chiark / gitweb /
6da5be4d7acfe5596d3141e1d64124f378a221ad
[hippotat.git] / 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()