X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?a=blobdiff_plain;f=hippotat%2F__init__.py;h=c1a6996bb862cb2a50cf08980ca3b35a5f679142;hb=ca386ea20592402ba35ced4f7f964f06a6263cab;hp=abc2a9fd687129eecd9732e258ac68168ed24711;hpb=d579a04817c1a141e715481a18fe7d71568c451d;p=hippotat.git diff --git a/hippotat/__init__.py b/hippotat/__init__.py index abc2a9f..c1a6996 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,21 +22,38 @@ from configparser import NoOptionError import collections import time +import codecs +import traceback import re as regexp -from twisted.python.constants import NamedConstant - 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] @@ -95,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 ---------- @@ -137,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 %s: %s' % (self._desc, 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('bufremain', d=self._buffer) def _maybe_packet(self, packet): - if len(packet): - self._on_packet(packet) + self._log('maybepacket', d=packet) + if len(packet): + self._on_packet(packet) def flush(self): + self._log('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): @@ -183,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) @@ -192,42 +221,57 @@ def queue_inbound(packet): class PacketQueue(): 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_debug(self, fn, pri, msg) - log_debug(pri, + def _log(self, dflag, msg, **kwargs): + log_debug(dflag, self._desc+' pq: '+msg, **kwargs) def append(self, packet): - log_data(DBG.QUEUE, packet, 'pq %s: append' % self._desc) + self._log(DBG.QUEUE, 'append', d=packet) self._pq.append((time.monotonic(), packet)) def nonempty(self): - log_debug(DBG.QUEUE, 'pq %s: nonempty ?' % self._desc) + 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) @@ -236,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 @@ -260,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__) @@ -317,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') @@ -329,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)