diff --git a/serving/fleet-manager/README.md b/serving/fleet-manager/README.md new file mode 100644 index 0000000..c0c0db3 --- /dev/null +++ b/serving/fleet-manager/README.md @@ -0,0 +1,99 @@ +# fleet-manager — server-side fleet control plane + +The management service for the diskless kiosk fleet, and the backend for the +admin SPA. It is the central place to **monitor**, **authorise**, and +**lifecycle-control** enrolled hosts. See +[`docs/fleet-management-design.md`](../../docs/fleet-management-design.md) and +[`docs/tpm-enrolment-design.md`](../../docs/tpm-enrolment-design.md) §5. + +Unlike `fleet-hostapi` / `fleet-enrol` (baked into the reproducible verity image, +stdlib-only), this runs **server-side** as its own container on the management +network, so external deps are fine (`grpc-go`, `influxdb-client-go`). + +## Two mTLS planes (distinct trust anchors) + +| Plane | Transport | Verifies | Who | Purpose | +|---|---|---|---|---| +| **admin** | HTTPS + mTLS | **dedicated admin CA** (`admin-pki.sh`) | admin browsers | JSON API + SPA (monitor / authorise / lifecycle) | +| **agent** | gRPC + mTLS | **fleet CA** (host TPM cert) | enrolled hosts | bidi stream: health/metrics up, desired-state down | + +The split is load-bearing: the agent plane uses the host's **reused TPM identity +cert** (the same one used for 802.1X EAP-TLS and from which the NVMe `hostnqn` +derives — "one identity, three stages"), while the admin plane uses a **separate** +CA so an enrolled host can never reach approve/deny/lifecycle. See the identity +discussion in the enrolment design §5. + +## Monitoring path (no host ever touches InfluxDB) + +``` +host agent ──gRPC bidi (mTLS, TPM cert)──▶ fleet-manager ──line protocol──▶ InfluxDB + ◀──── desired-state (target ver / kiosk_url / reboot / drain) ──┘ (+ inventory health) +SPA ──GET /api/hosts/{serial}/metrics──▶ fleet-manager ──Flux──▶ InfluxDB ──JSON──▶ SPA renders charts +``` + +Hosts hold **no InfluxDB credential** and never reach it; the manager is the only +writer/reader, holding the Influx token server-side. Identity is taken from the +mTLS cert (anti-spoofing: a host can only write its own series). Push-only, no +inbound host port. Graphs are rendered **in the SPA**, not Grafana. + +## Inventory record + +Keyed by TPM **cert serial** (= hostnqn key). Born at enrolment approval, filled +by the agent stream and admin action: + +``` +Host (cert_serial) + ├─ identity : cert_serial, hostnqn, mac, ek_hash, starship_name + ├─ binding : asset_tag, switch_port, location (recorded at approval) + ├─ enrolled : approver, at, fingerprint + ├─ version : image_version, rootfs_digest, kernel (agent-reported) + ├─ health : load, cpu, temp, tmpfs%, nvme_root, auth, selinux, chrony, kiosk (latest snapshot; series in InfluxDB) + ├─ usage : kiosk_url, session_uptime, chromium_crashes, display + └─ desired : target_version, kiosk_url, reboot_at, drain (SERVER-AUTHORITATIVE) +``` + +Persisted as a single JSON file (atomic writes) on a mounted volume — stdlib-only, +inspectable, reproducible; a DB swap is a drop-in behind the `Store` interface. + +## API (admin plane) + +| Route | Status | +|---|---| +| `GET /healthz` | ✅ ckpt 1 | +| `GET /api/hosts` | ✅ ckpt 1 — fleet overview | +| `GET /api/hosts/{serial}` | ✅ ckpt 1 — host detail | +| `PUT /api/hosts/{serial}/desired` | ✅ ckpt 1 — set desired-state (audited to admin CN) | +| `GET /api/enrol/pending`, `POST /api/enrol/{id}/approve\|deny` | ⏳ ckpt 3 (drives the broker) | +| `GET /api/hosts/{serial}/metrics` | ⏳ ckpt 4 (InfluxDB/Flux) | + +Agent plane (gRPC `FleetAgent.Session` bidi) — ⏳ ckpt 2. + +## Config (env) + +| Var | Default | Notes | +|---|---|---| +| `FLEET_MGR_ADMIN_ADDR` | `127.0.0.1:9500` | admin listener | +| `FLEET_MGR_ADMIN_CERT` / `_KEY` | — | admin server cert/key (mTLS) | +| `FLEET_MGR_ADMIN_CLIENT_CA` | — | **dedicated admin CA** (NOT the fleet CA) | +| `FLEET_MGR_ADMIN_INSECURE` | `false` | plaintext, no client-cert — **dev only** | +| `FLEET_MGR_INVENTORY` | `/var/lib/fleet-manager/inventory.json` | inventory file (mount a volume) | +| `FLEET_MGR_SEED` | `false` | inject demo hosts so the SPA has data — dev only | +| `HOSTNQN_BASE` | `nqn.2026-06.org.example.fleet:host-` | hostnqn prefix (match the broker) | + +Agent-plane + InfluxDB env land with their checkpoints. + +## Dev run + +```bash +FLEET_MGR_ADMIN_INSECURE=true FLEET_MGR_SEED=true \ + FLEET_MGR_ADMIN_ADDR=127.0.0.1:9500 FLEET_MGR_INVENTORY=/tmp/inv.json \ + go run . +curl -s localhost:9500/api/hosts | jq '.hosts[] | {name:.identity.starship_name, run:.version.image_version, want:.desired.target_version, online}' +``` + +## Build status + +- **Checkpoint 1 (done):** Go skeleton, JSON-file inventory `Store`, admin JSON API + (monitor + lifecycle), admin mTLS wiring, dev seed. Stdlib-only. +- Next: gRPC agent plane (ckpt 2) → authorise/broker fold-in (ckpt 3) → InfluxDB + read API (ckpt 4) → themed SPA (ckpt 5) → Docker/compose + PKI + docs (ckpt 6). diff --git a/serving/fleet-manager/admin.go b/serving/fleet-manager/admin.go new file mode 100644 index 0000000..423c9e8 --- /dev/null +++ b/serving/fleet-manager/admin.go @@ -0,0 +1,139 @@ +// admin.go — the admin JSON API (the SPA's backend) served on the mTLS admin +// plane. Checkpoint 1 covers the MONITOR read surface + LIFECYCLE desired-state; +// the AUTHORISE routes (enrolment approve/deny) land in checkpoint 3 and the +// InfluxDB-backed metrics routes in checkpoint 4. +// +// Every mutating action is attributed to the admin's mTLS client-cert CN +// (adminCN), the same audit convention the broker uses. +package main + +import ( + "encoding/json" + "net/http" + "time" +) + +func (s *server) routes(mux *http.ServeMux) { + // Liveness (no auth-sensitive data; also reachable in insecure dev mode). + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write([]byte("ok\n")) + }) + + // MONITOR — fleet overview + per-host detail. + mux.HandleFunc("GET /api/hosts", s.handleListHosts) + mux.HandleFunc("GET /api/hosts/{serial}", s.handleGetHost) + + // LIFECYCLE — set server-authoritative desired-state for one host. + mux.HandleFunc("PUT /api/hosts/{serial}/desired", s.handleSetDesired) + + // AUTHORISE routes (/api/enrol/*) — checkpoint 3. + // METRICS routes (/api/hosts/{serial}/metrics) — checkpoint 4. +} + +func (s *server) handleListHosts(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{"hosts": s.store.List()}) +} + +func (s *server) handleGetHost(w http.ResponseWriter, r *http.Request) { + serial := r.PathValue("serial") + h, ok := s.store.Get(serial) + if !ok { + writeErr(w, http.StatusNotFound, "no such host") + return + } + writeJSON(w, http.StatusOK, h) +} + +func (s *server) handleSetDesired(w http.ResponseWriter, r *http.Request) { + serial := r.PathValue("serial") + + var d Desired + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10)).Decode(&d); err != nil { + writeErr(w, http.StatusBadRequest, "invalid desired-state JSON: "+err.Error()) + return + } + // Provenance: the client-cert CN is authoritative; ignore any body-supplied value. + d.UpdatedBy = adminCN(r) + + h, ok, err := s.store.SetDesired(serial, d) + if err != nil { + writeErr(w, http.StatusInternalServerError, "persist failed: "+err.Error()) + return + } + if !ok { + writeErr(w, http.StatusNotFound, "no such host") + return + } + // checkpoint 2: after persisting, push the new desired-state down the host's + // live agent stream here (if connected); offline hosts get it on reconnect. + writeJSON(w, http.StatusOK, h) +} + +// adminCN returns the CN of the admin's mTLS client cert, for audit attribution. +// In insecure dev mode there is no peer cert, so it reports a sentinel. +func adminCN(r *http.Request) string { + if r.TLS != nil && len(r.TLS.PeerCertificates) > 0 { + if cn := r.TLS.PeerCertificates[0].Subject.CommonName; cn != "" { + return cn + } + return "unknown-admin" + } + return "dev-insecure" +} + +// ---- JSON response helpers ------------------------------------------------ + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(code) + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = enc.Encode(v) +} + +func writeErr(w http.ResponseWriter, code int, msg string) { + writeJSON(w, code, map[string]string{"error": msg}) +} + +// ---- dev seed ------------------------------------------------------------- + +// seedHosts injects a couple of demo records so the SPA has something to render +// before real enrolment/agent data exists. Gated by FLEET_MGR_SEED=true; never +// overwrites an existing record. Returns how many it added. +func seedHosts(store Store, hostnqnBase string) int { + now := time.Now().UTC() + demo := []*Host{ + { + Identity: Identity{CertSerial: "a1b2c3d4e5f60001", Hostnqn: hostnqnBase + "a1b2c3d4e5f60001", MAC: "de:ad:be:ef:00:01", StarshipName: "comet-beacon-54"}, + Binding: Binding{AssetTag: "KIOSK-0001", SwitchPort: "sw1/12", Location: "Foyer"}, + Enrolled: Enrolled{Approver: "seed", At: now.Add(-72 * time.Hour), Fingerprint: "amber-castle-river-delta-orbit-nova"}, + Version: Version{ImageVersion: "0.5.0", Kernel: "5.14.0-687.10.1.el9_8.0.1.x86_64"}, + Health: Health{LoadAvg: 0.42, CPUPct: 7, TempC: 44, MemUsedPct: 31, TmpfsUsedPct: 58, NVMeRoot: "ok", Auth: "up", SELinux: "enforcing", Chrony: "synced", KioskActive: true}, + Usage: Usage{KioskURL: "http://boot.fleet.lab/kiosk/", SessionUptime: "3h 12m", Display: "1920x1080"}, + Desired: Desired{TargetVersion: "0.5.0", KioskURL: "http://boot.fleet.lab/kiosk/"}, + Online: true, FirstSeen: now.Add(-72 * time.Hour), LastSeen: now.Add(-15 * time.Second), + }, + { + Identity: Identity{CertSerial: "a1b2c3d4e5f60002", Hostnqn: hostnqnBase + "a1b2c3d4e5f60002", MAC: "de:ad:be:ef:00:02", StarshipName: "orion-drifter-07"}, + Binding: Binding{AssetTag: "KIOSK-0002", SwitchPort: "sw1/13", Location: "Lab B"}, + Enrolled: Enrolled{Approver: "seed", At: now.Add(-48 * time.Hour), Fingerprint: "violet-harbor-signal-quartz-piano-lunar"}, + Version: Version{ImageVersion: "0.4.2", Kernel: "5.14.0-687.10.1.el9_8.0.1.x86_64"}, + Health: Health{LoadAvg: 0.11, CPUPct: 3, TempC: 39, MemUsedPct: 27, TmpfsUsedPct: 41, NVMeRoot: "ok", Auth: "up", SELinux: "enforcing", Chrony: "synced", KioskActive: true}, + Usage: Usage{KioskURL: "http://boot.fleet.lab/kiosk/", SessionUptime: "1d 4h 9m", Display: "1920x1080"}, + Desired: Desired{TargetVersion: "0.5.0", KioskURL: "http://boot.fleet.lab/kiosk/"}, // running 0.4.2 → drift + Online: false, FirstSeen: now.Add(-48 * time.Hour), LastSeen: now.Add(-9 * time.Minute), + }, + } + added := 0 + for _, h := range demo { + if _, exists := store.Get(h.Identity.CertSerial); exists { + continue + } + if store.Upsert(h) == nil { + added++ + } + } + return added +} diff --git a/serving/fleet-manager/fleet-manager b/serving/fleet-manager/fleet-manager new file mode 100755 index 0000000..6cc8ecb Binary files /dev/null and b/serving/fleet-manager/fleet-manager differ diff --git a/serving/fleet-manager/go.mod b/serving/fleet-manager/go.mod new file mode 100644 index 0000000..96b3464 --- /dev/null +++ b/serving/fleet-manager/go.mod @@ -0,0 +1,15 @@ +// fleet-manager — server-side fleet control plane for the diskless kiosk fleet: +// inventory, enrolment authorisation, lifecycle (desired-state) control, and the +// metrics gateway that fronts InfluxDB. Backend for the admin SPA. See +// docs/fleet-management-design.md and docs/tpm-enrolment-design.md §5. +// +// Unlike fleet-hostapi / fleet-enrol (baked into the reproducible verity image, +// so stdlib-only), this is a SERVER-SIDE service that ships as its own container +// on the mgmt network — external deps are acceptable here. They are added at the +// build checkpoint that needs them: +// - google.golang.org/grpc + protobuf : the agent bidi stream (checkpoint 2) +// - github.com/influxdata/influxdb-client-go/v2 : metrics read/write (ckpt 2/4) +// Checkpoint 1 (this commit) is stdlib-only. +module fleet-manager + +go 1.22 diff --git a/serving/fleet-manager/inventory.go b/serving/fleet-manager/inventory.go new file mode 100644 index 0000000..baa0d97 --- /dev/null +++ b/serving/fleet-manager/inventory.go @@ -0,0 +1,115 @@ +// inventory.go — the fleet inventory record and the Store abstraction. +// +// The Host record is the join key the rest of the fleet-manager hangs off. It is +// BORN at enrolment approval (identity/binding/enrolled), then FILLED by the +// steady-state agent stream (version/health/usage) and by admin action (desired). +// Schema mirrors docs/fleet-management-design.md "Inventory record" and +// docs/tpm-enrolment-design.md "Inventory record born at enrolment". +// +// Keyed by the TPM cert SERIAL (lowercase hex) — identical to the value the NVMe +// hostnqn derives from and the value the agent's mTLS leaf presents, so the cert +// an agent shows IS its inventory key (no identity translation). See §5 identity +// discussion. +package main + +import "time" + +// Identity — minted at enrolment, immutable afterwards. CertSerial is the primary +// key; Hostnqn = HOSTNQN_BASE + CertSerial (as the broker computes it on approve). +type Identity struct { + CertSerial string `json:"cert_serial"` // lowercase hex — PRIMARY KEY + Hostnqn string `json:"hostnqn,omitempty"` // nqn...:host- + MAC string `json:"mac,omitempty"` // advisory (DeviceContext at enrol) + EKHash string `json:"ek_hash,omitempty"` // TPM EK hash (advisory) + StarshipName string `json:"starship_name,omitempty"` // dnsmasq fleet-namer join +} + +// Binding — the physical↔crypto tie the human records during the OOB approval. +type Binding struct { + AssetTag string `json:"asset_tag,omitempty"` + SwitchPort string `json:"switch_port,omitempty"` + Location string `json:"location,omitempty"` +} + +// Enrolled — provenance of the approval that birthed this record. +type Enrolled struct { + Approver string `json:"approver,omitempty"` // admin CN from the approve mTLS + At time.Time `json:"at,omitempty"` // approval timestamp + Fingerprint string `json:"fingerprint,omitempty"` // the human-verified anchor +} + +// Version — what the host is actually running (agent-reported). +type Version struct { + ImageVersion string `json:"image_version,omitempty"` + RootfsDigest string `json:"rootfs_digest,omitempty"` + Kernel string `json:"kernel,omitempty"` +} + +// Health — the LATEST snapshot only. The full time-series lives in InfluxDB +// (written by the manager as agent reports arrive); this cached last value is +// what the fleet-overview list renders without a per-host Flux query. +type Health struct { + LoadAvg float64 `json:"load_avg,omitempty"` + CPUPct float64 `json:"cpu_pct,omitempty"` + TempC float64 `json:"temp_c,omitempty"` + MemUsedPct float64 `json:"mem_used_pct,omitempty"` + TmpfsUsedPct float64 `json:"tmpfs_used_pct,omitempty"` // overlay RAM pressure (ties to #17) + NVMeRoot string `json:"nvme_root,omitempty"` // ok|degraded|down + Auth string `json:"auth,omitempty"` // 802.1x/macsec: up|down + SELinux string `json:"selinux,omitempty"` // enforcing|permissive + Chrony string `json:"chrony,omitempty"` // synced|unsynced + KioskActive bool `json:"kiosk_active"` +} + +// Usage — kiosk-session facts (agent-reported). +type Usage struct { + KioskURL string `json:"kiosk_url,omitempty"` + SessionUptime string `json:"session_uptime,omitempty"` + ChromiumCrashes int `json:"chromium_crashes,omitempty"` + Display string `json:"display,omitempty"` +} + +// Desired — SERVER-AUTHORITATIVE target state. Set via the admin lifecycle API, +// persisted here, and pushed to the host over its live agent stream (or delivered +// on its next reconnect if offline). Inventory is the source of truth; the stream +// is only the delivery optimisation. +type Desired struct { + TargetVersion string `json:"target_version,omitempty"` + KioskURL string `json:"kiosk_url,omitempty"` + RebootAt *time.Time `json:"reboot_at,omitempty"` + Drain bool `json:"drain,omitempty"` + UpdatedBy string `json:"updated_by,omitempty"` // admin CN + UpdatedAt time.Time `json:"updated_at,omitempty"` +} + +// Host is the full inventory record. +type Host struct { + Identity Identity `json:"identity"` + Binding Binding `json:"binding"` + Enrolled Enrolled `json:"enrolled"` + Version Version `json:"version"` + Health Health `json:"health"` + Usage Usage `json:"usage"` + Desired Desired `json:"desired"` + FirstSeen time.Time `json:"first_seen"` + LastSeen time.Time `json:"last_seen"` // last agent report / stream activity + Online bool `json:"online"` // agent stream currently connected +} + +// Store is the inventory persistence abstraction. The v1 implementation is a +// JSON-file store (jsonStore); the interface keeps a DB swap a drop-in later. +// All methods are safe for concurrent use. +type Store interface { + // List returns a snapshot of every host (copies — safe to mutate). + List() []*Host + // Get returns a copy of one host by cert serial. + Get(serial string) (*Host, bool) + // Upsert creates or replaces a host record (used by enrolment approval). + Upsert(h *Host) error + // SetDesired replaces the desired-state block and returns the updated host. + SetDesired(serial string, d Desired) (*Host, bool, error) + // ReportState applies an agent snapshot (version/health/usage + last-seen). + ReportState(serial string, v Version, h Health, u Usage, at time.Time) (*Host, bool, error) + // SetOnline flips the live-stream presence flag (and stamps last-seen on connect). + SetOnline(serial string, online bool, at time.Time) +} diff --git a/serving/fleet-manager/main.go b/serving/fleet-manager/main.go new file mode 100644 index 0000000..3f5396f --- /dev/null +++ b/serving/fleet-manager/main.go @@ -0,0 +1,151 @@ +// fleet-manager — server-side control plane for the diskless kiosk fleet. +// +// Two planes, mirroring the enrol-broker's two-listener pattern but with distinct +// trust anchors (see docs/tpm-enrolment-design.md §5 "identity"): +// +// - ADMIN plane (this checkpoint): HTTPS + mTLS verified by the DEDICATED admin +// CA (admin-pki.sh) — NOT the fleet CA — so an enrolled host cert can never +// reach approve/deny/lifecycle. Serves the JSON API and (later) the SPA to +// admin browsers. +// - AGENT plane (checkpoint 2): gRPC + mTLS verified by the FLEET CA — the +// host's reused TPM identity cert. Bidi stream: health/metrics up, +// desired-state down. +// +// Checkpoint 1 wires only the admin plane + inventory + a seed host. +package main + +import ( + "crypto/tls" + "crypto/x509" + "log" + "net/http" + "os" + "strconv" + "time" +) + +type config struct { + adminAddr string + adminInsecure bool + adminCert string + adminKey string + adminClientCA string // dedicated admin CA (verifies admin browser certs) + inventoryPath string + seed bool + hostnqnBase string +} + +func loadConfig() config { + return config{ + adminAddr: env("FLEET_MGR_ADMIN_ADDR", "127.0.0.1:9500"), + adminInsecure: env("FLEET_MGR_ADMIN_INSECURE", "") == "true", + adminCert: env("FLEET_MGR_ADMIN_CERT", ""), + adminKey: env("FLEET_MGR_ADMIN_KEY", ""), + adminClientCA: env("FLEET_MGR_ADMIN_CLIENT_CA", ""), + inventoryPath: env("FLEET_MGR_INVENTORY", "/var/lib/fleet-manager/inventory.json"), + seed: env("FLEET_MGR_SEED", "") == "true", + hostnqnBase: env("HOSTNQN_BASE", "nqn.2026-06.org.example.fleet:host-"), + } +} + +// server bundles the shared dependencies the handlers close over. +type server struct { + cfg config + store Store +} + +func main() { + log.SetFlags(log.LstdFlags | log.LUTC) + cfg := loadConfig() + + store, err := newJSONStore(cfg.inventoryPath) + if err != nil { + log.Fatalf("inventory: %v", err) + } + if cfg.seed { + if n := seedHosts(store, cfg.hostnqnBase); n > 0 { + log.Printf("seeded %d demo host(s) (FLEET_MGR_SEED=true — dev only)", n) + } + } + + s := &server{cfg: cfg, store: store} + + mux := http.NewServeMux() + s.routes(mux) + + srv := &http.Server{ + Addr: cfg.adminAddr, + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, + ReadTimeout: 15 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 60 * time.Second, + } + + if cfg.adminInsecure { + log.Printf("WARNING: admin plane INSECURE (plaintext, no client-cert auth) on %s — dev/lab only", cfg.adminAddr) + log.Printf("fleet-manager admin (insecure) listening on %s", cfg.adminAddr) + log.Fatal(srv.ListenAndServe()) + } + + tlsCfg, err := adminTLS(cfg) + if err != nil { + log.Fatalf("admin TLS: %v", err) + } + srv.TLSConfig = tlsCfg + log.Printf("fleet-manager admin (mTLS) listening on %s", cfg.adminAddr) + log.Fatal(srv.ListenAndServeTLS("", "")) // certs are in TLSConfig +} + +// adminTLS builds the mTLS server config: our server cert + REQUIRE a client cert +// signed by the dedicated admin CA. Reusing the fleet CA here would let any host +// cert authenticate to the admin plane — privilege escalation — so it is a hard +// error to omit the admin client CA in secure mode. +func adminTLS(cfg config) (*tls.Config, error) { + if cfg.adminCert == "" || cfg.adminKey == "" { + return nil, errString("FLEET_MGR_ADMIN_CERT and FLEET_MGR_ADMIN_KEY are required (or set FLEET_MGR_ADMIN_INSECURE=true for dev)") + } + if cfg.adminClientCA == "" { + return nil, errString("FLEET_MGR_ADMIN_CLIENT_CA is required — the dedicated admin CA that verifies admin browser certs (must NOT be the fleet CA)") + } + cert, err := tls.LoadX509KeyPair(cfg.adminCert, cfg.adminKey) + if err != nil { + return nil, err + } + caPEM, err := os.ReadFile(cfg.adminClientCA) + if err != nil { + return nil, err + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caPEM) { + return nil, errString("no certificates parsed from FLEET_MGR_ADMIN_CLIENT_CA") + } + return &tls.Config{ + MinVersion: tls.VersionTLS12, + Certificates: []tls.Certificate{cert}, + ClientAuth: tls.RequireAndVerifyClientCert, + ClientCAs: pool, + }, nil +} + +// ---- tiny helpers --------------------------------------------------------- + +func env(k, def string) string { + if v := os.Getenv(k); v != "" { + return v + } + return def +} + +func envInt(k string, def int) int { + if v := os.Getenv(k); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return def +} + +type errString string + +func (e errString) Error() string { return string(e) } diff --git a/serving/fleet-manager/store.go b/serving/fleet-manager/store.go new file mode 100644 index 0000000..b32cb33 --- /dev/null +++ b/serving/fleet-manager/store.go @@ -0,0 +1,167 @@ +// store.go — jsonStore: a mutex-guarded, JSON-file-backed inventory Store. +// +// Rationale for a file (not a DB) in v1: stdlib-only, no cgo, trivially +// reproducible/inspectable, and a fleet of kiosks is small. Writes are atomic +// (temp file + rename) so a crash mid-save can't corrupt the inventory. The Store +// interface keeps a SQLite/Postgres swap a drop-in if scale ever demands it. +// +// On disk: a single JSON object { "": {Host...}, ... } at the +// configured path (default /var/lib/fleet-manager/inventory.json), which in the +// deployed container is a mounted volume so the record survives restarts (unlike +// the broker's deliberately-ephemeral PENDING store). +package main + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "sync" + "time" +) + +type jsonStore struct { + mu sync.Mutex + path string + items map[string]*Host // keyed by cert serial +} + +// newJSONStore loads the inventory file if present (absent = empty fleet). +func newJSONStore(path string) (*jsonStore, error) { + s := &jsonStore{path: path, items: map[string]*Host{}} + b, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return s, nil // fresh fleet + } + return nil, fmt.Errorf("read inventory %s: %w", path, err) + } + if len(b) == 0 { + return s, nil + } + if err := json.Unmarshal(b, &s.items); err != nil { + return nil, fmt.Errorf("parse inventory %s: %w", path, err) + } + if s.items == nil { + s.items = map[string]*Host{} + } + return s, nil +} + +// save persists the current map atomically. Caller must hold s.mu. +func (s *jsonStore) save() error { + if s.path == "" { + return nil // in-memory mode (tests / seed-only) + } + if dir := filepath.Dir(s.path); dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o750); err != nil { + return err + } + } + b, err := json.MarshalIndent(s.items, "", " ") + if err != nil { + return err + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, b, 0o640); err != nil { + return err + } + return os.Rename(tmp, s.path) +} + +func clone(h *Host) *Host { + if h == nil { + return nil + } + c := *h + return &c +} + +func (s *jsonStore) List() []*Host { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]*Host, 0, len(s.items)) + for _, h := range s.items { + out = append(out, clone(h)) + } + // Stable order: online first, then most-recently-seen, then serial. + sort.Slice(out, func(i, j int) bool { + if out[i].Online != out[j].Online { + return out[i].Online + } + if !out[i].LastSeen.Equal(out[j].LastSeen) { + return out[i].LastSeen.After(out[j].LastSeen) + } + return out[i].Identity.CertSerial < out[j].Identity.CertSerial + }) + return out +} + +func (s *jsonStore) Get(serial string) (*Host, bool) { + s.mu.Lock() + defer s.mu.Unlock() + h, ok := s.items[serial] + return clone(h), ok +} + +func (s *jsonStore) Upsert(h *Host) error { + if h == nil || h.Identity.CertSerial == "" { + return fmt.Errorf("upsert: host needs a cert serial") + } + s.mu.Lock() + defer s.mu.Unlock() + if existing, ok := s.items[h.Identity.CertSerial]; ok { + h.FirstSeen = existing.FirstSeen // preserve birth time on re-approval + } else if h.FirstSeen.IsZero() { + h.FirstSeen = time.Now().UTC() + } + s.items[h.Identity.CertSerial] = clone(h) + return s.save() +} + +func (s *jsonStore) SetDesired(serial string, d Desired) (*Host, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + h, ok := s.items[serial] + if !ok { + return nil, false, nil + } + d.UpdatedAt = time.Now().UTC() + h.Desired = d + if err := s.save(); err != nil { + return nil, true, err + } + return clone(h), true, nil +} + +func (s *jsonStore) ReportState(serial string, v Version, h Health, u Usage, at time.Time) (*Host, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + host, ok := s.items[serial] + if !ok { + return nil, false, nil + } + host.Version = v + host.Health = h + host.Usage = u + host.LastSeen = at + if err := s.save(); err != nil { + return nil, true, err + } + return clone(host), true, nil +} + +func (s *jsonStore) SetOnline(serial string, online bool, at time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + h, ok := s.items[serial] + if !ok { + return + } + h.Online = online + if online { + h.LastSeen = at + } + _ = s.save() // presence is best-effort; a failed persist isn't fatal +}