server

   1 #!/usr/bin/python3
   2
   3 import twisted
   4
   5 #import twisted.web.server import Site
   6 #from twisted.web.resource import Resource
   7 from twisted.web.server import NOT_DONE_YET
   8 from twisted.internet import reactor
   9
  10 from optparse import OptionParser
  11 from configparser import ConfigParser
  12 from configparser import NoOptionError
  13 import ipaddress
  14
  15 import collections
  16
  17 import syslog
  18
  19 clients = { }
  20
  21 def ipaddress(input):
  22   try:
  23     r = ipaddress.IPv4Address(input)
  24   except AddressValueError:
  25     r = ipaddress.IPv6Address(input)
  26   return r
  27
  28 def ipnetwork(input):
  29   try:
  30     r = ipaddress.IPv4Network(input)
  31   except NetworkValueError:
  32     r = ipaddress.IPv6Network(input)
  33   return r
  34
  35 defcfg = '''
  36 [DEFAULT]
  37 max_batch_down = 65536
  38 max_queue_time = 10
  39 max_request_time = 54
  40
  41 [virtual]
  42 mtu = 1500
  43 # network
  44 # [host]
  45 # [relay]
  46
  47 [server]
  48 ipif = userv root ipif %(host),%(relay),%(mtu),slip %(network)
  49 addrs = 127.0.0.1 ::1
  50 port = 80
  51
  52 [limits]
  53 max_batch_down = 262144
  54 max_queue_time = 121
  55 max_request_time = 121
  56 '''
  57
  58 #---------- "router" ----------
  59
  60 def route(packet. daddr):
  61   try: client = clients[daddr]
  62   except KeyError: dclient = None
  63   if dclient is not None:
  64     dclient.queue_outbound(packet)
  65   else if daddr = host or daddr not in network:
  66     queue_inbound(packet)
  67   else if daddr = relay:
  68     log_discard(packet, saddr, daddr, 'relay')
  69   else:
  70     log_discard(packet, saddr, daddr, 'no client')
  71
  72 def log_discard(packet, saddr, daddr, why):
  73   syslog.syslog(syslog.LOG_DEBUG,
  74                 'discarded packet %s -> %s (%s)' % (saddr, daddr, why))
  75
  76 #---------- ipif (slip subprocess) ----------
  77
  78 class IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol):
  79   def __init__(self):
  80     self._buffer = b''
  81   def connectionMade(self): pass
  82   def outReceived(self, data):
  83     buffer += data
  84     packets = slip_decode(buffer)
  85     buffer = packets.pop()
  86     for packet in packets:
  87       (saddr, daddr) = packet_addrs(packet)
  88       route(packet, daddr)
  89   def processEnded(self, status):
  90     status.raiseException()
  91
  92 def start_ipif():
  93   global ipif
  94   ipif = IpifProcessProtocol()
  95   reactor.spawnProcess(ipif,
  96                        '/bin/sh',['-c', ipif_command],
  97                        childFDs={0:'w', 1:'r', 2:2})
  98
  99 def queue_inbound(packet):
 100   ipif.transport.write(slip_delimiter)
 101   ipif.transport.write(slip_encode(packet))
 102   ipif.transport.write(slip_delimiter)
 103
 104 #---------- client ----------
 105
 106 class Client():
 107   def __init__(self, ip, cs):
 108     # instance data members
 109     self._ip = ip
 110     self._cs = cs
 111     self.pw = cfg.get(cs, 'password')
 112     self._rq = collections.deque() # requests
 113     self._pq = collections.deque() # packets
 114     # plus from config:
 115     #  .max_batch_down
 116     #  .max_queue_time
 117     #  .max_request_time
 118     for k in ('max_batch_down','max_queue_time','max_request_time'):
 119       req = cfg.getint(cs, k)
 120       limit = cfg.getint('limits',k)
 121       self.__dict__[k] = min(req, limit)
 122
 123     def process_arriving_data(self, d):
 124       for packet in slip_decode(d):
 125         (saddr, daddr) = packet_addrs(packet)
 126         if saddr != self._ip:
 127           raise ValueError('wrong source address %s' % saddr)
 128         route(packet, daddr)
 129
 130     def _req_cancel(self, request):
 131       request.finish()
 132
 133     def _req_error(self, err, request):
 134       self._req_cancel(request)
 135
 136     def queue_outbound(self, packet):
 137       self._pq.append((time.monotonic(), packet))
 138
 139     def http_request(self, request):
 140       request.setHeader('Content-Type','application/octet-stream')
 141       reactor.callLater(self.max_request_time, self._req_cancel, request)
 142       request.notifyFinish().addErrback(self._req_error, request)
 143       self._rq.append(request)
 144       self._check_outbound()
 145
 146     def _check_outbound(self):
 147       while True:
 148         try: request = self._rq[0]
 149         except IndexError: request = None
 150         if request and request.finished:
 151           self._rq.popleft()
 152           continue
 153
 154         # now request is an unfinished request, or None
 155         try: (queuetime, packet) = self._pq[0]
 156         except: IndexError:
 157           # no packets, oh well
 158           break
 159
 160         age = time.monotonic() - queuetime
 161         if age > self.max_queue_time:
 162           self._pq.popleft()
 163           continue
 164
 165         if request is None:
 166           # no request
 167           break
 168
 169         # request, and also some non-expired packets
 170         while True:
 171           try: (dummy, packet) = self._pq[0]
 172           except IndexError: break
 173
 174           encoded = slip_encode(packet)
 175
 176           if request.sentLength > 0:
 177             if (request.sentLength + len(slip_delimiter)
 178                 + len(encoded) > self.max_batch_down):
 179               break
 180             request.write(slip_delimiter)
 181
 182           request.write(encoded)
 183           self._pq.popLeft()
 184
 185         assert(request.sentLength)
 186         self._rq.popLeft()
 187         request.finish()
 188         # round again, looking for more to do
 189
 190 class IphttpResource(twisted.web.resource.Resource):
 191   def render_POST(self, request):
 192     # find client, update config, etc.
 193     ci = ipaddress(request.args['i'])
 194     c = clients[ci]
 195     pw = request.args['pw']
 196     if pw != c.pw: raise ValueError('bad password')
 197
 198     # update config
 199     for r, w in (('mbd', 'max_batch_down'),
 200                  ('mqt', 'max_queue_time'),
 201                  ('mrt', 'max_request_time')):
 202       try: v = request.args[r]
 203       except KeyError: continue
 204       v = int(v)
 205       c.__dict__[w] = v
 206
 207     try: d = request.args['d']
 208     except KeyError: d = ''
 209
 210     c.process_arriving_data(d)
 211     c.new_request(request)
 212
 213 def start_http():
 214   resource = IphttpResource()
 215   sitefactory = twisted.web.server.Site(resource)
 216   for addrspec in cfg.get('server','addresses').split():
 217     try:
 218       addr = ipaddress.IPv4Address(addrspec)
 219       endpointfactory = twisted.internet.endpoints.TCP4ServerEndpoint
 220     except AddressValueError:
 221       addr = ipaddress.IPv6Address(addrspec)
 222       endpointfactory = twisted.internet.endpoints.TCP6ServerEndpoint
 223     ep = endpointfactory(reactor, cfg.getint('server','port'), addr)
 224     ep.listen(sitefactory)
 225
 226 #---------- config and setup ----------
 227
 228 def process_cfg():
 229   global network
 230   global host
 231   global relay
 232   global ipif_command
 233
 234   network = ipnetwork(cfg.get('virtual','network'))
 235   if network.num_addresses < 3 + 2:
 236     raise ValueError('network needs at least 2^3 addresses')
 237
 238   try:
 239     host = cfg.get('virtual','host')
 240   except NoOptionError:
 241     host = network.hosts().next()
 242
 243   try:
 244     relay = cfg.get('virtual','relay')
 245   except OptionError:
 246     for search in network.hosts():
 247       if search = host: continue
 248       relay = search
 249       break
 250
 251   for cs in cfg.sections():
 252     if not (':' in cs or '.' in cs): continue
 253     ci = ipaddress(cs)
 254     if ci not in network:
 255       raise ValueError('client %s not in network' % ci)
 256     if ci in clients:
 257       raise ValueError('multiple client cfg sections for %s' % ci)
 258     clients[ci] = Client(ci, cs)
 259
 260   iic_vars = { }
 261   for k in ('host','relay','mtu','network'):
 262     iic_vars[k] = globals()[k]
 263
 264   ipif_command = cfg.get('server','ipif', vars=iic_vars)
 265
 266 def startup():
 267   op = OptionParser()
 268   op.add_option('-c', '--config', dest='configfile',
 269                 default='/etc/hippottd/server.conf')
 270   global opts
 271   (opts, args) = op.parse_args()
 272   if len(args): op.error('no non-option arguments please')
 273
 274   cfg = ConfigParser()
 275   cfg.read_string(defcfg)
 276   cfg.read_file(opts['configfile'])
 277   process_cfg()
 278
 279   start_ipif()
 280   start_http()