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 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 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)
    return u.rapport()


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