Skip to content

D
dmlib
Commit c23cda02 authored by nextime's avatar nextime
Browse files
Options

Moved all generic libs to nexlibs

parent 87bf22d9
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 4 additions and 2342 deletions
dmlib/daemonizer.py deleted 100644 → 0
View file @ 87bf22d9
#############################################################################
#
# $Id: daemonizer.py,v 1.3 2005/01/22 01:05:44 irmen Exp $
# Run Pyro servers as daemon processes on Unix/Linux.
# This won't work on other operating systems such as Windows.
# Author: Jeff Bauer (jbauer@rubic.com)
# This software is released under the MIT software license.
# Based on an earlier daemonize module by Jeffery Kunce
# Updated by Luis Camaano to double-fork-detach.
#
# This is part of "Pyro" - Python Remote Objects
# which is (c) Irmen de Jong - irmen@users.sourceforge.net
#
#############################################################################
import sys, os, time
from signal import SIGINT, SIGHUP
import logging
import pwd, grp
import getopt
log = logging.getLogger( 'DaemonStarter' )
class DaemonizerException:
def __init__(self, msg):
self.msg = msg
def __str__(self):
return self.msg
class Daemonizer:
"""
Daemonizer is a class wrapper to run a Pyro server program
in the background as daemon process. The only requirement
is for the derived class to implement a main_loop() method.
See Test class below for an example.
The following command line operations are provided to support
typical /etc/init.d startup/shutdown on Unix systems:
start | stop | restart
In addition, a daemonized program can be called with arguments:
status - check if process is still running
debug - run the program in non-daemon mode for testing
Note: Since Daemonizer uses fork(), it will not work on non-Unix
systems.
"""
def __init__(self, pidfile=None):
log.debug("Calling Daemonizer orig. init")
if not pidfile:
self.pidfile = "/tmp/%s.pid" % self.__class__.__name__.lower()
else:
self.pidfile = pidfile
self.chuid=False
self.chuid_actions=False
def become_daemon(self, root_dir='/'):
log.debug("Become Daemon")
if os.fork() != 0: # launch child and ...
os._exit(0) # kill off parent
log.debug("Calling setsid")
os.setsid()
log.debug("Calling chdir")
os.chdir(root_dir)
log.debug("Calling umask")
os.umask(0)
log.debug("Calling fork")
if os.fork() != 0: # fork again so we are not a session leader
os._exit(0)
log.debug("Wrapping stdin")
sys.stdin.close()
sys.__stdin__ = sys.stdin
log.debug("Wrapping stdout")
sys.stdout.close()
sys.stdout = sys.__stdout__ = _NullDevice()
log.debug("Wrapping stderr")
sys.stderr.close()
sys.stderr = sys.__stderr__ = _NullDevice()
#sys.stderr = sys.__stderr__ = log.debug
log.debug("Daemonized")
#for fd in range(1024):
# try:
# os.close(fd)
# except OSError:
# pass
def daemon_start(self, start_as_daemon=1, root_dir='/'):
log.debug("Daemon Start")
if start_as_daemon:
self.become_daemon(root_dir)
log.debug("Checking if the process is already running")
if self.is_process_running():
msg = "Unable to start server. Process is already running."
raise DaemonizerException(msg)
log.debug("Writing pid file")
f = open(self.pidfile, 'w')
f.write("%s" % os.getpid())
f.close()
log.debug("Calling main_loop")
if self.chuid:
log.debug("Change running privileges to "+str(self.chuid))
print 'Change running privileges to', str(self.chuid)
self.change_uid(self.chuid)
self.main_loop()
def daemon_stop(self):
pid = self.get_pid()
try:
os.kill(pid, SIGINT) # SIGTERM is too harsh...
time.sleep(1)
try:
os.unlink(self.pidfile)
except OSError:
pass
except IOError:
pass
def daemon_reload(self):
pid = self.get_pid()
try:
os.kill(pid, SIGHUP)
except:
pass
def get_pid(self):
try:
f = open(self.pidfile)
pid = int(f.readline().strip())
f.close()
except IOError:
pid = None
return pid
def is_process_running(self):
pid = self.get_pid()
if pid:
try:
os.kill(pid, 0)
return 1
except OSError:
pass
return 0
def main_loop(self):
"""NOTE: This method must be implemented in the derived class."""
msg = "main_loop method not implemented in derived class: %s" % \
self.__class__.__name__
raise DaemonizerException(msg)
def change_uid(self, uid):
c_user = uid
c_group = None
print os.getuid
if os.getuid() == 0:
if ':' in c_user:
c_user, c_group = c_user.split(":", 1)
cpw = pwd.getpwnam(c_user)
c_uid = cpw.pw_uid
if c_group:
cgr = grp.getgrnam(c_group)
c_gid = cgr.gr_gid
else:
c_gid = cpw.pw_gid
c_group = grp.getgrgid(cpw.pw_gid).gr_name
c_groups = []
for item in grp.getgrall():
if c_user in item.gr_mem:
c_groups.append(item.gr_gid)
if c_gid not in c_groups:
c_groups.append(c_gid)
if callable(self.chuid_actions):
self.chuid_actions(c_uid, c_gid)
os.setgid(c_gid)
os.setgroups(c_groups)
os.setuid(c_uid)
def process_command_line(self, argv, verbose=1, usagestr=None, chuid_actions=False):
usage = "usage: %s [options] start | stop | reload | restart | status | debug " \
"(run as non-daemon)\n\n" \
% os.path.basename(argv[0])
usage += "OPTIONS:\n"
usage += "\t--chuid=<username[:group]>\tchange username and group of the running process\n"
try:
optlist, args = getopt.getopt(argv[1:], "", ['chuid='])
for opt in optlist:
if opt[0] == '--chuid':
self.chuid=opt[1]
self.chuid_actions=chuid_actions
except:
pass
if usagestr:
usage += usagestr
if len(argv) < 2:
print usage
raise SystemExit
else:
operation = argv[len(argv)-1]
pid = self.get_pid()
if operation == 'status':
if self.is_process_running():
print "Server process %s is running." % pid
else:
print "Server is not running."
elif operation == 'start':
if self.is_process_running():
print "Server process %s is already running." % pid
raise SystemExit
else:
if verbose:
print "Starting server process."
self.daemon_start(1, os.path.abspath(os.path.dirname(argv[0])))
elif operation == 'stop':
if self.is_process_running():
self.daemon_stop()
if verbose:
print "Server process %s stopped." % pid
else:
print "Server process %s is not running." % pid
raise SystemExit
elif operation == 'reload':
if self.is_process_running():
if verbose:
print "Reloading server process."
self.daemon_reload()
else:
if verbose:
print "Server isn't running. Starting it."
self.daemon_start(1, os.path.abspath(os.path.dirname(argv[0])))
elif operation == 'restart':
self.daemon_stop()
if verbose:
print "Restarting server process."
self.daemon_start(1, os.path.abspath(os.path.dirname(argv[0])))
elif operation == 'debug':
self.daemon_start(0, os.path.abspath(os.path.dirname(argv[0])))
else:
print "Unknown operation:", operation
raise SystemExit
class _NullDevice:
"""A substitute for stdout/stderr that writes to nowhere."""
def isatty(self, *a, **kw):
return False
def write(self, s):
pass
def flush(self, s):
pass
class Test(Daemonizer):
def __init__(self):
Daemonizer.__init__(self)
def main_loop(self):
while 1:
time.sleep(1)
if __name__ == "__main__":
test = Test()
test.process_command_line(sys.argv)
dmlib/dmcrypt.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
import struct
import base64
from Crypto.Cipher import AES
try:
import hashlib
md5 = hashlib
md5.new = hashlib.md5
sha1 = hashlib.sha1
except:
import md5
import sha1
DMHASH=[
0x18a677e4,
0x11de784a,
0x51ae08b6,
0x22679cb0
]
def DMHash128(data):
import copy
reshash=copy.deepcopy(DMHASH)
i=0
while(len(data)<16):
#data+=struct.pack('B', struct.unpack('B', data[i])[0]^struct.unpack('B', data[len(data)-1])[0])
data+=struct.pack('<B', struct.unpack('<B', data[i])[0]^struct.unpack('<B', data[len(data)-1] if ord(data[len(data)-1]) else data[len(data)-2])[0])
i+=1
if not ord(data[len(data)-1]):
data=data[:-1]
data=struct.unpack('<4I', data[:16])
i=0
while(i<4):
reshash[i]=reshash[i]^data[i]
i+=1
return struct.pack('<4I', *reshash)
def DMHash256(data):
import copy
reshash=copy.deepcopy(DMHASH)
reshash+=reshash
i=0
while(len(data)<32):
#data+=struct.pack('B', struct.unpack('B', data[i])[0]^struct.unpack('B', data[len(data)-1])[0])
data+=struct.pack('<B', struct.unpack('<B', data[i])[0]^struct.unpack('<B', data[len(data)-1] if ord(data[len(data)-1]) else data[len(data)-2])[0])
i+=1
if not ord(data[len(data)-1]):
data=data[:-1]
data=struct.unpack('<8I', data[:32])
i=0
while(i<8):
reshash[i]=reshash[i]^data[i]
i+=1
i=0
while(i<4):
reshash[i]=reshash[i]^reshash[i+4]
i+=1
return struct.pack('<8I', *reshash)
class ARCFOUR(object):
def __init__(self,key,drop_n_bytes=768):
self.ksa = self.make_key(key,drop_n_bytes)
def encrypt_byte(self, i,j,S):
i = (i+1) % 256
j = (j+S[i]) % 256
S[i], S[j] = S[j],S[i]
K = S[(S[i] + S[j])%256]
return (i,j,K)
def make_key(self, key, drop_n_bytes):
#The key-scheduling algorithm (KSA)
S = [i for i in range(256)]
j = 0
for i in range(256):
j = (j + S[i] + ord(key[i % len(key)])) % 256
S[i], S[j] = S[j],S[i]
self.i = 0
self.j = 0
#Do the RC4-drop[(nbytes)]
if drop_n_bytes:
#i = j = 0
for dropped in range(drop_n_bytes):
self.i,self.j,K = self.encrypt_byte(self.i,self.j,S)
return S
def __crypt(self, message):
#The pseudo-random generation algorithm (PRGA)
S = list(self.ksa) #make a deep copy of you KSA array, gets modified
combined = []
counter = 0
i,j = self.i, self.j
for c in message:
i,j,K = self.encrypt_byte(i,j,S)
combined.append(struct.pack('B', (K ^ ord(c))))
crypted = (''.join(combined))
return crypted
def encode(self,message,encodeBase64=True):
crypted = self.__crypt(message)
if encodeBase64:
crypted = base64.urlsafe_b64encode(crypted)
return crypted
def decode(self,message,encodedBase64=True):
if encodedBase64:
message = base64.urlsafe_b64decode(message.encode())
return self.__crypt(message)
class XTEABlock(object):
cleandata=""
encdata=""
key=""
delta = 0x9e3779b9L
mask = 0xffffffffL
rounds=32
def __init__(self, key):
self.key=(int(key[0:][:8], 16), int(key[8:][:8], 16), int(key[16:][:8], 16), int(key[24:][:8], 16))
def setCleanData(self, data):
self.cleandata=data
self._encrypt(data)
def setEncryptData(self, data):
self.encdata=data
self._decrypt(data)
def _encrypt(self, data):
block=struct.unpack("<2L",data)
v0,v1=block[0], block[1]
sum=0L
for round in range(self.rounds):
v0 = (v0 + (((v1<<4 ^ v1>>5) + v1) ^ (sum + self.key[sum & 3]))) & self.mask
sum = (sum + self.delta) & self.mask
v1 = (v1 + (((v0<<4 ^ v0>>5) + v0) ^ (sum + self.key[sum>>11 & 3]))) & self.mask
self.encdata=struct.pack("<2L",v0,v1)
def _decrypt(self, data):
block=struct.unpack("<2L",data)
v0,v1=block[0], block[1]
sum = self.delta*self.rounds
for round in range(self.rounds):
v1 = (v1 - (((v0<<4 ^ v0>>5) + v0) ^ (sum + self.key[sum>>11 & 3]))) & self.mask
sum = (sum-self.delta) & self.mask;
v0 = (v0 - (((v1<<4 ^ v1>>5) + v1) ^ (sum + self.key[sum & 3]))) & self.mask
self.cleandata=struct.pack("<2L",v0,v1)
def raw_xxtea(v, n, k):
assert type(v) == type([])
assert type(k) == type([]) or type(k) == type(())
assert type(n) == type(1)
def MX():
return ((z>>5)^(y<<2)) + ((y>>3)^(z<<4))^(sum^y) + (k[(p & 3)^e]^z)
def u32(x):
return x & 0xffffffffL
y = v[0]
sum = 0
DELTA = 0x9e3779b9
if n > 1: # Encoding
z = v[n-1]
q = 6 + 52 / n
while q > 0:
q -= 1
sum = u32(sum + DELTA)
e = u32(sum >> 2) & 3
p = 0
while p < n - 1:
y = v[p+1]
z = v[p] = u32(v[p] + MX())
p += 1
y = v[0]
z = v[n-1] = u32(v[n-1] + MX())
return 0
elif n < -1: # Decoding
n = -n
q = 6 + 52 / n
sum = u32(q * DELTA)
while sum != 0:
e = u32(sum >> 2) & 3
p = n - 1
while p > 0:
z = v[p-1]
y = v[p] = u32(v[p] - MX())
p -= 1
z = v[n-1]
y = v[0] = u32(v[0] - MX())
sum = u32(sum - DELTA)
return 0
return 1
class BTEABlock(object):
cleandata=""
encdata=""
key=""
def __init__(self, key):
self.key=(int(key[0:][:8], 16), int(key[8:][:8], 16), int(key[16:][:8], 16), int(key[24:][:8], 16))
def setCleanData(self, data):
self.cleandata=data
self._encrypt(data)
def setEncryptData(self, data):
#print len(data)
self.encdata=data
self._decrypt(data)
def _encrypt(self, data):
ldata = len(data) / 4
block=list(struct.unpack("<%dL" % ldata ,data))
if raw_xxtea(block, ldata, self.key) == 0:
self.encdata=struct.pack("<%dL" % ldata, *block)
def _decrypt(self, data):
ldata = len(data) / 4
#print ldata, len(data)
block=list(struct.unpack("<%dL" % ldata ,data))
#print 'AAA', block
#print block
if raw_xxtea(block, -ldata, self.key) == 0:
#print 'AAA', block
self.cleandata=struct.pack("<%dL" % ldata ,*block)
class AES256(object):
cleandata=""
encdata=""
key=""
iv=""
def __init__(self, key, iv):
self.key=key
self.iv=iv
def setCleanData(self, data):
self.cleandata=data
self._encrypt(data)
def setEncryptData(self, data):
#print len(data)
self.encdata=data
self._decrypt(data)
def _encrypt(self, data):
ldata = len(data) / 4
key=struct.pack("<8L", *self.key)
iv=struct.pack("<4L", *self.iv)
obj=AES.new(key, AES.MODE_CBC, iv)
#block=list(struct.unpack("<%dL" % ldata ,data))
#enc=obj.encrypt(block)
#self.encdata=struct.pack("<%dL" % ldata, *enc)
self.encdata=obj.encrypt(data)
def _decrypt(self, data):
ldata = len(data) / 4
key=struct.pack("<8L", *self.key)
iv=struct.pack("<4L", *self.iv)
obj=AES.new(key, AES.MODE_CBC, iv)
#block=list(struct.unpack("<%dL" % ldata ,data))
#dec=obj.decrypt(block)
#self.cleandata=struct.pack("<%dL" % ldata ,*dec)
self.cleandata=obj.decrypt(data)
BLOCK_SIZE = 32
INTERRUPT = u'\u0001'
PAD = u'\u0000'
def AddPadding(data, interrupt, pad, block_size):
new_data = ''.join([data, interrupt])
new_data_len = len(new_data)
remaining_len = block_size - new_data_len
to_pad_len = remaining_len % block_size
pad_string = pad * to_pad_len
return ''.join([new_data, pad_string])
def StripPadding(data, interrupt, pad):
return data.rstrip(pad).rstrip(interrupt)
def B64AESEncrypt(key, iv, data):
c = md5.new()
c.update(key)
i = md5.new()
i.update(iv)
cipher = AES.new(c.hexdigest(), AES.MODE_CBC, i.hexdigest()[:16])
padded = AddPadding(data, INTERRUPT, PAD, BLOCK_SIZE)
encrypted = cipher.encrypt(padded)
return base64.b64encode(encrypted)
def B64AESDecrypt(key, iv, data):
c = md5.new()
c.update(key)
i = md5.new()
i.update(iv)
cipher = AES.new(c.hexdigest(), AES.MODE_CBC, i.hexdigest()[:16])
decoded = base64.b64decode(data)
try:
decrypted = cipher.decrypt(decoded)
except:
decrypted = ""
return decrypted
return StripPadding(decrypted, INTERRUPT, PAD)
dmlib/objgraph.py deleted 100644 → 0
View file @ 87bf22d9
This diff is collapsed.
dmlib/singleton.py deleted 100644 → 0
View file @ 87bf22d9
"""
__version__ = "$Revision: 1.2 $"
__date__ = "$Date: 2004/04/16 14:55:00 $"
"""
"""
A Python Singleton mixin class that makes use of some of the ideas
found at http://c2.com/cgi/wiki?PythonSingleton. Just inherit
from it and you have a singleton. No code is required in
subclasses to create singleton behavior -- inheritance from
Singleton is all that is needed.
Assume S is a class that inherits from Singleton. Useful behaviors
are:
1) Getting the singleton:
S.getInstance()
returns the instance of S. If none exists, it is created.
2) The usual idiom to construct an instance by calling the class, i.e.
S()
is disabled for the sake of clarity. If it were allowed, a programmer
who didn't happen notice the inheritance from Singleton might think he
was creating a new instance. So it is felt that it is better to
make that clearer by requiring the call of a class method that is defined in
Singleton. An attempt to instantiate via S() will restult in an SingletonException
being raised.
3) If S.__init__(.) requires parameters, include them in the
first call to S.getInstance(.). If subsequent calls have parameters,
a SingletonException is raised.
4) As an implementation detail, classes that inherit
from Singleton may not have their own __new__
methods. To make sure this requirement is followed,
an exception is raised if a Singleton subclass includ
es __new__. This happens at subclass instantiation
time (by means of the MetaSingleton metaclass.
By Gary Robinson, grobinson@transpose.com. No rights reserved --
placed in the public domain -- which is only reasonable considering
how much it owes to other people's version which are in the
public domain. The idea of using a metaclass came from
a comment on Gary's blog (see
http://www.garyrobinson.net/2004/03/python_singleto.html#comments).
Not guaranteed to be fit for any particular purpose.
RDS - 2004-04-16
To make a class a Singleton, inhert from singleton.Singleton.
Call <Class>.getInstance() to get the single instance of <Class>.
Any arguments passed to <Class>.getInstance() will be passed on
to your <Class>.__init__.
"""
class SingletonException(Exception):
pass
class MetaSingleton(type):
def __new__(metaclass, strName, tupBases, dict):
if dict.has_key('__new__'):
raise SingletonException, 'Can not override __new__ in a Singleton'
return super(MetaSingleton,metaclass).__new__(metaclass, strName, tupBases, dict)
def __call__(cls, *lstArgs, **dictArgs):
raise SingletonException, 'Singletons may only be instantiated through getInstance()'
class Singleton(object):
__metaclass__ = MetaSingleton
def getInstance(cls, *lstArgs):
"""
Call this to instantiate an instance or retrieve the existing instance.
If the singleton requires args to be instantiated, include them the first
time you call getInstance.
"""
if cls._isInstantiated():
if len(lstArgs) != 0:
raise SingletonException, 'If no supplied args, singleton must already be instantiated, or __init__ must require no args'
else:
if len(lstArgs) != cls._getConstructionArgCountNotCountingSelf():
raise SingletonException, 'If the singleton requires __init__ args, supply them on first instantiation'
instance = cls.__new__(cls)
instance.__init__(*lstArgs)
cls.cInstance = instance
return cls.cInstance
getInstance = classmethod(getInstance)
def _isInstantiated(cls):
return hasattr(cls, 'cInstance')
_isInstantiated = classmethod(_isInstantiated)
def _getConstructionArgCountNotCountingSelf(cls):
return cls.__init__.im_func.func_code.co_argcount - 1
_getConstructionArgCountNotCountingSelf = classmethod(_getConstructionArgCountNotCountingSelf)
def _forgetClassInstanceReferenceForTesting(cls):
"""
This is designed for convenience in testing -- sometimes you
want to get rid of a singleton during test code to see what
happens when you call getInstance() under a new situation.
To really delete the object, all external references to it
also need to be deleted.
"""
delattr(cls,'cInstance')
_forgetClassInstanceReferenceForTesting = classmethod(_forgetClassInstanceReferenceForTesting)
if __name__ == '__main__':
import unittest
class PublicInterfaceTest(unittest.TestCase):
def testReturnsSameObject(self):
"""
Demonstrates normal use -- just call getInstance and it returns a singleton instance
"""
class A(Singleton):
def __init__(self):
super(A, self).__init__()
a1 = A.getInstance()
a2 = A.getInstance()
self.assertEquals(id(a1), id(a2))
def testInstantiateWithMultiArgConstructor(self):
"""
If the singleton needs args to construct, include them in the first
call to get instances.
"""
class B(Singleton):
def __init__(self, arg1, arg2):
super(B, self).__init__()
self.arg1 = arg1
self.arg2 = arg2
b1 = B.getInstance('arg1 value', 'arg2 value')
b2 = B.getInstance()
self.assertEquals(b1.arg1, 'arg1 value')
self.assertEquals(b1.arg2, 'arg2 value')
self.assertEquals(id(b1), id(b2))
def testTryToInstantiateWithoutNeededArgs(self):
class B(Singleton):
def __init__(self, arg1, arg2):
super(B, self).__init__()
self.arg1 = arg1
self.arg2 = arg2
self.assertRaises(SingletonException, B.getInstance)
def testTryToInstantiateWithoutGetInstance(self):
"""
Demonstrates that singletons can ONLY be instantiated through
getInstance, as long as they call Singleton.__init__ during construction.
If this check is not required, you don't need to call Singleton.__init__().
"""
class A(Singleton):
def __init__(self):
super(A, self).__init__()
self.assertRaises(SingletonException, A)
def testDontAllowNew(self):
def instantiatedAnIllegalClass():
class A(Singleton):
def __init__(self):
super(A, self).__init__()
def __new__(metaclass, strName, tupBases, dict):
return super(MetaSingleton,metaclass).__new__(metaclass, strName, tupBases, dict)
self.assertRaises(SingletonException, instantiatedAnIllegalClass)
def testDontAllowArgsAfterConstruction(self):
class B(Singleton):
def __init__(self, arg1, arg2):
super(B, self).__init__()
self.arg1 = arg1
self.arg2 = arg2
b1 = B.getInstance('arg1 value', 'arg2 value')
self.assertRaises(SingletonException, B, 'arg1 value', 'arg2 value')
unittest.main()
\ No newline at end of file
dmlib/utils/__init__.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
dmlib/utils/blockingdefer.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
from twisted.internet import defer, reactor, threads
import time
class BlockingDefer(object):
result = None
errors = None
done = None
timeout = False
def run(self, d, timeout=10):
start = time.time()
d.addCallbacks(self.callbackDone, self.callbackError)
while not self.done:
if int(time.time()-start) > timeout:
self.timeout = True
self.done = True
reactor.iterate(0.05)
time.sleep(.05)
if self.timeout:
raise "Timeout Error"
else:
if not self.errors:
return self.result
else:
raise self.errors
def callbackError(self, err=True):
self.errors = err
self.done = True
def callbackDone(self, result):
self.result = result
self.done = True
def blockingDeferred(d, timeout=5):
bd=BlockingDefer()
ret=bd.run(d, timeout)
del bd
return ret
dmlib/utils/genutils.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
try:
import hashlib
md5 = hashlib
md5.new = hashlib.md5
sha1 = hashlib.sha1
except:
import md5
import sha1
class ConvenienceCaller(object):
"""
This metaclass build and abstraction of an object instance so
that other objects can call it as it is a module, and permit
to the original object to setup a specific callback
to the caller so it can manage the calls originated by the caller
with local methods
"""
def __init__(self, callerfunc):
"""
@params callerfunc: function/method to be used as callback
inside the original class of the abstract object
"""
self.callerfunc=callerfunc
def __getattribute__(self, name):
"""
This do the magic, transforming every called method in a call
to the callerfunction of the original object
This method isn't intended to be called directly!
"""
callerfunc = object.__getattribute__(self, 'callerfunc')
return callerfunc(name)
from ConfigParser import SafeConfigParser
def revlist(l): l.reverse(); return l
class FakeObject(object):
pass
class configFile(SafeConfigParser):
def __init__(self, config):
self.configfile = config
SafeConfigParser.__init__(self)
def readConfig(self):
return self.read(self.configfile)
def writeConfig(self):
fd = open(self.configfile, "w")
self.write(fd)
fd.close()
return True
def getOptions(self, section):
res = {}
for opt in self.items(section):
try:
res[opt[0]] = opt[1]
except:
pass
return res
class CircularList(list):
"""
A list that wraps around instead of throwing an index error.
Works like a regular list:
>>> cl = CircularList([1,2,3])
>>> cl
[1, 2, 3]
>>> cl[0]
1
>>> cl[-1]
3
>>> cl[2]
3
Except wraps around:
>>> cl[3]
1
>>> cl[-4]
3
Slices work
>>> cl[0:2]
[1, 2]
but only in range.
"""
def __getitem__(self, key):
# try normal list behavior
try:
return super(CircularList, self).__getitem__(key)
except IndexError:
pass
# key can be either integer or slice object,
# only implementing int now.
try:
index = int(key)
index = index % self.__len__()
return super(CircularList, self).__getitem__(index)
except ValueError:
raise TypeError
class CircularList2(list):
def __init__(self, sequence):
list.__init__(self, sequence)
self.i = 0
def set_index(self, i):
if i not in range(len(self)):
raise IndexError, 'Can\'t set index out of range'
else:
self.i = i
def next(self, n=1):
if self == []:
return None
if n < 0:
return self.prev(abs(n))
if self.i not in range(len(self)):
self.i = len(self) - 1
if self.i + n >= len(self):
i = self.i
self.set_index(0)
return self.next(n - len(self) + i)
else:
self.set_index(self.i + n)
return self[self.i]
def prev(self, n=1):
if self == []:
return None
if n < 0:
return self.next(abs(n))
if self.i not in range(len(self)):
self.i = len(self) - 1
if self.i - n < 0:
i = self.i
self.set_index(len(self) - 1)
return self.prev(n - i - 1)
else:
self.i -= n
return self[self.i]
class SliceCircular(CircularList2):
def getdata(self, howmany=1):
if(howmany > len(self)):
howmany=len(self)
i=self.i
distance=int((howmany-1)/2)
ret=[]
self.prev(distance+1)
while howmany > 0:
ret.append(self.next())
howmany-=1
self.i=i
return ret
def invertWord(word):
import struct
return struct.pack('<2B', struct.unpack('<2B', str(word[:2]))[1],
struct.unpack('<2B', str(word[:2]))[0])
def isIp(addr):
ip=addr.split(".")
if len(ip)==4:
for n in ip:
if not unicode(n).isnumeric() or int(n) <0 or int(n) > 255:
return False
return True
return False
def is_number(s):
try:
float(s) # for int, long and float
except ValueError:
try:
complex(s) # for complex
except ValueError:
return False
except:
return False
return True
def isTrue(d):
if str(d).lower() in ["1", "true", "yes", "si", "y"]:
return True
return False
def board_syspwd(cfgpwd):
if len(cfgpwd)>4:
return cfgpwd
return 'domotika'
def devs_adminpwd(cfgpwd):
if len(cfgpwd)>4:
return cfgpwd
return 'domotika'
def ip_match(mask, ip):
if mask=='255.255.255.255' or mask=='0.0.0.0':
return True
else:
return mask==ip
def hashPwd(pwd):
if pwd and len(pwd)>3:
s=sha1()
s.update(pwd)
return s.hexdigest()
return False
dmlib/utils/mail.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
import smtplib
from email import MIMEText, MIMEMultipart
from OpenSSL.SSL import SSLv3_METHOD
from twisted.internet import reactor
from twisted.mail.smtp import ESMTPSenderFactory
from twisted.internet.ssl import ClientContextFactory
from twisted.internet.defer import Deferred
import logging
from cStringIO import StringIO
log = logging.getLogger( 'SkylivedCore' )
# XXX: ATTENZIONE: la libreria email e' molto cambiata in python 2.5, verifica
# e rendi il tutto compatibile!
def sendmail(
authenticationUsername, authenticationSecret,
fromAddress, toAddress,
messageFile,
smtpHost, smtpPort=25, requiredAuth=False
):
"""
@param authenticationUsername: The username with which to authenticate.
@param authenticationSecret: The password with which to authenticate.
@param fromAddress: The SMTP reverse path (ie, MAIL FROM)
@param toAddress: The SMTP forward path (ie, RCPT TO)
@param messageFile: A file-like object containing the headers and body of
the message to send.
@param smtpHost: The MX host to which to connect.
@param smtpPort: The port number to which to connect.
@return: A Deferred which will be called back when the message has been
sent or which will errback if it cannot be sent.
"""
# Create a context factory which only allows SSLv3 and does not verify
# the peer's certificate.
contextFactory = ClientContextFactory()
contextFactory.method = SSLv3_METHOD
resultDeferred = Deferred()
senderFactory = ESMTPSenderFactory(
authenticationUsername,
authenticationSecret,
fromAddress,
toAddress,
messageFile,
resultDeferred,
contextFactory=contextFactory,
requireAuthentication=requiredAuth)
reactor.connectTCP(smtpHost, smtpPort, senderFactory)
return resultDeferred
def cbSentMessage(result):
"""
Called when the message has been sent.
Report success to the user and then stop the reactor.
"""
log.info("Message sent "+str(result))
def ebSentMessage(err):
"""
Called if the message cannot be sent.
Report the failure to the user and then stop the reactor.
"""
log.info("Error sending message "+str(err))
class GenericEmail:
def __init__(self, server="127.0.0.1"):
self.Subject = "This is a default subject"
self.From = "me"
self.To = "you"
self.Reply = "you"
self.Cc = False
self.server = server
self.serverport = 25
self.username = 'test@mail.com'
self.password = 'passwd'
self.msg=MIMEText.MIMEText("Default message")
def SetTo(self, to):
self.To=to
#self.Reply=to
def SetCc(self, cc):
self.Cc=cc
def SetSubject(self, subject):
self.Subject=subject
def SetFrom(self, From):
self.From=From
self.Reply=From
def SetReply(self, reply):
self.Reply=reply
def Send(self):
msg=self.msg
msg['Subject'] = self.Subject
msg['From'] = self.From
self._from = self.From
if "<" in self.From:
self._from = ""
start=False
for c in self.From:
if start and c=='>':
start=False
if start:
self._from += c
if c == '<':
start=True
msg['To'] = self.To
msg['Reply-To'] = self.Reply
if self.Cc:
msg['Cc'] = self.Cc
#s=smtplib.SMTP(self.server)
#s.connect()
#s.sendmail(self.From, [self.To], msg.as_string())
#s.close()
toAddress = self.To
if self.Cc:
try:
toAddress.append(self.Cc)
except:
toAddress = [self.To, self.Cc]
result = sendmail(
self.username,
self.password,
self._from,
toAddress,
StringIO(msg.as_string()),
self.server,
self.serverport)
result.addCallbacks(cbSentMessage, ebSentMessage)
class TextEmail(GenericEmail):
def SetMsg(self, msg):
self.msg=MIMEText.MIMEText(msg)
class HTMLEmail(GenericEmail):
def __init__(self, *args, **kwargs):
GenericEmail.__init__(self, *args, **kwargs)
self.msg=MIMEMultipart.MIMEMultipart('alternative')
self._txtmsg = MIMEText.MIMEText('default txt', 'plain')
self._htmlmsg = MIMEText.MIMEText('<html><body>default html</body></html>', 'html')
def SetTextMsg(self, msg):
self._txtmsg = MIMEText.MIMEText(msg, 'plain')
def SetHtmlMsg(self, msg):
self._htmlmsg = MIMEText.MIMEText(msg, 'html')
def Send(self):
self.msg.attach(self._htmlmsg)
self.msg.attach(self._txtmsg)
return GenericEmail.Send(self)
dmlib/utils/pwgen.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
from random import Random
lefthand="789yuiophjklbnmYUIOPHJKLBNM"
righthand="123456qwertyasdfgzxcvQWERTYASDFGZXCV"
allchars=lefthand+righthand
def GeneratePwd(leng=8, alt=False):
rng = Random()
pwd=""
for i in range(leng):
if not alt:
pwd+=rng.choice(allchars)
else:
if i%2:
pwd+=rng.choice(lefthand)
else:
pwd+=rng.choice(righthand)
return pwd
def GenerateHexKey(leng=8):
import struct
rng = Random()
pwd=[]
for i in range(leng):
pwd.append(rng.randint(0, 255))
return struct.pack('<%dB' % leng, *pwd)
def generateIV128(pwdhash):
from genutils import invertWord
import struct
res=""
i=0
while(i<16):
a=struct.unpack('<H',invertWord(pwdhash[i:i+2]))[0]
b=struct.unpack('<H',pwdhash[i+16:i+16+2])[0]
res+=struct.pack('<H', a^b)
i+=2
return res
dmlib/utils/validators.py deleted 100644 → 0
View file @ 87bf22d9
###########################################################################
# Copyright (c) 2011-2014 Unixmedia S.r.l. <info@unixmedia.it>
# Copyright (c) 2011-2014 Franco (nextime) Lanza <franco@unixmedia.it>
#
# Domotika System Controller Daemon "domotikad" [http://trac.unixmedia.it]
#
# This file is part of domotikad.
#
# domotikad is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
from zope.interface import implements
from twisted.internet import defer
import formal
from formal import iformal
REGEX_IP_ADDRESS=('b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?).)'
'{3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)b')
REGEX_EMAIL="^[a-zA-Z0-9]+([\._\-a-zA-Z0-9]+)*@[a-zA-Z0-9]+([._\-a-zA-Z0-9]+[\.][a-zA-Z]{2,5})+$"
REGEX_URI=("(http|https|mail|telnet|ftp|imap|irc|file):\/\/[a-z0-9]+"
"([\-\.]{1}[a-z0-9]+)*\.[a-z]{2,5}((:[0-9]{1,5})?\/.*)?$")
class GenericValidator(object):
def __init__(self, errmsg=u'Cannot validate'):
self.errmsg = errmsg
def check(self, value):
return False
def validate(self, field, value):
if not self.check(value):
raise formal.FieldValidationError(self.errmsg)
class GenericWordListValidator(GenericValidator):
def __init__(self, wordlist=[], errmsg=u'Word not in list'):
self.wordlist = wordlist
GenericValidator.__init__(self, errmsg)
def check(self, value):
if value in self.wordlist:
return True
else:
return False
class YesNoValidator(GenericWordListValidator):
def __init__(self, errmsg=u'Only yes or not are valid values'):
GenericWordListValidator.__init__(self, ['yes', 'no'], errmsg)
class AcceptPolicy(GenericWordListValidator):
def __init__(self, errmsg=u'You must accept our terms of usage!'):
GenericWordListValidator.__init__(self, ['yes'], errmsg)
dmlib/utils/webutils.py deleted 100644 → 0
View file @ 87bf22d9
This diff is collapsed.
setup.py
View file @ c23cda02
......@@ -3,11 +3,11 @@
from distutils.core import setup
setup(name='dmlib',
version='0.1',
version='0.2',
description='Python Domotika Libs',
author='Franco (nextime) Lanza',
author_email='franco@unixmedia.it',
url='http://www.unixmedia.it/',
packages=['dmlib', 'dmlib/utils'],
author_email='franco@nexlab.it',
url='http://www.nexlab.net/',
packages=['dmlib'],
)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment