diff --git a/.dockerignore b/.dockerignore index 0d34905..3d7f8fa 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,5 +1,7 @@ .git .gitignore +.gocache +.gobuild .pi bin/ *.md diff --git a/.gitignore b/.gitignore index 0cb4905..5688e0e 100644 --- a/.gitignore +++ b/.gitignore @@ -34,6 +34,15 @@ build/ local.properties *.apk +# MQTT lab material (Phase 3): generated certificates, accounts and passwords +deploy/mosquitto/tls/*.crt +deploy/mosquitto/tls/*.key +deploy/mosquitto/tls/*.csr +deploy/mosquitto/tls/*.srl +deploy/mosquitto/conf.d/*.passwd +deploy/mosquitto/build/ +deploy/mosquitto/clients.env + # OS / editors .DS_Store *.swp diff --git a/Makefile b/Makefile index 898f9c3..8894355 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,19 @@ -IMAGE ?= webrestrict/control:0.2.0-phase2 -BUILD_VERSION ?= 0.2.0-phase2 +IMAGE ?= webrestrict/control:0.3.0-phase3 +BUILD_VERSION ?= 0.3.0-phase3 -.PHONY: help test vet build up down restart logs ps hash-password fmt +# Bounded docker helper for the Go toolchain. The container runs as the current +# user, so the repository and the module cache stay owned by you. +GO_IMAGE ?= golang:1.25-alpine +GO_RUN = timeout 600 docker run --rm --network host --user "$$(id -u):$$(id -g)" \ + -e HOME=/src/.gocache -e GOMODCACHE=/src/.gocache/mod -e GOCACHE=/src/.gocache/build \ + -v "$$PWD":/src -w /src $(GO_IMAGE) + +# Scratch broker used by `make mqtt-verify` (see deploy/mosquitto/README.md). +MQTT_VERIFY_PORT ?= 18883 +MQTT_IMAGE ?= eclipse-mosquitto:latest + +.PHONY: help test vet build up down restart logs ps hash-password deps fmt \ + mqtt-tls mqtt-verify mqtt-config help: @echo "make test run go test ./... in golang:1.25-alpine" @@ -10,21 +22,24 @@ help: @echo "make up docker compose up -d --build" @echo "make down docker compose down" @echo "make logs follow container logs" + @echo "make ps docker compose ps" @echo "make hash-password generate a bcrypt hash (PASSWORD=...)" + @echo "make deps go mod tidy + download (writes go.mod/go.sum)" + @echo "make mqtt-tls generate the lab CA and broker certificate" + @echo "make mqtt-config build the scratch broker config in deploy/mosquitto/build/" + @echo "make mqtt-verify prove TLS + credentials + topic ACLs on a scratch broker" test: - docker run --rm --network host \ - -v "$$PWD":/src -w /src \ - -v webrestrict-go-mod:/go/pkg/mod \ - -v webrestrict-go-build:/root/.cache/go-build \ - golang:1.25-alpine sh -c 'go test ./...' + $(GO_RUN) go test ./... vet: - docker run --rm --network host \ - -v "$$PWD":/src -w /src \ - -v webrestrict-go-mod:/go/pkg/mod \ - -v webrestrict-go-build:/root/.cache/go-build \ - golang:1.25-alpine sh -c 'go vet ./...' + $(GO_RUN) go vet ./... + +deps: + $(GO_RUN) sh -c 'go mod tidy' + +fmt: + $(GO_RUN) sh -c 'go fmt ./...' build: docker build --build-arg VERSION=$(BUILD_VERSION) -t $(IMAGE) . @@ -47,3 +62,17 @@ ps: hash-password: @test -n "$(PASSWORD)" || (echo "usage: make hash-password PASSWORD=secret"; exit 2) docker run --rm --entrypoint /app/control $(IMAGE) hash-password '$(PASSWORD)' + +# --- MQTT --------------------------------------------------------------- +# These targets never touch the shared Mosquitto. They build the material and +# prove it on a scratch broker on $(MQTT_VERIFY_PORT). See deploy/mosquitto/README.md. + +mqtt-tls: + deploy/mosquitto/generate-tls.sh + +mqtt-config: mqtt-tls + deploy/mosquitto/build-config.sh + +mqtt-verify: mqtt-config + MQTT_VERIFY_PORT=$(MQTT_VERIFY_PORT) MQTT_IMAGE=$(MQTT_IMAGE) \ + deploy/mosquitto/verify-isolation.sh diff --git a/README.md b/README.md index 82a3848..046005a 100644 --- a/README.md +++ b/README.md @@ -5,8 +5,8 @@ Internet Use Control project (`SYSTEM_MAP.md`). It holds schedules, blocklists, per-device state and usage, and serves the web dashboard. Only it publishes device config. -This repository currently contains **Phase 2 — Devices and state** -(AGENTS.md §11), on top of the Phase 1 skeleton: +This repository currently contains **Phase 3 — MQTT** (AGENTS.md §11), on top of +Phases 1 and 2: - Go service with its own image (`Dockerfile`) and compose file. - SQLite schema for the whole data model (§7), applied through versioned @@ -22,9 +22,13 @@ This repository currently contains **Phase 2 — Devices and state** - **Dashboard cards**: name, child, platform, status, current state, time left. - **Manual lock/unlock** from the dashboard, persisted in `state` and written to the append-only `audit` log (FR-15). +- **MQTT**: each device's desired state is published *retained* on + `restrict//config`; device status and telemetry are subscribed and + written to SQLite; presence comes from the broker's Last Will and Testament. + The broker is transport only — SQLite stays the source of truth. -Enforcement, MQTT, schedules, blocklists and the clients arrive in later -phases. Nothing here touches the router, Pi-hole, or DNS. +Enforcement, schedules, blocklists, the admin API and the kids' clients arrive +in later phases. Nothing here touches the router, Pi-hole, or DNS. ## Quick start (on `.13`) @@ -40,7 +44,7 @@ docker compose up -d --build # 3. Prove it. curl -s http://127.0.0.1:8791/health -# {"status":"ok","service":"control-service","version":"0.2.0-phase2",...} +# {"status":"ok","service":"control-service","version":"0.3.0-phase3",...} ``` Open in a browser. It redirects to `/login`. Sign in @@ -65,15 +69,91 @@ reads only the process environment. | `CONTROL_ADMIN_PASSWORD_HASH` | — | bcrypt hash (option B, preferred) | | `CONTROL_APP_ENV` | `production` | Label shown in the footer | | `CONTROL_LOG_LEVEL` | `info` | `debug`/`info`/`warn`/`error` | +| `CONTROL_MQTT_ENABLED` | `true` | Turn the broker link on or off | +| `CONTROL_MQTT_BROKER` | `tcp://host.docker.internal:1883` | Broker URL; `tls://…:8883` once the TLS listener exists | +| `CONTROL_MQTT_CLIENT_ID` | `control-service` | MQTT client id | +| `CONTROL_MQTT_USERNAME` | — | Control Service broker account | +| `CONTROL_MQTT_PASSWORD` | — | its password | +| `CONTROL_MQTT_CA_FILE` | `/etc/webrestrict/tls/ca.crt` | PEM bundle that verifies the broker (TLS only) | +| `CONTROL_MQTT_INSECURE` | `false` | **Temporary.** Allow a plaintext, credential-less broker | Generate a bcrypt hash without host Go: ```bash -docker run --rm --entrypoint /app/control webrestrict/control:0.2.0-phase2 \ +docker run --rm --entrypoint /app/control webrestrict/control:0.3.0-phase3 \ hash-password 'your-password' ``` -## Endpoints (Phase 2) +`CONTROL_MQTT_USERNAME`/`CONTROL_MQTT_PASSWORD` are required unless +`CONTROL_MQTT_INSECURE=true`, and a non-TLS broker is refused unless +`CONTROL_MQTT_INSECURE=true` (AGENTS.md §10 rule 6). The shared Mosquitto on +`.13` has no TLS listener yet, so the running deployment sets the escape hatch +and logs a warning. `deploy/mosquitto/README.md` holds the TLS listener, the +per-device credentials and the topic ACL, verified but not yet applied — see +that file for the apply and rollback steps. + +## MQTT (Phase 3) + +Broker: the shared **Mosquitto on `.13`**. A second broker is not added +(§10 rule 6). + +| Topic | Direction | Retained | Payload | +|---|---|---|---| +| `restrict//config` | service → device | yes | desired state | +| `restrict//status` | device → service | yes | presence (`online`) | +| `restrict//telemetry` | device → service | no | usage totals for a day | + +- `` is `restrict/_` from `devices.topic`, which is stored when + the device is created and never changes. `` is its MQTT username. +- The service publishes each device's config **retained** on startup, after every + lock, unlock, expiry, create and edit, and again every 15 minutes so a broker + that lost its retained store is repaired from SQLite. +- A device registers the will `{"online": false}` on its status topic, so the + broker reports the drop the moment the connection dies. That is presence + (FR-6); no polling. +- Reports are written to SQLite. Telemetry is a rolling daily total per + (device, day, app, domain), so a repeated report replaces the row rather than + adding to it (migration 002). + +### Config payload (v1) + +```json +{"v":1,"device_id":1,"name":"Harry's phone","platform":"android", + "locked":true,"until":"2026-10-13T21:00:00Z","allow":[],"block":[], + "generated_at":"2026-10-13T03:30:27Z"} +``` + +`allow` and `block` are always arrays, never `null`. They are read from the +`allowlist` and `blocklist` tables, which get their editors in Phase 6; until +then they are empty. A client must ignore unknown fields, so this shape can grow. + +### Status payload (device → service) + +```json +{"online":true,"app":"com.spotify.music","at":"2026-10-13T03:30:27Z"} +``` + +The will the device registers on the same topic is `{"online":false}`. A report +with no `online` field counts as online, because the device just talked to us. + +### Telemetry payload (device → service) + +```json +{"day":"2026-10-13", + "apps":[{"name":"com.spotify.music","seconds":600}], + "domains":[{"name":"youtube.com","seconds":120}]} +``` + +A flat single-row form (`{"app":"…","seconds":600}`) is also accepted. Seconds +are totals for `day`, in `YYYY-MM-DD`; a missing or malformed day uses today. + +List the MQTT identity of every device (tokens, no secrets): + +```bash +docker exec webrestrict-control /app/control mqtt-clients +``` + +## Endpoints (Phase 2–3) | Method | Path | Notes | |---|---|---| @@ -91,7 +171,10 @@ docker run --rm --entrypoint /app/control webrestrict/control:0.2.0-phase2 \ | `POST` | `/devices/{id}/unlock` | Unlocks now. | Every form POST needs the `csrf` field from the session cookie. State changes -append to the `audit` table. Devices are never hard-deleted (§10 rule 10). +append to the `audit` table and republish the device's retained config. Devices +are never hard-deleted (§10 rule 10). + +Phase 3 adds no HTTP endpoints. ## Development @@ -103,16 +186,28 @@ make vet # go vet ./... make build # build the container image make up # docker compose up -d --build make logs # follow container logs +make mqtt-verify # prove TLS + credentials + ACLs on a throwaway broker ``` +`make test` and `make vet` run the container as your own user and keep the module +cache in `.gocache/` (gitignored), so nothing in the repository ends up owned by +root. `make mqtt-verify` never touches the shared Mosquitto: it stages a scratch +broker on `127.0.0.1:18883` and removes it afterwards. + ## Layout ``` -cmd/control/ service entry point (+ hash-password helper) +cmd/control/ service entry point (+ hash-password, mqtt-clients) internal/config/ environment configuration internal/store/ SQLite open, migrations, audit, devices + state -internal/store/migrations/001_init.sql + devices.go device registry, desired state, lock expiry + reports.go presence and telemetry written from broker reports + lists.go allow/block lists and device lookup by topic + migrations/001_init.sql, 002_telemetry_unique.sql internal/auth/ signed-cookie session login + CSRF +internal/mqtt/ the broker link: retained config out, reports in + payload.go topic and payload contract for the kids' clients + link.go paho client, subscriptions, publish, TLS verification internal/web/ HTTP handlers, templates, static assets devices.go device registry + manual lock/unlock handlers templates/ login, dashboard, device form, HTMX partials @@ -121,6 +216,7 @@ Dockerfile own image (multi-stage, CGO-free) docker-compose.yml own container, port 8791 environment.d/ control-service.env.example (real file gitignored) deploy/caddy/ Caddy vhost for webrestrict.lab.audasmedia.com.au +deploy/mosquitto/ TLS listener, credentials, topic ACL, verifier ``` ## Caddy @@ -129,10 +225,12 @@ deploy/caddy/ Caddy vhost for webrestrict.lab.audasmedia.com.au `webrestrict.lab.audasmedia.com.au`. The terminating Caddy runs on `.35` (sam-Ubuntu1); see that file for the two proxy patterns and the apply commands. -## Non-goals (Phase 2) +## Non-goals (Phase 3) -- No MQTT publishing (Phase 3). -- No schedules, blocklists or allowlists (Phase 6). -- No telemetry or usage screens (Phase 7). +- No schedules, blocklists or allowlists (Phase 6). The config payload reads + those tables, but nothing edits them yet. +- No usage screens (Phase 7). Telemetry is stored, not yet shown. - No admin API (Phase 8). +- No kids' clients (Phases 4–5). The topic and payload contract above is what + they implement. - No Hub contact (Phase 9). diff --git a/TRANSPORT_DESIGN.md b/TRANSPORT_DESIGN.md index 830894e..dc691a5 100644 --- a/TRANSPORT_DESIGN.md +++ b/TRANSPORT_DESIGN.md @@ -46,6 +46,50 @@ ground. | `restrict//cmd` | service → device | no | one-shot command | | `restrict//ack` | device → service | no | command result | +`config`, `status` and `telemetry` are **implemented** (Phase 3, +`internal/mqtt/`). `cmd` and `ack` are still design only: no phase uses them yet. + +`` is `restrict/_`, stored in `devices.topic` when the device +is created, and never changes. `` is also the device's MQTT username. + +### Payloads, as implemented (Phase 3) + +`restrict//config`, retained, written by the Control Service only: + +```json +{"v":1,"device_id":1,"name":"Harry's phone","platform":"android", + "locked":true,"until":"2026-10-13T21:00:00Z","allow":[],"block":[], + "generated_at":"2026-10-13T03:30:27Z"} +``` + +`allow` and `block` are always arrays. `until` is `null` when the lock has no end +time. Clients must ignore unknown fields; `v` is the schema version. + +`restrict//status`, retained, written by the device: + +```json +{"online":true,"app":"com.spotify.music","at":"2026-10-13T03:30:27Z"} +``` + +The device publishes `{"online": true}` when it comes up and registers the will +`{"online": false}` on the same topic in its CONNECT packet. The broker then +publishes that will the moment the connection drops without DISCONNECT, which is +the whole of presence detection (FR-6). A status payload with no `online` field +counts as online, because the device just talked to us. + +`restrict//telemetry`, not retained, written by the device: + +```json +{"day":"2026-10-13", + "apps":[{"name":"com.spotify.music","seconds":600}], + "domains":[{"name":"youtube.com","seconds":120}]} +``` + +Seconds are **totals for the day**, not deltas. A repeated report replaces the +row for (device, day, app, domain), so a device may resend freely (migration +`002_telemetry_unique.sql`). A flat single-row form (`{"app":"…","seconds":600}`) +is also accepted. + ### Example — the "phone off, desktop on" case ``` @@ -66,6 +110,20 @@ reconnects is locked again **immediately**, with no waiting. own** status. Without this, a child can publish a fake "unlocked". - Rotate device credentials when a device is removed. +**Status (Phase 3).** Written, verified, not yet applied: +`deploy/mosquitto/conf.d/webrestrict.conf` is the TLS listener, `.acl` is the +topic policy, and `deploy/mosquitto/README.md` is the apply/rollback runbook. +`make mqtt-verify` proves the policy on a throwaway broker. The Control Service +is configured for TLS plus credentials, and refuses a plaintext, +credential-less broker unless `CONTROL_MQTT_INSECURE=true` is set explicitly. + +The shared Mosquitto on `.13` still has only its original anonymous `1883` +listener, so the running deployment uses that escape hatch and logs a warning. +Applying the TLS listener means editing the shared broker's configuration and +compose file, adding port 8883 and restarting it — a change to a shared +household service, so it waits for Sam's approval instead of being applied +unilaterally. + --- ## Keep the HTTP API diff --git a/cmd/control/main.go b/cmd/control/main.go index 6162eb4..4e93f25 100644 --- a/cmd/control/main.go +++ b/cmd/control/main.go @@ -1,8 +1,8 @@ // Command control is the Internet Use Control Control Service. // // It is the single source of truth for the family's device rules -// (SYSTEM_MAP.md). Phase 2: device registry, desired state in SQLite, manual -// lock/unlock from the dashboard, and the audit log. +// (SYSTEM_MAP.md). Phase 3: publishes each device's desired state to the broker, +// retained, and consumes device presence and telemetry reports from it. package main import ( @@ -22,6 +22,7 @@ import ( "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/auth" "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/config" + "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/mqtt" "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store" "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/web" ) @@ -29,12 +30,20 @@ import ( // version is injected with -ldflags at build time. var version = "dev" +// republishInterval is how often the retained config is refreshed from SQLite. +// The broker is transport, not storage: if its retained store is ever lost, the +// service repairs it from the database without a restart. +const republishInterval = 15 * time.Minute + func main() { if len(os.Args) > 1 { switch os.Args[1] { case "hash-password": hashPassword(os.Args[2:]) return + case "mqtt-clients": + mqttClients(os.Args[2:]) + return case "version": fmt.Println(version) return @@ -77,7 +86,46 @@ func run() error { log.Warn("admin password is plaintext; set CONTROL_ADMIN_PASSWORD_HASH to use bcrypt") } - srv, err := web.New(cfg, st, am, log) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // The broker link (Phase 3). It is optional on purpose: the dashboard must + // keep working while the broker is down, and the device keeps its last state + // (AGENTS.md §10 rule 2). + var ( + link *mqtt.Link + pub web.Publisher + ) + if cfg.MQTTEnabled { + link, err = mqtt.New(mqtt.Options{ + Broker: cfg.MQTTBroker, + ClientID: cfg.MQTTClientID, + Username: cfg.MQTTUsername, + Password: cfg.MQTTPassword, + CAFile: cfg.MQTTCAFile, + Actor: "device", + Log: log, + Devices: st, + Source: st, + Reports: st, + OnConnected: func() { republish(ctx, log, link) }, + }) + if err != nil { + return err + } + pub = link + link.Start(ctx) + defer link.Stop() + go republishLoop(ctx, log, link) + if !cfg.MQTTTLS() { + log.Warn("mqtt is running without TLS; the shared broker has no TLS listener yet (AGENTS.md §10 rule 6)", + "broker", cfg.MQTTBroker) + } + } else { + log.Warn("mqtt is disabled; device config stays in SQLite and is not published") + } + + srv, err := web.New(cfg, st, am, log, pub) if err != nil { return err } @@ -109,16 +157,95 @@ func run() error { case sig := <-stop: log.Info("shutting down", "signal", sig.String()) } + cancel() - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - if err := httpServer.Shutdown(ctx); err != nil { + ctxShutdown, cancelShutdown := context.WithTimeout(context.Background(), 10*time.Second) + defer cancelShutdown() + if err := httpServer.Shutdown(ctxShutdown); err != nil { return fmt.Errorf("shutdown: %w", err) } log.Info("stopped", "uptime", time.Since(started).Round(time.Second).String()) return nil } +// republish pushes every device's retained config. It runs once per connect and +// then periodically, so the broker's view of the desired state is rebuilt from +// SQLite, the source of truth. +func republish(ctx context.Context, log *slog.Logger, link *mqtt.Link) { + if link == nil { + return + } + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + n, err := link.PublishAll(ctx) + if err != nil { + log.Error("mqtt publish all", "published", n, "err", err) + return + } + log.Info("mqtt retained config refreshed", "devices", n) +} + +func republishLoop(ctx context.Context, log *slog.Logger, link *mqtt.Link) { + ticker := time.NewTicker(republishInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if !link.Connected() { + continue + } + republish(ctx, log, link) + } + } +} + +// mqttClients prints the MQTT identity of every registered device, one device +// per line, as tab-separated fields: +// +// \t\t\t\t +// +// provision-clients.sh uses it to create one broker account per device. Real +// tabs, not padded columns: names contain spaces, so a space-separated field +// would not be safe to parse. The token is the MQTT username and the middle +// level of the device's topics (deploy/mosquitto/README.md). It is not a secret +// and this never prints one. +func mqttClients(args []string) { + dbPath := env("CONTROL_DB_PATH", "/data/control.db") + for i := 0; i < len(args); i++ { + if (args[i] == "--db" || args[i] == "-db") && i+1 < len(args) { + dbPath = args[i+1] + i++ + } + } + + st, err := store.Open(dbPath) + if err != nil { + fmt.Fprintln(os.Stderr, "control:", err) + os.Exit(1) + } + defer st.Close() + + devices, err := st.ListDevices(context.Background()) + if err != nil { + fmt.Fprintln(os.Stderr, "control:", err) + os.Exit(1) + } + + for _, dev := range devices { + token := strings.TrimSuffix(strings.TrimPrefix(dev.Topic, mqtt.TopicPrefix+"/"), mqtt.SuffixConfig) + fmt.Printf("%s\t%s\t%s\t%s\t%s\n", token, dev.Topic, dev.Platform, dev.Child, dev.Name) + } +} + +func env(key, fallback string) string { + if v := strings.TrimSpace(os.Getenv(key)); v != "" { + return v + } + return fallback +} + // hashPassword prints a bcrypt hash for the password given as an argument or // read from stdin. It never prints the password itself. func hashPassword(args []string) { diff --git a/deploy/mosquitto/README.md b/deploy/mosquitto/README.md new file mode 100644 index 0000000..4573e11 --- /dev/null +++ b/deploy/mosquitto/README.md @@ -0,0 +1,149 @@ +# Phase 3 MQTT security for the shared Mosquitto + +This directory holds the **broker-side** half of Phase 3 (AGENTS.md §11, §10 +rule 6; `TRANSPORT_DESIGN.md` "Security"). The Go half lives in +`internal/mqtt/`. + +The transport itself — retained config per device, status and telemetry +subscriptions, LWT presence — is implemented and running against the shared +Mosquitto on `.13:1883`. What follows is the credential and TLS layer that +onboards the kids' devices. + +## What is here + +| File | What it is | +|---|---| +| `conf.d/webrestrict.conf` | the TLS listener on 8883 with authentication | +| `conf.d/webrestrict.acl` | the topic ACL: one device, one topic set | +| `conf.d/webrestrict.passwd` | **generated** hashed accounts (gitignored) | +| `generate-tls.sh` | builds the lab CA and the broker certificate | +| `build-config.sh` | writes the scratch broker config used by the verifier | +| `provision-clients.sh` | creates one broker account per registered device | +| `verify-isolation.sh` | proves the ACL on a throwaway broker | +| `clients.env` | **generated** device credentials (gitignored) | + +## The design + +Two listeners on the one broker. No second broker (§10 rule 6), no router +change, no DNS change. + +| Listener | Auth | Who uses it | +|---|---|---| +| `1883` (existing, unchanged) | anonymous | n8n and the rest of the lab | +| `8883` (new, TLS) | required account | the kids' devices and the Control Service | + +`per_listener_settings true` is what keeps the two planes apart: the plain +listener stays anonymous, the TLS listener requires credentials. It must appear +in the main `mosquitto.conf` **before the first `listener` line**. + +### Topic ACL + +```acl +user control-service +topic write restrict/+/config +topic read restrict/+/status +topic read restrict/+/telemetry + +pattern read restrict/%u/config +pattern write restrict/%u/status +pattern write restrict/%u/telemetry +``` + +- Only the Control Service publishes config (§10 rule 5). It cannot read config + and cannot write status or telemetry. +- A device's MQTT username **is** its topic token — the middle level of its + config topic. For `restrict/harry_phone_2/config` the account is + `harry_phone_2`. +- `%u` expands to that username, so a device can only ever touch its own topics. + Adding a device is therefore one account, no ACL edit. +- A device publishing to its own config topic is dropped by the broker, so a + child cannot publish a fake "unlocked". + +List the tokens the Control Service expects: + +```bash +docker exec webrestrict-control /app/control mqtt-clients +``` + +## Prove it without touching anything shared + +```bash +make mqtt-verify # TLS + CA + scratch broker on 127.0.0.1:18883 +``` + +`verify-isolation.sh` stages a throwaway copy of the config, runs its own +Mosquitto container on a spare port, and asserts the whole policy: the plain +listener is still anonymous, the TLS listener rejects anonymous clients and a +wrong password, a device reads only its own config, writes only its own status +and telemetry, cannot write another device's topics, and **cannot overwrite its +own config to fake an unlock**. + +## Why this is not applied to the shared broker yet + +Applying it changes files outside this project (`mosquitto.conf` and the +Mosquitto `docker-compose.yml`), adds port 8883 to a family-facing service, and +restarts that service. The Phase 3 guardrail is explicit: *do not change the +shared Mosquitto config.* So the change is prepared, generated and verified — +**not applied**. Applying it is a one-command step after Sam approves it, and is +tracked as the follow-up on AUD-80. + +## Applying it (after approval) + +Everything below is additive. Nothing existing is rewritten except the three +lines in step 2, and a backup is taken first. + +```bash +# 1. Credentials and certificates. +docker exec webrestrict-control /app/control mqtt-clients # tokens, no secrets +deploy/mosquitto/provision-clients.sh # writes conf.d/webrestrict.passwd + clients.env +make mqtt-tls # writes tls/{ca,server}.{crt,key} + +# 2. Copy the material into the shared broker's config directory. +BROKER_DIR=/home/sam/Docker/Containers/mosquitto +sudo -u sam cp "$BROKER_DIR/mosquitto.conf" \ + "$BROKER_DIR/mosquitto.conf.bak.$(date +%Y%m%d-%H%M%S)" +install -d -m 755 "$BROKER_DIR/conf.d" "$BROKER_DIR/tls" +install -m 644 deploy/mosquitto/conf.d/webrestrict.conf "$BROKER_DIR/conf.d/" +install -m 644 deploy/mosquitto/conf.d/webrestrict.acl "$BROKER_DIR/conf.d/" +install -m 644 deploy/mosquitto/conf.d/webrestrict.passwd "$BROKER_DIR/conf.d/" +install -m 644 deploy/mosquitto/tls/* "$BROKER_DIR/tls/" +# then add these three lines to mosquitto.conf: +# per_listener_settings true (before the first listener) +# include_dir /mosquitto/config/conf.d +# (leave `listener 1883 0.0.0.0` and `allow_anonymous true` exactly as they are) + +# 3. Publish 8883 and restart, bounded. +# add - "8883:8883" to the mosquitto service ports in +# $BROKER_DIR/docker-compose.yml, then: +cd "$BROKER_DIR" +timeout 120 docker compose up -d +timeout 30 docker logs --tail 20 mosquitto # expect both listeners to open + +# 4. Prove the plain plane still works (n8n depends on it). +timeout 40 docker run --rm --network host --entrypoint sh eclipse-mosquitto:latest -c \ + 'mosquitto_sub -h 127.0.0.1 -p 1883 -t lab/after -C 1 -W 10 & sleep 1; \ + mosquitto_pub -h 127.0.0.1 -p 1883 -t lab/after -m ok; wait' + +# 5. Point the Control Service at TLS. +# environment.d/control-service.env: +# CONTROL_MQTT_BROKER=tls://host.docker.internal:8883 +# CONTROL_MQTT_INSECURE=false +# CONTROL_MQTT_USERNAME=control-service +# CONTROL_MQTT_PASSWORD= +timeout 120 docker compose up -d --build +timeout 30 docker logs --tail 30 webrestrict-control # expect "mqtt connected" over tls +``` + +### Roll back + +```bash +cd /home/sam/Docker/Containers/mosquitto +cp mosquitto.conf.bak. mosquitto.conf +rm -rf conf.d tls +# remove the "8883:8883" port line from docker-compose.yml +timeout 120 docker compose up -d +``` + +Rolling back the broker needs no change in this project: set +`CONTROL_MQTT_INSECURE=true` and `CONTROL_MQTT_BROKER=tcp://host.docker.internal:1883` +in `environment.d/control-service.env` and restart the service. diff --git a/deploy/mosquitto/build-config.sh b/deploy/mosquitto/build-config.sh new file mode 100755 index 0000000..6b462bb --- /dev/null +++ b/deploy/mosquitto/build-config.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +# +# Build the scratch broker configuration used by `make mqtt-verify`. +# +# It mirrors the proposed change to the shared Mosquitto on `.13` exactly: +# the existing plain listener is left as it is, and the webrestrict listener is +# added beside it. The verification runs this config on a separate, throwaway +# container, so nothing shared is touched. +# +# build/mosquitto.conf the scratch main configuration (generated) +# conf.d/ the real listener, ACL and password files + +set -euo pipefail + +here="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +out="$here/build" + +mkdir -p "$out" + +cat >"$out/mosquitto.conf" <<'EOF' +# GENERATED by deploy/mosquitto/build-config.sh — do not edit. +# +# The scratch broker for `make mqtt-verify`. It is the proposed shape of the +# shared Mosquitto on `.13`: +# +# * the plain 1883 listener is unchanged, still anonymous, so n8n and the rest +# of the lab keep working; +# * the 8883 listener (conf.d/webrestrict.conf) is TLS-only and needs an +# account; +# * `per_listener_settings true` is what keeps those two planes apart. It +# belongs in the main configuration, before the first listener, and is the +# only change to the existing lines of mosquitto.conf. +persistence false +log_dest stdout +connection_messages true + +per_listener_settings true + +listener 1883 0.0.0.0 +allow_anonymous true + +include_dir /mosquitto/config/conf.d +EOF + +echo "build-config: wrote $out/mosquitto.conf" diff --git a/deploy/mosquitto/conf.d/webrestrict.acl b/deploy/mosquitto/conf.d/webrestrict.acl new file mode 100644 index 0000000..9e3e779 --- /dev/null +++ b/deploy/mosquitto/conf.d/webrestrict.acl @@ -0,0 +1,28 @@ +# Topic ACL for the Internet Use Control MQTT plane (AGENTS.md §10 rule 6). +# +# Included by conf.d/webrestrict.conf, which is reached only over TLS on 8883. +# A device may read only its own config and write only its own status and +# telemetry. Without this, a child can publish a fake "unlocked" (TRANSPORT_ +# DESIGN.md "Security"). +# +# The MQTT username of a device IS its topic token: the middle level of its +# config topic. For restrict/harry_phone_2/config the account is +# `harry_phone_2`. List the tokens with: +# +# docker exec webrestrict-control /app/control mqtt-clients +# +# and create the accounts with deploy/mosquitto/provision-clients.sh. + +# The Control Service. It publishes every device's config and reads every +# device's reports. It cannot read configs and cannot write status or telemetry. +user control-service +topic write restrict/+/config +topic read restrict/+/status +topic read restrict/+/telemetry + +# One rule set for every device. %u expands to the authenticated username, so a +# device is confined to its own topics by construction — no per-device edit is +# needed when a device is added. +pattern read restrict/%u/config +pattern write restrict/%u/status +pattern write restrict/%u/telemetry diff --git a/deploy/mosquitto/conf.d/webrestrict.conf b/deploy/mosquitto/conf.d/webrestrict.conf new file mode 100644 index 0000000..6b2b667 --- /dev/null +++ b/deploy/mosquitto/conf.d/webrestrict.conf @@ -0,0 +1,35 @@ +# Per-device MQTT security for the Internet Use Control Control Service. +# +# APPLY WITH deploy/mosquitto/README.md. This fragment is included by the shared +# Mosquitto on `.13`; it does not replace its configuration (AGENTS.md §11 +# Phase 3, §10 rule 6). +# +# Two listeners, two planes: +# * 1883 stays exactly as it is — plain, anonymous. n8n and the rest of the +# lab keep working (N8N_DEFAULT_MQTT_BROKER_URL=mqtt://192.168.20.13:1883). +# * 8883 is new and is the ONLY listener the kids' devices use: TLS, a +# required account, and a topic ACL. +# +# `per_listener_settings true` belongs in the main mosquitto.conf, before the +# first listener, not here. It is what keeps the 1883 plane anonymous while the +# 8883 plane requires credentials. + +listener 8883 0.0.0.0 + +# No anonymous devices. A device reads only its own config and writes only its +# own status and telemetry (webrestrict.acl). +allow_anonymous false +password_file /mosquitto/config/conf.d/webrestrict.passwd +acl_file /mosquitto/config/conf.d/webrestrict.acl + +# TLS. The lab CA is generated by generate-tls.sh; every device and the Control +# Service trust the same CA, so a rogue broker on the LAN cannot hand a device +# its rules. +tls_version tlsv1.2 +cafile /mosquitto/config/tls/ca.crt +certfile /mosquitto/config/tls/server.crt +keyfile /mosquitto/config/tls/server.key + +# A device proves who it is with a password; it does not need its own client +# certificate. +require_certificate false diff --git a/deploy/mosquitto/generate-tls.sh b/deploy/mosquitto/generate-tls.sh new file mode 100755 index 0000000..a58d9fd --- /dev/null +++ b/deploy/mosquitto/generate-tls.sh @@ -0,0 +1,113 @@ +#!/usr/bin/env bash +# +# Generate the lab CA and the broker certificate for the Internet Use Control +# MQTT plane (AGENTS.md §11 Phase 3, §10 rule 6). +# +# The CA is what makes "TLS" mean something on a home LAN: every device and the +# Control Service trust this CA, and the broker presents a certificate it signed. +# A rogue machine cannot impersonate the broker and hand a child new rules. +# +# Output (all gitignored, mode 0644 so the Mosquitto container user can read it): +# +# tls/ca.crt the lab CA certificate — ship this to every device +# tls/ca.key the CA private key — keep offline, never ship +# tls/server.crt the broker certificate +# tls/server.key the broker private key +# +# Re-run with --force to replace an existing set. Existing files are kept +# otherwise, so this is safe to call from `make mqtt-config`. +# +# Hostname override: +# MQTT_TLS_HOSTNAME=broker.example.lan MQTT_TLS_SANS=DNS:broker.example.lan,IP:10.0.0.5 \ +# deploy/mosquitto/generate-tls.sh --force +# +# The certificates are NOT for the public internet. They exist to authenticate +# the broker to the kids' devices on the LAN. + +set -euo pipefail + +here="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +tls_dir="$here/tls" + +hostname="${MQTT_TLS_HOSTNAME:-mosquitto.lab.audasmedia.com.au}" +sans="${MQTT_TLS_SANS:-DNS:${hostname},DNS:localhost,DNS:host.docker.internal,IP:127.0.0.1,IP:192.168.20.13}" +days="${MQTT_TLS_DAYS:-3650}" + +force=0 +case "${1:-}" in + --force | -f) force=1 ;; + --help | -h) + sed -n '2,30p' "$0" | sed 's/^# \{0,1\}//' + exit 0 + ;; + "") ;; + *) + echo "usage: $(basename "$0") [--force]" >&2 + exit 2 + ;; +esac + +mkdir -p "$tls_dir" +work_dir="$PWD" +cd "$tls_dir" + +if [[ -s "$tls_dir/ca.crt" && -s "$tls_dir/server.crt" && $force -eq 0 ]]; then + echo "generate-tls: $tls_dir already holds a CA and a server certificate; leaving it alone." + echo "generate-tls: re-run with --force to replace it." + exit 0 +fi + +# Prefer a host openssl. `.13` has no openssl, so fall back to a bounded +# container run (never an unbounded `docker run`, in line with the project +# guardrails). The container needs root to install the package, so it hands the +# generated files back to the calling user before it exits. +openssl_run() { + if command -v openssl >/dev/null 2>&1; then + openssl "$@" + return + fi + timeout 300 docker run --rm --network host \ + -e HOST_UID="$(id -u)" -e HOST_GID="$(id -g)" \ + -v "$tls_dir":/tls -w /tls alpine:3.21 sh -c ' + apk add --no-cache openssl >/dev/null 2>&1 || exit 1 + openssl "$@" || exit 1 + chown -R "$HOST_UID:$HOST_GID" /tls' -- "$@" +} + +echo "generate-tls: CA + broker certificate for $hostname" +echo "generate-tls: SANs $sans" + +# 1. The lab CA. +openssl_run req -x509 -newkey rsa:4096 -sha256 -days "$days" -nodes \ + -keyout ca.key -out ca.crt \ + -subj "/O=Family Home Lab/CN=Internet Use Control Lab CA" \ + -addext "basicConstraints=critical,CA:TRUE,pathlen:0" \ + -addext "keyUsage=critical,keyCertSign,cRLSign" + +# 2. The broker certificate, signed by the CA. The extension file lives in the +# certificate directory so the same relative path works for a host openssl and +# for the fallback container, which is started with that directory as its cwd. +ext=.ext.cnf +cat >"$ext" <&2 + exit 1 +} + +command -v docker >/dev/null 2>&1 || die "docker is not available" + +# head reads a fixed number of bytes from the device and cut consumes its whole +# input, so no stage of this pipeline is signalled with SIGPIPE. +rand_pass() { + head -c 32 /dev/urandom | base64 | tr -d '/+=' | cut -c1-32 +} + +[[ -n "$service_pass" ]] || service_pass="$(rand_pass)" + +# The device list comes from the service, so the broker accounts can never +# drift from the devices the Control Service knows about. +clients="$(timeout 30 docker exec "$container" /app/control mqtt-clients)" || + die "could not read the device list from container '$container' (is it running?)" + +tmp_dir="$(mktemp -d)" +chmod 700 "$tmp_dir" +trap 'rm -rf "$tmp_dir"' EXIT + +add_user() { # user password (creates the file on the first call) + local user="$1" pass="$2" args=(-b) + [[ -s "$tmp_dir/webrestrict.passwd" ]] || args+=(-c) + timeout 60 docker run --rm --user "$(id -u):$(id -g)" \ + -v "$tmp_dir":/cfg --entrypoint mosquitto_passwd "$broker_image" \ + "${args[@]}" /cfg/webrestrict.passwd "$user" "$pass" >/dev/null || + die "mosquitto_passwd failed for $user" +} + +add_user "$service_user" "$service_pass" + +: >"$tmp_dir/clients.env" +{ + echo "# GENERATED by deploy/mosquitto/provision-clients.sh — secrets, do not commit." + echo "#" + echo "# Control Service account. Put these two lines in" + echo "# environment.d/control-service.env, then set CONTROL_MQTT_INSECURE=false" + echo "# and CONTROL_MQTT_BROKER=tls://host.docker.internal:8883." + echo "CONTROL_MQTT_USERNAME=$service_user" + echo "CONTROL_MQTT_PASSWORD=$service_pass" + echo + echo "# One account per device. The name is the device's topic token; the" + echo "# password goes into the device's own MQTT settings." + echo "# tokenpasswordconfig topic" +} >>"$tmp_dir/clients.env" + +count=0 +while IFS=$'\t' read -r token topic platform child name; do + # mqtt-clients prints real tabs; names contain spaces, so the split matters. + token="$(printf '%s' "${token:-}" | tr -d '[:space:]')" + [[ -n "$token" ]] || continue + pass="$(rand_pass)" + add_user "$token" "$pass" + printf '%s\t%s\t%s\n' "$token" "$pass" "$topic" >>"$tmp_dir/clients.env" + count=$((count + 1)) + echo "provision-clients: account '$token' ($name · $platform) -> $topic" +done <<<"$clients" + +[[ $count -gt 0 ]] || echo "provision-clients: no devices registered yet; only the Control Service account was created." + +install -m 600 "$tmp_dir/webrestrict.passwd" "$passwd_file" +install -m 600 "$tmp_dir/clients.env" "$secrets_file" + +echo "provision-clients: wrote $passwd_file ($((count + 1)) accounts)" +echo "provision-clients: wrote $secrets_file — read it, do not commit it." diff --git a/deploy/mosquitto/tls/README.md b/deploy/mosquitto/tls/README.md new file mode 100644 index 0000000..81db477 --- /dev/null +++ b/deploy/mosquitto/tls/README.md @@ -0,0 +1,25 @@ +# The lab CA and broker certificate + +`generate-tls.sh` writes the certificate material here: + +| File | Purpose | Who gets it | +|---|---|---| +| `ca.crt` | the lab CA certificate | every device and the Control Service | +| `ca.key` | the CA private key | **not** a device; keep it off-device | +| `server.crt` | the broker certificate | the broker | +| `server.key` | the broker private key | the broker | + +Everything in this directory except this README is gitignored: certificates and +keys are generated per host and never committed. + +```bash +make mqtt-tls # writes the four files (skips them if they exist) +make mqtt-tls-force # use generate-tls.sh --force to replace them +``` + +The broker runs as user `mosquitto` (uid 1883) inside its container, and this +directory is bind-mounted read-only, so the files are created mode **0644**. That +makes the private keys readable on `.13` by any local user. It is the pragmatic +choice for a LAN-only broker whose key must be readable by a container user that +does not exist on the host. Replace it with a group-mapped setup if `.13` ever +gets untrusted local users. diff --git a/deploy/mosquitto/verify-isolation.sh b/deploy/mosquitto/verify-isolation.sh new file mode 100755 index 0000000..ea1fe6c --- /dev/null +++ b/deploy/mosquitto/verify-isolation.sh @@ -0,0 +1,256 @@ +#!/usr/bin/env bash +# +# Prove the Phase 3 MQTT security design on a throwaway broker. +# +# This script NEVER touches the shared Mosquitto on `.13`. It builds a scratch +# configuration with the same shape as the proposed shared change, runs it in its +# own container on a spare port, and asserts the policy in +# conf.d/webrestrict.acl: +# +# * the plain listener is still anonymous, so n8n and the lab keep working; +# * the TLS listener refuses anonymous clients and a wrong password; +# * a device can read only its own config; +# * a device can write only its own status and telemetry; +# * a device cannot fake an unlock by publishing to its own config topic. +# +# Usage: make mqtt-verify +# MQTT_VERIFY_PORT host port for the TLS listener (default 18883) + +set -euo pipefail + +here="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +broker_image="${MQTT_IMAGE:-eclipse-mosquitto:latest}" +tls_port="${MQTT_VERIFY_PORT:-18883}" +plain_port=$((tls_port + 1)) +name="webrestrict-mqtt-verify" +service_user="control-service" +# A throwaway password. head reads a fixed number of bytes from the device and +# cut consumes its whole input, so no stage of this pipeline is signalled with +# SIGPIPE (which `set -o pipefail` would turn into a failure). +rand_pass() { head -c 32 /dev/urandom | base64 | tr -d '/+=' | cut -c1-32; } +service_pass="$(rand_pass)" +dev_a_pass="$(rand_pass)" +dev_b_pass="$(rand_pass)" + +work="$(mktemp -d)" +chmod 755 "$work" +cleanup() { + timeout 30 docker rm -f "$name" >/dev/null 2>&1 || true + rm -rf "$work" +} +trap cleanup EXIT + +pass=0 +fail=0 +ok() { + echo " PASS $1" + pass=$((pass + 1)) +} +no() { + echo " FAIL $1" + fail=$((fail + 1)) +} + +die() { + echo "mqtt-verify: $*" >&2 + exit 1 +} + +command -v docker >/dev/null 2>&1 || die "docker is not available" + +[[ -s "$here/conf.d/webrestrict.conf" ]] || die "run build-config.sh first (make mqtt-config)" +[[ -s "$here/tls/ca.crt" ]] || die "no CA; run generate-tls.sh first (make mqtt-tls)" + +# --- stage a throwaway config root ----------------------------------------- +echo "mqtt-verify: staging a scratch config in $work" +mkdir -p "$work/build" "$work/conf.d" "$work/tls" +cp "$here/conf.d/webrestrict.conf" "$work/conf.d/" +cp "$here/conf.d/webrestrict.acl" "$work/conf.d/" +cp "$here/tls/ca.crt" "$here/tls/server.crt" "$here/tls/server.key" "$work/tls/" +if [[ -s "$here/build/mosquitto.conf" ]]; then + cp "$here/build/mosquitto.conf" "$work/build/" +else + "$here/build-config.sh" >/dev/null + cp "$here/build/mosquitto.conf" "$work/build/" +fi + +# Throwaway accounts, generated with the broker's own tool. mosquitto_passwd +# refuses to overwrite an existing file, so only the first account creates it. +rm -f "$work/conf.d/webrestrict.passwd" +for pair in "$service_user:$service_pass" "testdev_a:$dev_a_pass" "testdev_b:$dev_b_pass"; do + user="${pair%%:*}" + secret="${pair#*:}" + args=(-b) + [[ -s "$work/conf.d/webrestrict.passwd" ]] || args+=(-c) + timeout 60 docker run --rm --user "$(id -u):$(id -g)" \ + -v "$work/conf.d":/cfg --entrypoint mosquitto_passwd "$broker_image" \ + "${args[@]}" /cfg/webrestrict.passwd "$user" "$secret" >/dev/null || + die "mosquitto_passwd failed" +done +chmod -R a+rX "$work" + +# --- run the scratch broker ------------------------------------------------- +timeout 30 docker rm -f "$name" >/dev/null 2>&1 || true +echo "mqtt-verify: starting $name (tls :$tls_port -> 8883, plain :$plain_port -> 1883)" +timeout 60 docker run -d --name "$name" \ + -v "$work":/mosquitto/config:ro \ + -p "127.0.0.1:$tls_port:8883" -p "127.0.0.1:$plain_port:1883" \ + "$broker_image" /usr/sbin/mosquitto -c /mosquitto/config/build/mosquitto.conf >/dev/null + +for _ in $(seq 1 40); do + if timeout 10 docker logs "$name" 2>&1 | grep -c 'mosquitto version .* running' >/dev/null; then break; fi + sleep 0.5 +done +sleep 1 +if [[ "$(timeout 20 docker inspect -f '{{.State.Running}}' "$name")" != "true" ]]; then + timeout 20 docker logs "$name" 2>&1 | tail -20 + die "the scratch broker did not stay up (see its log above)" +fi + +# --- helpers ---------------------------------------------------------------- +# mosquitto_pub/sub run in a client container that shares the host network. +pub() { # user pass topic payload [retain] + local retain="" + [[ -n "${5:-}" ]] && retain="-r" + timeout 40 docker run --rm --network host -v "$work":/work:ro \ + --entrypoint mosquitto_pub "$broker_image" \ + -h 127.0.0.1 -p "$tls_port" --cafile /work/tls/ca.crt \ + -u "$1" -P "$2" -t "$3" -m "$4" $retain 2>&1 +} + +# read retains everything published to the topic; prints what it got, or "" on +# timeout. +read_topic() { # user pass topic wait_secs + timeout 40 docker run --rm --network host -v "$work":/work:ro \ + --entrypoint mosquitto_sub "$broker_image" \ + -h 127.0.0.1 -p "$tls_port" --cafile /work/tls/ca.crt \ + -u "$1" -P "$2" -t "$3" -C 1 -W "$4" 2>/dev/null || true +} + +# roundtrip publishes as one account and reads as another, so a write permission +# and a read permission are both exercised. It prints what the reader received, +# or nothing. +roundtrip() { # pub_user pub_pass sub_user sub_pass topic payload + timeout 60 docker run --rm --network host -e PORT="$tls_port" -v "$work":/work:ro \ + --entrypoint sh "$broker_image" -c ' + mosquitto_sub -h 127.0.0.1 -p "$PORT" --cafile /work/tls/ca.crt \ + -u "$1" -P "$2" -t "$3" -C 1 -W 10 >/tmp/out 2>/dev/null & + s=$!; sleep 1 + mosquitto_pub -h 127.0.0.1 -p "$PORT" --cafile /work/tls/ca.crt \ + -u "$4" -P "$5" -t "$3" -m "$6" >/dev/null 2>&1 || true + wait $s; cat /tmp/out' \ + sh "$3" "$4" "$5" "$1" "$2" "$6" 2>/dev/null || true +} + +echo +echo "== 1. the plain listener is unchanged (the lab keeps working) ==" +plain="$(timeout 60 docker run --rm --network host --entrypoint sh "$broker_image" -c ' + mosquitto_sub -h 127.0.0.1 -p '"$plain_port"' -t lab/verify -C 1 -W 10 >/tmp/out 2>&1 & + s=$!; sleep 1 + mosquitto_pub -h 127.0.0.1 -p '"$plain_port"' -t lab/verify -m plain-plane-ok >/dev/null 2>&1 || true + wait $s; cat /tmp/out' 2>/dev/null || true)" +if [[ "$plain" == "plain-plane-ok" ]]; then + ok "anonymous client on the plain listener can publish and subscribe" +else + no "anonymous client on the plain listener was blocked (got '$plain')" +fi + +echo +echo "== 2. the TLS listener rejects what it should ==" +if timeout 40 docker run --rm --network host --entrypoint mosquitto_pub "$broker_image" \ + -h 127.0.0.1 -p "$tls_port" -t restrict/testdev_a/config -m nope >/dev/null 2>&1; then + no "a plain TCP client was accepted on the TLS port" +else + ok "a plain TCP client is rejected on the TLS port (TLS really is on)" +fi + +if timeout 40 docker run --rm --network host -v "$work":/work:ro --entrypoint mosquitto_pub "$broker_image" \ + -h 127.0.0.1 -p "$tls_port" --cafile /work/tls/ca.crt \ + -t restrict/testdev_a/config -m nope >/dev/null 2>&1; then + no "an anonymous client was accepted over TLS" +else + ok "an anonymous client is refused over TLS" +fi + +if timeout 40 docker run --rm --network host -v "$work":/work:ro --entrypoint mosquitto_pub "$broker_image" \ + -h 127.0.0.1 -p "$tls_port" --cafile /work/tls/ca.crt \ + -u testdev_a -P wrong-password -t restrict/testdev_a/status -m nope >/dev/null 2>&1; then + no "a wrong password was accepted" +else + ok "a wrong password is refused" +fi + +echo +echo "== 3. one device is confined to its own topics ==" +payload='{"v":1,"locked":true,"src":"control"}' +if pub "$service_user" "$service_pass" restrict/testdev_a/config "$payload" retain >/dev/null; then + ok "the Control Service can publish retained config" +else + no "the Control Service could not publish config" +fi + +got="$(read_topic testdev_a "$dev_a_pass" restrict/testdev_a/config 5)" +if [[ "$got" == "$payload" ]]; then + ok "a device reads its own retained config" +else + no "a device did not read its own config (got '$got')" +fi + +got="$(read_topic testdev_b "$dev_b_pass" restrict/testdev_a/config 4)" +if [[ -z "$got" ]]; then + ok "a device cannot read another device's config" +else + no "a device read another device's config (got '$got')" +fi + +got="$(roundtrip testdev_a "$dev_a_pass" "$service_user" "$service_pass" restrict/testdev_a/status '{"online":true}')" +if [[ "$got" == '{"online":true}' ]]; then + ok "a device writes its own status and the Control Service reads it" +else + no "a device's own status did not arrive (got '$got')" +fi + +got="$(roundtrip testdev_a "$dev_a_pass" "$service_user" "$service_pass" restrict/testdev_a/telemetry '{"app":"com.spotify.music","seconds":60}')" +if [[ "$got" == *com.spotify.music* ]]; then + ok "a device writes its own telemetry and the Control Service reads it" +else + no "a device's own telemetry did not arrive (got '$got')" +fi + +got="$(roundtrip testdev_a "$dev_a_pass" "$service_user" "$service_pass" restrict/testdev_b/status '{"online":true}')" +if [[ -z "$got" ]]; then + ok "a device cannot write another device's status" +else + no "a device wrote another device's status (got '$got')" +fi + +got="$(roundtrip testdev_a "$dev_a_pass" "$service_user" "$service_pass" restrict/testdev_b/telemetry '{"app":"x","seconds":1}')" +if [[ -z "$got" ]]; then + ok "a device cannot write another device's telemetry" +else + no "a device wrote another device's telemetry (got '$got')" +fi + +echo +echo "== 4. a device cannot fake an unlock ==" +pub testdev_a "$dev_a_pass" restrict/testdev_a/config '{"v":1,"locked":false,"src":"device"}' retain >/dev/null || true +got="$(read_topic testdev_a "$dev_a_pass" restrict/testdev_a/config 5)" +if [[ "$got" == "$payload" ]]; then + ok "a device's publish to its own config topic was ignored (still $payload)" +else + no "a device overwrote its own config (got '$got')" +fi + +echo +echo "== 5. the Control Service cannot be spoofed either ==" +echo +got="$(roundtrip "$service_user" "$service_pass" "$service_user" "$service_pass" restrict/testdev_a/status '{"online":false}')" +if [[ -z "$got" ]]; then + ok "the Control Service identity cannot publish status (read-only there)" +else + no "the Control Service identity published status (got '$got')" +fi + +echo +echo "mqtt-verify: $pass passed, $fail failed" +[[ $fail -eq 0 ]] || exit 1 diff --git a/docker-compose.yml b/docker-compose.yml index bdad663..3d5d358 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,6 +1,9 @@ # Internet Use Control — Control Service (own image, own container, own port). # Deploy on .13: cd /home/sam/paseo/projects/internet_use_control && docker compose up -d --build # The service is the single source of truth (SYSTEM_MAP.md). Only it publishes config. +# +# MQTT is transport, not storage: the service publishes retained config to the +# shared Mosquitto on `.13:1883` and writes what devices report into SQLite. name: webrestrict @@ -9,16 +12,24 @@ services: build: context: . args: - VERSION: "0.2.0-phase2" - image: webrestrict/control:0.2.0-phase2 + VERSION: "0.3.0-phase3" + image: webrestrict/control:0.3.0-phase3 container_name: webrestrict-control restart: unless-stopped env_file: - environment.d/control-service.env ports: - "8791:8791" + # The broker is the Mosquitto container on the host. No host IP is baked + # in: the gateway is resolved at start, so CONTROL_MQTT_BROKER can stay + # "tcp://host.docker.internal:1883". + extra_hosts: + - "host.docker.internal:host-gateway" volumes: - webrestrict-data:/data + # The lab CA for broker TLS verification. Read-only and optional: the + # file is absent until the TLS listener exists. + - ./deploy/mosquitto/tls:/etc/webrestrict/tls:ro healthcheck: test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8791/health"] interval: 30s diff --git a/environment.d/control-service.env.example b/environment.d/control-service.env.example index 6d4aeba..ac4f7eb 100644 --- a/environment.d/control-service.env.example +++ b/environment.d/control-service.env.example @@ -34,5 +34,33 @@ CONTROL_ADMIN_USERNAME=sam CONTROL_ADMIN_PASSWORD=change-me # Option B (preferred): a bcrypt hash; leave CONTROL_ADMIN_PASSWORD empty. -# Generate: docker run --rm --entrypoint /app/control webrestrict/control:0.1.0-phase1 hash-password 'your-password' +# Generate: docker run --rm --entrypoint /app/control webrestrict/control:0.3.0-phase3 hash-password 'your-password' # CONTROL_ADMIN_PASSWORD_HASH=$2a$10$... + +# --- MQTT (Phase 3) -------------------------------------------------------- +# The broker is transport only. SQLite stays the source of truth, and only the +# Control Service publishes config (AGENTS.md §10 rules 5-6). + +# Turn the broker link on or off. With it off the dashboard still works and the +# config stays in SQLite, unpublished. +CONTROL_MQTT_ENABLED=true + +# The shared Mosquitto on `.13`, seen from inside this container. The compose +# file maps host.docker.internal to the host gateway. +# When the TLS listener exists, change this to tls://host.docker.internal:8883. +CONTROL_MQTT_BROKER=tcp://host.docker.internal:1883 + +# The Control Service identity. The broker ACL gives it write access to every +# config topic and read access to every status and telemetry topic, nothing else. +CONTROL_MQTT_CLIENT_ID=control-service +# CONTROL_MQTT_USERNAME=control-service +# CONTROL_MQTT_PASSWORD=change-me + +# PEM bundle used to verify the broker certificate. Required for a tls:// broker. +CONTROL_MQTT_CA_FILE=/etc/webrestrict/tls/ca.crt + +# TEMPORARY. The shared Mosquitto has no TLS listener yet, so a plaintext, +# credential-less link is allowed. The service refuses to start without +# credentials or TLS unless this is true, and it logs a warning when it is. +# Set to false as soon as deploy/mosquitto/ is applied (AGENTS.md §10 rule 6). +CONTROL_MQTT_INSECURE=true diff --git a/go.mod b/go.mod index b73e1cb..15abf8b 100644 --- a/go.mod +++ b/go.mod @@ -3,17 +3,21 @@ module gitea.lab.audasmedia.com.au/sam/internet_use_control go 1.25 require ( - golang.org/x/crypto v0.31.0 + github.com/eclipse/paho.mqtt.golang v1.5.1 + golang.org/x/crypto v0.42.0 modernc.org/sqlite v1.34.5 ) require ( github.com/dustin/go-humanize v1.0.1 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v0.1.9 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect - golang.org/x/sys v0.28.0 // indirect + golang.org/x/net v0.44.0 // indirect + golang.org/x/sync v0.17.0 // indirect + golang.org/x/sys v0.36.0 // indirect modernc.org/libc v1.55.3 // indirect modernc.org/mathutil v1.6.0 // indirect modernc.org/memory v1.8.0 // indirect diff --git a/go.sum b/go.sum index aa3cb2b..50cd4d7 100644 --- a/go.sum +++ b/go.sum @@ -1,22 +1,30 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= +github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo= github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= -golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= -golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= +golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI= +golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8= golang.org/x/mod v0.16.0 h1:QX4fJ0Rr5cPQCF7O9lh9Se4pmwfwskqZfq5moyldzic= golang.org/x/mod v0.16.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= +golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= -golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k= +golang.org/x/sys v0.36.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw= golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc= modernc.org/cc/v4 v4.21.4 h1:3Be/Rdo1fpr8GrQ7IVw9OHtplU4gWbb+wNgeoBMmGLQ= diff --git a/internal/config/config.go b/internal/config/config.go index e966698..629292e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -37,6 +37,25 @@ type Config struct { AppEnv string // Version is injected at build time. Version string + + // MQTTEnabled turns the broker link on. When false the service runs + // HTTP-only and never touches the broker. + MQTTEnabled bool + // MQTTBroker is the broker URL, for example "tls://mosquitto:8883" or + // "tcp://127.0.0.1:1883". + MQTTBroker string + // MQTTClientID is the MQTT client id of the Control Service. + MQTTClientID string + // MQTTUsername and MQTTPassword are the Control Service credentials + // (AGENTS.md §10 rule 6: per-device credentials). + MQTTUsername string + MQTTPassword string + // MQTTCAFile is the PEM bundle used to verify the broker certificate. + MQTTCAFile string + // MQTTInsecure allows a plaintext, credential-less broker link. It exists + // only until the shared Mosquitto gets the TLS listener; never set it in + // production. + MQTTInsecure bool } // Load reads and validates the configuration. @@ -49,6 +68,13 @@ func Load() (*Config, error) { AdminPasswordHash: os.Getenv("CONTROL_ADMIN_PASSWORD_HASH"), CookieSecure: envBool("CONTROL_COOKIE_SECURE", true), AppEnv: env("CONTROL_APP_ENV", "production"), + MQTTEnabled: envBool("CONTROL_MQTT_ENABLED", true), + MQTTBroker: env("CONTROL_MQTT_BROKER", DefaultMQTTBroker), + MQTTClientID: env("CONTROL_MQTT_CLIENT_ID", "control-service"), + MQTTUsername: os.Getenv("CONTROL_MQTT_USERNAME"), + MQTTPassword: os.Getenv("CONTROL_MQTT_PASSWORD"), + MQTTCAFile: env("CONTROL_MQTT_CA_FILE", "/etc/webrestrict/tls/ca.crt"), + MQTTInsecure: envBool("CONTROL_MQTT_INSECURE", false), } secret := os.Getenv("CONTROL_SESSION_SECRET") @@ -76,9 +102,45 @@ func Load() (*Config, error) { } cfg.LogLevel = lvl + if err := cfg.validateMQTT(); err != nil { + return nil, err + } + return cfg, nil } +// DefaultMQTTBroker is where the shared Mosquitto on `.13` lives, as seen from +// inside this project's container. The compose file maps host.docker.internal +// to the host gateway. +const DefaultMQTTBroker = "tcp://host.docker.internal:1883" + +// MQTTTLS reports whether the broker URL asks for TLS. +func (c *Config) MQTTTLS() bool { + return strings.HasPrefix(c.MQTTBroker, "tls://") || strings.HasPrefix(c.MQTTBroker, "ssl://") +} + +// validateMQTT refuses a link that carries neither TLS nor credentials, unless +// CONTROL_MQTT_INSECURE explicitly allows it (AGENTS.md §10 rule 6). +func (c *Config) validateMQTT() error { + if !c.MQTTEnabled { + return nil + } + if c.MQTTBroker == "" { + return errors.New("CONTROL_MQTT_BROKER must not be empty when MQTT is enabled") + } + if c.MQTTUsername == "" || c.MQTTPassword == "" { + if !c.MQTTInsecure { + return errors.New("CONTROL_MQTT_USERNAME and CONTROL_MQTT_PASSWORD are required; " + + "set CONTROL_MQTT_INSECURE=true only for the pre-TLS lab broker") + } + } + if !c.MQTTTLS() && !c.MQTTInsecure { + return errors.New("CONTROL_MQTT_BROKER must use tls:// (AGENTS.md §10 rule 6); " + + "set CONTROL_MQTT_INSECURE=true only for the pre-TLS lab broker") + } + return nil +} + func env(key, fallback string) string { if v := strings.TrimSpace(os.Getenv(key)); v != "" { return v diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5529d70..24207cd 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -11,6 +11,9 @@ func TestLoadDefaultsAndValidation(t *testing.T) { t.Setenv("CONTROL_ADMIN_PASSWORD", "hunter2") t.Setenv("CONTROL_ADDR", "") t.Setenv("CONTROL_DB_PATH", "") + // The lab broker has no TLS listener yet, so the pre-TLS escape hatch is + // the current deployment default (see CONTROL_MQTT_INSECURE). + t.Setenv("CONTROL_MQTT_INSECURE", "true") cfg, err := Load() if err != nil { @@ -50,3 +53,56 @@ func TestLoadRequiresSecretAndPassword(t *testing.T) { t.Error("expected error when no admin password is set") } } + +func TestMQTTDefaultsAndGuards(t *testing.T) { + t.Setenv("CONTROL_SESSION_SECRET", "0123456789abcdef") + t.Setenv("CONTROL_ADMIN_PASSWORD", "hunter2") + + // Enabled by default, pointed at the shared Mosquitto on `.13`. + t.Setenv("CONTROL_MQTT_INSECURE", "true") + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if !cfg.MQTTEnabled { + t.Error("MQTTEnabled = false, want true by default") + } + if cfg.MQTTBroker != DefaultMQTTBroker { + t.Errorf("MQTTBroker = %q, want %q", cfg.MQTTBroker, DefaultMQTTBroker) + } + if cfg.MQTTTLS() { + t.Error("MQTTTLS = true for the default plaintext broker, want false") + } + + // Without the escape hatch, a plaintext credential-less broker is refused + // (AGENTS.md §10 rule 6). + t.Setenv("CONTROL_MQTT_INSECURE", "false") + if _, err := Load(); err == nil { + t.Error("expected an error for a plaintext broker without CONTROL_MQTT_INSECURE") + } + + // TLS plus credentials is the supported shape. + t.Setenv("CONTROL_MQTT_BROKER", "tls://mosquitto.lab.audasmedia.com.au:8883") + t.Setenv("CONTROL_MQTT_USERNAME", "control-service") + t.Setenv("CONTROL_MQTT_PASSWORD", "secret") + cfg, err = Load() + if err != nil { + t.Fatalf("Load with TLS: %v", err) + } + if !cfg.MQTTTLS() { + t.Error("MQTTTLS = false for a tls:// broker, want true") + } + + // MQTT can be switched off entirely. + t.Setenv("CONTROL_MQTT_ENABLED", "false") + t.Setenv("CONTROL_MQTT_BROKER", "") + t.Setenv("CONTROL_MQTT_USERNAME", "") + t.Setenv("CONTROL_MQTT_PASSWORD", "") + cfg, err = Load() + if err != nil { + t.Fatalf("Load with MQTT disabled: %v", err) + } + if cfg.MQTTEnabled { + t.Error("MQTTEnabled = true, want false when CONTROL_MQTT_ENABLED=false") + } +} diff --git a/internal/mqtt/link.go b/internal/mqtt/link.go new file mode 100644 index 0000000..dc49a1f --- /dev/null +++ b/internal/mqtt/link.go @@ -0,0 +1,414 @@ +// The broker link: connect, subscribe to device reports, publish retained +// config. Everything that reaches SQLite is handled here (AGENTS.md §11 +// Phase 3). +package mqtt + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "log/slog" + "net/url" + "os" + "strings" + "sync" + "sync/atomic" + "time" + + paho "github.com/eclipse/paho.mqtt.golang" + + "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store" +) + +// DeviceLookup finds a device from a topic it wrote to. +type DeviceLookup interface { + DeviceByConfigTopic(ctx context.Context, topic string) (*store.Device, error) +} + +// ReportSink stores what a device reported. +type ReportSink interface { + SetPresence(ctx context.Context, id int64, online bool, at time.Time, actor string) (bool, error) + InsertTelemetry(ctx context.Context, id int64, day string, entries []store.TelemetryEntry) (int, error) +} + +// ConfigSource reads the desired state that is published to devices. +type ConfigSource interface { + ListDevices(ctx context.Context) ([]store.Device, error) + DeviceLists(ctx context.Context, id int64) (allow, block []string, err error) +} + +// Options configures the link. +type Options struct { + // Broker is a URL such as "tls://mosquitto.lab:8883" or, until the shared + // broker gets its TLS listener, "tcp://host.docker.internal:1883". + Broker string + // ClientID identifies the Control Service on the broker. + ClientID string + // Username and Password are the Control Service credentials. The broker's + // ACL gives this identity write access to every config topic and read + // access to every status and telemetry topic — and nothing else. + Username string + Password string + // CAFile is a PEM bundle used to verify the broker certificate. + CAFile string + // KeepAlive is the MQTT keepalive. + KeepAlive time.Duration + // ConnectTimeout bounds one connection attempt. + ConnectTimeout time.Duration + + // Actor is the audit-log actor for reports that arrive over the broker. + Actor string + // Log receives structured logs. + Log *slog.Logger + + // Devices and Reports are required. + Devices DeviceLookup + // Source is required to publish config. + Source ConfigSource + // Reports is required to store device reports. + Reports ReportSink + // OnConnected runs after every successful (re)connect. The service uses it + // to republish every device's retained config. + OnConnected func() +} + +// Link is the live connection to the broker. +type Link struct { + opts Options + log *slog.Logger + + client paho.Client + + connected atomic.Bool + received atomic.Int64 + published atomic.Int64 + lastMsg atomic.Int64 // unix nanos + lastErrMu sync.Mutex + lastErrMsg string +} + +// New builds the link. It does not connect; call Start. +func New(opts Options) (*Link, error) { + if opts.Devices == nil || opts.Source == nil || opts.Reports == nil { + return nil, errors.New("mqtt: Devices, Source and Reports are required") + } + if strings.TrimSpace(opts.Broker) == "" { + return nil, errors.New("mqtt: Broker is required") + } + if opts.ClientID == "" { + opts.ClientID = "control-service" + } + if opts.KeepAlive <= 0 { + opts.KeepAlive = 30 * time.Second + } + if opts.ConnectTimeout <= 0 { + opts.ConnectTimeout = 10 * time.Second + } + if opts.Actor == "" { + opts.Actor = "device" + } + if opts.Log == nil { + opts.Log = slog.Default() + } + + l := &Link{opts: opts, log: opts.Log} + + clientOpts := paho.NewClientOptions() + clientOpts.AddBroker(opts.Broker) + clientOpts.SetClientID(opts.ClientID) + clientOpts.SetUsername(opts.Username) + clientOpts.SetPassword(opts.Password) + clientOpts.SetKeepAlive(opts.KeepAlive) + clientOpts.SetPingTimeout(min(10*time.Second, opts.KeepAlive)) + clientOpts.SetConnectTimeout(opts.ConnectTimeout) + clientOpts.SetWriteTimeout(opts.ConnectTimeout) + // The service is a publisher of retained truth and a listener for reports: + // reconnect forever, and re-subscribe on every connect. + clientOpts.SetAutoReconnect(true) + clientOpts.SetConnectRetry(true) + clientOpts.SetConnectRetryInterval(5 * time.Second) + clientOpts.SetMaxReconnectInterval(60 * time.Second) + // Clean session: the broker keeps the retained config, so the service does + // not need a queue of stale reports. + clientOpts.SetCleanSession(true) + clientOpts.SetOrderMatters(false) + + if isTLSBroker(opts.Broker) { + tlsCfg, err := tlsConfig(opts.Broker, opts.CAFile) + if err != nil { + return nil, err + } + clientOpts.SetTLSConfig(tlsCfg) + } + + clientOpts.SetOnConnectHandler(l.handleConnect) + clientOpts.SetConnectionLostHandler(func(_ paho.Client, err error) { + l.connected.Store(false) + l.setErr(err) + l.log.Warn("mqtt connection lost", "broker", opts.Broker, "err", err) + }) + + l.client = paho.NewClient(clientOpts) + return l, nil +} + +// Start connects in the background. A broker outage must never stop the HTTP +// service: the dashboard keeps working and the link reconnects on its own. +func (l *Link) Start(ctx context.Context) { + l.log.Info("mqtt connecting", "broker", l.opts.Broker, "client_id", l.opts.ClientID, + "tls", isTLSBroker(l.opts.Broker), "username", l.opts.Username) + go func() { + if token := l.client.Connect(); token.Wait() && token.Error() != nil { + l.setErr(token.Error()) + l.log.Error("mqtt connect", "err", token.Error()) + } + }() +} + +// Stop disconnects cleanly. The broker then publishes nothing on our behalf; +// the service's own will is not registered, because only the service publishes +// config and a stale config must survive a service restart. +func (l *Link) Stop() { + if l.client == nil { + return + } + if l.client.IsConnected() { + l.client.Unsubscribe(StatusSubscription, TelemetrySubscription) + } + l.client.Disconnect(250) + l.connected.Store(false) +} + +// Broker returns the configured broker URL. +func (l *Link) Broker() string { return l.opts.Broker } + +// Connected reports the current connection state. +func (l *Link) Connected() bool { return l.connected.Load() } + +// Stats returns counters for the dashboard. +func (l *Link) Stats() (received, published int64, lastMessageAt time.Time, lastErr string) { + ns := l.lastMsg.Load() + if ns > 0 { + lastMessageAt = time.Unix(0, ns) + } + l.lastErrMu.Lock() + lastErr = l.lastErrMsg + l.lastErrMu.Unlock() + return l.received.Load(), l.published.Load(), lastMessageAt, lastErr +} + +func (l *Link) setErr(err error) { + if err == nil { + return + } + l.lastErrMu.Lock() + l.lastErrMsg = err.Error() + l.lastErrMu.Unlock() +} + +// handleConnect runs on every (re)connect, including the first one. +func (l *Link) handleConnect(c paho.Client) { + l.connected.Store(true) + l.setErr(nil) + l.log.Info("mqtt connected", "broker", l.opts.Broker) + + for _, sub := range []struct { + topic string + handler paho.MessageHandler + }{ + {StatusSubscription, l.handleStatusMessage}, + {TelemetrySubscription, l.handleTelemetryMessage}, + } { + token := c.Subscribe(sub.topic, 1, sub.handler) + if !token.WaitTimeout(l.opts.ConnectTimeout) { + l.log.Warn("mqtt subscribe timed out", "topic", sub.topic) + continue + } + if err := token.Error(); err != nil { + l.setErr(err) + l.log.Error("mqtt subscribe", "topic", sub.topic, "err", err) + continue + } + l.log.Info("mqtt subscribed", "topic", sub.topic) + } + + if l.opts.OnConnected != nil { + l.opts.OnConnected() + } +} + +// handleStatusMessage turns a device presence report into a store update. +func (l *Link) handleStatusMessage(_ paho.Client, msg paho.Message) { + l.noteMessage() + + configTopic, ok := deviceFromTopic(msg.Topic(), SuffixStatus) + if !ok { + l.log.Warn("mqtt status on an unexpected topic", "topic", msg.Topic()) + return + } + + status, err := ParseStatus(msg.Payload()) + if err != nil { + l.log.Warn("mqtt status rejected", "topic", msg.Topic(), "err", err) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + dev, err := l.opts.Devices.DeviceByConfigTopic(ctx, configTopic) + if err != nil { + l.log.Warn("mqtt status for an unknown device", "topic", msg.Topic(), "err", err) + return + } + + online := status.IsOnline() + changed, err := l.opts.Reports.SetPresence(ctx, dev.ID, online, status.Timestamp(time.Now()), l.opts.Actor) + if err != nil { + l.log.Error("mqtt presence", "device", dev.Name, "err", err) + return + } + if changed { + l.log.Info("presence changed", "device", dev.Name, "topic", msg.Topic(), + "online", online, "retained", msg.Retained()) + } +} + +// handleTelemetryMessage stores a device usage report. +func (l *Link) handleTelemetryMessage(_ paho.Client, msg paho.Message) { + l.noteMessage() + + configTopic, ok := deviceFromTopic(msg.Topic(), SuffixTelemetry) + if !ok { + l.log.Warn("mqtt telemetry on an unexpected topic", "topic", msg.Topic()) + return + } + + report, err := ParseTelemetry(msg.Payload()) + if err != nil { + l.log.Warn("mqtt telemetry rejected", "topic", msg.Topic(), "err", err) + return + } + + rows := report.Rows() + if len(rows) == 0 { + l.log.Warn("mqtt telemetry had no usable rows", "topic", msg.Topic()) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + dev, err := l.opts.Devices.DeviceByConfigTopic(ctx, configTopic) + if err != nil { + l.log.Warn("mqtt telemetry for an unknown device", "topic", msg.Topic(), "err", err) + return + } + + day := report.DayOrDefault(time.Now()) + n, err := l.opts.Reports.InsertTelemetry(ctx, dev.ID, day, rows) + if err != nil { + l.log.Error("mqtt telemetry", "device", dev.Name, "err", err) + return + } + l.log.Info("telemetry stored", "device", dev.Name, "day", day, "rows", n) +} + +func (l *Link) noteMessage() { + l.received.Add(1) + l.lastMsg.Store(time.Now().UnixNano()) +} + +// PublishDeviceConfig publishes one device's desired state, retained, on the +// topic stored with the device. It is the single writer path for config +// (AGENTS.md §10 rule 5): only the Control Service publishes. +func (l *Link) PublishDeviceConfig(ctx context.Context, dev store.Device) error { + allow, block, err := l.opts.Source.DeviceLists(ctx, dev.ID) + if err != nil { + return err + } + payload, err := Config{ + V: ConfigVersion, + DeviceID: dev.ID, + Name: dev.Name, + Platform: dev.Platform, + Locked: dev.Locked, + Until: dev.Until, + Allow: allow, + Block: block, + GeneratedAt: time.Now().UTC(), + }.Marshal() + if err != nil { + return err + } + return l.Publish(dev.Topic, payload) +} + +// Publish sends one retained message and waits for the broker to accept it. +func (l *Link) Publish(topic string, payload []byte) error { + if !l.client.IsConnected() { + return errors.New("mqtt: not connected") + } + token := l.client.Publish(topic, 1, true, payload) + if !token.WaitTimeout(l.opts.ConnectTimeout) { + return fmt.Errorf("mqtt: publish to %s timed out", topic) + } + if err := token.Error(); err != nil { + return fmt.Errorf("mqtt: publish to %s: %w", topic, err) + } + l.published.Add(1) + return nil +} + +// PublishAll republishes every device's retained config. It runs after every +// connect, so a broker that lost its retained store is repaired from SQLite — +// the database stays the source of truth. +func (l *Link) PublishAll(ctx context.Context) (int, error) { + devices, err := l.opts.Source.ListDevices(ctx) + if err != nil { + return 0, err + } + published := 0 + var firstErr error + for _, dev := range devices { + if err := l.PublishDeviceConfig(ctx, dev); err != nil { + l.log.Error("mqtt publish config", "device", dev.Name, "err", err) + if firstErr == nil { + firstErr = err + } + continue + } + published++ + } + return published, firstErr +} + +// isTLSBroker reports whether the broker URL asks for TLS. +func isTLSBroker(broker string) bool { + return strings.HasPrefix(broker, "tls://") || strings.HasPrefix(broker, "ssl://") +} + +// tlsConfig verifies the broker against the lab CA. Every device and the +// Control Service use the same CA, so a rogue broker on the LAN cannot take +// over a device's rules. +func tlsConfig(broker, caFile string) (*tls.Config, error) { + if caFile == "" { + return nil, errors.New("mqtt: a TLS broker needs CONTROL_MQTT_CA_FILE") + } + pem, err := os.ReadFile(caFile) + if err != nil { + return nil, fmt.Errorf("mqtt: read CA %s: %w", caFile, err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + return nil, fmt.Errorf("mqtt: no certificates in %s", caFile) + } + + cfg := &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12} + if u, err := url.Parse(broker); err == nil && u.Hostname() != "" { + cfg.ServerName = u.Hostname() + } + return cfg, nil +} diff --git a/internal/mqtt/payload.go b/internal/mqtt/payload.go new file mode 100644 index 0000000..74374e2 --- /dev/null +++ b/internal/mqtt/payload.go @@ -0,0 +1,223 @@ +// Package mqtt is the Control Service's link to the broker (AGENTS.md §10 +// rules 5–6, §11 Phase 3; TRANSPORT_DESIGN.md). +// +// The broker is transport only. This package publishes each device's desired +// state, retained, on restrict//config, and consumes the reports that +// devices write back. SQLite stays the source of truth (SYSTEM_MAP.md): nothing +// here is authoritative, and a report is handed straight to the store. +// +// Topics (TRANSPORT_DESIGN.md "Topic design"): +// +// restrict//config service → device, retained +// restrict//status device → service, retained (presence) +// restrict//telemetry device → service, not retained +// +// "config" keeps the exact topic the device table already stores, so a device's +// identity never changes when it is renamed. +package mqtt + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store" +) + +// TopicPrefix is the first topic level for every message this service touches. +const TopicPrefix = "restrict" + +// Topic suffixes, appended to "restrict/". +const ( + SuffixConfig = "/config" + SuffixStatus = "/status" + SuffixTelemetry = "/telemetry" +) + +// StatusSubscription and TelemetrySubscription are the wildcard topics the +// Control Service reads. A device may write only its own status and telemetry +// (AGENTS.md §10 rule 6). +const ( + StatusSubscription = TopicPrefix + "/+/status" + TelemetrySubscription = TopicPrefix + "/+/telemetry" +) + +// StatusTopic returns the status topic for a device's config topic. +func StatusTopic(configTopic string) string { + return withSuffix(configTopic, SuffixStatus) +} + +// TelemetryTopic returns the telemetry topic for a device's config topic. +func TelemetryTopic(configTopic string) string { + return withSuffix(configTopic, SuffixTelemetry) +} + +func withSuffix(configTopic, suffix string) string { + return strings.TrimSuffix(configTopic, SuffixConfig) + suffix +} + +// deviceFromTopic turns "restrict/harry_phone_2/status" back into the config +// topic "restrict/harry_phone_2/config" so the store can find the device. +func deviceFromTopic(topic, suffix string) (string, bool) { + if !strings.HasPrefix(topic, TopicPrefix+"/") || !strings.HasSuffix(topic, suffix) { + return "", false + } + base := strings.TrimSuffix(topic, suffix) + if strings.Count(base, "/") != 1 || len(base) <= len(TopicPrefix)+1 { + return "", false + } + return base + SuffixConfig, true +} + +// Config is the desired state for one device. It is published retained on +// restrict//config: a device that reconnects is configured at once, +// with no poll and no waiting (TRANSPORT_DESIGN.md). +// +// The shape is versioned with "v" and is consumed by the kids' clients in +// Phases 4 and 5. Add fields; do not repurpose them. +type Config struct { + V int `json:"v"` + DeviceID int64 `json:"device_id"` + Name string `json:"name"` + Platform string `json:"platform"` + Locked bool `json:"locked"` + Until *time.Time `json:"until"` + Allow []string `json:"allow"` + Block []string `json:"block"` + GeneratedAt time.Time `json:"generated_at"` +} + +// ConfigVersion is the current schema version of Config. +const ConfigVersion = 1 + +// Marshal renders the payload. Lists are always present, never null, so a +// client can parse without null checks. +func (c Config) Marshal() ([]byte, error) { + if c.Allow == nil { + c.Allow = []string{} + } + if c.Block == nil { + c.Block = []string{} + } + b, err := json.Marshal(c) + if err != nil { + return nil, fmt.Errorf("marshal config: %w", err) + } + return b, nil +} + +// Status is a presence report from a device. +// +// The device publishes a retained {"online": true} when it comes up and +// registers the will {"online": false} on the same topic, so the broker +// publishes "offline" the moment the connection drops without a clean +// goodbye. That is the whole of presence detection (FR-6). +type Status struct { + Online *bool `json:"online"` + App string `json:"app"` + Locked *bool `json:"locked"` + At string `json:"at"` +} + +// ParseStatus reads a status payload. A report with no "online" field means the +// device is talking, so it counts as online. +func ParseStatus(payload []byte) (Status, error) { + var st Status + if err := json.Unmarshal(payload, &st); err != nil { + return Status{}, fmt.Errorf("parse status: %w", err) + } + if len(payload) == 0 { + return Status{}, fmt.Errorf("parse status: empty payload") + } + return st, nil +} + +// IsOnline reports the presence the payload asserts. +func (s Status) IsOnline() bool { return s.Online == nil || *s.Online } + +// At parses the report timestamp, falling back to the current time. +func (s Status) Timestamp(now time.Time) time.Time { + if s.At == "" { + return now + } + for _, layout := range []string{time.RFC3339, time.RFC3339Nano} { + if t, err := time.Parse(layout, s.At); err == nil { + return t + } + } + return now +} + +// UsageEntry is one app or domain total. +type UsageEntry struct { + Name string `json:"name"` + Seconds int `json:"seconds"` +} + +// Telemetry is a rolling daily usage report from a device. Devices report +// totals for the day, not deltas, so a repeated report replaces the row +// (migration 002). +// +// Two shapes are accepted: the list form +// +// {"day":"2026-10-13","apps":[{"name":"com.spotify.music","seconds":600}], +// "domains":[{"name":"youtube.com","seconds":120}]} +// +// and the flat single-row form +// +// {"app":"com.spotify.music","seconds":600} +type Telemetry struct { + Day string `json:"day"` + Apps []UsageEntry `json:"apps"` + Domains []UsageEntry `json:"domains"` + App string `json:"app"` + Domain string `json:"domain"` + Seconds int `json:"seconds"` +} + +// ParseTelemetry reads a telemetry payload. +func ParseTelemetry(payload []byte) (Telemetry, error) { + var t Telemetry + if len(payload) == 0 { + return Telemetry{}, fmt.Errorf("parse telemetry: empty payload") + } + if err := json.Unmarshal(payload, &t); err != nil { + return Telemetry{}, fmt.Errorf("parse telemetry: %w", err) + } + return t, nil +} + +// DayOrDefault returns the day the report belongs to, in the given location. A +// device may send its own local day; anything that is not YYYY-MM-DD is +// ignored and today's date is used instead. +func (t Telemetry) DayOrDefault(now time.Time) string { + if t.Day != "" { + if d, err := time.Parse("2006-01-02", t.Day); err == nil { + return d.Format("2006-01-02") + } + } + return now.Format("2006-01-02") +} + +// Rows flattens the payload into telemetry rows. The app and the domain are +// kept in separate columns; a row with neither is dropped. +func (t Telemetry) Rows() []store.TelemetryEntry { + out := make([]store.TelemetryEntry, 0, len(t.Apps)+len(t.Domains)+1) + for _, e := range t.Apps { + if e.Name == "" || e.Seconds < 0 { + continue + } + out = append(out, store.TelemetryEntry{App: e.Name, Seconds: e.Seconds}) + } + for _, e := range t.Domains { + if e.Name == "" || e.Seconds < 0 { + continue + } + out = append(out, store.TelemetryEntry{Domain: e.Name, Seconds: e.Seconds}) + } + if (t.App != "" || t.Domain != "") && t.Seconds >= 0 { + out = append(out, store.TelemetryEntry{App: t.App, Domain: t.Domain, Seconds: t.Seconds}) + } + return out +} diff --git a/internal/mqtt/payload_test.go b/internal/mqtt/payload_test.go new file mode 100644 index 0000000..e63715e --- /dev/null +++ b/internal/mqtt/payload_test.go @@ -0,0 +1,211 @@ +package mqtt + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store" +) + +func TestTopicHelpers(t *testing.T) { + const config = "restrict/harry_phone_2/config" + if got := StatusTopic(config); got != "restrict/harry_phone_2/status" { + t.Errorf("StatusTopic = %q", got) + } + if got := TelemetryTopic(config); got != "restrict/harry_phone_2/telemetry" { + t.Errorf("TelemetryTopic = %q", got) + } + + for _, tc := range []struct { + topic string + suffix string + want string + ok bool + }{ + {"restrict/harry_phone_2/status", SuffixStatus, config, true}, + {"restrict/harry_phone_2/telemetry", SuffixTelemetry, config, true}, + {"restrict/harry_phone_2/config", SuffixStatus, "", false}, + {"restrict/status", SuffixStatus, "", false}, + {"restrict/a/b/status", SuffixStatus, "", false}, + {"other/harry_phone_2/status", SuffixStatus, "", false}, + } { + got, ok := deviceFromTopic(tc.topic, tc.suffix) + if ok != tc.ok || got != tc.want { + t.Errorf("deviceFromTopic(%q, %q) = (%q, %v), want (%q, %v)", + tc.topic, tc.suffix, got, ok, tc.want, tc.ok) + } + } +} + +func TestConfigMarshal(t *testing.T) { + until := time.Date(2026, 10, 13, 21, 0, 0, 0, time.UTC) + b, err := Config{ + V: ConfigVersion, + DeviceID: 2, + Name: "harry_phone", + Platform: store.PlatformAndroid, + Locked: true, + Until: &until, + GeneratedAt: time.Date(2026, 10, 13, 3, 0, 0, 0, time.UTC), + }.Marshal() + if err != nil { + t.Fatalf("Marshal: %v", err) + } + + // The lists must be present as arrays, never null, so a client can parse + // this without null checks. + if !strings.Contains(string(b), `"allow":[]`) || !strings.Contains(string(b), `"block":[]`) { + t.Errorf("payload has null lists: %s", b) + } + + var back Config + if err := json.Unmarshal(b, &back); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if !back.Locked || back.Until == nil || !back.Until.Equal(until) { + t.Errorf("round trip lost the lock: %+v", back) + } + if back.V != ConfigVersion || back.DeviceID != 2 { + t.Errorf("round trip lost the identity: %+v", back) + } +} + +func TestParseStatus(t *testing.T) { + if _, err := ParseStatus(nil); err == nil { + t.Error("expected an error for an empty payload") + } + if _, err := ParseStatus([]byte("{oops")); err == nil { + t.Error("expected an error for malformed JSON") + } + + // An explicit offline (the will payload the broker publishes on a dropped + // connection) must read as offline. + off, err := ParseStatus([]byte(`{"online":false}`)) + if err != nil { + t.Fatalf("ParseStatus: %v", err) + } + if off.IsOnline() { + t.Error(`{"online":false} read as online`) + } + + // A report with no "online" field means the device is talking. + on, err := ParseStatus([]byte(`{"app":"com.spotify.music"}`)) + if err != nil { + t.Fatalf("ParseStatus: %v", err) + } + if !on.IsOnline() { + t.Error("a report with no online field read as offline") + } + if on.App != "com.spotify.music" { + t.Errorf("App = %q", on.App) + } + + at := on.Timestamp(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)) + if !at.Equal(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)) { + t.Errorf("Timestamp with no at field = %v, want the fallback", at) + } + at = Status{At: "2026-10-13T07:30:00Z"}.Timestamp(time.Now()) + if at.UTC().Format(time.RFC3339) != "2026-10-13T07:30:00Z" { + t.Errorf("Timestamp = %v", at) + } +} + +func TestTelemetryRows(t *testing.T) { + payload := []byte(`{ + "day": "2026-10-13", + "apps": [{"name": "com.spotify.music", "seconds": 600}, {"name": "", "seconds": 5}], + "domains": [{"name": "youtube.com", "seconds": 120}] + }`) + report, err := ParseTelemetry(payload) + if err != nil { + t.Fatalf("ParseTelemetry: %v", err) + } + rows := report.Rows() + if len(rows) != 2 { + t.Fatalf("Rows = %+v, want 2 rows", rows) + } + // Apps first, then domains. + if rows[0].App != "com.spotify.music" || rows[0].Seconds != 600 { + t.Errorf("rows[0] = %+v", rows[0]) + } + if rows[1].Domain != "youtube.com" || rows[1].Seconds != 120 { + t.Errorf("rows[1] = %+v", rows[1]) + } + if got := report.DayOrDefault(time.Now()); got != "2026-10-13" { + t.Errorf("DayOrDefault = %q, want the reported day", got) + } + + // The flat single-row form, and a bad day falling back to today. + flat, err := ParseTelemetry([]byte(`{"app":"roblox","seconds":60,"day":"yesterday"}`)) + if err != nil { + t.Fatalf("ParseTelemetry flat: %v", err) + } + flatRows := flat.Rows() + if len(flatRows) != 1 || flatRows[0].App != "roblox" { + t.Fatalf("flat Rows = %+v", flatRows) + } + now := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC) + if got := flat.DayOrDefault(now); got != "2026-03-04" { + t.Errorf("DayOrDefault with a bad day = %q, want 2026-03-04", got) + } + + // Negative totals are dropped, not stored. + bad, err := ParseTelemetry([]byte(`{"apps":[{"name":"x","seconds":-1}]}`)) + if err != nil { + t.Fatalf("ParseTelemetry: %v", err) + } + if len(bad.Rows()) != 0 { + t.Errorf("negative usage stored: %+v", bad.Rows()) + } + + if _, err := ParseTelemetry([]byte("not json")); err == nil { + t.Error("expected an error for malformed JSON") + } +} + +func TestNewRejectsIncompleteOptions(t *testing.T) { + if _, err := New(Options{Broker: "tcp://127.0.0.1:1883"}); err == nil { + t.Error("expected an error without a device lookup, source and sink") + } + _, err := New(Options{ + Broker: "tls://broker:8883", + CAFile: "/nonexistent/ca.crt", + Devices: stubLookup{}, + Source: stubSource{}, + Reports: stubSink{}, + }) + if err == nil { + t.Fatal("expected an error for a missing CA bundle") + } + if !strings.Contains(err.Error(), "ca.crt") { + t.Errorf("error = %v, want it to name the CA file", err) + } +} + +type stubLookup struct{} + +func (stubLookup) DeviceByConfigTopic(ctx context.Context, topic string) (*store.Device, error) { + return nil, errors.New("not implemented") +} + +type stubSource struct{} + +func (stubSource) ListDevices(ctx context.Context) ([]store.Device, error) { return nil, nil } + +func (stubSource) DeviceLists(ctx context.Context, id int64) ([]string, []string, error) { + return nil, nil, nil +} + +type stubSink struct{} + +func (stubSink) SetPresence(ctx context.Context, id int64, online bool, at time.Time, actor string) (bool, error) { + return false, nil +} + +func (stubSink) InsertTelemetry(ctx context.Context, id int64, day string, e []store.TelemetryEntry) (int, error) { + return 0, nil +} diff --git a/internal/store/lists.go b/internal/store/lists.go new file mode 100644 index 0000000..741df9c --- /dev/null +++ b/internal/store/lists.go @@ -0,0 +1,74 @@ +// Per-device allow and block lists, and device lookup by MQTT topic +// (AGENTS.md §7 blocklist/allowlist, §11 Phase 3). +// +// The editors for these lists arrive in Phase 6. Phase 3 already publishes them +// in the device config, so a device that is onboarded later gets the real lists +// without another protocol change. Migration 001 created both tables. +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" +) + +// DeviceLists returns the site/app values that stay allowed and the values that +// are blocked for one device. A row with a NULL device_id applies to every +// device (§7). Rows are returned in insertion order, de-duplicated. +func (s *Store) DeviceLists(ctx context.Context, id int64) (allow, block []string, err error) { + if allow, err = s.listValues(ctx, "allowlist", id); err != nil { + return nil, nil, err + } + if block, err = s.listValues(ctx, "blocklist", id); err != nil { + return nil, nil, err + } + return allow, block, nil +} + +func (s *Store) listValues(ctx context.Context, table string, id int64) ([]string, error) { + // The table name is a package constant, never caller input. + rows, err := s.db.QueryContext(ctx, + `SELECT value FROM `+table+` WHERE device_id = ? OR device_id IS NULL ORDER BY id`, id) + if err != nil { + return nil, fmt.Errorf("list %s: %w", table, err) + } + defer rows.Close() + + out := make([]string, 0, 4) + seen := make(map[string]struct{}, 4) + for rows.Next() { + var v string + if err := rows.Scan(&v); err != nil { + return nil, fmt.Errorf("scan %s: %w", table, err) + } + v = strings.TrimSpace(v) + if v == "" { + continue + } + if _, dup := seen[v]; dup { + continue + } + seen[v] = struct{}{} + out = append(out, v) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list %s: %w", table, err) + } + return out, nil +} + +// DeviceByConfigTopic finds a device from a topic the device wrote to, for +// example "restrict/harry_phone_2/status". Only the Control Service reads the +// topic table, so a topic the service does not know is ignored. +func (s *Store) DeviceByConfigTopic(ctx context.Context, topic string) (*Device, error) { + dev, err := scanDevice(s.db.QueryRowContext(ctx, selectDevice+` WHERE d.topic = ?`, topic)) + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("device by topic: %w", err) + } + return dev, nil +} diff --git a/internal/store/migrations/002_telemetry_unique.sql b/internal/store/migrations/002_telemetry_unique.sql new file mode 100644 index 0000000..45aba5a --- /dev/null +++ b/internal/store/migrations/002_telemetry_unique.sql @@ -0,0 +1,13 @@ +-- Migration 002 — Phase 3. Telemetry reports arrive over MQTT and repeat. +-- +-- A device reports a rolling daily total ("Spotify: 600 s today"), not a delta, +-- so one row per (device, day, app, domain) is the correct shape: a repeat +-- report replaces the previous total instead of adding to it. Devices may omit +-- the app or the domain, so both default to '' — SQLite treats NULLs in a +-- UNIQUE index as distinct, which would defeat the upsert. + +UPDATE telemetry SET app = '' WHERE app IS NULL; +UPDATE telemetry SET domain = '' WHERE domain IS NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_telemetry_daily + ON telemetry(device_id, day, app, domain); diff --git a/internal/store/reports.go b/internal/store/reports.go new file mode 100644 index 0000000..64fed1e --- /dev/null +++ b/internal/store/reports.go @@ -0,0 +1,131 @@ +// Reports that arrive from devices over MQTT: presence and telemetry +// (AGENTS.md §7, §11 Phase 3). +// +// The broker is transport only. Every report is written to SQLite here, because +// SQLite is the source of truth (SYSTEM_MAP.md). A malformed or hostile report +// must never take the service down: callers validate first, and a failed insert +// is logged and dropped. +package store + +import ( + "context" + "fmt" + "time" +) + +// SetPresence records that a device was seen and whether it is online now. It +// returns whether the online flag changed. +// +// Presence comes from the broker's Last Will and Testament: a device publishes +// a retained "online" status, and the broker publishes its retained will +// ("online": false) when the connection drops without a clean disconnect. +// +// last_seen is refreshed on every report. The audit log only gets an entry when +// the online flag changes, so a chatty device does not flood the log (FR-15). +func (s *Store) SetPresence(ctx context.Context, id int64, online bool, at time.Time, actor string) (bool, error) { + if at.IsZero() { + at = time.Now() + } + at = at.UTC() + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return false, fmt.Errorf("set presence: %w", err) + } + defer func() { _ = tx.Rollback() }() + + var current int + if err := tx.QueryRowContext(ctx, `SELECT online FROM devices WHERE id = ?`, id).Scan(¤t); err != nil { + return false, fmt.Errorf("set presence: %w", err) + } + + if _, err := tx.ExecContext(ctx, + `UPDATE devices SET online = ?, last_seen = ?, updated_at = datetime('now') WHERE id = ?`, + boolToInt(online), at.Format(time.RFC3339), id); err != nil { + return false, fmt.Errorf("set presence: %w", err) + } + + changed := (current != 0) != online + if changed { + action, detail := "device.offline", "last will received" + if online { + action, detail = "device.online", "status report" + } + if err := insertAudit(ctx, tx, actor, action, &id, detail); err != nil { + return false, err + } + } + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("commit presence: %w", err) + } + return changed, nil +} + +// TelemetryEntry is one app or domain row of a usage report. +type TelemetryEntry struct { + App string + Domain string + Seconds int +} + +// InsertTelemetry upserts a rolling daily usage total per (device, day, app, +// domain). Reports are repeats, not deltas: a later report for the same key +// replaces the earlier total (migration 002). It returns the number of rows +// written. +func (s *Store) InsertTelemetry(ctx context.Context, id int64, day string, entries []TelemetryEntry) (int, error) { + if len(entries) == 0 { + return 0, nil + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("insert telemetry: %w", err) + } + defer func() { _ = tx.Rollback() }() + + n := 0 + for _, e := range entries { + if e.Seconds < 0 { + continue + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO telemetry (device_id, app, domain, seconds, day) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(device_id, day, app, domain) DO UPDATE SET + seconds = excluded.seconds, + created_at = datetime('now')`, + id, e.App, e.Domain, e.Seconds, day); err != nil { + return n, fmt.Errorf("insert telemetry: %w", err) + } + n++ + } + if err := tx.Commit(); err != nil { + return n, fmt.Errorf("commit telemetry: %w", err) + } + return n, nil +} + +// TelemetryForDay returns one device's usage rows for a day, biggest first. +func (s *Store) TelemetryForDay(ctx context.Context, id int64, day string) ([]TelemetryEntry, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT app, domain, seconds FROM telemetry + WHERE device_id = ? AND day = ? + ORDER BY seconds DESC`, id, day) + if err != nil { + return nil, fmt.Errorf("telemetry for day: %w", err) + } + defer rows.Close() + + out := make([]TelemetryEntry, 0, 8) + for rows.Next() { + var e TelemetryEntry + if err := rows.Scan(&e.App, &e.Domain, &e.Seconds); err != nil { + return nil, fmt.Errorf("scan telemetry: %w", err) + } + out = append(out, e) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("telemetry for day: %w", err) + } + return out, nil +} diff --git a/internal/store/reports_test.go b/internal/store/reports_test.go new file mode 100644 index 0000000..55efd6a --- /dev/null +++ b/internal/store/reports_test.go @@ -0,0 +1,187 @@ +package store + +import ( + "context" + "testing" + "time" +) + +func newTestDevice(t *testing.T) (*Store, *Device) { + t.Helper() + st := testStore(t) + dev, err := st.CreateDevice(t.Context(), "Harry's phone", "Harry", PlatformAndroid, "sam") + if err != nil { + t.Fatalf("CreateDevice: %v", err) + } + return st, dev +} + +func TestSetPresenceAuditsOnlyTransitions(t *testing.T) { + st, dev := newTestDevice(t) + ctx := context.Background() + + // The first "online" flips the flag, so it is audited. + changed, err := st.SetPresence(ctx, dev.ID, true, time.Now(), "device") + if err != nil { + t.Fatalf("SetPresence: %v", err) + } + if !changed { + t.Error("changed = false on the first online report, want true") + } + + // A repeat report only refreshes last_seen. + changed, err = st.SetPresence(ctx, dev.ID, true, time.Now().Add(time.Minute), "device") + if err != nil { + t.Fatalf("SetPresence: %v", err) + } + if changed { + t.Error("changed = true on a repeat report, want false") + } + + // The will payload flips it back. + changed, err = st.SetPresence(ctx, dev.ID, false, time.Now(), "device") + if err != nil { + t.Fatalf("SetPresence: %v", err) + } + if !changed { + t.Error("changed = false when the device went offline, want true") + } + + got, err := st.GetDevice(ctx, dev.ID) + if err != nil { + t.Fatalf("GetDevice: %v", err) + } + if got.Online { + t.Error("Online = true after the will, want false") + } + if got.LastSeen == nil { + t.Fatal("LastSeen is nil, want the report time") + } + + audit, err := st.ListAudit(ctx, 50) + if err != nil { + t.Fatalf("ListAudit: %v", err) + } + var online, offline int + for _, e := range audit { + switch e.Action { + case "device.online": + online++ + case "device.offline": + offline++ + } + } + if online != 1 || offline != 1 { + t.Errorf("audit has %d online and %d offline entries, want 1 and 1", online, offline) + } +} + +func TestInsertTelemetryReplacesTheDailyTotal(t *testing.T) { + st, dev := newTestDevice(t) + ctx := context.Background() + const day = "2026-10-13" + + n, err := st.InsertTelemetry(ctx, dev.ID, day, []TelemetryEntry{ + {App: "com.spotify.music", Seconds: 600}, + {Domain: "youtube.com", Seconds: 120}, + }) + if err != nil { + t.Fatalf("InsertTelemetry: %v", err) + } + if n != 2 { + t.Errorf("rows = %d, want 2", n) + } + + // Devices report rolling totals, so a repeat report replaces the row. + if _, err := st.InsertTelemetry(ctx, dev.ID, day, []TelemetryEntry{ + {App: "com.spotify.music", Seconds: 900}, + }); err != nil { + t.Fatalf("InsertTelemetry repeat: %v", err) + } + + rows, err := st.TelemetryForDay(ctx, dev.ID, day) + if err != nil { + t.Fatalf("TelemetryForDay: %v", err) + } + if len(rows) != 2 { + t.Fatalf("rows = %+v, want 2 (a replace, not an insert)", rows) + } + if rows[0].App != "com.spotify.music" || rows[0].Seconds != 900 { + t.Errorf("rows[0] = %+v, want the replaced total 900", rows[0]) + } + if rows[1].Seconds != 120 { + t.Errorf("rows[1] = %+v", rows[1]) + } + + // Another day is a different row. + if _, err := st.InsertTelemetry(ctx, dev.ID, "2026-10-14", []TelemetryEntry{ + {App: "com.spotify.music", Seconds: 60}, + }); err != nil { + t.Fatalf("InsertTelemetry next day: %v", err) + } + rows, err = st.TelemetryForDay(ctx, dev.ID, day) + if err != nil { + t.Fatalf("TelemetryForDay: %v", err) + } + if len(rows) != 2 || rows[0].Seconds != 900 { + t.Errorf("rows for %s = %+v, want the day isolated", day, rows) + } + + // A negative total is dropped, and an empty report is a no-op. + if n, err := st.InsertTelemetry(ctx, dev.ID, day, []TelemetryEntry{{App: "x", Seconds: -5}}); err != nil || n != 0 { + t.Errorf("negative telemetry = (%d, %v), want (0, nil)", n, err) + } + if n, err := st.InsertTelemetry(ctx, dev.ID, day, nil); err != nil || n != 0 { + t.Errorf("empty telemetry = (%d, %v), want (0, nil)", n, err) + } +} + +func TestDeviceListsAndTopicLookup(t *testing.T) { + st, dev := newTestDevice(t) + ctx := context.Background() + + // The Phase 6 editors do not exist yet, so seed the tables directly. + seed := []struct { + table string + kind string + value string + }{ + {"allowlist", "app", "com.spotify.music"}, + {"allowlist", "site", "school.example.edu"}, + {"blocklist", "app", "com.roblox.client"}, + } + for _, s := range seed { + if _, err := st.DB().ExecContext(ctx, + `INSERT INTO `+s.table+` (device_id, kind, value) VALUES (?, ?, ?)`, + dev.ID, s.kind, s.value); err != nil { + t.Fatalf("seed %s: %v", s.table, err) + } + } + // A NULL device_id applies to every device, and duplicates collapse. + if _, err := st.DB().ExecContext(ctx, + `INSERT INTO allowlist (device_id, kind, value) VALUES (NULL, 'app', 'com.spotify.music')`); err != nil { + t.Fatalf("seed global allow: %v", err) + } + + allow, block, err := st.DeviceLists(ctx, dev.ID) + if err != nil { + t.Fatalf("DeviceLists: %v", err) + } + if len(allow) != 2 || allow[0] != "com.spotify.music" || allow[1] != "school.example.edu" { + t.Errorf("allow = %v", allow) + } + if len(block) != 1 || block[0] != "com.roblox.client" { + t.Errorf("block = %v", block) + } + + found, err := st.DeviceByConfigTopic(ctx, dev.Topic) + if err != nil { + t.Fatalf("DeviceByConfigTopic: %v", err) + } + if found.ID != dev.ID { + t.Errorf("DeviceByConfigTopic = %d, want %d", found.ID, dev.ID) + } + if _, err := st.DeviceByConfigTopic(ctx, "restrict/nope_99/config"); err != ErrNotFound { + t.Errorf("unknown topic error = %v, want ErrNotFound", err) + } +} diff --git a/internal/web/devices.go b/internal/web/devices.go index e057c1d..f56c83d 100644 --- a/internal/web/devices.go +++ b/internal/web/devices.go @@ -140,6 +140,10 @@ func auditLabel(action string) string { return "Unlocked" case "state.expired": return "Lock expired" + case "device.online": + return "Device online" + case "device.offline": + return "Device offline" case "login.success": return "Signed in" case "login.failed": @@ -221,6 +225,7 @@ func (s *Server) handleDeviceCreate(w http.ResponseWriter, r *http.Request) { return } s.log.Info("device created", "id", dev.ID, "name", dev.Name, "topic", dev.Topic) + s.publishConfig(ctx, *dev) http.Redirect(w, r, "/?ok="+url.QueryEscape("Added "+dev.Name), http.StatusSeeOther) } @@ -294,6 +299,7 @@ func (s *Server) handleDeviceUpdate(w http.ResponseWriter, r *http.Request) { return } s.log.Info("device updated", "id", dev.ID, "name", dev.Name) + s.publishConfig(ctx, *dev) http.Redirect(w, r, "/?ok="+url.QueryEscape("Saved "+dev.Name), http.StatusSeeOther) } @@ -329,6 +335,7 @@ func (s *Server) handleDeviceLock(w http.ResponseWriter, r *http.Request) { return } s.log.Info("device locked", "id", dev.ID, "name", dev.Name, "minutes", minutes) + s.publishConfig(ctx, *dev) msg := "Locked " + dev.Name if minutes > 0 { msg = fmt.Sprintf("Locked %s for %d min", dev.Name, minutes) @@ -358,6 +365,7 @@ func (s *Server) handleDeviceUnlock(w http.ResponseWriter, r *http.Request) { return } s.log.Info("device unlocked", "id", dev.ID, "name", dev.Name) + s.publishConfig(ctx, *dev) http.Redirect(w, r, "/?ok="+url.QueryEscape("Unlocked "+dev.Name), http.StatusSeeOther) } diff --git a/internal/web/devices_test.go b/internal/web/devices_test.go index 13249ad..d867304 100644 --- a/internal/web/devices_test.go +++ b/internal/web/devices_test.go @@ -44,7 +44,7 @@ func newTestApp(t *testing.T) *testApp { AppEnv: "test", DBPath: ":memory:", } - srv, err := New(cfg, st, am, slog.New(slog.NewTextHandler(io.Discard, nil))) + srv, err := New(cfg, st, am, slog.New(slog.NewTextHandler(io.Discard, nil)), nil) if err != nil { t.Fatalf("web.New: %v", err) } diff --git a/internal/web/server.go b/internal/web/server.go index a30dcfc..88e15e5 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -32,10 +32,22 @@ type Server struct { auth *auth.Manager tmpl *template.Template log *slog.Logger + pub Publisher } -// New builds a Server and parses the embedded templates. -func New(cfg *config.Config, st *store.Store, am *auth.Manager, log *slog.Logger) (*Server, error) { +// Publisher pushes device config to the broker and reports the link's health. +// Only the Control Service publishes config (AGENTS.md §10 rule 5). It is nil +// when MQTT is switched off, and every call site tolerates that. +type Publisher interface { + Broker() string + Connected() bool + Stats() (received, published int64, lastMessageAt time.Time, lastErr string) + PublishDeviceConfig(ctx context.Context, dev store.Device) error + PublishAll(ctx context.Context) (int, error) +} + +// New builds a Server and parses the embedded templates. pub may be nil. +func New(cfg *config.Config, st *store.Store, am *auth.Manager, log *slog.Logger, pub Publisher) (*Server, error) { funcs := template.FuncMap{ "year": func() int { return time.Now().Year() }, "now": func() time.Time { return time.Now() }, @@ -48,7 +60,24 @@ func New(cfg *config.Config, st *store.Store, am *auth.Manager, log *slog.Logger if err != nil { return nil, fmt.Errorf("parse templates: %w", err) } - return &Server{cfg: cfg, store: st, auth: am, tmpl: tmpl, log: log}, nil + return &Server{cfg: cfg, store: st, auth: am, tmpl: tmpl, log: log, pub: pub}, nil +} + +// publishConfig pushes one device's desired state to the broker, retained. A +// publish failure never fails the HTTP request: SQLite already holds the new +// state, and the link republishes everything when it reconnects, so the broker +// catches up on its own. +func (s *Server) publishConfig(ctx context.Context, dev store.Device) { + if s.pub == nil { + s.log.Debug("mqtt disabled; config kept in SQLite only", "device", dev.Name) + return + } + if err := s.pub.PublishDeviceConfig(ctx, dev); err != nil { + s.log.Error("mqtt publish config", "device", dev.Name, "topic", dev.Topic, "err", err) + return + } + s.log.Info("mqtt config published", "device", dev.Name, "topic", dev.Topic, + "locked", dev.Locked) } // Handler returns the fully wrapped HTTP handler. @@ -234,6 +263,15 @@ type statusData struct { DBPath string ServerTime string ServerUp string + + // MQTT link health (Phase 3). + MQTTEnabled bool + MQTTConnected bool + MQTTBroker string + MQTTReceived int64 + MQTTPublished int64 + MQTTLastMessage string + MQTTLastError string } type dashboardData struct { @@ -259,6 +297,12 @@ func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { s.log.Error("expire locks", "err", err) } else if n > 0 { s.log.Info("expired timed locks", "count", n) + // The lock lifted, so the retained config is stale. Republish. + if s.pub != nil { + if _, err := s.pub.PublishAll(ctx); err != nil { + s.log.Error("mqtt republish after expiry", "err", err) + } + } } devs, err := s.store.ListDevices(ctx) if err != nil { @@ -290,12 +334,25 @@ func (s *Server) handleStatusPartial(w http.ResponseWriter, r *http.Request) { } func (s *Server) buildStatusData(ctx context.Context) statusData { - return statusData{ + data := statusData{ DBHealthy: s.store.Ping(ctx) == nil, DBPath: s.cfg.DBPath, ServerTime: time.Now().Format("Mon 02 Jan 2006 15:04:05"), ServerUp: time.Since(serverStarted).Round(time.Second).String(), } + if s.pub != nil { + data.MQTTEnabled = true + data.MQTTConnected = s.pub.Connected() + data.MQTTBroker = s.pub.Broker() + rx, tx, last, lastErr := s.pub.Stats() + data.MQTTReceived = rx + data.MQTTPublished = tx + data.MQTTLastError = lastErr + if !last.IsZero() { + data.MQTTLastMessage = last.Local().Format("15:04:05") + } + } + return data } func (s *Server) render(w http.ResponseWriter, _ *http.Request, name string, data any) { diff --git a/internal/web/templates/partials/status.html b/internal/web/templates/partials/status.html index 11d1a05..108c8a3 100644 --- a/internal/web/templates/partials/status.html +++ b/internal/web/templates/partials/status.html @@ -18,6 +18,22 @@ {{if .DBHealthy}}ok{{else}}error{{end}} +
+ +

Broker

+

+ {{if not .MQTTEnabled}}MQTT switched off{{else if .MQTTConnected}}{{.MQTTBroker}}{{else}}{{.MQTTBroker}} · reconnecting{{end}} +

+ + + {{if not .MQTTEnabled}}disabled{{else if .MQTTConnected}}connected{{else}}offline{{end}} + + {{if .MQTTEnabled}} +

{{.MQTTPublished}} published · {{.MQTTReceived}} received{{if .MQTTLastMessage}} · last report {{.MQTTLastMessage}}{{end}}

+ {{if .MQTTLastError}}

last error: {{.MQTTLastError}}

{{end}} + {{end}} +
+

Uptime