// 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" "crypto/rand" "errors" "fmt" "io" "log/slog" "math/big" "net/http" "os" "os/signal" "strconv" "strings" "syscall" "time" "golang.org/x/crypto/bcrypt" "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/alert" "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/hub" "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 // Maintenance (Phase 7): the daily usage rollup is rebuilt periodically so it // is repaired from the detail table, and the append-only audit log is capped so // it cannot grow without bound (AGENTS.md §7 retention). const ( maintenanceInterval = 15 * time.Minute auditMaxRows = 20000 ) 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 "keycode": keycode(os.Args[2:]) return case "rollup": rollup(os.Args[2:]) return case "alerts": alertsCommand(os.Args[2:]) return case "usage": usageCommand(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 } // Backfill the daily rollup from any telemetry written before Phase 7, then // cap the audit log. Both are idempotent. if n, err := st.RebuildRollups(context.Background()); err != nil { return err } else if n > 0 { log.Info("usage rollups rebuilt", "rows", n) } if n, err := st.CapAudit(context.Background(), auditMaxRows); err != nil { log.Error("audit cap", "err", err) } else if n > 0 { log.Info("audit log capped", "removed", n, "max", auditMaxRows) } // Phase 8: seed the shared device keycode from environment.d. The stored // value is the runtime copy that `control keycode` can change; the // environment is the deploy-time source. A malformed hash stops start-up // rather than silently disabling the keycode path. if seeded, err := st.SeedKeycodeHash(context.Background(), cfg.DeviceKeycodeHash); err != nil { return err } else if seeded { log.Info("device keycode seeded from CONTROL_DEVICE_KEYCODE_HASH") } 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() go maintenanceLoop(ctx, log, st) // 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 } // Phase 9: the alert engine turns a limit hit or a too-long offline spell // into one Apprise message to Sam (FR-13). It is optional; /ask and // /notify keep working without it. if cfg.AlertsEnabled { engine := alert.New(st, srv.Apprise(), log, cfg.AlertOfflineAfter, cfg.ScheduleLocation) engine.Interval = cfg.AlertInterval go engine.Run(ctx) if !cfg.AppriseConfigured() { log.Warn("alerts are enabled but Apprise is not configured; alerts will be recorded as undelivered") } else { log.Info("alert engine started", "offline_after", cfg.AlertOfflineAfter.String()) } } else { log.Info("alert engine disabled") } 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 } } // maintenanceLoop keeps derived data healthy without blocking the service: it // rebuilds the daily rollups and caps the audit log on a fixed interval. func maintenanceLoop(ctx context.Context, log *slog.Logger, st *store.Store) { ticker := time.NewTicker(maintenanceInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: runMaintenance(ctx, log, st) } } } func runMaintenance(ctx context.Context, log *slog.Logger, st *store.Store) { mctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() if n, err := st.RebuildRollups(mctx); err != nil { log.Error("usage rollups", "err", err) } else if n > 0 { log.Debug("usage rollups rebuilt", "rows", n) } if n, err := st.CapAudit(mctx, auditMaxRows); err != nil { log.Error("audit cap", "err", err) } else if n > 0 { log.Info("audit log capped", "removed", n, "max", auditMaxRows) } } // alertsCommand prints the most recent alerts, newest first, one per line as // tab-separated fields: // // \t\t\t\t\t // // It is a read-only operational view. The Phase 9 verification script uses it // to check that an alert was raised and delivered. func alertsCommand(args []string) { dbPath := env("CONTROL_DB_PATH", "/data/control.db") limit := 20 for i := 0; i < len(args); i++ { switch args[i] { case "--db", "-db": if i+1 < len(args) { dbPath = args[i+1] i++ } case "--limit", "-limit": if i+1 < len(args) { if n, err := strconv.Atoi(args[i+1]); err == nil { limit = n } i++ } } } st, err := store.Open(dbPath) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } defer st.Close() if err := st.Migrate(context.Background()); err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } list, err := st.ListAlerts(context.Background(), limit) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } for _, a := range list { delivered := "pending" if a.Delivered { delivered = "delivered" } device := a.DeviceName if device == "" { device = "-" } fmt.Printf("%d\t%s\t%s\t%s\t%s\t%s\n", a.ID, a.Kind, delivered, a.CreatedAt.Format(time.RFC3339), device, a.Message) } } // usageCommand prints one day's per-device screen time. --detail reads the raw // telemetry detail table instead of the derived rollup, so an operator can // check the daily answer independently. func usageCommand(args []string) { dbPath := env("CONTROL_DB_PATH", "/data/control.db") day := time.Now().Format("2006-01-02") detail := false for i := 0; i < len(args); i++ { switch args[i] { case "--db", "-db": if i+1 < len(args) { dbPath = args[i+1] i++ } case "--day", "-day": if i+1 < len(args) { day = args[i+1] i++ } case "--detail", "-detail": detail = true } } st, err := store.Open(dbPath) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } defer st.Close() ctx := context.Background() if err := st.Migrate(ctx); err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } source := "rollup" var rows []store.DeviceDayUsage if detail { source = "detail" rows, err = st.DetailUsageByDeviceForDay(ctx, day) } else { rows, err = st.UsageByDeviceForDay(ctx, day) } if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } fmt.Printf("day=%s source=%s\n", day, source) total := 0 for _, u := range rows { screen := hub.ScreenSeconds(u.AppSeconds, u.DomainSeconds) total += screen fmt.Printf("%s\t%s\tapp=%d\tdomain=%d\tscreen=%d\n", u.Name, u.Child, u.AppSeconds, u.DomainSeconds, screen) } fmt.Printf("total_screen=%d\n", total) } // rollup rebuilds the daily usage rollup from the telemetry detail table. It is // idempotent and safe to run while the service is up; it exits non-zero on a // failure so a script can tell. func rollup(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() if err := st.Migrate(context.Background()); err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } n, err := st.RebuildRollups(context.Background()) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } fmt.Printf("rebuilt %d daily rollup rows from %s\n", n, dbPath) } // 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 } // keycode sets the shared device keycode (AGENTS.md §10 rule 3). It reads the // code from a positional argument or stdin, or generates a random six-digit // code with --generate, and stores only its bcrypt hash in SQLite. The code // itself is printed once, and only when it was generated here, so it can be set // on each device; it is never logged by the running service. func keycode(args []string) { dbPath := env("CONTROL_DB_PATH", "/data/control.db") generate := false code := "" for i := 0; i < len(args); i++ { switch args[i] { case "--db", "-db": if i+1 < len(args) { dbPath = args[i+1] i++ } case "--generate", "-generate": generate = true default: if code == "" { code = args[i] } } } if generate { n, err := rand.Int(rand.Reader, big.NewInt(1000000)) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } code = fmt.Sprintf("%06d", n.Int64()) } else if code == "" { b, err := io.ReadAll(os.Stdin) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } code = strings.TrimSpace(string(b)) } if len(code) < 4 || len(code) > 64 { fmt.Fprintln(os.Stderr, "control: keycode must be 4 to 64 characters") os.Exit(2) } st, err := store.Open(dbPath) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } defer st.Close() if err := st.Migrate(context.Background()); err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } hash, err := bcrypt.GenerateFromPassword([]byte(code), bcrypt.DefaultCost) if err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } if err := st.SetKeycodeHash(context.Background(), string(hash), "admin"); err != nil { fmt.Fprintln(os.Stderr, "control:", err) os.Exit(1) } if generate { fmt.Printf("device keycode set: %s\n", code) return } fmt.Println("device keycode updated") } // 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)) }