fleet-manager: checkpoint 1 — service skeleton + inventory store + admin API

First slice of the server-side fleet control plane (serving/fleet-manager/), the
backend for the admin SPA (monitor / authorise / lifecycle). See
docs/fleet-management-design.md + tpm-enrolment-design.md §5.

This checkpoint (stdlib-only):
- Inventory record (identity/binding/enrolled/version/health/usage/desired),
  keyed by TPM cert serial = hostnqn key, per the design's 'inventory record born
  at enrolment'. JSON-file Store (atomic writes, mutex) behind an interface so a
  DB swap is a drop-in.
- Admin plane: mTLS server verified by the DEDICATED admin CA (never the fleet CA,
  so a host cert can't reach admin) + a plaintext dev mode. JSON API:
  GET /api/hosts, GET /api/hosts/{serial}, PUT /api/hosts/{serial}/desired
  (desired-state audited to the admin client-cert CN).
- Dev seed (FLEET_MGR_SEED) so the SPA has data before real enrolment/agent flow.

Smoke-tested: list/get/desired-PUT (incl. version-drift + drain), 404s, atomic
persistence. go build + go vet + gofmt clean.

Agent gRPC plane, broker-fold-in authorise, InfluxDB metrics, and the themed SPA
follow in checkpoints 2-6.
This commit is contained in:
Audiobuzz
2026-07-04 03:25:54 +00:00
parent 68dee79832
commit 915a15b6a1
7 changed files with 686 additions and 0 deletions
+99
View File
@@ -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).
+139
View File
@@ -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
}
Binary file not shown.
+15
View File
@@ -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
+115
View File
@@ -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-<serial>
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)
}
+151
View File
@@ -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) }
+167
View File
@@ -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 { "<cert_serial>": {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
}