UmbrelApps/tests/test_agent_certificates.py

"""Toetst de certificaatlezer en de keuze-guard van de agent.

Waarom deze test bestaat: de agent leest een X.509-certificaat met een eigen
DER-lezer, omdat de standaardbibliotheek er geen heeft en de app bij het starten
niets bijinstalleert. Zo'n lezer faalt niet met een foutmelding maar met een
verkeerd antwoord, en een certificaatdatum die er een jaar naast zit valt nooit
op. Daarom wordt hij hier vergeleken met `ssl`, dat een onafhankelijke
implementatie is.

Draaien:

    python tests/test_agent_certificates.py

De test laadt `tools/electrum-gate/agent.py` rechtstreeks: het bestand dat de
Dockerfile in de image zet. Tot 0.1.0 heette dat `agent.py.template` en stond het
in de app-map; de eerste toets hieronder is uit die tijd en blijft staan, want
het script haalt zijn instellingen nog altijd uit de omgeving en niet uit een
ingevulde constante.
"""

import sys

# Vóór de imports, want anders is het te laat: Python legt bytecode naast agent.py
# zodra die geïmporteerd wordt, en die rommel hoort niet in de repo. Een keer is
# zo'n .pyc meegegaan in een commit.
sys.dont_write_bytecode = True

import importlib.machinery  # noqa: E402
import importlib.util  # noqa: E402
import os  # noqa: E402
import shutil  # noqa: E402
import socket  # noqa: E402
import ssl  # noqa: E402
import subprocess  # noqa: E402
import tempfile  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
AGENT = os.path.join(HERE, os.pardir, "tools", "electrum-gate", "agent.py")


def load_agent():
    """Laadt agent.py als module, buiten sys.path om."""
    loader = importlib.machinery.SourceFileLoader("gate_agent", AGENT)
    spec = importlib.util.spec_from_file_location("gate_agent", AGENT, loader=loader)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class Uitslag:
    def __init__(self):
        self.goed = 0
        self.fout = []

    def check(self, naam, gelukt, uitleg=""):
        if gelukt:
            self.goed += 1
        else:
            self.fout.append(naam + ((" - " + uitleg) if uitleg else ""))

    def rapport(self):
        print()
        print("%d goed, %d fout" % (self.goed, len(self.fout)))
        for f in self.fout:
            print("  FOUT: " + f)
        return 0 if not self.fout else 1


def der(tag, inhoud):
    """Eén DER-element, met de lengte in korte of lange vorm."""
    if len(inhoud) < 0x80:
        return bytes([tag, len(inhoud)]) + inhoud
    lengte = len(inhoud).to_bytes((len(inhoud).bit_length() + 7) // 8, "big")
    return bytes([tag, 0x80 | len(lengte)]) + lengte + inhoud


def nep_certificaat(tijdtag, tijdtekst):
    """Het kleinste certificaat waarin not_after zijn validity vindt."""
    tijd = der(tijdtag, tijdtekst.encode("ascii"))
    return der(0x30, der(0x30, der(0x30, tijd + tijd)))


def test_template_is_invulbaar_zonder_schade(u):
    """Een restant uit de template-tijd dat blijft staan.

    Het bestand zit sinds 0.1.0 in de image en gaat door geen envsubst meer. Maar
    een accolade-variabele in Python-code is ook zonder umbreld een fout, en als
    iemand dit bestand ooit weer als template in een app-map zet, is dit de toets
    die het meteen ziet.
    """
    with open(AGENT, "r", encoding="utf-8") as f:
        inhoud = f.read()
    u.check("agent.py bevat geen accolade-variabelen",
            "${" not in inhoud,
            "een envsubst-stap zou die invullen en de Python-code slopen")


def test_einddatum_gelijk_aan_ssl(agent, u):
    """De echte certificaten uit de certificaatwinkel van het besturingssysteem.

    Dit is de belangrijkste toets, want `ssl` levert hier het antwoordblad. Zijn
    er geen certificaten te vinden, dan wordt dit overgeslagen en gemeld, in
    plaats van stil als geslaagd geteld.
    """
    ctx = ssl.create_default_context()
    geparseerd = ctx.get_ca_certs(binary_form=False)
    ders = ctx.get_ca_certs(binary_form=True)
    if not ders or len(ders) != len(geparseerd):
        print("OVERGESLAGEN: geen CA-certificaten beschikbaar op dit systeem, "
              "dus de vergelijking met ssl is niet gedaan")
        return

    afwijkingen = []
    for i, d in enumerate(ders):
        verwacht = int(ssl.cert_time_to_seconds(geparseerd[i]["notAfter"]))
        try:
            gekregen = agent.not_after(d)
        except Exception as e:  # noqa: BLE001
            afwijkingen.append("cert %d wierp %r" % (i, e))
            continue
        if gekregen != verwacht:
            afwijkingen.append("cert %d: verwacht %d, kreeg %d"
                               % (i, verwacht, gekregen))
    u.check("einddatum van %d CA-certificaten gelijk aan ssl" % len(ders),
            not afwijkingen, "; ".join(afwijkingen[:3]))


def test_beide_tijdvormen(agent, u):
    """UTCTime en GeneralizedTime, en beide takken van de eeuwregel.

    Het CA-materiaal gebruikt uitsluitend UTCTime met jaren na 2000, dus zonder
    deze zelfgebouwde certificaten blijven twee takken ongetoetst.
    """
    gevallen = [
        (0x18, "20700101000000Z", "Jan  1 00:00:00 2070 GMT"),
        (0x18, "99991231235959Z", "Dec 31 23:59:59 9999 GMT"),
        (0x17, "491231235959Z", "Dec 31 23:59:59 2049 GMT"),
        (0x17, "500101000000Z", "Jan  1 00:00:00 1950 GMT"),
    ]
    for tag, tekst, ssl_tekst in gevallen:
        verwacht = int(ssl.cert_time_to_seconds(ssl_tekst))
        gekregen = agent.not_after(nep_certificaat(tag, tekst))
        u.check("tijd %s" % tekst, gekregen == verwacht,
                "verwacht %d, kreeg %d" % (verwacht, gekregen))


def test_tijd_zonder_zulu_faalt_hard(agent, u):
    """Het niet-gelukkige pad: een tijd die DER niet toestaat.

    Dit is met opzet een uitzondering en geen benadering. Een certificaatdatum die
    er een uur naast zit, valt nooit op.
    """
    slecht = nep_certificaat(0x18, "20700101000000")
    try:
        agent.not_after(slecht)
        u.check("tijd zonder Z faalt", False, "er kwam een antwoord uit")
    except ValueError:
        u.check("tijd zonder Z faalt", True)
    except Exception as e:  # noqa: BLE001
        u.check("tijd zonder Z faalt", False, "verkeerde uitzondering: %r" % e)


def test_certificaat_zonder_validity_faalt(agent, u):
    """Een certificaat waar niets in staat, mag geen datum opleveren."""
    leeg = der(0x30, der(0x30, der(0x02, b"\x01")))
    try:
        agent.not_after(leeg)
        u.check("geen validity faalt", False, "er kwam een antwoord uit")
    except ValueError:
        u.check("geen validity faalt", True)


def test_live_certificaat_met_san(agent, u):
    """Een echt servercertificaat, want CA-certificaten hebben geen dNSName.

    Zonder netwerk wordt dit overgeslagen en blijft subjectAltName ongetoetst.
    Dat is gemeld, niet verzwegen.
    """
    try:
        with ssl.create_default_context().wrap_socket(
            socket.create_connection(("example.com", 443), timeout=6),
            server_hostname="example.com",
        ) as s:
            binair = s.getpeercert(binary_form=True)
            info = s.getpeercert()
    except Exception as e:  # noqa: BLE001
        print("OVERGESLAGEN: geen verbinding voor de SAN-toets (%s), "
              "subjectAltName blijft dus ongetoetst" % e)
        return

    verwacht = sorted(v for k, v in info.get("subjectAltName", ()) if k == "DNS")
    gekregen = sorted(agent.dns_names(binair))
    u.check("subjectAltName gelijk aan ssl", gekregen == verwacht,
            "verwacht %r, kreeg %r" % (verwacht[:3], gekregen[:3]))
    u.check("einddatum van een levend certificaat gelijk aan ssl",
            agent.not_after(binair) == int(ssl.cert_time_to_seconds(info["notAfter"])))


def test_keuze_negeert_onbekende_id(agent, u):
    """De guard. Dit is het pad waar een fout stil TLS zou uitschakelen.

    Wijst de keuze naar iets wat niet bestaat, dan mag de app niet doen alsof.
    Hij valt terug op wat er wel is, en meldt dat.
    """
    with tempfile.TemporaryDirectory() as tmp:
        agent.CONFIG_DIR = tmp
        agent.SELECTED_FILE = os.path.join(tmp, "selected-cert")

        nu = 2_000_000_000
        certs = [
            {"id": "zoraxy/kort", "source": "zoraxy", "name": "kort",
             "cert": "/x", "key": "/y", "not_after": nu + 86400},
            {"id": "zoraxy/lang", "source": "zoraxy", "name": "lang",
             "cert": "/a", "key": "/b", "not_after": nu + 400 * 86400},
        ]

        agent.write_selection("zoraxy/bestaat-niet")
        gekozen, waarom = agent.choose(certs)
        u.check("onbekende keuze wordt niet gehonoreerd",
                gekozen is None, "koos %r" % (gekozen and gekozen["id"]))
        u.check("en er wordt ook niet zomaar een ander gepakt",
                "2 certificates found" in waarom, "waarom was %r" % waarom)
        # De reden noemt het aantal en niet de namen. Op de Umbrel van de
        # gebruiker zijn dat er veertien, en die opsomming maakte van de melding
        # een muur tekst die zegt wat de keuzelijst eronder al toont.
        u.check("de reden somt de kandidaten niet op",
                "zoraxy/kort" not in waarom and "zoraxy/lang" not in waarom,
                "waarom was %r" % waarom)

        # Eén kandidaat is geen keuze, dus die mag wel automatisch.
        gekozen, waarom = agent.choose(certs[:1])
        u.check("bij precies één certificaat kiest de app zelf",
                gekozen is not None and gekozen["id"] == "zoraxy/kort",
                "koos %r" % (gekozen and gekozen["id"]))

        agent.write_selection("zoraxy/kort")
        gekozen, waarom = agent.choose(certs)
        u.check("een bestaande keuze wint van de automaat",
                gekozen["id"] == "zoraxy/kort", "koos %r" % gekozen["id"])

        onleesbaar = [dict(certs[0], unreadable="stuk")]
        gekozen, waarom = agent.choose(onleesbaar)
        u.check("een onleesbaar certificaat wordt niet gekozen",
                gekozen is None, "koos %r" % (gekozen and gekozen["id"]))


def test_keuze_zonder_geldig_certificaat(agent, u):
    """Alles verlopen: dan liever niets dan een verlopen certificaat opdringen."""
    with tempfile.TemporaryDirectory() as tmp:
        agent.CONFIG_DIR = tmp
        agent.SELECTED_FILE = os.path.join(tmp, "selected-cert")
        verlopen = [{"id": "own/oud", "source": "own", "name": "oud",
                     "cert": "/a", "key": "/b", "not_after": 1}]
        gekozen, waarom = agent.choose(verlopen)
        u.check("alles verlopen levert geen keuze op", gekozen is None)
        u.check("en de reden wordt genoemd", "expired" in waarom,
                "waarom was %r" % waarom)

        gekozen, waarom = agent.choose([])
        u.check("een lege lijst levert geen keuze op", gekozen is None)


def test_paren_vinden(agent, u):
    """De drie naamvormen van de bronnen, in een echte map."""
    with tempfile.TemporaryDirectory() as tmp:
        # Zoraxy-vorm
        open(os.path.join(tmp, "a.example.org.pem"), "w").close()
        open(os.path.join(tmp, "a.example.org.key"), "w").close()
        # Een certificaat zonder sleutel hoort niet mee te komen
        open(os.path.join(tmp, "b.example.org.pem"), "w").close()
        # Nginx Proxy Manager-vorm
        os.mkdir(os.path.join(tmp, "npm-7"))
        open(os.path.join(tmp, "npm-7", "fullchain.pem"), "w").close()
        open(os.path.join(tmp, "npm-7", "privkey.pem"), "w").close()
        # Een map zonder sleutel hoort ook niet mee te komen
        os.mkdir(os.path.join(tmp, "npm-8"))
        open(os.path.join(tmp, "npm-8", "fullchain.pem"), "w").close()

        namen = sorted(n for n, _, _ in agent._pairs_in(tmp))
        u.check("alleen volledige paren", namen == ["a.example.org", "npm-7"],
                "vond %r" % namen)

        u.check("een map die niet bestaat levert een lege lijst",
                agent._pairs_in(os.path.join(tmp, "weg")) == [])


def openssl_paar(map_, naam, cn, dagen=365, wachtwoord=None, san=True):
    """Een echt zelfondertekend paar. Geeft (certpad, keypad) of None.

    Echt en niet zelfgebouwd, want de controle die hier getoetst wordt is die van
    OpenSSL zelf: hoort deze sleutel bij dit certificaat. Een met de hand
    geknutseld DER-certificaat heeft geen sleutel en bewijst daar dus niets over.
    """
    if not shutil.which("openssl"):
        return None
    cert = os.path.join(map_, naam + ".crt")
    key = os.path.join(map_, naam + ".key")
    opdracht = [
        "openssl", "req", "-x509", "-newkey", "rsa:2048",
        "-keyout", key, "-out", cert, "-days", str(dagen),
        # De schuine streep is in -subj het scheidingsteken tussen velden, dus een
        # hostnaam die er zelf een bevat moet ontsnapt worden. Zonder dit sloeg de
        # padtruc-toets zichzelf stil over, en dat is precies de toets die er moet
        # zijn.
        "-subj", "/CN=" + cn.replace("/", "\\/"),
    ]
    if san:
        opdracht += ["-addext", "subjectAltName=DNS:" + cn]
    if wachtwoord:
        opdracht += ["-passout", "pass:" + wachtwoord]
    else:
        opdracht += ["-nodes"]
    klaar = subprocess.run(opdracht, capture_output=True)
    if klaar.returncode != 0:
        # De laatste regel, want openssl vult stderr met voortgangspuntjes en de
        # echte reden staat onderaan.
        regels = [r for r in klaar.stderr.decode("utf-8", "replace").splitlines()
                  if r.strip()]
        print("OVERGESLAGEN: openssl gaf een fout bij %s (%s)"
              % (naam, regels[-1][:160] if regels else "geen melding"))
        return None
    return cert, key


class Tijdreis:
    """Een tijdmodule die alleen over `time()` liegt.

    Nodig om de vervaldatum-guard te toetsen: een certificaat dat al verlopen ís
    kan openssl 3.1 niet maken, en de datum vooruitzetten kan wel.
    """

    def __init__(self, echt, wanneer):
        self._echt = echt
        self._wanneer = wanneer

    def time(self):
        return self._wanneer

    def __getattr__(self, naam):
        return getattr(self._echt, naam)


def test_upload(agent, u):
    """Het uploadpad, en vooral wat het weigert.

    Dit is het enige pad waarlangs iets van buiten bestanden neerzet, dus de
    toetsen gaan over de niet-gelukkige gevallen. De duurste fout die hier
    voorkomen moet worden is een sleutel die niet bij het certificaat hoort: die
    levert een nginx op die niet meer herlaadt, en dat is dezelfde klasse storing
    als waardoor 0.0.3 helemaal niet startte.
    """
    with tempfile.TemporaryDirectory() as tmp:
        bron = os.path.join(tmp, "bron")
        doel = os.path.join(tmp, "own")
        os.makedirs(bron)
        agent.UPLOAD_DIR = doel
        agent.SOURCES = [("Own folder", doel)]

        # Uit zonder map. Eerst, want dit hoort niet stil door te vallen naar een
        # pad dat toevallig bestaat.
        agent.UPLOAD_DIR = ""
        naam, fout = agent.accept_upload("x", "y")
        u.check("zonder uploadmap weigert de agent",
                naam is None and "switched off" in (fout or ""),
                "gaf %r, %r" % (naam, fout))
        agent.UPLOAD_DIR = doel

        naam, fout = agent.accept_upload("hallo", "-----BEGIN PRIVATE KEY-----")
        u.check("rommel in plaats van een certificaat wordt geweigerd",
                naam is None and "does not contain a certificate" in (fout or ""),
                "gaf %r, %r" % (naam, fout))

        naam, fout = agent.accept_upload(
            "-----BEGIN CERTIFICATE-----\nAAAA\n-----END CERTIFICATE-----", "hallo")
        u.check("rommel in plaats van een sleutel wordt geweigerd",
                naam is None and "does not contain a private key" in (fout or ""),
                "gaf %r, %r" % (naam, fout))

        u.check("een weigering laat niets achter in de doelmap",
                not os.path.isdir(doel) or os.listdir(doel) == [],
                "er staat: %r" % (os.path.isdir(doel) and os.listdir(doel)))

        paar = openssl_paar(bron, "goed", "gate.example.org")
        if paar is None:
            print("OVERGESLAGEN: geen openssl, dus het gelukkige pad en de "
                  "sleutelcontrole van deze toets blijven ongetoetst")
            return
        cert_tekst = open(paar[0], encoding="ascii").read()
        key_tekst = open(paar[1], encoding="ascii").read()

        # De belangrijkste toets van dit bestand na de datumvergelijking: een
        # sleutel van een ánder paar hoort hier te stranden.
        ander = openssl_paar(bron, "ander", "andere.example.org")
        naam, fout = agent.accept_upload(cert_tekst,
                                         open(ander[1], encoding="ascii").read())
        u.check("een sleutel van een ander paar wordt geweigerd",
                naam is None and "does not go with" in (fout or ""),
                "gaf %r, %r" % (naam, fout))
        u.check("en ook dan blijft de doelmap leeg",
                not os.path.isdir(doel) or os.listdir(doel) == [],
                "er staat: %r" % (os.path.isdir(doel) and os.listdir(doel)))

        # Een versleutelde sleutel: nginx kan die niet openen zonder wachtwoord.
        met_slot = openssl_paar(bron, "slot", "slot.example.org",
                                wachtwoord="geheim")
        if met_slot:
            naam, fout = agent.accept_upload(
                open(met_slot[0], encoding="ascii").read(),
                open(met_slot[1], encoding="ascii").read())
            u.check("een versleutelde sleutel wordt geweigerd",
                    naam is None and "encrypted" in (fout or ""),
                    "gaf %r, %r" % (naam, fout))

        # Verlopen. De klok gaat vooruit in plaats van het certificaat achteruit.
        echt = agent.time
        agent.time = Tijdreis(echt, echt.time() + 400 * 86400)
        try:
            naam, fout = agent.accept_upload(cert_tekst, key_tekst)
        finally:
            agent.time = echt
        u.check("een verlopen certificaat wordt geweigerd",
                naam is None and "expired on" in (fout or ""),
                "gaf %r, %r" % (naam, fout))

        # Een certificaat zonder bruikbare naam: er is dan niets om een wallet
        # tegen te vergelijken, dus plaatsen heeft geen zin.
        naamloos = openssl_paar(bron, "naamloos", "!!!", san=False)
        if naamloos:
            naam, fout = agent.accept_upload(
                open(naamloos[0], encoding="ascii").read(),
                open(naamloos[1], encoding="ascii").read())
            u.check("een certificaat zonder hostnaam wordt geweigerd",
                    naam is None and "no host name" in (fout or ""),
                    "gaf %r, %r" % (naam, fout))

        # En dan het gelukkige pad.
        naam, fout = agent.accept_upload(cert_tekst, key_tekst)
        u.check("een geldig paar wordt aangenomen",
                naam == "gate.example.org" and fout is None,
                "gaf %r, %r" % (naam, fout))
        u.check("en staat als paar in de doelmap",
                sorted(os.listdir(doel)) == ["gate.example.org.key",
                                             "gate.example.org.pem"],
                "er staat: %r" % sorted(os.listdir(doel)))
        u.check("de scanner vindt het meteen",
                [n for n, _, _ in agent._pairs_in(doel)] == ["gate.example.org"])
        u.check("de id verwijst naar de bron uit de omgeving",
                agent.upload_id(naam) == "Own folder/gate.example.org",
                "gaf %r" % agent.upload_id(naam))

        # Nog een keer hetzelfde: dat vervangt, en laat geen .tmp achter. Een
        # vernieuwd certificaat is precies waarom dit pad bestaat.
        naam, fout = agent.accept_upload(cert_tekst, key_tekst)
        u.check("hetzelfde nog eens vervangt zonder klagen",
                naam == "gate.example.org" and fout is None,
                "gaf %r, %r" % (naam, fout))
        u.check("en er blijft geen tijdelijk bestand liggen",
                not [n for n in os.listdir(doel) if n.endswith(".tmp")],
                "er staat: %r" % sorted(os.listdir(doel)))

        # Een naam met padtrucs erin kan niet uit het doel ontsnappen, want de
        # naam komt uit het certificaat en niet uit het verzoek.
        stout = openssl_paar(bron, "stout", "../../etc/nginx/evil", san=False)
        if stout:
            naam, fout = agent.accept_upload(
                open(stout[0], encoding="ascii").read(),
                open(stout[1], encoding="ascii").read())
            u.check("een hostnaam met padtekens levert een platte naam op",
                    fout is None and "/" not in (naam or "/") and ".." not in (naam or ".."),
                    "gaf %r, %r" % (naam, fout))
            u.check("en het bestand staat in de doelmap zelf",
                    fout is not None or os.path.isfile(
                        os.path.join(doel, naam + ".pem")))


def test_sessieregels_indelen(agent, u):
    """De drie soorten sessieregels uit de nginx-log.

    Het onderscheid tussen `probe` en `refused` is het punt: een doorgestuurde
    poort wordt gescand, en zonder dat onderscheid gaf elke scan een rode regel
    "refused" terwijl er niets geweigerd is. De grens ligt bij de bytes en niet bij
    de duur of de status, en dat is precies wat hier vastligt.
    """
    with tempfile.TemporaryDirectory() as tmp:
        agent.STREAM_LOG = os.path.join(tmp, "stream.log")
        with open(agent.STREAM_LOG, "w", encoding="utf-8") as f:
            # Een echte sessie, een scan, en een verbinding die wél iets
            # doorgaf en daarna omviel.
            f.write("2026-08-20T12:39:29+02:00 200 95300 218000 240.5\n")
            f.write("2026-08-20T12:38:29+02:00 500 0 0 1.001\n")
            f.write("2026-08-20T12:40:29+02:00 502 120 0 0.5\n")
            # Een scan die lang open bleef staan: nog steeds een scan.
            f.write("2026-08-20T12:41:29+02:00 500 0 0 45.0\n")
            f.write("rommel die geen sessieregel is\n")

        regels = agent.read_stream_log(0)
        soorten = [r["event"] for r in regels]
        u.check("de drie soorten komen er in de juiste orde uit",
                soorten == ["disconnect", "probe", "refused", "probe"],
                "gaf %r" % soorten)

        u.check("een onleesbare regel wordt overgeslagen", len(regels) == 4,
                "gaf %d regels" % len(regels))

        probe = regels[1]
        u.check("de scan houdt zijn status in de notitie",
                probe.get("note") == "nothing exchanged, status 500",
                "gaf %r" % probe.get("note"))
        u.check("en zijn bytes staan op nul",
                probe["bytes_in"] == 0 and probe["bytes_out"] == 0)

        stuk = regels[2]
        u.check("een sessie die wél verkeer had heet refused",
                stuk.get("note") == "session ended with status 502",
                "gaf %r" % stuk.get("note"))

        echt = regels[0]
        # 240 en niet 241: round() in Python rondt een halve naar even af.
        u.check("een geslaagde sessie heeft geen notitie nodig",
                "note" not in echt and echt["seconds"] == 240,
                "gaf %r" % echt)


def test_open_verbindingen(agent, u):
    """De verbindingsteller, en de logregel die eruit volgt.

    Dit gaat via `build_status`, dus via de ronde die de agent echt draait, en
    niet via een interne helper. Twee dingen moeten kloppen die makkelijk stil
    fout gaan: een teller die er niet is mag geen nul worden, en een
    connect-regel moet de volgende ronde overleven. Die laatste zat er eerst
    naast: `build_status` neemt de eigen gebeurtenissen van de vorige ronde over
    aan de hand van een opsomming van soorten, en `connect` stond er niet in.
    Een regel die na een minuut weer verdwijnt is nutteloos.
    """
    with tempfile.TemporaryDirectory() as tmp:
        agent.SOURCES = []
        agent.ELECTRUM_HOST = ""  # dan faalt de probe meteen, zonder netwerk
        agent.STATE_DIR = tmp
        agent.CONFIG_DIR = os.path.join(tmp, "config")
        agent.SELECTED_FILE = os.path.join(agent.CONFIG_DIR, "selected-cert")
        agent.STATUS_FILE = os.path.join(tmp, "status.json")
        agent.CERT_CONF = os.path.join(tmp, "cert.conf")
        agent.RELOAD_FLAG = os.path.join(tmp, "reload")
        agent.STREAM_LOG = os.path.join(tmp, "stream.log")
        agent.SESSIONS_FILE = os.path.join(tmp, "sessions")

        def ronde():
            status = agent.build_status([])
            agent.write_status(status)
            return status

        # Geen tellerbestand: dan hoort er niets in de status te staan. Nul zou
        # hier "geen wallet verbonden" beweren op grond van niets.
        status = ronde()
        u.check("zonder teller geen bewering over verbindingen",
                "open_connections" not in status["tls"],
                "stond er: %r" % status["tls"])

        def zet_teller(tekst):
            with open(agent.SESSIONS_FILE, "w", encoding="utf-8") as f:
                f.write(tekst)

        zet_teller("0\n")
        status = ronde()
        u.check("een teller van nul komt wel in de status",
                status["tls"].get("open_connections") == 0)
        u.check("nul verbindingen levert geen connect-regel op",
                not [e for e in status["log"] if e.get("event") == "connect"])

        zet_teller("1\n")
        status = ronde()
        connects = [e for e in status["log"] if e.get("event") == "connect"]
        u.check("een verbinding erbij levert één connect-regel op",
                len(connects) == 1, "regels: %r" % connects)
        u.check("en die regel noemt hoeveel er nu openstaan",
                connects and connects[0].get("note") == "1 open now",
                "regel: %r" % (connects[0] if connects else None))

        # Zelfde aantal, dus geen nieuwe regel. En de vorige moet blijven staan:
        # dit is de toets die de vergeten opsomming vond.
        status = ronde()
        connects = [e for e in status["log"] if e.get("event") == "connect"]
        u.check("een gelijk aantal levert geen tweede regel op",
                len(connects) == 1, "regels: %r" % connects)

        zet_teller("rommel")
        status = ronde()
        u.check("een onleesbare teller wordt niets, niet nul",
                "open_connections" not in status["tls"],
                "stond er: %r" % status["tls"])
        u.check("en de eerdere connect-regel staat er nog",
                len([e for e in status["log"] if e.get("event") == "connect"]) == 1)

        # Van onbekend naar een aantal hoort ook een regel op te leveren. Dit is
        # het geval "de wallet hing er al voordat de agent begon", en dat is
        # precies wanneer iemand op de pagina komt kijken. Zonder deze regel blijft
        # het log dan uren stil.
        zet_teller("2\n")
        status = ronde()
        connects = [e for e in status["log"] if e.get("event") == "connect"]
        u.check("na een onbekend aantal levert een verbinding alsnog een regel op",
                len(connects) == 2, "regels: %r" % connects)
        u.check("en die noemt het aantal dat er nu staat",
                connects[-1].get("note") == "2 open now",
                "regel: %r" % connects[-1])


def main():
    u = Uitslag()
    test_template_is_invulbaar_zonder_schade(u)
    agent = load_agent()
    test_einddatum_gelijk_aan_ssl(agent, u)
    test_beide_tijdvormen(agent, u)
    test_tijd_zonder_zulu_faalt_hard(agent, u)
    test_certificaat_zonder_validity_faalt(agent, u)
    test_live_certificaat_met_san(agent, u)
    test_keuze_negeert_onbekende_id(agent, u)
    test_keuze_zonder_geldig_certificaat(agent, u)
    test_paren_vinden(agent, u)
    # Deze twee als laatste, want ze verzetten de module-instellingen naar een
    # tijdelijke map.
    test_upload(agent, u)
    test_sessieregels_indelen(agent, u)
    test_open_verbindingen(agent, u)
    return u.rapport()


if __name__ == "__main__":
    sys.exit(main())