chiark / gitweb /
wip debug
[hippotat.git] / hippotat / __init__.py
1 # -*- python -*-
2
3 import signal
4 signal.signal(signal.SIGINT, signal.SIG_DFL)
5
6 import sys
7
8 from zope.interface import implementer
9
10 import twisted
11 from twisted.internet import reactor
12 import twisted.internet.endpoints
13 import twisted.logger
14 from twisted.logger import LogLevel
15 import twisted.python.constants
16 from twisted.python.constants import NamedConstant
17
18 import ipaddress
19 from ipaddress import AddressValueError
20
21 from optparse import OptionParser
22 from configparser import ConfigParser
23 from configparser import NoOptionError
24
25 from functools import partial
26
27 import collections
28 import time
29 import codecs
30 import traceback
31
32 import re as regexp
33
34 import hippotat.slip as slip
35
36 class DBG(twisted.python.constants.Names):
37   INIT = NamedConstant()
38   ROUTE = NamedConstant()
39   DROP = NamedConstant()
40   FLOW = NamedConstant()
41   HTTP = NamedConstant()
42   TWISTED = NamedConstant()
43   QUEUE = NamedConstant()
44   HTTP_CTRL = NamedConstant()
45   QUEUE_CTRL = NamedConstant()
46   HTTP_FULL = NamedConstant()
47   CTRL_DUMP = NamedConstant()
48   SLIP_FULL = NamedConstant()
49   DATA_COMPLETE = NamedConstant()
50
51 _hex_codec = codecs.getencoder('hex_codec')
52
53 #---------- logging ----------
54
55 org_stderr = sys.stderr
56
57 log = twisted.logger.Logger()
58
59 debug_set = set()
60 debug_def_detail = DBG.HTTP
61
62 def log_debug(dflag, msg, idof=None, d=None):
63   if dflag not in debug_set: return
64   #print('---------------->',repr((dflag, msg, idof, d)), file=sys.stderr)
65   if idof is not None:
66     msg = '[%#x] %s' % (id(idof), msg)
67   if d is not None:
68     trunc = ''
69     if not DBG.DATA_COMPLETE in debug_set:
70       if len(d) > 64:
71         d = d[0:64]
72         trunc = '...'
73     d = _hex_codec(d)[0].decode('ascii')
74     msg += ' ' + d + trunc
75   log.info('{dflag} {msgcore}', dflag=dflag, msgcore=msg)
76
77 @implementer(twisted.logger.ILogFilterPredicate)
78 class LogNotBoringTwisted:
79   def __call__(self, event):
80     yes = twisted.logger.PredicateResult.yes
81     no  = twisted.logger.PredicateResult.no
82     try:
83       if event.get('log_level') != LogLevel.info:
84         return yes
85       dflag = event.get('dflag')
86       if dflag                         in debug_set: return yes
87       if dflag is None and DBG.TWISTED in debug_set: return yes
88       return no
89     except Exception:
90       print(traceback.format_exc(), file=org_stderr)
91       return yes
92
93 #---------- default config ----------
94
95 defcfg = '''
96 [DEFAULT]
97 #[<client>] overrides
98 max_batch_down = 65536           # used by server, subject to [limits]
99 max_queue_time = 10              # used by server, subject to [limits]
100 target_requests_outstanding = 3  # must match; subject to [limits] on server
101 http_timeout = 30                # used by both } must be
102 http_timeout_grace = 5           # used by both }  compatible
103 max_requests_outstanding = 4     # used by client
104 max_batch_up = 4000              # used by client
105 http_retry = 5                   # used by client
106
107 #[server] or [<client>] overrides
108 ipif = userv root ipif %(local)s,%(peer)s,%(mtu)s,slip %(rnets)s
109 # extra interpolations:  %(local)s        %(peer)s          %(rnet)s
110 #  obtained   on server  [virtual]server  [virtual]relay    [virtual]network
111 #      from   on client  <client>         [virtual]server   [virtual]routes
112
113 [virtual]
114 mtu = 1500
115 routes = ''
116 # network = <prefix>/<len>  # mandatory for server
117 # server  = <ipaddr>   # used by both, default is computed from `network'
118 # relay   = <ipaddr>   # used by server, default from `network' and `server'
119 #  default server is first host in network
120 #  default relay is first host which is not server
121
122 [server]
123 # addrs = 127.0.0.1 ::1    # mandatory for server
124 port = 80                  # used by server
125 # url              # used by client; default from first `addrs' and `port'
126
127 # [<client-ip4-or-ipv6-address>]
128 # password = <password>    # used by both, must match
129
130 [limits]
131 max_batch_down = 262144           # used by server
132 max_queue_time = 121              # used by server
133 http_timeout = 121                # used by server
134 target_requests_outstanding = 10  # used by server
135 '''
136
137 # these need to be defined here so that they can be imported by import *
138 cfg = ConfigParser()
139 optparser = OptionParser()
140
141 _mimetrans = bytes.maketrans(b'-'+slip.esc, slip.esc+b'-')
142 def mime_translate(s):
143   # SLIP-encoded packets cannot contain ESC ESC.
144   # Swap `-' and ESC.  The result cannot contain `--'
145   return s.translate(_mimetrans)
146
147 class ConfigResults:
148   def __init__(self, d = { }):
149     self.__dict__ = d
150   def __repr__(self):
151     return 'ConfigResults('+repr(self.__dict__)+')'
152
153 c = ConfigResults()
154
155 def log_discard(packet, iface, saddr, daddr, why):
156   log_debug(DBG.DROP,
157             'discarded packet [%s] %s -> %s: %s' % (iface, saddr, daddr, why),
158             d=packet)
159
160 #---------- packet parsing ----------
161
162 def packet_addrs(packet):
163   version = packet[0] >> 4
164   if version == 4:
165     addrlen = 4
166     saddroff = 3*4
167     factory = ipaddress.IPv4Address
168   elif version == 6:
169     addrlen = 16
170     saddroff = 2*4
171     factory = ipaddress.IPv6Address
172   else:
173     raise ValueError('unsupported IP version %d' % version)
174   saddr = factory(packet[ saddroff           : saddroff + addrlen   ])
175   daddr = factory(packet[ saddroff + addrlen : saddroff + addrlen*2 ])
176   return (saddr, daddr)
177
178 #---------- address handling ----------
179
180 def ipaddr(input):
181   try:
182     r = ipaddress.IPv4Address(input)
183   except AddressValueError:
184     r = ipaddress.IPv6Address(input)
185   return r
186
187 def ipnetwork(input):
188   try:
189     r = ipaddress.IPv4Network(input)
190   except NetworkValueError:
191     r = ipaddress.IPv6Network(input)
192   return r
193
194 #---------- ipif (SLIP) subprocess ----------
195
196 class SlipStreamDecoder():
197   def __init__(self, desc, on_packet):
198     self._buffer = b''
199     self._on_packet = on_packet
200     self._desc = desc
201     self._log('__init__')
202
203   def _log(self, msg, **kwargs):
204     log_debug(DBG.SLIP_FULL, 'slip %s: %s' % (self._desc, msg), **kwargs)
205
206   def inputdata(self, data):
207     self._log('inputdata', d=data)
208     packets = slip.decode(data)
209     packets[0] = self._buffer + packets[0]
210     self._buffer = packets.pop()
211     for packet in packets:
212       self._maybe_packet(packet)
213     self._log('bufremain', d=self._buffer)
214
215   def _maybe_packet(self, packet):
216     self._log('maybepacket', d=packet)
217     if len(packet):
218       self._on_packet(packet)
219
220   def flush(self):
221     self._log('flush')
222     self._maybe_packet(self._buffer)
223     self._buffer = b''
224
225 class _IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol):
226   def __init__(self, router):
227     self._router = router
228     self._decoder = SlipStreamDecoder('ipif', self.slip_on_packet)
229   def connectionMade(self): pass
230   def outReceived(self, data):
231     self._decoder.inputdata(data)
232   def slip_on_packet(self, packet):
233     (saddr, daddr) = packet_addrs(packet)
234     if saddr.is_link_local or daddr.is_link_local:
235       log_discard(packet, 'ipif', saddr, daddr, 'link-local')
236       return
237     self._router(packet, saddr, daddr)
238   def processEnded(self, status):
239     status.raiseException()
240
241 def start_ipif(command, router):
242   global ipif
243   ipif = _IpifProcessProtocol(router)
244   reactor.spawnProcess(ipif,
245                        '/bin/sh',['sh','-xc', command],
246                        childFDs={0:'w', 1:'r', 2:2},
247                        env=None)
248
249 def queue_inbound(packet):
250   log_debug(DBG.FLOW, "queue_inbound", d=packet)
251   ipif.transport.write(slip.delimiter)
252   ipif.transport.write(slip.encode(packet))
253   ipif.transport.write(slip.delimiter)
254
255 #---------- packet queue ----------
256
257 class PacketQueue():
258   def __init__(self, desc, max_queue_time):
259     self._desc = desc
260     assert(desc + '')
261     self._max_queue_time = max_queue_time
262     self._pq = collections.deque() # packets
263
264   def _log(self, dflag, msg, **kwargs):
265     log_debug(dflag, self._desc+' pq: '+msg, **kwargs)
266
267   def append(self, packet):
268     self._log(DBG.QUEUE, 'append', d=packet)
269     self._pq.append((time.monotonic(), packet))
270
271   def nonempty(self):
272     self._log(DBG.QUEUE, 'nonempty ?')
273     while True:
274       try: (queuetime, packet) = self._pq[0]
275       except IndexError:
276         self._log(DBG.QUEUE, 'nonempty ? empty.')
277         return False
278
279       age = time.monotonic() - queuetime
280       if age > self._max_queue_time:
281         # strip old packets off the front
282         self._log(DBG.QUEUE, 'dropping (old)', d=packet)
283         self._pq.popleft()
284         continue
285
286       self._log(DBG.QUEUE, 'nonempty ? nonempty.')
287       return True
288
289   def process(self, sizequery, moredata, max_batch):
290     # sizequery() should return size of batch so far
291     # moredata(s) should add s to batch
292     self._log(DBG.QUEUE, 'process...')
293     while True:
294       try: (dummy, packet) = self._pq[0]
295       except IndexError:
296         self._log(DBG.QUEUE, 'process... empty')
297         break
298
299       self._log(DBG.QUEUE_CTRL, 'process... packet', d=packet)
300
301       encoded = slip.encode(packet)
302       sofar = sizequery()  
303
304       self._log(DBG.QUEUE_CTRL,
305                 'process... (sofar=%d, max=%d) encoded' % (sofar, max_batch),
306                 d=encoded)
307
308       if sofar > 0:
309         if sofar + len(slip.delimiter) + len(encoded) > max_batch:
310           self._log(DBG.QUEUE_CTRL, 'process... overflow')
311           break
312         moredata(slip.delimiter)
313
314       moredata(encoded)
315       self._pq.popleft()
316
317 #---------- error handling ----------
318
319 _crashing = False
320
321 def crash(err):
322   global _crashing
323   _crashing = True
324   print('========== CRASH ==========', err,
325         '===========================', file=sys.stderr)
326   try: reactor.stop()
327   except twisted.internet.error.ReactorNotRunning: pass
328
329 def crash_on_defer(defer):
330   defer.addErrback(lambda err: crash(err))
331
332 def crash_on_critical(event):
333   if event.get('log_level') >= LogLevel.critical:
334     crash(twisted.logger.formatEvent(event))
335
336 #---------- config processing ----------
337
338 def process_cfg_common_always():
339   global mtu
340   c.mtu = cfg.get('virtual','mtu')
341
342 def process_cfg_ipif(section, varmap):
343   for d, s in varmap:
344     try: v = getattr(c, s)
345     except AttributeError: continue
346     setattr(c, d, v)
347
348   #print(repr(c))
349
350   c.ipif_command = cfg.get(section,'ipif', vars=c.__dict__)
351
352 def process_cfg_network():
353   c.network = ipnetwork(cfg.get('virtual','network'))
354   if c.network.num_addresses < 3 + 2:
355     raise ValueError('network needs at least 2^3 addresses')
356
357 def process_cfg_server():
358   try:
359     c.server = cfg.get('virtual','server')
360   except NoOptionError:
361     process_cfg_network()
362     c.server = next(c.network.hosts())
363
364 class ServerAddr():
365   def __init__(self, port, addrspec):
366     self.port = port
367     # also self.addr
368     try:
369       self.addr = ipaddress.IPv4Address(addrspec)
370       self._endpointfactory = twisted.internet.endpoints.TCP4ServerEndpoint
371       self._inurl = b'%s'
372     except AddressValueError:
373       self.addr = ipaddress.IPv6Address(addrspec)
374       self._endpointfactory = twisted.internet.endpoints.TCP6ServerEndpoint
375       self._inurl = b'[%s]'
376   def make_endpoint(self):
377     return self._endpointfactory(reactor, self.port, self.addr)
378   def url(self):
379     url = b'http://' + (self._inurl % str(self.addr).encode('ascii'))
380     if self.port != 80: url += b':%d' % self.port
381     url += b'/'
382     return url
383     
384 def process_cfg_saddrs():
385   try: port = cfg.getint('server','port')
386   except NoOptionError: port = 80
387
388   c.saddrs = [ ]
389   for addrspec in cfg.get('server','addrs').split():
390     sa = ServerAddr(port, addrspec)
391     c.saddrs.append(sa)
392
393 def process_cfg_clients(constructor):
394   c.clients = [ ]
395   for cs in cfg.sections():
396     if not (':' in cs or '.' in cs): continue
397     ci = ipaddr(cs)
398     pw = cfg.get(cs, 'password')
399     pw = pw.encode('utf-8')
400     constructor(ci,cs,pw)
401
402 #---------- startup ----------
403
404 def common_startup():
405   optparser.add_option('-c', '--config', dest='configfile',
406                        default='/etc/hippotat/config')
407
408   def dfs_less_detailed(dl):
409     return [df for df in DBG.iterconstants() if df <= dl]
410
411   def ds_default(od,os,dl,op):
412     global debug_set
413     debug_set = set(dfs_less_detailed(debug_def_detail))
414
415   def ds_select(od,os, spec, op):
416     last_df = next(DBG.iterconstants())
417     mutator = debug_set.add
418
419     for it in spec.split(','):
420
421       if not len(it):
422         for df in dfs_less_detailed(last_df):
423           mutator(df)
424         continue
425
426       if it.startswith('-'):
427         mutator = debug_set.discard
428         it = it[1:]
429       else:
430         mutator = debug_set.add
431         
432       try:
433         df = DBG.lookupByName(it)
434       except ValueError:
435         optparser.error('unknown debug flag %s in --debug-select' % it)
436       mutator(df)
437       last_df = df
438
439   optparser.add_option('-D', '--debug',
440                        nargs=0,
441                        action='callback',
442                        help='enable default debug (to stdout)',
443                        callback= ds_default)
444
445   optparser.add_option('--debug-select',
446                        nargs=1,
447                        type='string',
448                        metavar='[-]DFLAG[,],...',
449                        help=
450 '''enable (or with -, disable) each specified DFLAG;
451 empty entry means do the same for all DFLAGS "more interesting"
452 than the last (or, all DFLAGs)''',
453                        action='callback',
454                        callback= ds_select)
455
456   (opts, args) = optparser.parse_args()
457   if len(args): optparser.error('no non-option arguments please')
458
459   print(repr(debug_set), file=sys.stderr)
460
461   re = regexp.compile('#.*')
462   cfg.read_string(re.sub('', defcfg))
463   cfg.read(opts.configfile)
464
465   log_formatter = twisted.logger.formatEventAsClassicLogText
466   stdout_obs = twisted.logger.FileLogObserver(sys.stdout, log_formatter)
467   stderr_obs = twisted.logger.FileLogObserver(sys.stderr, log_formatter)
468   pred = twisted.logger.LogLevelFilterPredicate(LogLevel.error)
469   stdsomething_obs = twisted.logger.FilteringLogObserver(
470     stderr_obs, [pred], stdout_obs
471   )
472   log_observer = twisted.logger.FilteringLogObserver(
473     stdsomething_obs, [LogNotBoringTwisted()]
474   )
475   #log_observer = stdsomething_obs
476   twisted.logger.globalLogBeginner.beginLoggingTo(
477     [ log_observer, crash_on_critical ]
478     )
479
480 def common_run():
481   log_debug(DBG.INIT, 'entering reactor')
482   if not _crashing: reactor.run()
483   print('CRASHED (end)', file=sys.stderr)