// 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 3: publishes each device's desired state to the broker, // retained, and consumes device presence and telemetry reports from it. package main import ( "context" "errors" "fmt" "io" "log/slog" "net/http" "os" "os/signal" "strings" "syscall" "time" "golang.org/x/crypto/bcrypt" "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" ) // 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 // republishPoll is how often the loop re-evaluates the next schedule boundary. // It bounds how late a transition can be after the schedule is edited, and it // keeps the service from sleeping past a boundary that did not exist yet when // the loop last looked. const republishPoll = 30 * time.Second 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 } } if err := run(); err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } } func run() error { started := time.Now() cfg, err := config.Load() if err != nil { return err } cfg.Version = version log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.LogLevel})) slog.SetDefault(log) st, err := store.Open(cfg.DBPath) if err != nil { return err } defer st.Close() if err := st.Migrate(context.Background()); err != nil { return err } am, err := auth.New(cfg.SessionSecret, cfg.SessionTTL, cfg.CookieSecure, cfg.AdminUsername, cfg.AdminPasswordHash, cfg.AdminPassword) if err != nil { return err } if !am.Hashed() { log.Warn("admin password is plaintext; set CONTROL_ADMIN_PASSWORD_HASH to use bcrypt") } 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", Location: cfg.ScheduleLocation, Log: log, Devices: st, Source: st, Reports: st, OnConnected: func() { republish(ctx, log, link, st) }, }) if err != nil { return err } pub = link link.Start(ctx) defer link.Stop() go republishLoop(ctx, log, link, st, cfg.ScheduleLocation) 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 } httpServer := &http.Server{ Addr: cfg.Addr, Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 30 * time.Second, WriteTimeout: 30 * time.Second, IdleTimeout: 60 * time.Second, } errCh := make(chan error, 1) go func() { log.Info("control service listening", "addr", cfg.Addr, "db", cfg.DBPath, "env", cfg.AppEnv, "version", version) if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { errCh <- err } }() stop := make(chan os.Signal, 1) signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM) select { case err := <-errCh: return err case sig := <-stop: log.Info("shutting down", "signal", sig.String()) } cancel() 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 // whenever the desired state may have changed, so the broker's view is rebuilt // from SQLite, the source of truth. It first clears any manual lock that has // expired, so the published state is never stale. func republish(ctx context.Context, log *slog.Logger, link *mqtt.Link, st *store.Store) { if link == nil { return } ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() if n, err := st.ExpireLocks(ctx); err != nil { log.Error("expire locks", "err", err) } else if n > 0 { log.Info("expired timed locks", "count", n) } 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) } // republishLoop refreshes the retained config when a schedule boundary is // crossed and otherwise on a long interval. It re-evaluates often instead of // sleeping until one computed boundary, so a schedule edited after the loop // started still turns on and off on time (FR-2) without a poll from the device. func republishLoop(ctx context.Context, log *slog.Logger, link *mqtt.Link, st *store.Store, loc *time.Location) { if loc == nil { loc = time.Local } lastEval := time.Now() lastPublish := lastEval for { select { case <-ctx.Done(): return case <-time.After(republishPoll): } if !link.Connected() { lastEval = time.Now() continue } now := time.Now() shouldPublish := !now.Before(lastPublish.Add(republishInterval)) // A boundary between the last evaluation and now has been crossed. if boundary, err := st.NextScheduleBoundary(ctx, lastEval.In(loc)); err != nil { log.Error("next schedule boundary", "err", err) } else if boundary != nil && !boundary.After(now) { shouldPublish = true } if shouldPublish { republish(ctx, log, link, st) lastPublish = time.Now() } lastEval = now } } // mqttClients prints the MQTT identity of every registered device, one device // per line, as tab-separated fields: // // \t\t\t\t // // provision-clients.sh uses it to create one broker account per device. Real // tabs, not padded columns: names contain spaces, so a space-separated field // would not be safe to parse. The token is the MQTT username and the middle // level of the device's topics (deploy/mosquitto/README.md). It is not a secret // and this never prints one. func mqttClients(args []string) { dbPath := env("CONTROL_DB_PATH", "/data/control.db") for i := 0; i < len(args); i++ { if (args[i] == "--db" || args[i] == "-db") && i+1 < len(args) { dbPath = args[i+1] i++ } } st, err := store.Open(dbPath) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } defer st.Close() devices, err := st.ListDevices(context.Background()) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } for _, dev := range devices { token := strings.TrimSuffix(strings.TrimPrefix(dev.Topic, mqtt.TopicPrefix+"/"), mqtt.SuffixConfig) fmt.Printf("%s\t%s\t%s\t%s\t%s\n", token, dev.Topic, dev.Platform, dev.Child, dev.Name) } } func env(key, fallback string) string { if v := strings.TrimSpace(os.Getenv(key)); v != "" { return v } return fallback } // hashPassword prints a bcrypt hash for the password given as an argument or // read from stdin. It never prints the password itself. func hashPassword(args []string) { var password string if len(args) > 0 { password = args[0] } else { b, err := io.ReadAll(os.Stdin) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } password = strings.TrimRight(string(b), "\r\n") } if password == "" { fmt.Fprintln(os.Stderr, "control: empty password") os.Exit(2) } hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } fmt.Println(string(hash)) }