client

   1 #!/usr/bin/python3
   2
   3 from hippotat import *
   4
   5 import twisted.web
   6 import twisted.web.client
   7
   8 import io
   9
  10 client_cs = None
  11
  12 def set_client(ci,cs,pw):
  13   global client_cs
  14   global password
  15   assert(client_cs is None)
  16   client_cs = cs
  17   c.client = ci
  18   c.max_outstanding = cfg.getint(cs, 'max_requests_outstanding')
  19   c.target_outstanding = cfg.getint(cs, 'target_requests_outstanding')
  20   password = pw
  21
  22 def process_cfg():
  23   global url
  24   global max_requests_outstanding
  25
  26   process_cfg_common_always()
  27   process_cfg_server()
  28
  29   try:
  30     c.url = cfg.get('server','url')
  31   except NoOptionError:
  32     process_cfg_saddrs()
  33     c.url = c.saddrs[0].url()
  34
  35   process_cfg_clients(set_client)
  36
  37   c.routes = cfg.get('virtual','routes')
  38   c.max_queue_time = cfg.getint(client_cs, 'max_queue_time')
  39   c.max_batch_up   = cfg.getint(client_cs, 'max_batch_up')
  40   c.http_timeout   = cfg.getint(client_cs, 'http_timeout')
  41   c.http_retry     = cfg.getint(client_cs, 'http_retry')
  42
  43   process_cfg_ipif(client_cs,
  44                    (('local', 'client'),
  45                     ('peer',  'server'),
  46                     ('rnets', 'routes')))
  47
  48 outstanding = 0
  49
  50 def start_client():
  51   global queue
  52   global agent
  53   queue = PacketQueue('up', c.max_queue_time)
  54   agent = twisted.web.client.Agent(reactor, connectTimeout = c.http_timeout)
  55
  56 def outbound(packet, saddr, daddr):
  57   #print('OUT ', saddr, daddr, repr(packet))
  58   queue.append(packet)
  59   check_outbound()
  60
  61 def crashy(): assert(False)
  62
  63 class ResponseConsumer(twisted.internet.protocol.Protocol):
  64   def __init__(self, req):
  65     self._req = req
  66     self._ssd = SlipStreamDecoder(crashy)
  67     self._log(DBG.HTTP_CTRL, '__init__')
  68
  69   def _log(self, dflag, msg, **kwargs):
  70     log_debug(dflag, 'RC ' + msg, idof=self._req, **kwargs)
  71
  72   def dataReceived(self, data):
  73     self._log(DBG.HTTP_CTRL, 'dataReceived', d=data)
  74     try:
  75       self._ssd.inputdata(mime_translate(data))
  76     except Exception as e:
  77       self._handleexception()
  78
  79   def connectionMade(self):
  80     self._log(DBG.HTTP_CTRL, 'connectionMade')
  81
  82   def connectionLost(self, reason):
  83     self._log(DBG.HTTP_CTRL, 'connectionLost ' + str(reason))
  84     if not reason.check(twisted.web.client.ResponseDone):
  85       self._asyncfailure(reason)
  86       return
  87     try:
  88       self._ssd.flush()
  89     except Exception as e:
  90       self._handleexception()
  91
  92   def _handleexception(self):
  93     self._asyncfailure(traceback.format_exc())
  94
  95   def _asyncfailure(self, reason):
  96     self._log(DBG.HTTP_CTRL, '_asyncFailure ' + str(reason))
  97     req_err(self._req, reason)
  98
  99 class ErrorResponseConsumer(twisted.internet.protocol.Protocol):
 100   def __init__(self, req, resp):
 101     self._req = req
 102     self._resp = resp
 103
 104     try:
 105       self._phrase = resp.phrase.decode('utf-8')
 106     except Exception:
 107       self._phrase = repr(resp.phrase)
 108
 109     self._log(DBG.HTTP_CTRL, '__init__ %d %s' % (resp.code, self._phrase))
 110
 111   def _log(self, dflag, msg, **kwargs):
 112     log_debug(dflag,'ERROR-RC '+msg, idof=self._req, **kwargs)
 113
 114   def connectionMade(self):
 115     self._m = b''
 116
 117   def dataReceived(self, data):
 118     self._log(DBG.HTTP_CTRL, 'dataReceived ' + repr(data))
 119     self._m += data
 120
 121   def connectionLost(self, reason):
 122     try:
 123       mbody = self._m.decode('utf-8')
 124     except Exception:
 125       mbody = repr(self._m)
 126     if not reason.check(twisted.web.client.ResponseDone):
 127       mbody += ' || ' + str(reason)
 128     req_err(self._req,
 129             "FAILED %d %s | %s"
 130             % (self._resp.code, self._phrase, mbody))
 131
 132 def req_ok(req, resp):
 133   log_debug(DBG.HTTP_CTRL,
 134             'req_ok %d %s %s' % (resp.code, repr(resp.phrase), str(resp)),
 135             idof=req)
 136   if resp.code == 200:
 137     rc = ResponseConsumer(req)
 138   else:
 139     rc = ErrorResponseConsumer(req, resp)
 140
 141   resp.deliverBody(rc)
 142
 143 def req_err(req, err):
 144   log_debug(DBG.HTTP_CTRL, 'req_err ' + str(err), idof=req)
 145   if isinstance(err, twisted.python.failure.Failure):
 146     err = err.getTraceback()
 147   print(err, file=sys.stderr)
 148   reactor.callLater(c.http_retry, (lambda: req_fin(req)))
 149
 150 def req_fin(req):
 151   global outstanding
 152   outstanding -= 1
 153   log_debug(DBG.HTTP_CTRL, 'req_fin OS=%d' % outstanding, idof=req)
 154   check_outbound()
 155
 156 def check_outbound():
 157   global outstanding
 158
 159   while True:
 160     if                          outstanding >= c.max_outstanding   : break
 161     if not queue.nonempty() and outstanding >= c.target_outstanding: break
 162
 163     d = b''
 164     def moredata(s): nonlocal d; d += s
 165     queue.process((lambda: len(d)),
 166                   moredata,
 167                   c.max_batch_up)
 168
 169     d = mime_translate(d)
 170
 171     crlf = b'\r\n'
 172     lf   =   b'\n'
 173     mime = (b'--b'                                        + crlf +
 174             b'Content-Type: text/plain; charset="utf-8"'  + crlf +
 175             b'Content-Disposition: form-data; name="m"'   + crlf + crlf +
 176             str(c.client)             .encode('ascii')    + crlf +
 177             password                                      + crlf +
 178             str(c.target_outstanding) .encode('ascii')    + crlf +
 179           ((
 180             b'--b'                                        + crlf +
 181             b'Content-Type: application/octet-stream'     + crlf +
 182             b'Content-Disposition: form-data; name="d"'   + crlf + crlf +
 183             d                                             + crlf
 184            ) if len(d) else b'')                               +
 185             b'--b--'                                      + crlf)
 186
 187     #df = open('data.dump.dbg', mode='wb')
 188     #df.write(mime)
 189     #df.close()
 190     # POST -use -c 'multipart/form-data; boundary="b"' http://localhost:8099/ <data.dump.dbg
 191
 192     log_debug(DBG.HTTP_FULL, 'requesting: ' + str(mime))
 193
 194     hh = { 'User-Agent': ['hippotat'],
 195            'Content-Type': ['multipart/form-data; boundary="b"'],
 196            'Content-Length': [str(len(mime))] }
 197
 198     bytesreader = io.BytesIO(mime)
 199     producer = twisted.web.client.FileBodyProducer(bytesreader)
 200
 201     req = agent.request(b'POST',
 202                         c.url,
 203                         twisted.web.client.Headers(hh),
 204                         producer)
 205
 206     outstanding += 1
 207     log_debug(DBG.HTTP_CTRL, 'request OS=%d' % outstanding, idof=req, d=d)
 208     req.addTimeout(c.http_timeout, reactor)
 209     req.addCallback((lambda resp: req_ok(req, resp)))
 210     req.addErrback((lambda err: req_err(req, err)))
 211
 212 common_startup()
 213 process_cfg()
 214 start_client()
 215 start_ipif(c.ipif_command, outbound)
 216 check_outbound()
 217 common_run()