Rip out old HTTPS code, replace with new asynchronous HTTP code (sic:

replacement does not yet support TLS!), beat on result with stick.

At this point the new code passes "make test" with persistent
connections disabled (but not with them enabled).  Have not yet tried
"make all-tests".  Currently logs an insane level of detail about HTTP
state, to aid in debugging; will need to be squelched later.

svn path=/rpkid/rpki/https.py; revision=2398

2 files changed, 446 insertions, 255 deletions

diff --git a/rpkid/rpki/https.py b/rpkid/rpki/https.py
index b461f0cf..558e548e 100644
--- a/rpkid/rpki/https.py
+++ b/rpkid/rpki/https.py
@@ -6,163 +6,480 @@ general version should use SQL anyway.
 
 $Id$
 
-Copyright (C) 2007--2008  American Registry for Internet Numbers ("ARIN")
+Copyright (C) 2009  Internet Systems Consortium ("ISC")
 
 Permission to use, copy, modify, and distribute this software for any
 purpose with or without fee is hereby granted, provided that the above
 copyright notice and this permission notice appear in all copies.
 
-THE SOFTWARE IS PROVIDED "AS IS" AND ARIN DISCLAIMS ALL WARRANTIES WITH
+THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
 REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
-AND FITNESS.  IN NO EVENT SHALL ARIN BE LIABLE FOR ANY SPECIAL, DIRECT,
+AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
 INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
 LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
 OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
 PERFORMANCE OF THIS SOFTWARE.
-"""
 
-import httplib, BaseHTTPServer, tlslite.api, traceback, urlparse, signal
-import rpki.x509, rpki.exceptions, rpki.log
+Portions copyright (C) 2007--2008  American Registry for Internet Numbers ("ARIN")
 
-# This should be wrapped somewhere in rpki.x509 eventually
-import POW
+Permission to use, copy, modify, and distribute this software for any
+purpose with or without fee is hereby granted, provided that the above
+copyright notice and this permission notice appear in all copies.
 
-# Do not set this to True for production use!
-disable_tls_certificate_validation_exceptions = False
+THE SOFTWARE IS PROVIDED "AS IS" AND ARIN DISCLAIMS ALL WARRANTIES WITH
+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+AND FITNESS.  IN NO EVENT SHALL ARIN BE LIABLE FOR ANY SPECIAL, DIRECT,
+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+PERFORMANCE OF THIS SOFTWARE.
+"""
 
-# Chatter about TLS certificates
-debug_tls_certs = False
+import sys, os, time, socket, asyncore, asynchat, traceback, urlparse
+import rpki.async, rpki.sundial, rpki.x509, rpki.exceptions, rpki.log
 
-# Debugging hack while converting to event-driven I/O model
-trace_synchronous_calls = True
+print "====== WARNING WARNING WARNING ======"
+print "THIS VERSION OF rpki.https DOES NOT SUPPORT TLS."
+print "CONNECTIONS ARE NOT SECURE."
+print "THIS IS A DEVELOPMENT VERSION, TLS WILL BE ADDED LATER."
+print "====== WARNING WARNING WARNING ======"
 
 rpki_content_type = "application/x-rpki"
 
-def tlslite_certChain(x509):
-  """Utility function to construct tlslite certChains."""
-  if isinstance(x509, rpki.x509.X509):
-    return tlslite.api.X509CertChain([x509.get_tlslite()])
-  else:
-    return tlslite.api.X509CertChain([x.get_tlslite() for x in x509])
-
-def build_https_ta_cache(certs):
-  """Build a dynamic TLS trust anchor cache."""
-
-  store = POW.X509Store()
-  for x in certs:
-    if rpki.https.debug_tls_certs:
-      rpki.log.debug("HTTPS dynamic trusted cert issuer %s [%s] subject %s [%s]" % (x.getIssuer(), x.hAKI(), x.getSubject(), x.hSKI()))
-    store.addTrust(x.get_POW())
-  return store
-
-class Checker(tlslite.api.Checker):
-  """Derived class to handle X.509 client certificate checking."""
 
-  ## @var refuse_tls_ca_certs
-  # Raise an exception upon receiving CA certificates via TLS rather
-  # than just quietly ignoring them.
+# ================================================================
 
-  refuse_tls_ca_certs = False
+debug = True
 
-  ## @var pem_dump_tls_certs
-  # Vile debugging hack
+want_persistent_client = False
+want_persistent_server = False
 
-  pem_dump_tls_certs = False
+idle_timeout_default   = rpki.sundial.timedelta(seconds = 60)
+active_timeout_default = rpki.sundial.timedelta(seconds = 15)
 
-  def __init__(self, trust_anchor = None, dynamic_https_trust_anchor = None):
-    """Initialize our modified certificate checker."""
-
-    self.dynamic_https_trust_anchor = dynamic_https_trust_anchor
-
-    if dynamic_https_trust_anchor is not None:
-      return
+default_http_version = (1, 0)
 
-    self.x509store = POW.X509Store()
+class http_message(object):
 
-    trust_anchor = rpki.x509.X509.normalize_chain(trust_anchor)
-    assert trust_anchor
+  software_name = "ISC RPKI library"
 
-    for x in trust_anchor:
-      if debug_tls_certs:
-        rpki.log.debug("HTTPS trusted cert issuer %s [%s] subject %s [%s]" % (x.getIssuer(), x.hAKI(), x.getSubject(), x.hSKI()))
-      self.x509store.addTrust(x.get_POW())
-      if self.pem_dump_tls_certs:
-        print x.get_PEM()
+  def __init__(self, version = None, body = None, headers = None):
+    self.version = version
+    self.body = body
+    self.headers = headers
+    self.normalize_headers()
 
-  def x509store_thunk(self):
-    if self.dynamic_https_trust_anchor is not None:
-      return self.dynamic_https_trust_anchor()
+  def normalize_headers(self, headers = None):
+    if headers is None:
+      headers = () if self.headers is None else self.headers.items()
+      translate_underscore = True
     else:
-      return self.x509store
-
-  def __call__(self, tlsConnection):
-    """POW/OpenSSL-based certificate checker.
-
-    Given our BPKI model, we're only interested in the TLS EE
-    certificates.
-    """
-
-    if tlsConnection._client:
-      chain = tlsConnection.session.serverCertChain
-      peer = "server"
+      translate_underscore = False
+    result = {}
+    for k,v in headers:
+      if translate_underscore:
+        k = k.replace("_", "-")
+      k = "-".join(s.capitalize() for s in k.split("-"))
+      v = v.strip()
+      if k in result:
+        result[k] += ", " + v
+      else:
+        result[k] = v
+    self.headers = result
+
+  @classmethod
+  def parse_from_wire(cls, headers):
+    self = cls()
+    headers = headers.split("\r\n")
+    self.parse_first_line(*headers.pop(0).split(None, 2))
+    for i in xrange(len(headers) - 2, -1, -1):
+      if headers[i + 1][0].isspace():
+        headers[i] += headers[i + 1]
+        del headers[i + 1]
+    self.normalize_headers([h.split(":", 1) for h in headers])
+    return self
+
+  def format(self):
+    s = self.format_first_line()
+    if self.body is not None:
+      assert isinstance(self.body, str)
+      self.headers["Content-Length"] = len(self.body)
+    for kv in self.headers.iteritems():
+      s += "%s: %s\r\n" % kv
+    s += "\r\n"
+    if self.body is not None:
+      s += self.body
+    return s
+
+  def __str__(self):
+    return self.format()
+
+  def parse_version(self, version):
+    if version[:5] != "HTTP/":
+      raise RuntimeError, "Couldn't parse version %s" % version
+    self.version = tuple(int(i) for i in version[5:].split("."))
+
+  def persistent(self):
+    c = self.headers.get("Connection")
+    if self.version == (1, 1):
+      return c is None or "close" not in c.lower()
+    elif self.version == (1, 0):
+      return c is not None and "keep-alive" in c.lower()
     else:
-      chain = tlsConnection.session.clientCertChain
-      peer = "client"
-
-    chain = [rpki.x509.X509(tlslite = chain.x509List[i]) for i in range(chain.getNumCerts())]
-
-    ee = None
-
-    for x in chain:
-
-      if debug_tls_certs:
-        rpki.log.debug("Received %s TLS %s cert issuer %s [%s] subject %s [%s]"
-                       % (peer, "CA" if x.is_CA() else "EE", x.getIssuer(), x.hAKI(), x.getSubject(), x.hSKI()))
-        if self.pem_dump_tls_certs:
-          print x.get_PEM()
-
-      if x.is_CA():
-        if self.refuse_tls_ca_certs:
-          raise rpki.exceptions.ReceivedTLSCACert
-        continue
+      return False
 
-      if ee is not None:
-        raise rpki.exceptions.MultipleTLSEECert, chain
-      ee = x
+class http_request(http_message):
+
+  def __init__(self, cmd = None, path = None, version = default_http_version, body = None, callback = None, **headers):
+    if cmd is not None and cmd != "POST" and body is not None:
+      raise RuntimeError
+    http_message.__init__(self, version = version, body = body, headers = headers)
+    self.cmd = cmd
+    self.path = path
+    self.callback = callback
+
+  def parse_first_line(self, cmd, path, version):
+    self.parse_version(version)
+    self.cmd = cmd
+    self.path = path
+
+  def format_first_line(self):
+    self.headers.setdefault("User-Agent", self.software_name)
+    return "%s %s HTTP/%d.%d\r\n" % (self.cmd, self.path, self.version[0], self.version[1])
+
+class http_response(http_message):
+
+  def __init__(self, code = None, reason = None, version = default_http_version, body = None, **headers):
+    http_message.__init__(self, version = version, body = body, headers = headers)
+    self.code = code
+    self.reason = reason
+
+  def parse_first_line(self, version, code, reason):
+    self.parse_version(version)
+    self.code = int(code)
+    self.reason = reason
+
+  def format_first_line(self):
+    self.headers.setdefault("Date", time.strftime("%a, %d %b %Y %T GMT"))
+    self.headers.setdefault("Server", self.software_name)
+    return "HTTP/%d.%d %s %s\r\n" % (self.version[0], self.version[1], self.code, self.reason)
+
+def logger(self, msg):
+  if debug:
+    rpki.log.debug("%r: %s" % (self, msg))
+
+class http_stream(asynchat.async_chat):
+
+  log = logger
+
+  idle_timeout = idle_timeout_default
+  active_timeout = active_timeout_default
+
+  def __init__(self, conn = None):
+    asynchat.async_chat.__init__(self, conn = conn)
+    self.buffer = []
+    self.timer = rpki.async.timer(self.handle_timeout)
+    self.restart()
+
+  def restart(self, idle = True):
+    assert not self.buffer
+    self.chunk_handler = None
+    self.set_terminator("\r\n\r\n")
+    timeout = self.idle_timeout if idle else self.active_timeout
+    if timeout is not None:
+      self.timer.set(timeout)
+    else:
+      self.timer.cancel()
 
-    result = self.x509store_thunk().verifyDetailed(ee.get_POW())
-    if not result[0]:
-      rpki.log.debug("TLS certificate validation result %s" % repr(result))
-      if disable_tls_certificate_validation_exceptions:
-        rpki.log.warn("DANGER WILL ROBINSON!  IGNORING TLS VALIDATION FAILURE!")
+  def update_active_timeout(self):
+    if self.active_timeout is not None:
+      self.timer.set(self.active_timeout)
+    else:
+      self.timer.cancel()
+
+  def collect_incoming_data(self, data):
+    """Buffer the data"""
+    self.buffer.append(data)
+    self.update_active_timeout()
+
+  def get_buffer(self):
+    val = "".join(self.buffer)
+    self.buffer = []
+    return val
+
+  def found_terminator(self):
+    self.update_active_timeout()
+    if self.chunk_handler:
+      self.chunk_handler()
+    elif not isinstance(self.get_terminator(), str):
+      self.handle_body()
+    else:
+      self.log("Got headers")
+      self.msg = self.parse_type.parse_from_wire(self.get_buffer())
+      if self.msg.version == (1, 1) and "chunked" in self.msg.headers.get("Transfer-Encoding", "").lower():
+        self.msg.body = []
+        self.chunk_handler = self.chunk_header
+        self.set_terminator("\r\n")
+      elif "Content-Length" in self.msg.headers:
+        self.set_terminator(int(self.msg.headers["Content-Length"]))
       else:
-        raise rpki.exceptions.TLSValidationError
+        self.handle_no_content_length()
+      
+  def chunk_header(self):
+    n = int(self.get_buffer().partition(";")[0], 16)
+    self.log("Chunk length %s" % n)
+    if n:
+      self.chunk_handler = self.chunk_body
+      self.set_terminator(n)
+    else:
+      self.msg.body = "".join(self.msg.body)
+      self.chunk_handler = self.chunk_discard_trailer
+
+  def chunk_body(self):
+    self.log("Chunk body")
+    self.msg.body += self.buffer
+    self.buffer = []
+    self.chunk_handler = self.chunk_discard_crlf
+    self.set_terminator("\r\n")
+
+  def chunk_discard_crlf(self):
+    self.log("Chunk CRLF")
+    s = self.get_buffer()
+    assert s == "", "Expected chunk CRLF, got '%s'" % s
+    self.chunk_handler = self.chunk_header
+
+  def chunk_discard_trailer(self):
+    self.log("Chunk trailer")
+    s = self.get_buffer()
+    assert s == "", "Expected end of chunk trailers, got '%s'" % s
+    self.chunk_handler = None
+    self.handle_message()
+
+  def handle_body(self):
+    self.msg.body = self.get_buffer()
+    self.handle_message()
+
+  def handle_error(self):
+    self.log("Error in HTTP stream handler")
+    print traceback.format_exc()
+    asyncore.close_all()
+
+  def handle_timeout(self):
+    self.log("Timeout, closing")
+    self.close()
+
+  def handle_close(self):
+    asynchat.async_chat.handle_close(self)
+    self.timer.cancel()
+    self.log("Closed")
+
+class http_server(http_stream):
+
+  parse_type = http_request
+
+  def __init__(self, conn, handlers):
+    self.handlers = handlers
+    http_stream.__init__(self, conn)
+    self.expect_close = not want_persistent_server
+
+  def handle_no_content_length(self):
+    self.handle_message()
+
+  def find_handler(self, path):
+    """Helper method to search self.handlers."""
+    for s, h in self.handlers:
+      if path.startswith(s):
+        return h
+    return None
 
-class httpsClient(tlslite.api.HTTPTLSConnection):
-  """Derived class to let us replace the default Checker."""
+  def handle_message(self):
+    if not self.msg.persistent():
+      self.expect_close = True
+    handler = self.find_handler(self.msg.path)
+    error = None
+    if self.msg.cmd != "POST":
+      error = 501, "No handler for method %s" % self.msg.cmd
+    elif self.msg.headers["Content-Type"] != rpki_content_type:
+      error = 415, "No handler for Content-Type %s" % self.headers["Content-Type"]
+    elif handler is None:
+      error = 404, "No handler for URL %s" % self.msg.path
+    if error is None:
+      handler(self.msg.body, self.msg.path, self.send_reply)
+    else:
+      self.send_error(*error)
+
+  def send_error(self, code, reason):
+    self.send_message(code = code, reason = reason)
+
+  def send_reply(self, code, body):
+    self.send_message(code = code, body = body)
+
+  def send_message(self, code, reason = "OK", body = None):
+    msg = http_response(code = code, reason = reason, body = body,
+                        Content_Type = rpki_content_type,
+                        Connection = "Close" if self.expect_close else "Keep-Alive")
+    self.push(msg.format())
+    if self.expect_close:
+      self.log("Closing")
+      self.timer.cancel()
+      self.close_when_done()
+    else:      
+      self.log("Listening for next message")
+      self.restart()
+
+class http_listener(asyncore.dispatcher):
+
+  log = logger
+
+  def __init__(self, handlers, port = 80, host = ""):
+    asyncore.dispatcher.__init__(self)
+    self.handlers = handlers
+    self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+    self.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    self.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+    self.bind((host, port))
+    self.listen(5)
+    self.log("Listening on (host, port) %r, handlers %r" % ((host, port), handlers))
+
+  def handle_accept(self):
+    self.log("Accepting connection")
+    server = http_server(conn = self.accept()[0], handlers = self.handlers)
+
+  def handle_error(self):
+    self.log("Error in HTTP listener")
+    print traceback.format_exc()
+    asyncore.close_all()
+
+class http_client(http_stream):
+
+  parse_type = http_response
+
+  def __init__(self, queue, hostport):
+    self.log("Creating new connection to %s" % repr(hostport))
+    http_stream.__init__(self)
+    self.queue = queue
+    self.state = "idle"
+    self.expect_close = not want_persistent_client
+    self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+    self.connect(hostport)
+
+  def handle_no_content_length(self):
+    self.set_terminator(None)
+
+  def send_request(self, msg):
+    self.log("Sending request")
+    assert self.state == "idle", "%r: state should be idle, is %s" % (self, self.state)
+    if msg is not None:
+      self.state = "request-sent"
+      msg.headers["Connection"] = "Close" if self.expect_close else "Keep-Alive"
+      self.push(msg.format())
+      self.restart(idle = False)
+
+  def handle_message(self):
+    if not self.msg.persistent():
+      self.expect_close = True
+    self.log("Message received, state %s" % self.state)
+    msg = None
+    if self.state == "request-sent":
+      msg = self.queue.done_with_request()
+    elif self.state == "idle":
+      self.log("Received unsolicited message")
+    elif self.state == "closing":
+      assert not self.msg.body
+      self.log("Ignoring empty response received while closing")
+      return
+    else:
+      raise RuntimeError, "[%r: Unexpected state]" % self
+    self.state = "idle"
+    if msg != None:
+      if self.msg.code != 200:
+        rpki.log.debug("HTTPS client returned failure")
+        msg.callback(rpki.exceptions.HTTPRequestFailed("HTTP request failed with status %s, reason %s, response %s" % (self.msg.code, self.msg.reason, self.msg.body)))
+      else:
+        msg.callback(self.msg.body)
+    msg = self.queue.next_request(not self.expect_close)
+    if msg is not None:
+      self.log("Got a new message to send from my queue")
+      self.send_request(msg)
+    elif self.expect_close:
+      self.log("Closing")
+      self.state = "closing"
+      self.queue.closing(self)
+      self.close_when_done()
+    else:
+      self.log("Idling")
+      self.timer.set(self.idle_timeout)
+
+  def handle_connect(self):
+    self.log("Connected")
+    self.send_request(self.queue.next_request(True))
+
+  def kickstart(self):
+    self.log("Kickstart")
+    assert self.state == "idle"
+    self.send_request(self.queue.next_request(True))
+
+  def handle_close(self):
+    http_stream.handle_close(self)
+    self.queue.closing(self)
+    if self.get_terminator() is None:
+      self.handle_body()
+
+class http_queue(object):
+
+  log = logger
+
+  def __init__(self, hostport):
+    self.log("Creating queue for %s" % repr(hostport))
+    self.hostport = hostport
+    self.client = None
+    self.queue = []
+
+  def request(self, *requests):
+    self.log("Adding requests %r" % requests)
+    need_kick = self.client is not None and not self.queue
+    self.queue.extend(requests)
+    if self.client is None:
+      self.client = http_client(self, self.hostport)
+    elif need_kick:
+      self.client.kickstart()
+
+  def done_with_request(self):
+    req = self.queue.pop(0)
+    self.log("Dequeuing request %r" % req)
+    return req
+
+  def next_request(self, usable):
+    if not self.queue:
+      self.log("Queue is empty")
+      return None
+    self.log("Queue: %r" % self.queue)
+    if usable:
+      self.log("Queue not empty and connection usable")
+      return self.queue[0]
+    else:
+      self.log("Queue not empty but connection not usable, spawning")
+      self.client = http_client(self, self.hostport)
+      self.log("Spawned connection %r" % self.client)
+      return None
 
-  def __init__(self, host, port = None,
-               client_cert = None, client_key = None,
-               server_ta = None, settings = None):
-    """Create a new httpsClient."""
+  def closing(self, client):
+    if client is self.client:
+      self.log("Removing client")
+      self.client = None
 
-    tlslite.api.HTTPTLSConnection.__init__(
-      self, host = host, port = port, settings = settings,
-      certChain = client_cert, privateKey = client_key)
 
-    self.checker = Checker(trust_anchor = server_ta)
+queues = {}
 
 def client(msg, client_key, client_cert, server_ta, url, timeout = 300, callback = None):
   """Open client HTTPS connection, send a message, wait for response.
 
-  This function wraps most of what one needs to do to send a message
-  over HTTPS and get a response.  The certificate checking isn't quite
-  up to snuff; it's better than with the other packages I've found,
-  but doesn't appear to handle subjectAltName extensions (sigh).
+  THIS VERSION DOES NOT DO TLS.  THIS IS EXPERIMENTAL CODE.  DO NOT
+  USE IN PRODUCTION UNTIL TLS SUPPORT HAS BEEN ADDED.
   """
 
   # This is an easy way to find synchronous calls that need conversion
-  if trace_synchronous_calls and callback is None:
+  if callback is None:
     raise RuntimeError, "Synchronous call to rpki.http.client()"
 
   u = urlparse.urlparse(url)
@@ -177,149 +494,20 @@ def client(msg, client_key, client_cert, server_ta, url, timeout = 300, callback
 
   rpki.log.debug("Contacting %s" % url)
 
-  if debug_tls_certs:
-    for cert in (client_cert,) if isinstance(client_cert, rpki.x509.X509) else client_cert:
-      rpki.log.debug("Sending client TLS cert issuer %s subject %s" % (cert.getIssuer(), cert.getSubject()))
-
-  # We could add a "settings = foo" argument to the following call to
-  # pass in a tlslite.HandshakeSettings object that would let us
-  # insist on, eg, particular SSL/TLS versions.
-
-  httpc = httpsClient(host         = u.hostname or "localhost",
-                      port         = u.port or 443,
-                      client_key   = client_key.get_tlslite(),
-                      client_cert  = tlslite_certChain(client_cert),
-                      server_ta    = server_ta)
-  httpc.connect()
-  httpc.sock.settimeout(timeout)
-  httpc.request("POST", u.path, msg, {"Content-Type" : rpki_content_type})
-  response = httpc.getresponse()
-  rpki.log.debug("HTTPS client returned")
-  r = response.read()
-  if response.status != httplib.OK:
-    rpki.log.debug("HTTPS client returned failure")
-    r = rpki.exceptions.HTTPRequestFailed("HTTP request failed with status %s, response %s" % (response.status, r))
-  if callback is not None:
-    rpki.log.debug("HTTPS client callback supplied, using it")
-    callback(r)
-  elif response.status == httplib.OK:
-    rpki.log.debug("HTTPS no client callback, returning success")
-    return r
-  else:
-    rpki.log.debug("HTTPS no client callback, raising exception")
-    raise r
-
-class requestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
-  """Derived type to supply POST handler and override logging."""
-
-  rpki_handlers = None                  # Subclass must bind
-
-  def rpki_find_handler(self):
-    """Helper method to search self.rpki_handlers."""
-    for s, h in self.rpki_handlers:
-      if self.path.startswith(s):
-        return h
-    return None
-
-  def do_POST(self):
-    """POST handler."""
-    try:
-      handler = self.rpki_find_handler()
-      if self.headers["Content-Type"] != rpki_content_type:
-        result = 415, "No handler for Content-Type %s" % self.headers["Content-Type"]
-      elif handler is None:
-        result = 404, "No handler found for URL " + self.path
-      else:
-        self.called_back = False
-        result = handler(query = self.rfile.read(int(self.headers["Content-Length"])),
-                         path  = self.path,
-                         cb    = self.do_POST_cb)
-        assert result is not None or self.called_back, "Missing HTTPS server callback from %s" % repr(handler)
-    except Exception, edata:
-      rpki.log.error(traceback.format_exc())
-      result = 500, "Unhandled exception %s" % edata
-    if result is not None:
-      self.do_POST_cb(result[0], result[1])
-
-  def do_POST_cb(self, rcode, rtext):
-    """Send result back to client."""
-    rpki.log.info("HTTPS server callback")
-    self.called_back = True
-    self.send_response(rcode)
-    self.send_header("Content-Type", rpki_content_type)
-    self.end_headers()
-    self.wfile.write(rtext)
-
-  def log_message(self, format, *args):
-    """Redirect HTTP server logging into our own logging system."""
-    if args:
-      rpki.log.info(format % args)
-    else:
-      rpki.log.info(format)
-
-class httpsServer(tlslite.api.TLSSocketServerMixIn, BaseHTTPServer.HTTPServer):
-  """Derived type to handle TLS aspects of HTTPS."""
-
-  rpki_sessionCache = None
-  rpki_server_key   = None
-  rpki_server_cert  = None
-  rpki_checker      = None
-  
-  def handshake(self, tlsConnection):
-    """TLS handshake handler."""
-    assert self.rpki_server_cert  is not None
-    assert self.rpki_server_key   is not None
-    assert self.rpki_sessionCache is not None
-
-    try:
-      #
-      # We could add a "settings = foo" argument to the following call
-      # to pass in a tlslite.HandshakeSettings object that would let
-      # us insist on, eg, particular SSL/TLS versions.
-      #
-      tlsConnection.handshakeServer(certChain    = self.rpki_server_cert,
-                                    privateKey   = self.rpki_server_key,
-                                    sessionCache = self.rpki_sessionCache,
-                                    checker      = self.rpki_checker,
-                                    reqCert      = True)
-      tlsConnection.ignoreAbruptClose = True
-      return True
-    except (tlslite.api.TLSError, rpki.exceptions.TLSValidationError), error:
-      rpki.log.warn("TLS handshake failure: " + str(error))
-      return False
+  request = http_request(cmd = "POST", path = u.path, body = msg, callback = callback,
+                         Host = u.hostname, Content_Type = rpki_content_type)
+  hostport = (u.hostname or "localhost", u.port or 80)
+  rpki.log.debug("Created request %r for %r" % (request, hostport))
+  if hostport not in queues:
+    queues[hostport] = http_queue(hostport)
+  queues[hostport].request(request)
 
-  def handle_error(self, request, client_address):
-    """Override SOcketServer error handling.  This may be wrong in the
-    long run, but at the moment I'm seeing the server hang while
-    trying to shut down, because the default handler is intercepting
-    ServerShuttingDown in certain states, for reasons unknown.
-    """
-
-    raise
-
-def server(handlers, server_key, server_cert, port = 4433, host ="", client_ta = None, dynamic_https_trust_anchor = None, catch_signals = (signal.SIGINT, signal.SIGTERM)):
+def server(handlers, server_key, server_cert, port = 4433, host ="", client_ta = None, dynamic_https_trust_anchor = None):
   """Run an HTTPS server and wait (forever) for connections."""
 
   if not isinstance(handlers, (tuple, list)):
     handlers = (("/", handlers),)
 
-  class boundRequestHandler(requestHandler):
-    rpki_handlers = handlers
-
-  httpd = httpsServer((host, port), boundRequestHandler)
-
-  httpd.rpki_server_key   = server_key.get_tlslite()
-  httpd.rpki_server_cert  = tlslite_certChain(server_cert)
-  httpd.rpki_sessionCache = tlslite.api.SessionCache()
-  httpd.rpki_checker      = Checker(trust_anchor = client_ta, dynamic_https_trust_anchor = dynamic_https_trust_anchor)
-
-  try:
-    def raiseServerShuttingDown(signum, frame):
-      raise rpki.exceptions.ServerShuttingDown
-    old_signal_handlers = tuple((sig, signal.signal(sig, raiseServerShuttingDown)) for sig in catch_signals)
-    httpd.serve_forever()
-  except rpki.exceptions.ServerShuttingDown:
-    pass
-  finally:
-    for sig, handler in old_signal_handlers:
-      signal.signal(sig, handler)
+  listener = http_listener(port = port, handlers = handlers)
+  rpki.async.event_loop()
+
diff --git a/rpkid/testbed.py b/rpkid/testbed.py
index f47f71c3..c2fa2df5 100644
--- a/rpkid/testbed.py
+++ b/rpkid/testbed.py
@@ -229,7 +229,9 @@ class main(object):
 
       rpki.async.iterator(self.db.engines, create_rpki_objects, self.created_rpki_objects)
 
-      # At this point we have gone into (pseudo) event-driven code.
+      rpki.async.event_loop()
+
+      # At this point we have gone into event-driven code.
       # See comments above about cleanup of this try/finally code
 
       rpki.log.info("All done")
@@ -667,6 +669,7 @@ class allocation(object):
     rpki.log.info("Call to rpkid %s returned" % self.name)
 
   def call_rpkid_cb(self, val):
+    rpki.log.info("Callback from to rpkid %s" % self.name)
     if isinstance(val, Exception):
       raise val
     msg, xml = rpki.left_right.cms_msg.unwrap(val, (self.rpkid_ta, self.rpkid_cert),