X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=hippotat.git;a=blobdiff_plain;f=hippotat%2F__init__.py;h=ae13eece753bb87b17131c48a32f3173aabe1ef8;hp=9bfe280e4c8c8d8b5d158a3b382f611034e5cccb;hb=db6ba5840b00c61fe6e576b54179ddeb30202b0e;hpb=4edf77a3f6ad7ec05ad0809fc4c0408e6ca6405c diff --git a/hippotat/__init__.py b/hippotat/__init__.py index 9bfe280..ae13eec 100644 --- a/hippotat/__init__.py +++ b/hippotat/__init__.py @@ -7,8 +7,11 @@ import sys import twisted from twisted.internet import reactor -from twisted.logger import LogLevel import twisted.internet.endpoints +import twisted.logger +from twisted.logger import LogLevel +import twisted.python.constants +from twisted.python.constants import NamedConstant import ipaddress from ipaddress import AddressValueError @@ -19,11 +22,39 @@ from configparser import NoOptionError import collections import time +import codecs +import traceback import re as regexp import hippotat.slip as slip +class DBG(twisted.python.constants.Names): + ROUTE = NamedConstant() + DROP = NamedConstant() + FLOW = NamedConstant() + HTTP = NamedConstant() + HTTP_CTRL = NamedConstant() + INIT = NamedConstant() + QUEUE = NamedConstant() + QUEUE_CTRL = NamedConstant() + HTTP_FULL = NamedConstant() + SLIP_FULL = NamedConstant() + +_hex_codec = codecs.getencoder('hex_codec') + +log = twisted.logger.Logger() + +def log_debug(dflag, msg, idof=None, d=None): + #print('---------------->',repr((dflag, msg, idof, d)), file=sys.stderr) + if idof is not None: + msg = '[%d] %s' % (id(idof), msg) + if d is not None: + #d = d[0:64] + d = _hex_codec(d)[0].decode('ascii') + msg += ' ' + d + log.info('{dflag} {msgcore}', dflag=dflag, msgcore=msg) + defcfg = ''' [DEFAULT] #[] overrides @@ -84,10 +115,10 @@ class ConfigResults: c = ConfigResults() -def log_discard(packet, saddr, daddr, why): - print('DROP ', saddr, daddr, why) -# syslog.syslog(syslog.LOG_DEBUG, -# 'discarded packet %s -> %s (%s)' % (saddr, daddr, why)) +def log_discard(packet, iface, saddr, daddr, why): + log_debug(DBG.DROP, + 'discarded packet [%s] %s -> %s: %s' % (iface, saddr, daddr, why), + d=packet) #---------- packet parsing ---------- @@ -126,38 +157,46 @@ def ipnetwork(input): #---------- ipif (SLIP) subprocess ---------- class SlipStreamDecoder(): - def __init__(self, on_packet): - # we will call packet() + def __init__(self, desc, on_packet): self._buffer = b'' self._on_packet = on_packet + self._desc = desc + self._log('__init__') + + def _log(self, msg, **kwargs): + log_debug(DBG.SLIP_FULL, 'slip '+msg, **kwargs) def inputdata(self, data): - #print('SLIP-GOT ', repr(data)) - self._buffer += data - packets = slip.decode(self._buffer) + self._log('inputdata', d=data) + data = self._buffer + data + self._buffer = b'' + packets = slip.decode(data) self._buffer = packets.pop() for packet in packets: self._maybe_packet(packet) + self._log('inputdata bufremain', d=self._buffer) def _maybe_packet(self, packet): - if len(packet): - self._on_packet(packet) + self._log('inputdata maybepacket', d=packet) + if len(packet): + self._on_packet(packet) def flush(self): + self._log('inputdata flush') self._maybe_packet(self._buffer) self._buffer = b'' class _IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol): def __init__(self, router): self._router = router - self._decoder = SlipStreamDecoder(self.slip_on_packet) + self._decoder = SlipStreamDecoder('ipif', self.slip_on_packet) def connectionMade(self): pass def outReceived(self, data): self._decoder.inputdata(data) def slip_on_packet(self, packet): (saddr, daddr) = packet_addrs(packet) if saddr.is_link_local or daddr.is_link_local: - log_discard(packet, saddr, daddr, 'link-local') + log_discard(packet, 'ipif', saddr, daddr, 'link-local') return self._router(packet, saddr, daddr) def processEnded(self, status): @@ -172,6 +211,7 @@ def start_ipif(command, router): env=None) def queue_inbound(packet): + log_debug(DBG.FLOW, "queue_inbound", d=packet) ipif.transport.write(slip.delimiter) ipif.transport.write(slip.encode(packet)) ipif.transport.write(slip.delimiter) @@ -179,38 +219,59 @@ def queue_inbound(packet): #---------- packet queue ---------- class PacketQueue(): - def __init__(self, max_queue_time): + def __init__(self, desc, max_queue_time): + self._desc = desc + assert(desc + '') self._max_queue_time = max_queue_time self._pq = collections.deque() # packets + def _log(self, dflag, msg, **kwargs): + log_debug(dflag, self._desc+' pq: '+msg, **kwargs) + def append(self, packet): + self._log(DBG.QUEUE, 'append', d=packet) self._pq.append((time.monotonic(), packet)) def nonempty(self): + self._log(DBG.QUEUE, 'nonempty ?') while True: try: (queuetime, packet) = self._pq[0] - except IndexError: return False + except IndexError: + self._log(DBG.QUEUE, 'nonempty ? empty.') + return False age = time.monotonic() - queuetime if age > self._max_queue_time: # strip old packets off the front + self._log(DBG.QUEUE, 'dropping (old)', d=packet) self._pq.popleft() continue + self._log(DBG.QUEUE, 'nonempty ? nonempty.') return True def process(self, sizequery, moredata, max_batch): # sizequery() should return size of batch so far # moredata(s) should add s to batch + self._log(DBG.QUEUE, 'process...') while True: try: (dummy, packet) = self._pq[0] - except IndexError: break + except IndexError: + self._log(DBG.QUEUE, 'process... empty') + break + + self._log(DBG.QUEUE_CTRL, 'process... packet', d=packet) encoded = slip.encode(packet) sofar = sizequery() + self._log(DBG.QUEUE_CTRL, + 'process... (sofar=%d, max=%d) encoded' % (sofar, max_batch), + d=encoded) + if sofar > 0: if sofar + len(slip.delimiter) + len(encoded) > max_batch: + self._log(DBG.QUEUE_CTRL, 'process... overflow') break moredata(slip.delimiter) @@ -219,7 +280,11 @@ class PacketQueue(): #---------- error handling ---------- +_crashing = False + def crash(err): + global _crashing + _crashing = True print('CRASH ', err, file=sys.stderr) try: reactor.stop() except twisted.internet.error.ReactorNotRunning: pass @@ -243,7 +308,7 @@ def process_cfg_ipif(section, varmap): except AttributeError: continue setattr(c, d, v) - print(repr(c)) + #print(repr(c)) c.ipif_command = cfg.get(section,'ipif', vars=c.__dict__) @@ -300,7 +365,11 @@ def process_cfg_clients(constructor): #---------- startup ---------- def common_startup(): - twisted.logger.globalLogPublisher.addObserver(crash_on_critical) + log_formatter = twisted.logger.formatEventAsClassicLogText + log_observer = twisted.logger.FileLogObserver(sys.stderr, log_formatter) + twisted.logger.globalLogBeginner.beginLoggingTo( + [ log_observer, crash_on_critical ] + ) optparser.add_option('-c', '--config', dest='configfile', default='/etc/hippotat/config') @@ -312,5 +381,6 @@ def common_startup(): cfg.read(opts.configfile) def common_run(): - reactor.run() + log_debug(DBG.INIT, 'entering reactor') + if not _crashing: reactor.run() print('CRASHED (end)', file=sys.stderr)