"""networked spam-signature detection server"""
from __future__ import division
import os
import sys
import SocketServer
import time
import gdbm
import cStringIO
import traceback
import threading
import pyzor
from pyzor import *
__author__ = pyzor.__author__
__version__ = pyzor.__version__
__revision__ = "$Id: server.py,v 1.29 2002-10-09 00:45:45 ftobin Exp $"
class AuthorizationError(pyzor.CommError):
"""signature was valid, but not permitted to
do the requested action"""
pass
class ACL(object):
__slots__ = ['entries']
default_allow = False
def __init__(self):
self.entries = []
def add_entry(self, entry):
typecheck(entry, ACLEntry)
self.entries.append(entry)
def allows(self, user, op):
typecheck(user, Username)
typecheck(op, Opname)
for entry in self.entries:
if entry.allows(user, op):
return True
if entry.denies(user, op):
return False
return self.default_allow
class ACLEntry(tuple):
all_keyword = 'all'.lower()
def __init__(self, v):
(user, op, allow) = v
typecheck(user, Username)
typecheck(op, Opname)
assert bool(allow) == allow
def user(self):
return self[0]
user = property(user)
def op(self):
return self[1]
op = property(op)
def allow(self):
return self[2]
allow = property(allow)
def allows(self, user, op):
return self._says(user, op, True)
def denies(self, user, op):
return self._says(user, op, False)
def _says(self, user, op, allow):
"""If allow is True, we return true if and only if we allow user to do op.
If allow is False, we return true if and only if we deny user to do op
"""
typecheck(user, Username)
typecheck(op, Opname)
assert bool(allow) == allow
return (self.allow == allow
and (self.user == user
or self.user.lower() == self.all_keyword)
and (self.op == op
or self.op.lower() == self.all_keyword))
class AccessFile(object):
# I started doing an iterator protocol for this, but it just
# got too complicated keeping track of everything on the line
__slots__ = ['file', 'output', 'lineno']
allow_keyword = 'allow'
deny_keyword = 'deny'
def __init__(self, f):
self.output = Output()
self.file = f
self.lineno = 0
def feed_into(self, acl):
typecheck(acl, ACL)
for orig_line in self.file:
self.lineno += 1
line = orig_line.strip()
if not line or line.startswith('#'):
continue
parts = line.split(':')
if len(parts) != 3:
self.output.warn("access file: invalid number of parts in line %d"
% self.lineno)
continue
(ops_str, users_str, allow_str) = parts
ops = []
for op_str in ops_str.split():
try:
op = Opname(op_str)
except ValueError, e:
self.output.warn("access file: invalid opname %s line %d: %s"
% (repr(op_str), self.lineno, e))
else:
ops.append(op)
users = []
for u in users_str.split():
try:
user = Username(u)
except ValueError, e:
self.output.warn("access file: invalid username %s line %d: %s"
% (repr(u), self.lineno, e))
else:
users.append(user)
allow_str = allow_str.strip()
if allow_str.lower() == self.allow_keyword:
allow = True
elif allow_str.lower() == self.deny_keyword:
allow = False
else:
self.output.warn("access file: invalid allow/deny keyword %s line %d"
% (repr(allow_str), self.lineno))
continue
for op in ops:
for user in users:
acl.add_entry(ACLEntry((user, op, allow)))
class Passwd(dict):
def __setitem__(self, k, v):
typecheck(k, pyzor.Username)
typecheck(v, long)
super(Passwd, self).__setitem__(k, v)
class PasswdFile(BasicIterator):
"""Iteration gives (Username, long) objects
Format of file is:
user : key
"""
__slots__ = ['file', 'output', 'lineno']
def __init__(self, f):
self.file = f
self.output = Output()
self.lineno = 0
def next(self):
while True:
orig_line = self.file.readline()
self.lineno += 1
if not orig_line:
raise StopIteration
line = orig_line.strip()
if not line or line.startswith('#'):
continue
fields = line.split(':')
fields = map(lambda x: x.strip(), fields)
if len(fields) != 2:
self.output.warn("passwd line %d is invalid (wrong number of parts)"
% self.lineno)
continue
try:
return (Username(fields[0]), long(fields[1], 16))
except ValueError, e:
self.output.warn("invalid passwd entry line %d: %s"
% (self.lineno, e))
class Log(object):
__slots__ = ['fp']
def __init__(self, fp=None):
self.fp = fp
def log(self, address, user=None, command=None, arg=None, code=None):
# we don't use defaults because we want to be able
# to pass in None
if user is None: user = ''
if command is None: command = ''
if arg is None: arg = ''
if code is None: code = -1
# We duplicate the time field merely so that
# humans can peruse through the entries without processing
ts = int(time.time())
if self.fp is not None:
self.fp.write("%s\n" %
','.join((("%d" % ts),
time.ctime(ts),
user,
address[0],
command,
repr(arg),
("%d" % code)
)))
self.fp.flush()
class Record(object):
"""Prefix conventions used in this class:
r = report (spam)
wl = whitelist
"""
__slots__ = ['r_count', 'r_entered', 'r_updated',
'wl_count', 'wl_entered', 'wl_updated',
]
fields = ('r_count', 'r_entered', 'r_updated',
'wl_count', 'wl_entered', 'wl_updated',
)
this_version = '1'
# epoch seconds
never = -1
def __init__(self, r_count=0, wl_count=0):
self.r_count = r_count
self.wl_count = wl_count
self.r_entered = self.never
self.r_updated = self.never
self.wl_entered = self.never
self.wl_updated = self.never
def wl_increment(self):
# overflow prevention
if self.wl_count < sys.maxint:
self.wl_count += 1
if self.wl_entered == self.never:
self.wl_entered = int(time.time())
self.wl_update()
def r_increment(self):
# overflow prevention
if self.r_count < sys.maxint:
self.r_count += 1
if self.r_entered == self.never:
self.r_entered = int(time.time())
self.r_update()
def r_update(self):
self.r_updated = int(time.time())
def wl_update(self):
self.wl_updated = int(time.time())
def __str__(self):
return "%s,%d,%d,%d,%d,%d,%d" \
% ((self.this_version,)
+ tuple(map(lambda x: getattr(self, x), self.fields)))
def from_str(self, s):
parts = s.split(',')
dispatch = None
version = parts[0]
if len(parts) == 3:
dispatch = self.from_str_0
elif version == '1':
dispatch = self.from_str_1
else:
raise StandardError, ("don't know how to handle db value %s"
% repr(s))
return apply(dispatch, (s,))
from_str = classmethod(from_str)
def from_str_0(self, s):
r = Record()
parts = s.split(',')
fields = ('r_count', 'r_entered', 'r_updated')
assert len(parts) == len(fields)
for i in range(len(parts)):
setattr(r, fields[i], int(parts[i]))
return r
from_str_0 = classmethod(from_str_0)
def from_str_1(self, s):
r = Record()
parts = s.split(',')[1:]
assert len(parts) == len(self.fields)
for i in range(len(parts)):
setattr(r, self.fields[i], int(parts[i]))
return r
from_str_1 = classmethod(from_str_1)
class DBHandle(Singleton):
__slots__ = ['output', 'initialized']
db_lock = threading.Lock()
max_age = 3600*24*30*4 # 3 months
db = None
sync_period = 60
reorganize_period = 3600*24 # 1 day
def __init__(self):
assert self.db is not None, "database was not initialized"
def initialize(self, fn, mode):
self.output = Output()
self.db = gdbm.open(fn, mode)
self.start_reorganizing()
self.start_syncing()
initialize = classmethod(initialize)
def apply_locking_method(self, method, varargs=(), kwargs={}):
# just so we don't carry around a mutable kwargs
if kwargs == {}:
kwargs = {}
self.output.debug("acquiring lock")
self.db_lock.acquire()
self.output.debug("acquired lock")
try:
result = apply(method, varargs, kwargs)
finally:
self.output.debug("releasing lock")
self.db_lock.release()
self.output.debug("released lock")
return result
apply_locking_method = classmethod(apply_locking_method)
def __getitem__(self, key):
return self.apply_locking_method(self._really_getitem, (key,))
def _really_getitem(self, key):
return self.db[key]
def __setitem__(self, key, value):
self.apply_locking_method(self._really_setitem, (key, value))
def _really_setitem(self, key, value):
self.db[key] = value
def start_syncing(self):
self.apply_locking_method(self._really_sync)
self.sync_timer = threading.Timer(self.sync_period,
self.start_syncing)
self.sync_timer.start()
start_syncing = classmethod(start_syncing)
def _really_sync(self):
self.db.sync()
_really_sync = classmethod(_really_sync)
def start_reorganizing(self):
self.apply_locking_method(self._really_reorganize)
self.reorganize_timer = threading.Timer(self.reorganize_period,
self.start_reorganizing)
self.reorganize_timer.start()
start_reorganizing = classmethod(start_reorganizing)
def _really_reorganize(self):
self.output.debug("reorganizing the database")
key = self.db.firstkey()
breakpoint = time.time() - self.max_age
while key is not None:
rec = Record.from_str(self.db[key])
delkey = None
if rec.r_updated < breakpoint:
self.output.debug("deleting key %s" % key)
delkey = key
key = self.db.nextkey(key)
if delkey:
del self.db[delkey]
self.db.reorganize()
_really_reorganize = classmethod(_really_reorganize)
class Server(SocketServer.ThreadingUDPServer, object):
max_packet_size = 8192
time_diff_allowance = 180
def __init__(self, address, log):
typecheck(log, Log)
self.output = Output()
RequestHandler.output = self.output
RequestHandler.log = log
self.output.debug('listening on %s' % str(address))
super(Server, self).__init__(address, RequestHandler)
def serve_forever(self):
self.pid = os.getpid()
super(Server, self).serve_forever()
def replace_log(self, newlog):
typecheck(newlog, Log)
RequestHandler.log = newlog
self.output.debug("changing logfile")
class RequestHandler(SocketServer.DatagramRequestHandler, object):
def setup(self):
super(RequestHandler, self).setup()
# This is to work around a bug in current versions
# of Python. The bug has been reported, and fixed
# in Python's CVS.
self.wfile = cStringIO.StringIO()
self.client_address = Address(self.client_address)
self.out_msg = Response()
self.user = None
self.op = None
self.op_arg = None
self.out_code = None
self.msg_thread = None
def handle(self):
try:
self._really_handle()
except UnsupportedVersionError, e:
self.handle_error(505, "Version Not Supported: %s" % e)
except NotImplementedError, e:
self.handle_error(501, "Not implemented: %s" % e)
except (ProtocolError, KeyError), e:
# We assume that KeyErrors are due to not
# finding a key in the RFC822 message
self.handle_error(400, "Bad request: %s" % e)
except AuthorizationError, e:
self.handle_error(401, "Unauthorized: %s" % e)
except SignatureError, e:
self.handle_error(401, "Unauthorized, Signature Error: %s" % e)
except Exception, e:
self.handle_error(500, "Internal Server Error: %s" % e)
traceback.print_exc()
self.out_msg.setdefault('Code', str(self.out_msg.ok_code))
self.out_msg.setdefault('Diag', 'OK')
self.out_msg.init_for_sending()
self.log.log(self.client_address, self.user, self.op, self.op_arg,
int(self.out_msg['Code']))
msg_str = str(self.out_msg)
self.output.debug("sending: %s" % repr(msg_str))
self.wfile.write(msg_str)
def _really_handle(self):
"""handle() without the exception handling"""
self.output.debug("received: %s" % repr(self.packet))
signed_msg = MacEnvelope(self.rfile)
self.user = Username(signed_msg['User'])
if self.user != pyzor.anonymous_user:
if self.server.passwd.has_key(self.user):
signed_msg.verify_sig(self.server.passwd[self.user])
else:
raise SignatureError, "unknown user"
self.in_msg = signed_msg.get_submsg(pyzor.Request)
self.msg_thread = self.in_msg.get_thread()
# We take the int() of the proto versions because
# if the int()'s are the same, then they should be compatible
if int(self.in_msg.get_protocol_version()) != int(proto_version):
raise UnsupportedVersionError
self.out_msg.set_thread(self.msg_thread)
self.op = Opname(self.in_msg.get_op())
if not self.server.acl.allows(self.user, self.op):
raise AuthorizationError, "user is unauthorized to request the operation"
self.output.debug("got a %s command from %s" %
(self.op, self.client_address))
if not self.dispatches.has_key(self.op):
raise NotImplementedError, "requested operation is not implemented"
dispatch = self.dispatches[self.op]
if dispatch is not None:
apply(dispatch, (self,))
def handle_error(self, code, s):
self.out_msg = ErrorResponse(code, s)
if self.msg_thread is None:
self.out_msg.set_thread(ThreadId(0))
else:
self.out_msg.set_thread(self.msg_thread)
def handle_check(self):
digest = self.in_msg['Op-Digest']
self.op_arg = digest
self.output.debug("request to check digest %s" % digest)
db = DBHandle()
try:
rec = Record.from_str(db[digest])
r_count = rec.r_count
wl_count = rec.wl_count
except KeyError:
r_count = 0
wl_count = 0
self.out_msg['Count'] = "%d" % r_count
self.out_msg['WL-Count'] = "%d" % wl_count
def handle_report(self):
digest = self.in_msg['Op-Digest']
self.op_arg = digest
self.output.debug("request to report digest %s" % digest)
db = DBHandle()
try:
rec = Record.from_str(db[digest])
except KeyError:
rec = Record()
rec.r_increment()
db[digest] = str(rec)
def handle_whitelist(self):
digest = self.in_msg['Op-Digest']
self.op_arg = digest
self.output.debug("request to whitelist digest %s" % digest)
db = DBHandle()
try:
rec = Record.from_str(db[digest])
except KeyError:
rec = Record()
rec.wl_increment()
db[digest] = str(rec)
def handle_info(self):
digest = self.in_msg['Op-Digest']
self.op_arg = digest
self.output.debug("request to check digest %s" % digest)
db = DBHandle()
try:
record = Record.from_str(db[digest])
except KeyError:
record = Record()
r_count = record.r_count
wl_count = record.wl_count
self.out_msg['Entered'] = "%d" % record.r_entered
self.out_msg['Updated'] = "%d" % record.r_updated
self.out_msg['WL-Entered'] = "%d" % record.wl_entered
self.out_msg['WL-Updated'] = "%d" % record.wl_updated
self.out_msg['Count'] = "%d" % r_count
self.out_msg['WL-Count'] = "%d" % wl_count
dispatches = { 'check': handle_check,
'report': handle_report,
'ping': None,
'info': handle_info,
'whitelist': handle_whitelist,
}
|