fixes

[hippotat] / server

#!/usr/bin/python3

from hippotat import *

import os

import twisted.internet
from twisted.web.server import NOT_DONE_YET

#import twisted.web.server import Site
#from twisted.web.resource import Resource

import syslog

import traceback

clients = { }

#---------- "router" ----------

def route(packet, iface, saddr, daddr):
  def lt(dest):
    log_debug(DBG.ROUTE, 'route: %s -> %s: %s' % (saddr,daddr,dest), d=packet)
  try: dclient = clients[daddr]
  except KeyError: dclient = None
  if dclient is not None:
    lt('client')
    dclient.queue_outbound(packet)
  elif daddr == c.server or daddr not in c.network:
    lt('inbound')
    queue_inbound(packet)
  elif daddr == relay:
    lt('discard relay')
    log_discard(packet, iface, saddr, daddr, 'relay')
  else:
    lt('discard no-client')
    log_discard(packet, iface, saddr, daddr, 'no-client')

#---------- client ----------

class Client():
  def __init__(self, ip, cs, pw):
    # instance data members
    self._ip = ip
    self._cs = cs
    self.pw = pw
    self._rq = collections.deque() # requests
    # self._pq = PacketQueue(...)
    # plus from config:
    #  .max_batch_down
    #  .max_queue_time
    #  .max_request_time
    #  .target_requests_outstanding

    if ip not in c.network:
      raise ValueError('client %s not in network' % ip)

    for k in ('max_batch_down','max_queue_time','max_request_time',
              'target_requests_outstanding'):
      req = cfg.getint(cs, k)
      limit = cfg.getint('limits',k)
      self.__dict__[k] = min(req, limit)

    self._pq = PacketQueue(ip, self.max_queue_time)

    if ip in clients:
      raise ValueError('multiple client cfg sections for %s' % ip)
    clients[ip] = self

    self._log(DBG.INIT, 'new')

  def _log(self, dflag, msg, **kwargs):
    log_debug(dflag, ('client %s: ' % self._ip)+msg, **kwargs)

  def process_arriving_data(self, d):
    self._log(DBG.FLOW, 'req data', d=d)
    for packet in slip.decode(d):
      (saddr, daddr) = packet_addrs(packet)
      if saddr != self._ip:
        raise ValueError('wrong source address %s' % saddr)
      route(packet, self._ip, saddr, daddr)

  def _req_cancel(self, request):
    self._log(DBG.HTTPCTRL, 'cancel', idof=request)
    request.finish()

  def _req_error(self, err, request):
    self._log(DBG.HTTPCTRL, 'error %s' % err, idof=request)
    self._req_cancel(request)

  def queue_outbound(self, packet):
    self._pq.append(packet)
    self._check_outbound()

  def new_request(self, request):
    request.setHeader('Content-Type','application/octet-stream')
    reactor.callLater(self.max_request_time, self._req_cancel, request)
    request.notifyFinish().addErrback(self._req_error, request)
    self._rq.append(request)
    self._check_outbound()

  def _check_outbound(self):
    log_debug(DBG.HTTPCTRL, 'CHKO')
    while True:
      try: request = self._rq[0]
      except IndexError: request = None
      if request and request.finished:
        self._log(DBG.HTTP_CTRL, 'CHKO req finished, discard', idof=request)
        self._rq.popleft()
        continue

      if not self._pq.nonempty():
        # no packets, oh well
        self._log(DBG.HTTP_CTRL, 'CHKO no packets, OUT-DONE', idof=request)
        break

      if request is None:
        # no request
        self._log(DBG.HTTP_CTRL, 'CHKO no request, OUT-DONE', idof=request)
        break

      self._log(DBG.HTTP_CTRL, 'CHKO processing', idof=request)
      # request, and also some non-expired packets
      self._pq.process((lambda: request.sentLength),
                       request.write,
                       self.max_batch_down)

      assert(request.sentLength)
      self._rq.popleft()
      request.finish()
      self._log(DBG.HTTP, 'complete', idof=request)
      # round again, looking for more to do

    while len(self._rq) > self.target_requests_outstanding:
      request = self._rq.popleft()
      self._log(DBG.HTTP, 'CHKO above target, returning empty', idof=request)
      request.finish()

def process_request(request, desca):
  # find client, update config, etc.
  metadata = request.args[b'm'][0]
  metadata = metadata.split(b'\r\n')
  (ci_s, pw, tro) = metadata[0:3]
  desca['m[0,2]'] = [ci_s, tro]
  ci_s = ci_s.decode('utf-8')
  tro = int(tro)
  desca['tro']= tro
  ci = ipaddr(ci_s)
  desca['ci'] = ci
  cl = clients[ci]
  if pw != cl.pw: raise ValueError('bad password')
  desca['pwok']=True

  if tro != cl.target_requests_outstanding:
    raise ValueError('tro must be %d' % cl.target_requests_outstanding)

  try:
    d = request.args[b'd']
    desca['d'] = d
  except KeyError: d = ''

  cl.process_arriving_data(d)
  cl.new_request(request)

def log_http(desca, msg):
  try:
    d = desca['d']
    del desca['d']
  except KeyError:
    d = None
  log_debug(DBG.HTTP, msg + repr(desca), d=d)

class IphttpResource(twisted.web.resource.Resource):
  isLeaf = True
  def render_POST(self, request):
    log_debug(DBG.HTTP_FULL,
              'req recv: ' + repr(request) + ' ' + repr(request.args),
              idof=id(request))
    desca = {'d': None}
    try: process_request(request, desca)
    except Exception as e:
      emsg = traceback.format_exc()
      log_http(desca, 'EXCEPTION ' + emsg)
      request.setHeader('Content-Type','text/plain; charset="utf-8"')
      request.setResponseCode(400)
      return (emsg + ' # ' + repr(desca) + '\r\n').encode('utf-8')
    log_http(desca, '... [%s]' % id(request))
    return NOT_DONE_YET

  def render_GET(self, request):
    log_debug(DBG.HTTP, 'GET request')
    return b'<html><body>hippotat</body></html>'

def start_http():
  resource = IphttpResource()
  site = twisted.web.server.Site(resource)
  for sa in c.saddrs:
    ep = sa.make_endpoint()
    crash_on_defer(ep.listen(site))
    log_debug(DBG.INIT, 'listening on %s' % sa)

#---------- config and setup ----------
        
def process_cfg():
  process_cfg_common_always()
  process_cfg_server()
  process_cfg_network()

  try:
    c.relay = cfg.get('virtual','relay')
  except NoOptionError:
    for search in c.network.hosts():
      if search == c.server: continue
      c.relay = search
      break

  process_cfg_saddrs()
  process_cfg_clients(Client)

  process_cfg_ipif('server',
                   (('local','server'),
                    ('peer', 'relay'),
                    ('rnets','network')))

common_startup()
process_cfg()
start_ipif(c.ipif_command, route)
start_http()
common_run()