chiark / gitweb /
wip
[hippotat.git] / server
1 #!/usr/bin/python3
2
3 import signal
4 signal.signal(signal.SIGINT, signal.SIG_DFL)
5
6 import sys
7 import os
8
9 import twisted
10 import twisted.internet
11 import twisted.internet.endpoints
12 from twisted.internet import reactor
13 from twisted.web.server import NOT_DONE_YET
14 from twisted.logger import LogLevel
15
16 import ipaddress
17 from ipaddress import AddressValueError
18
19 #import twisted.web.server import Site
20 #from twisted.web.resource import Resource
21
22 from optparse import OptionParser
23 from configparser import ConfigParser
24 from configparser import NoOptionError
25
26 import collections
27
28 import syslog
29
30 clients = { }
31
32 def ipaddr(input):
33   try:
34     r = ipaddress.IPv4Address(input)
35   except AddressValueError:
36     r = ipaddress.IPv6Address(input)
37   return r
38
39 def ipnetwork(input):
40   try:
41     r = ipaddress.IPv4Network(input)
42   except NetworkValueError:
43     r = ipaddress.IPv6Network(input)
44   return r
45
46 defcfg = '''
47 [DEFAULT]
48 max_batch_down = 65536
49 max_queue_time = 10
50 max_request_time = 54
51
52 [virtual]
53 mtu = 1500
54 # network
55 # [host]
56 # [relay]
57
58 [server]
59 ipif = userv root ipif %(host)s,%(relay)s,%(mtu)s,slip %(network)s
60 addrs = 127.0.0.1 ::1
61 port = 8099
62
63 [limits]
64 max_batch_down = 262144
65 max_queue_time = 121
66 max_request_time = 121
67 '''
68
69 #---------- error handling ----------
70
71 def crash(err):
72   print('CRASH ', err, file=sys.stderr)
73   try: reactor.stop()
74   except twisted.internet.error.ReactorNotRunning: pass
75
76 def crash_on_defer(defer):
77   defer.addErrback(lambda err: crash(err))
78
79 def crash_on_critical(event):
80   if event.get('log_level') >= LogLevel.critical:
81     crash(twisted.logger.formatEvent(event))
82
83 #---------- "router" ----------
84
85 def route(packet, saddr, daddr):
86   print('TRACE ', saddr, daddr, packet)
87   try: client = clients[daddr]
88   except KeyError: dclient = None
89   if dclient is not None:
90     dclient.queue_outbound(packet)
91   elif saddr.is_link_local or daddr.is_link_local:
92     log_discard(packet, saddr, daddr, 'link-local')
93   elif daddr == host or daddr not in network:
94     print('TRACE INBOUND ', saddr, daddr, packet)
95     queue_inbound(packet)
96   elif daddr == relay:
97     log_discard(packet, saddr, daddr, 'relay')
98   else:
99     log_discard(packet, saddr, daddr, 'no client')
100
101 def log_discard(packet, saddr, daddr, why):
102   print('DROP ', saddr, daddr, why)
103 #  syslog.syslog(syslog.LOG_DEBUG,
104 #                'discarded packet %s -> %s (%s)' % (saddr, daddr, why))
105
106 #---------- ipif (slip subprocess) ----------
107
108 class IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol):
109   def __init__(self):
110     self._buffer = b''
111   def connectionMade(self): pass
112   def outReceived(self, data):
113     #print('RECV ', repr(data))
114     self._buffer += data
115     packets = slip_decode(self._buffer)
116     self._buffer = packets.pop()
117     for packet in packets:
118       if not len(packet): continue
119       (saddr, daddr) = packet_addrs(packet)
120       route(packet, saddr, daddr)
121   def processEnded(self, status):
122     status.raiseException()
123
124 def start_ipif():
125   global ipif
126   ipif = IpifProcessProtocol()
127   reactor.spawnProcess(ipif,
128                        '/bin/sh',['sh','-xc', ipif_command],
129                        childFDs={0:'w', 1:'r', 2:2})
130
131 def queue_inbound(packet):
132   ipif.transport.write(slip_delimiter)
133   ipif.transport.write(slip_encode(packet))
134   ipif.transport.write(slip_delimiter)
135
136 #---------- SLIP handling ----------
137
138 slip_end = b'\300'
139 slip_esc = b'\333'
140 slip_esc_end = b'\334'
141 slip_esc_esc = b'\335'
142 slip_delimiter = slip_end
143
144 def slip_encode(packet):
145   return (packet
146           .replace(slip_esc, slip_esc + slip_esc_esc)
147           .replace(slip_end, slip_esc + slip_esc_end))
148
149 def slip_decode(data):
150   print('DECODE ', repr(data))
151   out = []
152   for packet in data.split(slip_end):
153     pdata = b''
154     while True:
155       eix = packet.find(slip_esc)
156       if eix == -1:
157         pdata += packet
158         break
159       #print('ESC ', repr((pdata, packet, eix)))
160       pdata += packet[0 : eix]
161       ck = packet[eix+1]
162       #print('ESC... %o' % ck)
163       if   ck == slip_esc_esc[0]: pdata += slip_esc
164       elif ck == slip_esc_end[0]: pdata += slip_end
165       else: raise ValueError('invalid SLIP escape')
166       packet = packet[eix+2 : ]
167     out.append(pdata)
168   print('DECODED ', repr(out))
169   return out
170
171 #---------- packet parsing ----------
172
173 def packet_addrs(packet):
174   version = packet[0] >> 4
175   if version == 4:
176     addrlen = 4
177     saddroff = 3*4
178     factory = ipaddress.IPv4Address
179   elif version == 6:
180     addrlen = 16
181     saddroff = 2*4
182     factory = ipaddress.IPv6Address
183   else:
184     raise ValueError('unsupported IP version %d' % version)
185   saddr = factory(packet[ saddroff           : saddroff + addrlen   ])
186   daddr = factory(packet[ saddroff + addrlen : saddroff + addrlen*2 ])
187   return (saddr, daddr)
188
189 #---------- client ----------
190
191 class Client():
192   def __init__(self, ip, cs):
193     # instance data members
194     self._ip = ip
195     self._cs = cs
196     self.pw = cfg.get(cs, 'password')
197     self._rq = collections.deque() # requests
198     self._pq = collections.deque() # packets
199     # plus from config:
200     #  .max_batch_down
201     #  .max_queue_time
202     #  .max_request_time
203     for k in ('max_batch_down','max_queue_time','max_request_time'):
204       req = cfg.getint(cs, k)
205       limit = cfg.getint('limits',k)
206       self.__dict__[k] = min(req, limit)
207
208     def process_arriving_data(self, d):
209       for packet in slip_decode(d):
210         (saddr, daddr) = packet_addrs(packet)
211         if saddr != self._ip:
212           raise ValueError('wrong source address %s' % saddr)
213         route(packet, saddr, daddr)
214
215     def _req_cancel(self, request):
216       request.finish()
217
218     def _req_error(self, err, request):
219       self._req_cancel(request)
220
221     def queue_outbound(self, packet):
222       self._pq.append((time.monotonic(), packet))
223
224     def http_request(self, request):
225       request.setHeader('Content-Type','application/octet-stream')
226       reactor.callLater(self.max_request_time, self._req_cancel, request)
227       request.notifyFinish().addErrback(self._req_error, request)
228       self._rq.append(request)
229       self._check_outbound()
230
231     def _check_outbound(self):
232       while True:
233         try: request = self._rq[0]
234         except IndexError: request = None
235         if request and request.finished:
236           self._rq.popleft()
237           continue
238
239         # now request is an unfinished request, or None
240         try: (queuetime, packet) = self._pq[0]
241         except IndexError:
242           # no packets, oh well
243           break
244
245         age = time.monotonic() - queuetime
246         if age > self.max_queue_time:
247           self._pq.popleft()
248           continue
249
250         if request is None:
251           # no request
252           break
253
254         # request, and also some non-expired packets
255         while True:
256           try: (dummy, packet) = self._pq[0]
257           except IndexError: break
258
259           encoded = slip_encode(packet)
260           
261           if request.sentLength > 0:
262             if (request.sentLength + len(slip_delimiter)
263                 + len(encoded) > self.max_batch_down):
264               break
265             request.write(slip_delimiter)
266
267           request.write(encoded)
268           self._pq.popLeft()
269
270         assert(request.sentLength)
271         self._rq.popLeft()
272         request.finish()
273         # round again, looking for more to do
274
275 class IphttpResource(twisted.web.resource.Resource):
276   def render_POST(self, request):
277     # find client, update config, etc.
278     ci = ipaddr(request.args['i'])
279     c = clients[ci]
280     pw = request.args['pw']
281     if pw != c.pw: raise ValueError('bad password')
282
283     # update config
284     for r, w in (('mbd', 'max_batch_down'),
285                  ('mqt', 'max_queue_time'),
286                  ('mrt', 'max_request_time')):
287       try: v = request.args[r]
288       except KeyError: continue
289       v = int(v)
290       c.__dict__[w] = v
291
292     try: d = request.args['d']
293     except KeyError: d = ''
294
295     c.process_arriving_data(d)
296     c.new_request(request)
297
298   def render_GET(self, request):
299     return b'<html><body>hippotit</body></html>'
300
301 def start_http():
302   resource = IphttpResource()
303   site = twisted.web.server.Site(resource)
304   for addrspec in cfg.get('server','addrs').split():
305     try:
306       addr = ipaddress.IPv4Address(addrspec)
307       endpointfactory = twisted.internet.endpoints.TCP4ServerEndpoint
308     except AddressValueError:
309       addr = ipaddress.IPv6Address(addrspec)
310       endpointfactory = twisted.internet.endpoints.TCP6ServerEndpoint
311     ep = endpointfactory(reactor, cfg.getint('server','port'), addr)
312     crash_on_defer(ep.listen(site))
313
314 #---------- config and setup ----------
315         
316 def process_cfg():
317   global network
318   global host
319   global relay
320   global ipif_command
321
322   network = ipnetwork(cfg.get('virtual','network'))
323   if network.num_addresses < 3 + 2:
324     raise ValueError('network needs at least 2^3 addresses')
325
326   try:
327     host = cfg.get('virtual','host')
328   except NoOptionError:
329     host = next(network.hosts())
330
331   try:
332     relay = cfg.get('virtual','relay')
333   except NoOptionError:
334     for search in network.hosts():
335       if search == host: continue
336       relay = search
337       break
338
339   for cs in cfg.sections():
340     if not (':' in cs or '.' in cs): continue
341     ci = ipaddr(cs)
342     if ci not in network:
343       raise ValueError('client %s not in network' % ci)
344     if ci in clients:
345       raise ValueError('multiple client cfg sections for %s' % ci)
346     clients[ci] = Client(ci, cs)
347
348   global mtu
349   mtu = cfg.get('virtual','mtu')
350
351   iic_vars = { }
352   for k in ('host','relay','mtu','network'):
353     iic_vars[k] = globals()[k]
354
355   ipif_command = cfg.get('server','ipif', vars=iic_vars)
356
357 def startup():
358   global cfg
359
360   op = OptionParser()
361   op.add_option('-c', '--config', dest='configfile',
362                 default='/etc/hippottd/server.conf')
363   global opts
364   (opts, args) = op.parse_args()
365   if len(args): op.error('no non-option arguments please')
366
367   twisted.logger.globalLogPublisher.addObserver(crash_on_critical)
368
369   cfg = ConfigParser()
370   cfg.read_string(defcfg)
371   cfg.read(opts.configfile)
372   process_cfg()
373
374   start_ipif()
375   start_http()
376
377 startup()
378 reactor.run()
379 print('CRASHED (end)', file=sys.stderr)