server

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