sysop/UmbrelApps
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
498b0fe69574e276e6dc7e81bb82c9718914b6d4
UmbrelApps/whatsnext-electrum-gate/agent.py.template
T
HarmenandClaude Opus 5 498b0fe695 De nieuwe widget vond meteen een fout, en die zat in de widget zelf 
De Status-widget meldde "Not answering" met Errno 111 op een gateway die gewoon
werkte. Die foutmelding was het bewijs: geweigerd betekent dat de naam oploste,
alleen niet naar ons. GATE_TLS_HOST stond op de servicenaam "server", en dat is
een naam die meer apps op een Umbrel gebruiken; op een gedeeld netwerk is het
dus een gok wie je krijgt. Nu de volledige containernaam, precies de vorm die
umbrelOS van APP_HOST verlangt en die twee regels hoger in dezelfde compose al
stond.

Wat dit zegt over de aanname van gisteren: er stond dat dit "dezelfde weg is die
nginx andersom gebruikt". Dat klopte voor de richting maar niet voor de naam, en
het was opgeschreven als redenering in plaats van als meting.

De rest zijn bijstellingen van de gebruiker na het bekijken van 0.0.17. Het menu
is drie kale punten met een accent bij hover in plaats van een knop, de eerste
regel heet Settings zonder de certificaatnaam erachter, de dialoog opent in het
midden in plaats van linksboven, de widgets zijn hoger met hun waarde langs de
onderrand, reactietijd en blokhoogte zijn omgedraaid zodat de reactietijd naast
de server staat waar hij over gaat, en het log is half zo hoog.

De manifest-toets ving onderweg een echte fout: bij het herschrijven van de
release notes was de sleutel releaseNotes zelf weggeknipt.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-27 16:04:13 +02:00

1190 lines
46 KiB
Plaintext
Raw Blame History

"""Electrum Gate agent: schrijft status.json en neemt de certificaatkeuze aan.
Waarom dit bestand een .template is en er toch geen accolade-variabelen in staan.
umbreld ververst bij een update alleen een whitelist van bestanden en *.template
staat daarin, dus een gewoon script naast de compose zou na een update ongewijzigd
blijven staan zonder dat er iets gemeld wordt. Maar umbreld vult bij elke start
ook elke accolade-variabele in, en die zou hier Python-code slopen. Daarom haalt
dit script zijn instellingen uit de omgeving, en zet de compose ze daar neer. Het
ingevulde agent.py is daardoor byte voor byte gelijk aan dit bestand, en dat is
precies wat de test in tests/ erover aanneemt.
Alleen de standaardbibliotheek. De app draait op een kant-en-klare python-image
en installeert bij het starten niets; dat was een van de dingen die de vorige
opzet fout deed.
"""
import calendar
import json
import os
import re
import socket
import ssl
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
# --- Instellingen uit de omgeving -------------------------------------------
STATE_DIR = os.environ.get("GATE_STATE_DIR", "/var/lib/gate")
CONFIG_DIR = os.path.join(STATE_DIR, "config")
STATUS_FILE = os.path.join(STATE_DIR, "status.json")
SELECTED_FILE = os.path.join(CONFIG_DIR, "selected-cert")
CERT_CONF = os.path.join(STATE_DIR, "cert.conf")
RELOAD_FLAG = os.path.join(STATE_DIR, "reload")
STREAM_LOG = os.path.join(STATE_DIR, "stream.log")
SESSIONS_FILE = os.path.join(STATE_DIR, "sessions")
# De map waarin een upload terechtkomt. Expliciet uit de omgeving en niet
# afgeleid uit GATE_CERT_SOURCES: van die bronnen is er precies één beschrijfbaar,
# en welke dat is hoort in de compose te staan naast de mount die het mogelijk
# maakt. Leeg betekent: uploaden staat uit.
UPLOAD_DIR = os.environ.get("GATE_UPLOAD_DIR", "")
UPLOAD_MAX = 96 * 1024
ELECTRUM_HOST = os.environ.get("GATE_ELECTRUM_HOST", "")
ELECTRUM_PORT = int(os.environ.get("GATE_ELECTRUM_PORT", "50001"))
TLS_PORT = int(os.environ.get("GATE_TLS_PORT", "50022"))
# De zelfcontrole: de agent verbindt met de eigen TLS-poort. Zie het plan
# Webinterface, OPEN.md punt 5.
#
# GATE_TLS_HOST is de nginx-container hiernaast. Niet localhost: dat is een andere
# container en dus een andere netwerk-namespace.
#
# De compose zet hier de volledige containernaam neer en niet de servicenaam. Die
# stond er eerst, en op het apparaat leverde dat `[Errno 111] Connection refused`
# op: de naam loste op, maar naar een andere container. Zie de toelichting in
# docker-compose.yml. De standaard hieronder is alleen de terugval voor een
# opstelling zonder die variabele.
#
# Waarom deze controle niet elke ronde draait: elke verbinding met poort 50022 is
# voor nginx een gewone sessie en levert dus een regel in het activiteitenlog op.
# Elke minuut een eigen regel maakt dat log onbruikbaar. Vijf minuten is vaak
# genoeg voor een waarde die zelden verandert, en na een certificaatwissel wordt
# er sowieso gemeten.
TLS_HOST = os.environ.get("GATE_TLS_HOST", "server")
SELF_CHECK_INTERVAL = int(os.environ.get("GATE_SELF_CHECK_INTERVAL", "300"))
INTERVAL = int(os.environ.get("GATE_INTERVAL", "60"))
API_PORT = int(os.environ.get("GATE_API_PORT", "8000"))
HISTORY_SECONDS = 24 * 3600
LOG_SECONDS = 24 * 3600
# Bronnaam naar map. Een bron die niet gemount is, bestaat simpelweg niet en
# wordt overgeslagen; dat is geen fout maar de normale toestand voor een
# reverse proxy die de gebruiker niet draait.
SOURCES = []
for stuk in os.environ.get("GATE_CERT_SOURCES", "").split(","):
stuk = stuk.strip()
if not stuk or "=" not in stuk:
continue
naam, pad = stuk.split("=", 1)
SOURCES.append((naam.strip(), pad.strip()))
# Vertaaltabel van backend-adres naar naam. Bewust niet meer dan dit: staat een
# adres er niet in, dan toont de pagina het adres zonder naam. Zie het plan
# Webinterface, PLAN.md paragraaf 4c: liever minder informatief dan onwaar.
BACKEND_NAMES = {
"10.21.21.10": "Electrs",
"10.21.21.200": "Fulcrum",
"10.21.21.199": "ElectrumX",
}
def log(bericht):
"""Naar stdout, want dat is wat docker logs laat zien."""
sys.stdout.write("[gate] " + bericht + "\n")
sys.stdout.flush()
# --- X.509 lezen ------------------------------------------------------------
# Een minimale DER-lezer voor de twee dingen die de app van een certificaat wil
# weten: tot wanneer het geldig is, en voor welke namen. De standaardbibliotheek
# heeft hier geen parser voor, en iets bijinstalleren is precies wat deze opzet
# niet doet.
TAG_SEQUENCE = 0x30
TAG_SET = 0x31
TAG_UTCTIME = 0x17
TAG_GENERALIZEDTIME = 0x18
TAG_OID = 0x06
TAG_OCTETSTRING = 0x04
TAG_DNSNAME = 0x82
TAG_EXTENSIONS = 0xA3
OID_SAN = bytes([0x55, 0x1D, 0x11])
OID_CN = bytes([0x55, 0x04, 0x03])
def _tlv(buf, pos):
"""Leest een DER-element en geeft (tag, inhoud, positie erna)."""
tag = buf[pos]
first = buf[pos + 1]
if first < 0x80:
length = first
start = pos + 2
else:
count = first & 0x7F
if count == 0 or count > 4:
raise ValueError("onbruikbaar lengteveld")
length = int.from_bytes(buf[pos + 2:pos + 2 + count], "big")
start = pos + 2 + count
end = start + length
if end > len(buf):
raise ValueError("element loopt buiten de buffer")
return tag, buf[start:end], end
def _children(buf):
"""Alle directe elementen in de inhoud van een samengesteld element."""
out = []
pos = 0
while pos < len(buf):
tag, body, pos = _tlv(buf, pos)
out.append((tag, body))
return out
def _time_to_unix(tag, body):
"""UTCTime of GeneralizedTime naar unix.
Alleen de Zulu-vorm, want dat schrijft DER voor. Een tijd met een
tijdzone-offset hoort hier hard te falen in plaats van er stil naast te
zitten: een certificaatdatum die een uur verkeerd is, valt nooit op.
"""
text = body.decode("ascii")
if not text.endswith("Z"):
raise ValueError("tijd zonder Z: " + text)
text = text[:-1]
if tag == TAG_UTCTIME:
if len(text) != 12:
raise ValueError("UTCTime met onverwachte lengte: " + text)
year = int(text[0:2])
# De eeuwregel uit RFC 5280: 50 en hoger hoort bij de vorige eeuw.
year += 1900 if year >= 50 else 2000
rest = text[2:]
elif tag == TAG_GENERALIZEDTIME:
if len(text) != 14:
raise ValueError("GeneralizedTime met onverwachte lengte: " + text)
year = int(text[0:4])
rest = text[4:]
else:
raise ValueError("geen tijd-tag: " + hex(tag))
return calendar.timegm((
year, int(rest[0:2]), int(rest[2:4]),
int(rest[4:6]), int(rest[6:8]), int(rest[8:10]), 0, 0, 0,
))
def _tbs(der):
"""De tbsCertificate-inhoud uit een DER-certificaat."""
tag, body, _ = _tlv(der, 0)
if tag != TAG_SEQUENCE:
raise ValueError("buitenste element is geen SEQUENCE")
tag, tbs, _ = _tlv(body, 0)
if tag != TAG_SEQUENCE:
raise ValueError("tbsCertificate is geen SEQUENCE")
return tbs
def not_after(der):
"""Einddatum als unix-tijdstip.
Zoekt het eerste SEQUENCE in tbsCertificate dat uit precies twee
tijd-elementen bestaat; dat is de validity en die vorm komt nergens anders
voor. Positioneel tellen zou ook kunnen, maar dan zit een certificaat met een
onverwacht optioneel veld er stil naast.
"""
for tag, body in _children(_tbs(der)):
if tag != TAG_SEQUENCE:
continue
try:
inner = _children(body)
except ValueError:
continue
if len(inner) != 2:
continue
if inner[0][0] not in (TAG_UTCTIME, TAG_GENERALIZEDTIME):
continue
if inner[1][0] not in (TAG_UTCTIME, TAG_GENERALIZEDTIME):
continue
return _time_to_unix(inner[1][0], inner[1][1])
raise ValueError("geen validity gevonden")
def _extensions(tbs):
"""De Extension-elementen, of een lege lijst."""
for tag, body in _children(tbs):
if tag != TAG_EXTENSIONS:
continue
outer = _children(body)
if not outer or outer[0][0] != TAG_SEQUENCE:
continue
return _children(outer[0][1])
return []
def dns_names(der):
"""Alle dNSName-waarden uit subjectAltName."""
names = []
for tag, ext in _extensions(_tbs(der)):
if tag != TAG_SEQUENCE:
continue
parts = _children(ext)
if not parts or parts[0][0] != TAG_OID or parts[0][1] != OID_SAN:
continue
payload = None
for ptag, pbody in parts[1:]:
if ptag == TAG_OCTETSTRING:
payload = pbody
if payload is None:
continue
gtag, gbody, _ = _tlv(payload, 0)
if gtag != TAG_SEQUENCE:
continue
for ntag, nbody in _children(gbody):
if ntag == TAG_DNSNAME:
names.append(nbody.decode("ascii", "replace"))
return names
def common_name(der):
"""De commonName uit het subject, of None.
Het subject is de tweede Name in tbsCertificate; de eerste is de issuer. Een
Name is te herkennen aan een SEQUENCE die alleen SET-elementen bevat, en dat
onderscheidt hem van de validity en van de sleutelinformatie.
"""
names = []
for tag, body in _children(_tbs(der)):
if tag != TAG_SEQUENCE:
continue
try:
kids = _children(body)
except ValueError:
continue
if kids and all(t == TAG_SET for t, _ in kids):
names.append(kids)
if len(names) < 2:
return None
for kids in names[1:]:
for _, setbody in kids:
for atag, attr in _children(setbody):
if atag != TAG_SEQUENCE:
continue
pair = _children(attr)
if len(pair) == 2 and pair[0][0] == TAG_OID and pair[0][1] == OID_CN:
return pair[1][1].decode("utf-8", "replace")
return None
def read_pem(path):
"""Het eerste certificaat uit een PEM-bestand, als DER."""
with open(path, "rb") as f:
return der_from_pem(f.read().decode("ascii", "replace"))
def der_from_pem(data):
"""Het eerste certificaat uit PEM-tekst, als DER.
Los van read_pem omdat een upload nog geen bestand is: die tekst komt uit een
verzoek en wordt pas weggeschreven als hij deze lezer en de sleutelcontrole
doorstaat.
"""
import base64
start = data.find("-----BEGIN CERTIFICATE-----")
if start < 0:
raise ValueError("geen certificaat in " + path)
end = data.find("-----END CERTIFICATE-----", start)
if end < 0:
raise ValueError("onafgesloten certificaat in " + path)
body = data[start + len("-----BEGIN CERTIFICATE-----"):end]
return base64.b64decode("".join(body.split()))
# --- Certificaten vinden ----------------------------------------------------
def _pairs_in(directory):
"""Kandidaat-paren in een map: (naam, certpad, keypad).
Drie vormen, want de bronnen benoemen verschillend. Zoraxy schrijft
naam.pem naast naam.key. Nginx Proxy Manager zet per certificaat een map met
fullchain.pem en privkey.pem, dus zonder het domein in de naam. Een eigen
upload mag beide.
"""
found = []
try:
entries = sorted(os.listdir(directory))
except OSError:
return found
for entry in entries:
full = os.path.join(directory, entry)
if os.path.isdir(full):
cert = os.path.join(full, "fullchain.pem")
key = os.path.join(full, "privkey.pem")
if os.path.isfile(cert) and os.path.isfile(key):
found.append((entry, cert, key))
continue
stem, ext = os.path.splitext(entry)
if ext not in (".pem", ".crt"):
continue
if stem in ("fullchain", "chain", "cert"):
continue
for keyname in (stem + ".key", stem + ".pem.key"):
key = os.path.join(directory, keyname)
if os.path.isfile(key):
found.append((stem, full, key))
break
return found
def scan_certificates():
"""Alles wat in de gemounte bronnen staat, met datum en domein."""
out = []
for source, directory in SOURCES:
for name, cert, key in _pairs_in(directory):
item = {
"id": source + "/" + name,
"source": source,
"name": name,
"cert": cert,
"key": key,
}
try:
der = read_pem(cert)
item["not_after"] = not_after(der)
names = dns_names(der)
item["domain"] = names[0] if names else common_name(der)
item["domains"] = names
except Exception as e: # noqa: BLE001 - een onleesbaar certificaat
# mag de rest van de lijst niet meenemen; hij wordt wel getoond,
# want hem verzwijgen maakt een verkeerde keuze onvindbaar.
item["unreadable"] = str(e)
item["domain"] = name
out.append(item)
return out
# --- Een certificaat aannemen -----------------------------------------------
# De pagina kan een certificaat en een sleutel opsturen. Dit is het enige pad
# waarlangs iets van buiten bestanden neerzet, dus hier staat de volledige
# controle, en die weigert liever dan dat hij plaatst.
#
# Waarom dit er is: /certs/own is zonder SSH niet te bereiken, dus die bron was
# alleen bruikbaar voor wie de shell op durft. Dat is de helft van de reden dat
# hij bestaat.
NAME_OK = re.compile(r"[^A-Za-z0-9._-]")
def _file_name(domain):
"""Een bestandsnaam uit de hostnaam van het certificaat zelf.
Niet uit iets wat de gebruiker meestuurt, en dat is de hele reden: een naam
die van buiten komt moet je tegen padtrucs verdedigen, een naam die je uit het
certificaat leest niet. Wat er niet in het alfabet zit gaat eruit, dus een
jokerteken-certificaat voor *.example.org wordt example.org.
"""
naam = NAME_OK.sub("", domain or "").strip(".-")
return naam[:64]
def accept_upload(cert_text, key_text):
"""Controleert een opgestuurd paar en zet het weg. Geeft (id, fout).
De volgorde is die van de foutmeldingen: eerst of het überhaupt een
certificaat is, dan of er een naam in staat, dan of het nog geldig is, en als
laatste de duurste controle, namelijk of de sleutel erbij hoort.
Die laatste is de belangrijkste en gaat via ssl.SSLContext.load_cert_chain.
Dat is dezelfde OpenSSL die nginx straks gebruikt, dus een paar dat hier
doorkomt komt daar ook door. Zonder deze controle levert een verkeerde sleutel
een nginx die niet meer herlaadt, en dat is dezelfde klasse storing als die
waardoor 0.0.3 helemaal niet startte.
"""
if not UPLOAD_DIR:
return None, "uploading is switched off in this installation"
cert_text = (cert_text or "").strip()
key_text = (key_text or "").strip()
if "-----BEGIN CERTIFICATE-----" not in cert_text:
return None, "that file does not contain a certificate"
if "-----BEGIN" not in key_text or "PRIVATE KEY-----" not in key_text:
return None, "that file does not contain a private key"
# Een versleutelde sleutel kan nginx niet openen zonder wachtwoord, en dat
# wachtwoord heeft deze app niet en wil hij ook niet hebben. Twee vormen: de
# PKCS#8-kop en de oudere PEM-kop met Proc-Type.
if "ENCRYPTED PRIVATE KEY" in key_text or "Proc-Type: 4,ENCRYPTED" in key_text:
return None, ("that private key is encrypted; nginx cannot open it "
"without a passphrase")
try:
der = der_from_pem(cert_text)
except Exception as e: # noqa: BLE001 - alles hier betekent hetzelfde
return None, "that certificate cannot be read (%s)" % e
try:
names = dns_names(der)
domain = names[0] if names else common_name(der)
except Exception: # noqa: BLE001
domain = None
naam = _file_name(domain)
if not naam:
return None, ("that certificate has no host name in it, so there is "
"nothing to match a wallet against")
try:
einde = not_after(der)
except Exception as e: # noqa: BLE001
return None, "that certificate has no readable end date (%s)" % e
if einde <= int(time.time()):
return None, "that certificate expired on " + time.strftime(
"%d-%m-%Y", time.gmtime(einde))
# Pas nu naar schijf, en eerst onder een tijdelijke naam. Die naam staat in
# dezelfde map als de eindbestemming en niet in /tmp, want os.replace over een
# bestandssysteemgrens faalt met EXDEV, en /certs/own is een bind-mount. Op
# ".tmp" ziet de scanner ze niet: die kijkt naar .pem en .crt.
os.makedirs(UPLOAD_DIR, exist_ok=True)
doel_cert = os.path.join(UPLOAD_DIR, naam + ".pem")
doel_key = os.path.join(UPLOAD_DIR, naam + ".key")
proef_cert = doel_cert + ".tmp"
proef_key = doel_key + ".tmp"
try:
with open(proef_cert, "w", encoding="ascii") as f:
f.write(cert_text + "\n")
with open(proef_key, "w", encoding="ascii") as f:
f.write(key_text + "\n")
os.chmod(proef_key, 0o600)
try:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
# Een wachtwoord meegeven, ook al hoort er geen bij: zonder dit gaat
# OpenSSL er zelf om vragen, en in een container zonder terminal is
# dat een fout die nergens uitkomt.
ctx.load_cert_chain(proef_cert, proef_key, password=lambda: "")
except Exception as e: # noqa: BLE001 - de reden is voor de pagina, en
# die is bijna altijd "de sleutel hoort niet bij dit certificaat".
return None, ("that key does not go with that certificate, or one of "
"the two is damaged (%s)" % _short(e))
vervangt = os.path.exists(doel_cert)
os.replace(proef_cert, doel_cert)
os.replace(proef_key, doel_key)
finally:
# Wat er na een weigering nog ligt, gaat weg. Anders blijft er bij elke
# mislukte poging een half paar in de map staan.
for pad in (proef_cert, proef_key):
try:
os.unlink(pad)
except OSError:
pass
os.chmod(doel_cert, 0o644)
# De sleutel is het enige geheim dat deze app aanraakt. nginx leest hem als
# root bij het laden van de configuratie, dus 0600 is genoeg.
os.chmod(doel_key, 0o600)
log("upload accepted: %s (%s)" % (naam, "replaced" if vervangt else "new"))
return naam, None
def upload_id(naam):
"""De id waaronder een upload in de lijst komt, of None.
De bronnaam komt uit GATE_CERT_SOURCES en niet uit een letterlijke tekst
hier: staat "Own folder" morgen anders in de compose, dan zou een vaste tekst
een id opleveren die nergens naar wijst, en dat is precies het soort fout dat
pas bij het kiezen opvalt.
"""
for naam_bron, pad in SOURCES:
if pad == UPLOAD_DIR:
return naam_bron + "/" + naam
return None
def _short(e):
"""De laatste, leesbare regel uit een OpenSSL-fout."""
tekst = str(e)
if len(tekst) > 200:
tekst = tekst[:200] + "..."
return tekst
def read_selection():
"""De gekozen id uit het configuratiebestand, of None."""
try:
with open(SELECTED_FILE, "r", encoding="utf-8") as f:
return f.read().strip() or None
except OSError:
return None
def write_selection(value):
os.makedirs(CONFIG_DIR, exist_ok=True)
tmp = SELECTED_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
f.write(value + "\n")
os.replace(tmp, SELECTED_FILE)
def choose(certs):
"""Welk certificaat gebruikt wordt, en waarom.
De keuze van de gebruiker wint, maar alleen als hij naar iets bestaands
wijst. Een onbekende id wordt genegeerd in plaats van gehonoreerd: anders
zou een typefout of een verwijderd certificaat de app laten starten met een
pad dat niet bestaat, en dan is er geen TLS meer.
"""
if not certs:
return None, "no certificates found in any source"
wanted = read_selection()
if wanted:
for c in certs:
if c["id"] == wanted and not c.get("unreadable"):
# Geen reden meesturen, en dat is de reden: er stond "your
# choice", en de gebruiker merkte op dat het altijd zijn keuze is
# (20-08-2026). Dit veld is er om een toestand uit te leggen die
# niet vanzelf spreekt, zoals een app die niets koos of er zelf
# een pakte. Een keuze die je zelf gemaakt hebt, spreekt vanzelf.
return c, None
log("selected certificate %r is missing or unreadable, "
"ignoring the selection" % wanted)
usable = [c for c in certs if not c.get("unreadable") and c.get("not_after")]
if not usable:
return None, "no readable certificate found"
now = int(time.time())
valid = [c for c in usable if c["not_after"] > now]
if not valid:
return None, "every certificate found has expired"
if len(valid) == 1:
return valid[0], "the only certificate found"
# Bewust géén gok bij meerdere kandidaten, ook niet "de nieuwste". Zo staat
# het besloten in het plan Configuratie, paragraaf 4c, en de reden is dat een
# verkeerd certificaat een verbinding oplevert die het lijkt te doen en bij de
# wallet stukloopt op naamverificatie. Dat is veel lastiger te vinden dan een
# app die weigert en zegt waarom. De pagina toont de kandidaten, dus kiezen
# kost één klik.
#
# Alleen het aantal, niet de namen. Hier stond eerst de hele opsomming, en op
# de Umbrel van de gebruiker leverde dat op 20-08-2026 veertien domeinnamen in
# één foutmelding op: een muur tekst die zegt wat de lijst eronder al toont, en
# die lijst is waar je klikt. Het aantal blijft er wel in, want dat verklaart
# waarom de app niets koos.
return None, ("%d certificates found, so it is not clear which one this app "
"should use; pick one below" % len(valid))
def apply_certificate(cert):
"""Schrijft het nginx-fragment en vraagt om een herlading als er iets wijzigt.
nginx herladen kan niet vanuit deze container zonder de Docker-socket, en die
is er bewust uit. Daarom zet deze functie een vlagbestand neer, en herlaadt de
nginx-container zichzelf zodra hij dat ziet.
"""
body = (
"# Door de agent geschreven; niet met de hand bewerken.\n"
"ssl_certificate %s;\n"
"ssl_certificate_key %s;\n"
) % (cert["cert"], cert["key"])
old = None
try:
with open(CERT_CONF, "r", encoding="utf-8") as f:
old = f.read()
except OSError:
pass
if old == body:
return False
tmp = CERT_CONF + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
f.write(body)
os.replace(tmp, CERT_CONF)
with open(RELOAD_FLAG, "w", encoding="utf-8") as f:
f.write(str(int(time.time())) + "\n")
log("certificate changed to %s, reload requested" % cert["id"])
return True
# --- De backend bevragen ----------------------------------------------------
def probe_backend():
"""Blokhoogte en reactietijd van de Electrum-server.
Eén verzoek over een verse verbinding. Het Electrum-protocol is
JSON-RPC met een regeleinde eraan, dus dit is een socket en niets meer.
"""
result = {"address": ELECTRUM_HOST + ":" + str(ELECTRUM_PORT), "reachable": False}
result["name"] = BACKEND_NAMES.get(ELECTRUM_HOST)
if not ELECTRUM_HOST:
result["error"] = "geen backend-adres in de omgeving"
return result
payload = json.dumps({
"id": 0,
"method": "blockchain.headers.subscribe",
"params": [],
}) + "\n"
start = time.monotonic()
try:
with socket.create_connection((ELECTRUM_HOST, ELECTRUM_PORT), timeout=10) as s:
s.sendall(payload.encode("ascii"))
s.settimeout(10)
chunks = []
while True:
chunk = s.recv(4096)
if not chunk:
break
chunks.append(chunk)
if b"\n" in chunk:
break
elapsed = (time.monotonic() - start) * 1000.0
line = b"".join(chunks).split(b"\n", 1)[0]
answer = json.loads(line.decode("utf-8"))
height = (answer.get("result") or {}).get("height")
result["reachable"] = True
result["response_ms"] = int(round(elapsed))
if isinstance(height, int):
result["height"] = height
except Exception as e: # noqa: BLE001 - elke fout betekent hetzelfde voor
# de pagina: de backend antwoordde niet.
result["error"] = str(e)
return result
# Wat er in het activiteitenlog komt als de uitkomst verandert. Engels, want dit
# belandt op de pagina. Alleen bij een wijziging: elke ronde dezelfde regel is
# ruis, en de toestand zelf staat in status.json.
SELF_CHECK_NOTES = {
"ok": "the gateway answered on its own TLS port",
"failed": "the gateway did not answer on its own TLS port",
"wrong-certificate": "the gateway answered with a different certificate than the one selected",
"off": "TLS is off; no certificate selected",
}
def probe_self(active):
"""Antwoordt de eigen TLS-poort, en met het gekozen certificaat?
Dit is de enige controle die gaat over wat deze app zelf doet. De agent kijkt
al of de Electrum-server antwoordt, maar dat zegt niets over de voordeur, en
de voordeur is de hele app. Eén verbinding met een afgemaakte handdruk bewijst
drie dingen tegelijk: nginx luistert, er ligt een certificaat, en het is te
gebruiken.
Bewust géén controle op vertrouwen. `verify_mode` staat op CERT_NONE en de
hostnaam wordt niet nagelopen, en dat is geen slordigheid maar het verschil
tussen deze controle en die van een wallet:
- een gebruiker mag een zelfondertekend certificaat uploaden, dat is een
ondersteunde bron. Verifiëren tegen de certificaatwinkel van het
besturingssysteem zou die opstelling als kapot melden terwijl hij werkt;
- de winkel hoeft in de image niet te bestaan. De bestaande test slaat de
vergelijking met `ssl` om die reden soms over, en een controle die zichzelf
overslaat op het apparaat is geen controle.
Wat er in plaats daarvan gebeurt is scherper: het getoonde certificaat wordt
byte voor byte vergeleken met het gekozen bestand. Dat vangt het geval waar
trust-verificatie juist blind voor is, namelijk dat nginx nog het vórige
certificaat serveert omdat een herlading niet gelukt is.
**Wat dit niet bewijst, en dat is minder dan het plan beloofde.** Er wordt na
de handdruk niets verstuurd, dus de doorverbinding naar de Electrum-server zit
er níet in. Dat is een bewuste ruil: één echt verzoek zou de hele keten
bewijzen, maar dan draagt de sessie bytes, en sessies met bytes worden nooit
uit het activiteitenlog gefilterd. Zie suppress_self_probes: een sessie waarin
iets is doorgegeven kan een storing zijn, en die verbergen is erger dan deze
controle mislopen. De backend wordt bovendien apart bevraagd door
probe_backend; wat tussen wal en schip valt is alleen de proxy_pass zelf.
Evenmin bewezen: de weg van buiten naar binnen. Deze verbinding loopt van
container naar container over het compose-netwerk, dus de poortmapping naar de
host en de doorstuurregel in de router blijven ongetest.
Geeft altijd een dict met een `state` terug, nooit een uitzondering:
- `off` er is geen certificaat gekozen, dus poort 50022 luistert
niet. Geen storing maar de normale begintoestand;
- `ok` handdruk afgemaakt en het gekozen certificaat kwam terug;
- `wrong-certificate` handdruk afgemaakt, maar met een ander certificaat;
- `failed` geen verbinding of geen handdruk, met de reden erbij.
"""
if not active:
return {"at": int(time.time()), "state": "off"}
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
# In deze volgorde: verify_mode op CERT_NONE zetten terwijl check_hostname
# nog aanstaat is een ValueError.
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
# De naam die een wallet ook zou sturen. Er is maar één server-blok, dus
# nginx kiest hier niets op; het houdt de meting alleen gelijk aan de echte
# situatie. Leeg mag niet, dus dan liever niets.
server_name = active.get("domain") or None
start = time.monotonic()
try:
with socket.create_connection((TLS_HOST, TLS_PORT), timeout=10) as raw:
with context.wrap_socket(raw, server_hostname=server_name) as tls:
presented = tls.getpeercert(binary_form=True)
elapsed = int(round((time.monotonic() - start) * 1000.0))
except Exception as e: # noqa: BLE001 - elke fout betekent hetzelfde voor de
# pagina: de voordeur antwoordde niet.
return {"at": int(time.time()), "state": "failed", "error": str(e)}
result = {"at": int(time.time()), "ms": elapsed}
try:
expected = read_pem(active["cert"])
except Exception: # noqa: BLE001 - een onleesbaar bestand is al elders
# gemeld; hier zou het alleen een tweede, vagere melding opleveren.
expected = None
if expected is not None and presented is not None and presented != expected:
result["state"] = "wrong-certificate"
return result
result["state"] = "ok"
return result
# --- Het activiteitenlog ----------------------------------------------------
# nginx schrijft per stream-sessie één regel, en pas bij het sluiten. Een wallet
# houdt zijn verbinding uren open, dus een lopende sessie staat er nog niet in.
# Dat is een bekende beperking en hij staat in het plan als open punt; hem hier
# verzinnen zou erger zijn dan hem missen.
LOG_LINE = re.compile(
r"^(?P<t>\S+)\s+(?P<status>\d+)\s+(?P<recv>\d+)\s+(?P<sent>\d+)\s+(?P<dur>[\d.]+)"
)
def read_stream_log(cutoff):
"""De sessieregels van nginx, omgezet naar logregels voor de pagina."""
out = []
try:
with open(STREAM_LOG, "r", encoding="utf-8", errors="replace") as f:
# Alleen het staartje; de lus draait elke minuut en het bestand
# groeit langzaam, maar ongelimiteerd lezen is een tijdbom.
lines = f.readlines()[-2000:]
except OSError:
return out
for raw in lines:
m = LOG_LINE.match(raw.strip())
if not m:
continue
try:
ended = int(calendar.timegm(time.strptime(
m.group("t")[:19], "%Y-%m-%dT%H:%M:%S")))
except ValueError:
continue
if ended < cutoff:
continue
duration = float(m.group("dur"))
status = m.group("status")
recv = int(m.group("recv"))
sent = int(m.group("sent"))
# Drie soorten, en het onderscheid tussen de laatste twee is er op verzoek
# van de gebruiker (20-08-2026). Een poort die in de router doorgestuurd
# staat wordt gescand, en elke scan leverde een regel "refused" op. Dat
# woord suggereert dat de app iets geweigerd heeft, terwijl er niets
# gebeurd is, en het log liep er vol met rood.
#
# Het onderscheid zit in de bytes en niet in de duur of de status. Deze
# tellers gaan over de doorgegeven verbinding en niet over de TLS-handdruk:
# dat is af te lezen aan de sessies van de gebruiker, waar een mislukte
# verbinding op 0 en 0 uitkwam terwijl er wél een handdruk geprobeerd is.
# Nul beide kanten op betekent dus: er is nooit iets doorgegeven.
#
# Waarom niet op de duur: een scan van tien seconden is nog steeds een
# scan, en een echte sessie die na een halve seconde omvalt is nog steeds
# een storing. De bytes zeggen wat er gebeurd is, de klok niet.
if status == "200":
event = "disconnect"
elif recv == 0 and sent == 0:
event = "probe"
else:
event = "refused"
entry = {
"t": ended,
"event": event,
"seconds": int(round(duration)),
"bytes_in": recv,
"bytes_out": sent,
}
if event == "probe":
# Feitelijk houden: dát het de handdruk was, weten we niet, alleen dat
# er niets doorgegeven is. Bij status 500 is de handdruk de bijna
# zekere oorzaak, maar bijna is hier niet genoeg.
entry["note"] = "nothing exchanged, status " + status
elif event == "refused":
entry["note"] = "session ended with status " + status
out.append(entry)
return out
def suppress_self_probes(entries, probes):
"""Haalt de sessieregels van de eigen zelfcontrole uit het log.
Nodig omdat die controle een echte verbinding met poort 50022 is: nginx ziet
geen verschil met een wallet en schrijft dus een regel. Die regel komt als
`probe` binnen, want er wordt niets doorgegeven, en zonder deze filter zou het
activiteitenlog voor het grootste deel uit onze eigen controles bestaan.
Er is niets in de logregel waaraan onze verbinding te herkennen is, en dat is
met opzet: er staat geen client-adres in en dat blijft zo. Wat er wél is, is
het moment. De agent onthoudt wanneer hij zelf verbond, en dat wordt hier
tegen de eindtijd van de sessie gelegd.
Twee dingen om te weten voordat je hierop leunt:
- **de speling is twee seconden**, want de logregel draagt de eindtijd van de
sessie en de agent noteert het moment vlak erna;
- **er kan een échte scan verdwijnen** die toevallig in datzelfde venster
eindigde. Per onthouden controle wordt hoogstens één regel weggelaten, dus
het blijft bij die ene. Dat is de prijs, en hij is lager dan een log dat
elke vijf minuten over onszelf gaat.
"""
rest = list(probes)
out = []
for e in entries:
if e.get("event") == "probe" and not e.get("bytes_in") and not e.get("bytes_out"):
match = None
for t in rest:
if abs(e.get("t", 0) - t) <= 2:
match = t
break
if match is not None:
rest.remove(match)
continue
out.append(e)
return out
def read_open_sessions():
"""Hoeveel wallet-verbindingen er nu openstaan, of None.
Geteld door de achtergrondlus van de nginx-container en hier alleen gelezen.
Die omweg via een bestand is nodig omdat /proc/net/tcp per
netwerk-namespace geldt: deze container ziet de sockets van nginx niet, hoe
graag hij ook zou willen. Zelfde patroon als de herlaadvlag, en om dezelfde
reden: geen Docker-socket.
None betekent "niet bekend" en niet "nul". Het verschil doet ertoe: een
nginx-container van vóór deze versie schrijft dit bestand niet, en dan hoort
de pagina niets te beweren in plaats van rustig nul te tonen.
"""
try:
with open(SESSIONS_FILE, "r", encoding="utf-8") as f:
waarde = int(f.read().strip())
except (OSError, ValueError):
return None
return waarde if waarde >= 0 else None
# --- De status samenstellen -------------------------------------------------
def read_previous():
try:
with open(STATUS_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except (OSError, ValueError):
return {}
def write_status(status):
tmp = STATUS_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(status, f, separators=(",", ":"))
os.replace(tmp, STATUS_FILE)
def build_status(events):
now = int(time.time())
previous = read_previous()
certs = scan_certificates()
active, why = choose(certs)
reloaded = apply_certificate(active) if active else False
if reloaded:
events.append({"t": now, "event": "reload", "note": "certificate changed"})
# De open verbindingen, en een logregel als er eentje bij is gekomen. Alleen
# bij een stijging: een gelijk aantal is geen gebeurtenis, en een daling
# wordt door nginx zelf gelogd zodra de sessie sluit, met duur en bytes erbij.
#
# Een onbekend vorig aantal geldt als nul, en dat is met opzet. Anders blijft
# het log stil over een wallet die al verbonden was voordat de agent begon, en
# dat is precies het geval waarin iemand op deze pagina komt kijken. Het kost
# één regel per herstart, en die zegt iets.
#
# Wat dit niet ziet: een wallet die binnen één ronde verbindt en weer
# weggaat. Daar staat de disconnect-regel van nginx tegenover, dus die sessie
# is niet onzichtbaar, alleen zonder connect-regel.
open_now = read_open_sessions()
was_open = (previous.get("tls") or {}).get("open_connections")
if not isinstance(was_open, int):
was_open = 0
if isinstance(open_now, int) and open_now > was_open:
events.append({
"t": now,
"event": "connect",
"note": "%d open now" % open_now,
})
backend = probe_backend()
if not backend.get("reachable"):
events.append({
"t": now,
"event": "refused",
"note": "the Electrum server did not answer",
})
# De zelfcontrole op de eigen TLS-poort. Niet elke ronde; zie
# SELF_CHECK_INTERVAL voor waarom.
previous_self = (previous.get("tls") or {}).get("self_check") or {}
probes = [t for t in previous.get("self_probes", [])
if isinstance(t, int) and t > now - LOG_SECONDS]
if reloaded:
# Nu meten zou vrijwel zeker "wrong-certificate" opleveren en dat zou
# onwaar zijn: nginx herlaadt binnen tien seconden en tot die tijd staat
# het vorige certificaat er nog. Een ronde overslaan en het dan meten
# geeft juist de interessantste meting die er is, want een herlading die
# niet aankomt is precies wat deze controle moet vinden.
self_check = {"state": "pending"}
elif now - int(previous_self.get("at") or 0) >= SELF_CHECK_INTERVAL:
self_check = probe_self(active)
if self_check.get("state") != "off":
# Ook bij een mislukking: een geweigerde handdruk is voor nginx nog
# steeds een sessie en levert dus een regel op. Alleen als er niet
# verbonden is, is er niets weg te filteren.
probes.append(self_check["at"])
was = previous_self.get("state")
if was and was != self_check["state"]:
events.append({
"t": now,
"event": "selfcheck",
"note": SELF_CHECK_NOTES.get(
self_check["state"], "self-check: " + self_check["state"]),
})
else:
self_check = previous_self
history = [h for h in previous.get("history", [])
if isinstance(h, dict) and h.get("t", 0) > now - HISTORY_SECONDS]
if "response_ms" in backend:
history.append({"t": now, "response_ms": backend["response_ms"]})
cutoff = now - LOG_SECONDS
entries = suppress_self_probes(read_stream_log(cutoff), probes)
# De eigen gebeurtenissen van de agent uit de vorige ronde weer meenemen; de
# sessieregels komen uit de nginx-log en staan daar al. Vergeet je er één in
# deze opsomming, dan verdwijnt dat soort regel na één minuut weer van de
# pagina, en dat is precies wat een connect-regel onbruikbaar maakt.
entries.extend(e for e in previous.get("log", [])
if isinstance(e, dict) and e.get("t", 0) > cutoff
and e.get("event") in ("start", "reload", "refused", "connect",
"selfcheck"))
entries.extend(events)
# Dubbele meldingen dat de backend niet antwoordt zijn ruis; één per ronde
# is genoeg en de rest zegt hetzelfde.
seen = set()
unique = []
for e in sorted(entries, key=lambda x: x.get("t", 0)):
key = (e.get("t"), e.get("event"), e.get("note"))
if key in seen:
continue
seen.add(key)
unique.append(e)
del events[:]
status = {
"generated_at": now,
"interval_seconds": INTERVAL,
# open_connections blijft weg als het niet bekend is. Zie
# read_open_sessions: nul beweren is hier erger dan zwijgen.
"tls": {"port": TLS_PORT, "self_check": self_check},
"backend": backend,
"history": history,
"log": unique[-200:],
# Wanneer de agent zelf verbond. Boekhouding voor suppress_self_probes en
# niet iets om te tonen; het venster is even lang als dat van het log,
# want daar wordt het tegen gelegd.
"self_probes": probes[-500:],
"certificates": [
{
"id": c["id"],
"source": c["source"],
"name": c["name"],
"domain": c.get("domain"),
"not_after": c.get("not_after"),
"unreadable": c.get("unreadable"),
}
for c in certs
],
"certificate": {"reason": why},
}
if isinstance(open_now, int):
status["tls"]["open_connections"] = open_now
if active:
status["tls"]["domain"] = active.get("domain")
status["certificate"] = {
"reason": why,
"source": active["source"],
"selected": active["id"],
"not_after": active.get("not_after"),
"last_reload": _reload_time(),
}
return status
def _reload_time():
"""Wanneer nginx voor het laatst om een herlading is gevraagd."""
for path in (RELOAD_FLAG + ".done", RELOAD_FLAG):
try:
return int(os.path.getmtime(path))
except OSError:
continue
return None
# --- De API -----------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
server_version = "ElectrumGate"
def log_message(self, fmt, *args):
"""Geen client-adres in de log. Zie het plan Webinterface, 4e."""
log("api " + (fmt % args))
def _send(self, code, body):
data = json.dumps(body).encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(data)
def do_PUT(self): # noqa: N802 - naam wordt door de basisklasse bepaald
if self.path.rstrip("/") != "/api/certificate":
self._send(404, {"error": "unknown path"})
return
length = int(self.headers.get("Content-Length") or 0)
if length <= 0 or length > 1024:
self._send(400, {"error": "expected a certificate id in the body"})
return
wanted = self.rfile.read(length).decode("utf-8", "replace").strip()
# De guard: alleen een id die we zelf gevonden hebben. Dit is het enige
# pad waarlangs iets van buiten de app-instellingen raakt, dus hier hoort
# de controle en niet verderop.
known = {c["id"]: c for c in scan_certificates()}
if wanted not in known:
self._send(400, {"error": "unknown certificate", "id": wanted})
return
if known[wanted].get("unreadable"):
self._send(400, {"error": "that certificate cannot be read",
"id": wanted})
return
write_selection(wanted)
log("selection saved: " + wanted)
WAKE.set()
self._send(200, {"selected": wanted})
def do_POST(self): # noqa: N802
"""Een geupload paar aannemen.
POST en niet PUT, want dit maakt iets nieuws aan in plaats van een
bestaande instelling te zetten. Het formaat is JSON met twee velden en
niet multipart: dat laatste zou een parser vragen die sinds Python 3.13
niet meer in de standaardbibliotheek zit, en de pagina kan de twee
bestanden zelf als tekst inlezen.
"""
if self.path.rstrip("/") != "/api/certificate/upload":
self._send(404, {"error": "unknown path"})
return
length = int(self.headers.get("Content-Length") or 0)
if length <= 0:
self._send(400, {"error": "expected a certificate and a key"})
return
if length > UPLOAD_MAX:
self._send(413, {"error": "that is larger than a certificate and a "
"key should ever be"})
return
try:
payload = json.loads(self.rfile.read(length).decode("utf-8"))
cert_text = payload["cert"]
key_text = payload["key"]
except Exception: # noqa: BLE001 - elke vorm van rommel is hetzelfde
self._send(400, {"error": "could not read the upload"})
return
naam, fout = accept_upload(cert_text, key_text)
if fout:
self._send(400, {"error": fout})
return
# Bewust géén automatische keuze. Uploaden is één handeling en kiezen is
# een andere; de pagina zet de nieuwe eroverheen in de lijst, dus het kost
# één klik. Zo blijft er precies één plek waar TLS van certificaat wisselt.
WAKE.set()
self._send(200, {"id": upload_id(naam)})
def do_GET(self): # noqa: N802
if self.path.rstrip("/") == "/api/certificate":
self._send(200, {"selected": read_selection()})
return
self._send(404, {"error": "unknown path"})
WAKE = threading.Event()
def serve_api():
server = ThreadingHTTPServer(("0.0.0.0", API_PORT), Handler)
log("api listening on port %d" % API_PORT)
server.serve_forever()
def main():
os.makedirs(CONFIG_DIR, exist_ok=True)
events = [{"t": int(time.time()), "event": "start",
"note": "listening on " + str(TLS_PORT)}]
threading.Thread(target=serve_api, daemon=True).start()
while True:
try:
write_status(build_status(events))
except Exception as e: # noqa: BLE001 - de lus mag nooit stoppen, want
# dan bevriest de pagina op oude gegevens zonder dat iemand het ziet.
# De pagina merkt het aan de leeftijd van generated_at.
log("round failed: %r" % e)
WAKE.wait(INTERVAL)
WAKE.clear()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink