From c1c2161350aebc07bc1c403b2538a1d1b1cdfbae Mon Sep 17 00:00:00 2001 From: lr00rl Date: Mon, 3 Aug 2026 23:36:27 -0700 Subject: [PATCH] [hephaestus] Keep optional NetGuard reality from starving core polls Wire the existing fail-whole collector to the accepted agent endpoint behind an explicit local opt-in. Run the report after core poll work and share a ten-second deadline across collection and POST so degraded discovery or transport remains visible without delaying task, monitor, and log-source service in the current cycle. Constraint: Existing bearer authentication and server payload contracts must remain unchanged Constraint: Partial collector output must never be reported as node ground truth Rejected: Fire-and-forget reporting | overlapping host discovery would lose per-cycle ordering and error ownership Rejected: Reuse the shared 30-second request ceiling synchronously | optional reporting could starve core poll work Confidence: high Scope-risk: moderate Directive: Keep guard reality low-trust and fail-whole; do not move it before core poll operations without re-evaluating latency Tested: Focused normal and race tests; next-cycle recovery and deadline tests; gopls; release-workflow check; install-integrity check; go vet ./...; go test -race -cover ./...; allowed-path and redaction scans Not-tested: Live ss/ip/nft discovery and a cross-process agent-to-server run were intentionally not executed --- README.md | 10 ++ cmd/lattice-agent/guard_reality_test.go | 192 ++++++++++++++++++++++++ cmd/lattice-agent/main.go | 43 +++++- 3 files changed, 241 insertions(+), 4 deletions(-) create mode 100644 cmd/lattice-agent/guard_reality_test.go diff --git a/README.md b/README.md index 30e4cc2..67fe64d 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,16 @@ memory/swap, platform, kernel, hostname, boot time, virtualization hint). They are collected with stdlib and local platform files such as `/proc` and `/etc/os-release`; missing fields are left empty and never block the agent. +NetGuard reality reporting is opt-in. Enable `-report-guard-reality` or set +`LATTICE_REPORT_GUARD_REALITY=1` to collect one complete read-only snapshot per +agent interval and post it to `/api/agent/guard-reality`. Collection invokes +local `ss`, `ip`, and `nft` commands with bounded output and a shared 10-second +collection-and-report deadline. If any step fails, the agent logs the failure +and sends no partial snapshot. Core task, monitor, and log-source polling runs +before this optional report, so a degraded collector cannot delay that work in +the current cycle. Reported facts are low-trust input for display, drift, and +suggestions only. They never mutate nftables or author policy. + ## Run ```sh diff --git a/cmd/lattice-agent/guard_reality_test.go b/cmd/lattice-agent/guard_reality_test.go new file mode 100644 index 0000000..53b3836 --- /dev/null +++ b/cmd/lattice-agent/guard_reality_test.go @@ -0,0 +1,192 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "reflect" + "testing" + "time" + + "github.com/LatticeNet/lattice-node-agent/internal/guardreality" + "github.com/LatticeNet/lattice-sdk/model" +) + +func TestReportGuardReality(t *testing.T) { + originalClient := httpClient + t.Cleanup(func() { httpClient = originalClient }) + + t.Run("disabled", func(t *testing.T) { + collectorCalls := 0 + requestCalls := 0 + httpClient = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + requestCalls++ + return testResponse(http.StatusOK, `{"ok":true}`), nil + })} + + err := reportGuardReality(context.Background(), agentConfig{}, func(context.Context, guardreality.Source, string) (model.GuardNodeReality, error) { + collectorCalls++ + return model.GuardNodeReality{}, errors.New("collector must not run") + }) + if err != nil { + t.Fatalf("disabled report error = %v", err) + } + if collectorCalls != 0 || requestCalls != 0 { + t.Fatalf("disabled report calls = collector %d, requests %d; want zero", collectorCalls, requestCalls) + } + }) + + t.Run("success", func(t *testing.T) { + collectedAt := time.Date(2026, time.August, 4, 6, 15, 0, 0, time.UTC) + wantReality := model.GuardNodeReality{ + NodeID: "node-a", + Listeners: []model.GuardListener{{ + Protocol: "tcp", + Port: 22, + Address: "::", + Process: "sshd", + }}, + Interfaces: []model.GuardInterface{{Name: "wg0", Addresses: []string{"2001:db8::2/128"}, Up: true}}, + ManagedSHA: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + ForeignTables: []string{"inet foreign"}, + NFTVersion: "1.1.0", + CollectedAt: collectedAt, + } + var body struct { + NodeID string `json:"node_id"` + Reality model.GuardNodeReality `json:"reality"` + } + requestCalls := 0 + httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + requestCalls++ + if r.Method != http.MethodPost || r.URL.Path != "/api/agent/guard-reality" { + return testResponse(http.StatusBadRequest, "bad request target"), nil + } + if r.Header.Get("Authorization") != "Bearer node-secret" { + return testResponse(http.StatusBadRequest, "missing bearer"), nil + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatal(err) + } + return testResponse(http.StatusOK, `{"ok":true}`), nil + })} + + collectorCalls := 0 + err := reportGuardReality(context.Background(), agentConfig{ + Server: "http://lattice.test", + NodeID: "node-a", + Token: "node-secret", + ReportGuardReality: true, + }, func(_ context.Context, _ guardreality.Source, nodeID string) (model.GuardNodeReality, error) { + collectorCalls++ + if nodeID != "node-a" { + t.Fatalf("collector node id = %q, want node-a", nodeID) + } + return wantReality, nil + }) + if err != nil { + t.Fatal(err) + } + if collectorCalls != 1 || requestCalls != 1 { + t.Fatalf("success calls = collector %d, requests %d; want one each", collectorCalls, requestCalls) + } + if body.NodeID != "node-a" || !reflect.DeepEqual(body.Reality, wantReality) { + t.Fatalf("unexpected report body: %+v", body) + } + }) + + t.Run("collect_failure", func(t *testing.T) { + requestCalls := 0 + httpClient = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + requestCalls++ + return testResponse(http.StatusOK, `{"ok":true}`), nil + })} + + err := reportGuardReality(context.Background(), agentConfig{ReportGuardReality: true}, func(context.Context, guardreality.Source, string) (model.GuardNodeReality, error) { + return model.GuardNodeReality{}, errors.New("nft unavailable") + }) + requireErrorContains(t, err, "collect guard reality") + requireErrorContains(t, err, "nft unavailable") + if requestCalls != 0 { + t.Fatalf("collection failure sent %d requests, want zero", requestCalls) + } + }) + + t.Run("server_failure", func(t *testing.T) { + requestCalls := 0 + httpClient = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + requestCalls++ + return testResponse(http.StatusServiceUnavailable, `{"error":{"code":"temporarily_unavailable","message":"retry later"}}`), nil + })} + + err := reportGuardReality(context.Background(), agentConfig{ + Server: "http://lattice.test", + NodeID: "node-a", + Token: "node-secret", + ReportGuardReality: true, + }, func(context.Context, guardreality.Source, string) (model.GuardNodeReality, error) { + return model.GuardNodeReality{NodeID: "node-a", CollectedAt: time.Now().UTC()}, nil + }) + requireErrorContains(t, err, "report guard reality") + requireErrorContains(t, err, "503 Service Unavailable") + if requestCalls != 1 { + t.Fatalf("server failure requests = %d, want one", requestCalls) + } + }) + + t.Run("next_cycle_after_failure", func(t *testing.T) { + requestCalls := 0 + httpClient = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + requestCalls++ + return testResponse(http.StatusOK, `{"ok":true}`), nil + })} + collectorCalls := 0 + collect := func(context.Context, guardreality.Source, string) (model.GuardNodeReality, error) { + collectorCalls++ + if collectorCalls == 1 { + return model.GuardNodeReality{}, errors.New("temporary collection failure") + } + return model.GuardNodeReality{NodeID: "node-a", CollectedAt: time.Now().UTC()}, nil + } + cfg := agentConfig{ + Server: "http://lattice.test", + NodeID: "node-a", + Token: "node-secret", + ReportGuardReality: true, + } + + firstErr := reportGuardReality(context.Background(), cfg, collect) + requireErrorContains(t, firstErr, "temporary collection failure") + if err := reportGuardReality(context.Background(), cfg, collect); err != nil { + t.Fatalf("next cycle did not recover: %v", err) + } + if collectorCalls != 2 || requestCalls != 1 { + t.Fatalf("two cycles made collector %d, requests %d calls; want 2 and 1", collectorCalls, requestCalls) + } + }) + + t.Run("deadline_bounds_post", func(t *testing.T) { + httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + <-r.Context().Done() + return nil, r.Context().Err() + })} + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + started := time.Now() + err := reportGuardReality(ctx, agentConfig{ + Server: "http://lattice.test", + NodeID: "node-a", + Token: "node-secret", + ReportGuardReality: true, + }, func(context.Context, guardreality.Source, string) (model.GuardNodeReality, error) { + return model.GuardNodeReality{NodeID: "node-a", CollectedAt: time.Now().UTC()}, nil + }) + requireErrorContains(t, err, "report guard reality") + requireErrorContains(t, err, "context deadline exceeded") + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("deadline took %s, want under one second", elapsed) + } + }) +} diff --git a/cmd/lattice-agent/main.go b/cmd/lattice-agent/main.go index 1fc06cc..2943b23 100644 --- a/cmd/lattice-agent/main.go +++ b/cmd/lattice-agent/main.go @@ -22,6 +22,7 @@ import ( "sync" "time" + "github.com/LatticeNet/lattice-node-agent/internal/guardreality" "github.com/LatticeNet/lattice-node-agent/internal/hostfacts" "github.com/LatticeNet/lattice-node-agent/internal/ipdiscover" "github.com/LatticeNet/lattice-node-agent/internal/metrics" @@ -61,6 +62,7 @@ const ( defaultDebugMaxLineBytes = 4096 defaultDebugMaxBatchLines = 100 debugSinkMaxLines = 1000 + guardRealityReportTimeout = 10 * time.Second ) type agentConfig struct { @@ -68,6 +70,7 @@ type agentConfig struct { NodeID string Token string Interval time.Duration + ReportGuardReality bool AllowExec bool AllowRoot bool NoExec bool @@ -179,6 +182,7 @@ func main() { flag.StringVar(&cfg.NodeID, "node-id", os.Getenv("LATTICE_NODE_ID"), "node id") flag.StringVar(&cfg.Token, "token", os.Getenv("LATTICE_NODE_TOKEN"), "node enrollment token") flag.DurationVar(&cfg.Interval, "interval", 10*time.Second, "metrics interval") + flag.BoolVar(&cfg.ReportGuardReality, "report-guard-reality", os.Getenv("LATTICE_REPORT_GUARD_REALITY") == "1", "report read-only NetGuard reality each interval") flag.BoolVar(&cfg.AllowExec, "allow-exec", os.Getenv("LATTICE_AGENT_ALLOW_EXEC") == "1", "allow bounded task execution") // -allow-root-exec opts in to running operator scripts while the agent is // uid 0. Without it, a root agent refuses tasks rather than executing @@ -303,7 +307,7 @@ func main() { log.Printf("warning: terminal sessions disabled because agent is running as root without -allow-root-exec") cfg.AllowTerminal = false } - debugf(cfg, "debug enabled: node=%s server=%s interval=%s allow_exec=%v allow_root_exec=%v allow_terminal=%v ssh_alerts=%v", cfg.NodeID, cfg.Server, cfg.Interval, cfg.AllowExec, cfg.AllowRoot, cfg.AllowTerminal, cfg.SSHAlerts) + debugf(cfg, "debug enabled: node=%s server=%s interval=%s report_guard_reality=%v allow_exec=%v allow_root_exec=%v allow_terminal=%v ssh_alerts=%v", cfg.NodeID, cfg.Server, cfg.Interval, cfg.ReportGuardReality, cfg.AllowExec, cfg.AllowRoot, cfg.AllowTerminal, cfg.SSHAlerts) // Probe interpreter availability once at startup so operators learn early // which allowlisted interpreters are missing, rather than only discovering it // when a task fails. Non-fatal; task-time resolution is unchanged. @@ -331,7 +335,7 @@ func main() { } else { applyAgentConfig(&cfg, agentCfg) } - log.Printf("lattice-agent connected node=%s server=%s allow_exec=%v allow_root_exec=%v task_cgroup=%v task_work_root=%v allow_terminal=%v terminal_transport=%s debug=%v", cfg.NodeID, cfg.Server, cfg.AllowExec, cfg.AllowRoot, cfg.taskCgroupConfig().Root != "", strings.TrimSpace(cfg.TaskWorkRoot) != "", cfg.AllowTerminal, cfg.TerminalTransport, cfg.Debug) + log.Printf("lattice-agent connected node=%s server=%s report_guard_reality=%v allow_exec=%v allow_root_exec=%v task_cgroup=%v task_work_root=%v allow_terminal=%v terminal_transport=%s debug=%v", cfg.NodeID, cfg.Server, cfg.ReportGuardReality, cfg.AllowExec, cfg.AllowRoot, cfg.taskCgroupConfig().Root != "", strings.TrimSpace(cfg.TaskWorkRoot) != "", cfg.AllowTerminal, cfg.TerminalTransport, cfg.Debug) if cfg.SSHAlerts { go watchSSHLogins(context.Background(), cfg) } @@ -379,6 +383,11 @@ func main() { if err := flushDebugEvents(cfg); err != nil { log.Printf("debug event report error: %v", err) } + reportCtx, cancelReport := context.WithTimeout(context.Background(), guardRealityReportTimeout) + if err := reportGuardReality(reportCtx, cfg, guardreality.Collect); err != nil { + log.Printf("guard reality error: %v", err) + } + cancelReport() <-ticker.C } } @@ -645,6 +654,24 @@ func reportMetrics(cfg agentConfig) error { }, nil) } +type guardRealityCollector func(context.Context, guardreality.Source, string) (model.GuardNodeReality, error) + +func reportGuardReality(ctx context.Context, cfg agentConfig, collect guardRealityCollector) error { + if !cfg.ReportGuardReality { + return nil + } + reality, err := collect(ctx, guardreality.Source{}, cfg.NodeID) + if err != nil { + return fmt.Errorf("collect guard reality: %w", err) + } + if err := postAgentJSONContext(ctx, cfg, "/api/agent/guard-reality", map[string]any{ + "reality": reality, + }, nil); err != nil { + return fmt.Errorf("report guard reality: %w", err) + } + return nil +} + // lastPublicProbe throttles outbound IP-echo requests so the agent does not hit // resolvers on every metrics tick. var lastPublicProbe time.Time @@ -1053,12 +1080,16 @@ func runTasks(cfg agentConfig, runner taskexec.Runner) error { } func postAgentJSON(cfg agentConfig, path string, payload map[string]any, out any) error { + return postAgentJSONContext(context.Background(), cfg, path, payload, out) +} + +func postAgentJSONContext(ctx context.Context, cfg agentConfig, path string, payload map[string]any, out any) error { if payload == nil { payload = map[string]any{} } payload["node_id"] = cfg.NodeID debugf(cfg, "agent post start: path=%s keys=%s", path, strings.Join(payloadKeys(payload), ",")) - if err := postJSON(cfg.Server+path, cfg.Token, payload, out); err != nil { + if err := postJSONContext(ctx, cfg.Server+path, cfg.Token, payload, out); err != nil { debugf(cfg, "agent post failed: path=%s err=%v", path, err) return err } @@ -1207,11 +1238,15 @@ func postAgentDebugBatch(cfg agentConfig, batch model.AgentDebugBatch) error { } func postJSON(url string, bearerToken string, payload any, out any) error { + return postJSONContext(context.Background(), url, bearerToken, payload, out) +} + +func postJSONContext(ctx context.Context, url string, bearerToken string, payload any, out any) error { data, err := json.Marshal(payload) if err != nil { return err } - req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(data)) if err != nil { return err }