UmbrelApps/tests/test_agent_zelfcontrole.py

"""Toetst de zelfcontrole van de agent: antwoordt de eigen TLS-poort?

Waarom deze test bestaat: dit is de enige controle die over de app zelf gaat, en
hij kan op twee manieren stil verkeerd zijn. Hij kan "ok" melden terwijl er iets
anders aan de lijn hangt dan het gekozen certificaat, en hij kan het
activiteitenlog vullen met zijn eigen verbindingen. Dat tweede is geen schoonheids-
foutje: elke meting is voor nginx een gewone sessie, dus zonder rem en zonder
filter gaat het log over onszelf in plaats van over wallets.

De nadruk ligt daarom op wat er níet moet gebeuren: niet meten vlak na een
herlading, niet meten als er net gemeten is, en de eigen sessieregel niet laten
staan.

Draaien:

    python tests/test_agent_zelfcontrole.py

Wat een echte TLS-server vraagt, vraagt openssl. Is dat er niet, dan slaan die
toetsen zichzelf over en zeggen dat erbij; de rest draait gewoon.
"""

import sys

# Zelfde reden als in test_agent_certificates.py: geen .pyc naast de app-map.
sys.dont_write_bytecode = True

import json  # noqa: E402
import os  # noqa: E402
import socket  # noqa: E402
import ssl  # noqa: E402
import tempfile  # noqa: E402
import threading  # noqa: E402
import time  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
if HERE not in sys.path:
    sys.path.insert(0, HERE)

# De buurtest is hier de gereedschapskist: hij laadt agent.py.template als module
# en maakt echte sleutelparen met openssl. Dat tweede keer overdoen zou een tweede
# versie opleveren die uit elkaar loopt.
from test_agent_certificates import Uitslag, load_agent, openssl_paar  # noqa: E402


class TlsServer:
    """Een echte TLS-server op een vrije poort van de loopback.

    Echt en niet nagebootst, want wat hier getoetst wordt is een handdruk. Een
    verzonnen server bewijst niets over `wrap_socket`.
    """

    def __init__(self, cert, key):
        self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self.context.load_cert_chain(cert, key)
        self.sock = socket.socket()
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(5)
        self.port = self.sock.getsockname()[1]
        self.stoppen = False
        self.thread = threading.Thread(target=self._loop, daemon=True)
        self.thread.start()

    def _loop(self):
        while not self.stoppen:
            try:
                rauw, _ = self.sock.accept()
            except OSError:
                return
            try:
                with self.context.wrap_socket(rauw, server_side=True) as tls:
                    # Wachten tot de client sluit. De agent sluit meteen na de
                    # handdruk, dus dit levert b"" op en niet een hangende lus.
                    tls.recv(1)
            except OSError:
                pass

    def close(self):
        self.stoppen = True
        try:
            self.sock.close()
        except OSError:
            pass


class KaleServer:
    """Luistert wel, maar spreekt geen TLS. Accepteert en sluit meteen."""

    def __init__(self):
        self.sock = socket.socket()
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(5)
        self.port = self.sock.getsockname()[1]
        self.stoppen = False
        self.thread = threading.Thread(target=self._loop, daemon=True)
        self.thread.start()

    def _loop(self):
        while not self.stoppen:
            try:
                rauw, _ = self.sock.accept()
            except OSError:
                return
            rauw.close()

    def close(self):
        self.stoppen = True
        try:
            self.sock.close()
        except OSError:
            pass


def dichte_poort():
    """Een poortnummer waar zeker niets luistert."""
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    poort = s.getsockname()[1]
    s.close()
    return poort


def richt(agent, poort):
    agent.TLS_HOST = "127.0.0.1"
    agent.TLS_PORT = poort


def test_zonder_certificaat_wordt_er_niet_verbonden(agent, u):
    """Geen keuze betekent geen poort 50022, en dus geen storing.

    Dit is de normale toestand van een verse installatie. Hem als fout melden zou
    de ergste soort onwaarheid zijn: precies daar staat al een foutmelding dat er
    een certificaat gekozen moet worden, en een tweede rode melding erbij leidt
    weg van de knop die het oplost.
    """
    richt(agent, dichte_poort())

    for leeg in (None, {}):
        uitkomst = agent.probe_self(leeg)
        u.check("zonder certificaat is de toestand 'off' (%r)" % (leeg,),
                uitkomst.get("state") == "off", "gaf %r" % uitkomst)
        u.check("en dan is er niets gemeten (%r)" % (leeg,),
                "ms" not in uitkomst and "error" not in uitkomst,
                "gaf %r" % uitkomst)


def test_niemand_luistert(agent, u):
    """Wel een certificaat, geen poort. Dat is wel een storing."""
    richt(agent, dichte_poort())
    uitkomst = agent.probe_self({"domain": "gate.example.org", "cert": "/bestaat/niet"})
    u.check("een dichte poort geeft 'failed'",
            uitkomst.get("state") == "failed", "gaf %r" % uitkomst)
    u.check("en de reden staat erbij",
            bool(uitkomst.get("error")), "gaf %r" % uitkomst)
    u.check("een mislukking meldt geen meettijd",
            "ms" not in uitkomst, "gaf %r" % uitkomst)


def test_luisteren_is_niet_genoeg(agent, u):
    """Iets dat luistert maar geen TLS spreekt, is ook een storing.

    Dit onderscheidt de controle van een simpele poortcheck. Een open poort
    bewijst niet dat er een bruikbaar certificaat achter zit, en juist dat is wat
    een wallet nodig heeft.
    """
    server = KaleServer()
    try:
        richt(agent, server.port)
        uitkomst = agent.probe_self({"domain": "gate.example.org", "cert": "/bestaat/niet"})
        u.check("luisteren zonder TLS geeft 'failed'",
                uitkomst.get("state") == "failed", "gaf %r" % uitkomst)
    finally:
        server.close()


def test_gelukkig_pad_en_verkeerd_certificaat(agent, u):
    """De handdruk lukt, en dan gaat het erom wélk certificaat er terugkomt.

    Het tweede geval is de reden dat deze controle niet op vertrouwen leunt maar
    op de bytes van het certificaat: nginx dat na een mislukte herlading nog het
    vorige certificaat serveert, is voor elke trust-controle in orde en is toch
    precies de storing die je zoekt.
    """
    with tempfile.TemporaryDirectory() as tmp:
        paar = openssl_paar(tmp, "gekozen", "gate.example.org")
        ander = openssl_paar(tmp, "ander", "andere.example.org")
        if not paar or not ander:
            print("OVERGESLAGEN: geen openssl, dus de handdruk is niet getoetst")
            return
        cert, key = paar
        ander_cert, _ = ander

        server = TlsServer(cert, key)
        try:
            richt(agent, server.port)

            uitkomst = agent.probe_self({"domain": "gate.example.org", "cert": cert})
            u.check("het gekozen certificaat aan de lijn geeft 'ok'",
                    uitkomst.get("state") == "ok", "gaf %r" % uitkomst)
            u.check("en een meettijd in milliseconden",
                    isinstance(uitkomst.get("ms"), int) and uitkomst["ms"] >= 0,
                    "gaf %r" % uitkomst)

            uitkomst = agent.probe_self({"domain": "andere.example.org",
                                         "cert": ander_cert})
            u.check("een ander certificaat aan de lijn wordt herkend",
                    uitkomst.get("state") == "wrong-certificate", "gaf %r" % uitkomst)
        finally:
            server.close()


def test_eigen_sessies_uit_het_log(agent, u):
    """De filter die het log over wallets laat gaan in plaats van over onszelf."""
    def regel(t, event="probe", inn=0, uit=0):
        return {"t": t, "event": event, "bytes_in": inn, "bytes_out": uit}

    entries = [regel(1000), regel(2000)]
    over = agent.suppress_self_probes(entries, [1001])
    u.check("een eigen meting verdwijnt uit het log",
            [e["t"] for e in over] == [2000], "over: %r" % over)

    over = agent.suppress_self_probes([regel(1000)], [1010])
    u.check("een scan buiten het venster blijft staan",
            len(over) == 1, "over: %r" % over)

    # Dit is de belangrijkste: een sessie waarin wél iets is doorgegeven, is per
    # definitie niet van ons. Die weglaten zou een storing verbergen.
    echt = regel(1000, event="refused", inn=120, uit=300)
    over = agent.suppress_self_probes([echt], [1000])
    u.check("een sessie met verkeer blijft staan, ook op hetzelfde moment",
            over == [echt], "over: %r" % over)

    # Eén onthouden meting laat hoogstens één regel weg. Anders zou een enkele
    # meting een reeks echte scans kunnen opslokken.
    over = agent.suppress_self_probes([regel(1000), regel(1001)], [1000])
    u.check("één meting laat hoogstens één regel weg",
            len(over) == 1, "over: %r" % over)

    u.check("zonder metingen verandert er niets",
            agent.suppress_self_probes([regel(1000)], []) == [regel(1000)])


def test_ronde_meet_niet_elke_keer(agent, u):
    """De rem, via `build_status`, dus via de ronde die de agent echt draait.

    Drie dingen op een rij die alle drie stil fout kunnen gaan:

    1. vlak na een certificaatwissel wordt er niet gemeten. nginx herlaadt binnen
       tien seconden, dus meten zou het vorige certificaat zien en dat als storing
       melden;
    2. de ronde daarna wordt er wel gemeten;
    3. de ronde daarna weer niet, want er is net gemeten. Zonder deze rem staat er
       elke minuut een eigen regel in het activiteitenlog.
    """
    with tempfile.TemporaryDirectory() as tmp:
        bron = os.path.join(tmp, "certs")
        os.makedirs(bron)
        paar = openssl_paar(bron, "gate", "gate.example.org")
        if not paar:
            print("OVERGESLAGEN: geen openssl, dus de rem is niet getoetst")
            return
        cert, key = paar

        agent.SOURCES = [("Own folder", bron)]
        agent.ELECTRUM_HOST = ""  # dan faalt de backend-probe 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")
        agent.SELF_CHECK_INTERVAL = 300

        server = TlsServer(cert, key)
        try:
            richt(agent, server.port)

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

            eerste = ronde()["tls"]["self_check"]
            u.check("vlak na de eerste keuze wordt er niet gemeten",
                    eerste.get("state") == "pending", "gaf %r" % eerste)

            tweede = ronde()["tls"]["self_check"]
            u.check("de ronde daarna wel, en die lukt",
                    tweede.get("state") == "ok", "gaf %r" % tweede)

            derde = ronde()
            u.check("en daarna niet opnieuw, want er is net gemeten",
                    derde["tls"]["self_check"].get("at") == tweede.get("at"),
                    "gaf %r" % derde["tls"]["self_check"])
            u.check("er is dus precies één eigen meting onthouden",
                    len(derde.get("self_probes", [])) == 1,
                    "gaf %r" % derde.get("self_probes"))

            # En nu het sluitstuk: de sessieregel die nginx van diezelfde meting
            # geschreven zou hebben, hoort niet in het log te komen. Een tweede
            # regel even ervoor is niet van ons en moet blijven staan.
            gemeten = tweede["at"]
            with open(agent.STREAM_LOG, "w", encoding="utf-8") as f:
                for t in (gemeten - 60, gemeten):
                    f.write("%s 500 0 0 0.010\n"
                            % time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime(t)))

            log = ronde()["log"]
            momenten = [e["t"] for e in log if e.get("event") == "probe"]
            u.check("de eigen sessieregel staat niet in het log",
                    gemeten not in momenten, "regels op: %r" % momenten)
            u.check("en de scan van iemand anders wel",
                    (gemeten - 60) in momenten, "regels op: %r" % momenten)
        finally:
            server.close()


def test_eerste_ronde_en_snelle_herhaling(agent, u):
    """De startrace, en de rem die een mislukking vasthield.

    AANLEIDING (gemeld door de gebruiker, 31-08-2026): na elke herstart of update
    las de statuskaart vijf minuten lang "Not answering" met
    `[Errno 111] Connection refused`, terwijl de gateway van buiten gewoon werkte.

    De oorzaak was een startvolgorde en geen storing. De compose laat `server` ná
    de agent starten (`depends_on: [agent]`), en `build_status` draait meteen bij
    het opstarten. De allereerste meting valt dus gegarandeerd in het gat waarin
    nginx nog niet luistert. Dat was op zichzelf te verdragen; het probleem was dat
    de volgende meting pas na SELF_CHECK_INTERVAL kwam.

    Twee dingen worden hier vastgelegd, en ze repareren allebei een helft:

    1. de eerste ronde van een proces meet niet, ook niet als het certificaat
       ongewijzigd is (dus met `reloaded` op False, en dát is het geval dat de
       bestaande toets hierboven NIET raakt: daar is de eerste ronde `pending`
       omdat het certificaat net gekozen is);
    2. na een mislukking wordt er na SELF_CHECK_RETRY opnieuw gemeten en niet pas
       na SELF_CHECK_INTERVAL.
    """
    with tempfile.TemporaryDirectory() as tmp:
        bron = os.path.join(tmp, "certs")
        os.makedirs(bron)
        paar = openssl_paar(bron, "gate", "gate.example.org")
        if not paar:
            print("OVERGESLAGEN: geen openssl, dus de startrace is niet getoetst")
            return
        cert, key = paar

        agent.SOURCES = [("Own folder", bron)]
        agent.ELECTRUM_HOST = ""
        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")
        agent.SELF_CHECK_INTERVAL = 300
        agent.SELF_CHECK_RETRY = 60

        server = TlsServer(cert, key)
        try:
            richt(agent, server.port)

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

            ronde()                      # certificaat gekozen: pending via reloaded
            tweede = ronde()["tls"]["self_check"]
            u.check("de meting lukt zodra het certificaat staat",
                    tweede.get("state") == "ok", "gaf %r" % tweede)

            # Nu doen alsof het proces opnieuw begint, met een ONGEWIJZIGD
            # certificaat. Vóór de reparatie mat hij hier gewoon door.
            agent.EERSTE_RONDE = True
            herstart = ronde()["tls"]["self_check"]
            u.check("de eerste ronde na een herstart meet niet",
                    herstart.get("state") == "pending", "gaf %r" % herstart)
            u.check("en zonder `at`, zodat de volgende ronde wél meet",
                    "at" not in herstart, "gaf %r" % herstart)

            derde = ronde()["tls"]["self_check"]
            u.check("de ronde daarna meet wel",
                    derde.get("state") == "ok", "gaf %r" % derde)

            # De tweede helft: een mislukking mag niet vijf minuten blijven staan.
            # De opgeslagen meting met de hand op 'failed' zetten, 90 seconden oud:
            # dat is meer dan SELF_CHECK_RETRY en minder dan SELF_CHECK_INTERVAL.
            def zet_vorige(state, ouderdom):
                with open(agent.STATUS_FILE, "r", encoding="utf-8") as f:
                    status = json.load(f)
                status["tls"]["self_check"] = {
                    "at": int(time.time()) - ouderdom, "state": state,
                }
                with open(agent.STATUS_FILE, "w", encoding="utf-8") as f:
                    json.dump(status, f)

            zet_vorige("failed", 90)
            na_fout = ronde()["tls"]["self_check"]
            u.check("na een mislukking wordt er binnen een minuut opnieuw gemeten",
                    na_fout.get("state") == "ok", "gaf %r" % na_fout)

            # En de tegenproef, want anders toetst het bovenstaande alleen dat er
            # ooit gemeten wordt: een GESLAAGDE meting van dezelfde ouderdom hoort
            # juist te blijven staan.
            zet_vorige("ok", 90)
            na_goed = ronde()["tls"]["self_check"]
            u.check("een geslaagde meting van 90 seconden oud blijft staan",
                    na_goed.get("at") == int(time.time()) - 90,
                    "gaf %r" % na_goed)
        finally:
            server.close()


def main():
    u = Uitslag()
    agent = load_agent()
    test_zonder_certificaat_wordt_er_niet_verbonden(agent, u)
    test_niemand_luistert(agent, u)
    test_luisteren_is_niet_genoeg(agent, u)
    test_gelukkig_pad_en_verkeerd_certificaat(agent, u)
    test_eigen_sessies_uit_het_log(agent, u)
    # Als laatste, want deze verzet de module-instellingen naar een tijdelijke map.
    test_ronde_meet_niet_elke_keer(agent, u)
    test_eerste_ronde_en_snelle_herhaling(agent, u)
    return u.rapport()


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