diff --git a/cmd/agent/main.go b/cmd/agent/main.go index c17965e..04320f1 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -32,6 +32,9 @@ import ( "github.com/RebuildStackCo/runtime-agent/internal/collector" "github.com/RebuildStackCo/runtime-agent/internal/config" + "github.com/RebuildStackCo/runtime-agent/internal/nodeauth" + "github.com/RebuildStackCo/runtime-agent/internal/nodeintake" + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" "github.com/RebuildStackCo/runtime-agent/internal/rollup" "github.com/RebuildStackCo/runtime-agent/internal/sink" ) @@ -94,31 +97,33 @@ func runController(ctx context.Context, logger *slog.Logger, args []string) erro return fmt.Errorf("loading configuration: %w", err) } - clientset, host, err := connect() + clientset, restConfig, err := connect() if err != nil { return fmt.Errorf("connecting to cluster: %w", err) } - logger.Info("connected to cluster", "host", host) + logger.Info("connected to cluster", "host", restConfig.Host) - return run(ctx, logger, clientset, cfg) + return run(ctx, logger, clientset, restConfig, cfg) } // connect builds a clientset from the in-cluster service account when -// running as a pod, falling back to the local kubeconfig otherwise. -func connect() (kubernetes.Interface, string, error) { +// running as a pod, falling back to the local kubeconfig otherwise. It returns +// the REST config too: the node-intake receiver builds its JWKS HTTP client +// from it, reusing the in-cluster CA and bearer credential (ADR 0010). +func connect() (kubernetes.Interface, *rest.Config, error) { config, err := rest.InClusterConfig() if err != nil { rules := clientcmd.NewDefaultClientConfigLoadingRules() config, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig(rules, &clientcmd.ConfigOverrides{}).ClientConfig() if err != nil { - return nil, "", fmt.Errorf("no in-cluster config and no kubeconfig: %w", err) + return nil, nil, fmt.Errorf("no in-cluster config and no kubeconfig: %w", err) } } clientset, err := kubernetes.NewForConfig(config) if err != nil { - return nil, "", err + return nil, nil, err } - return clientset, config.Host, nil + return clientset, config, nil } // coverageInterval is how often the aggregate coverage counters are logged. @@ -126,7 +131,7 @@ const coverageInterval = time.Minute // run is the agent's lifecycle: it starts, works until ctx is canceled, and // returns. -func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interface, cfg config.Config) error { +func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interface, restConfig *rest.Config, cfg config.Config) error { logger.Info("agent starting", "version", version) defer logger.Info("agent stopping") @@ -258,14 +263,28 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac } } }() - var wg sync.WaitGroup - var mu sync.Mutex - var errs []error - for name, run := range map[string]func(context.Context) error{ + + tasks := map[string]func(context.Context) error{ "pods": podWatcher.Run, "nodes": nodeWatcher.Run, "usage": usagePoller.Run, - } { + } + + // The node-intake receiver is optional (only the ebpf/node profile ships a + // DaemonSet). When enabled it becomes one more lifecycle task, validating + // node tokens locally against the cluster JWKS (ADR 0010). + if cfg.NodeIntake.Enabled { + intake, err := buildNodeIntake(logger, restConfig, cfg.NodeIntake) + if err != nil { + return fmt.Errorf("configuring node intake: %w", err) + } + tasks["node-intake"] = intake.Run + } + + var wg sync.WaitGroup + var mu sync.Mutex + var errs []error + for name, run := range tasks { wg.Add(1) go func() { defer wg.Done() @@ -281,3 +300,47 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac logCoverage() return errors.Join(errs...) } + +// buildNodeIntake constructs the node-intake receiver: a token verifier backed +// by the cluster JWKS (fetched over the API server's in-cluster transport, no +// TokenReview) and an HTTP server that decodes node reports. For this slice the +// report is authenticated, decoded, and logged in aggregate; the join into a +// payload lands in the next slice. +func buildNodeIntake(logger *slog.Logger, restConfig *rest.Config, cfg config.NodeIntake) (*nodeintake.Server, error) { + addr := cfg.ListenAddress + if addr == "" { + addr = config.DefaultNodeIntakeListenAddress + } + audience := cfg.Audience + if audience == "" { + audience = config.DefaultNodeIntakeAudience + } + + httpClient, err := rest.HTTPClientFor(restConfig) + if err != nil { + return nil, fmt.Errorf("building JWKS http client: %w", err) + } + keys := &nodeauth.CachingKeySource{ + Source: &nodeauth.HTTPKeySource{IssuerBaseURL: restConfig.Host, Client: httpClient}, + } + verifier := nodeauth.NewVerifier(keys, audience, nodeauth.WithExpectedSubject(cfg.ExpectedSubject)) + + onReport := func(id nodeauth.Identity, report nodescan.Report) { + // Aggregate-only: identities of filtered-out binaries are already gone + // (the node filtered them, ADR 0009); here we log what arrived and from + // whom. The per-workload join and payload come in the next slice. + logger.Info("node report received", + "from_subject", id.Subject, + "node", report.Node, + "binaries", len(report.Binaries), + "processes_scanned", report.Counters.ProcessesScanned, + "go_found", report.Counters.GoFound, + "filtered_infra", report.Counters.FilteredInfra, + "unreadable", report.Counters.Unreadable, + ) + } + + handler := nodeintake.NewHandler(verifier, logger, onReport) + logger.Info("node intake enabled", "addr", addr, "audience", audience, "expected_subject", cfg.ExpectedSubject) + return nodeintake.NewServer(addr, handler, logger), nil +} diff --git a/cmd/agent/main_test.go b/cmd/agent/main_test.go index 24b363b..be4f388 100644 --- a/cmd/agent/main_test.go +++ b/cmd/agent/main_test.go @@ -8,6 +8,7 @@ import ( "time" "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/rest" "github.com/RebuildStackCo/runtime-agent/internal/config" ) @@ -17,7 +18,9 @@ func TestRunStopsOnContextCancel(t *testing.T) { logger := slog.New(slog.NewTextHandler(io.Discard, nil)) done := make(chan error, 1) - go func() { done <- run(ctx, logger, fake.NewClientset(), config.Config{}) }() + // A minimal REST config; node intake is disabled in this config, so it is + // never used to build a JWKS client. + go func() { done <- run(ctx, logger, fake.NewClientset(), &rest.Config{}, config.Config{}) }() cancel() select { diff --git a/cmd/agent/node.go b/cmd/agent/node.go index 1f9ec0a..3f43172 100644 --- a/cmd/agent/node.go +++ b/cmd/agent/node.go @@ -4,30 +4,48 @@ import ( "context" "flag" "log/slog" + "os" "time" "github.com/RebuildStackCo/runtime-agent/internal/nodescan" ) +// defaultControllerTokenPath is where the DaemonSet's projected +// serviceAccountToken volume mounts the controller-audience token (ADR 0010). +// It is a mount path, not a secret value. +const defaultControllerTokenPath = "/var/run/secrets/rebuildstack.co/controller-token/token" // #nosec G101 -- filesystem path, not an embedded credential + // runNode is the node role's lifecycle. It scans on-node processes for Go -// build information and writes the result to the structured log. It builds NO -// Kubernetes client and opens no external connection: the DaemonSet's -// ServiceAccount holds zero RBAC and everything the scanner reads is under -// /proc (ADR 0009). Delivery of these findings to the controller is a later -// slice; here the output is the log line. +// build information, writes the result to the structured log, and — when a +// controller endpoint is configured — ships the result to the controller over +// HTTP, authenticated by a projected ServiceAccount token (ADR 0010). It builds +// NO Kubernetes client: the DaemonSet's ServiceAccount holds zero RBAC and the +// projected token is audience-bound to the controller, so it cannot reach the +// API server (ADR 0009). Everything the scanner reads is under /proc. With no +// endpoint configured the output is the log line only, as in the first slice. func runNode(ctx context.Context, logger *slog.Logger, args []string) error { fs := flag.NewFlagSet(roleNode, flag.ExitOnError) procRoot := fs.String("proc", "/proc", "path to the proc filesystem to scan") interval := fs.Duration("interval", time.Minute, "scan interval; 0 runs a single pass and exits") + endpoint := fs.String("controller-endpoint", "", "controller URL to POST scan reports to; empty runs log-only") + tokenPath := fs.String("token-path", defaultControllerTokenPath, "path to the projected controller-audience ServiceAccount token") if err := fs.Parse(args); err != nil { return err } + // The node identity for the report; the DaemonSet sets NODE_NAME via the + // downward API. Empty is acceptable (the controller keys facts by pod UID + // and container ID, not node name). + node := os.Getenv("NODE_NAME") + shipper := newReportShipper(*endpoint, *tokenPath, node) + scanner := nodescan.NewScanner(*procRoot, nodescan.NewModuleFilter(nodescan.DefaultInfraModulePrefixes)) logger.Info("node scanner starting", "version", version, "proc_root", *procRoot, "interval", interval.String(), + "controller_endpoint", *endpoint, + "node", node, ) defer logger.Info("node scanner stopping") @@ -60,6 +78,21 @@ func runNode(ctx context.Context, logger *slog.Logger, args []string) error { "filtered_infra", res.Counters.FilteredInfra, "unreadable", res.Counters.Unreadable, ) + + // Ship to the controller when configured. Best-effort: a delivery + // failure is logged and the next pass retries — the controller + // rebuilds inventory from re-scans, so a lost report costs nothing + // (ADR 0010). + if shipper != nil { + if err := shipper.ship(ctx, res); err != nil { + logger.Error("shipping report to controller failed", "error", err) + } else { + logger.Info("report shipped to controller", + "go_found", res.Counters.GoFound, + "endpoint", *endpoint, + ) + } + } } scanOnce() diff --git a/cmd/agent/nodeship.go b/cmd/agent/nodeship.go new file mode 100644 index 0000000..ced4dcd --- /dev/null +++ b/cmd/agent/nodeship.go @@ -0,0 +1,101 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" + + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" +) + +// reportShipper delivers a node scan result to the controller over HTTP +// (ADR 0010). The node always initiates; authentication is a projected +// ServiceAccount token read fresh from tokenPath on every send, so the +// kubelet's rotation of that file needs no restart. Nothing is persisted on the +// node — the token lives only on the kubelet-managed tmpfs path (ADR 0008). +type reportShipper struct { + endpoint string + tokenPath string + node string + client *http.Client +} + +// newReportShipper builds a shipper for endpoint, or returns nil when endpoint +// is empty — the log-only mode that keeps the node role usable without a +// controller (and keeps the existing node e2e unchanged). +func newReportShipper(endpoint, tokenPath, node string) *reportShipper { + if endpoint == "" { + return nil + } + return &reportShipper{ + endpoint: endpoint, + tokenPath: tokenPath, + node: node, + client: &http.Client{ + Timeout: 30 * time.Second, + }, + } +} + +// ship POSTs one scan result. It re-reads the token per call (rotation), sets +// the bearer credential, and treats any non-2xx as an error. Delivery is +// best-effort by design: a failure is logged by the caller and the next scan +// pass retries — the controller reconstructs inventory from re-scans, so a lost +// report costs nothing (ADR 0010, loss-harmless). +func (s *reportShipper) ship(ctx context.Context, res nodescan.Result) error { + token, err := s.readToken() + if err != nil { + return fmt.Errorf("reading controller token: %w", err) + } + + report := nodescan.Report{ + Node: s.node, + Binaries: res.Binaries, + Counters: res.Counters, + } + body, err := json.Marshal(report) + if err != nil { + return fmt.Errorf("encoding report: %w", err) + } + + // The endpoint is the controller URL from the node's own configuration + // (a flag set by the DaemonSet manifest / ConfigMap, invariant 1), not + // caller-influenced input; the node has exactly one place to send. + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.endpoint, bytes.NewReader(body)) // #nosec G704 -- endpoint is operator-set config, not tainted input + if err != nil { + return fmt.Errorf("building request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := s.client.Do(req) // #nosec G704 -- endpoint is operator-set config, not tainted input + if err != nil { + return fmt.Errorf("posting report: %w", err) + } + defer func() { _ = resp.Body.Close() }() + // Drain a bounded amount so the connection can be reused. + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("controller rejected report: status %d", resp.StatusCode) + } + return nil +} + +// readToken reads and trims the projected token file. +func (s *reportShipper) readToken() (string, error) { + raw, err := os.ReadFile(s.tokenPath) // #nosec G304,G703 -- path is an operator-set flag, the projected-token mount + if err != nil { + return "", err + } + token := strings.TrimSpace(string(raw)) + if token == "" { + return "", fmt.Errorf("token file %s is empty", s.tokenPath) + } + return token, nil +} diff --git a/cmd/agent/nodeship_test.go b/cmd/agent/nodeship_test.go new file mode 100644 index 0000000..4596ade --- /dev/null +++ b/cmd/agent/nodeship_test.go @@ -0,0 +1,146 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" +) + +// capturedRequest records what the fake controller received. +type capturedRequest struct { + mu sync.Mutex + auth string + report nodescan.Report + hits int +} + +func fakeController(t *testing.T, captured *capturedRequest, status int) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + captured.mu.Lock() + defer captured.mu.Unlock() + captured.hits++ + captured.auth = r.Header.Get("Authorization") + _ = json.NewDecoder(r.Body).Decode(&captured.report) + w.WriteHeader(status) + })) + t.Cleanup(srv.Close) + return srv +} + +func writeToken(t *testing.T, content string) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "token") + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatal(err) + } + return path +} + +func sampleResult() nodescan.Result { + return nodescan.Result{ + Binaries: []nodescan.BinaryInfo{{ + PID: 7, GoVersion: "go1.26.1", MainModule: "example.com/app", + PGO: true, PodUID: "pod-uid", ContainerID: "container-id", + }}, + Counters: nodescan.Counters{ProcessesScanned: 5, GoFound: 1, FilteredInfra: 2, Unreadable: 1}, + } +} + +func TestNewReportShipperNilWithoutEndpoint(t *testing.T) { + if s := newReportShipper("", "/tok", "node"); s != nil { + t.Fatal("shipper should be nil (log-only) when no endpoint is configured") + } +} + +func TestReportShipperShipsWithBearerToken(t *testing.T) { + captured := &capturedRequest{} + srv := fakeController(t, captured, http.StatusAccepted) + tokenPath := writeToken(t, " secret-token\n") // padded to check trimming + + s := newReportShipper(srv.URL, tokenPath, "kind-worker") + if err := s.ship(context.Background(), sampleResult()); err != nil { + t.Fatalf("ship: %v", err) + } + + captured.mu.Lock() + defer captured.mu.Unlock() + if captured.auth != "Bearer secret-token" { + t.Errorf("Authorization = %q, want %q", captured.auth, "Bearer secret-token") + } + if captured.report.Node != "kind-worker" || len(captured.report.Binaries) != 1 { + t.Errorf("received report = %+v", captured.report) + } + if captured.report.Binaries[0].MainModule != "example.com/app" { + t.Errorf("received binary = %+v", captured.report.Binaries[0]) + } + if captured.report.Counters.FilteredInfra != 2 { + t.Errorf("received counters = %+v", captured.report.Counters) + } +} + +func TestReportShipperReadsTokenPerSend(t *testing.T) { + captured := &capturedRequest{} + srv := fakeController(t, captured, http.StatusAccepted) + tokenPath := writeToken(t, "first-token") + + s := newReportShipper(srv.URL, tokenPath, "n") + if err := s.ship(context.Background(), sampleResult()); err != nil { + t.Fatalf("first ship: %v", err) + } + captured.mu.Lock() + first := captured.auth + captured.mu.Unlock() + + // Rotate the token file; the next send must pick up the new value. + if err := os.WriteFile(tokenPath, []byte("rotated-token"), 0o600); err != nil { + t.Fatal(err) + } + if err := s.ship(context.Background(), sampleResult()); err != nil { + t.Fatalf("second ship: %v", err) + } + captured.mu.Lock() + second := captured.auth + captured.mu.Unlock() + + if first != "Bearer first-token" { + t.Errorf("first auth = %q", first) + } + if second != "Bearer rotated-token" { + t.Errorf("second auth = %q, want the rotated token", second) + } +} + +func TestReportShipperErrorsOnNon2xx(t *testing.T) { + captured := &capturedRequest{} + srv := fakeController(t, captured, http.StatusUnauthorized) + tokenPath := writeToken(t, "tok") + + s := newReportShipper(srv.URL, tokenPath, "n") + if err := s.ship(context.Background(), sampleResult()); err == nil { + t.Fatal("ship should error on a non-2xx response") + } +} + +func TestReportShipperErrorsOnMissingToken(t *testing.T) { + captured := &capturedRequest{} + srv := fakeController(t, captured, http.StatusAccepted) + + s := newReportShipper(srv.URL, filepath.Join(t.TempDir(), "absent"), "n") + if err := s.ship(context.Background(), sampleResult()); err == nil { + t.Fatal("ship should error when the token file is absent") + } + captured.mu.Lock() + defer captured.mu.Unlock() + if captured.hits != 0 { + t.Error("no request should be sent without a token") + } +} diff --git a/deploy/node-daemonset.yaml b/deploy/node-daemonset.yaml index e226180..f195b34 100644 --- a/deploy/node-daemonset.yaml +++ b/deploy/node-daemonset.yaml @@ -40,7 +40,11 @@ spec: app.kubernetes.io/component: node spec: serviceAccountName: runtime-agent-node - # No API credential in the container: the node role holds no client. + # No *API* credential in the container: the default ServiceAccount token + # (audience = the API server) is never mounted, so the node role cannot + # call the API server. The projected token below is a *different* token, + # audience-bound to the controller (ADR 0010): it authenticates only to + # the controller and is rejected by the API server. Zero RBAC still holds. automountServiceAccountToken: false # Share the host PID namespace so the scanner sees every process on the # node and can open its /proc//exe (ADR 0009). This is the only @@ -66,6 +70,18 @@ spec: - /host/proc - -interval - 60s + # Deliver findings to the controller over the cluster network only + # (ADR 0010). The node always initiates; the controller Service is + # the single in-cluster target. Remove this arg to run log-only. + - -controller-endpoint + - http://runtime-agent-controller.runtime-agent.svc:8080/v1/node-inventory + env: + # The node name for the report; the controller keys facts by pod UID + # and container ID, so this is for observability, not attribution. + - name: NODE_NAME + valueFrom: + fieldRef: + fieldPath: spec.nodeName securityContext: privileged: false allowPrivilegeEscalation: false @@ -94,8 +110,25 @@ spec: readOnly: true # Pick up mounts that appear on the host after the pod starts. mountPropagation: HostToContainer + - name: controller-token + mountPath: /var/run/secrets/rebuildstack.co/controller-token + readOnly: true volumes: - name: host-proc hostPath: path: /proc type: Directory + # The one credential the node holds (ADR 0010): a projected + # ServiceAccount token, audience-bound to the controller, kubelet-issued + # and rotated on this tmpfs path. It is NOT the API-server token — the + # API server rejects this audience — so the node still cannot reach the + # API. Nothing is persisted: the token lives only here, for this pod's + # lifetime. The audience must match the controller's configured value + # (config.nodeIntake.audience; default "rebuildstack-controller"). + - name: controller-token + projected: + sources: + - serviceAccountToken: + audience: rebuildstack-controller + expirationSeconds: 3600 + path: token diff --git a/docs/security.md b/docs/security.md index 454a307..7323437 100644 --- a/docs/security.md +++ b/docs/security.md @@ -42,6 +42,11 @@ runtime-agent node # DaemonSet (optional) — eBPF profiling and Go bi ``` - The node role sends data to the controller over the cluster network only. +- The controller exposes an in-cluster-only HTTP receiver for those node + reports (ADR 0010): it accepts data pushes from the node DaemonSet and + answers ack/error only. It is not reachable from outside the cluster and + returns nothing the node acts on — the one-way rule ([§1](#1-design-principles), + principle 2) applied one level down. - The controller aggregates, filters, and ships data to `` over mTLS. - Configuration is read exclusively from the Helm-rendered ConfigMap. The @@ -91,6 +96,16 @@ bound to nothing and its token is not mounted — so it appears in no row here. | `nodes/proxy` | get | Poll each kubelet's `/stats/summary` and `/metrics/cadvisor` through the API server for usage counters: CPU, memory working set, CFS throttling, PSI where exposed (ADR 0006). **Honest disclosure:** this verb technically permits any kubelet GET endpoint through the API server, including node logs. The agent calls exactly the two stats paths above — auditable, since all kubelet access lives in a single poll loop | | its own identity Secret | get, update (namespaced Role, `resourceNames`) | Persist the in-cluster-generated key and certificate across rescheduling (ADR 0008). Helm pre-creates the Secret; the agent owns only its content. No `create` (cannot be name-scoped), no `list`/`watch`/`delete` — the agent can neither enumerate nor touch any other Secret. Not emitted when `persistence.enabled: true` | +**Authenticating node reports adds no RBAC.** When the node DaemonSet is +installed (ADR 0010), the controller authenticates each node's projected +ServiceAccount token by verifying its signature against the cluster's published +JWKS **locally**. It does **not** call `TokenReview`, so no `create` on +`authentication.k8s.io/tokenreviews` — or any other verb — is added for this. +Resolving the cluster's OIDC discovery and JWKS endpoints is a read-only +non-resource URL `GET`, granted cluster-wide to ServiceAccounts by the built-in +`system:service-account-issuer-discovery` role on typical clusters; it is a +read, never a write. + ### Cluster-wide vs. namespace-scoped installation - **Cluster-wide (default):** a single ClusterRole with the rules above. @@ -114,7 +129,7 @@ bound to nothing and its token is not mounted — so it appears in no row here. | Pod pprof endpoints | controller → pods | Pull `/debug/pprof/profile` and `/debug/pprof/heap` from Go services that already expose them | Only pods that pass the workload filters are probed; ports taken from `containerPorts`, never blind scans. `pprof`/`ebpf` profiles only | | kubelet stats, proxied | controller → API server | Poll `/stats/summary` and `/metrics/cadvisor` on every kubelet for usage counters (ADR 0006) | Goes through the API server proxy (`nodes/proxy`, §4) — the agent opens **no direct connection to kubelets**. A direct-kubelet transport for very large clusters would be a documented change here, not a silent one | | Backend egress | controller → one fixed domain | Ship aggregated rollups and filtered profiles | mTLS, pinned domain. The only cross-boundary connection in the system. A NetworkPolicy restricting controller egress to this domain plus in-cluster targets is shipped with the chart | -| Node role | node → controller only | Deliver profiles and binary metadata for aggregation | The DaemonSet has **no** external egress | +| Node → controller reports | node → controller, in-cluster only | Deliver on-node Go build-info findings (and, later, profiles) for aggregation (ADR 0010) | Plain HTTP on the cluster network; the node always initiates and authenticates with a projected controller-audience ServiceAccount token, validated locally (no `TokenReview`). The controller's receiver answers ack/error only, is not reachable from outside the cluster, and a shipped NetworkPolicy restricts it to the DaemonSet. The DaemonSet has **no** external egress. Honest disclosure: the token and the (already-filtered) facts travel in cleartext in-cluster; a party that can already sniff pod traffic could replay the short-lived token to post fabricated inventory facts for this one cluster — bounded, one-way, no monetary or read capability (ADR 0010) | --- @@ -238,17 +253,25 @@ It has two functions with different privilege needs, documented separately below (ADR 0009). Whichever runs, one property holds for both: **the node role has no Kubernetes API access whatsoever.** Its ServiceAccount is bound to no Role, RoleBinding, ClusterRole, or ClusterRoleBinding — not one RBAC rule -exists for it — and `automountServiceAccountToken: false` keeps its token out -of the container. It appears in none of the RBAC tables in [§4](#4-kubernetes-api-access-rbac) -because it holds nothing. This is enforced by the API server (an unbound -identity cannot authorize anything), not by agent configuration. +exists for it — and `automountServiceAccountToken: false` keeps the **default +API token** out of the container. When the node delivers findings to the +controller (ADR 0010) it mounts exactly one credential: a projected token +**audience-bound to the controller**, which the API server rejects (wrong +audience) and which grants nothing anyway (the ServiceAccount holds zero RBAC). +So the node still cannot reach the API — now by two independent barriers, wrong +audience *and* no grant — and it appears in none of the RBAC tables in +[§4](#4-kubernetes-api-access-rbac) because it holds nothing there. This is +enforced by the API server (an unbound identity cannot authorize anything), not +by agent configuration. ### 7.1 The Go-binary scanner (the current node role) This is what the node DaemonSet runs today. It enumerates processes under `/proc`, reads the Go build information embedded in each executable, ties it to a pod and container through the process cgroup, filters infrastructure on the -node, and reports the result. It loads no eBPF and opens no socket. +node, and reports the result. It loads no eBPF; the only socket it opens is the +outbound connection to the controller to deliver its findings (ADR 0010) — never +to the internet, never to the API server. | Privilege | Why | |---|---| @@ -397,9 +420,10 @@ Stated explicitly so it does not have to be asked: - No cloud provider credentials, IAM roles, or billing API access. Node pricing uses a static price table plus your stated discount. - No external egress from nodes — only the controller crosses the boundary. -- No API access from the node role at all — its ServiceAccount holds zero RBAC - and its token is not mounted ([§7](#7-node-privileges), ADR 0009). The node - role never calls the Kubernetes API. +- No API access from the node role at all — its ServiceAccount holds zero RBAC, + and the only token it mounts is audience-bound to the controller, which the + API server rejects ([§7](#7-node-privileges), ADR 0009/0010). The default API + token is never mounted, and the node role never calls the Kubernetes API. - No dynamic configuration from the backend — the agent cannot be reconfigured remotely. diff --git a/internal/config/config.go b/internal/config/config.go index 9246393..9c1bcc3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,10 +12,39 @@ import ( // Config is the root of the agent configuration file. type Config struct { - Filters Filters `json:"filters"` - Spool Spool `json:"spool"` + Filters Filters `json:"filters"` + Spool Spool `json:"spool"` + NodeIntake NodeIntake `json:"nodeIntake"` } +// NodeIntake configures the controller's receiver for node-role reports +// (ADR 0010). It is off unless enabled, so a controller-only install (no node +// DaemonSet) opens no port. When enabled, the controller validates each node +// token locally against the cluster JWKS — no TokenReview, no new RBAC verb. +type NodeIntake struct { + // Enabled turns the receiver on. Off by default. + Enabled bool `json:"enabled"` + // ListenAddress is where the receiver listens (net/http form, e.g. + // ":8080"). Empty selects the built-in default when enabled. + ListenAddress string `json:"listenAddress"` + // Audience is the projected-token audience the receiver requires; it must + // match the audience the node DaemonSet's serviceAccountToken projection + // requests. Empty selects the built-in default. + Audience string `json:"audience"` + // ExpectedSubject pins the accepted token subject to the node role's + // ServiceAccount, e.g. + // "system:serviceaccount:runtime-agent:runtime-agent-node". Empty accepts + // any subject that satisfies the audience — set it in production. + ExpectedSubject string `json:"expectedSubject"` +} + +// Defaults for the node-intake receiver, applied when enabled and the +// corresponding field is empty. +const ( + DefaultNodeIntakeListenAddress = ":8080" + DefaultNodeIntakeAudience = "rebuildstack-controller" +) + // Spool configures the local payload sink (ADR 0003, ADR 0007). The // directory's durability is the installation's choice: emptyDir by default, // a PersistentVolume for those who want unacknowledged data to survive diff --git a/internal/nodeintake/handler.go b/internal/nodeintake/handler.go new file mode 100644 index 0000000..eeda173 --- /dev/null +++ b/internal/nodeintake/handler.go @@ -0,0 +1,123 @@ +// Package nodeintake is the controller's receiver for the node→controller +// inventory channel (ADR 0010). The node initiates every connection and pushes +// its already-filtered scan facts here; the receiver authenticates the caller +// with a projected ServiceAccount token verified locally (nodeauth), decodes +// the report, and hands it to a callback. The response carries only an +// acknowledgment or an error — never configuration or commands: this ingress is +// one-way, mirroring the agent↔backend contract one level down (ADR 0001). +package nodeintake + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "strings" + + "github.com/RebuildStackCo/runtime-agent/internal/nodeauth" + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" +) + +// defaultMaxBodyBytes caps a single node report. A node's kept-binary list is +// small (customer Go workloads on one node), so this is generous headroom, not +// a tuned limit; its purpose is to refuse a runaway or hostile body outright. +const defaultMaxBodyBytes = 4 << 20 // 4 MiB + +// TokenVerifier validates a bearer token and returns the caller's identity. +// *nodeauth.Verifier satisfies it; tests supply a fake. +type TokenVerifier interface { + Verify(ctx context.Context, token string) (nodeauth.Identity, error) +} + +// Handler authenticates and decodes node reports, then invokes onReport with +// the verified caller identity and the report. onReport must not block the +// request goroutine for long; it does the cheap in-memory join in this design. +type Handler struct { + verifier TokenVerifier + onReport func(nodeauth.Identity, nodescan.Report) + logger *slog.Logger + maxBodyBytes int64 +} + +// NewHandler builds a receiver handler. onReport may be nil (the report is +// authenticated and decoded, then dropped — useful before the join exists). +func NewHandler(verifier TokenVerifier, logger *slog.Logger, onReport func(nodeauth.Identity, nodescan.Report)) *Handler { + return &Handler{ + verifier: verifier, + onReport: onReport, + logger: logger, + maxBodyBytes: defaultMaxBodyBytes, + } +} + +// ServeHTTP implements the receiver. It accepts POST with a bearer token, +// verifies it, decodes a nodescan.Report, and acknowledges. Every failure path +// returns a status and a short reason; no failure reveals whether a token was +// well-formed-but-untrusted versus malformed beyond the coarse 401. +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + token, ok := bearerToken(r) + if !ok { + http.Error(w, "missing bearer token", http.StatusUnauthorized) + return + } + identity, err := h.verifier.Verify(r.Context(), token) + if err != nil { + // Do not echo the verifier's reason to the caller; log it for the + // operator instead. A caller learns only "unauthorized". + h.logger.Warn("node report rejected: token verification failed", "error", err) + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + var report nodescan.Report + dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, h.maxBodyBytes)) + dec.DisallowUnknownFields() + if err := dec.Decode(&report); err != nil { + if _, ok := errors.AsType[*http.MaxBytesError](err); ok { + http.Error(w, "report too large", http.StatusRequestEntityTooLarge) + return + } + http.Error(w, "malformed report", http.StatusBadRequest) + return + } + // Reject trailing data after the JSON object — a well-formed report is a + // single object, nothing more. + if dec.More() { + http.Error(w, "malformed report", http.StatusBadRequest) + return + } + + if h.onReport != nil { + h.onReport(identity, report) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(struct { + Status string `json:"status"` + }{Status: "accepted"}) +} + +// bearerToken extracts the token from an "Authorization: Bearer " +// header, case-insensitively on the scheme. +func bearerToken(r *http.Request) (string, bool) { + h := r.Header.Get("Authorization") + if h == "" { + return "", false + } + const scheme = "bearer " + if len(h) < len(scheme) || !strings.EqualFold(h[:len(scheme)], scheme) { + return "", false + } + token := strings.TrimSpace(h[len(scheme):]) + if token == "" { + return "", false + } + return token, true +} diff --git a/internal/nodeintake/handler_test.go b/internal/nodeintake/handler_test.go new file mode 100644 index 0000000..5d36fbd --- /dev/null +++ b/internal/nodeintake/handler_test.go @@ -0,0 +1,169 @@ +package nodeintake + +import ( + "context" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/RebuildStackCo/runtime-agent/internal/nodeauth" + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" +) + +// fakeVerifier accepts one fixed token and rejects everything else. +type fakeVerifier struct { + accept string + identity nodeauth.Identity +} + +func (f fakeVerifier) Verify(_ context.Context, token string) (nodeauth.Identity, error) { + if token != f.accept { + return nodeauth.Identity{}, errors.New("invalid token") + } + return f.identity, nil +} + +const goodToken = "good-token" + +func testHandler(t *testing.T, onReport func(nodeauth.Identity, nodescan.Report)) *Handler { + t.Helper() + v := fakeVerifier{accept: goodToken, identity: nodeauth.Identity{ + Subject: "system:serviceaccount:runtime-agent:runtime-agent-node", + Namespace: "runtime-agent", + ServiceAccount: "runtime-agent-node", + }} + return NewHandler(v, slog.New(slog.NewTextHandler(io.Discard, nil)), onReport) +} + +func post(t *testing.T, h http.Handler, token, body string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPost, reportPath, strings.NewReader(body)) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +const validBody = `{"node":"kind-worker","binaries":[{"pid":42,"go_version":"go1.26.1","main_module":"example.com/app","pgo":true,"pod_uid":"abc","container_id":"def"}],"counters":{"processes_scanned":10,"go_found":1,"filtered_infra":3,"unreadable":2}}` + +func TestHandlerAcceptsValidReport(t *testing.T) { + var got *nodescan.Report + var gotID nodeauth.Identity + h := testHandler(t, func(id nodeauth.Identity, r nodescan.Report) { + gotID = id + got = &r + }) + + rec := post(t, h, goodToken, validBody) + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want %d; body %q", rec.Code, http.StatusAccepted, rec.Body.String()) + } + if got == nil { + t.Fatal("onReport not invoked") + } + if got.Node != "kind-worker" || len(got.Binaries) != 1 { + t.Errorf("decoded report = %+v, want node kind-worker with 1 binary", got) + } + if got.Binaries[0].MainModule != "example.com/app" || !got.Binaries[0].PGO { + t.Errorf("decoded binary = %+v", got.Binaries[0]) + } + if got.Counters.FilteredInfra != 3 || got.Counters.Unreadable != 2 { + t.Errorf("decoded counters = %+v", got.Counters) + } + if gotID.ServiceAccount != "runtime-agent-node" { + t.Errorf("identity = %+v, want the node ServiceAccount", gotID) + } +} + +func TestHandlerRejectsMissingToken(t *testing.T) { + called := false + h := testHandler(t, func(nodeauth.Identity, nodescan.Report) { called = true }) + rec := post(t, h, "", validBody) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401", rec.Code) + } + if called { + t.Error("onReport invoked for an unauthenticated request") + } +} + +func TestHandlerRejectsBadToken(t *testing.T) { + called := false + h := testHandler(t, func(nodeauth.Identity, nodescan.Report) { called = true }) + rec := post(t, h, "wrong-token", validBody) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401", rec.Code) + } + if called { + t.Error("onReport invoked for a rejected token") + } +} + +func TestHandlerRejectsMalformedBody(t *testing.T) { + h := testHandler(t, func(nodeauth.Identity, nodescan.Report) { + t.Error("onReport invoked for a malformed body") + }) + rec := post(t, h, goodToken, `{"node": "x", not json`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", rec.Code) + } +} + +func TestHandlerRejectsUnknownFields(t *testing.T) { + h := testHandler(t, func(nodeauth.Identity, nodescan.Report) { + t.Error("onReport invoked for a body with unknown fields") + }) + rec := post(t, h, goodToken, `{"node":"x","surprise":true}`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 (unknown field rejected)", rec.Code) + } +} + +func TestHandlerRejectsTrailingData(t *testing.T) { + h := testHandler(t, func(nodeauth.Identity, nodescan.Report) { + t.Error("onReport invoked for a body with trailing data") + }) + rec := post(t, h, goodToken, validBody+`{"another":"object"}`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 (trailing data rejected)", rec.Code) + } +} + +func TestHandlerRejectsOversizeBody(t *testing.T) { + h := testHandler(t, func(nodeauth.Identity, nodescan.Report) { + t.Error("onReport invoked for an oversize body") + }) + h.maxBodyBytes = 64 // tiny cap for the test + rec := post(t, h, goodToken, validBody) + if rec.Code != http.StatusRequestEntityTooLarge { + t.Fatalf("status = %d, want 413", rec.Code) + } +} + +func TestHandlerRejectsWrongMethod(t *testing.T) { + h := testHandler(t, nil) + req := httptest.NewRequest(http.MethodGet, reportPath, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("status = %d, want 405", rec.Code) + } + if allow := rec.Header().Get("Allow"); allow != http.MethodPost { + t.Errorf("Allow header = %q, want POST", allow) + } +} + +func TestHandlerNilOnReportStillAccepts(t *testing.T) { + h := NewHandler(fakeVerifier{accept: goodToken}, slog.New(slog.NewTextHandler(io.Discard, nil)), nil) + rec := post(t, h, goodToken, validBody) + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want 202 with a nil onReport", rec.Code) + } +} diff --git a/internal/nodeintake/server.go b/internal/nodeintake/server.go new file mode 100644 index 0000000..48565e9 --- /dev/null +++ b/internal/nodeintake/server.go @@ -0,0 +1,66 @@ +package nodeintake + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net/http" + "time" +) + +// The receiver's path. Nodes POST their reports here. +const reportPath = "/v1/node-inventory" + +// Server runs the node-intake HTTP receiver until its context is canceled. +// It is one more long-lived task in the controller's lifecycle, alongside the +// watchers and the usage poller. +type Server struct { + addr string + handler http.Handler + logger *slog.Logger +} + +// NewServer wires a receiver listening on addr that dispatches the report path +// to handler. addr follows net/http conventions (e.g. ":8080"). +func NewServer(addr string, handler http.Handler, logger *slog.Logger) *Server { + mux := http.NewServeMux() + mux.Handle(reportPath, handler) + return &Server{addr: addr, handler: mux, logger: logger} +} + +// Run serves until ctx is canceled, then shuts down gracefully. It returns nil +// on a clean shutdown and an error only if the listener fails to start or serve +// for a reason other than the shutdown. +func (s *Server) Run(ctx context.Context) error { + srv := &http.Server{ + Addr: s.addr, + Handler: s.handler, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 2 * time.Minute, + } + + serveErr := make(chan error, 1) + go func() { + s.logger.Info("node intake receiver listening", "addr", s.addr, "path", reportPath) + serveErr <- srv.ListenAndServe() + }() + + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + return fmt.Errorf("node intake shutdown: %w", err) + } + s.logger.Info("node intake receiver stopped") + return nil + case err := <-serveErr: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return fmt.Errorf("node intake serve: %w", err) + } +} diff --git a/internal/nodescan/report.go b/internal/nodescan/report.go new file mode 100644 index 0000000..672b174 --- /dev/null +++ b/internal/nodescan/report.go @@ -0,0 +1,15 @@ +package nodescan + +// Report is the node role's scan result as delivered to the controller +// (ADR 0010): the kept (non-infrastructure) binaries and the aggregate counters +// for one pass, tagged with the node that produced them. It is the wire +// contract of the node→controller channel — the node marshals it, the +// controller unmarshals it — and carries only already-filtered facts (the +// on-node module filter, ADR 0009, ran before this shape exists). No identity +// of a filtered-out or unreadable binary appears here; those are only the +// counters (CLAUDE.md invariant 6). +type Report struct { + Node string `json:"node"` + Binaries []BinaryInfo `json:"binaries"` + Counters Counters `json:"counters"` +}