wip client

[hippotat] / hippotat / __init__.py

# -*- python -*-

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

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

import ipaddress
from ipaddress import AddressValueError

import hippotat.slip as slip

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

import collections

cfg = ConfigParser()
optparser = OptionParser()

#---------- 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 _IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol):
  def __init__(self, router):
    self._buffer = b''
    self._router = router
  def connectionMade(self): pass
  def outReceived(self, data):
    #print('RECV ', repr(data))
    self._buffer += data
    packets = slip.decode(self._buffer)
    self._buffer = packets.pop()
    for packet in packets:
      if not len(packet): continue
      (saddr, daddr) = packet_addrs(packet)
      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 popleft(self):
    # caller must have checked nonempty
    try: (dummy, packet) = self._pq[0]
    except IndexError: return None
    return packet

#---------- 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))

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

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

def common_startup(defcfg):
  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')

  cfg.read_string(defcfg)
  cfg.read(opts.configfile)

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