HarmenandClaude Opus 5
67ed9b603b
umbrelOS leest per store één repo, dus twee apps in twee repo's kan niet. Deze repo is de store en bevat vanaf nu Electrum Gate en het werk aan Evolu Relay. Opgezet als verse repo op verzoek van de gebruiker: de historie van ElectrumTLS en van EvoluRelay komt niet mee. Dat heeft één gevolg dat verder gaat dan opruimen. In de historie van ElectrumTLS staat het domein van de gebruiker en het certificaatpad, van vóór de opschoning van 19-08. Die komt hier niet in. Zolang die repo op de Git-server blijft staan verandert dat niets, dus het weghalen ervan is het laatste stuk van open punt 3 van het plan Appstore, en geen bijzaak. De store zelf hoefde niet te veranderen: store-id whatsnext, en dus blijft het app-id whatsnext-electrum-gate. Dat hangt aan het store-id en niet aan de URL, dus voor umbrelOS is dit dezelfde app in een andere store. Dat de store op 19-08 naar de maker genoemd werd in plaats van naar deze ene app, betaalt zich hier uit. Wat de documentatie betreft is dit één wortel voor beide apps, en dat was de reden om samen te voegen en niet de prijs ervan: de appstore-spec, het pinnen van images en de werkwijze golden al voor allebei en stonden in twee repo's naast elkaar. De kruisverwijzing die daarvoor nodig was (Referenties/Umbrel-appstore.md in de oude EvoluRelay-repo) is verdwenen; wat daarin stond over de plekken waar de relay een ander geval is, staat nu als ontwerp in het masterplan Umbrelapp §4. Botsende namen kregen een achtervoegsel met de app, en alleen die: Publicatie werd Publicatie-Gate en Publicatie-Relay, CHANGELOG.md werd CHANGELOG-electrum-gate.md. Proefopstelling kreeg 007, tussen de twee bestaande nummers, zodat de bovenkant van de reeks op tier-orde blijft staan. CONTINUE_HERE.md heeft een kolom App, maar de tiers lopen over beide apps heen: er is één volgorde van werken. Electrum Gate gaat naar 0.0.15, want website, repo, support, submission en icon wijzen nu naar UmbrelApps en zonder versieverhoging rolt dat niet uit. De release notes leggen aan de gebruiker uit dat hij de store opnieuw moet toevoegen. Of een geïnstalleerde app een wisseling van store-URL overleeft is nog steeds niet uitgezocht; dat blijkt bij het omzetten. Twee dingen in de plannen van Electrum Gate waren door deze verhuizing niet meer waar en zijn bijgewerkt: de taak "de repo hernoemen" in fase 7 is afgevinkt, en de repo-vorm in PLAN.md §4a toonde nog de store-id electrumtls, die al sinds fase 7 achterhaald was. Tests: 39 goed 0 fout en 54 goed 0 fout, niets overgeslagen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
626 lines
27 KiB
Python
626 lines
27 KiB
Python
"""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 `agent.py.template` rechtstreeks. Dat kan omdat dat bestand geen
|
|
accolade-variabelen bevat en de invulling door umbreld hem dus onveranderd laat;
|
|
de eerste toets hieronder controleert precies dat.
|
|
"""
|
|
|
|
import sys
|
|
|
|
# Vóór de imports, want anders is het te laat: Python legt bytecode naast
|
|
# agent.py.template zodra die geïmporteerd wordt, en die rommel hoort niet in de
|
|
# app-map. 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__))
|
|
TEMPLATE = os.path.join(HERE, os.pardir, "whatsnext-electrum-gate", "agent.py.template")
|
|
|
|
|
|
def load_agent():
|
|
"""Laadt agent.py.template als module.
|
|
|
|
Met een expliciete loader, want importlib kijkt normaal naar de extensie en
|
|
.template staat daar niet tussen.
|
|
"""
|
|
loader = importlib.machinery.SourceFileLoader("gate_agent", TEMPLATE)
|
|
spec = importlib.util.spec_from_file_location("gate_agent", TEMPLATE, 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):
|
|
"""De aanname waar deze hele test op rust."""
|
|
with open(TEMPLATE, "r", encoding="utf-8") as f:
|
|
inhoud = f.read()
|
|
u.check("template bevat geen accolade-variabelen",
|
|
"${" not in inhoud,
|
|
"umbreld 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())
|