Eén app store, twee apps
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>
This commit is contained in:
Harmen
@@ -0,0 +1,625 @@
|
||||
"""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())
|
||||
@@ -0,0 +1,400 @@
|
||||
"""Toetst dat de pagina omhoog komt zonder certificaat.
|
||||
|
||||
Waarom deze test bestaat. In 0.0.3 startte de app niet: het stream-blok van
|
||||
nginx had een 'listen 50022 ssl' met een include naar de cert.conf van de agent,
|
||||
en nginx weigert te starten als dat certificaat er niet is. Daardoor kwam ook de
|
||||
web-UI niet omhoog, en dat is precies de pagina waarop je een certificaat kiest.
|
||||
Wie geen reverse proxy draait, of twee kandidaten heeft, zag dus niets en kon
|
||||
niets. De symptomen wezen ergens anders heen: de app_proxy meldde alleen dat de
|
||||
server niet te bereiken was.
|
||||
|
||||
De reparatie is een verplaatsing, en die is niet aan de code af te lezen: het
|
||||
stream-blok staat nu in stream.conf.template en wordt door het command-blok van
|
||||
de compose pas in /var/lib/gate/tls/ gezet als cert.conf bestaat. nginx.conf
|
||||
haalt die map op met een jokerteken, en een jokerteken dat niets matcht is voor
|
||||
nginx geen fout. Eén iemand die dat blok "netjes" terugzet in nginx.conf en de
|
||||
klem is terug, zonder dat een van de andere toetsen iets merkt.
|
||||
|
||||
Draaien:
|
||||
|
||||
python tests/test_server_start_zonder_certificaat.py
|
||||
|
||||
Er staan inmiddels een paar toetsen bij die niet over het starten gaan maar wel
|
||||
over dezelfde soort fout: een die niets meldt en pas opvalt als iemand het toevallig
|
||||
ziet. De verbindingsteller die stil nul telt, een mount die weer alleen-lezen wordt,
|
||||
en twee taglines die uit elkaar lopen. Ze staan hier en niet in een derde bestand,
|
||||
want de suite heeft geen runner en een bestand dat niemand aanroept toetst niets.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
sys.dont_write_bytecode = True
|
||||
|
||||
import os # noqa: E402
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
APP = os.path.join(HERE, os.pardir, "whatsnext-electrum-gate")
|
||||
|
||||
NGINX_CONF = os.path.join(APP, "nginx.conf.template")
|
||||
STREAM_CONF = os.path.join(APP, "stream.conf.template")
|
||||
COMPOSE = os.path.join(APP, "docker-compose.yml")
|
||||
MANIFEST = os.path.join(APP, "umbrel-app.yml")
|
||||
PAGINA = os.path.join(APP, "index.html.template")
|
||||
|
||||
|
||||
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 lees(pad):
|
||||
with open(pad, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
def zonder_commentaar(tekst):
|
||||
"""De regels die nginx daadwerkelijk uitvoert.
|
||||
|
||||
Zonder dit zou elke toets hieronder afgaan op de uitleg erboven, en die
|
||||
noemt juist de dingen die er niet meer mogen staan.
|
||||
"""
|
||||
regels = []
|
||||
for regel in tekst.splitlines():
|
||||
kaal = regel.strip()
|
||||
if not kaal or kaal.startswith("#"):
|
||||
continue
|
||||
regels.append(kaal)
|
||||
return regels
|
||||
|
||||
|
||||
def test_pagina_hangt_niet_aan_een_certificaat(u):
|
||||
"""Geen TLS-directive in nginx.conf, want die blokkeert de start."""
|
||||
regels = zonder_commentaar(lees(NGINX_CONF))
|
||||
|
||||
u.check("nginx.conf heeft geen listen met ssl",
|
||||
not any("listen" in r and "ssl" in r for r in regels),
|
||||
"een 'listen ssl' zonder certificaat laat nginx niet starten, "
|
||||
"en dan is er ook geen pagina om er een te kiezen")
|
||||
|
||||
u.check("nginx.conf includet cert.conf niet rechtstreeks",
|
||||
not any("cert.conf" in r for r in regels),
|
||||
"een include van een bestand dat er nog niet is, is een startfout")
|
||||
|
||||
u.check("nginx.conf haalt het stream-blok met een jokerteken op",
|
||||
any(r.startswith("include") and "/var/lib/gate/tls/*" in r
|
||||
for r in regels),
|
||||
"zonder jokerteken is een lege map alsnog een startfout")
|
||||
|
||||
u.check("de pagina zelf luistert nog wel",
|
||||
any(r.startswith("listen 80") for r in regels))
|
||||
|
||||
|
||||
def test_stream_blok_staat_apart(u):
|
||||
"""Het TLS-deel is compleet, en het is een template."""
|
||||
u.check("stream.conf.template bestaat", os.path.isfile(STREAM_CONF),
|
||||
"anders komt het bij een update niet mee: alleen *.template, "
|
||||
"docker-compose.yml, exports.sh, torrc, hooks en umbrel-app.yml "
|
||||
"worden ververst")
|
||||
if not os.path.isfile(STREAM_CONF):
|
||||
return
|
||||
|
||||
tekst = lees(STREAM_CONF)
|
||||
regels = zonder_commentaar(tekst)
|
||||
|
||||
u.check("stream.conf luistert op 50022 met ssl",
|
||||
any("listen 50022 ssl" in r for r in regels))
|
||||
u.check("stream.conf includet de cert.conf van de agent",
|
||||
any("include /var/lib/gate/cert.conf" in r for r in regels))
|
||||
u.check("stream.conf includet het log_format",
|
||||
any("include /var/lib/gate/stream-log.conf" in r for r in regels))
|
||||
u.check("stream.conf verwijst naar de backend uit de omgeving",
|
||||
any("proxy_pass ${APP_ELECTRS_NODE_IP}" in r for r in regels))
|
||||
|
||||
# Dezelfde architectuurregel als voor de andere templates: umbreld vult elke
|
||||
# accolade-variabele in, dus een nginx-variabele met een dollarteken wordt
|
||||
# hier stil leeggemaakt. Vandaar dat het log_format uit een ander bestand
|
||||
# komt.
|
||||
losse_dollars = []
|
||||
for regel in regels:
|
||||
pos = regel.find("$")
|
||||
while pos >= 0:
|
||||
if regel[pos + 1:pos + 2] != "{":
|
||||
losse_dollars.append(regel)
|
||||
break
|
||||
pos = regel.find("$", pos + 1)
|
||||
u.check("stream.conf heeft geen nginx-variabelen",
|
||||
not losse_dollars,
|
||||
"de template-invulling maakt die leeg: %r" % losse_dollars[:2])
|
||||
|
||||
|
||||
def test_compose_wacht_niet_op_de_agent(u):
|
||||
"""Het command-blok start nginx altijd, en zet TLS erbij als het kan."""
|
||||
tekst = lees(COMPOSE)
|
||||
regels = zonder_commentaar(tekst)
|
||||
|
||||
u.check("geen wachtlus op cert.conf",
|
||||
not any("! -f /var/lib/gate/cert.conf" in r and "while" in r
|
||||
for r in regels),
|
||||
"dat is precies de klem uit 0.0.3: geen certificaat, geen nginx, "
|
||||
"geen pagina, dus geen manier om een certificaat te kiezen")
|
||||
|
||||
u.check("stream.conf is in de server gemount",
|
||||
any("stream.conf:/etc/nginx/stream.conf" in r for r in regels),
|
||||
"anders is er niets om in /var/lib/gate/tls/ te zetten")
|
||||
|
||||
u.check("het stream-blok wordt weggezet als cert.conf bestaat",
|
||||
any("cp /etc/nginx/stream.conf /var/lib/gate/tls/" in r
|
||||
for r in regels))
|
||||
|
||||
u.check("en weggehaald als cert.conf er niet is",
|
||||
any("rm -f /var/lib/gate/tls/stream.conf" in r for r in regels),
|
||||
"een ingetrokken keuze moet de poort ook echt sluiten")
|
||||
|
||||
u.check("een mislukte herlading breekt de wachtlus niet af",
|
||||
any("if ! nginx -s reload" in r for r in regels),
|
||||
"onder 'set -e' zou de lus verdwijnen en daarna pikt niets meer "
|
||||
"een wijziging op")
|
||||
|
||||
|
||||
def test_uploadpad(u):
|
||||
"""De drie dingen buiten de agent die uploaden mogelijk maken.
|
||||
|
||||
Alle drie kunnen ze stil wegvallen: een mount die weer `:ro` wordt geeft een
|
||||
foutmelding pas als iemand iets uploadt, en een te kleine limiet in nginx
|
||||
geeft een 413 die niets uitlegt.
|
||||
"""
|
||||
compose = zonder_commentaar(lees(COMPOSE))
|
||||
nginx = zonder_commentaar(lees(NGINX_CONF))
|
||||
|
||||
eigen = [r for r in compose if "/certs/own" in r and r.startswith("-")]
|
||||
u.check("de eigen certificaatmap is twee keer gemount",
|
||||
len(eigen) == 2, "gevonden: %r" % eigen)
|
||||
u.check("en precies één daarvan is beschrijfbaar",
|
||||
len([r for r in eigen if not r.endswith(":ro")]) == 1,
|
||||
"zonder :ro kan de agent niets neerzetten, met :ro overal ook niet: %r"
|
||||
% eigen)
|
||||
|
||||
u.check("GATE_UPLOAD_DIR staat in de omgeving van de agent",
|
||||
any(r.startswith("GATE_UPLOAD_DIR:") for r in compose),
|
||||
"leeg of afwezig betekent: uploaden staat uit")
|
||||
|
||||
u.check("nginx heeft een eigen locatie voor de upload",
|
||||
any("location = /api/certificate/upload" in r for r in nginx))
|
||||
# De limiet moet groter zijn dan die van /api/, anders kapt nginx het verzoek
|
||||
# af voordat de agent er iets over kan zeggen.
|
||||
limieten = []
|
||||
for regel in nginx:
|
||||
if regel.startswith("client_max_body_size"):
|
||||
limieten.append(regel.rstrip(";").split()[-1])
|
||||
u.check("en een grotere limiet dan het gewone api-pad",
|
||||
len(limieten) == 2 and limieten[0] != limieten[1],
|
||||
"gevonden limieten: %r" % limieten)
|
||||
|
||||
|
||||
def test_verbindingsteller(u):
|
||||
"""De teller in de nginx-container, en de poort in hex.
|
||||
|
||||
Hoort hier omdat het om hetzelfde `command`-blok gaat. Het gevaarlijke deel
|
||||
is de hex: de teller zoekt in /proc/net/tcp op de poort in hexadecimale vorm,
|
||||
en verandert de TLS-poort ooit, dan telt hij stil nul. Een teller die altijd
|
||||
nul zegt, ziet eruit als "geen wallet verbonden" en niet als een fout.
|
||||
"""
|
||||
tekst = lees(COMPOSE)
|
||||
regels = zonder_commentaar(tekst)
|
||||
|
||||
teller = [r for r in regels if "/proc/net/tcp" in r]
|
||||
u.check("de teller leest /proc/net/tcp", bool(teller),
|
||||
"dat is de enige plek waar de open verbindingen van deze container "
|
||||
"staan; de agent zit in een andere netwerk-namespace")
|
||||
|
||||
u.check("en schrijft het aantal weg voor de agent",
|
||||
any("/var/lib/gate/sessions" in r for r in regels))
|
||||
|
||||
# De poort uit de omgeving van de agent is de waarheid; de teller moet
|
||||
# dezelfde poort in hex zoeken.
|
||||
poorten = [r for r in regels if r.startswith("GATE_TLS_PORT:")]
|
||||
u.check("GATE_TLS_PORT staat in de compose", len(poorten) == 1,
|
||||
"gevonden: %r" % poorten)
|
||||
if not poorten or not teller:
|
||||
return
|
||||
|
||||
poort = int(poorten[0].split(":", 1)[1].strip().strip('"'))
|
||||
hexpoort = "%X" % poort
|
||||
# Over de regels heen zoeken, want de awk-regel staat achter een
|
||||
# backslash-vervolg en dan valt de hex op een eigen regel.
|
||||
u.check("de teller zoekt de TLS-poort in hex (%d is %s)" % (poort, hexpoort),
|
||||
(":" + hexpoort) in " ".join(regels),
|
||||
"nergens in het command-blok staat :%s, dus telt de teller nul"
|
||||
% hexpoort)
|
||||
|
||||
|
||||
def test_data_onder_data(u):
|
||||
"""Alles wat de app zelf schrijft staat onder `data/`.
|
||||
|
||||
De conventie van de officiele appstore, afgelezen aan echte apps: electrs
|
||||
mount `${APP_DATA_DIR}/data/electrs`, mempool `${APP_DATA_DIR}/data`. Deze app
|
||||
had `runtime/` en `certs/` naast de templates staan. Verplaatst op 20-08-2026
|
||||
met het oog op publicatie.
|
||||
|
||||
Waarom er een toets op staat: bij de volgende mount die iemand toevoegt is dit
|
||||
precies het detail dat je vergeet, en niets gaat er stuk van. Het valt pas op
|
||||
bij het inleveren.
|
||||
"""
|
||||
regels = zonder_commentaar(lees(COMPOSE))
|
||||
mounts = [r[1:].strip() for r in regels
|
||||
if r.startswith("-") and "${APP_DATA_DIR}" in r]
|
||||
u.check("er zijn mounts uit de app-datamap", bool(mounts))
|
||||
|
||||
fout = []
|
||||
for mount in mounts:
|
||||
host = mount.strip('"').split(":")[0].replace("${APP_DATA_DIR}/", "")
|
||||
# Een bestand naast de compose mag: dat zijn de ingevulde templates en het
|
||||
# icoon, en die hóren daar omdat de whitelist ze daar verft. Een map die de
|
||||
# app zelf vult, hoort onder data/.
|
||||
if "." in os.path.basename(host):
|
||||
continue
|
||||
if not host.startswith("data/"):
|
||||
fout.append(host)
|
||||
u.check("elke map die de app zelf vult staat onder data/",
|
||||
not fout, "deze niet: %r" % fout)
|
||||
|
||||
u.check("de gedeelde toestand staat onder data/",
|
||||
any("${APP_DATA_DIR}/data/runtime:" in m for m in mounts))
|
||||
u.check("de eigen certificaatmap staat onder data/",
|
||||
any("${APP_DATA_DIR}/data/certs:" in m for m in mounts))
|
||||
|
||||
# En het manifest moet die paden ook noemen, anders wijst backupIgnore naar
|
||||
# iets wat niet bestaat en gaat de sessielog alsnog mee in elke back-up.
|
||||
manifest = lees(MANIFEST)
|
||||
u.check("backupIgnore staat in het manifest", "backupIgnore:" in manifest)
|
||||
for pad in ("data/runtime/stream.log", "data/runtime/status.json"):
|
||||
u.check("backupIgnore noemt %s" % pad, pad in manifest)
|
||||
|
||||
|
||||
# De volgorde die de packaging-documentatie van de officiele appstore voorschrijft.
|
||||
# Wat de spec niet noemt, zoals icon en backupIgnore, hoort daarachter en niet
|
||||
# ertussen: dan is de kop van het manifest letterlijk goed.
|
||||
MANIFEST_ORDE = [
|
||||
"manifestVersion", "id", "category", "name", "version", "tagline",
|
||||
"description", "releaseNotes", "developer", "website", "dependencies",
|
||||
"repo", "support", "port", "gallery", "path",
|
||||
]
|
||||
|
||||
|
||||
def manifest_velden():
|
||||
"""De veldnamen op het hoogste niveau, in de volgorde van het bestand."""
|
||||
velden = []
|
||||
for regel in lees(MANIFEST).splitlines():
|
||||
if not regel or regel[0] in " #-":
|
||||
continue
|
||||
if ":" in regel:
|
||||
velden.append(regel.split(":", 1)[0])
|
||||
return velden
|
||||
|
||||
|
||||
def test_manifest_volgorde(u):
|
||||
"""De velden staan in de voorgeschreven volgorde.
|
||||
|
||||
Geen smaak: de officiele appstore schrijft deze volgorde voor. Een toets
|
||||
hierop omdat een nieuw veld standaard onderaan of middenin belandt, en er
|
||||
niets van stukgaat; het valt pas op bij het inleveren.
|
||||
"""
|
||||
velden = manifest_velden()
|
||||
kern = [v for v in velden if v in MANIFEST_ORDE]
|
||||
|
||||
ontbreekt = [v for v in MANIFEST_ORDE if v not in velden]
|
||||
u.check("alle voorgeschreven velden staan in het manifest",
|
||||
not ontbreekt, "mist: %r" % ontbreekt)
|
||||
u.check("en in de voorgeschreven volgorde",
|
||||
kern == MANIFEST_ORDE, "de volgorde is nu: %r" % kern)
|
||||
|
||||
# De extra's horen achter de voorgeschreven reeks. Zo is de kop van het
|
||||
# bestand letterlijk op orde en hoeft er bij inlevering alleen iets weg.
|
||||
if kern == MANIFEST_ORDE and velden:
|
||||
laatste_kern = velden.index(MANIFEST_ORDE[-1])
|
||||
extra_ervoor = [v for v in velden[:laatste_kern] if v not in MANIFEST_ORDE]
|
||||
u.check("en wat de spec niet noemt staat erachter",
|
||||
not extra_ervoor, "deze staan ertussen: %r" % extra_ervoor)
|
||||
|
||||
u.check("de lege standaard-inloggegevens zijn eruit",
|
||||
"defaultUsername" not in velden and "defaultPassword" not in velden,
|
||||
"deze app heeft geen eigen inlog, dus een leeg veld zegt niets")
|
||||
|
||||
|
||||
def test_releasenotes_hebben_een_geschiedenis(u):
|
||||
"""Eén verhaal over deze versie, dan één regel per eerdere versie.
|
||||
|
||||
Deze toets bestaat om een echte fout: drie versies achter elkaar kwam er een
|
||||
nieuwe kop bovenop de release notes terwijl de oude tekst eronder bleef staan.
|
||||
In 0.0.9 stond er daardoor drie keer "Earlier releases" en twee keer dezelfde
|
||||
regel over 0.0.4. Dat is zichtbare tekst in de appstore, en niemand leest zijn
|
||||
eigen release notes nog een keer na.
|
||||
"""
|
||||
tekst = lees(MANIFEST)
|
||||
u.check("er staat precies een keer een geschiedenis-kop",
|
||||
tekst.count("Earlier releases:") == 1,
|
||||
"gevonden: %d keer" % tekst.count("Earlier releases:"))
|
||||
|
||||
# Elke eerdere versie mag maar een keer genoemd worden.
|
||||
dubbel = []
|
||||
for versie in ("0.0.4", "0.0.5", "0.0.6", "0.0.7", "0.0.8"):
|
||||
aantal = tekst.count("\n " + versie + " ")
|
||||
if aantal > 1:
|
||||
dubbel.append((versie, aantal))
|
||||
u.check("en geen versie wordt twee keer beschreven",
|
||||
not dubbel, "dubbel: %r" % dubbel)
|
||||
|
||||
|
||||
def test_tagline_is_overal_dezelfde(u):
|
||||
"""De regel onder de app-naam staat op twee plekken en moet gelijk zijn.
|
||||
|
||||
Besloten door de gebruiker op 20-08-2026, nadat hij zag dat de winkel en de
|
||||
pagina iets anders zeiden. Eén app, één regel. Deze toets bestaat omdat de
|
||||
twee plekken ver uit elkaar staan: niemand die de tagline in het manifest
|
||||
aanpast, opent daarna de pagina.
|
||||
"""
|
||||
tagline = None
|
||||
for regel in lees(MANIFEST).splitlines():
|
||||
if regel.startswith("tagline:"):
|
||||
tagline = regel.split(":", 1)[1].strip().strip('"')
|
||||
break
|
||||
|
||||
u.check("het manifest heeft een tagline", bool(tagline))
|
||||
if not tagline:
|
||||
return
|
||||
|
||||
u.check("en de pagina gebruikt woordelijk dezelfde regel",
|
||||
tagline in lees(PAGINA),
|
||||
"de pagina noemt hem niet: %r" % tagline)
|
||||
|
||||
|
||||
def main():
|
||||
u = Uitslag()
|
||||
test_pagina_hangt_niet_aan_een_certificaat(u)
|
||||
test_stream_blok_staat_apart(u)
|
||||
test_compose_wacht_niet_op_de_agent(u)
|
||||
test_verbindingsteller(u)
|
||||
test_uploadpad(u)
|
||||
test_data_onder_data(u)
|
||||
test_manifest_volgorde(u)
|
||||
test_releasenotes_hebben_een_geschiedenis(u)
|
||||
test_tagline_is_overal_dezelfde(u)
|
||||
return u.rapport()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user