UmbrelApps/whatsnext-evolu-relay/agent.py.template

# ═══════════════════════════════════════════════════════════════════════════════
#  De agent van Evolu Relay.
#
#  Hij bedient de statuspagina en doet verder niets: lezen wat het relay-proces
#  heeft opgeschreven, en opdrachten van de pagina in de postbus leggen.
#
#  Waarom de agent niet zelf beslist wie er binnen mag: dat beleid hoort bij het
#  proces dat de verbindingen aanneemt, en dat is de relay. Twee processen die in
#  dezelfde allowlist schrijven is een wedloop die je een keer per jaar treft en
#  dan niet kunt reproduceren. De agent schrijft daarom uitsluitend command.json,
#  en het relay-proces past hem toe en ruimt hem op.
#
#  LET OP: umbreld haalt dit bestand bij elke start door envsubst. Er mag dus geen
#  dollarteken in staan, ook niet in een regex of een tekst. Een accolade-variabele
#  die niet bestaat wordt leeg, en dat sloopt Python-code zonder foutmelding.
#  tests/test_appstore_vorm.py controleert dat.
# ═══════════════════════════════════════════════════════════════════════════════

import json
import os
import socket
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

STATE_DIR = Path(os.environ.get("RELAY_STATE_DIR", "/var/lib/relay"))
OWNERS_FILE = STATE_DIR / "owners.json"
COMMAND_FILE = STATE_DIR / "command.json"
DATABASE_FILE = STATE_DIR / "evolu-relay.db"

# De labels die de gebruiker aan een eigenaar-id hangt. Een eigen bestand, en dat
# is de kern van deze keuze: owners.json is van het relay-proces en labels.json is
# van de agent, dus er is per bestand precies één schrijver. Dat is dezelfde
# afspraak die de postbus hierboven oplevert, en de reden staat bovenaan dit
# bestand.
#
# Wat het bovendien oplevert: een label is meteen opgeslagen en niet pas als de
# relay de postbus leegmaakt, en labelen blijft werken als de relay omgevallen is.
# Dat mag, want de relay hoeft dit niet te weten: een label zegt niets over wie er
# binnen mag.
LABELS_FILE = STATE_DIR / "labels.json"

RELAY_HOST = os.environ.get("RELAY_HOST", "")
RELAY_PORT = int(os.environ.get("RELAY_PORT", "4000"))
PUBLIC_PORT = int(os.environ.get("RELAY_PUBLIC_PORT", "3852"))
API_PORT = int(os.environ.get("RELAY_API_PORT", "8000"))
PROBE_INTERVAL = int(os.environ.get("RELAY_PROBE_INTERVAL", "15"))

# Wat de pagina mag vragen. Expliciet en niet doorgeven wat er binnenkomt: dit
# bestand wordt door een ander proces uitgevoerd, en een onbekende actie hoort
# hier te stranden en niet daar.
ALLOWED_ACTIONS = ("set-learning", "block", "allow", "forget")

MAX_BODY_BYTES = 4096
MAX_OWNER_ID_LENGTH = 256

# Een label is een herkenpunt en geen aantekenveld: het staat op de pagina naast
# een id en moet daar op één regel passen.
MAX_LABEL_LENGTH = 48

# Een bovengrens op het aantal labels. De pagina zit achter de inlog van umbrelOS,
# dus dit is geen verdediging tegen een aanvaller maar tegen een lus die per
# ongeluk blijft schrijven. Ruim boven het aantal eigenaars dat iemand ooit heeft.
MAX_LABELS = 200

# Hoelang de pagina de deur voor nieuwe eigenaars openzet. Dezelfde waarde staat
# in de pagina; die stuurt hem mee en het relay-proces begrenst hem nog een keer.
# Hier staat hij omdat de agent hem moet toestaan, niet omdat hij hem kiest.
MAX_LEARNING_SECONDS = 3600

# Eén schrijver per bestand is de afspraak, maar de agent zelf is meerdradig:
# ThreadingHTTPServer geeft elk verzoek zijn eigen draad. Twee labels die op
# hetzelfde moment binnenkomen zouden elkaar dus kunnen overschrijven, want
# labelen is lezen-wijzigen-schrijven. Dit slot maakt daar één handeling van.
labels_lock = threading.Lock()

# Door de achtergrondlus bijgewerkt, door de webserver gelezen. Een dict wordt in
# zijn geheel vervangen en nooit ter plekke aangepast, zodat een lezer altijd een
# samenhangend beeld heeft zonder slot.
probe = {"reachable": None, "checked": None}


def relay_reachable():
    """Kan de relay een TCP-verbinding aannemen.

    Bewust niet meer dan dat. De relay is een WebSocket-server en antwoordt niet
    op een gewoon verzoek; een handdruk nabouwen om de pagina groen te krijgen is
    meer code dan het waard is. Wat dit wel uitsluit is de meest voorkomende
    storing: het proces is omgevallen.
    """
    if not RELAY_HOST:
        return None
    try:
        with socket.create_connection((RELAY_HOST, RELAY_PORT), timeout=3):
            return True
    except OSError:
        return False


def probe_loop():
    global probe
    while True:
        # In zijn geheel vervangen en niet ter plekke aanpassen: een lezer ziet
        # dan altijd een samenhangend beeld, zonder dat er een slot nodig is.
        probe = {
            "reachable": relay_reachable(),
            "checked": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }
        time.sleep(PROBE_INTERVAL)


def read_owners():
    """De allowlist zoals het relay-proces hem heeft achtergelaten."""
    try:
        with OWNERS_FILE.open("r", encoding="utf-8") as handle:
            data = json.load(handle)
    except FileNotFoundError:
        # Nog nooit geschreven. Dat is de normale toestand vlak na een
        # installatie: het relay-proces schrijft pas bij de eerste wijziging.
        return {"state": None, "problem": "nog-niet-aangemaakt"}
    except (OSError, ValueError):
        return {"state": None, "problem": "onleesbaar"}

    if not isinstance(data, dict):
        return {"state": None, "problem": "onleesbaar"}
    return {"state": data, "problem": None}


def read_labels():
    """De labels, of een lege verzameling.

    Bewust vergevingsgezind, en dat is het omgekeerde van hoe owners.json gelezen
    wordt. Daar hangt aan een half begrepen bestand de vraag wie er binnen mag, en
    dan is weigeren het antwoord. Hier gaat het om een naam naast een id: is het
    onleesbaar, dan is het ergste gevolg dat je de rauwe ids ziet.
    """
    try:
        with LABELS_FILE.open("r", encoding="utf-8") as handle:
            data = json.load(handle)
    except (OSError, ValueError):
        return {}

    if not isinstance(data, dict):
        return {}

    schoon = {}
    for owner_id, label in data.items():
        if not isinstance(owner_id, str) or not isinstance(label, str):
            continue
        if not owner_id or len(owner_id) > MAX_OWNER_ID_LENGTH:
            continue
        label = label.strip()
        if label:
            schoon[owner_id] = label[:MAX_LABEL_LENGTH]
    return schoon


def write_labels(labels):
    """Schrijft de labels. Eerst een tijdelijk bestand en dan hernoemen.

    Hernoemen binnen dezelfde map is atomair, dus een onderbroken schrijfactie
    laat geen half bestand achter. Dezelfde constructie als de postbus.
    """
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    temporary = LABELS_FILE.with_suffix(".json.tmp")
    with temporary.open("w", encoding="utf-8") as handle:
        json.dump(labels, handle, indent=2, sort_keys=True)
        handle.write("\n")
    temporary.replace(LABELS_FILE)


def parse_moment(value):
    """Een ISO-tijdstip uit owners.json als datetime, of None.

    Het relay-proces schrijft `new Date().toISOString()`, dus met milliseconden en
    met een Z erachter. `fromisoformat` neemt die Z sinds Python 3.11; de image is
    python:3-alpine en dus nieuwer. Faalt het alsnog, dan is None het antwoord en
    beslist de aanroeper.
    """
    if not isinstance(value, str):
        return None
    try:
        when = datetime.fromisoformat(value)
    except ValueError:
        return None
    if when.tzinfo is None:
        return when.replace(tzinfo=timezone.utc)
    return when


def learning_facts(state):
    """De leerstand zoals de pagina hem hoort te zien.

    Het veld in het bestand is niet het hele antwoord: staat er een tijdstip in dat
    verstreken is, dan is de deur dicht, ook al staat `learning` nog op true. Het
    relay-proces ruimt dat op in zijn eigen lus, en tussen het aflopen en die ronde
    zit een seconde of twee. De pagina hoort daar niet "open" te tonen.

    De resterende tijd wordt hier uitgerekend en niet in de browser. Dat is met
    opzet: dan telt de klok van de Umbrel en niet die van de bezoeker, en die twee
    lopen niet per definitie gelijk.
    """
    learning = state.get("learning")
    until = state.get("learningUntil")

    if learning is not True:
        return {"learning": learning, "learningUntil": None, "learningSecondsLeft": None}

    when = parse_moment(until)
    if when is None:
        # Geen tijdslot: open tot de gebruiker hem zelf sluit. Dat is de
        # begintoestand van een verse installatie.
        return {"learning": True, "learningUntil": None, "learningSecondsLeft": None}

    resterend = (when - datetime.now(timezone.utc)).total_seconds()
    if resterend <= 0:
        return {"learning": False, "learningUntil": None, "learningSecondsLeft": 0}
    return {
        "learning": True,
        "learningUntil": until,
        "learningSecondsLeft": int(resterend),
    }


def database_facts():
    try:
        stat = DATABASE_FILE.stat()
    except OSError:
        return {"bytes": None, "modified": None}
    return {
        "bytes": stat.st_size,
        "modified": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)),
    }


def met_label(entries, labels):
    """Hangt het label van de gebruiker aan elke regel.

    Een kopie en niet ter plekke: wat hier binnenkomt komt uit het bestand van het
    relay-proces, en daar horen wij niets aan toe te voegen.
    """
    resultaat = []
    for entry in entries:
        regel = dict(entry)
        regel["label"] = labels.get(entry.get("id"))
        resultaat.append(regel)
    return resultaat


def build_status():
    owners = read_owners()
    state = owners["state"] or {}
    labels = read_labels()

    # Ontbreekt de staat, dan is 'learning' onbekend en niet 'false'. De pagina
    # hoort dat verschil te tonen: onbekend is een reden om te kijken, uit is een
    # keuze. learning_facts() geeft None door zoals het binnenkwam.
    learning = learning_facts(state)

    return {
        "relay": {
            "reachable": probe["reachable"],
            "checked": probe["checked"],
            "publicPort": PUBLIC_PORT,
        },
        "owners": {
            "problem": owners["problem"],
            "learning": learning["learning"],
            "learningUntil": learning["learningUntil"],
            "learningSecondsLeft": learning["learningSecondsLeft"],
            "allowed": met_label(
                [
                    entry
                    for entry in state.get("owners", [])
                    if isinstance(entry, dict) and entry.get("allowed") is True
                ],
                labels,
            ),
            "blocked": met_label(
                [
                    entry
                    for entry in state.get("owners", [])
                    if isinstance(entry, dict) and entry.get("allowed") is False
                ],
                labels,
            ),
            "rejected": met_label(
                [entry for entry in state.get("rejected", []) if isinstance(entry, dict)],
                labels,
            ),
        },
        "database": database_facts(),
        # Ligt er nog een opdracht, dan heeft het relay-proces hem nog niet
        # opgepakt. De pagina kan dat tonen in plaats van te doen alsof er niets
        # gebeurd is.
        "pendingCommand": COMMAND_FILE.exists(),
    }


def valid_command(payload):
    """Geeft de opdracht terug, of een foutmelding.

    Streng aan deze kant, want dit is de enige plek waar iets van buiten in de
    postbus belandt.
    """
    if not isinstance(payload, dict):
        return None, "geen object"

    action = payload.get("action")
    if action not in ALLOWED_ACTIONS:
        return None, "onbekende actie"

    if action == "set-learning":
        value = payload.get("value")
        if not isinstance(value, bool):
            return None, "waarde moet true of false zijn"

        seconds = payload.get("seconds")
        if seconds is None:
            # Zonder tijdslot: open tot de gebruiker hem zelf sluit. Dat pad blijft
            # bestaan voor een verse installatie, waar een venster van twee minuten
            # zou aflopen terwijl je nog aan het koppelen bent.
            return {"action": action, "value": value}, None

        # isinstance(True, int) is in Python waar, dus een boolean zou hier als
        # aantal seconden doorglippen. Vandaar de uitsluiting.
        if isinstance(seconds, bool) or not isinstance(seconds, int):
            return None, "seconds moet een heel getal zijn"
        if seconds <= 0 or seconds > MAX_LEARNING_SECONDS:
            return None, "seconds valt buiten het toegestane bereik"
        if value is not True:
            return None, "seconds hoort alleen bij openzetten"
        return {"action": action, "value": value, "seconds": seconds}, None

    owner_id = payload.get("ownerId")
    if not isinstance(owner_id, str) or not owner_id or len(owner_id) > MAX_OWNER_ID_LENGTH:
        return None, "ontbrekende of te lange ownerId"
    return {"action": action, "ownerId": owner_id}, None


def valid_label(payload):
    """Geeft (ownerId, label) terug, of een foutmelding.

    Een leeg label is geen fout maar de manier om er een weg te halen: dan hoeft er
    geen tweede opdracht te bestaan voor iets dat de gebruiker als hetzelfde veld
    ziet.
    """
    if not isinstance(payload, dict):
        return None, None, "geen object"

    owner_id = payload.get("ownerId")
    if not isinstance(owner_id, str) or not owner_id or len(owner_id) > MAX_OWNER_ID_LENGTH:
        return None, None, "ontbrekende of te lange ownerId"

    label = payload.get("label")
    if label is None:
        label = ""
    if not isinstance(label, str):
        return None, None, "label moet tekst zijn"

    # Regeleindes eruit: dit is één regel naast een id, en een label met een
    # nieuwe regel erin zou de lijst uit elkaar trekken.
    label = " ".join(label.split()).strip()
    if len(label) > MAX_LABEL_LENGTH:
        return None, None, "label is te lang"

    return owner_id, label, None


def apply_label(owner_id, label):
    """Zet of haalt een label weg. Geeft een foutmelding terug, of None.

    Onder het slot, want dit is lezen-wijzigen-schrijven en de agent bedient
    meerdere verzoeken tegelijk.
    """
    with labels_lock:
        labels = read_labels()
        if label:
            if owner_id not in labels and len(labels) >= MAX_LABELS:
                return "er zijn al te veel labels"
            labels[owner_id] = label
        else:
            if owner_id not in labels:
                # Niets te doen, en dat is geen fout: de pagina stuurt een leeg
                # label als je het veld leegmaakt, ook als er nog niets stond.
                return None
            del labels[owner_id]

        try:
            write_labels(labels)
        except OSError as error:
            return "label kon niet worden weggeschreven: " + str(error)
    return None


def write_command(command):
    """Legt de opdracht in de postbus.

    Eerst een tijdelijk bestand en dan hernoemen: het relay-proces kijkt op zijn
    eigen moment en mag geen half bestand aantreffen.
    """
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    temporary = COMMAND_FILE.with_suffix(".json.tmp")
    with temporary.open("w", encoding="utf-8") as handle:
        json.dump(command, handle)
        handle.write("\n")
    temporary.replace(COMMAND_FILE)


class Handler(BaseHTTPRequestHandler):
    # De standaardregel van BaseHTTPRequestHandler noemt de naam van de server en
    # de Python-versie. Dat hoeft niemand te weten.
    server_version = "evolu-relay-agent"
    sys_version = ""

    def log_message(self, format, *args):
        # Geen toegangslog. Elke regel zou het adres van de bezoeker bevatten en
        # de pagina zit achter de inlog van umbrelOS; er valt niets te zien wat
        # het bewaren waard is.
        return

    def _send(self, status, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path.rstrip("/") in ("/api/status", "/status"):
            self._send(200, build_status())
            return
        self._send(404, {"error": "onbekend pad"})

    def _read_payload(self):
        """Het verzoek als object, of None als er al een fout verstuurd is."""
        try:
            length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            self._send(400, {"error": "lengte ontbreekt"})
            return None

        if length <= 0 or length > MAX_BODY_BYTES:
            self._send(400, {"error": "lege of te grote opdracht"})
            return None

        try:
            return json.loads(self.rfile.read(length).decode("utf-8"))
        except (UnicodeDecodeError, ValueError):
            self._send(400, {"error": "onleesbare opdracht"})
            return None

    def do_POST(self):
        pad = self.path.rstrip("/")

        # Een label gaat NIET via de postbus, en dat is de enige uitzondering op
        # die regel. De reden dat opdrachten er wel door gaan, is dat het
        # relay-proces beslist wie er binnen mag en dat twee schrijvers in die
        # allowlist een wedloop zou zijn. Een label zegt niets over toegang, staat
        # in een eigen bestand met de agent als enige schrijver, en is meteen
        # opgeslagen in plaats van na de volgende ronde van de relay.
        if pad in ("/api/label", "/label"):
            payload = self._read_payload()
            if payload is None:
                return

            owner_id, label, problem = valid_label(payload)
            if problem is not None:
                self._send(400, {"error": problem})
                return

            problem = apply_label(owner_id, label)
            if problem is not None:
                self._send(500, {"error": problem})
                return

            self._send(200, {"ownerId": owner_id, "label": label})
            return

        if pad not in ("/api/command", "/command"):
            self._send(404, {"error": "onbekend pad"})
            return

        payload = self._read_payload()
        if payload is None:
            return

        command, problem = valid_command(payload)
        if command is None:
            self._send(400, {"error": problem})
            return

        # Eén opdracht tegelijk. Ligt er nog een, dan zou schrijven hem stil
        # overschrijven en verdwijnt de vorige zonder dat iemand het merkt.
        if COMMAND_FILE.exists():
            self._send(409, {"error": "vorige opdracht is nog niet verwerkt"})
            return

        try:
            write_command(command)
        except OSError as error:
            self._send(500, {"error": "opdracht kon niet worden weggeschreven: " + str(error)})
            return

        self._send(202, {"accepted": command})


def main():
    threading.Thread(target=probe_loop, daemon=True).start()
    ThreadingHTTPServer(("0.0.0.0", API_PORT), Handler).serve_forever()


if __name__ == "__main__":
    main()