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