feat: node → controller transport (ADR 0010)

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,39 +97,41 @@ 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.
 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
+}

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 {

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()

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
+}