333 lines
13 KiB
Python
333 lines
13 KiB
Python
|
"""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())
|