Phase 3: MQTT transport — retained config, status/telemetry, LWT presence, TLS/ACL design

- internal/mqtt: publish each device's desired state retained on
  restrict/<device>/config; subscribe to restrict/+/status and
  restrict/+/telemetry and write reports to SQLite; publish on every state
  change, republish on connect and every 15 minutes from SQLite, so the broker
  is repaired from the source of truth.
- store: presence (devices.online and last_seen, audited only on transitions),
  telemetry upsert per (device, day, app, domain) via migration 002, allow and
  block lists read into the config payload, device lookup by topic.
- config: MQTT settings. TLS plus per-device credentials are required; the
  pre-TLS shared broker needs an explicit CONTROL_MQTT_INSECURE escape hatch.
- web: republish config after lock, unlock, expiry, create and edit; broker card
  on the status partial.
- deploy/mosquitto: TLS listener, topic ACL, credential provisioning and a
  verifier that proves the policy on a throwaway broker. Not applied to the
  shared Mosquitto on .13 — that change needs Sam's approval.
- cmd/control: mqtt-clients prints the MQTT identity of every device.
- paho.mqtt.golang v1.5.1, which pulls golang.org/x/net and with it the
  golang.org/x/crypto bump to v0.42.0.

Verified on .13: retained config delivered to a fresh subscriber; a client
killed without DISCONNECT drove the broker's retained will and the service
flipped the device offline; telemetry landed in SQLite; make mqtt-verify passed
13/13; the service connected over TLS with credentials and a CA.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
2026-10-13 14:37:43 +11:00
parent 7c88d3f52b
commit 14b304e0b0
31 changed files with 2624 additions and 47 deletions

View File

@@ -1,5 +1,7 @@
.git
.gitignore
.gocache
.gobuild
.pi
bin/
*.md

9
.gitignore vendored
View File

@@ -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

View File

@@ -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

126
README.md
View File

@@ -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/<device>/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 <http://127.0.0.1:8791/> 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/<device>/config` | service → device | yes | desired state |
| `restrict/<device>/status` | device → service | yes | presence (`online`) |
| `restrict/<device>/telemetry` | device → service | no | usage totals for a day |
- `<device>` is `restrict/<slug>_<id>` from `devices.topic`, which is stored when
the device is created and never changes. `<id>` 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).

View File

@@ -46,6 +46,50 @@ ground.
| `restrict/<device>/cmd` | service → device | no | one-shot command |
| `restrict/<device>/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.
`<device>` is `restrict/<slug>_<id>`, stored in `devices.topic` when the device
is created, and never changes. `<id>` is also the device's MQTT username.
### Payloads, as implemented (Phase 3)
`restrict/<device>/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/<device>/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/<device>/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

View File

@@ -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:
//
// <token>\t<config topic>\t<platform>\t<child>\t<name>
//
// 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) {

149
deploy/mosquitto/README.md Normal file
View File

@@ -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=<from clients.env>
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.<timestamp> 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.

View File

@@ -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"

View File

@@ -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

View File

@@ -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

113
deploy/mosquitto/generate-tls.sh Executable file
View File

@@ -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" <<EOF
basicConstraints=critical,CA:FALSE
keyUsage=critical,digitalSignature,keyEncipherment
extendedKeyUsage=serverAuth
subjectAltName=$sans
EOF
trap 'rm -f "$tls_dir/$ext"' EXIT
openssl_run req -newkey rsa:2048 -sha256 -nodes \
-keyout server.key -out server.csr \
-subj "/O=Family Home Lab/CN=$hostname"
openssl_run x509 -req -in server.csr \
-CA ca.crt -CAkey ca.key -CAcreateserial \
-out server.crt -days "$days" -sha256 -extfile "$ext"
rm -f server.csr ca.srl "$ext"
chmod 644 ca.crt ca.key server.crt server.key
cd "$work_dir"
echo "generate-tls: wrote $tls_dir/{ca.crt,ca.key,server.crt,server.key}"
echo "generate-tls: ship ca.crt to every device; keep ca.key and server.key off-device."

View File

@@ -0,0 +1,100 @@
#!/usr/bin/env bash
#
# Create one broker account per registered device, plus the Control Service's
# own account (AGENTS.md §11 Phase 3, §10 rule 6).
#
# The MQTT username of a device is its topic token — the middle level of its
# config topic. The device list comes from the Control Service itself:
#
# docker exec webrestrict-control /app/control mqtt-clients
#
# Because conf.d/webrestrict.acl is written with `%u`, no per-device ACL edit is
# needed: adding the account is the whole onboarding step.
#
# Output (both gitignored, mode 0600 for the secrets):
#
# conf.d/webrestrict.passwd hashed accounts, read by the broker
# clients.env the plaintext values, for the devices and .env
#
# Nothing secret is printed to the terminal. Read clients.env to collect them.
set -euo pipefail
here="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
passwd_file="$here/conf.d/webrestrict.passwd"
secrets_file="$here/clients.env"
container="${CONTROL_CONTAINER:-webrestrict-control}"
broker_image="${MQTT_IMAGE:-eclipse-mosquitto:latest}"
service_user="${CONTROL_MQTT_USERNAME:-control-service}"
service_pass="${CONTROL_MQTT_PASSWORD:-}"
die() {
echo "provision-clients: $*" >&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 "# token<TAB>password<TAB>config 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."

View File

@@ -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.

View File

@@ -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

View File

@@ -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

View File

@@ -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

8
go.mod
View File

@@ -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

16
go.sum
View File

@@ -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=

View File

@@ -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

View File

@@ -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")
}
}

414
internal/mqtt/link.go Normal file
View File

@@ -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
}

223
internal/mqtt/payload.go Normal file
View File

@@ -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/<device>/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/<device>/config service → device, retained
// restrict/<device>/status device → service, retained (presence)
// restrict/<device>/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/<device>".
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/<device>/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
}

View File

@@ -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
}

74
internal/store/lists.go Normal file
View File

@@ -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
}

View File

@@ -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);

131
internal/store/reports.go Normal file
View File

@@ -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(&current); 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
}

View File

@@ -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)
}
}

View File

@@ -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)
}

View File

@@ -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)
}

View File

@@ -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) {

View File

@@ -18,6 +18,22 @@
<span class="status-pill status-pill--muted">{{if .DBHealthy}}ok{{else}}error{{end}}</span>
</article>
<article class="tool-card">
<div class="tool-card__icon tool-card__icon--purple" aria-hidden="true">⇄</div>
<h3 class="tool-card__title">Broker</h3>
<p class="tool-card__desc">
{{if not .MQTTEnabled}}MQTT switched off{{else if .MQTTConnected}}{{.MQTTBroker}}{{else}}{{.MQTTBroker}} · reconnecting{{end}}
</p>
<span class="status-pill {{if not .MQTTEnabled}}status-pill--muted{{else if .MQTTConnected}}status-pill--on{{else}}status-pill--off{{end}}">
<span class="dot {{if .MQTTConnected}}dot--on{{else}}dot--off{{end}}"></span>
{{if not .MQTTEnabled}}disabled{{else if .MQTTConnected}}connected{{else}}offline{{end}}
</span>
{{if .MQTTEnabled}}
<p class="tool-card__desc">{{.MQTTPublished}} published · {{.MQTTReceived}} received{{if .MQTTLastMessage}} · last report {{.MQTTLastMessage}}{{end}}</p>
{{if .MQTTLastError}}<p class="tool-card__desc">last error: {{.MQTTLastError}}</p>{{end}}
{{end}}
</article>
<article class="tool-card">
<div class="tool-card__icon tool-card__icon--purple" aria-hidden="true">⏱</div>
<h3 class="tool-card__title">Uptime</h3>