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:
414
internal/mqtt/link.go
Normal file
414
internal/mqtt/link.go
Normal 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
|
||||
}
|
||||
Reference in New Issue
Block a user