reorg SlipStreamDecoder again

[hippotat] / hippotat / __init__.py

# -*- python -*-

import signal
signal.signal(signal.SIGINT, signal.SIG_DFL)

import sys

import twisted
from twisted.internet import reactor
from twisted.logger import LogLevel
import twisted.internet.endpoints

import ipaddress
from ipaddress import AddressValueError

from optparse import OptionParser
from configparser import ConfigParser
from configparser import NoOptionError

import collections

import re as regexp

import hippotat.slip as slip

defcfg = '''
[DEFAULT]
#[<client>] overrides
max_batch_down = 65536           # used by server, subject to [limits]
max_queue_time = 10              # used by server, subject to [limits]
max_request_time = 54            # used by server, subject to [limits]
target_requests_outstanding = 3  # must match; subject to [limits] on server
max_requests_outstanding = 4     # used by client
max_batch_up = 4000              # used by client
http_timeout = 30                # used by client

#[server] or [<client>] overrides
ipif = userv root ipif %(local)s,%(peer)s,%(mtu)s,slip %(rnets)s
# extra interpolations:  %(local)s        %(peer)s          %(rnet)s
#  obtained   on server  [virtual]server  [virtual]relay    [virtual]network
#      from   on client  <client>         [virtual]server   [virtual]routes

[virtual]
mtu = 1500
routes = ''
# network = <prefix>/<len>  # mandatory for server
# server  = <ipaddr>   # used by both, default is computed from `network'
# relay   = <ipaddr>   # used by server, default from `network' and `server'
#  default server is first host in network
#  default relay is first host which is not server

[server]
# addrs = 127.0.0.1 ::1    # mandatory for server
port = 80                  # used by server
# url              # used by client; default from first `addrs' and `port'

# [<client-ip4-or-ipv6-address>]
# password = <password>    # used by both, must match

[limits]
max_batch_down = 262144           # used by server
max_queue_time = 121              # used by server
max_request_time = 121            # used by server
target_requests_outstanding = 10  # used by server
'''

# these need to be defined here so that they can be imported by import *
cfg = ConfigParser()
optparser = OptionParser()

_mimetrans = str.maketrans(b'-'+slip.esc, slip.esc+'-')
def mime_translate(s):
  # SLIP-encoded packets cannot contain ESC ESC.
  # Swap `-' and ESC.  The result cannot contain `--'
  return s.translate(_mimetrans)

class ConfigResults:
  def __init__(self, d = { }):
    self.__dict__ = d
  def __repr__(self):
    return 'ConfigResults('+repr(self.__dict__)+')'

c = ConfigResults()

def log_discard(packet, saddr, daddr, why):
  print('DROP ', saddr, daddr, why)
#  syslog.syslog(syslog.LOG_DEBUG,
#                'discarded packet %s -> %s (%s)' % (saddr, daddr, why))

#---------- packet parsing ----------

def packet_addrs(packet):
  version = packet[0] >> 4
  if version == 4:
    addrlen = 4
    saddroff = 3*4
    factory = ipaddress.IPv4Address
  elif version == 6:
    addrlen = 16
    saddroff = 2*4
    factory = ipaddress.IPv6Address
  else:
    raise ValueError('unsupported IP version %d' % version)
  saddr = factory(packet[ saddroff           : saddroff + addrlen   ])
  daddr = factory(packet[ saddroff + addrlen : saddroff + addrlen*2 ])
  return (saddr, daddr)

#---------- address handling ----------

def ipaddr(input):
  try:
    r = ipaddress.IPv4Address(input)
  except AddressValueError:
    r = ipaddress.IPv6Address(input)
  return r

def ipnetwork(input):
  try:
    r = ipaddress.IPv4Network(input)
  except NetworkValueError:
    r = ipaddress.IPv6Network(input)
  return r

#---------- ipif (SLIP) subprocess ----------

class SlipStreamDecoder():
  def __init__(self, on_packet):
    # we will call packet(<packet>)
    self._buffer = b''
    self._on_packet = on_packet

  def inputdata(self, data):
    #print('SLIP-GOT ', repr(data))
    self._buffer += data
    packets = slip.decode(self._buffer)
    self._buffer = packets.pop()
    for packet in packets:
      self._maybe_packet(packet)

  def _maybe_packet(self, packet):
      if len(packet):
        self._on_packet(packet)

  def flush(self):
    self._maybe_packet(self._buffer)
    self._buffer = b''

class _IpifProcessProtocol(SlipProtocol):
  def __init__(self, router):
    self._router = router
    self._decoder = SlipStreamDecoder(self.slip_on_packet)
  def connectionMade(self): pass
  def outReceived(self, data):
    self._decoder.inputdata(data)
  def slip_on_packet(self, packet):
    (saddr, daddr) = packet_addrs(packet)
    if saddr.is_link_local or daddr.is_link_local:
      log_discard(packet, saddr, daddr, 'link-local')
      return
    self._router(packet, saddr, daddr)
  def processEnded(self, status):
    status.raiseException()

def start_ipif(command, router):
  global ipif
  ipif = _IpifProcessProtocol(router)
  reactor.spawnProcess(ipif,
                       '/bin/sh',['sh','-xc', command],
                       childFDs={0:'w', 1:'r', 2:2})

def queue_inbound(packet):
  ipif.transport.write(slip.delimiter)
  ipif.transport.write(slip.encode(packet))
  ipif.transport.write(slip.delimiter)

#---------- packet queue ----------

class PacketQueue():
  def __init__(self, max_queue_time):
    self._max_queue_time = max_queue_time
    self._pq = collections.deque() # packets

  def append(self, packet):
    self._pq.append((time.monotonic(), packet))

  def nonempty(self):
    while True:
      try: (queuetime, packet) = self._pq[0]
      except IndexError: return False

      age = time.monotonic() - queuetime
      if age > self.max_queue_time:
        # strip old packets off the front
        self._pq.popleft()
        continue

      return True

  def process(self, sizequery, moredata, max_batch):
    # sizequery() should return size of batch so far
    # moredata(s) should add s to batch
    while True:
      try: (dummy, packet) = self._pq[0]
      except IndexError: break

      encoded = slip.encode(packet)
      sofar = sizequery()  

      if sofar > 0:
        if sofar + len(slip.delimiter) + len(encoded) > max_batch:
          break
        moredata(slip.delimiter)

      moredata(encoded)
      self._pq.popLeft()

#---------- error handling ----------

def crash(err):
  print('CRASH ', err, file=sys.stderr)
  try: reactor.stop()
  except twisted.internet.error.ReactorNotRunning: pass

def crash_on_defer(defer):
  defer.addErrback(lambda err: crash(err))

vdef crash_on_critical(event):
  if event.get('log_level') >= LogLevel.critical:
    crash(twisted.logger.formatEvent(event))

#---------- config processing ----------

def process_cfg_common_always():
  global mtu
  c.mtu = cfg.get('virtual','mtu')

def process_cfg_ipif(section, varmap):
  for d, s in varmap:
    try: v = getattr(c, s)
    except AttributeError: continue
    setattr(c, d, v)

  print(repr(c))

  c.ipif_command = cfg.get(section,'ipif', vars=c.__dict__)

def process_cfg_network():
  c.network = ipnetwork(cfg.get('virtual','network'))
  if c.network.num_addresses < 3 + 2:
    raise ValueError('network needs at least 2^3 addresses')

def process_cfg_server():
  try:
    c.server = cfg.get('virtual','server')
  except NoOptionError:
    process_cfg_network()
    c.server = next(c.network.hosts())

class ServerAddr():
  def __init__(self, port, addrspec):
    self.port = port
    # also self.addr
    try:
      self.addr = ipaddress.IPv4Address(addrspec)
      self._endpointfactory = twisted.internet.endpoints.TCP4ServerEndpoint
      self._inurl = '%s'
    except AddressValueError:
      self.addr = ipaddress.IPv6Address(addrspec)
      self._endpointfactory = twisted.internet.endpoints.TCP6ServerEndpoint
      self._inurl = '[%s]'
  def make_endpoint(self):
    return self._endpointfactory(reactor, self.port, self.addr)
  def url(self):
    url = 'http://' + (self._inurl % self.addr)
    if self.port != 80: url += ':%d' % self.port
    url += '/'
    return url
    
def process_cfg_saddrs():
  try: port = cfg.getint('server','port')
  except NoOptionError: port = 80

  c.saddrs = [ ]
  for addrspec in cfg.get('server','addrs').split():
    sa = ServerAddr(port, addrspec)
    c.saddrs.append(sa)

def process_cfg_clients(constructor):
  c.clients = [ ]
  for cs in cfg.sections():
    if not (':' in cs or '.' in cs): continue
    ci = ipaddr(cs)
    pw = cfg.get(cs, 'password')
    constructor(ci,cs,pw)

#---------- startup ----------

def common_startup():
  twisted.logger.globalLogPublisher.addObserver(crash_on_critical)

  optparser.add_option('-c', '--config', dest='configfile',
                       default='/etc/hippotat/config')
  (opts, args) = optparser.parse_args()
  if len(args): optparser.error('no non-option arguments please')

  re = regexp.compile('#.*')
  cfg.read_string(re.sub('', defcfg))
  cfg.read(opts.configfile)

def common_run():
  reactor.run()
  print('CRASHED (end)', file=sys.stderr)