#!/usr/bin/python3 import signal signal.signal(signal.SIGINT, signal.SIG_DFL) import sys import os import twisted import twisted.internet import twisted.internet.endpoints from twisted.internet import reactor from twisted.web.server import NOT_DONE_YET from twisted.logger import LogLevel #import twisted.web.server import Site #from twisted.web.resource import Resource from optparse import OptionParser from configparser import ConfigParser from configparser import NoOptionError import collections import syslog from hippotat import * clients = { } defcfg = ''' [DEFAULT] max_batch_down = 65536 max_queue_time = 10 max_request_time = 54 [virtual] mtu = 1500 # network # [host] # [relay] [server] ipif = userv root ipif %(host)s,%(relay)s,%(mtu)s,slip %(network)s addrs = 127.0.0.1 ::1 port = 8099 [limits] max_batch_down = 262144 max_queue_time = 121 max_request_time = 121 ''' #---------- error handling ---------- def crash(err): print('CRASH ', err, file=sys.stderr) try: reactor.stop() except twisted.internet.error.ReactorNotRunning: pass def crash_on_defer(defer): defer.addErrback(lambda err: crash(err)) def crash_on_critical(event): if event.get('log_level') >= LogLevel.critical: crash(twisted.logger.formatEvent(event)) #---------- "router" ---------- def route(packet, saddr, daddr): print('TRACE ', saddr, daddr, packet) try: client = clients[daddr] except KeyError: dclient = None if dclient is not None: dclient.queue_outbound(packet) elif saddr.is_link_local or daddr.is_link_local: log_discard(packet, saddr, daddr, 'link-local') elif daddr == host or daddr not in network: print('TRACE INBOUND ', saddr, daddr, packet) queue_inbound(packet) elif daddr == relay: log_discard(packet, saddr, daddr, 'relay') else: log_discard(packet, saddr, daddr, 'no client') 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)) #---------- client ---------- class Client(): def __init__(self, ip, cs): # instance data members self._ip = ip self._cs = cs self.pw = cfg.get(cs, 'password') self._rq = collections.deque() # requests self._pq = collections.deque() # packets # plus from config: # .max_batch_down # .max_queue_time # .max_request_time for k in ('max_batch_down','max_queue_time','max_request_time'): req = cfg.getint(cs, k) limit = cfg.getint('limits',k) self.__dict__[k] = min(req, limit) def process_arriving_data(self, d): for packet in slip.decode(d): (saddr, daddr) = packet_addrs(packet) if saddr != self._ip: raise ValueError('wrong source address %s' % saddr) route(packet, saddr, daddr) def _req_cancel(self, request): request.finish() def _req_error(self, err, request): self._req_cancel(request) def queue_outbound(self, packet): self._pq.append((time.monotonic(), packet)) def http_request(self, request): request.setHeader('Content-Type','application/octet-stream') reactor.callLater(self.max_request_time, self._req_cancel, request) request.notifyFinish().addErrback(self._req_error, request) self._rq.append(request) self._check_outbound() def _check_outbound(self): while True: try: request = self._rq[0] except IndexError: request = None if request and request.finished: self._rq.popleft() continue # now request is an unfinished request, or None try: (queuetime, packet) = self._pq[0] except IndexError: # no packets, oh well break age = time.monotonic() - queuetime if age > self.max_queue_time: self._pq.popleft() continue if request is None: # no request break # request, and also some non-expired packets while True: try: (dummy, packet) = self._pq[0] except IndexError: break encoded = slip.encode(packet) if request.sentLength > 0: if (request.sentLength + len(slip.delimiter) + len(encoded) > self.max_batch_down): break request.write(slip.delimiter) request.write(encoded) self._pq.popLeft() assert(request.sentLength) self._rq.popLeft() request.finish() # round again, looking for more to do class IphttpResource(twisted.web.resource.Resource): isLeaf = True def render_POST(self, request): # find client, update config, etc. ci = ipaddr(request.args['i']) c = clients[ci] pw = request.args['pw'] if pw != c.pw: raise ValueError('bad password') # update config for r, w in (('mbd', 'max_batch_down'), ('mqt', 'max_queue_time'), ('mrt', 'max_request_time')): try: v = request.args[r] except KeyError: continue v = int(v) c.__dict__[w] = v try: d = request.args['d'] except KeyError: d = '' c.process_arriving_data(d) c.new_request(request) def render_GET(self, request): return b'hippotat' def start_http(): resource = IphttpResource() site = twisted.web.server.Site(resource) for addrspec in cfg.get('server','addrs').split(): try: addr = ipaddress.IPv4Address(addrspec) endpointfactory = twisted.internet.endpoints.TCP4ServerEndpoint except AddressValueError: addr = ipaddress.IPv6Address(addrspec) endpointfactory = twisted.internet.endpoints.TCP6ServerEndpoint ep = endpointfactory(reactor, cfg.getint('server','port'), addr) crash_on_defer(ep.listen(site)) #---------- config and setup ---------- def process_cfg(): global network global host global relay global ipif_command network = ipnetwork(cfg.get('virtual','network')) if network.num_addresses < 3 + 2: raise ValueError('network needs at least 2^3 addresses') try: host = cfg.get('virtual','host') except NoOptionError: host = next(network.hosts()) try: relay = cfg.get('virtual','relay') except NoOptionError: for search in network.hosts(): if search == host: continue relay = search break for cs in cfg.sections(): if not (':' in cs or '.' in cs): continue ci = ipaddr(cs) if ci not in network: raise ValueError('client %s not in network' % ci) if ci in clients: raise ValueError('multiple client cfg sections for %s' % ci) clients[ci] = Client(ci, cs) global mtu mtu = cfg.get('virtual','mtu') iic_vars = { } for k in ('host','relay','mtu','network'): iic_vars[k] = globals()[k] ipif_command = cfg.get('server','ipif', vars=iic_vars) def startup(): global cfg op = OptionParser() op.add_option('-c', '--config', dest='configfile', default='/etc/hippottd/server.conf') global opts (opts, args) = op.parse_args() if len(args): op.error('no non-option arguments please') twisted.logger.globalLogPublisher.addObserver(crash_on_critical) cfg = ConfigParser() cfg.read_string(defcfg) cfg.read(opts.configfile) process_cfg() start_ipif(ipif_command, route) start_http() startup() reactor.run() print('CRASHED (end)', file=sys.stderr)