X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?a=blobdiff_plain;f=server;h=f1b9ca4ff7b8de0a6db27e74e05b9f44680c35e3;hb=87a7c0c717c2bd9380e93e4ab25baebbc1e3aa6a;hp=a139e6e240d3d19e24847c54ff77c1698bc52459;hpb=5bae5ba3df0f2d33782972d94bb850bd8e22d68b;p=hippotat.git diff --git a/server b/server index a139e6e..f1b9ca4 100755 --- a/server +++ b/server @@ -1,42 +1,27 @@ #!/usr/bin/python3 -import twisted +from hippotat import * -import twisted.web.server import Site -from twisted.web.resource import Resource -from twisted.web.server import NOT_DONE_YET -from twisted.internet import reactor +import sys +import os -from optparse import OptionParser -from configparser import ConfigParser -from configparser import NoOptionError -import ipaddress +import twisted.internet +import twisted.internet.endpoints +from twisted.web.server import NOT_DONE_YET -import collections +#import twisted.web.server import Site +#from twisted.web.resource import Resource import syslog clients = { } -def ipaddress(input): - try: - r = ipaddress.IPv4Address(input) - except AddressValueError: - r = ipaddress.IPv6Address(input) - return r - -def ipnetwork(input): - try: - r = ipaddress.IPv4Network(input) - except NetworkValueError: - r = ipaddress.IPv6Network(input) - return r - defcfg = ''' [DEFAULT] max_batch_down = 65536 max_queue_time = 10 max_request_time = 54 +target_requests_outstanding = 3 [virtual] mtu = 1500 @@ -45,47 +30,42 @@ mtu = 1500 # [relay] [server] -ipif = userv root ipif %(host),%(relay),%(mtu),slip %(network) +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 ''' -class IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol): - def __init__(self): - self._buffer = b'' - def connectionMade(self): pass - def outReceived(self, data): - buffer += data - packets = slip_decode(buffer) - buffer = packets.pop() - for packet in packets: - (saddr, daddr) = packet_addrs(packet) - route(packet, daddr) - -def start_ipif(): - reactor.spawnProcess(IpifProcessProtocol(), - '/bin/sh',['-c', ipif_command], - childFDs={0:'w', 1:'r', 2:2}) - -def log_discard(packet, saddr, daddr, why): - syslog.syslog(syslog.LOG_DEBUG, - 'discarded packet %s -> %s (%s)' % (saddr, daddr, why)) +#---------- "router" ---------- -def route(packet. daddr): +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) - else if daddr = host or daddr not in network: + elif saddr.is_link_local or daddr.is_link_local: + log_discard(packet, saddr, daddr, 'link-local') + elif daddr == c.host or daddr not in c.network: + print('TRACE INBOUND ', saddr, daddr, packet) queue_inbound(packet) - else if daddr = relay: + 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 @@ -93,22 +73,25 @@ class Client(): self._cs = cs self.pw = cfg.get(cs, 'password') self._rq = collections.deque() # requests - self._pq = collections.deque() # packets + # self._pq = PacketQueue(...) # plus from config: # .max_batch_down # .max_queue_time # .max_request_time - for k in ('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): + for packet in slip.decode(d): (saddr, daddr) = packet_addrs(packet) if saddr != self._ip: raise ValueError('wrong source address %s' % saddr) - route(packet, daddr) + route(packet, saddr, daddr) def _req_cancel(self, request): request.finish() @@ -117,7 +100,7 @@ class Client(): self._req_cancel(request) def queue_outbound(self, packet): - self._pq.append((time.monotonic(), packet)) + self._pq.append(packet) def http_request(self, request): request.setHeader('Content-Type','application/octet-stream') @@ -134,15 +117,8 @@ class Client(): self._rq.popleft() continue - # now request is an unfinished request, or None - try: (queuetime, packet) = self._pq[0] - except: IndexError: + if not self._pq.nonempty(): # no packets, oh well - break - - age = time.monotonic() - queuetime - if age > self.max_queue_time: - self._pq.popleft() continue if request is None: @@ -151,16 +127,16 @@ class Client(): # request, and also some non-expired packets while True: - try: (dummy, packet) = self._pq[0] - except IndexError: break + packet = self.pq.popleft() + if packet is None: break - encoded = slip_encode(packet) + encoded = slip.encode(packet) if request.sentLength > 0: - if (request.sentLength + len(slip_delimiter) + if (request.sentLength + len(slip.delimiter) + len(encoded) > self.max_batch_down): break - request.write(slip_delimiter) + request.write(slip.delimiter) request.write(encoded) self._pq.popLeft() @@ -170,48 +146,15 @@ class Client(): request.finish() # round again, looking for more to do -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 = network.hosts().next() - - try: - relay = cfg.get('virtual','relay') - except OptionError: - 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 = ipaddress(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) - - iic_vars = { } - for k in ('host','relay','mtu','network'): - iic_vars[k] = globals()[k] - - ipif_command = cfg.get('server','ipif', vars=iic_vars) + while len(self._rq) > self.target_requests_outstanding: + request = self._rq.popleft() + request.finish() -class FormPage(Resource): +class IphttpResource(twisted.web.resource.Resource): + isLeaf = True def render_POST(self, request): # find client, update config, etc. - ci = ipaddress(request.args['i']) + ci = ipaddr(request.args['i']) c = clients[ci] pw = request.args['pw'] if pw != c.pw: raise ValueError('bad password') @@ -219,7 +162,8 @@ class FormPage(Resource): # update config for r, w in (('mbd', 'max_batch_down'), ('mqt', 'max_queue_time'), - ('mrt', 'max_request_time')): + ('mrt', 'max_request_time'), + ('tro', 'target_requests_outstanding')): try: v = request.args[r] except KeyError: continue v = int(v) @@ -231,18 +175,63 @@ class FormPage(Resource): c.process_arriving_data(d) c.new_request(request) -def startup(): - 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') - - cfg = ConfigParser() - cfg.read_string(defcfg) - cfg.read_file(opts['configfile']) - process_cfg() - - start_ipif() - start_http() + 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(): + process_cfg_common_always() + + c.network = ipnetwork(cfg.get('virtual','network')) + if c.network.num_addresses < 3 + 2: + raise ValueError('network needs at least 2^3 addresses') + + try: + c.host = cfg.get('virtual','host') + except NoOptionError: + c.host = next(c.network.hosts()) + + try: + c.relay = cfg.get('virtual','relay') + except NoOptionError: + for search in c.network.hosts(): + if search == c.host: continue + c.relay = search + break + + for cs in cfg.sections(): + if not (':' in cs or '.' in cs): continue + ci = ipaddr(cs) + if ci not in c.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) + + print(repr(c)) + + iic_vars = { } + for k in ('host','relay','mtu','network'): + iic_vars[k] = getattr(c,k) + + c.ipif_command = cfg.get('server','ipif', vars=iic_vars) + +common_startup(defcfg) +process_cfg() +start_ipif(c.ipif_command, route) +start_http() +common_run()