e523eae3 — eitri

Makefile

cmd/eitri-agent/main.go

Old		New
@@ -1,5 +1,5 @@
1	// eitri-agent: BYO-hardware agent. Enrolls the host and then runs the	1	// eitri-agent: BYO-hardware agent. Enrolls the host via "join <blob>" and
2	// reconcile + sync loop indefinitely.	2	// then runs the reconcile + sync loop indefinitely.
3	package main	3	package main
4		4
5	import (	5	import (
@@ -28,21 +28,21 @@ import (
28	"github.com/a73x/eitri/internal/agent/seed"	28	"github.com/a73x/eitri/internal/agent/seed"
29	"github.com/a73x/eitri/internal/agent/state"	29	"github.com/a73x/eitri/internal/agent/state"
30	"github.com/a73x/eitri/internal/agent/syncclient"	30	"github.com/a73x/eitri/internal/agent/syncclient"
		31	"github.com/a73x/eitri/internal/joinblob"
31	)	32	)
32		33
33	func main() {	34	func main() {
34	stateDir := flag.String("state-dir", "/var/lib/eitri-agent", "agent state directory")	35	stateDir := flag.String("state-dir", "/var/lib/eitri-agent", "agent state directory")
35	chBin := flag.String("ch-bin", "cloud-hypervisor", "path to cloud-hypervisor binary")	36	chBin := flag.String("ch-bin", "cloud-hypervisor", "path to cloud-hypervisor binary")
36	firmware := flag.String("firmware", "/usr/share/eitri/hypervisor-fw", "path to hypervisor-fw")	37	firmware := flag.String("firmware", "/usr/share/eitri/hypervisor-fw", "path to hypervisor-fw")
37	server := flag.String("server", "", "HTTP server URL (e.g. http://localhost:8080)")	38	overlayKind := flag.String("overlay", "tailscale", "overlay kind: tailscale or none")
38	quicAddr := flag.String("quic-addr", "", "server QUIC address (e.g. localhost:8443)")
39	token := flag.String("token", "", "enrollment token")
40	overlayKind := flag.String("overlay", "tailscale", "overlay kind: tailscale or none")
41	overlayAuthkey := flag.String("overlay-authkey", "", "overlay auth key (optional, for initial enroll on dedicated hosts — requires --manage-overlay)")	39	overlayAuthkey := flag.String("overlay-authkey", "", "overlay auth key (optional, for initial enroll on dedicated hosts — requires --manage-overlay)")
42	manageOverlay := flag.Bool("manage-overlay", false, "allow agent to additively modify overlay route advertisement (opt-in; for dedicated hosts)")	40	manageOverlay := flag.Bool("manage-overlay", false, "allow agent to additively modify overlay route advertisement (opt-in; for dedicated hosts)")
43	noMasqIfaces := flag.String("no-masquerade-ifaces", "", "comma-separated interfaces to exclude from NAT masquerade (e.g. wg0)")	41	noMasqIfaces := flag.String("no-masquerade-ifaces", "", "comma-separated interfaces to exclude from NAT masquerade (e.g. wg0)")
44	tombstoneGrace := flag.Duration("tombstone-grace", 5*time.Minute, "quarantine period for tombstoned VMs before destroy")	42	tombstoneGrace := flag.Duration("tombstone-grace", 5*time.Minute, "quarantine period for tombstoned VMs before destroy")
45	vanishGrace := flag.Duration("vanish-grace", time.Hour, "quarantine period for VMs that vanished without a tombstone")	43	vanishGrace := flag.Duration("vanish-grace", time.Hour, "quarantine period for VMs that vanished without a tombstone")
		44	stepTimeout := flag.Duration("step-timeout", 15*time.Minute, "watchdog bound on one WHOLE reconcile step — all VMs, summed (0 disables); keep above the 10m image-download timeout")
		45	imageCacheMaxGB := flag.Int64("image-cache-max-gb", 20, "evict least-recently-used cached base images beyond this size (0 = never evict)")
46	flag.Parse()	46	flag.Parse()
47		47
48	st, err := state.Open(*stateDir)	48	st, err := state.Open(*stateDir)
@@ -51,18 +51,26 @@ func main() {
51	os.Exit(1)	51	os.Exit(1)
52	}	52	}
53		53
54	if flag.Arg(0) == "enroll" {	54	if flag.Arg(0) == "join" {
55	runEnroll(st, server, quicAddr, token, stateDir, *overlayKind)	55	runJoin(st, flag.Arg(1), *overlayKind)
56	return	56	return
57	}	57	}
58		58
59	runAgent(st, stateDir, chBin, firmware, overlayKind, overlayAuthkey, noMasqIfaces, manageOverlay, tombstoneGrace, *vanishGrace)	59	runAgent(st, stateDir, chBin, firmware, overlayKind, overlayAuthkey, noMasqIfaces, manageOverlay, tombstoneGrace, vanishGrace, stepTimeout, *imageCacheMaxGB)
60	}	60	}
61		61
62	// runEnroll handles the "enroll" subcommand.	62	// runJoin handles the "join <blob>" subcommand: decode the join blob, enroll,
63	func runEnroll(st *state.Store, server, quicAddr, token, stateDir, overlayKind string) {	63	// and persist identity — pinning the server cert from the blob (the enroll
64	if server == "" \|\| quicAddr == "" \|\| token == "" {	64	// response's fingerprint is ignored, so the blob is the sole trust root).
65	fmt.Fprintln(os.Stderr, "enroll requires --server, --quic-addr, and --token")	65	func runJoin(st *state.Store, blob, overlayKind string) {
		66	if blob == "" {
		67	fmt.Fprintln(os.Stderr, "usage: eitri-agent join <join-blob>")
		68	os.Exit(1)
		69	}
		70	f, err := joinblob.Decode(blob)
		71	if err != nil {
		72	// Never echo the blob itself — it carries a bearer token.
		73	fmt.Fprintf(os.Stderr, "invalid join blob: %v\n", err)
66	os.Exit(1)	74	os.Exit(1)
67	}	75	}
68		76
@@ -70,9 +78,8 @@ func runEnroll(st *state.Store, server, quicAddr, token, stateDir, overlayKind s
70	if err != nil {	78	if err != nil {
71	hostname = "unknown"	79	hostname = "unknown"
72	}	80	}
73
74	body, err := json.Marshal(map[string]string{	81	body, err := json.Marshal(map[string]string{
75	"token": token,	82	"token": f.Token,
76	"name": hostname,	83	"name": hostname,
77	"os": runtime.GOOS,	84	"os": runtime.GOOS,
78	"arch": runtime.GOARCH,	85	"arch": runtime.GOARCH,
@@ -84,24 +91,31 @@ func runEnroll(st *state.Store, server, quicAddr, token, stateDir, overlayKind s
84	os.Exit(1)	91	os.Exit(1)
85	}	92	}
86		93
87	resp, err := http.Post(server+"/api/v1/enroll", "application/json", bytes.NewReader(body))	94	client := &http.Client{Timeout: 30 * time.Second}
		95	resp, err := client.Post(f.HTTPURL+"/api/v1/enroll", "application/json", bytes.NewReader(body))
88	if err != nil {	96	if err != nil {
89	slog.Error("enroll request", "err", err)	97	slog.Error("enroll request", "err", err)
90	os.Exit(1)	98	os.Exit(1)
91	}	99	}
92	defer resp.Body.Close()	100	defer resp.Body.Close()
93		101	respBody, readErr := io.ReadAll(resp.Body)
94	respBody, _ := io.ReadAll(resp.Body)	102	if readErr != nil {
		103	slog.Error("read enroll response", "err", readErr)
		104	os.Exit(1)
		105	}
		106	if resp.StatusCode == http.StatusForbidden {
		107	fmt.Fprintln(os.Stderr, "enroll rejected: token already used or expired — mint a new join token")
		108	os.Exit(1)
		109	}
95	if resp.StatusCode != http.StatusCreated {	110	if resp.StatusCode != http.StatusCreated {
96	fmt.Fprintf(os.Stderr, "enroll failed (HTTP %d): %s\n", resp.StatusCode, respBody)	111	fmt.Fprintf(os.Stderr, "enroll failed (HTTP %d): %s\n", resp.StatusCode, respBody)
97	os.Exit(1)	112	os.Exit(1)
98	}	113	}
99		114
100	var result struct {	115	var result struct {
101	HostID string `json:"host_id"`	116	HostID string `json:"host_id"`
102	Credential string `json:"credential"`	117	Credential string `json:"credential"`
103	BridgeCIDR string `json:"bridge_cidr"`	118	BridgeCIDR string `json:"bridge_cidr"`
104	ServerCertSHA256 string `json:"server_cert_sha256"`
105	}	119	}
106	if err := json.Unmarshal(respBody, &result); err != nil {	120	if err := json.Unmarshal(respBody, &result); err != nil {
107	slog.Error("parse enroll response", "err", err)	121	slog.Error("parse enroll response", "err", err)
@@ -112,14 +126,13 @@ func runEnroll(st *state.Store, server, quicAddr, token, stateDir, overlayKind s
112	HostID: result.HostID,	126	HostID: result.HostID,
113	Credential: result.Credential,	127	Credential: result.Credential,
114	BridgeCIDR: result.BridgeCIDR,	128	BridgeCIDR: result.BridgeCIDR,
115	ServerQUICAddr: quicAddr,	129	ServerQUICAddr: f.QUICAddr,
116	ServerCertSHA256: result.ServerCertSHA256,	130	ServerCertSHA256: f.CertFP, // authoritative; response fingerprint ignored
117	}	131	}
118	if err := st.SaveIdentity(id); err != nil {	132	if err := st.SaveIdentity(id); err != nil {
119	slog.Error("save identity", "err", err)	133	slog.Error("save identity", "err", err)
120	os.Exit(1)	134	os.Exit(1)
121	}	135	}
122
123	fmt.Printf("Enrolled: host_id=%s bridge_cidr=%s\n", result.HostID, result.BridgeCIDR)	136	fmt.Printf("Enrolled: host_id=%s bridge_cidr=%s\n", result.HostID, result.BridgeCIDR)
124	}	137	}
125		138
@@ -146,10 +159,10 @@ func splitComma(s string) []string {
146	}	159	}
147		160
148	// runAgent handles the normal (no subcommand) run mode.	161	// runAgent handles the normal (no subcommand) run mode.
149	func runAgent(st *state.Store, stateDir, chBin, firmware, overlayKind, overlayAuthkey, noMasqIfacesStr string, manageOverlay bool, tombstoneGrace, vanishGrace time.Duration) {	162	func runAgent(st *state.Store, stateDir, chBin, firmware, overlayKind, overlayAuthkey, noMasqIfacesStr string, manageOverlay bool, tombstoneGrace, vanishGrace, stepTimeout time.Duration, imageCacheMaxGB int64) {
150	id, ok := st.Identity()	163	id, ok := st.Identity()
151	if !ok {	164	if !ok {
152	fmt.Fprintln(os.Stderr, "not enrolled — run with 'enroll' subcommand first")	165	fmt.Fprintln(os.Stderr, "not enrolled — run with 'join <blob>' subcommand first")
153	os.Exit(1)	166	os.Exit(1)
154	}	167	}
155		168
@@ -211,6 +224,14 @@ func runAgent(st *state.Store, stateDir, chBin, firmware, overlayKind, overlayAu
211		224
212	prov := cloudhv.New(st, chBin, firmware, realRunner)	225	prov := cloudhv.New(st, chBin, firmware, realRunner)
213	cache := imagecache.New(st.ImagesDir(), realRunner)	226	cache := imagecache.New(st.ImagesDir(), realRunner)
		227	// Clamp before shifting: GB<<30 overflows int64 for absurd flag values —
		228	// same trap class cloudhv's maxDiskGB comment documents. 1 PiB is beyond
		229	// any real cache; anything above disables eviction just like 0 would.
		230	if imageCacheMaxGB < 0 \|\| imageCacheMaxGB > 1<<20 {
		231	slog.Warn("image-cache-max-gb out of range [0, 2^20]; disabling eviction", "value", imageCacheMaxGB)
		232	imageCacheMaxGB = 0
		233	}
		234	cache.MaxBytes = imageCacheMaxGB << 30
214		235
215	engine := &reconcile.Engine{	236	engine := &reconcile.Engine{
216	St: st,	237	St: st,
@@ -223,6 +244,7 @@ func runAgent(st *state.Store, stateDir, chBin, firmware, overlayKind, overlayAu
223	TombstoneGrace: tombstoneGrace,	244	TombstoneGrace: tombstoneGrace,
224	VanishGrace: vanishGrace,	245	VanishGrace: vanishGrace,
225	MaxCreateAttempts: 3,	246	MaxCreateAttempts: 3,
		247	StepTimeout: stepTimeout,
226	}	248	}
227		249
228	// Compile-time interface satisfaction checks.	250	// Compile-time interface satisfaction checks.

cmd/eitri-server/main.go

Old		New
@@ -11,6 +11,7 @@ import (
11	"os"	11	"os"
12	"time"	12	"time"
13		13
		14	"github.com/a73x/eitri/internal/joinblob"
14	"github.com/a73x/eitri/internal/server/api"	15	"github.com/a73x/eitri/internal/server/api"
15	"github.com/a73x/eitri/internal/server/hub"	16	"github.com/a73x/eitri/internal/server/hub"
16	"github.com/a73x/eitri/internal/server/registry"	17	"github.com/a73x/eitri/internal/server/registry"
@@ -30,6 +31,16 @@ type config struct {
30	CIDRPool string `json:"cidr_pool"`	31	CIDRPool string `json:"cidr_pool"`
31	DefaultImageURL string `json:"default_image_url"`	32	DefaultImageURL string `json:"default_image_url"`
32	DefaultImageSHA string `json:"default_image_sha256"`	33	DefaultImageSHA string `json:"default_image_sha256"`
		34	AdvertiseHTTP string `json:"advertise_http"`
		35	AdvertiseQUIC string `json:"advertise_quic"`
		36	// CredentialMaxAge optionally bounds host credential age (Go duration,
		37	// e.g. "2160h" for 90 days). Empty/zero disables — revocation via
		38	// POST /api/v1/hosts/{id}/revoke-credential is the primary mechanism;
		39	// max-age forces periodic re-enrollment and is opt-in defense-in-depth.
		40	CredentialMaxAge string `json:"credential_max_age"`
		41	// AuditRetention bounds the audit_log age (Go duration; default "2160h" =
		42	// 90 days; "0" disables pruning). Pruned at startup and daily.
		43	AuditRetention string `json:"audit_retention"`
33	}	44	}
34		45
35	func main() {	46	func main() {
@@ -49,6 +60,10 @@ func main() {
49	slog.Error("admin_token and host_secret are required")	60	slog.Error("admin_token and host_secret are required")
50	os.Exit(1)	61	os.Exit(1)
51	}	62	}
		63	if cfg.AdvertiseHTTP == "" \|\| cfg.AdvertiseQUIC == "" {
		64	slog.Error("advertise_http and advertise_quic are required (the addresses agents use to reach this server, e.g. http://192.168.0.190:8080 and 192.168.0.190:8443)")
		65	os.Exit(1)
		66	}
52		67
53	st, err := store.Open(cfg.DBPath, cfg.CIDRPool)	68	st, err := store.Open(cfg.DBPath, cfg.CIDRPool)
54	if err != nil {	69	if err != nil {
@@ -67,12 +82,79 @@ func main() {
67	os.Exit(1)	82	os.Exit(1)
68	}	83	}
69		84
		85	// Log the identity agents pin — the operator verifies this out-of-band
		86	// during the rotation ceremony (docs/cert-rotation.md step 3).
		87	slog.Info("server cert", "fingerprint", certFP)
		88
		89	// Rotation nudge: the agent pin ignores expiry so nothing breaks at
		90	// NotAfter, but a long-lived key is a widening forgery window. Warn while
		91	// inside the renewal window — at startup AND daily, because servers here
		92	// are long-lived daemons that can cross into the window (or past expiry)
		93	// without ever restarting. See docs/cert-rotation.md for the ceremony.
		94	warnIfRenewalDue := func() {
		95	notAfter, due := transport.CertRenewalDue(certPEM, time.Now())
		96	if !due {
		97	return
		98	}
		99	if notAfter.IsZero() {
		100	slog.Warn("server cert unparseable — inspect server.crt (docs/cert-rotation.md)")
		101	return
		102	}
		103	slog.Warn("server cert renewal due — rotate and re-enroll agents (docs/cert-rotation.md)",
		104	"not_after", notAfter.Format(time.RFC3339))
		105	}
		106	warnIfRenewalDue()
		107
		108	// Audit retention: bound the append-only log (default 90 days, 0 disables).
		109	auditRetention := 90 * 24 * time.Hour
		110	if cfg.AuditRetention != "" {
		111	auditRetention, err = time.ParseDuration(cfg.AuditRetention)
		112	if err != nil {
		113	slog.Error("audit_retention invalid", "err", err)
		114	os.Exit(1)
		115	}
		116	if auditRetention < 0 {
		117	// Almost certainly a typo — refuse rather than silently keeping
		118	// the audit log forever ("0" is the explicit disable spelling).
		119	slog.Error("audit_retention must be >= 0", "value", cfg.AuditRetention)
		120	os.Exit(1)
		121	}
		122	}
		123	pruneAudit := func() {
		124	if auditRetention <= 0 {
		125	return
		126	}
		127	if n, err := st.PruneAudit(auditRetention); err != nil {
		128	slog.Warn("audit prune failed", "err", err)
		129	} else if n > 0 {
		130	slog.Info("audit pruned", "rows", n, "retention", auditRetention)
		131	}
		132	}
		133	pruneAudit()
		134
		135	// Daily housekeeping: cert-renewal nudge + audit retention.
		136	go func() {
		137	for range time.Tick(24 * time.Hour) {
		138	warnIfRenewalDue()
		139	pruneAudit()
		140	}
		141	}()
		142
		143	// Fail fast if the advertised addresses are non-empty but malformed (e.g. a
		144	// URL with no scheme): otherwise every enroll-token mint would 500 at runtime.
		145	if _, err := joinblob.Encode(cfg.AdvertiseHTTP, cfg.AdvertiseQUIC, "startup-probe", certFP); err != nil {
		146	slog.Error("advertise_http/advertise_quic invalid", "err", err)
		147	os.Exit(1)
		148	}
		149
70	reg := registry.New(time.Now)	150	reg := registry.New(time.Now)
71	h := hub.New()	151	h := hub.New()
72		152
73	a := api.New(api.Config{AdminToken: cfg.AdminToken, HostSecret: []byte(cfg.HostSecret),	153	a := api.New(api.Config{AdminToken: cfg.AdminToken, HostSecret: []byte(cfg.HostSecret),
74	DefaultImage: api.DefaultImage{URL: cfg.DefaultImageURL, SHA256: cfg.DefaultImageSHA},	154	DefaultImage: api.DefaultImage{URL: cfg.DefaultImageURL, SHA256: cfg.DefaultImageSHA},
75	ServerCertSHA256: certFP},	155	ServerCertSHA256: certFP,
		156	AdvertiseHTTP: cfg.AdvertiseHTTP,
		157	AdvertiseQUIC: cfg.AdvertiseQUIC},
76	st, reg, h)	158	st, reg, h)
77		159
78	tlsConf, err := transport.ServerTLS(certPEM, keyPEM)	160	tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
@@ -86,7 +168,15 @@ func main() {
86	slog.Error("quic listen", "err", err)	168	slog.Error("quic listen", "err", err)
87	os.Exit(1)	169	os.Exit(1)
88	}	170	}
89	svc := syncsvc.New(st, reg, h, []byte(cfg.HostSecret))	171	var maxCredAge time.Duration
		172	if cfg.CredentialMaxAge != "" {
		173	maxCredAge, err = time.ParseDuration(cfg.CredentialMaxAge)
		174	if err != nil {
		175	slog.Error("credential_max_age invalid", "err", err)
		176	os.Exit(1)
		177	}
		178	}
		179	svc := syncsvc.New(st, reg, h, []byte(cfg.HostSecret), maxCredAge)
90	go func() {	180	go func() {
91	slog.Info("quic listening", "addr", cfg.QUICListen)	181	slog.Info("quic listening", "addr", cfg.QUICListen)
92	if err := svc.Serve(context.Background(), lis); err != nil {	182	if err := svc.Serve(context.Background(), lis); err != nil {

docs/architecture.md

Old		New
@@ -26,11 +26,11 @@ side effects on real infrastructure live in the agent.
26	\| # \| Invariant \| Enforced by \|	26	\| # \| Invariant \| Enforced by \|
27	\|---\|-----------\|-------------\|	27	\|---\|-----------\|-------------\|
28	\| R1 \| The control plane and the data plane never import each other (even transitively). \| `internal/arch` `TestControlAndDataPlaneAreDisjoint` (production, transitive) + `depguard` `server-no-agent` / `agent-no-server` (non-test files). \|	28	\| R1 \| The control plane and the data plane never import each other (even transitively). \| `internal/arch` `TestControlAndDataPlaneAreDisjoint` (production, transitive) + `depguard` `server-no-agent` / `agent-no-server` (non-test files). \|
29	\| R2 \| No `internal/server` package shells out — the server is pure control plane. \| `internal/arch` `TestServerNeverShellsOut` + `depguard` `server-no-exec`. \|	29	\| R2 \| No `internal/server` package shells out — the server is pure control plane. Checked transitively: an internal wrapper around `os/exec` cannot smuggle a shell-out in. \| `internal/arch` `TestServerNeverShellsOut` (transitive) + `depguard` `server-no-exec` (direct, fast in-editor). \|
30	\| R3 \| The wire contract (`pb`, `transport`) imports no other internal package, so a heavy dependency can't leak across the boundary into both binaries. \| `internal/arch` `TestWireContractIsLeaf`. Behavior pinned by `transport` round-trip contract tests. \|	30	\| R3 \| The wire contract (`pb`, `transport`) imports no other internal package, so a heavy dependency can't leak across the boundary into both binaries. \| `internal/arch` `TestWireContractIsLeaf`. Behavior pinned by `transport` round-trip contract tests. \|
31	\| R4 \| Pure domain packages (`agent/state`, `agent/seed`, `agent/ipalloc`, `server/registry`) don't depend on the transport stack (HTTP/QUIC/`transport`). `server/store` may use `transport` (cert helpers) but not HTTP/QUIC. \| `internal/arch` `TestDomainDoesNotImportTransportStack` + `depguard` `domain-no-transport`. \|	31	\| R4 \| Pure domain packages (`agent/state`, `agent/seed`, `agent/ipalloc`, `server/registry`) don't depend on the transport stack (HTTP/QUIC/`transport`). `server/store` may use `transport` (cert helpers) but not HTTP/QUIC. \| `internal/arch` `TestDomainDoesNotImportTransportStack` + `depguard` `domain-no-transport`. \|
32	\| R5 \| The reconcile boundary interfaces (`Provisioner`, `NetEnv`, `Overlay`) stay consumer-owned and small; the IPAM seam (`NetEnv.AllocateIP`/`GuestNetwork`) is where a future central allocator plugs in. \| Convention (below) + `ireturn` allow-list keeps the seams' interface returns honest. \|	32	\| R5 \| The reconcile boundary interfaces (`Provisioner`, `NetEnv`, `Overlay`) stay consumer-owned and small; the IPAM seam (`NetEnv.AllocateIP`/`GuestNetwork`) is where a future central allocator plugs in. \| Convention (below) + `ireturn` allow-list keeps the seams' interface returns honest. \|
33	\| R6 \| All external process execution in the data plane funnels through `agent/exec.Runner`. The sole exception is `agent/cloudhv`, which launches the long-lived cloud-hypervisor process directly. \| `internal/arch` `TestOnlyCloudhvImportsOsExecInDataPlane`. \|	33	\| R6 \| All external process execution in the data plane funnels through `agent/exec.Runner`. The sole exception is `agent/cloudhv`, which launches the long-lived cloud-hypervisor process directly. Checked transitively (reaching `os/exec` via the sanctioned `cloudhv` is fine). \| `internal/arch` `TestOnlyCloudhvImportsOsExecInDataPlane` (transitive). \|
34		34
35	> The `internal/arch` tests shell out to `go list`, so Go's test cache can't see	35	> The `internal/arch` tests shell out to `go list`, so Go's test cache can't see
36	> edges changing elsewhere in the module. Always run them with `-count=1`	36	> edges changing elsewhere in the module. Always run them with `-count=1`

docs/cert-rotation.md

@@ @@ -0,0 +1,51 @@ @@
+# Server certificate rotation
+The server's QUIC identity is a self-signed ECDSA cert generated on first
+start (`transport.GenerateServerCert`, 2-year validity) and persisted next to
+the database (`server.crt` / `server.key`). Agents trust it by **fingerprint
+pin** carried in the join blob — not by CA path and **not by expiry**: an
+expired cert keeps working for already-enrolled agents. Rotation is therefore
+never an emergency at `NotAfter`; it is hygiene that bounds how long a stolen
+`server.key` stays useful.
+The server logs its cert fingerprint at every startup, and warns — at startup
+and daily thereafter — when the cert is within 90 days of expiry
+(`transport.CertRenewalDue`).
+## Why rotate
+- `server.key` compromise: anyone holding it can impersonate the server to
+  every agent pinning that cert's fingerprint, until every agent re-pins.
+- The validity period (2 years) is the *scheduled* cadence; rotate immediately
+  on suspected key exposure.
+## Ceremony
+Rotating the cert changes its fingerprint, which invalidates every agent's
+pin. Each agent must re-enroll to pick up the new fingerprint.
+. Stop the server.
+. Move the old cert+key aside: `mv server.crt server.crt.old && mv
+   server.key server.key.old` (in the DB directory).
+. Start the server — it generates and persists a fresh 2-year cert and logs
+   the new fingerprint.
+. For each host: mint a join token (`POST /api/v1/enroll-tokens`), copy the
+   `join` blob, run `eitri-agent join <blob>` on the host, **then restart the
+   agent daemon** (e.g. `systemctl restart eitri-agent`). `join` only rewrites
+   the on-disk identity; the running daemon holds its identity in memory and
+   keeps pinning the old fingerprint until restarted. Running VMs are
+   untouched — they survive agent restarts by design, and the agent's
+   reconcile state is independent of its server identity.
+. Delete `server.crt.old` / `server.key.old` once every host has reconnected
+   (watch `GET /api/v1/hosts` for `online: true`).
+Until a host re-enrolls, its agent logs `server cert pin mismatch` and backs
+off — VMs keep running, but the host is dark to the control plane. Rotate
+during a window where that is acceptable, host by host.
+## Out of scope (future)
+- Overlap rotation (serving old+new certs simultaneously) — needs dual-cert
+  listener support.
+- Pushing new pins over the existing authenticated channel (would remove the
+  re-enroll requirement).

docs/credential-revocation.md

@@ @@ -0,0 +1,42 @@ @@
+# Host credential compromise & revocation
+Each agent authenticates its QUIC session with a bearer credential:
+```
+<host_id>.<generation>.<issued_unix>.<hmac-sha256>
+```
+The HMAC (keyed by the server's `host_secret`) covers all three fields. The
+`generation` is compared against the host row's `cred_generation` on every
+Hello **and every report (~10s)**, so revocation takes effect on live
+sessions within one tick.
+## Single credential leaked (the common case)
+```
+curl -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
+  http://server:8080/api/v1/hosts/<host_id>/revoke-credential
+```
+- Bumps that host's generation: the leaked credential is dead fleet-wide
+  within ~10s; every other host is untouched.
+- The host's VMs keep running (the agent reconciles autonomously); the host
+  is dark to the control plane until re-enrolled.
+- Re-enroll: mint a join token, `eitri-agent join <blob>` on the host,
+  restart the agent daemon.
+- The revocation is recorded in the audit log (`host.credential.revoke`).
+## Server `host_secret` leaked (the disaster case)
+Rotate `host_secret` in `server.json` and restart — this revokes **every**
+host credential at once; re-enroll each host as above.
+## Optional max-age
+`credential_max_age` in `server.json` (Go duration, e.g. `"2160h"`) rejects
+credentials older than the window — at Hello and, like revocation, on every
+report tick, so a live session cannot outlive the window. Off by default:
+there is no automatic renewal channel yet, so expiry trades credential
+lifetime against operator toil. Generation revocation is the primary
+mechanism.

docs/openapi.json

Old		New
@@ -1,6 +1,24 @@
1	{	1	{
2	"components": {	2	"components": {
3	"schemas": {	3	"schemas": {
		4	"AuditEvent": {
		5	"properties": {
		6	"action": {
		7	"type": "string"
		8	},
		9	"at": {
		10	"format": "date-time",
		11	"type": "string"
		12	},
		13	"detail": {}
		14	},
		15	"required": [
		16	"action",
		17	"at",
		18	"detail"
		19	],
		20	"type": "object"
		21	},
4	"Capacity": {	22	"Capacity": {
5	"properties": {	23	"properties": {
6	"disk_gb": {	24	"disk_gb": {
@@ -125,11 +143,15 @@
125	},	143	},
126	"EnrollTokenResponse": {	144	"EnrollTokenResponse": {
127	"properties": {	145	"properties": {
		146	"join": {
		147	"type": "string"
		148	},
128	"token": {	149	"token": {
129	"type": "string"	150	"type": "string"
130	}	151	}
131	},	152	},
132	"required": [	153	"required": [
		154	"join",
133	"token"	155	"token"
134	],	156	],
135	"type": "object"	157	"type": "object"
@@ -219,6 +241,17 @@
219	],	241	],
220	"type": "object"	242	"type": "object"
221	},	243	},
		244	"StreamTicketResponse": {
		245	"properties": {
		246	"ticket": {
		247	"type": "string"
		248	}
		249	},
		250	"required": [
		251	"ticket"
		252	],
		253	"type": "object"
		254	},
222	"VM": {	255	"VM": {
223	"properties": {	256	"properties": {
224	"actual_power": {	257	"actual_power": {
@@ -305,6 +338,52 @@
305	},	338	},
306	"openapi": "3.1.0",	339	"openapi": "3.1.0",
307	"paths": {	340	"paths": {
		341	"/api/v1/audit": {
		342	"get": {
		343	"parameters": [
		344	{
		345	"description": "max rows to return (default 100, cap 1000)",
		346	"in": "query",
		347	"name": "limit",
		348	"required": false,
		349	"schema": {
		350	"type": "string"
		351	}
		352	}
		353	],
		354	"responses": {
		355	"200": {
		356	"content": {
		357	"application/json": {
		358	"schema": {
		359	"items": {
		360	"$ref": "#/components/schemas/AuditEvent"
		361	},
		362	"type": "array"
		363	}
		364	}
		365	},
		366	"description": "success"
		367	},
		368	"default": {
		369	"content": {
		370	"text/plain": {
		371	"schema": {
		372	"type": "string"
		373	}
		374	}
		375	},
		376	"description": "error (plain text)"
		377	}
		378	},
		379	"security": [
		380	{
		381	"adminToken": []
		382	}
		383	],
		384	"summary": "Newest audit log rows."
		385	}
		386	},
308	"/api/v1/enroll": {	387	"/api/v1/enroll": {
309	"post": {	388	"post": {
310	"requestBody": {	389	"requestBody": {
@@ -339,7 +418,7 @@
339	"description": "error (plain text)"	418	"description": "error (plain text)"
340	}	419	}
341	},	420	},
342	"summary": "Redeem a one-time enrollment token: a new host joins the fleet and receives its credential. Unauthenticated; the token is the proof."	421	"summary": "Redeem a one-time enrollment token: a new host joins the fleet and receives its credential. Unauthenticated but rate-limited; the token is the proof."
343	}	422	}
344	},	423	},
345	"/api/v1/enroll-tokens": {	424	"/api/v1/enroll-tokens": {
@@ -371,16 +450,16 @@
371	"adminToken": []	450	"adminToken": []
372	}	451	}
373	],	452	],
374	"summary": "Mint a one-time host enrollment token."	453	"summary": "Mint a one-time host enrollment token plus the join blob agents consume."
375	}	454	}
376	},	455	},
377	"/api/v1/events": {	456	"/api/v1/events": {
378	"get": {	457	"get": {
379	"parameters": [	458	"parameters": [
380	{	459	{
381	"description": "admin token",	460	"description": "one-time stream ticket",
382	"in": "query",	461	"in": "query",
383	"name": "token",	462	"name": "ticket",
384	"required": false,	463	"required": false,
385	"schema": {	464	"schema": {
386	"type": "string"	465	"type": "string"
@@ -482,6 +561,73 @@
482	"summary": "Decommission a host: tombstone its VMs and drain gracefully (202)."	561	"summary": "Decommission a host: tombstone its VMs and drain gracefully (202)."
483	}	562	}
484	},	563	},
		564	"/api/v1/hosts/{id}/revoke-credential": {
		565	"post": {
		566	"parameters": [
		567	{
		568	"in": "path",
		569	"name": "id",
		570	"required": true,
		571	"schema": {
		572	"type": "string"
		573	}
		574	}
		575	],
		576	"responses": {
		577	"204": {
		578	"description": "success"
		579	},
		580	"default": {
		581	"content": {
		582	"text/plain": {
		583	"schema": {
		584	"type": "string"
		585	}
		586	}
		587	},
		588	"description": "error (plain text)"
		589	}
		590	},
		591	"security": [
		592	{
		593	"adminToken": []
		594	}
		595	],
		596	"summary": "Revoke a host's outstanding credential by bumping its generation; the host stays dark until re-enrolled."
		597	}
		598	},
		599	"/api/v1/stream-tickets": {
		600	"post": {
		601	"responses": {
		602	"201": {
		603	"content": {
		604	"application/json": {
		605	"schema": {
		606	"$ref": "#/components/schemas/StreamTicketResponse"
		607	}
		608	}
		609	},
		610	"description": "success"
		611	},
		612	"default": {
		613	"content": {
		614	"text/plain": {
		615	"schema": {
		616	"type": "string"
		617	}
		618	}
		619	},
		620	"description": "error (plain text)"
		621	}
		622	},
		623	"security": [
		624	{
		625	"adminToken": []
		626	}
		627	],
		628	"summary": "Mint a one-time short-TTL ticket for the SSE stream — the only credential that ever rides in a URL."
		629	}
		630	},
485	"/api/v1/vms": {	631	"/api/v1/vms": {
486	"get": {	632	"get": {
487	"responses": {	633	"responses": {

docs/shape.html

Old		New
@@ -64,7 +64,8 @@
64	"internal/agent/reconcile",	64	"internal/agent/reconcile",
65	"internal/agent/seed",	65	"internal/agent/seed",
66	"internal/agent/state",	66	"internal/agent/state",
67	"internal/agent/syncclient"	67	"internal/agent/syncclient",
		68	"internal/joinblob"
68	]	69	]
69	},	70	},
70	{	71	{
@@ -80,6 +81,7 @@
80	"plane": "binaries",	81	"plane": "binaries",
81	"synopsis": "eitri-server: single-node control plane (Phase 1: static admin token, no TLS termination here — front with a reverse proxy for TLS).",	82	"synopsis": "eitri-server: single-node control plane (Phase 1: static admin token, no TLS termination here — front with a reverse proxy for TLS).",
82	"imports": [	83	"imports": [
		84	"internal/joinblob",
83	"internal/server/api",	85	"internal/server/api",
84	"internal/server/hub",	86	"internal/server/hub",
85	"internal/server/registry",	87	"internal/server/registry",
@@ -183,6 +185,12 @@
183	"imports": []	185	"imports": []
184	},	186	},
185	{	187	{
		188	"importPath": "internal/joinblob",
		189	"plane": "wire",
		190	"synopsis": "Package joinblob encodes and decodes the single-paste enrollment token (\"join blob\") an agent uses to enroll: it carries the server's HTTP base URL, its QUIC address, a one-shot enrollment token, and the server's TLS cert fingerprint for out-of-band pinning.",
		191	"imports": []
		192	},
		193	{
186	"importPath": "internal/pb",	194	"importPath": "internal/pb",
187	"plane": "wire",	195	"plane": "wire",
188	"synopsis": "",	196	"synopsis": "",
@@ -193,6 +201,7 @@
193	"plane": "control",	201	"plane": "control",
194	"synopsis": "Package api implements the admin REST API and the unauthenticated enrollment endpoint.",	202	"synopsis": "Package api implements the admin REST API and the unauthenticated enrollment endpoint.",
195	"imports": [	203	"imports": [
		204	"internal/joinblob",
196	"internal/server/api/types",	205	"internal/server/api/types",
197	"internal/server/hosttoken",	206	"internal/server/hosttoken",
198	"internal/server/hub",	207	"internal/server/hub",
@@ -218,7 +227,7 @@
218	{	227	{
219	"importPath": "internal/server/hosttoken",	228	"importPath": "internal/server/hosttoken",
220	"plane": "control",	229	"plane": "control",
221	"synopsis": "Package hosttoken mints and verifies host credentials: \"\u003chost_id\u003e.\u003chex hmac-sha256\u003e\".",	230	"synopsis": "Package hosttoken mints and verifies generation-versioned host credentials.",
222	"imports": []	231	"imports": []
223	},	232	},
224	{	233	{

docs/shape.json

Old		New
@@ -13,7 +13,8 @@
13	"internal/agent/reconcile",	13	"internal/agent/reconcile",
14	"internal/agent/seed",	14	"internal/agent/seed",
15	"internal/agent/state",	15	"internal/agent/state",
16	"internal/agent/syncclient"	16	"internal/agent/syncclient",
		17	"internal/joinblob"
17	]	18	]
18	},	19	},
19	{	20	{
@@ -29,6 +30,7 @@
29	"plane": "binaries",	30	"plane": "binaries",
30	"synopsis": "eitri-server: single-node control plane (Phase 1: static admin token, no TLS termination here — front with a reverse proxy for TLS).",	31	"synopsis": "eitri-server: single-node control plane (Phase 1: static admin token, no TLS termination here — front with a reverse proxy for TLS).",
31	"imports": [	32	"imports": [
		33	"internal/joinblob",
32	"internal/server/api",	34	"internal/server/api",
33	"internal/server/hub",	35	"internal/server/hub",
34	"internal/server/registry",	36	"internal/server/registry",
@@ -132,6 +134,12 @@
132	"imports": []	134	"imports": []
133	},	135	},
134	{	136	{
		137	"importPath": "internal/joinblob",
		138	"plane": "wire",
		139	"synopsis": "Package joinblob encodes and decodes the single-paste enrollment token (\"join blob\") an agent uses to enroll: it carries the server's HTTP base URL, its QUIC address, a one-shot enrollment token, and the server's TLS cert fingerprint for out-of-band pinning.",
		140	"imports": []
		141	},
		142	{
135	"importPath": "internal/pb",	143	"importPath": "internal/pb",
136	"plane": "wire",	144	"plane": "wire",
137	"synopsis": "",	145	"synopsis": "",
@@ -142,6 +150,7 @@
142	"plane": "control",	150	"plane": "control",
143	"synopsis": "Package api implements the admin REST API and the unauthenticated enrollment endpoint.",	151	"synopsis": "Package api implements the admin REST API and the unauthenticated enrollment endpoint.",
144	"imports": [	152	"imports": [
		153	"internal/joinblob",
145	"internal/server/api/types",	154	"internal/server/api/types",
146	"internal/server/hosttoken",	155	"internal/server/hosttoken",
147	"internal/server/hub",	156	"internal/server/hub",
@@ -167,7 +176,7 @@
167	{	176	{
168	"importPath": "internal/server/hosttoken",	177	"importPath": "internal/server/hosttoken",
169	"plane": "control",	178	"plane": "control",
170	"synopsis": "Package hosttoken mints and verifies host credentials: \"\u003chost_id\u003e.\u003chex hmac-sha256\u003e\".",	179	"synopsis": "Package hosttoken mints and verifies generation-versioned host credentials.",
171	"imports": []	180	"imports": []
172	},	181	},
173	{	182	{

internal/agent/cloudhv/cloudhv.go

Old		New
@@ -70,10 +70,46 @@ func (p *Provisioner) buildArgs(spec state.VMSpec) []string {
70	}	70	}
71	}	71	}
72		72
		73	// maxDiskGB caps a VM disk at 1 PiB (2^20 GiB) — far beyond any real host,
		74	// and small enough that DiskGB<<30 can never overflow int64 (2^50 max).
		75	const maxDiskGB = 1 << 20
		76
		77	// permanentError marks provisioning failures no retry can fix (the disk
		78	// guards); reconcile checks the Permanent() marker and terminal-fails
		79	// immediately instead of burning its retry budget.
		80	type permanentError struct{ err error }
		81
		82	func (e permanentError) Error() string { return e.err.Error() }
		83	func (e permanentError) Unwrap() error { return e.err }
		84	func (e permanentError) Permanent() bool { return true }
		85
		86	func permanentf(format string, args ...any) error {
		87	return permanentError{err: fmt.Errorf(format, args...)}
		88	}
		89
73	// PrepareDisk creates the VM disk by making a reflink copy of basePath	90	// PrepareDisk creates the VM disk by making a reflink copy of basePath
74	// (instant on XFS/btrfs; silent full-copy fallback on ext4) and then	91	// (instant on XFS/btrfs; silent full-copy fallback on ext4) and then
75	// truncating it to spec.DiskGB gigabytes.	92	// truncating it to spec.DiskGB gigabytes.
		93	//
		94	// truncate -s sets an EXACT size, so a target smaller than the base image
		95	// would silently chop the guest filesystem. PrepareDisk refuses to shrink:
		96	// spec.DiskGB must be in [1, maxDiskGB] and cover the base image. The range
		97	// check runs first so the byte computation is overflow-safe (a naive
		98	// DiskGB<<30 wraps to a small positive value for e.g. 2^34+10, bypassing the
		99	// guard). Pinned by TestPrepareDiskRefusesToShrinkBaseImage and
		100	// TestPrepareDiskShrinkGuardEdgeCases.
76	func (p *Provisioner) PrepareDisk(ctx context.Context, spec state.VMSpec, basePath string) error {	101	func (p *Provisioner) PrepareDisk(ctx context.Context, spec state.VMSpec, basePath string) error {
		102	base, err := os.Stat(basePath)
		103	if err != nil {
		104	return fmt.Errorf("stat base image %s: %w", basePath, err)
		105	}
		106	if spec.DiskGB < 1 \|\| spec.DiskGB > maxDiskGB {
		107	return permanentf("disk_gb %d out of range [1, %d]", spec.DiskGB, int64(maxDiskGB))
		108	}
		109	if targetBytes := spec.DiskGB << 30; targetBytes < base.Size() {
		110	return permanentf("disk_gb %d (%d bytes) is smaller than base image %s (%d bytes) — shrinking would corrupt the guest",
		111	spec.DiskGB, targetBytes, basePath, base.Size())
		112	}
77	diskPath := p.st.DiskPath(spec.VMID)	113	diskPath := p.st.DiskPath(spec.VMID)
78	// Ensure VM directory exists.	114	// Ensure VM directory exists.
79	if err := os.MkdirAll(p.st.VMDir(spec.VMID), 0o700); err != nil {	115	if err := os.MkdirAll(p.st.VMDir(spec.VMID), 0o700); err != nil {
@@ -97,12 +133,18 @@ func (p *Provisioner) pidPath(vmID string) string {
97	// Boot spawns a cloud-hypervisor process for spec. The process is placed in	133	// Boot spawns a cloud-hypervisor process for spec. The process is placed in
98	// its own session (Setsid) so it survives an agent restart. A goroutine calls	134	// its own session (Setsid) so it survives an agent restart. A goroutine calls
99	// cmd.Wait to reap the child when it exits.	135	// cmd.Wait to reap the child when it exits.
100	func (p *Provisioner) Boot(ctx context.Context, vmID string, spec state.VMSpec) error {	136	//
		137	// The ctx parameter is deliberately NOT wired to the process: the VM's
		138	// lifetime must not be tied to the agent's (exec.CommandContext SIGKILLs the
		139	// child on ctx cancel, which would hard-power-off every VM on a graceful
		140	// agent stop). Stopping a VM is exclusively the job of Shutdown/Kill, driven
		141	// by the reconcile loop. Pinned by TestBootedVMSurvivesCtxCancellation.
		142	func (p *Provisioner) Boot(_ context.Context, vmID string, spec state.VMSpec) error {
101	// Remove stale socket from a previous run.	143	// Remove stale socket from a previous run.
102	_ = os.Remove(p.st.SocketPath(vmID))	144	_ = os.Remove(p.st.SocketPath(vmID))
103		145
104	args := p.buildArgs(spec)	146	args := p.buildArgs(spec)
105	cmd := exec.CommandContext(ctx, p.chBin, args...)	147	cmd := exec.Command(p.chBin, args...)
106	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}	148	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
107		149
108	// Guest console output (serial) goes to serial.log (via --serial file=…).	150	// Guest console output (serial) goes to serial.log (via --serial file=…).

internal/agent/cloudhv/cloudhv_test.go

@@ @@ -2,11 +2,15 @@ package cloudhv @@
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+	"github.com/a73x/eitri/internal/agent/state"
+	"github.com/stretchr/testify/assert"
@@ @@ -35,6 +39,18 @@ func TestBuildArgs(t *testing.T) { @@
+	assert.Contains(t, joined, "tap=eit-vm1,mac="+MAC("vm1"))
+// sparseFile creates a sparse file of the given size and returns its path.
+// Sparse: no real disk space is consumed regardless of the nominal size.
+func sparseFile(t *testing.T, size int64) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "base.raw")
+	f, err := os.Create(path)
+	require.NoError(t, err)
+	require.NoError(t, f.Truncate(size))
+	require.NoError(t, f.Close())
+	return path
+}
+func TestPrepareDiskUsesReflinkAndResizes(t *testing.T) {
+	var cmds []string
+	run := func(ctx context.Context, name string, args ...string) (string, error) {
@@ @@ -43,14 +59,91 @@ func TestPrepareDiskUsesReflinkAndResizes(t *testing.T) { @@
+	st, _ := state.Open(t.TempDir())
+	p := New(st, "ch", "fw", run)
+	base := sparseFile(t, 1<<20) // 1 MiB base, well under the 10G target
+	require.NoError(t, p.PrepareDisk(context.Background(),
+		state.VMSpec{VMID: "vm1", DiskGB: 10}, base))
+	joined := strings.Join(cmds, "\n")
+	// reflink=auto: instant on XFS/btrfs, silent full-copy fallback on ext4 (spec)
+	assert.Contains(t, joined, "cp --reflink=auto "+base+" "+st.DiskPath("vm1"))
+	assert.Contains(t, joined, "truncate -s 10G "+st.DiskPath("vm1"))
+// TestPrepareDiskRefusesToShrinkBaseImage pins the never-shrink guard:
+// truncate -s sets an EXACT size, so a DiskGB smaller than the base image
+// would silently corrupt the guest filesystem. PrepareDisk must refuse
+// before running any command.
+func TestPrepareDiskRefusesToShrinkBaseImage(t *testing.T) {
+	var cmds []string
+	run := func(ctx context.Context, name string, args ...string) (string, error) {
+		cmds = append(cmds, name+" "+strings.Join(args, " "))
+		return "", nil
+	}
+	st, _ := state.Open(t.TempDir())
+	p := New(st, "ch", "fw", run)
+	base := sparseFile(t, 2<<30) // sparse 2 GiB base
+	err := p.PrepareDisk(context.Background(),
+		state.VMSpec{VMID: "vm1", DiskGB: 1}, base)
+	require.Error(t, err, "shrinking below the base image must be rejected")
+	assert.Contains(t, err.Error(), "smaller than base image")
+	assert.Empty(t, cmds, "no command may run once the shrink is detected")
+}
+// TestPrepareDiskFailsOnMissingBaseImage: a stat failure on the base image is
+// a real error (cp would fail anyway) and must surface before any command.
+func TestPrepareDiskFailsOnMissingBaseImage(t *testing.T) {
+	var cmds []string
+	run := func(ctx context.Context, name string, args ...string) (string, error) {
+		cmds = append(cmds, name+" "+strings.Join(args, " "))
+		return "", nil
+	}
+	st, _ := state.Open(t.TempDir())
+	p := New(st, "ch", "fw", run)
+	err := p.PrepareDisk(context.Background(),
+		state.VMSpec{VMID: "vm1", DiskGB: 10}, "/nonexistent/base.raw")
+	require.Error(t, err)
+	assert.Empty(t, cmds)
+}
+// TestBootedVMSurvivesCtxCancellation pins the VM-lifetime contract: the
+// cloud-hypervisor process must NOT die when the context passed to Boot is
+// cancelled. The agent's root context is cancelled on every graceful agent
+// stop (SIGINT/SIGTERM in main), and VMs are meant to survive agent restarts
+// (that is why Boot uses Setsid). Killing the VM is exclusively the job of
+// the reconcile Shutdown/Kill path.
+func TestBootedVMSurvivesCtxCancellation(t *testing.T) {
+	st, err := state.Open(t.TempDir())
+	require.NoError(t, err)
+	vmID := "vm-survive-test"
+	// Ensure the VM directory exists (Boot writes ch.log + pidfile inside it).
+	require.NoError(t, st.SaveVM(state.Record{Spec: state.VMSpec{VMID: vmID}}))
+	// Fake cloud-hypervisor: ignores its CLI args and sleeps. exec replaces the
+	// shell, so the pidfile PID is the sleep itself and the cleanup Kill reaps
+	// it directly (no orphaned child if the shell wouldn't exec its tail).
+	fakeCH := filepath.Join(t.TempDir(), "fake-ch")
+	require.NoError(t, os.WriteFile(fakeCH, []byte("#!/bin/sh\nexec sleep 60\n"), 0o755))
+	p := New(st, fakeCH, "fw", nil)
+	ctx, cancel := context.WithCancel(context.Background())
+	require.NoError(t, p.Boot(ctx, vmID, state.VMSpec{VMID: vmID, VCPUs: 1, MemMB: 128}))
+	t.Cleanup(func() { _ = p.Kill(context.Background(), vmID) })
+	require.True(t, p.Running(vmID), "process must be alive right after Boot")
+	cancel()
+	// The process must still be alive well after cancellation. Poll instead of
+	// a single sleep so a kill-on-cancel regression fails fast and reliably
+	// (SIGKILL from exec.CommandContext lands and is reaped within
+	// milliseconds, so 300ms is orders-of-magnitude margin).
+	deadline := time.Now().Add(300 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		require.True(t, p.Running(vmID),
+			"cloud-hypervisor process died after ctx cancellation — VM lifetime must not be tied to the agent's context")
+		time.Sleep(50 * time.Millisecond)
+	}
+}
+// TestShutdownFallsBackToSIGTERMOn500 verifies fix 3: a non-2xx HTTP response
+// from the cloud-hypervisor socket is treated as failure and the SIGTERM
+// fallback path is taken.
@@ @@ -116,3 +209,73 @@ func TestShutdownSucceedsOn204(t *testing.T) { @@
+	p := New(st, "ch", "fw", nil)
+	assert.NoError(t, p.Shutdown(context.Background(), vmID))
+// TestPrepareDiskShrinkGuardEdgeCases pins the guard's arithmetic: an exact
+// fit is allowed (truncate to same size is a no-op), and an absurd DiskGB
+// that would overflow a byte computation (DiskGB<<30) must still be caught —
+// 2^34+10 wraps to a small positive byte count if computed naively.
+func TestPrepareDiskShrinkGuardEdgeCases(t *testing.T) {
+	newP := func(t *testing.T, cmds *[]string) *Provisioner {
+		run := func(ctx context.Context, name string, args ...string) (string, error) {
+			*cmds = append(*cmds, name)
+			return "", nil
+		}
+		st, _ := state.Open(t.TempDir())
+		return New(st, "ch", "fw", run)
+	}
+	t.Run("exact fit is allowed", func(t *testing.T) {
+		var cmds []string
+		p := newP(t, &cmds)
+		base := sparseFile(t, 2<<30) // exactly 2 GiB
+		require.NoError(t, p.PrepareDisk(context.Background(),
+			state.VMSpec{VMID: "vm1", DiskGB: 2}, base))
+		assert.NotEmpty(t, cmds)
+	})
+	t.Run("overflow-sized disk_gb does not bypass the guard", func(t *testing.T) {
+		var cmds []string
+		p := newP(t, &cmds)
+		base := sparseFile(t, 2<<30)
+		// (1<<34)+10 << 30 wraps to +10 GiB... no: ((1<<34)+10)*2^30 mod 2^64
+		// wraps to 10 GiB-ish positive — either way the request is absurd and
+		// must not run cp/truncate with a nonsense size. DiskGB=2^34+10 > any
+		// real disk; the guard must reject or the size math must be exact.
+		err := p.PrepareDisk(context.Background(),
+			state.VMSpec{VMID: "vm1", DiskGB: (1 << 34) + 10}, base)
+		if err == nil {
+			// Accepting it is only sound if the target genuinely covers the
+			// base, which it does mathematically (2^34+10 GiB >> 2 GiB) — but
+			// then truncate would run with a size beyond off_t. Reject instead.
+			t.Fatal("absurd disk_gb accepted; overflow in the guard arithmetic")
+		}
+		assert.Empty(t, cmds, "no command may run for an absurd disk_gb")
+	})
+}
+// TestDiskGuardErrorsArePermanent pins that the never-shrink and range
+// guards mark their errors with the reconcile-consumed Permanent() marker:
+// no retry can ever fix a disk_gb below the base image.
+func TestDiskGuardErrorsArePermanent(t *testing.T) {
+	st, _ := state.Open(t.TempDir())
+	p := New(st, "ch", "fw", func(ctx context.Context, name string, args ...string) (string, error) {
+		return "", nil
+	})
+	isPermanent := func(err error) bool {
+		var m interface{ Permanent() bool }
+		return errors.As(err, &m) && m.Permanent()
+	}
+	base := sparseFile(t, 2<<30)
+	err := p.PrepareDisk(context.Background(), state.VMSpec{VMID: "v", DiskGB: 1}, base)
+	require.Error(t, err)
+	assert.True(t, isPermanent(err), "shrink guard error must be permanent")
+	err = p.PrepareDisk(context.Background(), state.VMSpec{VMID: "v", DiskGB: (1 << 34) + 10}, base)
+	require.Error(t, err)
+	assert.True(t, isPermanent(err), "range guard error must be permanent")
+	err = p.PrepareDisk(context.Background(), state.VMSpec{VMID: "v", DiskGB: 10}, "/nonexistent/base.raw")
+	require.Error(t, err)
+	assert.False(t, isPermanent(err), "stat failure may be transient (NFS blip, cache re-fetch)")
+}

internal/agent/imagecache/imagecache.go

Old		New
@@ -1,6 +1,9 @@
1	// Package imagecache downloads, verifies, and raw-converts base images.	1	// Package imagecache downloads, verifies, and raw-converts base images.
2	// Layout: <dir>/<sha256>.raw — keyed by checksum (spec). LRU eviction is	2	// Layout: <dir>/<sha256>.raw — keyed by checksum (spec). The cache is
3	// deferred (Phase 1 hosts pin one or two images).	3	// size-capped LRU: a hit refreshes the file's mtime, and after every
		4	// successful Ensure the oldest images beyond MaxBytes are evicted (never the
		5	// one just ensured). Eviction is safe for running VMs: PrepareDisk copies
		6	// (reflink) the base, so nothing references it after create.
4	package imagecache	7	package imagecache
5		8
6	import (	9	import (
@@ -9,10 +12,12 @@ import (
9	"encoding/hex"	12	"encoding/hex"
10	"fmt"	13	"fmt"
11	"io"	14	"io"
		15	"log/slog"
12	"net/http"	16	"net/http"
13	"os"	17	"os"
14	"path/filepath"	18	"path/filepath"
15	"regexp"	19	"regexp"
		20	"sort"
16	"time"	21	"time"
17		22
18	"github.com/a73x/eitri/internal/agent/exec"	23	"github.com/a73x/eitri/internal/agent/exec"
@@ -34,6 +39,14 @@ type Cache struct {
34	dir string	39	dir string
35	run exec.Runner	40	run exec.Runner
36	http httpDoer	41	http httpDoer
		42
		43	// MaxBytes caps the summed size of cached images; 0 disables eviction.
		44	// Best-effort: the just-ensured image is never evicted, even if it alone
		45	// exceeds the cap. Accounting uses APPARENT size (fi.Size), which
		46	// overstates sparse raw images — the error direction is safe (evicts too
		47	// eagerly, never too late), but a huge-virtual-size image can pin the
		48	// cache over cap; use block-based accounting if that ever bites.
		49	MaxBytes int64
37	}	50	}
38		51
39	// New returns a cache rooted at dir. Its HTTP client carries a generous timeout	52	// New returns a cache rooted at dir. Its HTTP client carries a generous timeout
@@ -51,6 +64,10 @@ func (c *Cache) Ensure(ctx context.Context, url, sha string) (string, error) {
51		64
52	final := filepath.Join(c.dir, sha+".raw")	65	final := filepath.Join(c.dir, sha+".raw")
53	if _, err := os.Stat(final); err == nil {	66	if _, err := os.Stat(final); err == nil {
		67	// Hit: refresh recency so frequently-used images sort as recent.
		68	now := time.Now()
		69	_ = os.Chtimes(final, now, now)
		70	c.evict(final)
54	return final, nil	71	return final, nil
55	}	72	}
56	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)	73	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
@@ -90,5 +107,49 @@ func (c *Cache) Ensure(ctx context.Context, url, sha string) (string, error) {
90	os.Remove(converting)	107	os.Remove(converting)
91	return "", fmt.Errorf("imagecache: rename to final: %w", err)	108	return "", fmt.Errorf("imagecache: rename to final: %w", err)
92	}	109	}
		110	c.evict(final)
93	return final, nil	111	return final, nil
94	}	112	}
		113
		114	// evict removes least-recently-used .raw images until the cache fits
		115	// MaxBytes, never touching keep (the image the current create is about to
		116	// use). Failures are logged-by-omission best-effort: eviction must never
		117	// fail an Ensure that already succeeded.
		118	func (c *Cache) evict(keep string) {
		119	if c.MaxBytes <= 0 {
		120	return
		121	}
		122	entries, err := filepath.Glob(filepath.Join(c.dir, "*.raw"))
		123	if err != nil {
		124	return
		125	}
		126	type img struct {
		127	path string
		128	size int64
		129	mtime time.Time
		130	}
		131	var imgs []img
		132	var total int64
		133	for _, p := range entries {
		134	fi, err := os.Stat(p)
		135	if err != nil {
		136	continue
		137	}
		138	total += fi.Size()
		139	if p != keep {
		140	imgs = append(imgs, img{p, fi.Size(), fi.ModTime()})
		141	}
		142	}
		143	sort.Slice(imgs, func(i, j int) bool { return imgs[i].mtime.Before(imgs[j].mtime) })
		144	for _, im := range imgs {
		145	if total <= c.MaxBytes {
		146	return
		147	}
		148	if os.Remove(im.path) == nil {
		149	total -= im.size
		150	// The one observable trace of eviction: cap-thrash (an oversized
		151	// image forcing constant re-downloads) is invisible without it.
		152	slog.Info("imagecache: evicted", "path", im.path, "size", im.size)
		153	}
		154	}
		155	}

internal/agent/imagecache/imagecache_test.go

@@ @@ -1,6 +1,7 @@ @@
+package imagecache
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
@@ @@ -10,6 +11,7 @@ import ( @@
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+	"github.com/a73x/eitri/internal/agent/exec"
+	"github.com/stretchr/testify/assert"
@@ @@ -126,3 +128,79 @@ func TestEnsureNoStrayFilesAfterSuccess(t *testing.T) { @@
+	assert.Len(t, entries, 1, "only the final .raw file should remain in cache dir")
+	assert.Equal(t, sum+".raw", entries[0].Name())
+// ensureBytes downloads body into c and returns the cached path.
+func ensureBytes(t *testing.T, c *Cache, body []byte) string {
+	t.Helper()
+	ts, sha := serve(t, body)
+	p, err := c.Ensure(context.Background(), ts.URL+"/img", sha)
+	require.NoError(t, err)
+	return p
+}
+// backdate pushes a file's mtime into the past to force LRU ordering.
+func backdate(t *testing.T, path string, d time.Duration) {
+	t.Helper()
+	old := time.Now().Add(-d)
+	require.NoError(t, os.Chtimes(path, old, old))
+}
+// TestEnsureEvictsLRUBeyondCap pins eviction: with MaxBytes set, the
+// least-recently-used image beyond the cap is removed after a successful
+// Ensure; the freshly-ensured image always survives.
+func TestEnsureEvictsLRUBeyondCap(t *testing.T) {
+	dir := t.TempDir()
+	c := New(dir, fakeRunner(t))
+	c.MaxBytes = 150
+	a := ensureBytes(t, c, bytes.Repeat([]byte("a"), 100))
+	backdate(t, a, time.Hour)
+	b := ensureBytes(t, c, bytes.Repeat([]byte("b"), 100)) // 200 > 150 → evict a
+	_, errA := os.Stat(a)
+	assert.True(t, os.IsNotExist(errA), "LRU image beyond the cap must be evicted")
+	_, errB := os.Stat(b)
+	assert.NoError(t, errB, "just-ensured image must never be evicted")
+}
+// TestEnsureCapZeroDisablesEviction pins the default: MaxBytes 0 keeps
+// everything (existing behavior).
+func TestEnsureCapZeroDisablesEviction(t *testing.T) {
+	dir := t.TempDir()
+	c := New(dir, fakeRunner(t))
+	a := ensureBytes(t, c, bytes.Repeat([]byte("a"), 100))
+	backdate(t, a, time.Hour)
+	_ = ensureBytes(t, c, bytes.Repeat([]byte("b"), 100))
+	_, err := os.Stat(a)
+	assert.NoError(t, err, "no eviction when MaxBytes is 0")
+}
+// TestCacheHitRefreshesRecency pins the LRU signal: a cache hit touches the
+// file so frequently-used images sort as recent.
+func TestCacheHitRefreshesRecency(t *testing.T) {
+	dir := t.TempDir()
+	c := New(dir, fakeRunner(t))
+	body := bytes.Repeat([]byte("a"), 50)
+	a := ensureBytes(t, c, body)
+	backdate(t, a, time.Hour)
+	before, _ := os.Stat(a)
+	_ = ensureBytes(t, c, body) // same sha → cache hit
+	after, err := os.Stat(a)
+	require.NoError(t, err)
+	assert.True(t, after.ModTime().After(before.ModTime()), "hit must refresh mtime")
+}
+// TestEvictionNeverRemovesEnsuredImageEvenOverCap pins the best-effort cap:
+// the just-ensured image survives even when it alone exceeds MaxBytes.
+func TestEvictionNeverRemovesEnsuredImageEvenOverCap(t *testing.T) {
+	dir := t.TempDir()
+	c := New(dir, fakeRunner(t))
+	c.MaxBytes = 10
+	a := ensureBytes(t, c, bytes.Repeat([]byte("a"), 100))
+	_, err := os.Stat(a)
+	assert.NoError(t, err, "cap is best-effort; the image in use survives")
+}

internal/agent/netenv/netenv.go

Old		New
@@ -7,6 +7,9 @@ import (
7	"context"	7	"context"
8	"fmt"	8	"fmt"
9	"net/netip"	9	"net/netip"
		10	"os"
		11	"path/filepath"
		12	"strconv"
10	"strings"	13	"strings"
11		14
12	"github.com/a73x/eitri/internal/agent/exec"	15	"github.com/a73x/eitri/internal/agent/exec"
@@ -20,6 +23,10 @@ const Bridge = "eitri0"
20	type Net struct {	23	type Net struct {
21	run exec.Runner	24	run exec.Runner
22	cidr netip.Prefix	25	cidr netip.Prefix
		26	// isTap reports whether an existing link is a tun/tap device. Injectable
		27	// for tests; production checks /sys/class/net/<name>/tun_flags, which
		28	// exists only for tun/tap links — no error-string parsing.
		29	isTap func(name string) bool
23	}	30	}
24		31
25	// New constructs a Net. cidr must be a valid IPv4 prefix (e.g. "10.77.1.0/24").	32	// New constructs a Net. cidr must be a valid IPv4 prefix (e.g. "10.77.1.0/24").
@@ -31,9 +38,38 @@ func New(run exec.Runner, cidr string) (*Net, error) {
31	if !p.Addr().Is4() {	38	if !p.Addr().Is4() {
32	return nil, fmt.Errorf("bridge CIDR must be IPv4, got %s", cidr)	39	return nil, fmt.Errorf("bridge CIDR must be IPv4, got %s", cidr)
33	}	40	}
34	return &Net{run: run, cidr: p}, nil	41	return &Net{run: run, cidr: p, isTap: sysfsIsTap}, nil
35	}	42	}
36		43
		44	// sysfsIsTap reports whether name is an L2 TAP link: /sys/class/net/<name>/
		45	// tun_flags exists exactly for tun-driver links, and bit 0x0002 (IFF_TAP)
		46	// distinguishes a TAP from an L3 TUN (which cannot join a bridge and would
		47	// fail later with a raw RTNETLINK error).
		48	func sysfsIsTap(name string) bool { return sysfsIsTapAt("/sys/class/net", name) }
		49
		50	// sysfsIsTapAt is sysfsIsTap with an injectable sysfs root (for tests).
		51	func sysfsIsTapAt(root, name string) bool {
		52	raw, err := os.ReadFile(filepath.Join(root, name, "tun_flags"))
		53	if err != nil {
		54	return false
		55	}
		56	flags, err := strconv.ParseUint(strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(string(raw)), "0x")), 16, 64)
		57	if err != nil {
		58	return false
		59	}
		60	const iffTap = 0x0002
		61	return flags&iffTap != 0
		62	}
		63
		64	// tapConflictError marks the name-collision failure permanent: a foreign
		65	// device squatting on the tap name never resolves by retrying (reconcile's
		66	// Permanent() marker — burns the budget at once when that lands).
		67	type tapConflictError struct{ err error }
		68
		69	func (e tapConflictError) Error() string { return e.err.Error() }
		70	func (e tapConflictError) Unwrap() error { return e.err }
		71	func (e tapConflictError) Permanent() bool { return true }
		72
37	// Gateway returns the host-side IP (.1) on the bridge, as a bare address string.	73	// Gateway returns the host-side IP (.1) on the bridge, as a bare address string.
38	func (n *Net) Gateway() string { return n.cidr.Masked().Addr().Next().String() }	74	func (n *Net) Gateway() string { return n.cidr.Masked().Addr().Next().String() }
39		75
@@ -161,9 +197,23 @@ func (n *Net) EnsureBridge(ctx context.Context, noMasqIfaces []string) error {
161		197
162	// CreateTap creates a TAP device and attaches it to the eitri0 bridge.	198	// CreateTap creates a TAP device and attaches it to the eitri0 bridge.
163	// tap should be the value from state.TapName(vmID).	199	// tap should be the value from state.TapName(vmID).
		200	//
		201	// Idempotent by existence check, not error-string parsing (same rationale as
		202	// EnsureBridge): re-running `ip tuntap add` on a live tap fails with
		203	// "ioctl(TUNSETIFF): Device or resource busy" — a message the tolerated()
		204	// allow-list can never chase across iproute2 versions. A create-retry hitting
		205	// the previous attempt's tap must not mask the retry's real error. Pinned by
		206	// TestCreateTapIdempotentWhenTapExists.
164	func (n *Net) CreateTap(ctx context.Context, tap string) error {	207	func (n *Net) CreateTap(ctx context.Context, tap string) error {
165	if _, err := n.best(ctx, "ip", "tuntap", "add", "dev", tap, "mode", "tap"); err != nil {	208	if _, err := n.run(ctx, "ip", "link", "show", "dev", tap); err != nil {
166	return err	209	if _, err := n.best(ctx, "ip", "tuntap", "add", "dev", tap, "mode", "tap"); err != nil {
		210	return err
		211	}
		212	} else if !n.isTap(tap) {
		213	// Name collision or stale device: silently enslaving a non-tap would
		214	// surface later as an illegible cloud-hypervisor failure. Fail loudly
		215	// here instead (pinned by TestCreateTapRejectsNonTapDevice).
		216	return tapConflictError{err: fmt.Errorf("link %s exists but is not a TAP device — name collision or stale interface; remove it or rename the VM", tap)}
167	}	217	}
168	if _, err := n.best(ctx, "ip", "link", "set", tap, "master", Bridge); err != nil {	218	if _, err := n.best(ctx, "ip", "link", "set", tap, "master", Bridge); err != nil {
169	return err	219	return err

internal/agent/netenv/netenv_test.go

@@ @@ -3,6 +3,8 @@ package netenv @@
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
@@ @@ -89,7 +91,11 @@ func TestEnsureBridgeIdempotentOnRestart(t *testing.T) { @@
+func TestTapLifecycle(t *testing.T) {
+	// Fresh tap: the existence probe fails, so the add runs.
+	errs := map[string]error{
+		"ip link show dev eit-abc123": errors.New(`Device "eit-abc123" does not exist.`),
+	}
+	run, calls := recorder(nil, errs)
+	n, _ := New(run, "10.77.1.0/24")
+	require.NoError(t, n.CreateTap(context.Background(), "eit-abc123"))
+	all := joinCalls(calls)
@@ @@ -213,7 +219,12 @@ func TestCreateTapToleratesAlreadyExists(t *testing.T) { @@
+			key := "ip tuntap add dev eit-x mode tap"
+			run, _ := recorder(
+				map[string]string{key: tc.out},
+				map[string]error{
+					key: errors.New(tc.errText),
+					// TOCTOU window: probe says absent, add still races an
+					// exists-style failure — the tolerance must still apply.
+					"ip link show dev eit-x": errors.New(`Device "eit-x" does not exist.`),
+				},
+			n, _ := New(run, "10.77.1.0/24")
+			require.NoError(t, n.CreateTap(context.Background(), "eit-x"),
@@ @@ -225,7 +236,10 @@ func TestCreateTapToleratesAlreadyExists(t *testing.T) { @@
+// CreateTap must still surface a genuine (non-tolerated) error.
+func TestCreateTapPropagatesRealError(t *testing.T) {
+	key := "ip tuntap add dev eit-x mode tap"
+	run, _ := recorder(nil, map[string]error{
+		key:                      errors.New("Operation not permitted"),
+		"ip link show dev eit-x": errors.New(`Device "eit-x" does not exist.`),
+	})
+	n, _ := New(run, "10.77.1.0/24")
+	err := n.CreateTap(context.Background(), "eit-x")
+	require.Error(t, err)
@@ @@ -248,3 +262,105 @@ func TestDeleteTapToleratesMissingDevice(t *testing.T) { @@
+	n3, _ := New(run3, "10.77.1.0/24")
+	require.Error(t, n3.DeleteTap(context.Background(), "eit-x"))
+// TestCreateTapIdempotentWhenTapExists is the regression test for the sandbox
+// failure: a create-retry re-ran CreateTap on a tap left by the previous
+// attempt (UP, enslaved to eitri0), and `ip tuntap add` failed with
+// "ioctl(TUNSETIFF): Device or resource busy" — which is NOT in the tolerated
+// string list, so the retry's real root-cause error (disk guard) was masked
+// by a tap artifact. Like EnsureBridge, CreateTap must existence-check the
+// link and skip the add instead of parsing error strings.
+func TestCreateTapIdempotentWhenTapExists(t *testing.T) {
+	out := map[string]string{
+		"ip link show dev eit-busy": "4: eit-busy: <NO-CARRIER,BROADCAST,MULTICAST,UP> master eitri0 state DOWN",
+	}
+	// If the code ever regresses to an unconditional add, fail it the
+	// real-world way.
+	errs := map[string]error{
+		"ip tuntap add dev eit-busy mode tap": errors.New("ioctl(TUNSETIFF): Device or resource busy"),
+	}
+	run, calls := recorder(out, errs)
+	n, err := New(run, "10.77.1.0/24")
+	require.NoError(t, err)
+	n.isTap = func(string) bool { return true } // the leftover IS a real tap
+	require.NoError(t, n.CreateTap(context.Background(), "eit-busy"),
+		"CreateTap on an existing tap must succeed")
+	all := joinCalls(calls)
+	assert.NotContains(t, all, "ip tuntap add", "tap exists → no add")
+	assert.Contains(t, all, "ip link set eit-busy master eitri0", "enslave stays (idempotent)")
+	assert.Contains(t, all, "ip link set eit-busy up", "up stays (idempotent)")
+}
+// TestCreateTapRejectsNonTapDevice pins the collision guard: when a link
+// with the tap's name exists but is NOT a tap (stale dummy/veth or a name
+// collision), CreateTap must fail loudly naming the conflict — silently
+// enslaving it would surface later as an illegible cloud-hypervisor error.
+func TestCreateTapRejectsNonTapDevice(t *testing.T) {
+	out := map[string]string{
+		"ip link show dev eit-clash": "5: eit-clash: <BROADCAST> state DOWN", // exists
+	}
+	run, calls := recorder(out, nil)
+	n, err := New(run, "10.77.1.0/24")
+	require.NoError(t, err)
+	n.isTap = func(tap string) bool { return false } // not a tun/tap device
+	err = n.CreateTap(context.Background(), "eit-clash")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "not a TAP device")
+	var p interface{ Permanent() bool }
+	assert.True(t, errors.As(err, &p) && p.Permanent(), "conflict never self-resolves — must be permanent")
+	all := joinCalls(calls)
+	assert.NotContains(t, all, "master", "must not enslave a conflicting device")
+	assert.NotContains(t, all, "ip tuntap add", "guard path must not attempt a create")
+}
+// TestCreateTapAcceptsExistingRealTap pins the counterpart: an existing
+// genuine tap passes the type check and is (idempotently) enslaved.
+func TestCreateTapAcceptsExistingRealTap(t *testing.T) {
+	out := map[string]string{
+		"ip link show dev eit-ok": "5: eit-ok: <NO-CARRIER,BROADCAST,MULTICAST,UP> master eitri0",
+	}
+	run, calls := recorder(out, nil)
+	n, err := New(run, "10.77.1.0/24")
+	require.NoError(t, err)
+	n.isTap = func(tap string) bool { return true }
+	require.NoError(t, n.CreateTap(context.Background(), "eit-ok"))
+	all := joinCalls(calls)
+	assert.NotContains(t, all, "ip tuntap add")
+	assert.Contains(t, all, "ip link set eit-ok master eitri0")
+}
+// TestSysfsIsTap exercises the production sysfs probe against a fabricated
+// /sys-like tree: an L2 TAP (IFF_TAP set), an L3 TUN (bit clear), a
+// non-tun-driver link (no tun_flags), and an absent link.
+func TestSysfsIsTap(t *testing.T) {
+	// The probe reads /sys/class/net/<name>/tun_flags; point it at a temp
+	// tree via a tiny indirection so the test needs no root or real devices.
+	root := t.TempDir()
+	write := func(name, flags string) {
+		dir := root + "/" + name
+		require.NoError(t, os.MkdirAll(dir, 0o755))
+		require.NoError(t, os.WriteFile(dir+"/tun_flags", []byte(flags), 0o644))
+	}
+	write("eit-tap", "0x0002")                           // IFF_TAP
+	write("eit-tun", "0x0001")                           // IFF_TUN (L3)
+	require.NoError(t, os.MkdirAll(root+"/eth0", 0o755)) // no tun_flags
+	isTap := func(name string) bool { return sysfsIsTapAt(root, name) }
+	assert.True(t, isTap("eit-tap"), "IFF_TAP link is a tap")
+	assert.False(t, isTap("eit-tun"), "IFF_TUN link is not an L2 tap")
+	assert.False(t, isTap("eth0"), "non-tun link is not a tap")
+	assert.False(t, isTap("absent"), "missing link is not a tap")
+}
+// TestTapConflictErrorIsPermanent pins the marker + unwrap on the conflict error.
+func TestTapConflictErrorIsPermanent(t *testing.T) {
+	inner := fmt.Errorf("boom")
+	e := tapConflictError{err: inner}
+	assert.Equal(t, "boom", e.Error())
+	assert.Equal(t, inner, e.Unwrap())
+	var p interface{ Permanent() bool }
+	assert.True(t, errors.As(error(e), &p) && p.Permanent())
+}

internal/agent/reconcile/reconcile.go

Old		New
@@ -19,6 +19,7 @@ package reconcile
19	import (	19	import (
20	"context"	20	"context"
21	"encoding/json"	21	"encoding/json"
		22	"errors"
22	"time"	23	"time"
23		24
24	"github.com/a73x/eitri/internal/agent/seed"	25	"github.com/a73x/eitri/internal/agent/seed"
@@ -77,6 +78,17 @@ type Engine struct {
77		78
78	// MaxCreateAttempts is the maximum number of create attempts before terminal failed.	79	// MaxCreateAttempts is the maximum number of create attempts before terminal failed.
79	MaxCreateAttempts int	80	MaxCreateAttempts int
		81
		82	// StepTimeout bounds one WHOLE Step call — the sum of all operations for
		83	// all VMs in that step, not each operation. Zero disables the watchdog.
		84	// A wedged operation (disk prep, seed build, image fetch) then fails with
		85	// the ctx error instead of freezing the reconcile loop forever; the next
		86	// tick retries, and ctx-expiry failures are refunded from the create retry
		87	// budget (see failCreate). Convergence across ticks is guaranteed because
		88	// completed image downloads are durably cached per-sha. Set comfortably
		89	// above the longest legitimate operation (imagecache's HTTP client allows
		90	// 10m for a first-time image download).
		91	StepTimeout time.Duration
80	}	92	}
81		93
82	// Step reconciles the desired snapshot against actual state and returns an	94	// Step reconciles the desired snapshot against actual state and returns an
@@ -85,6 +97,15 @@ type Engine struct {
85	// The algorithm is level-triggered: every call re-examines full state and	97	// The algorithm is level-triggered: every call re-examines full state and
86	// drives toward desired. Idempotent under repeated identical snapshots.	98	// drives toward desired. Idempotent under repeated identical snapshots.
87	func (e Engine) Step(ctx context.Context, snap pb.DesiredStateSnapshot) *pb.ActualStateReport {	99	func (e Engine) Step(ctx context.Context, snap pb.DesiredStateSnapshot) *pb.ActualStateReport {
		100	// Watchdog: bound the whole step so one wedged operation cannot freeze the
		101	// reconcile loop (and with it all reports for this host) forever. Pinned by
		102	// TestStepTimeoutBoundsSlowOperations.
		103	if e.StepTimeout > 0 {
		104	var cancel context.CancelFunc
		105	ctx, cancel = context.WithTimeout(ctx, e.StepTimeout)
		106	defer cancel()
		107	}
		108
88	rep := &pb.ActualStateReport{}	109	rep := &pb.ActualStateReport{}
89		110
90	// ── 1. Epoch fence ───────────────────────────────────────────────────────	111	// ── 1. Epoch fence ───────────────────────────────────────────────────────
@@ -183,6 +204,11 @@ func (e Engine) Step(ctx context.Context, snap pb.DesiredStateSnapshot) *pb.Ac
183		204
184	if now.Sub(*rec.QuarantinedAt) >= grace {	205	if now.Sub(*rec.QuarantinedAt) >= grace {
185	// Grace expired: destroy the VM.	206	// Grace expired: destroy the VM.
		207	// NOTE: this may run with an expired step ctx (watchdog). Safe today
		208	// because cloudhv.Kill ignores ctx (SIGKILL via pidfile) and
		209	// Shutdown falls back to a ctx-free SIGTERM — a future Provisioner
		210	// that honors ctx here would skip the destroy until a later tick,
		211	// which the level-triggered loop tolerates but delays.
186	_ = e.Prov.Kill(ctx, id)	212	_ = e.Prov.Kill(ctx, id)
187	_ = e.Net.DeleteTap(ctx, state.TapName(id))	213	_ = e.Net.DeleteTap(ctx, state.TapName(id))
188	_ = e.St.DeleteVM(id)	214	_ = e.St.DeleteVM(id)
@@ -228,6 +254,15 @@ func (e Engine) Step(ctx context.Context, snap pb.DesiredStateSnapshot) *pb.Ac
228	// create attempts to create a new VM from desired state d.	254	// create attempts to create a new VM from desired state d.
229	// rec is the existing (potentially stale) record, ok indicates whether one exists.	255	// rec is the existing (potentially stale) record, ok indicates whether one exists.
230	func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record, ok bool, rep *pb.ActualStateReport) {	256	func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record, ok bool, rep *pb.ActualStateReport) {
		257	// A fired step watchdog is the STEP's failure, not this VM's: don't start
		258	// an attempt (which would burn retry budget) with a dead context. Report
		259	// and let the next tick — with a fresh budget — do the work. Pinned by
		260	// TestWatchdogExpiryDoesNotBurnCreateAttempts.
		261	if err := ctx.Err(); err != nil {
		262	addReport(rep, d.VmId, rec.IP, "stopped", "creating", "step timeout: "+err.Error())
		263	return
		264	}
		265
231	// Fix 2: if the desired spec differs from the stored spec, the user edited the	266	// Fix 2: if the desired spec differs from the stored spec, the user edited the
232	// VM definition. Reset CreateAttempts so the new spec gets a fresh retry budget	267	// VM definition. Reset CreateAttempts so the new spec gets a fresh retry budget
233	// instead of being permanently terminal-failed due to the old spec's failures.	268	// instead of being permanently terminal-failed due to the old spec's failures.
@@ -328,8 +363,31 @@ func (e Engine) create(ctx context.Context, d pb.VMDesired, rec state.Record,
328	addReport(rep, d.VmId, rec.IP, power, "ready", "")	363	addReport(rep, d.VmId, rec.IP, power, "ready", "")
329	}	364	}
330		365
331	// failCreate records a failed create attempt and appends a report row.	366	// permanent reports whether err (anywhere in its chain) carries the
		367	// consumer-owned permanence marker — the provisioner's way of saying no
		368	// retry can ever succeed (e.g. the disk-shrink guard). Consumer-side
		369	// interface per the R5 convention: reconcile declares it, cloudhv implements.
		370	func permanent(err error) bool {
		371	var p interface{ Permanent() bool }
		372	return errors.As(err, &p) && p.Permanent()
		373	}
		374
		375	// failCreate records a failed create attempt and appends a report row. When
		376	// the step context has expired, the failure belongs to the watchdog, not the
		377	// VM: the attempt is refunded so wedged steps can never drive a healthy VM to
		378	// terminal failed (see TestWatchdogExpiryDoesNotBurnCreateAttempts). A
		379	// Permanent() error spends the whole budget at once — retrying a permanent
		380	// misconfiguration only wastes tap/image work across three ticks (pinned by
		381	// TestPermanentCreateErrorFailsTerminallyInOneAttempt). Ordering: the ctx
		382	// refund wins over permanence — a permanent error surfacing under an expired
		383	// ctx is refunded this tick and, being deterministic, terminal-fails on the
		384	// next tick's fresh ctx. Keeps the watchdog invariant unconditional.
332	func (e Engine) failCreate(ctx context.Context, rec state.Record, err error, rep pb.ActualStateReport) {	385	func (e Engine) failCreate(ctx context.Context, rec state.Record, err error, rep pb.ActualStateReport) {
		386	if ctx.Err() != nil {
		387	rec.CreateAttempts-- // refund: the step died mid-operation
		388	} else if permanent(err) {
		389	rec.CreateAttempts = e.MaxCreateAttempts // terminal now; retry cannot succeed
		390	}
333	rec.LastError = err.Error()	391	rec.LastError = err.Error()
334	_ = e.St.SaveVM(rec)	392	_ = e.St.SaveVM(rec)
335		393
@@ -362,8 +420,16 @@ func (e Engine) converge(ctx context.Context, d pb.VMDesired, rec state.Record
362		420
363	// Persistent lost VM.	421	// Persistent lost VM.
364	if d.PowerState == "running" {	422	if d.PowerState == "running" {
365	// Restart: tap dies on reboot, recreate it.	423	// Restart: tap dies on reboot, recreate it. A CreateTap failure
366	_ = e.Net.CreateTap(ctx, state.TapName(d.VmId))	424	// (e.g. a foreign device squatting on the name) is reported with
		425	// ITS message — letting Boot fail instead yields an illegible
		426	// cloud-hypervisor error for the same root cause.
		427	if err := e.Net.CreateTap(ctx, state.TapName(d.VmId)); err != nil {
		428	rec.LastError = err.Error()
		429	_ = e.St.SaveVM(rec)
		430	addReport(rep, d.VmId, rec.IP, "stopped", "failed", rec.LastError)
		431	return
		432	}
367	if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {	433	if err := e.Prov.Boot(ctx, d.VmId, rec.Spec); err != nil {
368	rec.LastError = err.Error()	434	rec.LastError = err.Error()
369	_ = e.St.SaveVM(rec)	435	_ = e.St.SaveVM(rec)

internal/agent/reconcile/reconcile_test.go

@@ @@ -2,6 +2,7 @@ package reconcile @@
+import (
+	"context"
+	"errors"
+	"net/netip"
+	"testing"
+	"time"
@@ @@ -17,18 +18,20 @@ import ( @@
+// ---- fakes ----
+type fakeProv struct {
+	running   map[string]bool
+	prepCalls int // total PrepareDisk invocations, including failed ones
+	prepared  []string
+	booted    []string
+	shutdown  []string
+	killed    []string
+	prepErr   error
+	bootErr   error // one-shot: consumed and cleared on first Boot call
+func newFakeProv() *fakeProv { return &fakeProv{running: map[string]bool{}} }
+func (f *fakeProv) PrepareDisk(_ context.Context, s state.VMSpec, _ string) error {
+	f.prepCalls++ // counted BEFORE the error short-circuit: total invocations
+	if f.prepErr != nil {
+		return f.prepErr
@@ @@ -63,7 +66,10 @@ type fakeNet struct { @@
+	cidr    string
+func (f *fakeNet) CreateTap(_ context.Context, t string) error {
+	f.taps = append(f.taps, t)
+	return nil
+}
+func (f *fakeNet) DeleteTap(_ context.Context, t string) error {
+	f.deleted = append(f.deleted, t)
+	return nil
+	// The quarantined VM must NOT appear in rep.Vms (it is quarantined, not active).
+	assert.Nil(t, findVM(rep, "vm1"), "quarantined VM must not appear in Vms on fence path")
+// TestStepTimeoutBoundsSlowOperations pins the Step watchdog: a wedged
+// operation inside Step (here: an image fetch that never returns until its
+// context is cancelled) must not block the reconcile loop forever. With
+// StepTimeout set, Step's context expires, the create fails with the ctx
+// error, and Step returns so the next tick can retry.
+func TestStepTimeoutBoundsSlowOperations(t *testing.T) {
+	f := setup(t)
+	f.eng.StepTimeout = 50 * time.Millisecond
+	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {
+		<-ctx.Done() // wedged until the watchdog fires
+		return "", ctx.Err()
+	}
+	done := make(chan *pb.ActualStateReport, 1)
+	go func() { done <- f.eng.Step(context.Background(), snap(1, vm("vm1"))) }()
+	select {
+	case rep := <-done:
+		row := findVM(rep, "vm1")
+		require.NotNil(t, row)
+		assert.Equal(t, "creating", row.Phase, "first failed attempt stays in creating (retry budget)")
+		assert.Contains(t, row.LastError, "context deadline exceeded")
+	case <-time.After(2 * time.Second):
+		t.Fatal("Step did not return: a wedged operation blocked the reconcile loop (no watchdog)")
+	}
+}
+// TestStepTimeoutZeroDisablesWatchdog: the zero value must not impose any
+// deadline (all pre-existing behavior and tests rely on unbounded Step).
+func TestStepTimeoutZeroDisablesWatchdog(t *testing.T) {
+	f := setup(t)
+	sawDeadline := false
+	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {
+		_, sawDeadline = ctx.Deadline()
+		return "/cache/x.raw", nil
+	}
+	rep := f.eng.Step(context.Background(), snap(1, vm("vm1")))
+	require.NotNil(t, findVM(rep, "vm1"))
+	assert.False(t, sawDeadline, "StepTimeout==0 must not set a deadline")
+}
+// TestWatchdogExpiryDoesNotBurnCreateAttempts pins that a fired step watchdog
+// is the STEP's failure, not any VM's: records must not lose retry budget when
+// the ctx has expired (three wedged steps would otherwise terminal-fail
+// perfectly healthy VMs — including ones that never got a turn because an
+// earlier VM in the randomized map order consumed the whole budget).
+func TestWatchdogExpiryDoesNotBurnCreateAttempts(t *testing.T) {
+	f := setup(t)
+	f.eng.StepTimeout = 50 * time.Millisecond
+	f.eng.Images = func(ctx context.Context, url, sha string) (string, error) {
+		<-ctx.Done() // every fetch wedges until the watchdog fires
+		return "", ctx.Err()
+	}
+	_ = f.eng.Step(context.Background(), snap(1, vm("vm1"), vm("vm2")))
+	recs, err := f.st.LoadVMs()
+	require.NoError(t, err)
+	for id, rec := range recs {
+		assert.Zero(t, rec.CreateAttempts,
+			"vm %s: watchdog expiry must not burn the retry budget", id)
+	}
+}
+// permErr is a test error carrying the consumer-owned permanence marker.
+type permErr struct{ msg string }
+func (e permErr) Error() string   { return e.msg }
+func (e permErr) Permanent() bool { return true }
+// TestPermanentCreateErrorFailsTerminallyInOneAttempt pins fast-fail: an
+// error marked Permanent() must not burn the retry budget across ticks —
+// the first attempt goes straight to terminal failed, and subsequent steps
+// do not retry the provisioner.
+func TestPermanentCreateErrorFailsTerminallyInOneAttempt(t *testing.T) {
+	f := setup(t)
+	f.prov.prepErr = permErr{"disk_gb 1 is smaller than base image"}
+	rep := f.eng.Step(context.Background(), snap(1, vm("vm1")))
+	row := findVM(rep, "vm1")
+	require.NotNil(t, row)
+	assert.Equal(t, "failed", row.Phase, "permanent error must be terminal on attempt 1")
+	assert.Contains(t, row.LastError, "smaller than base image")
+	require.Equal(t, 1, f.prov.prepCalls, "exactly one PrepareDisk invocation")
+	rep = f.eng.Step(context.Background(), snap(1, vm("vm1")))
+	row = findVM(rep, "vm1")
+	require.NotNil(t, row)
+	assert.Equal(t, "failed", row.Phase)
+	assert.Equal(t, 1, f.prov.prepCalls, "no further provisioner attempts after a permanent failure")
+}
+// TestTransientCreateErrorStillRetries pins the counterpart: unmarked errors
+// keep the existing bounded-retry behavior (phase creating until the budget
+// is spent).
+func TestTransientCreateErrorStillRetries(t *testing.T) {
+	f := setup(t)
+	f.prov.prepErr = errors.New("cp: reflink failed") // unmarked -> transient
+	for i := 1; i < f.eng.MaxCreateAttempts; i++ {
+		rep := f.eng.Step(context.Background(), snap(1, vm("vm1")))
+		row := findVM(rep, "vm1")
+		require.NotNil(t, row)
+		assert.Equal(t, "creating", row.Phase, "attempt %d stays in the retry budget", i)
+	}
+	rep := f.eng.Step(context.Background(), snap(1, vm("vm1")))
+	row := findVM(rep, "vm1")
+	require.NotNil(t, row)
+	assert.Equal(t, "failed", row.Phase, "budget spent -> terminal failed")
+	assert.Equal(t, f.eng.MaxCreateAttempts, f.prov.prepCalls, "one invocation per budgeted attempt")
+}

internal/agent/syncclient/client_test.go

Old		New
@@ -81,7 +81,7 @@ func (h *serverHarness) start(addr string) {
81	h.addr = lis.Addr().String()	81	h.addr = lis.Addr().String()
82	ctx, cancel := context.WithCancel(context.Background())	82	ctx, cancel := context.WithCancel(context.Background())
83	h.cancel = cancel	83	h.cancel = cancel
84	svc := syncsvc.New(h.st, h.reg, h.hub, h.secret)	84	svc := syncsvc.New(h.st, h.reg, h.hub, h.secret, 0)
85	go svc.Serve(ctx, lis) //nolint:errcheck	85	go svc.Serve(ctx, lis) //nolint:errcheck
86	}	86	}
87		87
@@ -97,9 +97,9 @@ func (h *serverHarness) stop() {
97	// enroll creates a host and returns a valid credential for it.	97	// enroll creates a host and returns a valid credential for it.
98	func (h *serverHarness) enroll() (hostID, cred string) {	98	func (h *serverHarness) enroll() (hostID, cred string) {
99	tok, _ := h.st.CreateEnrollmentToken()	99	tok, _ := h.st.CreateEnrollmentToken()
100	host, err := h.st.RedeemEnrollmentToken(tok, "h", "linux", "amd64", "cloudhv", "")	100	host, err := h.st.RedeemEnrollmentToken(tok, "h", "linux", "amd64", "cloudhv", "", "")
101	require.NoError(h.t, err)	101	require.NoError(h.t, err)
102	return host.ID, hosttoken.Mint(h.secret, host.ID)	102	return host.ID, hosttoken.Mint(h.secret, host.ID, host.CredGeneration, time.Now())
103	}	103	}
104		104
105	func newClient(t testing.T, addr, fp, hostID, cred string) Client {	105	func newClient(t testing.T, addr, fp, hostID, cred string) Client {

internal/arch/arch_test.go

Old		New
@@ -154,18 +154,14 @@ func TestDomainDoesNotImportTransportStack(t *testing.T) {
154	}	154	}
155		155
156	// R2: the server is a pure control plane. It expresses intent as desired state	156	// R2: the server is a pure control plane. It expresses intent as desired state
157	// and never actuates VMs itself, so no package under internal/server may shell	157	// and never actuates VMs itself, so no package under internal/server may reach
158	// out. A direct import of os/exec is the canonical signal of a violation.	158	// os/exec — directly OR through an internal wrapper package (checking only
		159	// direct imports would let internal/util/somewrapper smuggle a shell-out in).
159	func TestServerNeverShellsOut(t *testing.T) {	160	func TestServerNeverShellsOut(t *testing.T) {
160	g := directImports(t)	161	g := directImports(t)
161	for pkg, deps := range g {	162	for pkg, offenders := range execViolations(g, module, "internal/server/", nil) {
162	if !has(pkg, "internal/server/") {	163	for _, o := range offenders {
163	continue	164	t.Errorf("control-plane package %s reaches os/exec via %s — the server never shells out", short(pkg), short(o))
164	}
165	for _, d := range deps {
166	if d == "os/exec" {
167	t.Errorf("control-plane package %s must not import os/exec — the server never shells out", short(pkg))
168	}
169	}	165	}
170	}	166	}
171	}	167	}
@@ -173,20 +169,16 @@ func TestServerNeverShellsOut(t *testing.T) {
173	// R6: all external process execution in the data plane funnels through	169	// R6: all external process execution in the data plane funnels through
174	// agent/exec.Runner, which keeps the host-touching packages mockable and	170	// agent/exec.Runner, which keeps the host-touching packages mockable and
175	// auditable. The sole exception is agent/cloudhv, which launches the	171	// auditable. The sole exception is agent/cloudhv, which launches the
176	// long-lived cloud-hypervisor process directly (exec.CommandContext) rather	172	// long-lived cloud-hypervisor process directly (exec.Command) rather than
177	// than through the one-shot Runner. Every other agent package must use Runner	173	// through the one-shot Runner. Every other agent package must use Runner and
178	// and must not import os/exec.	174	// must not reach os/exec — directly or through an internal wrapper (reaching
		175	// it via the sanctioned cloudhv is fine).
179	func TestOnlyCloudhvImportsOsExecInDataPlane(t *testing.T) {	176	func TestOnlyCloudhvImportsOsExecInDataPlane(t *testing.T) {
180	g := directImports(t)	177	g := directImports(t)
181	allowed := module + "/internal/agent/cloudhv"	178	allowed := map[string]bool{module + "/internal/agent/cloudhv": true}
182	for pkg, deps := range g {	179	for pkg, offenders := range execViolations(g, module, "internal/agent/", allowed) {
183	if !has(pkg, "internal/agent/") \|\| pkg == allowed {	180	for _, o := range offenders {
184	continue	181	t.Errorf("data-plane package %s reaches os/exec via %s — use agent/exec.Runner instead", short(pkg), short(o))
185	}
186	for _, d := range deps {
187	if d == "os/exec" {
188	t.Errorf("data-plane package %s must not import os/exec — use agent/exec.Runner instead", short(pkg))
189	}
190	}	182	}
191	}	183	}
192	}	184	}

internal/arch/doc.go

Old		New
@@ -1,6 +1,6 @@
1	// Package arch holds executable architecture fitness functions for the Eitri	1	// Package arch holds executable architecture fitness functions for the Eitri
2	// module. It contains no production code — only tests (arch_test.go) that read	2	// module. It contains no production code — only _test.go files that read the
3	// the real package import graph via `go list` and fail the build when a	3	// real package import graph via `go list` and fail the build when a
4	// documented architectural invariant is violated.	4	// documented architectural invariant is violated.
5	//	5	//
6	// The invariants and their rationale are documented in docs/architecture.md.	6	// The invariants and their rationale are documented in docs/architecture.md.

internal/arch/execwalk_test.go

@@ @@ -0,0 +1,109 @@ @@
+package arch
+import (
+	"sort"
+	"strings"
+	"testing"
+)
+// TestExecViolationsDetectsWrappers pins the transitive os/exec detector on a
+// fake graph: a control-plane package must be flagged whether it imports
+// os/exec directly OR reaches it through an internal wrapper package (the
+// loophole a direct-import check misses). Allowed packages (cloudhv for R6)
+// are sanctioned exec users: reaching os/exec *via* them is fine, and they are
+// themselves exempt.
+func TestExecViolationsDetectsWrappers(t *testing.T) {
+	const m = "github.com/a73x/eitri"
+	graph := map[string][]string{
+		m + "/internal/server/api":       {m + "/internal/util/execwrap", "net/http"},
+		m + "/internal/util/execwrap":    {"os/exec", "fmt"},
+		m + "/internal/server/store":     {m + "/internal/transport", "database/sql"},
+		m + "/internal/transport":        {"crypto/tls"},
+		m + "/internal/server/hub":       {"os/exec"}, // direct violation
+		m + "/internal/agent/reconcile":  {m + "/internal/agent/cloudhv"},
+		m + "/internal/agent/cloudhv":    {"os/exec", m + "/internal/util/spawnhelper"}, // sanctioned (allowed)
+		m + "/internal/util/spawnhelper": {"os/exec"},
+		// multi-hop: guarded -> clean intermediate -> wrapper -> os/exec
+		m + "/internal/server/syncsvc": {m + "/internal/util/clean"},
+		m + "/internal/util/clean":     {m + "/internal/util/execwrap"},
+	}
+	t.Run("server: direct and wrapper violations found, clean pkg not flagged", func(t *testing.T) {
+		v := execViolations(graph, m, "internal/server/", nil)
+		if got := v[m+"/internal/server/api"]; len(got) != 1 || got[0] != m+"/internal/util/execwrap" {
+			t.Errorf("api must be flagged via internal/util/execwrap, got %v", got)
+		}
+		if got := v[m+"/internal/server/hub"]; len(got) != 1 || got[0] != m+"/internal/server/hub" {
+			t.Errorf("hub must be flagged as a direct importer, got %v", got)
+		}
+		if got := v[m+"/internal/server/syncsvc"]; len(got) != 1 || got[0] != m+"/internal/util/execwrap" {
+			t.Errorf("syncsvc must be flagged via the multi-hop chain, got %v", got)
+		}
+		if len(v[m+"/internal/server/store"]) != 0 {
+			t.Errorf("store is clean but was flagged: %v", v[m+"/internal/server/store"])
+		}
+	})
+	t.Run("agent: allowed package is exempt and does not taint importers", func(t *testing.T) {
+		allowed := map[string]bool{m + "/internal/agent/cloudhv": true}
+		v := execViolations(graph, m, "internal/agent/", allowed)
+		// Includes the documented design decision: spawnhelper (a non-allowed
+		// exec user) is reachable ONLY through cloudhv, so it is sanctioned by
+		// extension — code in the plane can only reach it via cloudhv's API.
+		if len(v) != 0 {
+			t.Errorf("no agent violations expected (cloudhv and its subtree sanctioned), got %v", v)
+		}
+	})
+}
+// execViolations returns, for every package whose module-relative path has the
+// given prefix, the set of packages through which os/exec becomes reachable:
+// the package itself (direct import) and/or any transitively-reached internal
+// package that imports os/exec. Walking never descends into (or flags)
+// packages in allowed — their exec use is sanctioned (R6: agent/cloudhv).
+//
+// This closes the wrapper loophole: a direct-import check misses an internal
+// package that wraps os/exec and is imported by the guarded plane.
+func execViolations(graph map[string][]string, module, prefix string, allowed map[string]bool) map[string][]string {
+	importsExec := func(pkg string) bool {
+		for _, d := range graph[pkg] {
+			if d == "os/exec" {
+				return true
+			}
+		}
+		return false
+	}
+	out := map[string][]string{}
+	for pkg := range graph {
+		rel := strings.TrimPrefix(pkg, module+"/")
+		if !strings.HasPrefix(rel, prefix) || allowed[pkg] {
+			continue
+		}
+		var offenders []string
+		if importsExec(pkg) {
+			offenders = append(offenders, pkg)
+		}
+		// Walk transitive internal deps, skipping sanctioned packages.
+		seen := map[string]bool{}
+		var walk func(string)
+		walk = func(p string) {
+			for _, d := range graph[p] {
+				if !strings.HasPrefix(d, module+"/") || seen[d] || allowed[d] {
+					continue
+				}
+				seen[d] = true
+				if importsExec(d) {
+					offenders = append(offenders, d)
+				}
+				walk(d)
+			}
+		}
+		walk(pkg)
+		if len(offenders) > 0 {
+			sort.Strings(offenders)
+			out[pkg] = offenders
+		}
+	}
+	return out
+}

internal/joinblob/joinblob.go

@@ @@ -0,0 +1,96 @@ @@
+// Package joinblob encodes and decodes the single-paste enrollment token
+// ("join blob") an agent uses to enroll: it carries the server's HTTP base URL,
+// its QUIC address, a one-shot enrollment token, and the server's TLS cert
+// fingerprint for out-of-band pinning. Dependency-free leaf shared by the
+// server (encode) and the agent (decode).
+package joinblob
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/url"
+	"regexp"
+	"strings"
+)
+// Prefix marks a join blob: human-recognizable and greppable.
+const Prefix = "eitri_join_"
+// version is the current blob format version.
+const version = 1
+// sha256Hex matches exactly 64 lowercase hex characters.
+var sha256Hex = regexp.MustCompile(`^[a-f0-9]{64}$`)
+// Fields is the decoded content of a join blob.
+type Fields struct {
+	HTTPURL  string // server HTTP base URL, including scheme (e.g. http://host:8080)
+	QUICAddr string // server QUIC address, host:port
+	Token    string // one-shot enrollment token
+	CertFP   string // server cert SHA-256 fingerprint (64 lowercase hex)
+}
+// wire is the JSON inside the base64url body. Terse keys keep the paste short.
+type wire struct {
+	V int    `json:"v"`
+	H string `json:"h"`
+	Q string `json:"q"`
+	T string `json:"t"`
+	F string `json:"f"`
+}
+// Encode validates its inputs and returns a prefixed join blob.
+func Encode(httpURL, quicAddr, token, certFP string) (string, error) {
+	if err := validate(httpURL, quicAddr, token, certFP); err != nil {
+		return "", err
+	}
+	raw, err := json.Marshal(wire{V: version, H: httpURL, Q: quicAddr, T: token, F: certFP})
+	if err != nil {
+		return "", fmt.Errorf("marshal join blob: %w", err)
+	}
+	return Prefix + base64.RawURLEncoding.EncodeToString(raw), nil
+}
+// Decode parses and validates a join blob. Surrounding whitespace/newlines
+// (common from copy-paste) are trimmed first.
+func Decode(blob string) (Fields, error) {
+	rest, ok := strings.CutPrefix(strings.TrimSpace(blob), Prefix)
+	if !ok {
+		return Fields{}, fmt.Errorf("not a join blob: missing %q prefix", Prefix)
+	}
+	raw, err := base64.RawURLEncoding.DecodeString(rest)
+	if err != nil {
+		return Fields{}, fmt.Errorf("join blob is not valid base64url: %w", err)
+	}
+	var w wire
+	if err := json.Unmarshal(raw, &w); err != nil {
+		return Fields{}, fmt.Errorf("join blob JSON malformed: %w", err)
+	}
+	if w.V != version {
+		return Fields{}, fmt.Errorf("unsupported join blob version %d (want %d)", w.V, version)
+	}
+	if err := validate(w.H, w.Q, w.T, w.F); err != nil {
+		return Fields{}, err
+	}
+	return Fields{HTTPURL: w.H, QUICAddr: w.Q, Token: w.T, CertFP: w.F}, nil
+}
+// validate enforces field rules with a distinct error per field.
+func validate(httpURL, quicAddr, token, certFP string) error {
+	u, err := url.Parse(httpURL)
+	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" || u.Path != "" {
+		return fmt.Errorf("http url %q must be an absolute http(s) URL with no path (no trailing slash)", httpURL)
+	}
+	if _, _, err := net.SplitHostPort(quicAddr); err != nil {
+		return fmt.Errorf("quic addr %q must be host:port", quicAddr)
+	}
+	if token == "" {
+		return fmt.Errorf("enrollment token is empty")
+	}
+	if !sha256Hex.MatchString(certFP) {
+		return fmt.Errorf("cert fingerprint must be 64 lowercase hex chars")
+	}
+	return nil
+}

internal/joinblob/joinblob_test.go

@@ @@ -0,0 +1,83 @@ @@
+package joinblob
+import (
+	"encoding/base64"
+	"strings"
+	"testing"
+)
+const goodFP = "18f2977d77dfea1b74aee14533bd21c34f789139e949c57023b7364894b7e5e9"
+func TestEncodeDecodeRoundTrip(t *testing.T) {
+	blob, err := Encode("http://192.168.0.190:8080", "192.168.0.190:8443", "tok123", goodFP)
+	if err != nil {
+		t.Fatalf("Encode: %v", err)
+	}
+	if !strings.HasPrefix(blob, Prefix) {
+		t.Fatalf("blob %q missing prefix %q", blob, Prefix)
+	}
+	got, err := Decode(blob)
+	if err != nil {
+		t.Fatalf("Decode: %v", err)
+	}
+	want := Fields{HTTPURL: "http://192.168.0.190:8080", QUICAddr: "192.168.0.190:8443", Token: "tok123", CertFP: goodFP}
+	if got != want {
+		t.Fatalf("round-trip mismatch:\n got %+v\nwant %+v", got, want)
+	}
+}
+func TestDecodeTrimsWhitespace(t *testing.T) {
+	blob, _ := Encode("http://h:8080", "h:8443", "t", goodFP)
+	if _, err := Decode("  " + blob + "\n"); err != nil {
+		t.Fatalf("Decode with surrounding whitespace: %v", err)
+	}
+}
+func TestEncodeRejectsBadInputs(t *testing.T) {
+	cases := map[string][4]string{
+		"http missing scheme": {"192.168.0.190:8080", "h:8443", "t", goodFP},
+		"quic not host:port":  {"http://h:8080", "hostonly", "t", goodFP},
+		"empty token":         {"http://h:8080", "h:8443", "", goodFP},
+		"short fp":            {"http://h:8080", "h:8443", "t", "abc"},
+		"uppercase fp":        {"http://h:8080", "h:8443", "t", strings.ToUpper(goodFP)},
+		"http with path":      {"http://h:8080/foo", "h:8443", "t", goodFP},
+	}
+	for name, in := range cases {
+		if _, err := Encode(in[0], in[1], in[2], in[3]); err == nil {
+			t.Errorf("%s: expected error, got nil", name)
+		}
+	}
+}
+func TestDecodeRejectsWrongVersion(t *testing.T) {
+	// A well-formed blob whose version is not 1 must be rejected. Built by hand
+	// because Encode only ever emits the current version.
+	raw := `{"v":2,"h":"http://h:8080","q":"h:8443","t":"t","f":"` + goodFP + `"}`
+	blob := Prefix + base64.RawURLEncoding.EncodeToString([]byte(raw))
+	if _, err := Decode(blob); err == nil {
+		t.Fatal("expected version-mismatch error, got nil")
+	}
+}
+func TestDecodeRejectsTrailingSlash(t *testing.T) {
+	raw := `{"v":1,"h":"http://h:8080/","q":"h:8443","t":"t","f":"` + goodFP + `"}`
+	blob := Prefix + base64.RawURLEncoding.EncodeToString([]byte(raw))
+	if _, err := Decode(blob); err == nil {
+		t.Fatal("expected trailing-slash rejection, got nil")
+	}
+}
+func TestDecodeRejectsMalformed(t *testing.T) {
+	good, _ := Encode("http://h:8080", "h:8443", "t", goodFP)
+	cases := map[string]string{
+		"no prefix":    strings.TrimPrefix(good, Prefix),
+		"bad base64":   Prefix + "!!!not-base64!!!",
+		"bad json":     Prefix + "YWJj", // base64url of "abc"
+		"empty string": "",
+	}
+	for name, blob := range cases {
+		if _, err := Decode(blob); err == nil {
+			t.Errorf("%s: expected error, got nil", name)
+		}
+	}
+}

internal/server/api/api.go

Old		New
@@ -6,13 +6,16 @@ import (
6	"crypto/sha256"	6	"crypto/sha256"
7	"crypto/subtle"	7	"crypto/subtle"
8	"database/sql"	8	"database/sql"
		9	"encoding/hex"
9	"encoding/json"	10	"encoding/json"
10	"errors"	11	"errors"
		12	"log/slog"
11	"net/http"	13	"net/http"
12	"regexp"	14	"regexp"
13	"strings"	15	"strings"
14	"time"	16	"time"
15		17
		18	"github.com/a73x/eitri/internal/joinblob"
16	"github.com/a73x/eitri/internal/server/api/types"	19	"github.com/a73x/eitri/internal/server/api/types"
17	"github.com/a73x/eitri/internal/server/hosttoken"	20	"github.com/a73x/eitri/internal/server/hosttoken"
18	"github.com/a73x/eitri/internal/server/hub"	21	"github.com/a73x/eitri/internal/server/hub"
@@ -40,20 +43,25 @@ type Config struct {
40	HostSecret []byte	43	HostSecret []byte
41	DefaultImage DefaultImage	44	DefaultImage DefaultImage
42	ServerCertSHA256 string	45	ServerCertSHA256 string
		46	AdvertiseHTTP string // HTTP base URL agents use to reach this server (scheme+host+port)
		47	AdvertiseQUIC string // QUIC host:port agents use to reach this server
43	}	48	}
44		49
45	// API is the HTTP handler container.	50	// API is the HTTP handler container.
46	type API struct {	51	type API struct {
47	cfg Config	52	cfg Config
48	st *store.Store	53	st *store.Store
49	reg *registry.Registry	54	reg *registry.Registry
50	hub *hub.Hub	55	hub *hub.Hub
51	notif *notifier	56	notif *notifier
		57	enrolls *ipLimiter // per-client-bucket brake on the unauthenticated enroll endpoint (v4: address, v6: /64)
		58	tickets *ticketStore // one-time SSE stream tickets
52	}	59	}
53		60
54	// New constructs an API.	61	// New constructs an API.
55	func New(cfg Config, st store.Store, reg registry.Registry, h hub.Hub) API {	62	func New(cfg Config, st store.Store, reg registry.Registry, h hub.Hub) API {
56	return &API{cfg: cfg, st: st, reg: reg, hub: h, notif: newNotifier()}	63	return &API{cfg: cfg, st: st, reg: reg, hub: h, notif: newNotifier(),
		64	enrolls: newIPLimiter(time.Now), tickets: newTicketStore(time.Now)}
57	}	65	}
58		66
59	// Handler returns the ServeMux with all routes registered from the declared	67	// Handler returns the ServeMux with all routes registered from the declared
@@ -174,7 +182,42 @@ func decodeJSON(w http.ResponseWriter, r *http.Request, v any) bool {
174	// validOverlays is the set of accepted overlay values at enrollment.	182	// validOverlays is the set of accepted overlay values at enrollment.
175	var validOverlays = map[string]bool{"tailscale": true, "none": true}	183	var validOverlays = map[string]bool{"tailscale": true, "none": true}
176		184
		185	// audit appends an audit row, mirrored to the live log. Failures are logged,
		186	// never fatal — the audit trail must not break the operation it records.
		187	// BEST-EFFORT: rows written here can be lost on a crash between the audited
		188	// operation and this insert. The one row that must be durable — host.enroll —
		189	// is written inside the redeem transaction by the store, not here.
		190	func (a *API) audit(action string, detail map[string]string) {
		191	raw, _ := json.Marshal(detail)
		192	if err := a.st.AppendAudit(action, string(raw)); err != nil {
		193	slog.Warn("audit append failed", "action", action, "err", err)
		194	}
		195	}
		196
		197	// truncate bounds attacker-controlled strings before they reach the audit
		198	// log (the name in a denied enroll is unauthenticated input).
		199	func truncate(s string, n int) string {
		200	if len(s) > n {
		201	return s[:n]
		202	}
		203	return s
		204	}
		205
		206	// tokenHashPrefix returns the first 8 hex chars of sha256(tok) — enough to
		207	// correlate mint/redeem audit rows without ever storing the secret.
		208	func tokenHashPrefix(tok string) string {
		209	sum := sha256.Sum256([]byte(tok))
		210	return hex.EncodeToString(sum[:])[:8]
		211	}
		212
177	func (a API) handleEnroll(w http.ResponseWriter, r http.Request) {	213	func (a API) handleEnroll(w http.ResponseWriter, r http.Request) {
		214	if !a.enrolls.allow(bucketKey(clientIP(r))) {
		215	httpError(w, "rate limited", http.StatusTooManyRequests)
		216	return
		217	}
		218	// Unauthenticated endpoint: bound the body so junk can't bloat memory or
		219	// the denied-audit trail. Legitimate enroll bodies are well under 1 KiB.
		220	r.Body = http.MaxBytesReader(w, r.Body, 16<<10)
178	var req types.EnrollRequest	221	var req types.EnrollRequest
179	if !decodeJSON(w, r, &req) {	222	if !decodeJSON(w, r, &req) {
180	return	223	return
@@ -187,12 +230,16 @@ func (a API) handleEnroll(w http.ResponseWriter, r http.Request) {
187	httpError(w, "overlay must be one of: tailscale, none", http.StatusBadRequest)	230	httpError(w, "overlay must be one of: tailscale, none", http.StatusBadRequest)
188	return	231	return
189	}	232	}
190	host, err := a.st.RedeemEnrollmentToken(req.Token, req.Name, req.OS, req.Arch, req.Provisioner, req.Overlay)	233	host, err := a.st.RedeemEnrollmentToken(req.Token, req.Name, req.OS, req.Arch, req.Provisioner, req.Overlay, clientIP(r))
191	if err != nil {	234	if err != nil {
		235	a.audit("host.enroll.denied", map[string]string{
		236	"remote": clientIP(r), "name": truncate(req.Name, 64),
		237	"token_hash_prefix": tokenHashPrefix(req.Token)})
192	httpError(w, "forbidden", http.StatusForbidden)	238	httpError(w, "forbidden", http.StatusForbidden)
193	return	239	return
194	}	240	}
195	cred := hosttoken.Mint(a.cfg.HostSecret, host.ID)	241	// host.enroll is audited durably inside the redeem transaction.
		242	cred := hosttoken.Mint(a.cfg.HostSecret, host.ID, host.CredGeneration, time.Now())
196	writeJSON(w, http.StatusCreated, types.EnrollResponse{	243	writeJSON(w, http.StatusCreated, types.EnrollResponse{
197	BridgeCIDR: host.BridgeCIDR,	244	BridgeCIDR: host.BridgeCIDR,
198	Credential: cred,	245	Credential: cred,
@@ -208,7 +255,14 @@ func (a API) handleCreateEnrollToken(w http.ResponseWriter, r http.Request) {
208	httpError(w, "internal error", http.StatusInternalServerError)	255	httpError(w, "internal error", http.StatusInternalServerError)
209	return	256	return
210	}	257	}
211	writeJSON(w, http.StatusCreated, types.EnrollTokenResponse{Token: tok})	258	a.audit("enroll-token.mint", map[string]string{
		259	"remote": clientIP(r), "token_hash_prefix": tokenHashPrefix(tok)})
		260	join, err := joinblob.Encode(a.cfg.AdvertiseHTTP, a.cfg.AdvertiseQUIC, tok, a.cfg.ServerCertSHA256)
		261	if err != nil {
		262	httpError(w, "internal error", http.StatusInternalServerError)
		263	return
		264	}
		265	writeJSON(w, http.StatusCreated, types.EnrollTokenResponse{Token: tok, Join: join})
212	}	266	}
213		267
214	// --- hosts ---	268	// --- hosts ---
@@ -240,22 +294,26 @@ func toHostResponse(h store.Host, st registry.HostState, ok bool, alloc store.Al
240	}	294	}
241		295
242	// snapshotHosts builds the wire host list (durable host rows merged with live	296	// snapshotHosts builds the wire host list (durable host rows merged with live
243	// registry state and server-computed allocation). Shared by GET /hosts and SSE.	297	// registry state and server-computed allocation) for GET /hosts.
244	func (a *API) snapshotHosts() ([]types.Host, error) {	298	func (a *API) snapshotHosts() ([]types.Host, error) {
245	hosts, err := a.st.ListHosts()	299	// Single-tx read: hosts and alloc must not mix two epochs (same property
246	if err != nil {	300	// the SSE stream needs; the unused vms read is cheap on these small
247	return nil, err	301	// control-plane tables).
248	}	302	hosts, alloc, _, err := a.st.Snapshot()
249	alloc, err := a.st.AllocatedByHost()
250	if err != nil {	303	if err != nil {
251	return nil, err	304	return nil, err
252	}	305	}
		306	return a.buildHostResponses(hosts, alloc), nil
		307	}
		308
		309	// buildHostResponses merges durable host rows with live registry state.
		310	func (a *API) buildHostResponses(hosts []store.Host, alloc map[string]store.Alloc) []types.Host {
253	out := make([]types.Host, len(hosts))	311	out := make([]types.Host, len(hosts))
254	for i, h := range hosts {	312	for i, h := range hosts {
255	st, ok := a.reg.Get(h.ID)	313	st, ok := a.reg.Get(h.ID)
256	out[i] = toHostResponse(h, st, ok, alloc[h.ID])	314	out[i] = toHostResponse(h, st, ok, alloc[h.ID])
257	}	315	}
258	return out, nil	316	return out
259	}	317	}
260		318
261	func (a API) handleListHosts(w http.ResponseWriter, r http.Request) {	319	func (a API) handleListHosts(w http.ResponseWriter, r http.Request) {
@@ -293,12 +351,18 @@ func toVMResponse(vm store.VM, actualPower, phase string) types.VM {
293	}	351	}
294		352
295	// snapshotVMs builds the wire VM list (durable VM rows merged with live	353	// snapshotVMs builds the wire VM list (durable VM rows merged with live
296	// actual-state from the registry). Shared by GET /vms and the SSE stream.	354	// actual-state from the registry). Used by GET /vms; the SSE stream uses
		355	// buildVMResponses over a single-tx store.Snapshot instead.
297	func (a *API) snapshotVMs() ([]types.VM, error) {	356	func (a *API) snapshotVMs() ([]types.VM, error) {
298	vms, err := a.st.ListVMs()	357	vms, err := a.st.ListVMs()
299	if err != nil {	358	if err != nil {
300	return nil, err	359	return nil, err
301	}	360	}
		361	return a.buildVMResponses(vms), nil
		362	}
		363
		364	// buildVMResponses merges durable VM rows with live registry actual-state.
		365	func (a *API) buildVMResponses(vms []store.VM) []types.VM {
302	out := make([]types.VM, len(vms))	366	out := make([]types.VM, len(vms))
303	for i, vm := range vms {	367	for i, vm := range vms {
304	var actualPower, phase string	368	var actualPower, phase string
@@ -313,7 +377,7 @@ func (a *API) snapshotVMs() ([]types.VM, error) {
313	}	377	}
314	out[i] = toVMResponse(vm, actualPower, phase)	378	out[i] = toVMResponse(vm, actualPower, phase)
315	}	379	}
316	return out, nil	380	return out
317	}	381	}
318		382
319	func (a API) handleListVMs(w http.ResponseWriter, r http.Request) {	383	func (a API) handleListVMs(w http.ResponseWriter, r http.Request) {
@@ -373,6 +437,12 @@ func validateCreateVM(req *types.CreateVMRequest) (string, int) {
373	if !sha256Hex.MatchString(req.ImageSHA256) {	437	if !sha256Hex.MatchString(req.ImageSHA256) {
374	return "invalid image_sha256", http.StatusBadRequest	438	return "invalid image_sha256", http.StatusBadRequest
375	}	439	}
		440	// Resource floors (post-defaults, so a zero has already become the default).
		441	// Negative values are never meaningful; a too-small disk_gb is also rejected
		442	// by the agent's never-shrink guard, but nonsense should fail at create time.
		443	if req.VCPUs < 1 \|\| req.MemMB < 1 \|\| req.DiskGB < 1 {
		444	return "vcpus, mem_mb and disk_gb must each be >= 1", http.StatusBadRequest
		445	}
376	return "", 0	446	return "", 0
377	}	447	}
378		448

internal/server/api/api_test.go

@@ @@ -9,6 +9,7 @@ import ( @@
+	"testing"
+	"time"
+	"github.com/a73x/eitri/internal/joinblob"
+	"github.com/a73x/eitri/internal/server/hub"
+	"github.com/a73x/eitri/internal/server/registry"
+	"github.com/a73x/eitri/internal/server/store"
+		DefaultImage: DefaultImage{
+			URL:    "https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.img",
+			SHA256: strings.Repeat("a", 64)},
+		AdvertiseHTTP:    "http://127.0.0.1:8080",
+		AdvertiseQUIC:    "127.0.0.1:8443",
+		ServerCertSHA256: strings.Repeat("c", 64),
+	}, st, registry.New(time.Now), h)
+	ts := httptest.NewServer(a.Handler())
+	t.Cleanup(ts.Close)
@@ @@ -322,6 +326,38 @@ func TestCreateVMNameValidation(t *testing.T) { @@
+// TestCreateVMResourceValidation pins the resource floors: post-defaults,
+// vcpus/mem_mb/disk_gb must each be >= 1. Zero means "use the default"; a
+// negative value is never meaningful (a tiny disk_gb would also truncate the
+// base image on the agent — the agent-side never-shrink guard is the backstop,
+// this rejects nonsense at create time).
+func TestCreateVMResourceValidation(t *testing.T) {
+	ts, _, _ := testServer(t)
+	out := enroll(t, ts)
+	tests := []struct {
+		name       string
+		body       map[string]any
+		wantStatus int
+	}{
+		{"negative disk_gb", map[string]any{"disk_gb": -1}, 400},
+		{"negative vcpus", map[string]any{"vcpus": -2}, 400},
+		{"negative mem_mb", map[string]any{"mem_mb": -512}, 400},
+		{"zero values take defaults", map[string]any{"disk_gb": 0, "vcpus": 0, "mem_mb": 0}, 201},
+		{"minimal explicit values", map[string]any{"disk_gb": 1, "vcpus": 1, "mem_mb": 1}, 201},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			body := map[string]any{"host_id": out["host_id"]}
+			for k, v := range tc.body {
+				body[k] = v
+			}
+			resp := do(t, "POST", ts.URL+"/api/v1/vms", "admintok", body)
+			assert.Equal(t, tc.wantStatus, resp.StatusCode)
+		})
+	}
+}
+func TestCreateVMSSHKeyValidation(t *testing.T) {
+	ts, _, _ := testServer(t)
+	out := enroll(t, ts)
@@ @@ -407,3 +443,158 @@ func TestCreateVMImageSHAAdmission(t *testing.T) { @@
+		assert.Equal(t, 201, resp.StatusCode)
+	})
+func TestMintEnrollTokenReturnsJoinBlob(t *testing.T) {
+	ts, _, _ := testServer(t)
+	resp := do(t, "POST", ts.URL+"/api/v1/enroll-tokens", "admintok", nil)
+	if resp.StatusCode != http.StatusCreated {
+		t.Fatalf("mint status = %d", resp.StatusCode)
+	}
+	var out struct {
+		Token string `json:"token"`
+		Join  string `json:"join"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&out))
+	if out.Token == "" || out.Join == "" {
+		t.Fatalf("expected token and join, got %+v", out)
+	}
+	f, err := joinblob.Decode(out.Join)
+	if err != nil {
+		t.Fatalf("join blob decode: %v", err)
+	}
+	if f.Token != out.Token {
+		t.Errorf("join token %q != response token %q", f.Token, out.Token)
+	}
+	// The blob must carry back exactly the configured advertise addresses and
+	// cert fingerprint (see testServer's api.Config) — not merely non-empty.
+	assert.Equal(t, "http://127.0.0.1:8080", f.HTTPURL)
+	assert.Equal(t, "127.0.0.1:8443", f.QUICAddr)
+	assert.Equal(t, strings.Repeat("c", 64), f.CertFP)
+}
+// TestEnrollmentIsAudited pins the audit trail: minting a token, a successful
+// enroll, and a denied enroll each leave a durable audit row (secrets appear
+// only as hash prefixes, never verbatim).
+func TestEnrollmentIsAudited(t *testing.T) {
+	ts, st, _ := testServer(t)
+	resp := do(t, "POST", ts.URL+"/api/v1/enroll-tokens", "admintok", nil)
+	require.Equal(t, 201, resp.StatusCode)
+	var tok map[string]string
+	json.NewDecoder(resp.Body).Decode(&tok)
+	resp = do(t, "POST", ts.URL+"/api/v1/enroll", "", map[string]string{
+		"token": tok["token"], "name": "host-a", "os": "linux", "arch": "amd64", "provisioner": "cloudhv"})
+	require.Equal(t, 201, resp.StatusCode)
+	resp = do(t, "POST", ts.URL+"/api/v1/enroll", "", map[string]string{
+		"token": "bogus", "name": "evil", "os": "linux", "arch": "amd64", "provisioner": "cloudhv"})
+	require.Equal(t, 403, resp.StatusCode)
+	rows, err := st.ListAudit(10)
+	require.NoError(t, err)
+	require.Len(t, rows, 3)
+	assert.Equal(t, "host.enroll.denied", rows[0].Action)
+	assert.Equal(t, "host.enroll", rows[1].Action)
+	assert.Contains(t, rows[1].Detail, "host-a")
+	assert.Equal(t, "enroll-token.mint", rows[2].Action)
+	for _, r := range rows {
+		assert.NotContains(t, r.Detail, tok["token"], "raw token must never reach the audit log")
+	}
+}
+// TestEnrollRateLimited pins the per-IP limiter on the unauthenticated enroll
+// endpoint: a burst beyond the limit returns 429 without touching the store.
+func TestEnrollRateLimited(t *testing.T) {
+	ts, _, _ := testServer(t)
+	var got429 bool
+	for i := 0; i < enrollBurst+3; i++ {
+		resp := do(t, "POST", ts.URL+"/api/v1/enroll", "", map[string]string{
+			"token": "bogus", "name": "x", "os": "linux", "arch": "amd64", "provisioner": "cloudhv"})
+		if resp.StatusCode == http.StatusTooManyRequests {
+			got429 = true
+			break
+		}
+		require.Equal(t, 403, resp.StatusCode, "pre-limit attempts fail auth, not rate limit")
+	}
+	assert.True(t, got429, "burst beyond enrollBurst must yield 429")
+}
+// TestRevokeCredentialEndpoint pins per-host revocation: POST
+// /api/v1/hosts/{id}/revoke-credential bumps the generation (204), leaves an
+// audit row, requires admin auth, and 404s unknown hosts.
+func TestRevokeCredentialEndpoint(t *testing.T) {
+	ts, st, _ := testServer(t)
+	out := enroll(t, ts)
+	resp := do(t, "POST", ts.URL+"/api/v1/hosts/"+out["host_id"]+"/revoke-credential", "admintok", nil)
+	require.Equal(t, 204, resp.StatusCode)
+	h, err := st.GetHost(out["host_id"])
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), h.CredGeneration)
+	rows, err := st.ListAudit(3)
+	require.NoError(t, err)
+	require.NotEmpty(t, rows)
+	assert.Equal(t, "host.credential.revoke", rows[0].Action)
+	assert.Contains(t, rows[0].Detail, out["host_id"])
+	resp = do(t, "POST", ts.URL+"/api/v1/hosts/deadbeef/revoke-credential", "admintok", nil)
+	assert.Equal(t, 404, resp.StatusCode)
+	resp = do(t, "POST", ts.URL+"/api/v1/hosts/"+out["host_id"]+"/revoke-credential", "wrong", nil)
+	assert.Equal(t, 401, resp.StatusCode)
+}
+// TestEnrollMintsGenerationCredential pins that the enroll response carries a
+// v2 credential bound to the host's current generation.
+func TestEnrollMintsGenerationCredential(t *testing.T) {
+	ts, _, _ := testServer(t)
+	out := enroll(t, ts)
+	parts := strings.Split(out["credential"], ".")
+	require.Len(t, parts, 4, "v2 credential is host_id.gen.issued.hmac")
+	assert.Equal(t, out["host_id"], parts[0])
+	assert.Equal(t, "1", parts[1], "fresh enrollment mints generation 1")
+}
+// TestAuditEndpoint pins the forensic read API: GET /api/v1/audit returns
+// newest-first rows (detail as embedded JSON), honors ?limit, requires admin.
+func TestAuditEndpoint(t *testing.T) {
+	ts, _, _ := testServer(t)
+	enroll(t, ts) // produces mint + enroll audit rows
+	resp := do(t, "GET", ts.URL+"/api/v1/audit", "admintok", nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var rows []struct {
+		At     time.Time       `json:"at"`
+		Action string          `json:"action"`
+		Detail json.RawMessage `json:"detail"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&rows))
+	require.Len(t, rows, 2)
+	assert.Equal(t, "host.enroll", rows[0].Action, "newest first")
+	assert.Equal(t, "enroll-token.mint", rows[1].Action)
+	assert.False(t, rows[0].At.IsZero())
+	assert.Contains(t, string(rows[0].Detail), "host-a")
+	resp = do(t, "GET", ts.URL+"/api/v1/audit?limit=1", "admintok", nil)
+	require.Equal(t, 200, resp.StatusCode)
+	rows = nil
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&rows))
+	assert.Len(t, rows, 1)
+	resp = do(t, "GET", ts.URL+"/api/v1/audit", "", nil)
+	assert.Equal(t, 401, resp.StatusCode)
+}
+// TestAuditEndpointLimitValidation pins the explicit reject-not-clamp
+// contract for ?limit.
+func TestAuditEndpointLimitValidation(t *testing.T) {
+	ts, _, _ := testServer(t)
+	for _, bad := range []string{"0", "1001", "-3", "abc"} {
+		resp := do(t, "GET", ts.URL+"/api/v1/audit?limit="+bad, "admintok", nil)
+		assert.Equal(t, 400, resp.StatusCode, "limit=%s must be rejected", bad)
+	}
+}

internal/server/api/decommission_api_test.go

Old		New
@@ -24,9 +24,12 @@ func apiServer(t testing.T) (httptest.Server, API, store.Store) {
24	require.NoError(t, err)	24	require.NoError(t, err)
25	t.Cleanup(func() { st.Close() })	25	t.Cleanup(func() { st.Close() })
26	a := New(Config{	26	a := New(Config{
27	AdminToken: "admintok",	27	AdminToken: "admintok",
28	HostSecret: []byte("hostsecret"),	28	HostSecret: []byte("hostsecret"),
29	DefaultImage: DefaultImage{URL: "http://img", SHA256: strings.Repeat("a", 64)},	29	DefaultImage: DefaultImage{URL: "http://img", SHA256: strings.Repeat("a", 64)},
		30	AdvertiseHTTP: "http://127.0.0.1:8080",
		31	AdvertiseQUIC: "127.0.0.1:8443",
		32	ServerCertSHA256: strings.Repeat("c", 64),
30	}, st, registry.New(time.Now), hub.New())	33	}, st, registry.New(time.Now), hub.New())
31	ts := httptest.NewServer(a.Handler())	34	ts := httptest.NewServer(a.Handler())
32	t.Cleanup(ts.Close)	35	t.Cleanup(ts.Close)
@@ -79,7 +82,7 @@ func TestEventsStreamSendsSnapshot(t *testing.T) {
79		82
80	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)	83	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
81	defer cancel()	84	defer cancel()
82	req, _ := http.NewRequestWithContext(ctx, "GET", ts.URL+"/api/v1/events?token=admintok", nil)	85	req, _ := http.NewRequestWithContext(ctx, "GET", ts.URL+"/api/v1/events?ticket="+mintTicket(t, ts.URL, "admintok"), nil)
83	resp, err := http.DefaultClient.Do(req)	86	resp, err := http.DefaultClient.Do(req)
84	require.NoError(t, err)	87	require.NoError(t, err)
85	defer resp.Body.Close()	88	defer resp.Body.Close()
@@ -107,6 +110,6 @@ func TestEventsStreamSendsSnapshot(t *testing.T) {
107		110
108	func TestEventsRejectsBadToken(t *testing.T) {	111	func TestEventsRejectsBadToken(t *testing.T) {
109	ts, _, _ := apiServer(t)	112	ts, _, _ := apiServer(t)
110	resp := do(t, "GET", ts.URL+"/api/v1/events?token=wrong", "", nil)	113	resp := do(t, "GET", ts.URL+"/api/v1/events?ticket=wrong", "", nil)
111	assert.Equal(t, http.StatusUnauthorized, resp.StatusCode)	114	assert.Equal(t, http.StatusUnauthorized, resp.StatusCode)
112	}	115	}

internal/server/api/events.go

Old		New
@@ -7,6 +7,7 @@ import (
7	"errors"	7	"errors"
8	"fmt"	8	"fmt"
9	"net/http"	9	"net/http"
		10	"strconv"
10	"time"	11	"time"
11		12
12	"github.com/a73x/eitri/internal/server/api/types"	13	"github.com/a73x/eitri/internal/server/api/types"
@@ -24,20 +25,73 @@ func (a API) handleDecommissionHost(w http.ResponseWriter, r http.Request) {
24	httpError(w, "internal error", http.StatusInternalServerError)	25	httpError(w, "internal error", http.StatusInternalServerError)
25	return	26	return
26	}	27	}
		28	a.audit("host.decommission", map[string]string{"host_id": id, "remote": clientIP(r)})
27	a.notif.notify()	29	a.notif.notify()
28	w.WriteHeader(http.StatusAccepted)	30	w.WriteHeader(http.StatusAccepted)
29	}	31	}
30		32
		33	// handleRevokeCredential bumps the host's credential generation, revoking its
		34	// outstanding credential WITHOUT rotating the fleet secret. The agent's live
		35	// session is closed by syncsvc within one report tick; the host stays dark
		36	// until the operator re-enrolls it with a fresh join blob.
		37	func (a API) handleRevokeCredential(w http.ResponseWriter, r http.Request) {
		38	id := r.PathValue("id")
		39	// Audit is written inside the bump transaction (a security action must
		40	// not be able to happen unrecorded).
		41	_, err := a.st.BumpCredGeneration(id, clientIP(r))
		42	switch {
		43	case errors.Is(err, sql.ErrNoRows):
		44	httpError(w, "host not found", http.StatusNotFound)
		45	return
		46	case err != nil:
		47	httpError(w, "internal error", http.StatusInternalServerError)
		48	return
		49	}
		50	w.WriteHeader(http.StatusNoContent)
		51	}
		52
		53	// handleListAudit returns the newest audit rows (default 100, ?limit=N caps
		54	// at 1000). Completes the forensic story: rows were previously reachable only
		55	// by opening the SQLite file.
		56	func (a API) handleListAudit(w http.ResponseWriter, r http.Request) {
		57	limit := 100
		58	if v := r.URL.Query().Get("limit"); v != "" {
		59	n, err := strconv.Atoi(v)
		60	if err != nil \|\| n < 1 \|\| n > 1000 {
		61	httpError(w, "limit must be 1..1000", http.StatusBadRequest)
		62	return
		63	}
		64	limit = n
		65	}
		66	rows, err := a.st.ListAudit(limit)
		67	if err != nil {
		68	httpError(w, "internal error", http.StatusInternalServerError)
		69	return
		70	}
		71	out := make([]types.AuditEvent, len(rows))
		72	for i, e := range rows {
		73	detail := json.RawMessage(e.Detail)
		74	if !json.Valid(detail) { // defensive: never emit invalid JSON
		75	detail, _ = json.Marshal(e.Detail)
		76	}
		77	out[i] = types.AuditEvent{At: e.At, Action: e.Action, Detail: detail}
		78	}
		79	writeJSON(w, http.StatusOK, out)
		80	}
		81
		82	// handleMintStreamTicket issues a one-time short-TTL ticket for the SSE
		83	// stream (admin-authenticated; the ticket is the only thing that ever
		84	// appears in a URL).
		85	func (a API) handleMintStreamTicket(w http.ResponseWriter, r http.Request) {
		86	writeJSON(w, http.StatusCreated, types.StreamTicketResponse{Ticket: a.tickets.mint()})
		87	}
		88
31	// handleEvents streams the fleet snapshot as Server-Sent Events. It pushes on	89	// handleEvents streams the fleet snapshot as Server-Sent Events. It pushes on
32	// every desired-state change (via the notifier) and re-checks on a 1s tick to	90	// every desired-state change (via the notifier) and re-checks on a 1s tick to
33	// catch agent-reported actual-state changes, sending only when the snapshot	91	// catch agent-reported actual-state changes, sending only when the snapshot
34	// actually changed. A periodic comment keeps the connection alive.	92	// actually changed. A periodic comment keeps the connection alive.
35	func (a API) handleEvents(w http.ResponseWriter, r http.Request) {	93	func (a API) handleEvents(w http.ResponseWriter, r http.Request) {
36	if a.cfg.AdminToken == "" {	94	if !a.tickets.consume(r.URL.Query().Get("ticket")) {
37	http.Error(w, "unauthorized", http.StatusUnauthorized)
38	return
39	}
40	if !constantTimeTokenMatch(r.URL.Query().Get("token"), a.cfg.AdminToken) {
41	http.Error(w, "unauthorized", http.StatusUnauthorized)	95	http.Error(w, "unauthorized", http.StatusUnauthorized)
42	return	96	return
43	}	97	}
@@ -96,15 +150,16 @@ func (a API) handleEvents(w http.ResponseWriter, r http.Request) {
96	}	150	}
97	}	151	}
98		152
99	// marshalSnapshot builds and JSON-encodes the current fleet snapshot.	153	// marshalSnapshot builds and JSON-encodes the current fleet snapshot. The
		154	// store reads happen in one transaction (store.Snapshot) so hosts, allocation
		155	// and VMs can never mix state from two different epochs.
100	func (a *API) marshalSnapshot() ([]byte, error) {	156	func (a *API) marshalSnapshot() ([]byte, error) {
101	hosts, err := a.snapshotHosts()	157	hosts, alloc, vms, err := a.st.Snapshot()
102	if err != nil {
103	return nil, err
104	}
105	vms, err := a.snapshotVMs()
106	if err != nil {	158	if err != nil {
107	return nil, err	159	return nil, err
108	}	160	}
109	return json.Marshal(types.StateSnapshot{Hosts: hosts, VMs: vms})	161	return json.Marshal(types.StateSnapshot{
		162	Hosts: a.buildHostResponses(hosts, alloc),
		163	VMs: a.buildVMResponses(vms),
		164	})
110	}	165	}

internal/server/api/events_test.go

@@ @@ -0,0 +1,95 @@ @@
+package api
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// mintTicket mints a one-time stream ticket via the admin API.
+func mintTicket(t *testing.T, ts string, token string) string {
+	t.Helper()
+	resp := do(t, "POST", ts+"/api/v1/stream-tickets", token, nil)
+	require.Equal(t, 201, resp.StatusCode)
+	var out map[string]string
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&out))
+	require.NotEmpty(t, out["ticket"])
+	return out["ticket"]
+}
+// readFirstSSEEvent connects to the events URL and returns true when a
+// "event: state" frame arrives before the deadline.
+func readFirstSSEEvent(t *testing.T, url string) (int, bool) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		return resp.StatusCode, false
+	}
+	sc := bufio.NewScanner(resp.Body)
+	for sc.Scan() {
+		if strings.HasPrefix(sc.Text(), "event: state") {
+			return 200, true
+		}
+	}
+	return 200, false
+}
+// TestStreamTicketFlow pins the SSE auth model: the long-lived admin token
+// never rides in a URL. A one-time short-TTL ticket is minted over an
+// authenticated POST; the events stream consumes it; replay fails; the old
+// ?token= path is gone.
+func TestStreamTicketFlow(t *testing.T) {
+	ts, _, _ := testServer(t)
+	t.Run("mint requires admin", func(t *testing.T) {
+		resp := do(t, "POST", ts.URL+"/api/v1/stream-tickets", "wrong", nil)
+		assert.Equal(t, 401, resp.StatusCode)
+	})
+	tick := mintTicket(t, ts.URL, "admintok")
+	t.Run("valid ticket streams", func(t *testing.T) {
+		code, gotState := readFirstSSEEvent(t, ts.URL+"/api/v1/events?ticket="+tick)
+		assert.Equal(t, 200, code)
+		assert.True(t, gotState, "first state frame must arrive")
+	})
+	t.Run("replay rejected", func(t *testing.T) {
+		code, _ := readFirstSSEEvent(t, ts.URL+"/api/v1/events?ticket="+tick)
+		assert.Equal(t, 401, code, "a consumed ticket must be rejected")
+	})
+	t.Run("garbage ticket rejected", func(t *testing.T) {
+		code, _ := readFirstSSEEvent(t, ts.URL+"/api/v1/events?ticket=nope")
+		assert.Equal(t, 401, code)
+	})
+	t.Run("admin token in query string is rejected", func(t *testing.T) {
+		code, _ := readFirstSSEEvent(t, ts.URL+"/api/v1/events?token=admintok")
+		assert.Equal(t, 401, code, "an admin token in the URL must not authenticate the stream")
+	})
+}
+// TestStreamTicketExpiry pins the TTL with an injected clock.
+func TestStreamTicketExpiry(t *testing.T) {
+	now := time.Unix(1_750_000_000, 0)
+	tk := newTicketStore(func() time.Time { return now })
+	tick := tk.mint()
+	now = now.Add(streamTicketTTL + time.Second)
+	assert.False(t, tk.consume(tick), "expired ticket must not be consumable")
+	fresh := tk.mint()
+	assert.True(t, tk.consume(fresh))
+	assert.False(t, tk.consume(fresh), "one-time: second consume fails")
+}

internal/server/api/ratelimit.go

@@ @@ -0,0 +1,110 @@ @@
+package api
+import (
+	"net"
+	"net/http"
+	"net/netip"
+	"sync"
+	"time"
+)
+// enrollBurst is the per-IP bucket capacity for POST /api/v1/enroll — the
+// only unauthenticated endpoint. Enrollment is operator-paced (one paste per
+// host), so a small burst with a slow refill comfortably covers legitimate
+// use while blunting token-guessing.
+const enrollBurst = 5
+// enrollRefillEvery is how often one token drips back into a bucket.
+const enrollRefillEvery = 30 * time.Second
+// maxLimiterEntries caps the bucket map so an address-spoofing flood cannot
+// grow it without bound; when full, entries idle past a full refill are
+// pruned, and if nothing is prunable the request is allowed (fail open — the
+// limiter is a brake, not the auth boundary).
+const maxLimiterEntries = 10_000
+// ipLimiter is a per-IP token bucket. now is injectable for tests.
+type ipLimiter struct {
+	mu      sync.Mutex
+	buckets map[string]*bucket
+	now     func() time.Time
+}
+type bucket struct {
+	tokens   float64
+	lastSeen time.Time
+}
+func newIPLimiter(now func() time.Time) *ipLimiter {
+	return &ipLimiter{buckets: map[string]*bucket{}, now: now}
+}
+// allow reports whether ip may proceed, consuming one token if so.
+func (l *ipLimiter) allow(ip string) bool {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	now := l.now()
+	b := l.buckets[ip]
+	if b == nil {
+		if len(l.buckets) >= maxLimiterEntries && !l.prune(now) {
+			return true // fail open; see maxLimiterEntries
+		}
+		b = &bucket{tokens: enrollBurst}
+		l.buckets[ip] = b
+	} else {
+		refill := float64(now.Sub(b.lastSeen)) / float64(enrollRefillEvery)
+		b.tokens = min(enrollBurst, b.tokens+refill)
+	}
+	b.lastSeen = now
+	if b.tokens < 1 {
+		return false
+	}
+	b.tokens--
+	return true
+}
+// prune drops buckets idle past a full refill; reports whether any were freed.
+func (l *ipLimiter) prune(now time.Time) bool {
+	idle := time.Duration(enrollBurst) * enrollRefillEvery
+	freed := false
+	for ip, b := range l.buckets {
+		if now.Sub(b.lastSeen) > idle {
+			delete(l.buckets, ip)
+			freed = true
+		}
+	}
+	return freed
+}
+// bucketKey normalizes a client IP into its rate-limit bucket: IPv4 (and
+// v4-mapped v6) per address; IPv6 by /64, since a single host trivially owns
+// billions of v6 addresses and per-address buckets would under-limit it.
+// Unparseable input is used raw (still a stable key).
+func bucketKey(ip string) string {
+	addr, err := netip.ParseAddr(ip)
+	if err != nil {
+		return ip
+	}
+	if addr.Is4() || addr.Is4In6() {
+		return addr.Unmap().String()
+	}
+	p, err := addr.Prefix(64)
+	if err != nil {
+		// Unreachable for a parsed addr (BitLen is always 128 here); kept as
+		// belt-and-braces so a refactor can't turn this into a panic.
+		return ip
+	}
+	return p.String()
+}
+// clientIP extracts the bare IP from r.RemoteAddr. Deployments fronted by a
+// reverse proxy see the proxy's address here — enroll rate limiting then
+// applies to the proxy as a whole, which is still a meaningful brake.
+func clientIP(r *http.Request) string {
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr
+	}
+	return host
+}

internal/server/api/ratelimit_test.go

@@ @@ -0,0 +1,81 @@ @@
+package api
+import (
+	"fmt"
+	"testing"
+	"time"
+	"github.com/stretchr/testify/assert"
+)
+// TestLimiterRefillAndCap pins the token-bucket math against a fake clock:
+// a drained bucket refills one token per enrollRefillEvery, capped at burst.
+func TestLimiterRefillAndCap(t *testing.T) {
+	now := time.Unix(1_750_000_000, 0)
+	l := newIPLimiter(func() time.Time { return now })
+	for i := 0; i < enrollBurst; i++ {
+		assert.True(t, l.allow("10.0.0.1"), "burst request %d", i)
+	}
+	assert.False(t, l.allow("10.0.0.1"), "burst exhausted")
+	now = now.Add(enrollRefillEvery)
+	assert.True(t, l.allow("10.0.0.1"), "one token refilled")
+	assert.False(t, l.allow("10.0.0.1"), "only one refilled")
+	// A long idle period refills to the cap, not beyond.
+	now = now.Add(24 * time.Hour)
+	for i := 0; i < enrollBurst; i++ {
+		assert.True(t, l.allow("10.0.0.1"), "cap request %d", i)
+	}
+	assert.False(t, l.allow("10.0.0.1"), "cap enforced")
+}
+// TestLimiterPruneAndFailOpen pins the capacity behavior: at the entry cap,
+// idle buckets are pruned to admit new IPs; when nothing is prunable the
+// limiter fails OPEN (it is a brake, not the auth boundary).
+func TestLimiterPruneAndFailOpen(t *testing.T) {
+	now := time.Unix(1_750_000_000, 0)
+	l := newIPLimiter(func() time.Time { return now })
+	for i := 0; i < maxLimiterEntries; i++ {
+		l.allow(fmt.Sprintf("10.%d.%d.%d", i>>16&0xff, i>>8&0xff, i&0xff))
+	}
+	assert.Len(t, l.buckets, maxLimiterEntries)
+	// Nothing is idle yet: a new IP must still be allowed (fail open) and
+	// must not grow the map.
+	assert.True(t, l.allow("192.0.2.1"), "fail open when full and nothing prunable")
+	assert.Len(t, l.buckets, maxLimiterEntries, "fail-open must not grow the map")
+	// After everything goes idle past a full refill, pruning admits new IPs.
+	now = now.Add(time.Duration(enrollBurst)*enrollRefillEvery + time.Second)
+	assert.True(t, l.allow("192.0.2.2"))
+	assert.Len(t, l.buckets, 1, "idle buckets pruned, new bucket tracked")
+}
+// TestBucketKeyGroupsIPv6BySlash64 pins the bucket-key normalization: IPv4
+// (and v4-mapped-v6) keep per-address buckets; IPv6 collapses to the /64 so a
+// single host's billions of addresses share one bucket; garbage stays raw.
+func TestBucketKeyGroupsIPv6BySlash64(t *testing.T) {
+	assert.Equal(t, "203.0.113.9", bucketKey("203.0.113.9"))
+	assert.Equal(t, "203.0.113.9", bucketKey("::ffff:203.0.113.9"), "v4-mapped stays per-address")
+	a := bucketKey("2001:db8:1:2:aaaa:bbbb:cccc:dddd")
+	b := bucketKey("2001:db8:1:2:1111:2222:3333:4444")
+	c := bucketKey("2001:db8:1:3::1")
+	assert.Equal(t, a, b, "same /64 must share a bucket")
+	assert.NotEqual(t, a, c, "different /64 must not")
+	assert.Equal(t, "not-an-ip", bucketKey("not-an-ip"))
+}
+// TestLimiterSharesBucketAcrossSameSlash64 pins end behavior: burst spent
+// from one v6 address exhausts the whole /64.
+func TestLimiterSharesBucketAcrossSameSlash64(t *testing.T) {
+	now := time.Unix(1_750_000_000, 0)
+	l := newIPLimiter(func() time.Time { return now })
+	for i := 0; i < enrollBurst; i++ {
+		assert.True(t, l.allow(bucketKey("2001:db8::1")))
+	}
+	assert.False(t, l.allow(bucketKey("2001:db8::2")), "same /64: bucket shared")
+	assert.True(t, l.allow(bucketKey("2001:db9::1")), "different /64: fresh bucket")
+}

internal/server/api/routes.go

Old		New
@@ -11,7 +11,8 @@ import (
11	type AuthTier int	11	type AuthTier int
12		12
13	const (	13	const (
14	AuthPublic AuthTier = iota // no auth (public material)	14	AuthPublic AuthTier = iota // no auth (public material / rate-limited)
		15	AuthTicket // one-time short-TTL ticket in ?ticket= (browser streams)
15	AuthAdmin // Authorization: Bearer <admin token>	16	AuthAdmin // Authorization: Bearer <admin token>
16	)	17	)
17		18
@@ -56,19 +57,21 @@ var routeTable = []Route{
56	Request: (*types.EnrollRequest)(nil),	57	Request: (*types.EnrollRequest)(nil),
57	Response: (*types.EnrollResponse)(nil),	58	Response: (*types.EnrollResponse)(nil),
58	Success: http.StatusCreated,	59	Success: http.StatusCreated,
59	Doc: "Redeem a one-time enrollment token: a new host joins the fleet and receives its credential. Unauthenticated; the token is the proof.",	60	Doc: "Redeem a one-time enrollment token: a new host joins the fleet and receives its credential. Unauthenticated but rate-limited; the token is the proof.",
60	handler: (*API).handleEnroll,	61	handler: (*API).handleEnroll,
61	},	62	},
62	// SSE live status. EventSource cannot set headers, so this endpoint	63	// SSE live status. EventSource cannot set headers, so the stream
63	// authenticates via a ?token= query param instead of the admin middleware.	64	// authenticates with a one-time short-TTL ticket minted via the
		65	// admin-authenticated POST /api/v1/stream-tickets — the long-lived admin
		66	// token never rides in a URL (proxy/access logs).
64	{	67	{
65	Method: "GET",	68	Method: "GET",
66	Path: "/api/v1/events",	69	Path: "/api/v1/events",
67	Auth: AuthPublic,	70	Auth: AuthTicket,
68	Kind: KindSSE,	71	Kind: KindSSE,
69	Response: (*types.StateSnapshot)(nil),	72	Response: (*types.StateSnapshot)(nil),
70	Success: http.StatusOK,	73	Success: http.StatusOK,
71	Query: []QueryParam{{Name: "token", Doc: "admin token"}},	74	Query: []QueryParam{{Name: "ticket", Doc: "one-time stream ticket"}},
72	Doc: "Live fleet state stream (Server-Sent Events); each 'state' event carries a StateSnapshot.",	75	Doc: "Live fleet state stream (Server-Sent Events); each 'state' event carries a StateSnapshot.",
73	handler: (*API).handleEvents,	76	handler: (*API).handleEvents,
74	},	77	},
@@ -81,7 +84,7 @@ var routeTable = []Route{
81	Kind: KindJSON,	84	Kind: KindJSON,
82	Response: (*types.EnrollTokenResponse)(nil),	85	Response: (*types.EnrollTokenResponse)(nil),
83	Success: http.StatusCreated,	86	Success: http.StatusCreated,
84	Doc: "Mint a one-time host enrollment token.",	87	Doc: "Mint a one-time host enrollment token plus the join blob agents consume.",
85	handler: (*API).handleCreateEnrollToken,	88	handler: (*API).handleCreateEnrollToken,
86	},	89	},
87	{	90	{
@@ -104,6 +107,36 @@ var routeTable = []Route{
104	handler: (*API).handleDecommissionHost,	107	handler: (*API).handleDecommissionHost,
105	},	108	},
106	{	109	{
		110	Method: "POST",
		111	Path: "/api/v1/hosts/{id}/revoke-credential",
		112	Auth: AuthAdmin,
		113	Kind: KindJSON,
		114	Success: http.StatusNoContent,
		115	Doc: "Revoke a host's outstanding credential by bumping its generation; the host stays dark until re-enrolled.",
		116	handler: (*API).handleRevokeCredential,
		117	},
		118	{
		119	Method: "GET",
		120	Path: "/api/v1/audit",
		121	Auth: AuthAdmin,
		122	Kind: KindJSON,
		123	Response: []types.AuditEvent(nil),
		124	Success: http.StatusOK,
		125	Query: []QueryParam{{Name: "limit", Doc: "max rows to return (default 100, cap 1000)"}},
		126	Doc: "Newest audit log rows.",
		127	handler: (*API).handleListAudit,
		128	},
		129	{
		130	Method: "POST",
		131	Path: "/api/v1/stream-tickets",
		132	Auth: AuthAdmin,
		133	Kind: KindJSON,
		134	Response: (*types.StreamTicketResponse)(nil),
		135	Success: http.StatusCreated,
		136	Doc: "Mint a one-time short-TTL ticket for the SSE stream — the only credential that ever rides in a URL.",
		137	handler: (*API).handleMintStreamTicket,
		138	},
		139	{
107	Method: "GET",	140	Method: "GET",
108	Path: "/api/v1/vms",	141	Path: "/api/v1/vms",
109	Auth: AuthAdmin,	142	Auth: AuthAdmin,

internal/server/api/routes_test.go

Old		New
@@ -34,7 +34,7 @@ func exemplarElem(t *testing.T, route Route, role string, v any) reflect.Type {
34	// `required` array for request schemas, so a type serving both roles would	34	// `required` array for request schemas, so a type serving both roles would
35	// get the wrong treatment on one of them).	35	// get the wrong treatment on one of them).
36	func TestRouteTable(t *testing.T) {	36	func TestRouteTable(t *testing.T) {
37	const wantRoutes = 9	37	const wantRoutes = 12
38	if len(routeTable) != wantRoutes {	38	if len(routeTable) != wantRoutes {
39	t.Fatalf("route table has %d entries, want %d — new endpoint? update this pin and cmd/eitri-apispec coverage together", len(routeTable), wantRoutes)	39	t.Fatalf("route table has %d entries, want %d — new endpoint? update this pin and cmd/eitri-apispec coverage together", len(routeTable), wantRoutes)
40	}	40	}

internal/server/api/testdata/audit-event.golden.json

@@ @@ -0,0 +1,8 @@ @@
+{
+  "at": "2026-07-27T12:02:00Z",
+  "action": "host.enroll",
+  "detail": {
+    "host_id": "h-1234",
+    "remote": "203.0.113.7"
+  }
+}

internal/server/api/testdata/enroll-token-response.golden.json

Old		New
@@ -1,3 +1,4 @@
1	{	1	{
		2	"join": "eitri-join-blob-base64url",
2	"token": "tok-secret-01"	3	"token": "tok-secret-01"
3	}	4	}

internal/server/api/testdata/stream-ticket-response.golden.json

@@ @@ -0,0 +1,3 @@ @@
+{
+  "ticket": "ticket-opaque-01"
+}

internal/server/api/ticket.go

@@ @@ -0,0 +1,58 @@ @@
+package api
+import (
+	"sync"
+	"time"
+	"github.com/a73x/eitri/internal/server/store"
+)
+// streamTicketTTL bounds how long a minted stream ticket stays redeemable.
+// It only needs to cover the gap between the SPA's mint call and the
+// EventSource connect — one minute is generous.
+const streamTicketTTL = time.Minute
+// ticketStore holds one-time SSE stream tickets in memory. Tickets are
+// ephemeral session bootstrap — a server restart just means the client mints
+// a fresh one on its next reconnect — so no durability is needed. now is
+// injectable for tests.
+type ticketStore struct {
+	mu      sync.Mutex
+	tickets map[string]time.Time // ticket → expiry
+	now     func() time.Time
+}
+func newTicketStore(now func() time.Time) *ticketStore {
+	return &ticketStore{tickets: map[string]time.Time{}, now: now}
+}
+// mint issues a fresh one-time ticket, pruning expired ones while it holds
+// the lock (mints are operator-paced; the map stays tiny).
+func (t *ticketStore) mint() string {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	now := t.now()
+	for k, exp := range t.tickets {
+		if now.After(exp) {
+			delete(t.tickets, k)
+		}
+	}
+	tick := store.RandHex(16)
+	t.tickets[tick] = now.Add(streamTicketTTL)
+	return tick
+}
+// consume redeems a ticket exactly once; expired or unknown tickets fail.
+// The map lookup is not constant-time by design: tickets are 128-bit
+// crypto-random, single-use, and 60s-TTL, so timing attacks are academic —
+// unlike the long-lived admin token, which does get constantTimeTokenMatch.
+func (t *ticketStore) consume(tick string) bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	exp, ok := t.tickets[tick]
+	if !ok {
+		return false
+	}
+	delete(t.tickets, tick) // one-time, even when expired
+	return !t.now().After(exp)
+}

internal/server/api/types/types.go

Old		New
@@ -6,6 +6,7 @@
6	package types	6	package types
7		7
8	import (	8	import (
		9	"encoding/json"
9	"time"	10	"time"
10	)	11	)
11		12
@@ -117,6 +118,7 @@ type EnrollResponse struct {
117		118
118	// EnrollTokenResponse answers POST /api/v1/enroll-tokens.	119	// EnrollTokenResponse answers POST /api/v1/enroll-tokens.
119	type EnrollTokenResponse struct {	120	type EnrollTokenResponse struct {
		121	Join string `json:"join"`
120	Token string `json:"token"`	122	Token string `json:"token"`
121	}	123	}
122		124
@@ -125,3 +127,16 @@ type CreateVMResponse struct {
125	ID string `json:"id"`	127	ID string `json:"id"`
126	Name string `json:"name"`	128	Name string `json:"name"`
127	}	129	}
		130
		131	// StreamTicketResponse answers POST /api/v1/stream-tickets.
		132	type StreamTicketResponse struct {
		133	Ticket string `json:"ticket"`
		134	}
		135
		136	// AuditEvent is the wire shape of one audit row, served by GET /api/v1/audit;
		137	// detail is embedded as raw JSON (it is always a marshaled object).
		138	type AuditEvent struct {
		139	At time.Time `json:"at"`
		140	Action string `json:"action"`
		141	Detail json.RawMessage `json:"detail"`
		142	}

internal/server/api/wire_golden_test.go

Old		New
@@ -129,6 +129,7 @@ func TestWireGolden(t *testing.T) {
129	})	129	})
130		130
131	goldenCheck(t, "enroll-token-response", types.EnrollTokenResponse{	131	goldenCheck(t, "enroll-token-response", types.EnrollTokenResponse{
		132	Join: "eitri-join-blob-base64url",
132	Token: "tok-secret-01",	133	Token: "tok-secret-01",
133	})	134	})
134		135
@@ -136,4 +137,14 @@ func TestWireGolden(t *testing.T) {
136	ID: "v-5678",	137	ID: "v-5678",
137	Name: "sandbox-abc123",	138	Name: "sandbox-abc123",
138	})	139	})
		140
		141	goldenCheck(t, "stream-ticket-response", types.StreamTicketResponse{
		142	Ticket: "ticket-opaque-01",
		143	})
		144
		145	goldenCheck(t, "audit-event", types.AuditEvent{
		146	At: base.Add(2 * time.Minute),
		147	Action: "host.enroll",
		148	Detail: json.RawMessage(`{"host_id":"h-1234","remote":"203.0.113.7"}`),
		149	})
139	}	150	}

internal/server/hosttoken/hosttoken.go

Old		New
@@ -1,34 +1,69 @@
1	// Package hosttoken mints and verifies host credentials: "<host_id>.<hex hmac-sha256>".	1	// Package hosttoken mints and verifies generation-versioned host credentials.
2	// Phase 1 has no revocation list (single-tenant; rotate the server secret to revoke all).	2	//
3	// Host IDs are hex strings and must not contain '.'; a dotted input fails verification	3	// Format:
4	// safely because the signature is computed over the full ID and will never match the	4	//
5	// truncated parse produced by strings.Cut.	5	// "<host_id>.<generation>.<issued_unix>.<hex hmac-sha256>"
		6	//
		7	// The HMAC covers the three dotted fields, binding identity, credential
		8	// generation, and issue time. Generation enables PER-HOST revocation: the
		9	// server compares the credential's generation against the host row's
		10	// cred_generation and rejects stale ones — bumping the row revokes that one
		11	// host without rotating the fleet-wide secret. issued_unix enables an
		12	// optional max-age policy (enforced by the caller; this package only signs
		13	// and parses).
		14	//
		15	// Host IDs are hex strings and must not contain '.'; a dotted input fails
		16	// verification safely because the signature is computed over the exact
		17	// parsed fields and will never match.
6	package hosttoken	18	package hosttoken
7		19
8	import (	20	import (
9	"crypto/hmac"	21	"crypto/hmac"
10	"crypto/sha256"	22	"crypto/sha256"
11	"encoding/hex"	23	"encoding/hex"
		24	"fmt"
		25	"strconv"
12	"strings"	26	"strings"
		27	"time"
13	)	28	)
14		29
15	func sign(secret []byte, hostID string) string {	30	// Claims is the verified content of a credential.
		31	type Claims struct {
		32	HostID string
		33	Generation int64
		34	IssuedAt time.Time
		35	}
		36
		37	func sign(secret []byte, payload string) string {
16	m := hmac.New(sha256.New, secret)	38	m := hmac.New(sha256.New, secret)
17	m.Write([]byte(hostID))	39	m.Write([]byte(payload))
18	return hex.EncodeToString(m.Sum(nil))	40	return hex.EncodeToString(m.Sum(nil))
19	}	41	}
20		42
21	func Mint(secret []byte, hostID string) string {	43	// Mint returns a signed credential for hostID at the given generation.
22	return hostID + "." + sign(secret, hostID)	44	func Mint(secret []byte, hostID string, generation int64, issuedAt time.Time) string {
		45	payload := fmt.Sprintf("%s.%d.%d", hostID, generation, issuedAt.Unix())
		46	return payload + "." + sign(secret, payload)
23	}	47	}
24		48
25	func Verify(secret []byte, cred string) (hostID string, ok bool) {	49	// Verify parses and authenticates cred: the four signed fields
26	id, sig, found := strings.Cut(cred, ".")	50	// host_id.generation.issued_unix.hmac. Anything else fails.
27	if !found \|\| id == "" {	51	func Verify(secret []byte, cred string) (Claims, bool) {
28	return "", false	52	parts := strings.Split(cred, ".")
		53	if len(parts) != 4 \|\| parts[0] == "" {
		54	return Claims{}, false
		55	}
		56	gen, err := strconv.ParseInt(parts[1], 10, 64)
		57	if err != nil {
		58	return Claims{}, false
		59	}
		60	issued, err := strconv.ParseInt(parts[2], 10, 64)
		61	if err != nil {
		62	return Claims{}, false
29	}	63	}
30	if !hmac.Equal([]byte(sig), []byte(sign(secret, id))) {	64	payload := strings.Join(parts[:3], ".")
31	return "", false	65	if !hmac.Equal([]byte(parts[3]), []byte(sign(secret, payload))) {
		66	return Claims{}, false
32	}	67	}
33	return id, true	68	return Claims{HostID: parts[0], Generation: gen, IssuedAt: time.Unix(issued, 0)}, true
34	}	69	}

internal/server/hosttoken/hosttoken_generation_test.go

@@ @@ -0,0 +1,59 @@ @@
+package hosttoken
+import (
+	"strings"
+	"testing"
+	"time"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// TestMintVerifyRoundTrip pins the credential contract:
+// host_id.gen.issued_unix.hmac, HMAC over the three dotted fields.
+func TestMintVerifyRoundTrip(t *testing.T) {
+	secret := []byte("s3cret")
+	issued := time.Unix(1_750_000_000, 0)
+	cred := Mint(secret, "abc123", 3, issued)
+	claims, ok := Verify(secret, cred)
+	require.True(t, ok)
+	assert.Equal(t, "abc123", claims.HostID)
+	assert.Equal(t, int64(3), claims.Generation)
+	assert.True(t, claims.IssuedAt.Equal(issued))
+}
+// TestVerifyRejectsTampering: flipping any field invalidates the HMAC, and a
+// wrong secret never verifies.
+func TestVerifyRejectsTampering(t *testing.T) {
+	secret := []byte("s3cret")
+	cred := Mint(secret, "abc123", 1, time.Unix(1_750_000_000, 0))
+	parts := strings.Split(cred, ".")
+	require.Len(t, parts, 4)
+	for i, repl := range []string{"otherhost", "9", "1750000001"} {
+		mut := make([]string, 4)
+		copy(mut, parts)
+		mut[i] = repl
+		_, ok := Verify(secret, strings.Join(mut, "."))
+		assert.False(t, ok, "tampered field %d must fail", i)
+	}
+	_, ok := Verify([]byte("wrong"), cred)
+	assert.False(t, ok)
+}
+// TestVerifyRejectsMalformed: anything that is not exactly the four signed
+// fields is rejected.
+func TestVerifyRejectsMalformed(t *testing.T) {
+	secret := []byte("s3cret")
+	for _, cred := range []string{
+		"", "abc123", "a.b.c",   // too few parts
+		"abc123.deadbeef",       // two parts
+		"a.b.c.d.e",             // too many parts
+		"abc123.x.170.deadbeef", // non-numeric generation
+		"abc123.1.x.deadbeef",   // non-numeric issued_unix
+	} {
+		_, ok := Verify(secret, cred)
+		assert.False(t, ok, "must reject %q", cred)
+	}
+}

internal/server/hosttoken/hosttoken_test.go

Old		New
@@ -2,20 +2,22 @@ package hosttoken
2		2
3	import (	3	import (
4	"testing"	4	"testing"
		5	"time"
		6
5	"github.com/stretchr/testify/assert"	7	"github.com/stretchr/testify/assert"
6	)	8	)
7		9
8	func TestMintedCredentialVerifiesAndRecoversHostID(t *testing.T) {	10	func TestMintedCredentialVerifiesAndRecoversHostID(t *testing.T) {
9	secret := []byte("server-secret")	11	secret := []byte("server-secret")
10	cred := Mint(secret, "host-123")	12	cred := Mint(secret, "host-123", 1, time.Unix(1_750_000_000, 0))
11	hostID, ok := Verify(secret, cred)	13	claims, ok := Verify(secret, cred)
12	assert.True(t, ok)	14	assert.True(t, ok)
13	assert.Equal(t, "host-123", hostID)	15	assert.Equal(t, "host-123", claims.HostID)
14	}	16	}
15		17
16	func TestVerifyRejectsTamperedAndWrongSecret(t *testing.T) {	18	func TestVerifyRejectsTamperedAndWrongSecret(t *testing.T) {
17	secret := []byte("server-secret")	19	secret := []byte("server-secret")
18	cred := Mint(secret, "host-123")	20	cred := Mint(secret, "host-123", 1, time.Unix(1_750_000_000, 0))
19	_, ok := Verify([]byte("other"), cred)	21	_, ok := Verify([]byte("other"), cred)
20	assert.False(t, ok, "wrong secret")	22	assert.False(t, ok, "wrong secret")
21	_, ok = Verify(secret, cred+"x")	23	_, ok = Verify(secret, cred+"x")

internal/server/store/allocation_test.go

Old		New
@@ -48,7 +48,7 @@ func TestAllocatedByHostSeparatesHosts(t *testing.T) {
48	s := newStore(t)	48	s := newStore(t)
49	h1 := enrollHost(t, s)	49	h1 := enrollHost(t, s)
50	tok, _ := s.CreateEnrollmentToken()	50	tok, _ := s.CreateEnrollmentToken()
51	h2, _ := s.RedeemEnrollmentToken(tok, "h2", "linux", "amd64", "cloudhv", "")	51	h2, _ := s.RedeemEnrollmentToken(tok, "h2", "linux", "amd64", "cloudhv", "", "")
52	vmWithResources(t, s, h1, "a", 2, 2048, 10)	52	vmWithResources(t, s, h1, "a", 2, 2048, 10)
53	vmWithResources(t, s, h2, "b", 8, 8192, 40)	53	vmWithResources(t, s, h2, "b", 8, 8192, 40)
54		54

internal/server/store/decommission_test.go

Old		New
@@ -49,7 +49,7 @@ func TestRemoveHostFreesCIDRForReuse(t *testing.T) {
49	h1 := enrollHost(t, s)	49	h1 := enrollHost(t, s)
50	assert.Equal(t, "10.77.1.0/24", h1.BridgeCIDR)	50	assert.Equal(t, "10.77.1.0/24", h1.BridgeCIDR)
51	tok2, _ := s.CreateEnrollmentToken()	51	tok2, _ := s.CreateEnrollmentToken()
52	h2, _ := s.RedeemEnrollmentToken(tok2, "b", "linux", "amd64", "cloudhv", "")	52	h2, _ := s.RedeemEnrollmentToken(tok2, "b", "linux", "amd64", "cloudhv", "", "")
53	assert.Equal(t, "10.77.2.0/24", h2.BridgeCIDR)	53	assert.Equal(t, "10.77.2.0/24", h2.BridgeCIDR)
54		54
55	// Decommission h1 and simulate the agent reaping its VMs (hard-delete).	55	// Decommission h1 and simulate the agent reaping its VMs (hard-delete).
@@ -64,7 +64,7 @@ func TestRemoveHostFreesCIDRForReuse(t *testing.T) {
64		64
65	// A new enrollment reuses h1's freed CIDR rather than allocating a fresh one.	65	// A new enrollment reuses h1's freed CIDR rather than allocating a fresh one.
66	tok3, _ := s.CreateEnrollmentToken()	66	tok3, _ := s.CreateEnrollmentToken()
67	h3, err := s.RedeemEnrollmentToken(tok3, "c", "linux", "amd64", "cloudhv", "")	67	h3, err := s.RedeemEnrollmentToken(tok3, "c", "linux", "amd64", "cloudhv", "", "")
68	require.NoError(t, err)	68	require.NoError(t, err)
69	assert.Equal(t, "10.77.1.0/24", h3.BridgeCIDR, "freed CIDR should be reused")	69	assert.Equal(t, "10.77.1.0/24", h3.BridgeCIDR, "freed CIDR should be reused")
70	}	70	}

internal/server/store/store.go

Old		New
@@ -9,11 +9,13 @@ import (
9	"crypto/sha256"	9	"crypto/sha256"
10	"database/sql"	10	"database/sql"
11	"encoding/hex"	11	"encoding/hex"
		12	"encoding/json"
12	"errors"	13	"errors"
13	"fmt"	14	"fmt"
14	"net/netip"	15	"net/netip"
15	"os"	16	"os"
16	"path/filepath"	17	"path/filepath"
		18	"strconv"
17	"strings"	19	"strings"
18	"time"	20	"time"
19		21
@@ -42,7 +44,11 @@ type Store struct {
42		44
43	type Host struct {	45	type Host struct {
44	ID, Name, OS, Arch, Provisioner, Overlay, BridgeCIDR, Status string	46	ID, Name, OS, Arch, Provisioner, Overlay, BridgeCIDR, Status string
45	EnrolledAt time.Time	47	// CredGeneration is the host's current credential generation. Credentials
		48	// minted at an older generation are rejected — bumping it revokes that
		49	// one host's outstanding credential without rotating the fleet secret.
		50	CredGeneration int64
		51	EnrolledAt time.Time
46	}	52	}
47		53
48	type VM struct {	54	type VM struct {
@@ -70,7 +76,8 @@ CREATE TABLE IF NOT EXISTS hosts (
70	overlay TEXT NOT NULL DEFAULT 'tailscale',	76	overlay TEXT NOT NULL DEFAULT 'tailscale',
71	bridge_cidr TEXT NOT NULL,	77	bridge_cidr TEXT NOT NULL,
72	status TEXT NOT NULL DEFAULT 'enrolled',	78	status TEXT NOT NULL DEFAULT 'enrolled',
73	enrolled_at DATETIME NOT NULL	79	enrolled_at DATETIME NOT NULL,
		80	cred_generation INTEGER NOT NULL DEFAULT 1
74	);	81	);
75		82
76	CREATE TABLE IF NOT EXISTS enrollment_tokens (	83	CREATE TABLE IF NOT EXISTS enrollment_tokens (
@@ -107,6 +114,19 @@ CREATE TABLE IF NOT EXISTS freed_cidrs (
107	bridge_cidr TEXT PRIMARY KEY	114	bridge_cidr TEXT PRIMARY KEY
108	);	115	);
109		116
		117	-- append-only operational audit trail (enrollment, decommission). Read via
		118	-- ListAudit / GET /api/v1/audit; rows are never UPDATEd, and the only DELETE
		119	-- is retention pruning (PruneAudit, driven by the server's audit_retention
		120	-- config). The at column is second-precision UTC RFC3339, which makes
		121	-- lexicographic comparison chronological -- PruneAudit's DELETE relies on
		122	-- every writer keeping that format.
		123	CREATE TABLE IF NOT EXISTS audit_log (
		124	id INTEGER PRIMARY KEY AUTOINCREMENT,
		125	at DATETIME NOT NULL,
		126	action TEXT NOT NULL,
		127	detail TEXT NOT NULL DEFAULT ''
		128	);
		129
110	INSERT INTO meta(key, value) VALUES ('epoch', '0') ON CONFLICT DO NOTHING;	130	INSERT INTO meta(key, value) VALUES ('epoch', '0') ON CONFLICT DO NOTHING;
111	INSERT INTO meta(key, value) VALUES ('next_cidr_index', '1') ON CONFLICT DO NOTHING;	131	INSERT INTO meta(key, value) VALUES ('next_cidr_index', '1') ON CONFLICT DO NOTHING;
112	`	132	`
@@ -180,7 +200,11 @@ func (s *Store) CreateEnrollmentToken() (string, error) {
180	return tok, nil	200	return tok, nil
181	}	201	}
182		202
183	func (s *Store) RedeemEnrollmentToken(tok, name, osName, arch, provisioner, overlay string) (Host, error) {	203	// RedeemEnrollmentToken atomically consumes tok and creates the host row.
		204	// remote (the enrolling client's IP) is recorded in a host.enroll audit row
		205	// written in the SAME transaction, so an enrolled host can never exist
		206	// without its durable audit record.
		207	func (s *Store) RedeemEnrollmentToken(tok, name, osName, arch, provisioner, overlay, remote string) (Host, error) {
184	if overlay == "" {	208	if overlay == "" {
185	overlay = "tailscale"	209	overlay = "tailscale"
186	}	210	}
@@ -248,6 +272,19 @@ func (s *Store) RedeemEnrollmentToken(tok, name, osName, arch, provisioner, over
248	return Host{}, fmt.Errorf("insert host: %w", err)	272	return Host{}, fmt.Errorf("insert host: %w", err)
249	}	273	}
250		274
		275	// Audit in the SAME tx: the durable record of who enrolled cannot be lost
		276	// once the enrollment itself commits. The token appears only as a hash
		277	// prefix (matches the enrollment_tokens.token_hash the mint row logs).
		278	tokSum := sha256.Sum256([]byte(tok))
		279	detail, _ := json.Marshal(map[string]string{
		280	"host_id": id, "name": name, "os": osName, "arch": arch,
		281	"remote": remote, "token_hash_prefix": hex.EncodeToString(tokSum[:])[:8],
		282	})
		283	if _, err := tx.Exec(`INSERT INTO audit_log(at, action, detail) VALUES (?, ?, ?)`,
		284	now.Format(time.RFC3339), "host.enroll", string(detail)); err != nil {
		285	return Host{}, fmt.Errorf("audit enroll: %w", err)
		286	}
		287
251	if err := tx.Commit(); err != nil {	288	if err := tx.Commit(); err != nil {
252	return Host{}, err	289	return Host{}, err
253	}	290	}
@@ -260,8 +297,9 @@ func (s *Store) RedeemEnrollmentToken(tok, name, osName, arch, provisioner, over
260	Provisioner: provisioner,	297	Provisioner: provisioner,
261	Overlay: overlay,	298	Overlay: overlay,
262	BridgeCIDR: bridgeCIDR,	299	BridgeCIDR: bridgeCIDR,
263	Status: "enrolled",	300	Status: "enrolled",
264	EnrolledAt: now,	301	CredGeneration: 1,
		302	EnrolledAt: now,
265	}, nil	303	}, nil
266	}	304	}
267		305
@@ -269,8 +307,8 @@ func (s *Store) GetHost(id string) (Host, error) {
269	var h Host	307	var h Host
270	var enrolledAt string	308	var enrolledAt string
271	err := s.db.QueryRow(	309	err := s.db.QueryRow(
272	`SELECT id, name, os, arch, provisioner, overlay, bridge_cidr, status, enrolled_at FROM hosts WHERE id=?`, id,	310	`SELECT id, name, os, arch, provisioner, overlay, bridge_cidr, status, enrolled_at, cred_generation FROM hosts WHERE id=?`, id,
273	).Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.Overlay, &h.BridgeCIDR, &h.Status, &enrolledAt)	311	).Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.Overlay, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration)
274	if err != nil {	312	if err != nil {
275	return Host{}, err	313	return Host{}, err
276	}	314	}
@@ -278,8 +316,43 @@ func (s *Store) GetHost(id string) (Host, error) {
278	return h, nil	316	return h, nil
279	}	317	}
280		318
281	func (s *Store) ListHosts() ([]Host, error) {	319	// BumpCredGeneration increments the host's credential generation, revoking
282	rows, err := s.db.Query(`SELECT id, name, os, arch, provisioner, overlay, bridge_cidr, status, enrolled_at FROM hosts`)	320	// every credential minted at the previous generation, and returns the new
		321	// value. The host.credential.revoke audit row is written in the SAME
		322	// transaction — a security action must not be able to happen unrecorded.
		323	// Errors (sql.ErrNoRows via %w) if the host does not exist.
		324	func (s *Store) BumpCredGeneration(id, remote string) (int64, error) {
		325	tx, err := s.db.Begin()
		326	if err != nil {
		327	return 0, err
		328	}
		329	defer tx.Rollback()
		330	var gen int64
		331	if err := tx.QueryRow(
		332	`UPDATE hosts SET cred_generation = cred_generation + 1 WHERE id=? RETURNING cred_generation`, id,
		333	).Scan(&gen); err != nil {
		334	return 0, fmt.Errorf("bump cred_generation for %s: %w", id, err)
		335	}
		336	detail, _ := json.Marshal(map[string]string{
		337	"host_id": id, "new_generation": strconv.FormatInt(gen, 10), "remote": remote,
		338	})
		339	if _, err := tx.Exec(`INSERT INTO audit_log(at, action, detail) VALUES (?, ?, ?)`,
		340	time.Now().UTC().Format(time.RFC3339), "host.credential.revoke", string(detail)); err != nil {
		341	return 0, fmt.Errorf("audit revoke: %w", err)
		342	}
		343	return gen, tx.Commit()
		344	}
		345
		346	// querier is the subset of sql.DB / sql.Tx the list helpers need, so the
		347	// same scan logic serves both the standalone reads and Snapshot's single-tx read.
		348	type querier interface {
		349	Query(query string, args ...any) (*sql.Rows, error)
		350	}
		351
		352	func (s *Store) ListHosts() ([]Host, error) { return listHosts(s.db) }
		353
		354	func listHosts(q querier) ([]Host, error) {
		355	rows, err := q.Query(`SELECT id, name, os, arch, provisioner, overlay, bridge_cidr, status, enrolled_at, cred_generation FROM hosts`)
283	if err != nil {	356	if err != nil {
284	return nil, err	357	return nil, err
285	}	358	}
@@ -288,7 +361,7 @@ func (s *Store) ListHosts() ([]Host, error) {
288	for rows.Next() {	361	for rows.Next() {
289	var h Host	362	var h Host
290	var enrolledAt string	363	var enrolledAt string
291	if err := rows.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.Overlay, &h.BridgeCIDR, &h.Status, &enrolledAt); err != nil {	364	if err := rows.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.Overlay, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration); err != nil {
292	return nil, err	365	return nil, err
293	}	366	}
294	h.EnrolledAt, _ = time.Parse(time.RFC3339, enrolledAt)	367	h.EnrolledAt, _ = time.Parse(time.RFC3339, enrolledAt)
@@ -406,8 +479,10 @@ type Alloc struct{ VCPUs, MemMB, DiskGB int64 }
406		479
407	// AllocatedByHost returns, per host, the resources allocated to its live	480	// AllocatedByHost returns, per host, the resources allocated to its live
408	// (non-tombstoned) VMs. Hosts with no live VMs are absent from the map.	481	// (non-tombstoned) VMs. Hosts with no live VMs are absent from the map.
409	func (s *Store) AllocatedByHost() (map[string]Alloc, error) {	482	func (s *Store) AllocatedByHost() (map[string]Alloc, error) { return allocatedByHost(s.db) }
410	rows, err := s.db.Query(`	483
		484	func allocatedByHost(q querier) (map[string]Alloc, error) {
		485	rows, err := q.Query(`
411	SELECT host_id, COALESCE(SUM(vcpus),0), COALESCE(SUM(mem_mb),0), COALESCE(SUM(disk_gb),0)	486	SELECT host_id, COALESCE(SUM(vcpus),0), COALESCE(SUM(mem_mb),0), COALESCE(SUM(disk_gb),0)
412	FROM vms WHERE deleted_at IS NULL GROUP BY host_id`)	487	FROM vms WHERE deleted_at IS NULL GROUP BY host_id`)
413	if err != nil {	488	if err != nil {
@@ -426,6 +501,58 @@ func (s *Store) AllocatedByHost() (map[string]Alloc, error) {
426	return out, rows.Err()	501	return out, rows.Err()
427	}	502	}
428		503
		504	// AuditEntry is one row of the append-only audit trail.
		505	type AuditEntry struct {
		506	At time.Time
		507	Action string
		508	Detail string
		509	}
		510
		511	// AppendAudit records an audit event. detail is a small JSON blob; keep
		512	// secrets out (hash prefixes, not tokens).
		513	func (s *Store) AppendAudit(action, detail string) error {
		514	_, err := s.db.Exec(`INSERT INTO audit_log(at, action, detail) VALUES (?, ?, ?)`,
		515	time.Now().UTC().Format(time.RFC3339), action, detail)
		516	return err
		517	}
		518
		519	// ListAudit returns up to limit audit entries, newest first.
		520	func (s *Store) ListAudit(limit int) ([]AuditEntry, error) {
		521	rows, err := s.db.Query(`SELECT at, action, detail FROM audit_log ORDER BY id DESC LIMIT ?`, limit)
		522	if err != nil {
		523	return nil, err
		524	}
		525	defer rows.Close()
		526	var out []AuditEntry
		527	for rows.Next() {
		528	var e AuditEntry
		529	var at string
		530	if err := rows.Scan(&at, &e.Action, &e.Detail); err != nil {
		531	return nil, err
		532	}
		533	e.At, _ = time.Parse(time.RFC3339, at)
		534	out = append(out, e)
		535	}
		536	return out, rows.Err()
		537	}
		538
		539	// PruneAudit deletes audit rows older than olderThan and reports how many
		540	// were removed. Retention keeps the append-only log bounded; the caller
		541	// (eitri-server) runs it at startup and daily.
		542	func (s *Store) PruneAudit(olderThan time.Duration) (int64, error) {
		543	// Guard the contract, not just the caller: a zero/negative window would
		544	// compute a now-or-future cutoff and wipe the entire forensic trail.
		545	if olderThan <= 0 {
		546	return 0, nil
		547	}
		548	cutoff := time.Now().UTC().Add(-olderThan).Format(time.RFC3339)
		549	res, err := s.db.Exec(`DELETE FROM audit_log WHERE at < ?`, cutoff)
		550	if err != nil {
		551	return 0, err
		552	}
		553	return res.RowsAffected()
		554	}
		555
429	// HostVMCount returns the number of VM rows for a host (live + tombstoned).	556	// HostVMCount returns the number of VM rows for a host (live + tombstoned).
430	// Rows are hard-deleted only after the agent acks destroy, so a count of 0 means	557	// Rows are hard-deleted only after the agent acks destroy, so a count of 0 means
431	// the host is fully drained.	558	// the host is fully drained.
@@ -532,8 +659,10 @@ func scanVM(rows *sql.Rows) (VM, error) {
532	return vm, nil	659	return vm, nil
533	}	660	}
534		661
535	func (s *Store) ListVMs() ([]VM, error) {	662	func (s *Store) ListVMs() ([]VM, error) { return listVMs(s.db) }
536	rows, err := s.db.Query(	663
		664	func listVMs(q querier) ([]VM, error) {
		665	rows, err := q.Query(
537	`SELECT id, host_id, name, image_url, image_sha256, cloud_init, ssh_authorized_key,	666	`SELECT id, host_id, name, image_url, image_sha256, cloud_init, ssh_authorized_key,
538	vcpus, mem_mb, disk_gb, persistent, power_state, status, last_error, assigned_ip,	667	vcpus, mem_mb, disk_gb, persistent, power_state, status, last_error, assigned_ip,
539	created_at, deleted_at FROM vms`,	668	created_at, deleted_at FROM vms`,
@@ -553,6 +682,31 @@ func (s *Store) ListVMs() ([]VM, error) {
553	return vms, rows.Err()	682	return vms, rows.Err()
554	}	683	}
555		684
		685	// Snapshot reads hosts, per-host allocation, and VMs in a single read
		686	// transaction, so the trio is mutually consistent — a concurrent desired-state
		687	// mutation between the reads cannot produce a payload mixing two epochs.
		688	// The SSE stream builds its fleet snapshot from this.
		689	func (s *Store) Snapshot() ([]Host, map[string]Alloc, []VM, error) {
		690	tx, err := s.db.Begin()
		691	if err != nil {
		692	return nil, nil, nil, err
		693	}
		694	defer tx.Rollback()
		695	hosts, err := listHosts(tx)
		696	if err != nil {
		697	return nil, nil, nil, err
		698	}
		699	alloc, err := allocatedByHost(tx)
		700	if err != nil {
		701	return nil, nil, nil, err
		702	}
		703	vms, err := listVMs(tx)
		704	if err != nil {
		705	return nil, nil, nil, err
		706	}
		707	return hosts, alloc, vms, tx.Commit()
		708	}
		709
556	func (s *Store) DesiredForHost(hostID string) (uint64, []VM, error) {	710	func (s *Store) DesiredForHost(hostID string) (uint64, []VM, error) {
557	tx, err := s.db.Begin()	711	tx, err := s.db.Begin()
558	if err != nil {	712	if err != nil {

internal/server/store/store_test.go

Old		New
@@ -3,6 +3,7 @@ package store
3	import (	3	import (
4	"path/filepath"	4	"path/filepath"
5	"testing"	5	"testing"
		6	"time"
6		7
7	"github.com/stretchr/testify/assert"	8	"github.com/stretchr/testify/assert"
8	"github.com/stretchr/testify/require"	9	"github.com/stretchr/testify/require"
@@ -20,7 +21,7 @@ func enrollHost(t testing.T, s Store) Host {
20	t.Helper()	21	t.Helper()
21	tok, err := s.CreateEnrollmentToken()	22	tok, err := s.CreateEnrollmentToken()
22	require.NoError(t, err)	23	require.NoError(t, err)
23	h, err := s.RedeemEnrollmentToken(tok, "host-a", "linux", "amd64", "cloudhv", "")	24	h, err := s.RedeemEnrollmentToken(tok, "host-a", "linux", "amd64", "cloudhv", "", "")
24	require.NoError(t, err)	25	require.NoError(t, err)
25	return h	26	return h
26	}	27	}
@@ -28,15 +29,15 @@ func enrollHost(t testing.T, s Store) Host {
28	func TestEnrollmentAssignsSequentialCIDRsAndIsOneTimeUse(t *testing.T) {	29	func TestEnrollmentAssignsSequentialCIDRsAndIsOneTimeUse(t *testing.T) {
29	s := newStore(t)	30	s := newStore(t)
30	tok1, _ := s.CreateEnrollmentToken()	31	tok1, _ := s.CreateEnrollmentToken()
31	h1, err := s.RedeemEnrollmentToken(tok1, "a", "linux", "amd64", "cloudhv", "")	32	h1, err := s.RedeemEnrollmentToken(tok1, "a", "linux", "amd64", "cloudhv", "", "")
32	require.NoError(t, err)	33	require.NoError(t, err)
33	assert.Equal(t, "10.77.1.0/24", h1.BridgeCIDR)	34	assert.Equal(t, "10.77.1.0/24", h1.BridgeCIDR)
34		35
35	_, err = s.RedeemEnrollmentToken(tok1, "b", "linux", "amd64", "cloudhv", "")	36	_, err = s.RedeemEnrollmentToken(tok1, "b", "linux", "amd64", "cloudhv", "", "")
36	assert.Error(t, err, "token must be one-time use")	37	assert.Error(t, err, "token must be one-time use")
37		38
38	tok2, _ := s.CreateEnrollmentToken()	39	tok2, _ := s.CreateEnrollmentToken()
39	h2, _ := s.RedeemEnrollmentToken(tok2, "b", "linux", "amd64", "cloudhv", "")	40	h2, _ := s.RedeemEnrollmentToken(tok2, "b", "linux", "amd64", "cloudhv", "", "")
40	assert.Equal(t, "10.77.2.0/24", h2.BridgeCIDR)	41	assert.Equal(t, "10.77.2.0/24", h2.BridgeCIDR)
41	}	42	}
42		43
@@ -110,7 +111,7 @@ func TestEnrollmentCIDRWorksForNonSlash16Pools(t *testing.T) {
110	require.NoError(t, err)	111	require.NoError(t, err)
111	defer s.Close()	112	defer s.Close()
112	tok, _ := s.CreateEnrollmentToken()	113	tok, _ := s.CreateEnrollmentToken()
113	h, err := s.RedeemEnrollmentToken(tok, "a", "linux", "amd64", "cloudhv", "")	114	h, err := s.RedeemEnrollmentToken(tok, "a", "linux", "amd64", "cloudhv", "", "")
114	require.NoError(t, err)	115	require.NoError(t, err)
115	assert.Equal(t, "192.168.5.0/24", h.BridgeCIDR, "1st /24 within the pool, 0th reserved")	116	assert.Equal(t, "192.168.5.0/24", h.BridgeCIDR, "1st /24 within the pool, 0th reserved")
116	}	117	}
@@ -120,11 +121,11 @@ func TestEnrollmentFailsWhenPoolExhausted(t *testing.T) {
120	require.NoError(t, err)	121	require.NoError(t, err)
121	defer s.Close()	122	defer s.Close()
122	tok1, _ := s.CreateEnrollmentToken()	123	tok1, _ := s.CreateEnrollmentToken()
123	h, err := s.RedeemEnrollmentToken(tok1, "a", "linux", "amd64", "cloudhv", "")	124	h, err := s.RedeemEnrollmentToken(tok1, "a", "linux", "amd64", "cloudhv", "", "")
124	require.NoError(t, err)	125	require.NoError(t, err)
125	assert.Equal(t, "10.9.9.0/24", h.BridgeCIDR)	126	assert.Equal(t, "10.9.9.0/24", h.BridgeCIDR)
126	tok2, _ := s.CreateEnrollmentToken()	127	tok2, _ := s.CreateEnrollmentToken()
127	_, err = s.RedeemEnrollmentToken(tok2, "b", "linux", "amd64", "cloudhv", "")	128	_, err = s.RedeemEnrollmentToken(tok2, "b", "linux", "amd64", "cloudhv", "", "")
128	assert.ErrorContains(t, err, "exhausted")	129	assert.ErrorContains(t, err, "exhausted")
129	}	130	}
130		131
@@ -140,13 +141,13 @@ func TestRedeemEnrollmentToken_OverlayPersistedAndDefaultsTailscale(t *testing.T
140	s := newStore(t)	141	s := newStore(t)
141	tok, _ := s.CreateEnrollmentToken()	142	tok, _ := s.CreateEnrollmentToken()
142	// Explicit overlay="none" must be persisted.	143	// Explicit overlay="none" must be persisted.
143	h, err := s.RedeemEnrollmentToken(tok, "host-b", "linux", "amd64", "cloudhv", "none")	144	h, err := s.RedeemEnrollmentToken(tok, "host-b", "linux", "amd64", "cloudhv", "none", "")
144	require.NoError(t, err)	145	require.NoError(t, err)
145	assert.Equal(t, "none", h.Overlay, "overlay must be 'none' as requested")	146	assert.Equal(t, "none", h.Overlay, "overlay must be 'none' as requested")
146		147
147	// Empty overlay → defaults to "tailscale".	148	// Empty overlay → defaults to "tailscale".
148	tok2, _ := s.CreateEnrollmentToken()	149	tok2, _ := s.CreateEnrollmentToken()
149	h2, err := s.RedeemEnrollmentToken(tok2, "host-c", "linux", "amd64", "cloudhv", "")	150	h2, err := s.RedeemEnrollmentToken(tok2, "host-c", "linux", "amd64", "cloudhv", "", "")
150	require.NoError(t, err)	151	require.NoError(t, err)
151	assert.Equal(t, "tailscale", h2.Overlay, "empty overlay must default to 'tailscale'")	152	assert.Equal(t, "tailscale", h2.Overlay, "empty overlay must default to 'tailscale'")
152	}	153	}
@@ -154,7 +155,7 @@ func TestRedeemEnrollmentToken_OverlayPersistedAndDefaultsTailscale(t *testing.T
154	func TestListHosts_ReturnsOverlay(t *testing.T) {	155	func TestListHosts_ReturnsOverlay(t *testing.T) {
155	s := newStore(t)	156	s := newStore(t)
156	tok, _ := s.CreateEnrollmentToken()	157	tok, _ := s.CreateEnrollmentToken()
157	_, err := s.RedeemEnrollmentToken(tok, "host-x", "linux", "amd64", "cloudhv", "none")	158	_, err := s.RedeemEnrollmentToken(tok, "host-x", "linux", "amd64", "cloudhv", "none", "")
158	require.NoError(t, err)	159	require.NoError(t, err)
159	hosts, err := s.ListHosts()	160	hosts, err := s.ListHosts()
160	require.NoError(t, err)	161	require.NoError(t, err)
@@ -181,3 +182,185 @@ func TestServerCertLoadOrCreatePersists(t *testing.T) {
181	require.NoError(t, err)	182	require.NoError(t, err)
182	require.NotEmpty(t, key)	183	require.NotEmpty(t, key)
183	}	184	}
		185
		186	// TestSnapshotReadsAllThreeConsistently pins the Snapshot contract: one call
		187	// returns hosts, per-host allocation, and VMs equal to the individual reads —
		188	// but sourced from a single read transaction so the trio can never mix state
		189	// from two different epochs (the SSE stream builds its payload from this).
		190	func TestSnapshotReadsAllThreeConsistently(t *testing.T) {
		191	s := newStore(t)
		192	h := enrollHost(t, s)
		193	require.NoError(t, s.CreateVM(VM{ID: "vm1", HostID: h.ID, Name: "one",
		194	ImageURL: "http://x/i", ImageSHA256: "abc", VCPUs: 2, MemMB: 1024, DiskGB: 10, PowerState: "running"}))
		195	require.NoError(t, s.CreateVM(VM{ID: "vm2", HostID: h.ID, Name: "two",
		196	ImageURL: "http://x/i", ImageSHA256: "abc", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "stopped"}))
		197
		198	hosts, alloc, vms, err := s.Snapshot()
		199	require.NoError(t, err)
		200
		201	wantHosts, err := s.ListHosts()
		202	require.NoError(t, err)
		203	wantAlloc, err := s.AllocatedByHost()
		204	require.NoError(t, err)
		205	wantVMs, err := s.ListVMs()
		206	require.NoError(t, err)
		207
		208	assert.Equal(t, wantHosts, hosts)
		209	assert.Equal(t, wantAlloc, alloc)
		210	assert.Equal(t, wantVMs, vms)
		211	assert.Equal(t, Alloc{VCPUs: 3, MemMB: 1536, DiskGB: 15}, alloc[h.ID])
		212	}
		213
		214	// TestSnapshotIsAtomicUnderConcurrentWrites pins the single-tx property
		215	// itself: alloc is derivable from vms, so any snapshot whose alloc disagrees
		216	// with its own vms mixed two epochs. A mutator hammers create/tombstone while
		217	// the main goroutine snapshots; the old three-call implementation released
		218	// the sole connection between reads and fails this with high probability.
		219	func TestSnapshotIsAtomicUnderConcurrentWrites(t *testing.T) {
		220	s := newStore(t)
		221	h := enrollHost(t, s)
		222
		223	done := make(chan struct{})
		224	go func() {
		225	defer close(done)
		226	for i := 0; i < 300; i++ {
		227	id := RandHex(8)
		228	_ = s.CreateVM(VM{ID: id, HostID: h.ID, Name: "vm-" + id,
		229	ImageURL: "http://x/i", ImageSHA256: "abc",
		230	VCPUs: 1, MemMB: 256, DiskGB: 1, PowerState: "running"})
		231	_ = s.TombstoneVM(id)
		232	}
		233	}()
		234
		235	for i := 0; i < 100; i++ {
		236	_, alloc, vms, err := s.Snapshot()
		237	require.NoError(t, err)
		238	derived := map[string]Alloc{}
		239	for _, vm := range vms {
		240	if vm.DeletedAt == nil {
		241	a := derived[vm.HostID]
		242	a.VCPUs += vm.VCPUs
		243	a.MemMB += vm.MemMB
		244	a.DiskGB += vm.DiskGB
		245	derived[vm.HostID] = a
		246	}
		247	}
		248	require.Equal(t, derived, alloc,
		249	"snapshot %d: alloc disagrees with its own vms — reads mixed two epochs", i)
		250	}
		251	<-done
		252	}
		253
		254	// TestAuditLogRoundTrip pins the audit trail contract: AppendAudit writes a
		255	// timestamped action+detail row; ListAudit returns newest-first with a limit.
		256	func TestAuditLogRoundTrip(t *testing.T) {
		257	s := newStore(t)
		258	require.NoError(t, s.AppendAudit("enroll-token.mint", `{"token_hash_prefix":"abcd1234"}`))
		259	require.NoError(t, s.AppendAudit("host.enroll", `{"host_id":"h1","name":"host-a"}`))
		260
		261	rows, err := s.ListAudit(10)
		262	require.NoError(t, err)
		263	require.Len(t, rows, 2)
		264	assert.Equal(t, "host.enroll", rows[0].Action, "newest first")
		265	assert.Contains(t, rows[0].Detail, "h1")
		266	assert.False(t, rows[0].At.IsZero())
		267
		268	one, err := s.ListAudit(1)
		269	require.NoError(t, err)
		270	require.Len(t, one, 1)
		271	assert.Equal(t, "host.enroll", one[0].Action)
		272	}
		273
		274	// TestRedeemWritesAuditRowAtomically pins audit durability: the host.enroll
		275	// audit row is written inside the SAME transaction as the redeem, so an
		276	// enrolled host can never exist without its durable audit record.
		277	func TestRedeemWritesAuditRowAtomically(t *testing.T) {
		278	s := newStore(t)
		279	tok, err := s.CreateEnrollmentToken()
		280	require.NoError(t, err)
		281	h, err := s.RedeemEnrollmentToken(tok, "host-a", "linux", "amd64", "cloudhv", "", "192.0.2.9")
		282	require.NoError(t, err)
		283
		284	rows, err := s.ListAudit(5)
		285	require.NoError(t, err)
		286	require.NotEmpty(t, rows)
		287	assert.Equal(t, "host.enroll", rows[0].Action)
		288	assert.Contains(t, rows[0].Detail, h.ID)
		289	assert.Contains(t, rows[0].Detail, "192.0.2.9")
		290	assert.NotContains(t, rows[0].Detail, tok, "raw token must never reach the audit log")
		291	}
		292
		293	// TestCredGenerationLifecycle pins per-host credential revocation: hosts
		294	// enroll at generation 1; BumpCredGeneration invalidates outstanding
		295	// credentials by incrementing the row; GetHost/ListHosts expose it.
		296	func TestCredGenerationLifecycle(t *testing.T) {
		297	s := newStore(t)
		298	h := enrollHost(t, s)
		299	assert.Equal(t, int64(1), h.CredGeneration, "fresh enrollment starts at generation 1")
		300
		301	got, err := s.GetHost(h.ID)
		302	require.NoError(t, err)
		303	assert.Equal(t, int64(1), got.CredGeneration)
		304
		305	gen, err := s.BumpCredGeneration(h.ID, "192.0.2.7")
		306	require.NoError(t, err)
		307	assert.Equal(t, int64(2), gen)
		308
		309	got, err = s.GetHost(h.ID)
		310	require.NoError(t, err)
		311	assert.Equal(t, int64(2), got.CredGeneration)
		312
		313	_, err = s.BumpCredGeneration("no-such-host", "")
		314	assert.Error(t, err, "unknown host must error")
		315	}
		316
		317
		318	// TestBumpCredGenerationAuditsAtomically pins that the revoke audit row is
		319	// written in the same transaction as the bump.
		320	func TestBumpCredGenerationAuditsAtomically(t *testing.T) {
		321	s := newStore(t)
		322	h := enrollHost(t, s)
		323	_, err := s.BumpCredGeneration(h.ID, "192.0.2.7")
		324	require.NoError(t, err)
		325	rows, err := s.ListAudit(1)
		326	require.NoError(t, err)
		327	require.Len(t, rows, 1)
		328	assert.Equal(t, "host.credential.revoke", rows[0].Action)
		329	assert.Contains(t, rows[0].Detail, h.ID)
		330	assert.Contains(t, rows[0].Detail, "192.0.2.7")
		331	}
		332
		333	// TestPruneAuditRemovesOnlyOldRows pins retention: rows older than the
		334	// window are deleted, newer rows survive, and the count is reported.
		335	func TestPruneAuditRemovesOnlyOldRows(t *testing.T) {
		336	s := newStore(t)
		337	// Insert directly so the timestamps are controlled.
		338	old := time.Now().UTC().Add(-100 * 24 * time.Hour).Format(time.RFC3339)
		339	_, err := s.db.Exec(`INSERT INTO audit_log(at, action, detail) VALUES (?, 'old.event', '{}')`, old)
		340	require.NoError(t, err)
		341	require.NoError(t, s.AppendAudit("new.event", "{}"))
		342
		343	n, err := s.PruneAudit(90 * 24 * time.Hour)
		344	require.NoError(t, err)
		345	assert.Equal(t, int64(1), n, "exactly the old row pruned")
		346
		347	rows, err := s.ListAudit(10)
		348	require.NoError(t, err)
		349	require.Len(t, rows, 1)
		350	assert.Equal(t, "new.event", rows[0].Action)
		351	}
		352
		353	// TestPruneAuditGuardsNonPositiveWindow pins the store-level contract: a
		354	// zero/negative retention must never delete anything.
		355	func TestPruneAuditGuardsNonPositiveWindow(t *testing.T) {
		356	s := newStore(t)
		357	require.NoError(t, s.AppendAudit("keep.me", "{}"))
		358	for _, d := range []time.Duration{0, -time.Hour} {
		359	n, err := s.PruneAudit(d)
		360	require.NoError(t, err)
		361	assert.Zero(t, n)
		362	}
		363	rows, err := s.ListAudit(5)
		364	require.NoError(t, err)
		365	assert.Len(t, rows, 1, "nothing may be deleted by a non-positive window")
		366	}

internal/server/syncsvc/syncsvc.go

Old		New
@@ -30,23 +30,29 @@ type Service struct {
30	reg *registry.Registry	30	reg *registry.Registry
31	hub *hub.Hub	31	hub *hub.Hub
32	secret []byte	32	secret []byte
		33	// maxCredAge, when non-zero, rejects credentials whose issued-at is older.
		34	// Zero disables the age check (default: expiry without an auto-renewal
		35	// channel would force periodic re-enrolls; per-host generation revocation
		36	// is the primary mechanism, max-age is opt-in defense-in-depth).
		37	maxCredAge time.Duration
33	// writeTimeout bounds each down-stream snapshot write (see defaultWriteTimeout).	38	// writeTimeout bounds each down-stream snapshot write (see defaultWriteTimeout).
34	writeTimeout time.Duration	39	writeTimeout time.Duration
35	}	40	}
36		41
37	// New constructs a Service with the production-default down-stream write timeout.	42	// New constructs a Service with the production-default down-stream write
38	func New(st store.Store, reg registry.Registry, h hub.Hub, secret []byte) Service {	43	// timeout. maxCredAge zero disables the credential age check.
39	return newWithWriteTimeout(st, reg, h, secret, defaultWriteTimeout)	44	func New(st store.Store, reg registry.Registry, h hub.Hub, secret []byte, maxCredAge time.Duration) Service {
		45	return newWithWriteTimeout(st, reg, h, secret, maxCredAge, defaultWriteTimeout)
40	}	46	}
41		47
42	// newWithWriteTimeout constructs a Service with an explicit down-stream write	48	// newWithWriteTimeout constructs a Service with an explicit down-stream write
43	// timeout. A zero timeout falls back to defaultWriteTimeout. Tests use this to	49	// timeout. A zero timeout falls back to defaultWriteTimeout. Tests use this to
44	// inject a short timeout; New keeps the public signature unchanged.	50	// inject a short timeout; New keeps the public signature unchanged.
45	func newWithWriteTimeout(st store.Store, reg registry.Registry, h hub.Hub, secret []byte, writeTimeout time.Duration) Service {	51	func newWithWriteTimeout(st store.Store, reg registry.Registry, h hub.Hub, secret []byte, maxCredAge, writeTimeout time.Duration) Service {
46	if writeTimeout <= 0 {	52	if writeTimeout <= 0 {
47	writeTimeout = defaultWriteTimeout	53	writeTimeout = defaultWriteTimeout
48	}	54	}
49	return &Service{st: st, reg: reg, hub: h, secret: secret, writeTimeout: writeTimeout}	55	return &Service{st: st, reg: reg, hub: h, secret: secret, maxCredAge: maxCredAge, writeTimeout: writeTimeout}
50	}	56	}
51		57
52	// Serve accepts QUIC connections until ctx is cancelled.	58	// Serve accepts QUIC connections until ctx is cancelled.
@@ -80,15 +86,26 @@ func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
80	// Auth: the Hello carries a Bearer host credential (string credential = 9).	86	// Auth: the Hello carries a Bearer host credential (string credential = 9).
81	cred := h.GetCredential()	87	cred := h.GetCredential()
82	// host_id in Hello is advisory; the authenticated identity comes from the credential (hosttoken.Verify), so a spoofed Hello.HostId cannot mislead us.	88	// host_id in Hello is advisory; the authenticated identity comes from the credential (hosttoken.Verify), so a spoofed Hello.HostId cannot mislead us.
83	hostID, ok := hosttoken.Verify(s.secret, cred)	89	claims, ok := hosttoken.Verify(s.secret, cred)
84	if !ok {	90	if !ok {
85	_ = conn.CloseWithError(transport.CodeAuthRejected, "invalid host credential")	91	_ = conn.CloseWithError(transport.CodeAuthRejected, "invalid host credential")
86	return	92	return
87	}	93	}
88	if _, err := s.st.GetHost(hostID); err != nil {	94	hostID := claims.HostID
		95	if s.maxCredAge > 0 && time.Since(claims.IssuedAt) > s.maxCredAge {
		96	_ = conn.CloseWithError(transport.CodeAuthRejected, "credential expired — re-enroll this host")
		97	return
		98	}
		99	hostRow, err := s.st.GetHost(hostID)
		100	if err != nil {
89	_ = conn.CloseWithError(transport.CodeAuthRejected, "host not found")	101	_ = conn.CloseWithError(transport.CodeAuthRejected, "host not found")
90	return	102	return
91	}	103	}
		104	// Per-host revocation: a credential minted at an older generation is dead.
		105	if claims.Generation != hostRow.CredGeneration {
		106	_ = conn.CloseWithError(transport.CodeAuthRejected, "credential revoked — re-enroll this host")
		107	return
		108	}
92	slog.Info("agent connected", "host", hostID, "provisioner", h.GetProvisioner(), "last_seen_epoch", h.GetLastSeenEpoch())	109	slog.Info("agent connected", "host", hostID, "provisioner", h.GetProvisioner(), "last_seen_epoch", h.GetLastSeenEpoch())
93		110
94	// Down-stream: server opens it; first write makes it visible to the agent.	111	// Down-stream: server opens it; first write makes it visible to the agent.
@@ -128,6 +145,24 @@ func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
128	return	145	return
129	}	146	}
130	if rep := msg.GetReport(); rep != nil {	147	if rep := msg.GetReport(); rep != nil {
		148	// Re-check the credential each report so a revoke (or max-age
		149	// expiry) lands within one tick (~10s) instead of waiting for the
		150	// session to drop naturally.
		151	row, err := s.st.GetHost(hostID)
		152	if err != nil {
		153	// Transient store failure — NOT an auth verdict. Close with a
		154	// plain code so the agent retries on its normal 5s backoff
		155	// instead of the permanent-auth 60s path.
		156	slog.Warn("credential re-check failed; dropping session", "host", hostID, "err", err)
		157	_ = conn.CloseWithError(0, "credential re-check unavailable")
		158	return
		159	}
		160	expired := s.maxCredAge > 0 && time.Since(claims.IssuedAt) > s.maxCredAge
		161	if row.CredGeneration != claims.Generation \|\| expired {
		162	slog.Info("credential no longer valid mid-session; closing", "host", hostID, "expired", expired)
		163	_ = conn.CloseWithError(transport.CodeAuthRejected, "credential revoked or expired — re-enroll this host")
		164	return
		165	}
131	s.applyReport(hostID, rep)	166	s.applyReport(hostID, rep)
132	}	167	}
133	select {	168	select {

internal/server/syncsvc/syncsvc_test.go

+	require.NoError(t, err)
+	t.Cleanup(func() { st.Close() })
+	tok, _ := st.CreateEnrollmentToken()
+	host, err := st.RedeemEnrollmentToken(tok, "h", "linux", "amd64", "cloudhv", "", "")
+	require.NoError(t, err)
+	reg := registry.New(time.Now)
+	addr, fp, _ := startTestServer(t, st, reg, h, secret, writeTimeout)
+	return &fixture{st: st, reg: reg, hub: h, addr: addr, fp: fp, secret: secret,
+		host: host, cred: hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now())}
+// startTestServer listens on 127.0.0.1:0 (random UDP port) and returns the addr,
+	require.NoError(t, err)
+	lis, err := quic.ListenAddr("127.0.0.1:0", tlsConf, &quic.Config{MaxIdleTimeout: 5 * time.Second})
+	require.NoError(t, err)
+	svc := newWithWriteTimeout(st, reg, h, secret, 0, writeTimeout)
+	ctx, cancel := context.WithCancel(context.Background())
+	go svc.Serve(ctx, lis) //nolint:errcheck
+	stop = func() { cancel(); lis.Close() }
@@ @@ -349,3 +349,121 @@ func TestWrongCertPinRejected(t *testing.T) { @@
+	assert.Contains(t, strings.ToLower(err.Error()), "fingerprint",
+		"expected cert fingerprint mismatch, got: %v", err)
+// TestStaleGenerationCredentialRejected pins per-host revocation: after
+// BumpCredGeneration, a credential minted at the old generation is rejected
+// with CodeAuthRejected, while a freshly-minted one connects.
+func TestStaleGenerationCredentialRejected(t *testing.T) {
+	f := setup(t)
+	oldCred := hosttoken.Mint(f.secret, f.host.ID, f.host.CredGeneration, time.Now())
+	_, err := f.st.BumpCredGeneration(f.host.ID, "")
+	require.NoError(t, err)
+	_, err = dial(t, f.addr, f.fp, f.host.ID, oldCred)
+	require.Error(t, err, "revoked-generation credential must be rejected")
+	var appErr *quic.ApplicationError
+	require.ErrorAs(t, err, &appErr)
+	assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
+	fresh := hosttoken.Mint(f.secret, f.host.ID, f.host.CredGeneration+1, time.Now())
+	c, err := dial(t, f.addr, f.fp, f.host.ID, fresh)
+	require.NoError(t, err, "current-generation credential must connect")
+	c.conn.CloseWithError(0, "")
+}
+// TestExpiredCredentialRejectedWhenMaxAgeSet pins the optional max-age
+// policy: with maxAge configured, a credential older than maxAge is rejected;
+// with maxAge zero (default) age is ignored.
+func TestExpiredCredentialRejectedWhenMaxAgeSet(t *testing.T) {
+	st, err := store.Open(t.TempDir()+"/db", "10.77.0.0/16")
+	require.NoError(t, err)
+	t.Cleanup(func() { st.Close() })
+	tok, _ := st.CreateEnrollmentToken()
+	host, err := st.RedeemEnrollmentToken(tok, "h", "linux", "amd64", "cloudhv", "", "")
+	require.NoError(t, err)
+	reg := registry.New(time.Now)
+	h := hub.New()
+	secret := []byte("s3cret")
+	certPEM, keyPEM, err := transport.GenerateServerCert()
+	require.NoError(t, err)
+	fp, err := transport.CertFingerprint(certPEM)
+	require.NoError(t, err)
+	tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
+	require.NoError(t, err)
+	lis, err := quic.ListenAddr("127.0.0.1:0", tlsConf, &quic.Config{MaxIdleTimeout: 5 * time.Second})
+	require.NoError(t, err)
+	svc := New(st, reg, h, secret, 30*24*time.Hour) // maxAge 30d
+	ctx, cancel := context.WithCancel(context.Background())
+	go svc.Serve(ctx, lis) //nolint:errcheck
+	t.Cleanup(func() { cancel(); lis.Close() })
+	addr := lis.Addr().String()
+	stale := hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now().Add(-31*24*time.Hour))
+	_, err = dial(t, addr, fp, host.ID, stale)
+	require.Error(t, err, "over-max-age credential must be rejected")
+	var appErr *quic.ApplicationError
+	require.ErrorAs(t, err, &appErr)
+	assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
+	fresh := hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now())
+	c, err := dial(t, addr, fp, host.ID, fresh)
+	require.NoError(t, err, "fresh credential must connect")
+	c.conn.CloseWithError(0, "")
+}
+// TestMaxAgeEnforcedMidSession pins that a live session does not outlive
+// credential_max_age: the per-report re-check closes it once the credential's
+// issued-at falls out of the window.
+func TestMaxAgeEnforcedMidSession(t *testing.T) {
+	st, err := store.Open(t.TempDir()+"/db", "10.77.0.0/16")
+	require.NoError(t, err)
+	t.Cleanup(func() { st.Close() })
+	tok, _ := st.CreateEnrollmentToken()
+	host, err := st.RedeemEnrollmentToken(tok, "h", "linux", "amd64", "cloudhv", "", "")
+	require.NoError(t, err)
+	reg := registry.New(time.Now)
+	h := hub.New()
+	secret := []byte("s3cret")
+	certPEM, keyPEM, err := transport.GenerateServerCert()
+	require.NoError(t, err)
+	fp, err := transport.CertFingerprint(certPEM)
+	require.NoError(t, err)
+	tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
+	require.NoError(t, err)
+	lis, err := quic.ListenAddr("127.0.0.1:0", tlsConf, &quic.Config{MaxIdleTimeout: 5 * time.Second})
+	require.NoError(t, err)
+	// Timing is truncation-aware: Mint stores issued_unix at SECOND
+	// granularity, so the observed age can read up to 0.999s older than
+	// real. maxAge 3s with a 1.5s-old credential leaves >1s margin on the
+	// Hello side (worst case 2.499s < 3s), and the 1.8s sleep pushes the
+	// real age to 3.3s — past maxAge even before truncation inflation.
+	svc := New(st, reg, h, secret, 3*time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
+	go svc.Serve(ctx, lis) //nolint:errcheck
+	t.Cleanup(func() { cancel(); lis.Close() })
+	cred := hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now().Add(-1500*time.Millisecond))
+	c, err := dial(t, lis.Addr().String(), fp, host.ID, cred)
+	require.NoError(t, err, "not-yet-expired credential connects")
+	defer c.conn.CloseWithError(0, "")
+	time.Sleep(1800 * time.Millisecond) // credential ages past maxCredAge
+	c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{Report: &pb.ActualStateReport{}}})
+	// The server must close the connection with CodeAuthRejected; the next
+	// read on the down-stream surfaces it.
+	var msg pb.ServerMessage
+	readErr := transport.ReadMsg(c.down, &msg, transport.DefaultMaxFrame)
+	if readErr == nil {
+		// First read may deliver the initial snapshot; the close lands next.
+		readErr = transport.ReadMsg(c.down, &msg, transport.DefaultMaxFrame)
+	}
+	require.Error(t, readErr)
+	var appErr *quic.ApplicationError
+	require.ErrorAs(t, readErr, &appErr)
+	assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
+}

internal/shape/classify.go

Old		New
@@ -30,7 +30,8 @@ func classify(rel string) Plane {
30	case strings.HasPrefix(rel, "internal/agent"):	30	case strings.HasPrefix(rel, "internal/agent"):
31	return PlaneData	31	return PlaneData
32	case strings.HasPrefix(rel, "internal/pb"),	32	case strings.HasPrefix(rel, "internal/pb"),
33	strings.HasPrefix(rel, "internal/transport"):	33	strings.HasPrefix(rel, "internal/transport"),
		34	strings.HasPrefix(rel, "internal/joinblob"):
34	return PlaneWire	35	return PlaneWire
35	case strings.HasPrefix(rel, "cmd/"):	36	case strings.HasPrefix(rel, "cmd/"):
36	return PlaneBinaries	37	return PlaneBinaries

internal/transport/tlsconf.go

Old		New
@@ -35,9 +35,13 @@ func GenerateServerCert() (certPEM, keyPEM []byte, err error) {
35	SerialNumber: big.NewInt(1),	35	SerialNumber: big.NewInt(1),
36	Subject: pkix.Name{CommonName: "eitri-server"},	36	Subject: pkix.Name{CommonName: "eitri-server"},
37	NotBefore: time.Now().Add(-time.Hour),	37	NotBefore: time.Now().Add(-time.Hour),
38	NotAfter: time.Now().AddDate(10, 0, 0),	38	// 2 years, not 10: the agent pin ignores expiry (an expired cert never
39	KeyUsage: x509.KeyUsageDigitalSignature,	39	// breaks the fleet), so long validity only widens the forgery window
40	ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},	40	// if server.key leaks. CertRenewalDue warns 90 days out; rotation
		41	// runbook: docs/cert-rotation.md.
		42	NotAfter: time.Now().AddDate(2, 0, 0),
		43	KeyUsage: x509.KeyUsageDigitalSignature,
		44	ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
41	}	45	}
42	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)	46	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
43	if err != nil {	47	if err != nil {
@@ -62,6 +66,27 @@ func CertFingerprint(certPEM []byte) (string, error) {
62	return hex.EncodeToString(sum[:]), nil	66	return hex.EncodeToString(sum[:]), nil
63	}	67	}
64		68
		69	// renewalWindow is how long before NotAfter CertRenewalDue starts reporting
		70	// true. 90 days gives the operator a comfortable rotation runway.
		71	const renewalWindow = 90 * 24 * time.Hour
		72
		73	// CertRenewalDue reports the cert's NotAfter and whether rotation is due (now
		74	// is within renewalWindow of expiry, or past it). An unparseable cert reports
		75	// due — fail loud so the operator looks at it. Note: the agent pin
		76	// (VerifyConnection) ignores expiry, so an expired cert never breaks the
		77	// running fleet; this only drives the operator-facing warning cadence.
		78	func CertRenewalDue(certPEM []byte, now time.Time) (notAfter time.Time, due bool) {
		79	block, _ := pem.Decode(certPEM)
		80	if block == nil {
		81	return time.Time{}, true
		82	}
		83	cert, err := x509.ParseCertificate(block.Bytes)
		84	if err != nil {
		85	return time.Time{}, true
		86	}
		87	return cert.NotAfter, now.After(cert.NotAfter.Add(-renewalWindow))
		88	}
		89
65	// ServerTLS builds the server's tls.Config from PEM cert+key with ALPN set.	90	// ServerTLS builds the server's tls.Config from PEM cert+key with ALPN set.
66	func ServerTLS(certPEM, keyPEM []byte) (*tls.Config, error) {	91	func ServerTLS(certPEM, keyPEM []byte) (*tls.Config, error) {
67	cert, err := tls.X509KeyPair(certPEM, keyPEM)	92	cert, err := tls.X509KeyPair(certPEM, keyPEM)

internal/transport/tlsconf_test.go

@@ @@ -3,7 +3,9 @@ package transport @@
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"testing"
+	"time"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
@@ @@ -46,3 +48,50 @@ func TestServerTLSHasALPN(t *testing.T) { @@
+	require.NoError(t, err)
+	assert.Contains(t, sc.NextProtos, ALPN)
+// TestGeneratedCertValidityIsTwoYears pins the cert lifetime at ~2 years. The
+// pin (VerifyConnection) ignores expiry, so a longer validity buys nothing —
+// it only widens the forgery window if server.key leaks. Two years sets the
+// rotation cadence the renewal warning drives.
+func TestGeneratedCertValidityIsTwoYears(t *testing.T) {
+	certPEM, _, err := GenerateServerCert()
+	require.NoError(t, err)
+	block, _ := pem.Decode(certPEM)
+	require.NotNil(t, block)
+	cert, err := x509.ParseCertificate(block.Bytes)
+	require.NoError(t, err)
+	lifetime := cert.NotAfter.Sub(cert.NotBefore)
+	assert.LessOrEqual(t, lifetime, 2*365*24*time.Hour+31*24*time.Hour,
+		"validity must be ~2 years, not the old 10")
+	assert.Greater(t, lifetime, 365*24*time.Hour, "validity must exceed 1 year")
+}
+// TestCertRenewalDue pins the renewal-warning helper: due when now is within
+// 90 days of NotAfter (or past it), not due before that, and an unparseable
+// cert reports due (fail-loud: an operator should look at a broken cert).
+func TestCertRenewalDue(t *testing.T) {
+	certPEM, _, err := GenerateServerCert()
+	require.NoError(t, err)
+	block, _ := pem.Decode(certPEM)
+	require.NotNil(t, block)
+	cert, err := x509.ParseCertificate(block.Bytes)
+	require.NoError(t, err)
+	notAfter, due := CertRenewalDue(certPEM, cert.NotAfter.Add(-180*24*time.Hour))
+	assert.False(t, due, "6 months out must not be due")
+	assert.Equal(t, cert.NotAfter, notAfter)
+	_, due = CertRenewalDue(certPEM, cert.NotAfter.Add(-30*24*time.Hour))
+	assert.True(t, due, "30 days out must be due")
+	_, due = CertRenewalDue(certPEM, cert.NotAfter.Add(24*time.Hour))
+	assert.True(t, due, "past expiry must be due")
+	_, due = CertRenewalDue([]byte("not a cert"), time.Now())
+	assert.True(t, due, "non-PEM input must report due (fail loud)")
+	junkDER := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: []byte("junk")})
+	_, due = CertRenewalDue(junkDER, time.Now())
+	assert.True(t, due, "valid PEM with garbage DER must report due (fail loud)")
+}

web/src/lib/api-types.ts

Old		New
@@ -4,6 +4,54 @@
4	*/	4	*/
5		5
6	export interface paths {	6	export interface paths {
		7	"/api/v1/audit": {
		8	parameters: {
		9	query?: never;
		10	header?: never;
		11	path?: never;
		12	cookie?: never;
		13	};
		14	/** Newest audit log rows. */
		15	get: {
		16	parameters: {
		17	query?: {
		18	/** @description max rows to return (default 100, cap 1000) */
		19	limit?: string;
		20	};
		21	header?: never;
		22	path?: never;
		23	cookie?: never;
		24	};
		25	requestBody?: never;
		26	responses: {
		27	/** @description success */
		28	200: {
		29	headers: {
		30	[name: string]: unknown;
		31	};
		32	content: {
		33	"application/json": components["schemas"]["AuditEvent"][];
		34	};
		35	};
		36	/** @description error (plain text) */
		37	default: {
		38	headers: {
		39	[name: string]: unknown;
		40	};
		41	content: {
		42	"text/plain": string;
		43	};
		44	};
		45	};
		46	};
		47	put?: never;
		48	post?: never;
		49	delete?: never;
		50	options?: never;
		51	head?: never;
		52	patch?: never;
		53	trace?: never;
		54	};
7	"/api/v1/enroll": {	55	"/api/v1/enroll": {
8	parameters: {	56	parameters: {
9	query?: never;	57	query?: never;
@@ -13,7 +61,7 @@ export interface paths {
13	};	61	};
14	get?: never;	62	get?: never;
15	put?: never;	63	put?: never;
16	/** Redeem a one-time enrollment token: a new host joins the fleet and receives its credential. Unauthenticated; the token is the proof. */	64	/** Redeem a one-time enrollment token: a new host joins the fleet and receives its credential. Unauthenticated but rate-limited; the token is the proof. */
17	post: {	65	post: {
18	parameters: {	66	parameters: {
19	query?: never;	67	query?: never;
@@ -62,7 +110,7 @@ export interface paths {
62	};	110	};
63	get?: never;	111	get?: never;
64	put?: never;	112	put?: never;
65	/** Mint a one-time host enrollment token. */	113	/** Mint a one-time host enrollment token plus the join blob agents consume. */
66	post: {	114	post: {
67	parameters: {	115	parameters: {
68	query?: never;	116	query?: never;
@@ -109,8 +157,8 @@ export interface paths {
109	get: {	157	get: {
110	parameters: {	158	parameters: {
111	query?: {	159	query?: {
112	/** @description admin token */	160	/** @description one-time stream ticket */
113	token?: string;	161	ticket?: string;
114	};	162	};
115	header?: never;	163	header?: never;
116	path?: never;	164	path?: never;
@@ -236,6 +284,96 @@ export interface paths {
236	patch?: never;	284	patch?: never;
237	trace?: never;	285	trace?: never;
238	};	286	};
		287	"/api/v1/hosts/{id}/revoke-credential": {
		288	parameters: {
		289	query?: never;
		290	header?: never;
		291	path?: never;
		292	cookie?: never;
		293	};
		294	get?: never;
		295	put?: never;
		296	/** Revoke a host's outstanding credential by bumping its generation; the host stays dark until re-enrolled. */
		297	post: {
		298	parameters: {
		299	query?: never;
		300	header?: never;
		301	path: {
		302	id: string;
		303	};
		304	cookie?: never;
		305	};
		306	requestBody?: never;
		307	responses: {
		308	/** @description success */
		309	204: {
		310	headers: {
		311	[name: string]: unknown;
		312	};
		313	content?: never;
		314	};
		315	/** @description error (plain text) */
		316	default: {
		317	headers: {
		318	[name: string]: unknown;
		319	};
		320	content: {
		321	"text/plain": string;
		322	};
		323	};
		324	};
		325	};
		326	delete?: never;
		327	options?: never;
		328	head?: never;
		329	patch?: never;
		330	trace?: never;
		331	};
		332	"/api/v1/stream-tickets": {
		333	parameters: {
		334	query?: never;
		335	header?: never;
		336	path?: never;
		337	cookie?: never;
		338	};
		339	get?: never;
		340	put?: never;
		341	/** Mint a one-time short-TTL ticket for the SSE stream — the only credential that ever rides in a URL. */
		342	post: {
		343	parameters: {
		344	query?: never;
		345	header?: never;
		346	path?: never;
		347	cookie?: never;
		348	};
		349	requestBody?: never;
		350	responses: {
		351	/** @description success */
		352	201: {
		353	headers: {
		354	[name: string]: unknown;
		355	};
		356	content: {
		357	"application/json": components["schemas"]["StreamTicketResponse"];
		358	};
		359	};
		360	/** @description error (plain text) */
		361	default: {
		362	headers: {
		363	[name: string]: unknown;
		364	};
		365	content: {
		366	"text/plain": string;
		367	};
		368	};
		369	};
		370	};
		371	delete?: never;
		372	options?: never;
		373	head?: never;
		374	patch?: never;
		375	trace?: never;
		376	};
239	"/api/v1/vms": {	377	"/api/v1/vms": {
240	parameters: {	378	parameters: {
241	query?: never;	379	query?: never;
@@ -396,6 +534,12 @@ export interface paths {
396	export type webhooks = Record<string, never>;	534	export type webhooks = Record<string, never>;
397	export interface components {	535	export interface components {
398	schemas: {	536	schemas: {
		537	AuditEvent: {
		538	action: string;
		539	/** Format: date-time */
		540	at: string;
		541	detail: unknown;
		542	};
399	Capacity: {	543	Capacity: {
400	disk_gb: number;	544	disk_gb: number;
401	mem_mb: number;	545	mem_mb: number;
@@ -434,6 +578,7 @@ export interface components {
434	server_cert_sha256: string;	578	server_cert_sha256: string;
435	};	579	};
436	EnrollTokenResponse: {	580	EnrollTokenResponse: {
		581	join: string;
437	token: string;	582	token: string;
438	};	583	};
439	Host: {	584	Host: {
@@ -458,6 +603,9 @@ export interface components {
458	hosts: components["schemas"]["Host"][];	603	hosts: components["schemas"]["Host"][];
459	vms: components["schemas"]["VM"][];	604	vms: components["schemas"]["VM"][];
460	};	605	};
		606	StreamTicketResponse: {
		607	ticket: string;
		608	};
461	VM: {	609	VM: {
462	actual_power: string;	610	actual_power: string;
463	assigned_ip: string;	611	assigned_ip: string;

web/src/lib/fleet.svelte.ts

Old		New
@@ -25,6 +25,11 @@ export const fleet = $state({
25		25
26	let es: EventSource \| null = null;	26	let es: EventSource \| null = null;
27		27
		28	// sseParseError marks that fleet.error came from the SSE stream itself (not a
		29	// user action), so the next well-formed event self-heals it — action errors
		30	// stay sticky until dismissed.
		31	let sseParseError = false;
		32
28	function authHeaders(): HeadersInit {	33	function authHeaders(): HeadersInit {
29	return { Authorization: `Bearer ${fleet.token}`, 'Content-Type': 'application/json' };	34	return { Authorization: `Bearer ${fleet.token}`, 'Content-Type': 'application/json' };
30	}	35	}
@@ -49,28 +54,85 @@ export function setToken(t: string) {
49	connect();	54	connect();
50	}	55	}
51		56
52	/** connect opens the SSE stream; on error it falls back to a one-shot refresh. */	57	let reconnectTimer: ReturnType<typeof setTimeout> \| null = null;
53	export function connect() {	58
		59	// connectGen guards against overlapping async connect() calls: only the
		60	// newest invocation may install its EventSource — a stale one closes its
		61	// orphan instead of clobbering (or later killing) the current stream.
		62	let connectGen = 0;
		63
		64	/** connect mints a one-time stream ticket (so the admin token never rides in
		65	* a URL) and opens the SSE stream. Tickets are single-use, so the browser's
		66	* built-in EventSource retry cannot work — on error we close the stream and
		67	* reconnect ourselves with a fresh ticket. */
		68	export async function connect() {
		69	es?.close(); // close first: clearing the token must also stop the stream
		70	if (reconnectTimer) {
		71	clearTimeout(reconnectTimer);
		72	reconnectTimer = null;
		73	}
54	if (!fleet.token) return;	74	if (!fleet.token) return;
55	es?.close();	75	const gen = ++connectGen;
56	es = new EventSource(`/api/v1/events?token=${encodeURIComponent(fleet.token)}`);	76	let ticket: string;
57	es.addEventListener('state', (e) => {	77	try {
		78	const r = await (await req('POST', '/api/v1/stream-tickets')).json();
		79	if (typeof r.ticket !== 'string' \|\| !r.ticket) throw new Error('malformed ticket response');
		80	ticket = r.ticket;
		81	} catch {
		82	if (gen !== connectGen) return; // superseded while minting
		83	fleet.connected = false;
		84	scheduleReconnect();
		85	return;
		86	}
		87	if (gen !== connectGen) return; // superseded while minting — drop the ticket
		88	const mine = new EventSource(`/api/v1/events?ticket=${encodeURIComponent(ticket)}`);
		89	es = mine;
		90	mine.addEventListener('state', (e) => {
58	try {	91	try {
59	const snap = JSON.parse((e as MessageEvent).data);	92	const snap = JSON.parse((e as MessageEvent).data);
60	fleet.hosts = snap.hosts ?? [];	93	fleet.hosts = snap.hosts ?? [];
61	fleet.vms = snap.vms ?? [];	94	fleet.vms = snap.vms ?? [];
62	fleet.connected = true;	95	fleet.connected = true;
63	fleet.error = '';	96	// Deliberately does NOT clear action errors: state events arrive
		97	// ~1/s, so auto-clearing here made action failures flash for under
		98	// a second — invisible in practice. They persist until dismissed
		99	// (dismissError) or the next successful refresh() (page load /
		100	// token re-set). Only the stream's OWN parse errors self-heal.
		101	if (sseParseError) {
		102	fleet.error = '';
		103	sseParseError = false;
		104	}
64	} catch (err) {	105	} catch (err) {
65	fleet.error = String(err);	106	fleet.error = String(err);
		107	sseParseError = true;
66	}	108	}
67	});	109	});
68	es.onerror = () => {	110	mine.onerror = () => {
		111	// Only the CURRENT stream may trigger reconnect churn — an orphaned
		112	// EventSource from a superseded connect() closes itself and stops.
		113	mine.close();
		114	if (es !== mine) return;
69	fleet.connected = false;	115	fleet.connected = false;
70	// EventSource auto-reconnects; surface the state but keep last data.	116	// The ticket is consumed: the built-in retry would replay a dead URL.
		117	// Reconnect with a fresh ticket instead; keep last data.
		118	scheduleReconnect();
71	};	119	};
72	}	120	}
73		121
		122	/** scheduleReconnect arms a single delayed reconnect (fresh ticket). */
		123	function scheduleReconnect() {
		124	if (reconnectTimer) return;
		125	reconnectTimer = setTimeout(() => {
		126	reconnectTimer = null;
		127	void connect();
		128	}, 3000);
		129	}
		130
		131	/** dismissError clears the sticky error banner. */
		132	export function dismissError() {
		133	fleet.error = '';
		134	}
		135
74	/** refresh does a one-shot fetch (used before SSE connects or as a fallback). */	136	/** refresh does a one-shot fetch (used before SSE connects or as a fallback). */
75	export async function refresh() {	137	export async function refresh() {
76	try {	138	try {
@@ -102,9 +164,9 @@ export async function decommissionHost(id: string) {
102	await req('DELETE', `/api/v1/hosts/${id}`);	164	await req('DELETE', `/api/v1/hosts/${id}`);
103	}	165	}
104		166
105	export async function createEnrollToken(): Promise<string> {	167	export async function createJoinBlob(): Promise<string> {
106	const r = await (await req('POST', '/api/v1/enroll-tokens')).json();	168	const r = await (await req('POST', '/api/v1/enroll-tokens')).json();
107	return r.token;	169	return r.join;
108	}	170	}
109		171
110	export function vmsForHost(id: string): VM[] {	172	export function vmsForHost(id: string): VM[] {

web/src/routes/+layout.svelte

Old		New
@@ -1,7 +1,7 @@
1	<script lang="ts">	1	<script lang="ts">
2	import favicon from '$lib/assets/favicon.svg';	2	import favicon from '$lib/assets/favicon.svg';
3	import { onMount } from 'svelte';	3	import { onMount } from 'svelte';
4	import { fleet, setToken, connect, refresh } from '$lib/fleet.svelte';	4	import { fleet, setToken, connect, refresh, dismissError } from '$lib/fleet.svelte';
5	let { children } = $props();	5	let { children } = $props();
6	let tokenInput = $state('');	6	let tokenInput = $state('');
7		7
@@ -18,6 +18,20 @@
18	setToken(tokenInput);	18	setToken(tokenInput);
19	refresh();	19	refresh();
20	}	20	}
		21
		22	// Dismissing the banner removes the focused button from the DOM, which
		23	// drops keyboard focus to <body>. Land it somewhere useful instead: the
		24	// fleet filter when present (overview), else the first main link/control.
		25	function dismissAndRefocus() {
		26	dismissError();
		27	// An open modal owns focus: dismissing the banner above it must not
		28	// send keyboard focus behind the overlay.
		29	const target =
		30	document.querySelector<HTMLElement>('.modal input, .modal select, .modal button') ??
		31	document.getElementById('fleet-filter') ??
		32	document.querySelector<HTMLElement>('main a, main button, main input');
		33	target?.focus();
		34	}
21	</script>	35	</script>
22		36
23	<svelte:head>	37	<svelte:head>
@@ -40,7 +54,10 @@
40	</header>	54	</header>
41		55
42	{#if fleet.error}	56	{#if fleet.error}
43	<div class="error">{fleet.error}</div>	57	<div class="error" role="alert">
		58	<span>{fleet.error}</span>
		59	<button type="button" class="dismiss" onclick={dismissAndRefocus} aria-label="dismiss error">✕</button>
		60	</div>
44	{/if}	61	{/if}
45		62
46	<main>	63	<main>
@@ -157,6 +174,19 @@
157	color: #ffb4b4;	174	color: #ffb4b4;
158	padding: 0.5rem 1rem;	175	padding: 0.5rem 1rem;
159	border-bottom: 1px solid #5a2025;	176	border-bottom: 1px solid #5a2025;
		177	display: flex;
		178	align-items: center;
		179	justify-content: space-between;
		180	gap: 0.6rem;
		181	}
		182	.error .dismiss {
		183	background: transparent;
		184	border: 1px solid #5a2025;
		185	color: #ffb4b4;
		186	padding: 0.1rem 0.45rem;
		187	}
		188	.error .dismiss:hover {
		189	background: #5a2025;
160	}	190	}
161	.hint {	191	.hint {
162	color: #6b7280;	192	color: #6b7280;

web/src/routes/+page.svelte

Old		New
@@ -5,7 +5,7 @@
5	setPower,	5	setPower,
6	deleteVM,	6	deleteVM,
7	decommissionHost,	7	decommissionHost,
8	createEnrollToken,	8	createJoinBlob,
9	vmsForHost,	9	vmsForHost,
10	vmPhase,	10	vmPhase,
11	vmPower,	11	vmPower,
@@ -16,8 +16,31 @@
16		16
17	let showCreate = $state(false);	17	let showCreate = $state(false);
18	let advanced = $state(false);	18	let advanced = $state(false);
19	let enrollToken = $state('');	19	let joinBlob = $state('');
20	let busy = $state('');	20	let busy = $state('');
		21	let filter = $state('');
		22
		23	// Case-insensitive substring match over the row's visible fields, so a
		24	// large fleet can be narrowed live from one box.
		25	const needle = $derived(filter.trim().toLowerCase());
		26	const shownHosts = $derived(
		27	fleet.hosts.filter(
		28	(h) =>
		29	!needle \|\|
		30	`${h.name} ${h.bridge_cidr} ${h.status} ${h.online ? 'online' : 'offline'}`
		31	.toLowerCase()
		32	.includes(needle)
		33	)
		34	);
		35	const shownVMs = $derived(
		36	fleet.vms.filter((v) => {
		37	if (!needle) return true;
		38	const hostName = fleet.hosts.find((h) => h.id === v.host_id)?.name ?? '';
		39	return `${v.name} ${hostName} ${vmPhase(v)} ${vmPower(v)} ${vmIP(v)}`
		40	.toLowerCase()
		41	.includes(needle);
		42	})
		43	);
21		44
22	let form = $state<CreateVMRequest>({ host_id: '' });	45	let form = $state<CreateVMRequest>({ host_id: '' });
23		46
@@ -75,35 +98,48 @@
75		98
76	async function addHost() {	99	async function addHost() {
77	try {	100	try {
78	enrollToken = await createEnrollToken();	101	joinBlob = await createJoinBlob();
79	} catch (err) {	102	} catch (err) {
80	fleet.error = String(err);	103	fleet.error = String(err);
81	}	104	}
82	}	105	}
83	</script>	106	</script>
84		107
		108	<div class="row">
		109	<input
		110	id="fleet-filter"
		111	class="filter"
		112	type="search"
		113	placeholder="filter hosts & VMs…"
		114	bind:value={filter}
		115	aria-label="filter hosts and VMs"
		116	/>
		117	</div>
		118
85	<section>	119	<section>
86	<div class="row">	120	<div class="row">
87	<h2>Hosts ({fleet.hosts.length})</h2>	121	<h2>Hosts ({needle ? `${shownHosts.length}/${fleet.hosts.length}` : fleet.hosts.length})</h2>
88	<button class="ghost" onclick={addHost}>+ Add host</button>	122	<button class="ghost" onclick={addHost}>+ Add host</button>
89	</div>	123	</div>
90		124
91	{#if enrollToken}	125	{#if joinBlob}
92	<div class="enroll">	126	<div class="enroll">
93	Run on the new host:	127	Run on the new host:
94	<code>eitri-agent --server <url> --quic-addr <addr> --token {enrollToken} enroll</code>	128	<code>eitri-agent --state-dir /var/lib/eitri-agent join {joinBlob}</code>
95	</div>	129	</div>
96	{/if}	130	{/if}
97		131
98	{#if fleet.hosts.length === 0}	132	{#if fleet.hosts.length === 0}
99	<p class="hint">No hosts enrolled yet.</p>	133	<p class="hint">No hosts enrolled yet.</p>
		134	{:else if shownHosts.length === 0}
		135	<p class="hint">No hosts match “{filter}”.</p>
100	{:else}	136	{:else}
101	<table>	137	<table>
102	<thead>	138	<thead>
103	<tr><th>Name</th><th>Status</th><th>VMs</th><th>CIDR</th><th>Used / total (vCPU · mem · disk)</th><th></th></tr>	139	<tr><th>Name</th><th>Status</th><th>VMs</th><th>CIDR</th><th>Used / total (vCPU · mem · disk)</th><th></th></tr>
104	</thead>	140	</thead>
105	<tbody>	141	<tbody>
106	{#each fleet.hosts as h (h.id)}	142	{#each shownHosts as h (h.id)}
107	<tr>	143	<tr>
108	<td><a href="/hosts/{h.id}">{h.name}</a></td>	144	<td><a href="/hosts/{h.id}">{h.name}</a></td>
109	<td>	145	<td>
@@ -133,19 +169,21 @@
133		169
134	<section>	170	<section>
135	<div class="row">	171	<div class="row">
136	<h2>VMs ({fleet.vms.length})</h2>	172	<h2>VMs ({needle ? `${shownVMs.length}/${fleet.vms.length}` : fleet.vms.length})</h2>
137	<button onclick={openCreate} disabled={fleet.hosts.length === 0}>+ Create VM</button>	173	<button onclick={openCreate} disabled={fleet.hosts.length === 0}>+ Create VM</button>
138	</div>	174	</div>
139		175
140	{#if fleet.vms.length === 0}	176	{#if fleet.vms.length === 0}
141	<p class="hint">No VMs.</p>	177	<p class="hint">No VMs.</p>
		178	{:else if shownVMs.length === 0}
		179	<p class="hint">No VMs match “{filter}”.</p>
142	{:else}	180	{:else}
143	<table>	181	<table>
144	<thead>	182	<thead>
145	<tr><th>Name</th><th>Host</th><th>Phase</th><th>Power</th><th>IP</th><th></th></tr>	183	<tr><th>Name</th><th>Host</th><th>Phase</th><th>Power</th><th>IP</th><th></th></tr>
146	</thead>	184	</thead>
147	<tbody>	185	<tbody>
148	{#each fleet.vms as v (v.id)}	186	{#each shownVMs as v (v.id)}
149	<tr>	187	<tr>
150	<td><a href="/vms/{v.id}">{v.name}</a></td>	188	<td><a href="/vms/{v.id}">{v.name}</a></td>
151	<td>{fleet.hosts.find((h) => h.id === v.host_id)?.name ?? v.host_id.slice(0, 8)}</td>	189	<td>{fleet.hosts.find((h) => h.id === v.host_id)?.name ?? v.host_id.slice(0, 8)}</td>
@@ -299,4 +337,8 @@
299	.hint {	337	.hint {
300	color: #6b7280;	338	color: #6b7280;
301	}	339	}
		340	.filter {
		341	margin-top: 0.8rem;
		342	width: 260px;
		343	}
302	</style>	344	</style>

Old		New
@@ -131,16 +131,18 @@ shape-check:
131	# entrypoint — every main() in cmd/. Rooting at the binaries (NOT -test) is what	131	# entrypoint — every main() in cmd/. Rooting at the binaries (NOT -test) is what
132	# catches production code kept alive only by its own tests; the fix is to remove	132	# catches production code kept alive only by its own tests; the fix is to remove
133	# it, wire it into a real path, or move it into a _test.go. The smoke/sandbox tags	133	# it, wire it into a real path, or move it into a _test.go. The smoke/sandbox tags
134	# compile the tag-gated code so it is analysed too. Three sanctioned exceptions,	134	# compile the tag-gated code so it is analysed too. Four sanctioned exceptions,
135	# all production code that only a CROSS-package test can reach (so none can be	135	# all production code that only a CROSS-package test can reach (so none can be
136	# a _test.go): internal/integration (the e2e/harness tree), reconcile.Engine.Stop	136	# a _test.go): internal/integration (the e2e/harness tree), reconcile.Engine.Stop
137	# (terminal teardown that must not run in production — it would report every VM as	137	# (terminal teardown that must not run in production — it would report every VM as
138	# vanished — used only by an integration test's cleanup), and store.Store.Close /	138	# vanished — used only by an integration test's cleanup), store.Store.Close /
139	# store.Store.Epoch (reachable only via the store's tests until the serial console	139	# store.Store.Epoch (reachable only via the store's tests until the serial console
140	# stream lands and flows the store through a Close()-bearing interface).	140	# stream lands and flows the store through a Close()-bearing interface), and
		141	# store.Store.AllocatedByHost (the single-tx Snapshot path now serves GET /hosts;
		142	# the standalone accessor is exercised only by the store's own allocation tests).
141	deadcode:	143	deadcode:
142	@out=$$(go run golang.org/x/tools/cmd/deadcode@$(DEADCODE_VERSION) -tags=smoke,sandbox ./... \	144	@out=$$(go run golang.org/x/tools/cmd/deadcode@$(DEADCODE_VERSION) -tags=smoke,sandbox ./... \
143	\| { grep -vE '^internal/integration/\|unreachable func: Engine\.Stop$$\|unreachable func: Store\.Close$$\|unreachable func: Store\.Epoch$$' \|\| true; }); \	145	\| { grep -vE '^internal/integration/\|unreachable func: Engine\.Stop$$\|unreachable func: Store\.Close$$\|unreachable func: Store\.Epoch$$\|unreachable func: Store\.AllocatedByHost$$' \|\| true; }); \
144	if [ -n "$$out" ]; then \	146	if [ -n "$$out" ]; then \
145	echo "deadcode: unreachable from any cmd/ entrypoint (remove it, wire it in, or move it to a _test.go):"; \	147	echo "deadcode: unreachable from any cmd/ entrypoint (remove it, wire it in, or move it to a _test.go):"; \
146	echo "$$out"; exit 1; \	148	echo "$$out"; exit 1; \