X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=hippotat.git;a=blobdiff_plain;f=server;h=db13f25954d01ff4ad24fe6fbd6bb4305e0d2995;hp=fcf7a81f5ed87e3dc13d05d5192adb08c92a0857;hb=650a3251620a44d22f30a4242cea6fff88ccfba5;hpb=3fba9787545d5da68dc247932817899f690edbf0 diff --git a/server b/server index fcf7a81..db13f25 100755 --- a/server +++ b/server @@ -1,48 +1,283 @@ -#!/usr/bin/python2 +#!/usr/bin/python3 -from twisted.web.server import Site -from twisted.web.resource import Resource +import signal +signal.signal(signal.SIGINT, signal.SIG_DFL) + +import sys +import os + +import twisted.internet +import twisted.internet.endpoints from twisted.web.server import NOT_DONE_YET -from twisted.internet import reactor +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 ConfigParser +import collections -import cgi +import syslog + +from hippotat import * clients = { } -def ipaddress(input): - try: - r = IPv4Address(input) - except AddressValueError: - r = IPv6Address(input) - return r +defcfg = ''' +[DEFAULT] +max_batch_down = 65536 +max_queue_time = 10 +max_request_time = 54 +target_requests_outstanding = 3 -def ipnetwork(input): - try: - r = IPv4Network(input) - except NetworkValueError: - r = IPv6Network(input) - return r +[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 +target_requests_outstanding = 10 +''' + +#---------- 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 = PacketQueue(...) + # plus from config: + # .max_batch_down + # .max_queue_time + # .max_request_time + # .target_requests_outstanding + for k in ('max_batch_down','max_queue_time','max_request_time', + 'target_requests_outstanding'): + req = cfg.getint(cs, k) + limit = cfg.getint('limits',k) + self.__dict__[k] = min(req, limit) + self._pq = PacketQueue(self.max_queue_time) + + 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(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 + + if not self._pq.nonempty(): + # no packets, oh well + continue + if request is None: + # no request + break + + # request, and also some non-expired packets + while True: + packet = self.pq.popleft() + if packet is None: 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 + + while len(self._rq) > self.target_requests_outstanding: + request = self._rq.popleft() + request.finish() + +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'), + ('tro', 'target_requests_outstanding')): + 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 ourself + 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') - network = ipnetwork(cfg.get('virtual','network') try: - ourself = cfg.get('virtual','server') - except ConfigParser.NoOptionError: - ourself = network.hosts().next() + 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 -class Client(): - def __init__(ip): - # instance data members - # ._ip - self._ip = IPv4Address(ip) - clients + 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) -class FormPage(Resource): - def render_POST(self, request): + 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)