path: root/rp/rcynic/rcynicng

#!/usr/bin/env python

# $Id$

"""
Reimplementation of rcynic in Python.  Work in progress.
"""

import os
import sys
import ssl
import time
import errno
import shutil
import socket
import logging
import argparse
import urlparse
import subprocess

import tornado.gen
import tornado.locks
import tornado.ioloop
import tornado.queues
import tornado.process
import tornado.httpclient

import rpki.POW
import rpki.sundial
import rpki.config
import rpki.autoconf
import rpki.relaxng

from rpki.oids import id_kp_bgpsec_router

from lxml.etree import ElementTree, Element, SubElement, Comment, XML

logger = logging.getLogger("rcynicng")

codes = rpki.POW.validation_status


class Status(object):
    """
    Validation status database, like validation_status_t in rcynic:tos.

    rcynic:tos version of this data structure is stored as an AVL
    tree, because the OpenSSL STACK_OF() sort-and-bsearch turned out
    to be a very poor choice for the input data.  Remains to be seen
    whether we need to do something like that here too.
    """

    db = dict()

    def __init__(self, uri):
        self.uri = uri
        self._timestamp = None
        self.status = set()

    def __str__(self):
        return "{my.timestamp} {my.uri} {status}".format(
            my = self, status = ",".join(str(s) for s in sorted(self.status)))

    @property
    def timestamp(self):
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self._timestamp))

    @classmethod
    def get(cls, uri):
        try:
            return cls.db[uri].status
        except KeyError:
            return None

    @classmethod
    def update(cls, uri):
        try:
            self = cls.db[uri]
        except KeyError:
            self = cls.db[uri] = cls(uri)
        self._timestamp = time.time()
        return self.status

    @classmethod
    def add(cls, uri, *codes):
        status = cls.update(uri)
        for code in codes:
            status.add(code)

    @classmethod
    def remove(cls, uri, *codes):
        if uri in cls.db:
            for code in codes:
                cls.db[uri].status.discard(code)

    @classmethod
    def test(cls, uri, code):
        return uri in cls.db and code in cls.db[uri].status


def install_object(obj):
    obj.obj.authenticated.add(authenticated)
    obj.obj.save()


class X509StoreCTX(rpki.POW.X509StoreCTX):

    @classmethod
    def subclass(cls, **kwargs):
        return type(cls.__name__, (cls,), kwargs)

    status = None

    def verify_callback(self, ok):
        err = self.getError()
        if err in (codes.X509_V_OK.code, codes.X509_V_ERR_SUBJECT_ISSUER_MISMATCH.code):
            return ok
        elif err == codes.X509_V_ERR_CRL_HAS_EXPIRED.code:
            return True
        elif err == codes.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT.code:
            self.status.add(codes.TRUST_ANCHOR_NOT_SELF_SIGNED)
            return ok
        else:
            self.status.add(codes.find(err))
            return ok


class X509(rpki.POW.X509):

    def __repr__(self):
        try:
            return "<X509 \"{}\" at 0x{:x}>".format(self.uri, id(self))
        except:
            return "<X509 at 0x{:x}>".format(id(self))

    @classmethod
    def store_if_new(cls, der, uri, retrieval):
        self = cls.derRead(der)
        aki  = self.getAKI()
        ski  = self.getSKI()
        return RPKIObject.objects.get_or_create(
            der  = der,
            defaults = dict(
                uri  = uri,
                aki  = "" if aki is None else aki.encode("hex"),
                ski  = "" if ski is None else ski.encode("hex"),
                sha256 = sha256(der).encode("hex"),
                retrieved = retrieval))

    @property
    def uri(self):
        return self.obj.uri

    @property
    def aki(self):
        return self.obj.aki

    @property
    def ski(self):
        return self.obj.ski

    @classmethod
    def load(cls, obj, cms = None):
        if cms is not None:
            # XXX Kludge to work around lack of subclass support in rpki.POW.CMS.certs().
            der = cms.certs()[0].derWrite()
        else:
            der = obj.der
        self = cls.derRead(der)
        self.obj = obj
        self.bc    = self.getBasicConstraints()
        self.eku   = self.getEKU()
        self.aia   = self.getAIA()
        self.sia   = self.getSIA()
        self.crldp = self.getCRLDP()
        self.is_ca = self.bc is not None and self.bc[0]
        self.caDirectory, self.rpkiManifest, self.signedObjectRepository, self.rpkiNotify \
                          = self.sia or (None, None, None, None)
        return self

    @staticmethod
    def count_uris(uris, scheme = "rsync://"):
        count = 0
        if uris is not None:
            for uri in uris:
                if uri.startswith(scheme):
                    count += 1
        return count

    def check(self, trusted, crl):
        #logger.debug("Starting checks for %r", self)
        status = Status.update(self.uri)
        is_ta = trusted is None
        is_routercert = (self.eku is not None and id_kp_bgpsec_router in self.eku and
                         not self.is_ca and self.uri.endswith(".cer"))
        if self.eku is not None and (self.is_ca or not self.uri.endswith(".cer")):
            status.add(codes.INAPPROPRIATE_EKU_EXTENSION)
        if is_ta and not self.is_ca:
            status.add(codes.MALFORMED_TRUST_ANCHOR)
        if is_ta and self.aia is not None:
            status.add(codes.AIA_EXTENSION_FORBIDDEN)
        if not is_ta and self.aia is None:
            status.add(codes.AIA_EXTENSION_MISSING)
        if is_routercert and self.sia is not None:
            status.add(codes.SIA_EXTENSION_FORBIDDEN)
        if not is_routercert and self.sia is None:
            status.add(codes.SIA_EXTENSION_MISSING)
        if is_ta and self.crldp is not None:
            status.add(codes.CRLDP_EXTENSION_FORBIDDEN)
        if not is_ta and self.crldp is None:
            status.add(codes.CRLDP_EXTENSION_MISSING)
        if not is_ta and not self.aki:
            status.add(codes.AKI_EXTENSION_MISSING)
        elif not is_ta and self.aki != trusted[0].ski:
            status.add(codes.AKI_EXTENSION_ISSUER_MISMATCH)
        serial = self.getSerial()
        if serial <= 0 or serial > 0x7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF:
            status.add(codes.BAD_CERTIFICATE_SERIAL_NUMBER)
        if self.getVersion() != 2:
            status.add(codes.WRONG_OBJECT_VERSION)
        n_rsync_caIssuers              = self.count_uris(self.aia)
        n_rsync_caDirectory            = self.count_uris(self.caDirectory)
        n_rsync_rpkiManifest           = self.count_uris(self.rpkiManifest)
        n_rsync_signedObjectRepository = self.count_uris(self.signedObjectRepository)
        if n_rsync_caIssuers > 1 or n_rsync_caDirectory > 1 or n_rsync_rpkiManifest > 1 or n_rsync_signedObjectRepository > 1:
            status.add(codes.MULTIPLE_RSYNC_URIS_IN_EXTENSION)
        if self.aia is not None and n_rsync_caIssuers == 0:
            status.add(codes.MALFORMED_AIA_EXTENSION)
        if self.is_ca:
            ok = n_rsync_caDirectory != 0 and n_rsync_rpkiManifest != 0 and n_rsync_signedObjectRepository == 0
        elif not is_routercert:
            ok = n_rsync_caDirectory == 0 and n_rsync_rpkiManifest == 0 and n_rsync_signedObjectRepository != 0
        else:
            ok = self.sia is None
        if not ok:
            status.add(codes.MALFORMED_SIA_EXTENSION)
        if not is_ta and self.count_uris(self.crldp) == 0:
            status.add(codes.MALFORMED_CRLDP_EXTENSION)
        self.checkRPKIConformance(status = status, eku = id_kp_bgpsec_router if is_routercert else None)
        try:
            self.verify(trusted = [self] if trusted is None else trusted, crl = crl, policy = "1.3.6.1.5.5.7.14.2",
                        context_class = X509StoreCTX.subclass(status = status))
        except rpki.POW.ValidationError as e:
            logger.debug("%r rejected: %s", self, e)
            status.add(codes.OBJECT_REJECTED)
        codes.normalize(status)
        #logger.debug("Finished checks for %r", self)
        return not any(s.kind == "bad" for s in status)


class CRL(rpki.POW.CRL):

    def __repr__(self):
        try:
            return "<CRL \"{}\" at 0x{:x}>".format(self.uri, id(self))
        except:
            return "<CRL at 0x{:x}>".format(id(self))

    @classmethod
    def store_if_new(cls, der, uri, retrieval):
        self = cls.derRead(der)
        aki  = self.getAKI()
        return RPKIObject.objects.get_or_create(
            der  = der,
            defaults = dict(
                uri  = uri,
                aki  = "" if aki is None else aki.encode("hex"),
                ski  = "",
                sha256 = sha256(der).encode("hex"),
                retrieved = retrieval))

    @property
    def uri(self):
        return self.obj.uri

    @property
    def aki(self):
        return self.obj.aki

    @property
    def ski(self):
        return ""

    @classmethod
    def load(cls, obj):
        self = cls.derRead(obj.der)
        self.obj = obj
        self.thisUpdate = self.getThisUpdate()
        self.nextUpdate = self.getNextUpdate()
        self.number     = self.getCRLNumber()
        return self

    def check(self, issuer):
        status = Status.update(self.uri)
        self.checkRPKIConformance(status = status, issuer = issuer)
        try:
            self.verify(issuer)
        except rpki.POW.ValidationError as e:
            logger.debug("%r rejected: %s", self, e)
            status.add(codes.OBJECT_REJECTED)
        codes.normalize(status)
        if self.getVersion() != 1:
            status.add(codes.WRONG_OBJECT_VERSION)
        now = rpki.sundial.now()
        if self.thisUpdate > now:
            status.add(codes.CRL_NOT_YET_VALID)
        if self.nextUpdate < now:
            status.add(codes.STALE_CRL_OR_MANIFEST)
        if self.number is None:
            status.add(codes.CRL_NUMBER_EXTENSION_MISSING)
        if self.number < 0:
            status.add(codes.CRL_NUMBER_IS_NEGATIVE)
        if self.number > 0x7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF:
            status.add(codes.CRL_NUMBER_OUT_OF_RANGE)
        if self.getIssuer() != issuer.getSubject():
            status.add(codes.CRL_ISSUER_NAME_MISMATCH)
        if not self.aki:
            status.add(codes.AKI_EXTENSION_MISSING)
        elif self.aki != issuer.ski:
            status.add(codes.AKI_EXTENSION_ISSUER_MISMATCH)

        return not any(s.kind == "bad" for s in status)


class CMS_Mixin(object):

    @classmethod
    def store_if_new(cls, der, uri, retrieval):
        self = cls.derRead(der)
        cert = self.certs()[0]
        aki  = cert.getAKI()
        ski  = cert.getSKI()
        return RPKIObject.objects.get_or_create(
            der  = der,
            defaults = dict(
                uri  = uri,
                aki  = "" if aki is None else aki.encode("hex"),
                ski  = "" if ski is None else ski.encode("hex"),
                sha256 = sha256(der).encode("hex"),
                retrieved = retrieval))

    @property
    def uri(self):
        return self.obj.uri

    @property
    def aki(self):
        return self.obj.aki

    @property
    def ski(self):
        return self.obj.ski


class Ghostbuster(rpki.POW.CMS, CMS_Mixin):

    def __repr__(self):
        try:
            return "<Ghostbuster \"{}\" at 0x{:x}>".format(self.uri, id(self))
        except:
            return "<Ghostbuster at 0x{:x}>".format(id(self))

    @classmethod
    def load(cls, obj):
        self        = cls.derRead(obj.der)
        self.obj    = obj
        self.ee     = X509.load(obj, self)
        self.vcard  = None
        return self

    def check(self, trusted, crl):
        status = Status.update(self.uri)
        self.ee.check(trusted = trusted, crl = crl)
        try:
            self.vcard = self.verify()
        except rpki.POW.ValidationError as e:
            logger.debug("%r rejected: %s", self, e)
            status.add(codes.OBJECT_REJECTED)
        self.checkRPKIConformance(status)
        codes.normalize(status)
        return not any(s.kind == "bad" for s in status)


class Manifest(rpki.POW.Manifest, CMS_Mixin):

    def __repr__(self):
        try:
            return "<Manifest \"{}\" at 0x{:x}>".format(self.uri, id(self))
        except:
            return "<Manifest at 0x{:x}>".format(id(self))

    @classmethod
    def load(cls, obj):
        self = cls.derRead(obj.der)
        self.obj = obj
        self.ee  = X509.load(obj, self)
        self.fah = None
        self.thisUpdate = None
        self.nextUpdate = None
        self.number     = None
        return self

    def check(self, trusted, crl):
        status = Status.update(self.uri)
        self.ee.check(trusted = trusted, crl = crl)
        try:
            self.verify()
        except rpki.POW.ValidationError as e:
            logger.debug("%r rejected: %s", self, e)
            status.add(codes.OBJECT_REJECTED)
        self.checkRPKIConformance(status)
        self.thisUpdate = self.getThisUpdate()
        self.nextUpdate = self.getNextUpdate()
        self.number     = self.getManifestNumber()
        self.fah        = self.getFiles()
        self.notBefore  = self.ee.getNotBefore()
        self.notAfter   = self.ee.getNotAfter()
        if self.thisUpdate < self.notBefore or self.nextUpdate > self.notAfter:
            status.add(codes.MANIFEST_INTERVAL_OVERRUNS_CERT)
        now = rpki.sundial.now()
        if self.thisUpdate > now:
            status.add(codes.MANIFEST_NOT_YET_VALID)
        if self.nextUpdate < now:
            status.add(codes.STALE_CRL_OR_MANIFEST)
        codes.normalize(status)
        return not any(s.kind == "bad" for s in status)

    def find_crl_candidate_hashes(self):
        for fn, digest in self.fah:
            if fn.endswith(".crl"):
                yield digest.encode("hex")


class ROA(rpki.POW.ROA, CMS_Mixin):

    def __repr__(self):
        try:
            return "<ROA \"{}\" at 0x{:x}>".format(self.uri, id(self))
        except:
            return "<ROA at 0x{:x}>".format(id(self))

    @classmethod
    def load(cls, obj):
        self = cls.derRead(obj.der)
        self.obj = obj
        self.ee         = X509.load(obj, self)
        self.asn        = None
        self.prefixes   = None
        return self

    def check(self, trusted, crl):
        status = Status.update(self.uri)
        self.ee.check(trusted = trusted, crl = crl)
        try:
            vcard = self.verify()
        except rpki.POW.ValidationError:
            status.add(codes.OBJECT_REJECTED)
        self.checkRPKIConformance(status)
        self.asn      = self.getASID()
        self.prefixes = self.getPrefixes()
        codes.normalize(status)
        return not any(s.kind == "bad" for s in status)


class_dispatch = dict(cer = X509,
                      crl = CRL,
                      gbr = Ghostbuster,
                      mft = Manifest,
                      roa = ROA)

def uri_to_class(uri):
    cls = class_dispatch.get(uri[-3:]) if len(uri) > 4 and uri[-4] == "." else None
    if cls is None:
        Status.add(uri, None, codes.UNKNOWN_OBJECT_TYPE_SKIPPED)
    return cls


# If we find ourselves using this same ordering for every retrieval from the RPKIObjects model, we
# can add it as a Meta option for the model and omit it in the query expressions, like this:
#
#    class RPKIObjects(models.Model):
#        ...
#        class Meta:
#            ordering = ["-retrieved__started"]
#
# https://docs.djangoproject.com/en/1.8/ref/models/querysets/#order-by
# https://docs.djangoproject.com/en/1.8/ref/models/options/#django.db.models.Options.ordering

def fetch_objects(**kwargs):
    for obj in RPKIObject.objects.filter(**kwargs).order_by("-retrieved__started"):
        cls = uri_to_class(obj.uri)
        if cls is not None:
            yield cls.load(obj)


class  WalkFrame(object):
    """
    Certificate tree walk stack frame.  This is basically just a
    preamble and a loop, broken out into several separate methods so
    that we can fork new tasks in the middle then resume processing of
    the current state machine (ie, this frame) when appropriate (eg,
    after an rsync or RRDP fetch completes).
    """

    def __init__(self, cer):
        self.cer = cer
        self.state = self.initial

    def __repr__(self):
        try:
            return "<WalkFrame \"{}\" at 0x{:x}".format(self.cer.uri, id(self))
        except:
            return "<WalkFrame at 0x{:x}>".format(id(self))

    @tornado.gen.coroutine
    def __call__(self, wsk):
        yield self.state(wsk)

    @tornado.gen.coroutine
    def initial(self, wsk):

        rsync_uri = first_rsync_uri(self.cer.caDirectory)
        rrdp_uri  = first_https_uri(self.cer.rpkiNotify)

        if args.prefer_rsync:
            uri = rsync_uri or rrdp_uri
        else:
            uri = rrdp_uri or rsync_uri 

        self.fetcher = Fetcher(uri)

        if not self.fetcher.needed():
            self.state = self.ready
        elif args.no_spawn_on_fetch:
            self.state = self.fetch
        else:
            self.state = self.fetch
            yield task_queue.put(wsk.clone())
            wsk.pop()

    @tornado.gen.coroutine
    def fetch(self, wsk):
        yield self.fetcher.fetch()
        self.state = self.ready

    @tornado.gen.coroutine
    def ready(self, wsk):
        self.trusted = wsk.trusted()

        #logger.debug("%r scanning products", self)

        # NB: CRL checks on manifest EE certificates deferred until we've picked a CRL.

        mft_candidates = []
        crl_candidates = []
        crl_candidate_hashes = set()

        for mft in fetch_objects(aki = self.cer.ski, uri__endswith = ".mft"):
            if mft.check(trusted = self.trusted, crl = None):
                mft_candidates.append(mft)
                crl_candidate_hashes.update(mft.find_crl_candidate_hashes())

        if not mft_candidates:
            wsk.pop()
            return

        for crl in fetch_objects(aki = self.cer.ski, uri__endswith = ".crl", sha256__in = crl_candidate_hashes):
            if crl.check(self.trusted[0]):
                crl_candidates.append(crl)

        mft_candidates.sort(reverse = True, key = lambda x: (x.number, x.thisUpdate, x.obj.retrieved.started))
        crl_candidates.sort(reverse = True, key = lambda x: (x.number, x.thisUpdate, x.obj.retrieved.started))

        if not crl_candidates:
            wsk.pop()
            return

        self.crl = crl_candidates[0]

        install_object(self.crl)
        Status.add(self.crl.uri, codes.OBJECT_ACCEPTED)

        #logger.debug("Picked CRL %s", self.crl.uri)

        for mft in mft_candidates:
            if self.crl.isRevoked(mft.ee):
                Status.add(mft.obj.uri, codes.MANIFEST_EE_REVOKED)
                continue
            self.mft = mft
            break
        else:
            wsk.pop()
            return

        install_object(self.mft)
        Status.add(self.mft.obj.uri, codes.OBJECT_ACCEPTED)

        self.stale_crl = Status.test(self.crl.uri, codes.STALE_CRL_OR_MANIFEST)
        self.stale_mft = Status.test(self.mft.uri, codes.STALE_CRL_OR_MANIFEST)

        # Issue warnings on mft and crl URI mismatches?

        # Use an explicit iterator so we can resume it; run loop in separate method, same reason.

        self.mft_iterator = iter(self.mft.getFiles())
        self.state        = self.loop

    @tornado.gen.coroutine
    def loop(self, wsk):

        #logger.debug("Processing %s", self.mft.uri)

        for fn, digest in self.mft_iterator:

            yield tornado.gen.moment

            uri = self.mft.uri[:self.mft.uri.rindex("/") + 1] + fn

            # Need general URI validator here?

            if uri == self.crl.uri:
                continue

            cls = uri_to_class(uri)

            if cls is None:
                continue

            if cls in (Manifest, CRL):
                Status.add(uri, None, codes.INAPPROPRIATE_OBJECT_TYPE_SKIPPED)
                continue

            for obj in fetch_objects(sha256 = digest.encode("hex")):
                
                # This can't happen as currently written, but we need to handle OBJECT_NOT_FOUND somehow,
                # this will do as a placeholder until we figure out how this should work.

                if obj is None:
                    Status.add(uri, codes.OBJECT_NOT_FOUND)
                    continue

                if self.stale_crl:
                    Status.add(uri, codes.TAINTED_BY_STALE_CRL)
                if self.stale_mft:
                    Status.add(uri, codes.TAINTED_BY_STALE_MANIFEST)

                if obj.check(trusted = self.trusted, crl = self.crl):
                    install_object(obj)
                    Status.add(uri, codes.OBJECT_ACCEPTED)
                    if cls is X509 and obj.is_ca:
                        wsk.push(obj)
                        return
                    break
                else:
                    Status.add(uri, codes.OBJECT_REJECTED)

        wsk.pop()


class WalkTask(object):
    """
    Task corresponding to one walk stack, roughly analgous to
    STACK_OF(walk_ctx_t) in rcynic:tos.
    """

    def __init__(self, wsk = None, cer = None):
        self.wsk = [] if wsk is None else wsk
        if cer is not None:
            self.push(cer)

    def __repr__(self):
        try:
            return "<WalkTask \"{}\" at 0x{:x}".format(self.wsk[-1].cer.uri, id(self))
        except:
            return "<WalkTask at 0x{:x}>".format(id(self))

    @tornado.gen.coroutine
    def __call__(self):
        while self.wsk:
            yield self.wsk[-1](wsk = self)

    def push(self, cer):
        self.wsk.append(WalkFrame(cer))

    def pop(self):
        return self.wsk.pop()

    def clone(self):
        return WalkTask(wsk = list(self.wsk))

    def trusted(self):
        stack = [w.cer for w in self.wsk]
        stack.reverse()
        return stack


def read_tals():
    for head, dirs, files in os.walk(args.tals):
        for fn in files:
            if fn.endswith(".tal"):
                furi = "file://" + os.path.abspath(os.path.join(head, fn))
                try:
                    with open(os.path.join(head, fn), "r") as f:
                        lines = [line.strip() for line in f]
                    blank = lines.index("")
                    uris = lines[:blank]
                    key  = rpki.POW.Asymmetric.derReadPublic("".join(lines[blank:]).decode("base64"))
                    if not uris or not all(uri.endswith(".cer") for uri in uris):
                        Status.add(furi, None, codes.MALFORMED_TAL_URI)
                    yield uris, key
                except:
                    Status.add(furi, None, codes.UNREADABLE_TRUST_ANCHOR_LOCATOR)


def uri_to_filename(uri, base = None):
    fn = uri[uri.index("://")+3:]
    if base is not None:
        fn = os.path.join(base, fn)
    return fn

def first_uri(uris, scheme):
    if uris is not None:
        for uri in uris:
            if uri.startswith(scheme):
                return uri
    return None

def first_rsync_uri(uris):
    return first_uri(uris, "rsync://")

def first_https_uri(uris):
    return first_uri(uris, "https://")

def sha256(bytes):
    d = rpki.POW.Digest(rpki.POW.SHA256_DIGEST)
    d.update(bytes)
    return d.digest()


class RRDP_ParseFailure(Exception):
    "Failure parsing RRDP message."

class DeadHost(Exception):
    "Host recently tried and known to be unavailable."


class Fetcher(object):
    """
    Network transfer methods and history database.

    At the moment this is rsync-only; eventually it will include
    support for HTTPS and RRDP.
    """

    # Internal protocol:
    #
    # - Instances which have just gotten to the query stage are not registered
    #
    # - Instances which are in progress are listed in .history and
    #   have a Condition object in .pending; instances which depend on
    #   this should wait for the condition, then return.
    #
    # - Instances which have completed are listed in .history and have
    # .pending set to None.

    _rsync_deadhosts = set()
    _rsync_history = dict()

    _https_deadhosts = set()
    _https_history = dict()

    def __init__(self, uri, ta = False):
        self.uri = uri
        self.ta  = ta
        self.pending = None
        self.status = None

    def _rsync_split_uri(self):
        return tuple(self.uri.rstrip("/").split("/")[2:])

    def _rsync_find(self, path):
        for i in xrange(1, len(path)):
            target = path[:i+1]
            try:
                return self._rsync_history[target]
            except KeyError:
                continue
        return None

    def needed(self):
        if args.no_fetch:
            return False
        if self.uri.startswith("rsync://"):
            return self._rsync_needed()
        if self.uri.startswith("https://"):
            return self._https_needed()
        raise ValueError

    def _rsync_needed(self):
        path = self._rsync_split_uri()
        if path[0] in self._rsync_deadhosts:
            return False
        entry = self._rsync_find(path)
        return entry is None or entry.pending is not None

    def _https_needed(self):
        netloc = urlparse.urlparse(self.uri).netloc
        if netloc in self._https_deadhosts:
            return False
        entry = self._https_history.get(self.uri)
        return entry is None or entry.pending is not None

    def fetch(self):
        if self.uri.startswith("rsync://"):
            return self._rsync_fetch()
        if self.uri.startswith("https://"):
            return self._https_fetch() if self.ta else self._rrdp_fetch()
        raise ValueError

    @tornado.gen.coroutine
    def _rsync_fetch(self):
        assert self.uri.startswith("rsync://") and (self.uri.endswith(".cer") if self.ta else self.uri.endswith("/"))

        if args.no_fetch:
            return
        path = self._rsync_split_uri()
        dead = path[0] in self._rsync_deadhosts
        other = self._rsync_find(path)
        if not dead and other is not None and other.pending is not None:
            yield other.pending.wait()
        if dead or other is not None:
            return

        self.pending = tornado.locks.Condition()
        self._rsync_history[path] = self

        try:
            path = uri_to_filename(self.uri, args.unauthenticated)
            cmd = ["rsync", "--update", "--times", "--copy-links", "--itemize-changes"]
            if self.uri.endswith("/"):
                cmd.append("--recursive")
                cmd.append("--delete")
            cmd.append(self.uri)
            cmd.append(path)

            dn = os.path.dirname(path)
            if not os.path.exists(dn):
                os.makedirs(dn)

            # We use the stdout close from rsync to detect when the  subprocess has finished.
            # There's a lovely tornado.process.Subprocess.wait_for_exit() method which does
            # exactly what one would think we'd want -- but Unix signal handling still hasn't
            # caught up to the software interrupt architecture ITS had forty years ago, so
            # signals still cause random "system call interrupted" failures in other libraries.
            # Nothing Tornado can do about this, so we avoid signals entirely and collect the
            # process exit status directly from the operating system.  In theory, the WNOHANG
            # isn't necessary here, we use it anyway to be safe in case theory is wrong.

            # If we need to add a timeout here to guard against rsync processes taking too long
            # (which has happened in the past with, eg, LACNIC), see tornado.gen.with_timeout()
            # (documented in the utility functions section of the tornado.gen page), which wraps
            # any future in a timeout.

            t0 = time.time()
            rsync = tornado.process.Subprocess(cmd, stdout = tornado.process.Subprocess.STREAM, stderr = subprocess.STDOUT)
            logger.debug("rsync[%s] started \"%s\"", rsync.pid, " ".join(cmd))
            output = yield rsync.stdout.read_until_close()
            pid, self.status = os.waitpid(rsync.pid, os.WNOHANG)
            t1 = time.time()
            if (pid, self.status) == (0, 0):
                logger.warn("rsync[%s] Couldn't get real exit status without blocking, sorry", rsync.pid)
            for line in output.splitlines():
                logger.debug("rsync[%s] %s", rsync.pid, line)
            logger.debug("rsync[%s] finished after %s seconds with status 0x%x", rsync.pid, t1 - t0, self.status)

            # Should do something with rsync result and validation status database here.

            retrieval = Retrieval.objects.create(
                uri        = self.uri,
                started    = rpki.sundial.datetime.fromtimestamp(t0),
                finished   = rpki.sundial.datetime.fromtimestamp(t1),
                successful = self.status == 0)

            for fn in self._rsync_walk(path):
                yield tornado.gen.moment
                uri = "rsync://" + fn[len(args.unauthenticated):].lstrip("/")
                cls = uri_to_class(uri)
                if cls is not None:
                    try:
                        with open(fn, "rb") as f:
                            cls.store_if_new(f.read(), uri, retrieval)
                    except:
                        Status.add(uri, codes.UNREADABLE_OBJECT)
                        logger.exception("Couldn't read %s from rsync tree", uri)

        finally:
            pending = self.pending
            self.pending = None
            pending.notify_all()

    def _rsync_walk(self, path):
        if self.uri.endswith("/"):
            for head, dirs, files in os.walk(path):
                for fn in files:
                    yield os.path.join(head, fn)
        elif os.path.exists(path):
            yield path

    @tornado.gen.coroutine
    def _https_fetch_url(self, url):

        if urlparse.urlparse(url).netloc in self._https_deadhosts:
            raise DeadHost

        # Should do something with deadhost processing below.  Looks
        # like errors such as HTTP timeout show up as
        # tornado.httpclient.HTTPError exceptions (which we could
        # suppress if we wanted to do so, but we probably don't).
        # HTTP timeout shows up in the logs as "HTTP 599".  See doc for:
        #
        #   tornado.httpclient.AsyncHTTPClient.fetch()
        #   tornado.httpclient.HTTPError

        try:
            ok = False
            t0 = time.time()
            client = tornado.httpclient.AsyncHTTPClient()
            response = yield client.fetch(url, validate_cert = False)
            # Might want to check response Content-Type here
            ok = True

        except tornado.httpclient.HTTPError as e:
            # Might want to check e.response here to figure out whether to add to _https_deadhosts.
            logger.info("HTTP error for %s: %s", url, e)
            raise

        except (socket.error, IOError, ssl.SSLError) as e:
            # Might want to check e.errno here to figure out whether to add to _https_deadhosts.
            logger.info("Network I/O error for %s: %s", url, e)
            raise

        except Exception as e:
            logger.exception("Error (%r) for %s", type(e), url)
            raise

        finally:
            t1 = time.time()
            logger.debug("Fetch of %s finished after %s seconds", url, t1 - t0)
            retrieval = Retrieval.objects.create(
                uri        = url,
                started    = rpki.sundial.datetime.fromtimestamp(t0),
                finished   = rpki.sundial.datetime.fromtimestamp(t1),
                successful = ok)

        raise tornado.gen.Return((retrieval, response.body if ok else None))

    @tornado.gen.coroutine
    def _https_fetch(self):

        if args.no_fetch:
            return

        other = self._https_history.get(self.uri)
        if other is not None and other.pending is not None:
            yield other.pending.wait()
            return

        self.pending = tornado.locks.Condition()
        self._rsync_history[self.uri] = self

        try:
            retrieval, der = yield self._https_fetch_url(self.uri)
            X509.store_if_new(der, self.uri, retrieval)
        except:
            logger.exception("Couldn't load %s", self.uri)

        finally:
            pending = self.pending
            self.pending = None
            pending.notify_all()

    @tornado.gen.coroutine
    def _rrdp_fetch_url(self, url, tag, hash = None, session_id = None, serial = None):
        
        retrieval, xml = yield self._https_fetch_url(url)

        if hash is not None and sha256(xml).encode("hex") != hash.lower():
            raise RRDP_ParseFailure("Expected RRDP hash %s for %s, got %s" % (
                hash.lower(), url, sha256(xml).encode("hex")))

        xml = XML(xml)
        rpki.relaxng.rrdp.schema.assertValid(xml)

        if tag is not None and xml.tag != rpki.relaxng.rrdp.xmlns + tag:
            raise RRDP_ParseFailure("Expected RRDP %s for %s, got %s" % (
                tag, delta_url, xml.tag))

        if session_id is not None and xml.get("session_id") != session_id:
            raise RRDP_ParseFailure("Expected RRDP session_id %s for %s, got %s" % (
                session_id, url, xml.get("session_id")))

        if serial is not None and long(xml.get("serial")) != long(serial):
            raise RRDP_ParseFailure("Expected RRDP serial %s for %s, got %s" % (
                serial, url, xml.get("serial")))
        
        raise tornado.gen.Return((retrieval, xml))

    @tornado.gen.coroutine
    def _rrdp_fetch(self):
        from django.db import transaction

        if args.no_fetch:
            return

        other = self._https_history.get(self.uri)
        if other is not None and other.pending is not None:
            yield other.pending.wait()
            return

        self.pending = tornado.locks.Condition()
        self._https_history[self.uri] = self

        try:
            retrieval, notification = yield self._rrdp_fetch_url(url = self.uri, tag = "notification")

            session_id = notification.get("session_id")
            serial = long(notification.get("serial"))

            snapshot = RRDPSnapshot.objects.filter(
                session_id = session_id).order_by("-retrieved__started").first()

            logger.debug("RRDP notification for %s session_id %s serial %s current snapshot %r",
                         self.uri, session_id, serial, snapshot)

            if snapshot is not None and snapshot.serial == serial:
                logger.debug("RRDP data for %s is up-to-date, nothing to do", self.uri)
                return
            
            delta_serials = dict((long(delta.get("serial")), (delta.get("uri"), delta.get("hash")))
                                 for delta in notification.iterchildren(rpki.relaxng.rrdp.xmlns + "delta"))

            if snapshot is None or snapshot.serial + 1 not in delta_serials:

                if snapshot is not None:
                    logger.debug("RRDP %s no deltas available for serial %s", self.uri, snapshot.serial)

                x = notification.find(rpki.relaxng.rrdp.xmlns + "snapshot")

                url, hash = x.get("uri"), x.get("hash")

                logger.debug("RRDP %s loading from snapshot %s serial %s", self.uri, url, serial)

                retrieval, x = yield self._rrdp_fetch_url(
                    url        = url,
                    hash       = hash,
                    tag        = "snapshot",
                    session_id = session_id,
                    serial     = serial)

                with transaction.atomic():
                    if snapshot is not None:
                        snapshot.delete()
                    snapshot = RRDPSnapshot.objects.create(retrieved = retrieval, session_id = session_id, serial = serial)
                    for x in x.iterchildren(rpki.relaxng.rrdp.xmlns + "publish"):
                        uri = x.get("uri")
                        cls = uri_to_class(uri)
                        if cls is None:
                            raise RRDP_ParseFailure("Unexpected URI %s" % uri)
                        obj, created = cls.store_if_new(x.text.decode("base64"), uri, retrieval)
                        obj.snapshot.add(snapshot)
                        obj.save()
                    logger.debug("RRDP %s committing snapshot %s serial %s", self.uri, url, serial)

            else:
                logger.debug("RRDP %s %s deltas (%s--%s)", self.uri, 
                             (serial - snapshot.serial), snapshot.serial, serial)

                for serial in xrange(snapshot.serial + 1, serial + 1):

                    url, hash = delta_serials[serial]

                    logger.debug("RRDP %s fetching %s serial %s", self.uri, url, serial)

                    retrieval, delta = yield self._rrdp_fetch_url(
                        url        = url,
                        hash       = hash,
                        tag        = "delta",
                        session_id = session_id,
                        serial     = serial)

                    logger.debug("RRDP %s loading delta %s serial %s", self.uri, url, serial)

                    with transaction.atomic():
                        snapshot.serial = serial
                        snapshot.save()

                        for x in delta.iterchildren(rpki.relaxng.rrdp.xmlns + "withdraw"):
                            snapshot.rpkiobject_set.remove(
                                snapshot.rpkiobject_set.get(sha256 = x.get("hash").lower()))
                            snapshot.save()

                        for x in delta.iterchildren(rpki.relaxng.rrdp.xmlns + "publish"):
                            uri = x.get("uri")
                            cls = uri_to_class(uri)
                            if cls is None:
                                raise RRDP_ParseFailure("Unexpected URI %s" % uri)
                            if x.get("hash", None) is not None:
                                snapshot.rpkiobject_set.remove(
                                    snapshot.rpkiobject_set.get(sha256 = x.get("hash").lower()))
                                snapshot.save()
                            obj, created = cls.store_if_new(x.text.decode("base64"), uri, retrieval)
                            obj.snapshot.add(snapshot)
                            obj.save()

                        logger.debug("RRDP %s committing delta %s serial %s", self.uri, url, serial)

                logger.debug("RRDP %s done processing deltas", self.uri)

        except (tornado.httpclient.HTTPError, socket.error, IOError, ssl.SSLError):
            pass                        # Already logged

        except RRDP_ParseFailure as e:
            logger.info("RRDP parse failure: %s", e)

        except:
            logger.exception("Couldn't load %s", self.uri)

        finally:
            pending = self.pending
            self.pending = None
            pending.notify_all()


class CheckTALTask(object):

    def __init__(self, uris, key):
        rsync_uri = first_rsync_uri(uris)
        https_uri = first_https_uri(uris)

        if args.prefer_rsync:
            self.uri = rsync_uri or https_uri
        else:
            self.uri = https_uri or rsync_uri 

        self.key = key

    def __repr__(self):
        return "<CheckTALTask: \"{}\">".format(self.uri)

    @tornado.gen.coroutine
    def __call__(self):
        yield Fetcher(self.uri, ta = True).fetch()
        for cer in fetch_objects(uri = self.uri):
            if self.check(cer):
                yield task_queue.put(WalkTask(cer = cer))
                break
        else:
            Status.add(self.uri, codes.TRUST_ANCHOR_SKIPPED)

    def check(self, cer):
        if self.key.derWritePublic() != cer.getPublicKey().derWritePublic():
            Status.add(self.uri, codes.TRUST_ANCHOR_KEY_MISMATCH)
            ok = False
        else:
            ok = cer.check(trusted = None, crl = None)
        if ok:
            install_object(cer)
            Status.add(self.uri, codes.OBJECT_ACCEPTED)
        else:
            Status.add(self.uri, codes.OBJECT_REJECTED)
        return ok


@tornado.gen.coroutine
def worker(meself):
    #
    # NB: This particular style of control loop REQUIRES an except
    #     clause, even if that except clause is just a pass statement.
    #
    while True:
        task = yield task_queue.get()
        try:
            logger.debug("Worker %s starting %s, queue length %s", meself, task, task_queue.qsize())
            yield task()
        except:
            logger.exception("Worker %s caught unhandled exception from %s", meself, task)
        finally:
            task_queue.task_done()
            logger.debug("Worker %s finished %s, queue length %s", meself, task, task_queue.qsize())


def final_report():
    # Clean up a bit to avoid confusing the user unnecessarily.
    for s in Status.db.itervalues():
        if codes.OBJECT_ACCEPTED in s.status:
            s.status.discard(codes.OBJECT_REJECTED)
    doc = Element("rcynic-summary") # rcynic-version = "", summary-version = "", reporting-hostname = ""
    labels = SubElement(doc, "labels")
    for code in codes.all():
        SubElement(labels, code.name).text = code.text
    for uri in Status.db:
        for sym in sorted(Status.db[uri].status):
            SubElement(doc, "validation_status",
                       timestamp  = str(Status.db[uri].timestamp),
                       status     = str(sym),
                       generation = "None"              # Historical relic, remove eventually
                       ).text     = uri
    #
    # Should generate <rsync_history/> elements here too, later
    #
    ElementTree(doc).write(file = args.xml_file, pretty_print = True)


def final_cleanup():
    from django.db import transaction, models

    def report(when):
        logger.debug("Database %s cleanup: %s Authenticated %s RRDPSnapshot %s RPKIObject %s Retrieval", when,
                     Authenticated.objects.all().count(), RRDPSnapshot.objects.all().count(),
                     RPKIObject.objects.all().count(), Retrieval.objects.all().count())

    report("before")

    with transaction.atomic():

        logger.debug("Flushing old authenticated sets")

        q = Authenticated.objects
        q = q.exclude(id = authenticated.id)
        q.delete()

        logger.debug("Flushing RRDP snapshots which don't contain anything in the (remaining) authenticated set")

        # We clean the RRDPSnapshot table in two steps.  First step is to remove snapshots which aren't related
        # to any RPKI object in the current authenticated set.

        q = RPKIObject.objects
        q = q.filter(authenticated = authenticated.id)
        q = q.exclude(snapshot = None)
        q = q.order_by("snapshot__id")
        q = q.values_list("snapshot__id", flat = True)
        q = q.distinct()
        q = RRDPSnapshot.objects.exclude(id__in = q)
        q.delete()

        # This is still too many snapshots, because an RPKI object can be (and usually are) present in more than
        # one snapshot for a given session.  So we still have to do something to prune out old snapshots from
        # sessions we're keeping.  I don't know a sane way to do this via the ORM as a single SQL query, so
        # we do it as a query followed by a loop.  If this is too slow, we can try doing it as raw SQL,
        # but that has portability issues.

        # well, maybe we want the raw sql interface after all, as the query is fairly simple:
        #
        # DELETE FROM rcynicdb_rrdpsnapshot
        # WHERE id IN
        #   (SELECT t2.id
        #    FROM (SELECT session_id, max(serial) AS max_serial FROM rcynicdb_rrdpsnapshot GROUP BY session_id) AS t1
        #    JOIN rcynicdb_rrdpsnapshot AS t2
        #    ON t1.session_id = t2.session_id
        #    WHERE serial < max_serial);
        #
        # Or, equivalently but slightly easier to read:
        #
        # DELETE FROM rcynicdb_rrdpsnapshot
        # WHERE id IN
        #   (SELECT id
        #    FROM (SELECT session_id, max(serial) AS max_serial FROM rcynicdb_rrdpsnapshot GROUP BY session_id) AS filter
        #    NATURAL JOIN rcynicdb_rrdpsnapshot
        #    WHERE serial < max_serial);
        #
        # See
        #  http://stackoverflow.com/questions/2317686/joining-2-sql-select-result-sets-into-one
        #  https://docs.djangoproject.com/en/1.8/topics/db/sql/#executing-custom-sql-directly

        # Except that this shouldn't be necessary after all, now that we reuse snapshot objects in place.

        if False:
            q = RRDPSnapshot.objects
            q = q.values("session_id")
            q = q.annotate(max_serial = models.Max("serial"))
            q = q.values_list("session_id", "max_serial")
            logger.debug("Annotation query for RRDPSnapshots gave us %r", list(q))
            for u, s in q:
                RRDPSnapshot.objects.filter(session_id = u, serial__lt = s).delete()

        logger.debug("Flushing RPKI objects which are in neither current authenticated set nor current RRDP snapshot")

        q = RPKIObject.objects
        q = q.exclude(authenticated = authenticated.id)
        q = q.filter(snapshot = None)
        q.delete()

        logger.debug("Flushing retrieval objects which are no longer related to any RPKI objects or RRDP snapshot")

        q = RPKIObject.objects
        q = q.order_by("retrieved__id")
        q = q.values_list("retrieved__id", flat = True)
        q = q.distinct()
        q = Retrieval.objects.exclude(id__in = q)
        q = q.filter(rrdpsnapshot = None)
        q.delete()

    report("after")


@tornado.gen.coroutine
def launcher():
    for i in xrange(args.workers):
        tornado.ioloop.IOLoop.current().spawn_callback(worker, i)

    yield [task_queue.put(CheckTALTask(uris, key)) for uris, key in read_tals()]
    yield task_queue.join()


class posint(int):
    def __init__(self, value):
        if self <= 0:
            raise ValueError


def main():
    global rpki

    os.environ.update(TZ = "UTC",
                      DJANGO_SETTINGS_MODULE = "rpki.django_settings.rcynic")
    time.tzset()

    parser = argparse.ArgumentParser(description = __doc__)

    parser.add_argument("-c", "--config", help = "override default location of configuration file")

    parser.add_argument("--authenticated",      default = "rcynic-data/authenticated")
    parser.add_argument("--unauthenticated",    default = "rcynic-data/unauthenticated")
    parser.add_argument("--xml-file",           default = "rcynicng.xml",       type = argparse.FileType("w"))

    parser.add_argument("--tals",               default = "sample-trust-anchors")

    parser.add_argument("--workers",            type = posint, default = 10)

    parser.add_argument("--no-fetch",           action = "store_true")
    parser.add_argument("--no-spawn-on-fetch",  action = "store_true")
    parser.add_argument("--no-migrate",         action = "store_true")

    parser.add_argument("--prefer-rsync",       action = "store_true")

    global args
    args = parser.parse_args()

    rpki.config.rpki_conf_envname = "RCYNIC_CONF"
    rpki.config.default_filename = os.path.join(rpki.autoconf.sysconfdir, "rcynic.conf")
    rpki.config.parser(set_filename = args.config, section = "rcynic")

    # If we're going to allow some options to be controlled from both
    # config file and command line, this is where we sort that out.
    # Doesn't play well with action=store_true.

    import django
    django.setup()

    if not args.no_migrate:
        # Not sure we should be doing this on every run, but sure simplifies things.
        import django.core.management
        django.core.management.call_command("migrate", verbosity = 0, interactive = False)

    import rpki.rcynicdb
    global Retrieval
    global Authenticated
    global RRDPSnapshot
    global RPKIObject
    Retrieval     = rpki.rcynicdb.models.Retrieval
    Authenticated = rpki.rcynicdb.models.Authenticated
    RRDPSnapshot  = rpki.rcynicdb.models.RRDPSnapshot
    RPKIObject    = rpki.rcynicdb.models.RPKIObject

    logging.basicConfig(level = logging.DEBUG, format = "%(asctime)s  %(message)s", datefmt = "%Y-%m-%d %H:%M:%S")

    global authenticated
    authenticated = Authenticated.objects.create(started  = rpki.sundial.datetime.now())

    global task_queue
    task_queue = tornado.queues.Queue()
    tornado.ioloop.IOLoop.current().run_sync(launcher)

    authenticated.finished = rpki.sundial.datetime.now()
    authenticated.save()

    final_report()

    final_cleanup()


if __name__ == "__main__":
    main()