mdw@git.distorted.org.uk Git - hippotat/blob - server

   1 #!/usr/bin/python3
   2
   3 from hippotat import *
   4
   5 import sys
   6 import os
   7
   8 import twisted.internet
   9 import twisted.internet.endpoints
  10 from twisted.web.server import NOT_DONE_YET
  11
  12 #import twisted.web.server import Site
  13 #from twisted.web.resource import Resource
  14
  15 import syslog
  16
  17 clients = { }
  18
  19 defcfg = '''
  20 [DEFAULT]
  21 max_batch_down = 65536
  22 max_queue_time = 10
  23 max_request_time = 54
  24 target_requests_outstanding = 3
  25 ipif = userv root ipif %(local)s,%(peer)s,%(mtu)s,slip %(rnets)s
  26
  27 [virtual]
  28 mtu = 1500
  29 # network
  30 # [server]
  31 # [relay]
  32
  33 [server]
  34 addrs = 127.0.0.1 ::1
  35 port = 8099
  36
  37 [limits]
  38 max_batch_down = 262144
  39 max_queue_time = 121
  40 max_request_time = 121
  41 target_requests_outstanding = 10
  42 '''
  43
  44 #---------- "router" ----------
  45
  46 def route(packet, saddr, daddr):
  47   print('TRACE ', saddr, daddr, packet)
  48   try: client = clients[daddr]
  49   except KeyError: dclient = None
  50   if dclient is not None:
  51     dclient.queue_outbound(packet)
  52   elif saddr.is_link_local or daddr.is_link_local:
  53     log_discard(packet, saddr, daddr, 'link-local')
  54   elif daddr == c.server or daddr not in c.network:
  55     print('TRACE INBOUND ', saddr, daddr, packet)
  56     queue_inbound(packet)
  57   elif daddr == relay:
  58     log_discard(packet, saddr, daddr, 'relay')
  59   else:
  60     log_discard(packet, saddr, daddr, 'no client')
  61
  62 def log_discard(packet, saddr, daddr, why):
  63   print('DROP ', saddr, daddr, why)
  64 #  syslog.syslog(syslog.LOG_DEBUG,
  65 #                'discarded packet %s -> %s (%s)' % (saddr, daddr, why))
  66
  67 #---------- client ----------
  68
  69 class Client():
  70   def __init__(self, ip, cs, pw):
  71     # instance data members
  72     self._ip = ip
  73     self._cs = cs
  74     self.pw = pw
  75     self._rq = collections.deque() # requests
  76     # self._pq = PacketQueue(...)
  77     # plus from config:
  78     #  .max_batch_down
  79     #  .max_queue_time
  80     #  .max_request_time
  81     #  .target_requests_outstanding
  82
  83     if ip not in c.network:
  84       raise ValueError('client %s not in network' % ip)
  85
  86     for k in ('max_batch_down','max_queue_time','max_request_time',
  87               'target_requests_outstanding'):
  88       req = cfg.getint(cs, k)
  89       limit = cfg.getint('limits',k)
  90       self.__dict__[k] = min(req, limit)
  91     self._pq = PacketQueue(self.max_queue_time)
  92
  93     if ip in clients:
  94       raise ValueError('multiple client cfg sections for %s' % ip)
  95     clients[ip] = self
  96
  97   def process_arriving_data(self, d):
  98     for packet in slip.decode(d):
  99       (saddr, daddr) = packet_addrs(packet)
 100       if saddr != self._ip:
 101         raise ValueError('wrong source address %s' % saddr)
 102       route(packet, saddr, daddr)
 103
 104   def _req_cancel(self, request):
 105     request.finish()
 106
 107   def _req_error(self, err, request):
 108     self._req_cancel(request)
 109
 110   def queue_outbound(self, packet):
 111     self._pq.append(packet)
 112
 113   def http_request(self, request):
 114     request.setHeader('Content-Type','application/octet-stream')
 115     reactor.callLater(self.max_request_time, self._req_cancel, request)
 116     request.notifyFinish().addErrback(self._req_error, request)
 117     self._rq.append(request)
 118     self._check_outbound()
 119
 120   def _check_outbound(self):
 121     while True:
 122       try: request = self._rq[0]
 123       except IndexError: request = None
 124       if request and request.finished:
 125         self._rq.popleft()
 126         continue
 127
 128       if not self._pq.nonempty():
 129         # no packets, oh well
 130         continue
 131
 132       if request is None:
 133         # no request
 134         break
 135
 136       # request, and also some non-expired packets
 137       while True:
 138         packet = self.pq.popleft()
 139         if packet is None: break
 140
 141         encoded = slip.encode(packet)
 142
 143         if request.sentLength > 0:
 144           if (request.sentLength + len(slip.delimiter)
 145               + len(encoded) > self.max_batch_down):
 146             break
 147           request.write(slip.delimiter)
 148
 149         request.write(encoded)
 150         self._pq.popLeft()
 151
 152       assert(request.sentLength)
 153       self._rq.popLeft()
 154       request.finish()
 155       # round again, looking for more to do
 156
 157     while len(self._rq) > self.target_requests_outstanding:
 158       request = self._rq.popleft()
 159       request.finish()
 160
 161 class IphttpResource(twisted.web.resource.Resource):
 162   isLeaf = True
 163   def render_POST(self, request):
 164     # find client, update config, etc.
 165     ci = ipaddr(request.args['i'])
 166     c = clients[ci]
 167     pw = request.args['pw']
 168     if pw != c.pw: raise ValueError('bad password')
 169
 170     # update config
 171     for r, w in (('mbd', 'max_batch_down'),
 172                  ('mqt', 'max_queue_time'),
 173                  ('mrt', 'max_request_time'),
 174                  ('tro', 'target_requests_outstanding')):
 175       try: v = request.args[r]
 176       except KeyError: continue
 177       v = int(v)
 178       c.__dict__[w] = v
 179
 180     try: d = request.args['d']
 181     except KeyError: d = ''
 182
 183     c.process_arriving_data(d)
 184     c.new_request(request)
 185
 186   def render_GET(self, request):
 187     return b'<html><body>hippotat</body></html>'
 188
 189 def start_http():
 190   resource = IphttpResource()
 191   site = twisted.web.server.Site(resource)
 192   for sa in c.saddrs:
 193     ep = sa.make_endpoint()
 194     crash_on_defer(ep.listen(site))
 195
 196 #---------- config and setup ----------
 197
 198 def process_cfg():
 199   process_cfg_common_always()
 200   process_cfg_server()
 201   process_cfg_network()
 202
 203   try:
 204     c.relay = cfg.get('virtual','relay')
 205   except NoOptionError:
 206     for search in c.network.hosts():
 207       if search == c.server: continue
 208       c.relay = search
 209       break
 210
 211   process_cfg_saddrs()
 212   process_cfg_clients(Client)
 213
 214   process_cfg_ipif('server',
 215                    (('local','server'),
 216                     ('peer','relay'),
 217                     ('rnets','network')))
 218
 219 common_startup(defcfg)
 220 process_cfg()
 221 start_ipif(c.ipif_command, route)
 222 start_http()
 223 common_run()