De app kijkt naar zijn eigen voordeur, en meet minder dan het plan beloofde
Open punt 5 van Webinterface, goedgekeurd en gebouwd. De agent verbindt met de eigen TLS-poort, maakt de handdruk af en vergelijkt het getoonde certificaat byte voor byte met het gekozen bestand. Die vergelijking blijkt waardevoller dan de handdruk. Ze vangt een certificaatwissel die nginx nooit heeft toegepast, en dat is precies het geval waar een controle op vertrouwen blind voor is. Er wordt daarom bewust niet tegen de certificaatwinkel van het besturingssysteem geverifieerd: een zelfondertekend certificaat uploaden is een ondersteunde bron en die opstelling zou dan als kapot gemeld worden. Twee dingen liepen anders dan het plan zei en staan nu rechtgezet in PLAN 4a2, OPEN punt 5 en de changelog. De belofte "luisteren, certificaat en doorverbinding in een keer" klopt voor twee van de drie: na de handdruk wordt er niets verstuurd. Een echt verzoek zou de sessie bytes geven, en sessies met bytes worden nooit uit het activiteitenlog gefilterd, want die kunnen een storing zijn. En dat filteren was de tweede verrassing. Elke meting is voor nginx een gewone sessie en levert dus een logregel op; zonder rem ging het log over onszelf in plaats van over wallets. Er zit nu een rem van vijf minuten op, er wordt niet gemeten vlak na een herlading omdat het vorige certificaat er dan nog staat, en de eigen regels worden weggelaten op grond van het moment. Tests: 22 nieuw, met de nadruk op de niet-gelukkige paden en op de rem, want dat is wat het log bruikbaar houdt. Alle vier de beslissende regels mutatie-getest. De changelog kreeg ook de ontbrekende 0.0.15 erbij. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Harmen
@@ -0,0 +1,332 @@
|
||||
"""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())
|
||||
Reference in New Issue
Block a user