path: root/rp/rcynic/rcynicng

#!/usr/bin/env python

# $Id$

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

import os
import sys
import time
import shutil
import logging
import argparse
import subprocess

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

import rpki.POW
import rpki.sundial

from rpki.oids import id_kp_bgpsec_router

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

logger = logging.getLogger("rcynicng")

codes = rpki.POW.validation_status

class Generation(object):

    all = []

    def __init__(self, name, tree):
        self.name = name
        self.tree = tree
        self.all.append(self)
        self.pos = len(self.all)
        setattr(self.__class__, name, self)

    def __hash__(self):
        return hash(self.name)

    def __cmp__(self, other):
        return cmp(self.pos, 0 if other is None else other.pos)

    def __str__(self):
        return self.name


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, generation):
        assert generation is None or isinstance(generation, Generation)
        self.uri = uri
        self._generation = generation
        self._timestamp = None
        self.status = set()

    def __str__(self):
        return "{my.timestamp} {my.uri} {status} {my.generation}".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))

    @property
    def generation(self):
        return str(self._generation)

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

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

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

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


def install_object(obj):
    fn = uri_to_filename(obj.uri, new_authenticated)
    dn = os.path.dirname(fn)
    #logger.debug("Installing %r by linking %s to %s", obj, obj.fn, fn)
    if not os.path.isdir(dn):
        os.makedirs(dn)
    os.link(obj.fn, fn)


def final_install():

    real_old = os.path.realpath(old_authenticated).rstrip("/")
    real_new = os.path.realpath(new_authenticated).rstrip("/")

    fn = args.authenticated.rstrip("/") + ".new"
    logger.debug("Symlinking %s to %s", os.path.basename(real_new), args.authenticated)
    if os.path.exists(fn):
        os.unlink(fn)
    os.symlink(os.path.basename(real_new), fn)
    os.rename(fn, args.authenticated)

    if os.path.isdir(real_old):
        fn = args.authenticated.rstrip("/") + ".old"
        logger.debug("Symlinking %s to %s", os.path.basename(real_old), fn)
        if os.path.exists(fn):
            os.unlink(fn)
        os.symlink(os.path.basename(real_old), fn)

    dn = os.path.dirname(args.authenticated.rstrip("/"))
    for fn in os.listdir(dn):
        fn = os.path.join(dn, fn)
        if fn.startswith(args.authenticated.rstrip("/") + ".") and os.path.realpath(fn) not in (real_new, real_old):
            logger.debug("Removing %s", fn)
            shutil.rmtree(fn)


class X509(rpki.POW.X509):

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

    @classmethod
    def derReadURI(cls, uri, generation, cms = None):
        fn = uri_to_filename(uri, generation.tree)
        if not os.path.exists(fn):
            Status.add(uri, generation, codes.OBJECT_NOT_FOUND)
            return None
        if cms is None:
            with open(fn, "rb") as f:
                der = f.read()
        else:
            # XXX awful kludge to work around current lack of subclass
            # support in rpki.POW.CMS.certs().
            der = cms.certs()[0].derWrite()
        self = cls.derRead(der)
        self.uri = uri
        self.fn  = fn
        self.generation = generation
        self.sha256 = sha256(der) if cms is None else None
        self.bc    = self.getBasicConstraints()
        self.aki   = self.getAKI()
        self.ski   = self.getSKI()
        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):
        status = Status.update(self.uri, self.generation)
        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)
        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)
        try:
            self.verify(trusted = [self] if trusted is None else trusted, crl = crl, status = status)
        except rpki.POW.ValidationError as e:
            logger.debug("%r rejected: %s", self, e)
            status.add(codes.OBJECT_REJECTED)
        codes.normalize(status)
        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, self.generation, id(self))
        except:
            return "<CRL at 0x{:x}>".format(id(self))

    @classmethod
    def derReadURI(cls, uri, generation):
        fn = uri_to_filename(uri, generation.tree)
        if not os.path.exists(fn):
            Status.add(uri, generation, codes.OBJECT_NOT_FOUND)
            return None
        with open(fn, "rb") as f:
            der = f.read()
        self = cls.derRead(der)
        self.uri = uri
        self.fn  = fn
        self.generation = generation
        self.sha256     = sha256(der)
        self.aki        = self.getAKI()
        self.thisUpdate = self.getThisUpdate()
        self.nextUpdate = self.getNextUpdate()
        self.number     = self.getCRLNumber()
        return self

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


class Ghostbuster(rpki.POW.CMS):

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

    @classmethod
    def derReadURI(cls, uri, generation):
        fn = uri_to_filename(uri, generation.tree)
        if not os.path.exists(fn):
            Status.add(uri, generation, codes.OBJECT_NOT_FOUND)
            return None
        with open(fn, "rb") as f:
            der = f.read()
        self = cls.derRead(der)
        self.uri = uri
        self.fn  = fn
        self.generation = generation
        self.sha256 = sha256(der)
        self.ee     = X509.derReadURI(uri, generation, self)
        self.vcard  = None
        return self

    def check(self, trusted, crl):
        status = Status.update(self.uri, self.generation)
        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)
        codes.normalize(status)
        return not any(s.kind == "bad" for s in status)


class Manifest(rpki.POW.Manifest):

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

    @classmethod
    def derReadURI(cls, uri, generation):
        fn = uri_to_filename(uri, generation.tree)
        if not os.path.exists(fn):
            Status.add(uri, generation, codes.OBJECT_NOT_FOUND)
            return None
        with open(fn, "rb") as f:
            der = f.read()
        self = cls.derRead(der)
        self.uri = uri
        self.fn  = fn
        self.ee  = X509.derReadURI(uri, generation, self)
        self.fah = None
        self.generation = generation
        self.thisUpdate = None
        self.nextUpdate = None
        self.number     = None
        return self

    def check(self, trusted, crl):
        status = Status.update(self.uri, self.generation)
        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.thisUpdate = self.getThisUpdate()
        self.nextUpdate = self.getNextUpdate()
        self.number     = self.getManifestNumber()
        self.fah        = self.getFiles()
        codes.normalize(status)
        return not any(s.kind == "bad" for s in status)

    def find_crl_uris(self):
        diruri = self.uri[:self.uri.rindex("/") + 1]
        for fn, digest in self.fah:
            if fn.endswith(".crl"):
                yield diruri + fn, digest


class ROA(rpki.POW.ROA):

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

    @classmethod
    def derReadURI(cls, uri, generation):
        fn = uri_to_filename(uri, generation.tree)
        if not os.path.exists(fn):
            Status.add(uri, generation, codes.OBJECT_NOT_FOUND)
            return None
        with open(fn, "rb") as f:
            der = f.read()
        self = cls.derRead(der)
        self.uri = uri
        self.fn  = fn
        self.generation = generation
        self.sha256     = sha256(der)
        self.ee         = X509.derReadURI(uri, generation, self)
        self.asn        = None
        self.prefixes   = None
        return self

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


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):
        self.diruri  = first_rsync_uri(self.cer.caDirectory)
        self.fetcher = Fetcher(self.diruri)

        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)

        mft_uri = first_rsync_uri(self.cer.rpkiManifest)

        crl_candidates = []

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

        current_mft = Manifest.derReadURI(mft_uri, Generation.current)
        if current_mft is not None and current_mft.check(trusted = self.trusted, crl = None):
            crl_candidates.extend(current_mft.find_crl_uris())
        else:
            current_mft = None

        backup_mft  = Manifest.derReadURI(mft_uri, Generation.backup)
        if backup_mft is not None and backup_mft.check(trusted = self.trusted, crl = None):
            crl_candidates.extend(backup_mft.find_crl_uris())
        else:
            backup_mft = None
            Status.remove(mft_uri, Generation.backup, codes.OBJECT_NOT_FOUND)

        if current_mft is None and backup_mft is None:
            wsk.pop()
            return

        crls = {}
        for uri, digest in crl_candidates:
            for generation in (Generation.current, Generation.backup):
                key = (uri, generation)
                if key not in crls:
                    crls[key] = CRL.derReadURI(uri, generation)

        self.crl = None
        for uri, digest in crl_candidates:
            for generation in (Generation.current, Generation.backup):
                crl = crls[uri, generation]
                if crl == self.crl:
                    continue
                if crl is None and generation == Generation.backup:
                    Status.remove(uri, generation, codes.OBJECT_NOT_FOUND)
                if crl is None:
                    continue
                if crl.sha256 != digest:
                    #Status.add(uri, generation, codes.DIGEST_MISMATCH)
                    continue
                if not crl.check(self.trusted[0]) or (self.crl is not None and crl.number < self.crl.number):
                    continue
                if self.crl is None or crl.number > self.crl.number or crl.thisUpdate > self.crl.thisUpdate:
                    self.crl = crl

        if self.crl is None:
            wsk.pop()
            return

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

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

        if current_mft is not None and self.crl.isRevoked(current_mft.ee):
            Status.add(current_mft.uri, current_mft.generation, codes.MANIFEST_EE_REVOKED)
            current_mft = None

        if backup_mft is not None and self.crl.isRevoked(backup_mft.ee):
            Status.add(backup_mft.uri, backup_mft.generation, codes.MANIFEST_EE_REVOKED)
            backup_mft = None

        if current_mft is not None:
            self.mft = current_mft
        elif backup_mft is not None:
            self.mft = backup_mft
        else:
            wsk.pop()
            return

        install_object(self.mft)
        Status.add(mft_uri, self.mft.generation, codes.OBJECT_ACCEPTED)

        now = rpki.sundial.now()

        self.stale_crl = now > self.crl.nextUpdate
        if self.stale_crl:
            logger.debug("Stale CRL %s %s", self.crl.nextUpdate, self.crl.uri)
            Status.add(self.crl.uri, self.crl.generation, codes.STALE_CRL_OR_MANIFEST)            

        self.stale_mft = now > self.mft.nextUpdate
        if self.stale_mft:
            logger.debug("Stale Manifest %s %s", self.mft.nextUpdate, self.mft.uri)
            Status.add(self.mft.uri, self.mft.generation, codes.STALE_CRL_OR_MANIFEST)

        # Use an explicit iterator so we can resume it later.
        # Run the loop in a separate method for the same reason.

        self.mft_iterator = iter(self.mft.getFiles())
        self.generation   = Generation.current
        self.state        = self.loop

    fns2 = dict(cer = X509, gbr = Ghostbuster, roa = ROA)

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

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

        for fn, digest in self.mft_iterator:

            yield tornado.gen.moment

            uri = self.diruri + fn
            cls = self.fns2.get(uri[-3:])

            # Need general URI validator here?

            if uri == self.crl.uri:
                continue

            if self.generation is Generation.backup and Status.test(uri, Generation.current, codes.OBJECT_ACCEPTED):
                #logger.debug("Current version of %s already accepted, skipping", uri)
                continue

            if uri[-4] != "." or cls is None:
                Status.add(uri, self.generation, codes.UNKNOWN_OBJECT_TYPE_SKIPPED)
                continue

            if self.stale_crl:
                logger.debug("%s tainted by stale CRL", uri)
                Status.add(uri, self.generation, codes.TAINTED_BY_STALE_CRL)

            if self.stale_mft:
                logger.debug("%s tainted by stale manifest", uri)
                Status.add(uri, self.generation, codes.TAINTED_BY_STALE_MANIFEST)

            obj = cls.derReadURI(uri, self.generation)

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

            ok  = obj.check(trusted = self.trusted, crl = self.crl)

            if obj.sha256 != digest:
                Status.add(uri, generation, codes.DIGEST_MISMATCH)
                ok = False

            if ok:
                install_object(obj)
                Status.add(uri, self.generation, codes.OBJECT_ACCEPTED)
            else:
                Status.add(uri, self.generation, codes.OBJECT_REJECTED)

            if ok and cls is X509 and obj.is_ca:
                wsk.push(obj)
                return

        if self.generation is Generation.current:
            self.mft_iterator = iter(self.mft.getFiles())
            self.generation   = Generation.backup
        else:
            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 root, dirs, files in os.walk(args.tals):
        for fn in files:
            if fn.endswith(".tal"):
                furi = "file://" + os.path.abspath(os.path.join(root, fn))
                try:
                    with open(os.path.join(root, fn), "r") as f:
                        lines = f.readlines()
                    uri = lines.pop(0).strip()
                    b64 = "".join(lines[lines.index("\n"):])
                    key = rpki.POW.Asymmetric.derReadPublic(b64.decode("base64"))
                    if not uri.endswith(".cer"):
                        Status.add(furi, None, codes.MALFORMED_TAL_URI)
                    yield uri, 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 sha256(bytes):
    d = rpki.POW.Digest(rpki.POW.SHA256_DIGEST)
    d.update(bytes)
    return d.digest()


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

    def __init__(self, uri):
        self.uri = uri
        self.pending = None
        self.status = None
        self.runtime = 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()
        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 fetch(self):
        if self.uri.startswith("rsync://"):
            return self._rsync_fetch()
        raise ValueError

    @tornado.gen.coroutine
    def _rsync_fetch(self):
        assert self.uri.startswith("rsync://") and (self.uri.endswith(".cer") or 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:
            cmd = ["rsync", "--update", "--times", "--copy-links", "--itemize-changes"]
            if self.uri.endswith("/"):
                cmd.append("--recursive")
                cmd.append("--delete")
            cmd.append(self.uri)
            cmd.append(uri_to_filename(self.uri, args.unauthenticated))

            dn = os.path.dirname(cmd[-1])
            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.

            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()
            self.runtime = t1 - t0
            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, self.runtime, self.status)

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

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


class CheckTALTask(object):

    def __init__(self, uri, key):
        self.uri = uri
        self.key = key

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

    @tornado.gen.coroutine
    def __call__(self):
        yield Fetcher(self.uri).fetch()
        if self.check(Generation.current):
            yield task_queue.put(WalkTask(cer = self.cer))
        elif self.check(Generation.backup):
            yield task_queue.put(WalkTask(cer = self.cer))
        else:
            Status.add(self.uri, None, codes.TRUST_ANCHOR_SKIPPED)

    def check(self, generation):
        self.cer = X509.derReadURI(self.uri, generation)
        ok = False
        if self.cer is None:
            Status.add(self.uri, generation, codes.UNREADABLE_TRUST_ANCHOR)
        elif self.key.derWritePublic() != self.cer.getPublicKey().derWritePublic():
            Status.add(self.uri, generation, codes.TRUST_ANCHOR_KEY_MISMATCH)
        else:
            ok = self.cer.check(trusted = None, crl = None)
        if ok:
            install_object(self.cer)
            Status.add(self.uri, generation, codes.OBJECT_ACCEPTED)
        else:
            Status.add(self.uri, generation, 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)
        if s.generation is Generation.backup:
            if Status.test(s.uri, Generation.current, codes.OBJECT_ACCEPTED):
                s.status.discard(codes.OBJECT_REJECTED)
            s.status.discard(codes.OBJECT_NOT_FOUND)
    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, generation in Status.db:
        for sym in sorted(Status.db[uri, generation].status):
            SubElement(doc, "validation_status",
                       timestamp  = str(Status.db[uri, generation].timestamp),
                       status     = str(sym),
                       generation = str(generation)
                       ).text     = uri
    #
    # Should generate <rsync_history/> elements here too, later
    #
    ElementTree(doc).write(file = args.xml_file, pretty_print = True)


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

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


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


def main():
    os.putenv("TZ", "UTC")
    time.tzset()

    parser = argparse.ArgumentParser(description = __doc__)

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

    global args
    args = parser.parse_args()

    global new_authenticated, old_authenticated
    new_authenticated = args.authenticated.rstrip("/") + time.strftime(".%Y-%m-%dT%H:%M:%SZ")
    old_authenticated = args.authenticated

    Generation("current", args.unauthenticated)
    Generation("backup",  old_authenticated)

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

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

    final_report()

    final_install()


if __name__ == "__main__":
    main()