diff --git a/Makefile b/Makefile index 7a42567..065488e 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test build compose-check sample e2e e2e-ha e2e-kubernetes e2e-down helm-lint +.PHONY: test build compose-check sample e2e e2e-ha e2e-kubernetes e2e-policy e2e-down helm-lint test: go test ./... @@ -47,6 +47,11 @@ e2e-ha: e2e-kubernetes: bash test/e2e/kubernetes/run.sh +# Builds a Calico cluster and proves the Agent egress NetworkPolicy is enforced. +# Set KEEP=true to leave the cluster running. +e2e-policy: + bash test/e2e/kubernetes/policy-check.sh + helm-lint: helm lint deploy/helm/agent-platform \ --set secrets.secretEncryptionKey=lint --set secrets.workerToken=lint \ diff --git a/README.md b/README.md index 327fd49..f08c4a9 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,8 @@ sessions while retaining a replayable audit trail. - Live per-Session Provider and model switching with immutable defaults - Runtime leases with same-Runner Worker restart recovery - Horizontally scalable Control Plane with a shared PostgreSQL tool-call queue -- Pluggable runtime backend: Docker containers or Kubernetes Pods +- Pluggable runtime backend: Docker containers or Kubernetes Pods, with Agent + egress enforcement verified against Calico - Helm chart deploying the platform onto Kubernetes with no Docker socket - Agent WebSocket reconnection with idempotent resend of unacknowledged events - Database-backed Session commands with Agent ACK and reconnect replay diff --git a/deploy/helm/README.md b/deploy/helm/README.md index 4b776b5..b508d94 100644 --- a/deploy/helm/README.md +++ b/deploy/helm/README.md @@ -60,8 +60,9 @@ cluster-scoped. The Control Plane Pod label `app.kubernetes.io/name: agent-platform-control-plane` is not cosmetic: the egress NetworkPolicy the worker writes selects Control Plane Pods by it. Renaming it cuts Agent Pods off from the platform on a CNI that -enforces policy. See [Kubernetes](../../docs/kubernetes.md) for what that policy -does and does not guarantee. +enforces policy — a case `make e2e-policy` covers directly. See +[Kubernetes](../../docs/kubernetes.md) for what that policy does and does not +guarantee. ## Ingress diff --git a/docs/kubernetes.md b/docs/kubernetes.md index 03f3b6d..9666f3e 100644 --- a/docs/kubernetes.md +++ b/docs/kubernetes.md @@ -49,6 +49,17 @@ before treating the allowlist as a control on Kubernetes. Failing to write the policy is logged and not fatal, because a cluster may withhold the permission while the proxy still mediates the traffic that goes through it. +Enforcement is verified against Calico rather than assumed. `make e2e-policy` +builds a kind cluster with Calico in place of kindnet and asserts that a Pod +carrying the Session labels cannot open a connection to the internet, that it +can still reach the Control Plane, and that DNS still resolves. + +Two details make those assertions mean something. The forbidden probe dials an +IP rather than a name, because an unresolvable name fails identically to a +blocked connection — that alone would let the test pass with no policy at all. +And every case runs an unlabelled Pod through the same probe first: if the +cluster simply has no route out, the test says so instead of reporting success. + ## stdio MCP servers Each call runs in its own throwaway Pod, created with stdin open and driven over diff --git a/docs/threat-model.md b/docs/threat-model.md index cd37fcd..8b5c9d5 100644 --- a/docs/threat-model.md +++ b/docs/threat-model.md @@ -25,7 +25,9 @@ proxy is only a boundary if the Agent cannot route around it: under Docker that comes from an internal network, under Kubernetes from a NetworkPolicy the worker writes. That policy is inert on a CNI that does not enforce NetworkPolicy, in which case Agent Pods have unrestricted egress regardless of the Manifest -allowlist. See docs/kubernetes.md. +allowlist. Enforcement is verified against Calico, including a check that an +Agent Pod cannot open a direct connection to the internet; see +docs/kubernetes.md. ## Important limitation diff --git a/internal/worker/kubernetes_client.go b/internal/worker/kubernetes_client.go index 6439363..1dc314b 100644 --- a/internal/worker/kubernetes_client.go +++ b/internal/worker/kubernetes_client.go @@ -218,6 +218,7 @@ func (c *kubeClient) applyNetworkPolicy(ctx context.Context, policy map[string]a // podStatus is the slice of a Pod's status this runtime reads. type podStatus struct { Phase string `json:"phase"` + PodIP string `json:"podIP"` ContainerStatuses []containerStatus `json:"containerStatuses"` } diff --git a/internal/worker/kubernetes_policy_test.go b/internal/worker/kubernetes_policy_test.go new file mode 100644 index 0000000..9b723ab --- /dev/null +++ b/internal/worker/kubernetes_policy_test.go @@ -0,0 +1,199 @@ +package worker + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" +) + +// The egress proxy is only a boundary if an Agent cannot route around it. Under +// Docker that comes from an internal network with no route out; under +// Kubernetes it comes from the NetworkPolicy the worker writes. A policy object +// is inert on a CNI that does not enforce it, so these tests assert the block +// actually happens rather than that the object exists. +// +// They need a cluster whose CNI enforces NetworkPolicy — Calico, Cilium, or +// Antrea — and are skipped otherwise: +// +// KUBERNETES_POLICY_ENFORCED=true go test ./internal/worker -run IntegrationPolicy +func policyRuntime(t *testing.T) *Kubernetes { + t.Helper() + if envOr("KUBERNETES_POLICY_ENFORCED", "") != "true" { + t.Skip("set KUBERNETES_POLICY_ENFORCED=true on a cluster whose CNI enforces NetworkPolicy") + } + runtime := integrationRuntime(t) + if err := runtime.EnsureNetworkPolicy(context.Background()); err != nil { + t.Fatalf("EnsureNetworkPolicy: %v", err) + } + return runtime +} + +// probePod runs one shell command and reports its exit code. labels decide +// whether the egress policy applies to it. +func probePod(t *testing.T, runtime *Kubernetes, name string, labels map[string]string, script string) int { + t.Helper() + pod := map[string]any{ + "apiVersion": "v1", + "kind": "Pod", + "metadata": map[string]any{ + "name": name, + "namespace": runtime.Namespace(), + "labels": labels, + }, + "spec": map[string]any{ + "restartPolicy": "Never", + "containers": []any{map[string]any{ + "name": agentContainerName, + "image": "busybox:1.37", + "imagePullPolicy": "IfNotPresent", + "command": []string{"sh", "-c", script}, + }}, + }, + } + t.Cleanup(func() { + _ = runtime.client.deletePod(context.Background(), name, 0) + }) + if err := runtime.client.createPod(context.Background(), pod); err != nil { + t.Fatalf("create probe pod: %v", err) + } + + deadline := time.Now().Add(120 * time.Second) + for time.Now().Before(deadline) { + status, err := runtime.client.getPod(context.Background(), name) + if err != nil { + t.Fatalf("read probe pod: %v", err) + } + if code, done := status.exitCode(); done { + return code + } + time.Sleep(500 * time.Millisecond) + } + t.Fatalf("probe pod %s never finished", name) + return -1 +} + +func sessionProbeLabels() map[string]string { + return sessionLabels(uuid.NewString()) +} + +// controlPlaneStandIn starts a listener carrying the label the policy allows +// egress to, and returns its address. The real Control Plane carries the same +// label; see the Helm chart. +func controlPlaneStandIn(t *testing.T, runtime *Kubernetes) string { + t.Helper() + name := "policy-control-plane-" + uuid.NewString()[:8] + pod := map[string]any{ + "apiVersion": "v1", + "kind": "Pod", + "metadata": map[string]any{ + "name": name, + "namespace": runtime.Namespace(), + "labels": map[string]string{ + "app.kubernetes.io/name": "agent-platform-control-plane", + }, + }, + "spec": map[string]any{ + "restartPolicy": "Never", + "containers": []any{map[string]any{ + "name": agentContainerName, + "image": "busybox:1.37", + "imagePullPolicy": "IfNotPresent", + // Answer repeatedly so the probe is not racing a single accept. + "command": []string{"sh", "-c", "while true; do echo ok | nc -l -p 8080; done"}, + }}, + }, + } + t.Cleanup(func() { + _ = runtime.client.deletePod(context.Background(), name, 0) + }) + if err := runtime.client.createPod(context.Background(), pod); err != nil { + t.Fatalf("create control plane stand-in: %v", err) + } + + deadline := time.Now().Add(120 * time.Second) + for time.Now().Before(deadline) { + status, err := runtime.client.getPod(context.Background(), name) + if err != nil { + t.Fatalf("read control plane stand-in: %v", err) + } + if status.Phase == "Running" && status.PodIP != "" { + // Give the listener a moment to bind. + time.Sleep(2 * time.Second) + return status.PodIP + } + time.Sleep(500 * time.Millisecond) + } + t.Fatal("the control plane stand-in never started") + return "" +} + +// externalProbe dials a public address by IP. Using an address rather than a +// name keeps DNS out of the result: a name would fail to resolve and look +// identical to a blocked connection, which is how this test could pass without +// the policy doing anything. +// +// It also models the actual threat. The danger is not that an Agent cannot +// reach the API server; it is that an Agent ignores HTTP_PROXY and talks +// straight to the internet, past the Manifest allowlist. +const externalProbe = "nc -z -w 5 1.1.1.1 443" + +func TestIntegrationPolicyBlocksDirectEgressFromAnAgentPod(t *testing.T) { + runtime := policyRuntime(t) + + // Positive control first. Without it, "blocked" could equally mean the + // cluster has no route out, and the test would pass while proving nothing. + unrestricted := probePod(t, runtime, "policy-unlabelled-"+uuid.NewString()[:8], + map[string]string{"agent-platform/probe": "unrestricted"}, externalProbe) + if unrestricted != 0 { + t.Skipf("an unlabelled pod has no route out either (exit %d); "+ + "this cluster cannot distinguish policy enforcement from no connectivity", unrestricted) + } + + restricted := probePod(t, runtime, "policy-session-"+uuid.NewString()[:8], + sessionProbeLabels(), externalProbe) + if restricted == 0 { + t.Fatal("an Agent pod reached the internet directly; the Manifest egress " + + "allowlist can be bypassed by ignoring the proxy") + } +} + +func TestIntegrationPolicyAllowsTheControlPlane(t *testing.T) { + runtime := policyRuntime(t) + address := controlPlaneStandIn(t, runtime) + + // The Agent Protocol, the model gateway, and the egress proxy all live + // behind this, so the policy has to let it through. + code := probePod(t, runtime, "policy-allowed-"+uuid.NewString()[:8], + sessionProbeLabels(), "nc -z -w 5 "+address+" 8080") + if code != 0 { + t.Fatalf("an Agent pod could not reach the Control Plane (exit %d); "+ + "the policy is too strict and Sessions would not run", code) + } +} + +// DNS is explicitly allowed, because without it the proxy host cannot be +// resolved and every Session would fail to start. +// +// The query uses a fully qualified name: busybox nslookup does not apply the +// pod's search domains the way a normal resolver does, and a short name returns +// NXDOMAIN even with no policy in place. +const dnsProbe = "nslookup kubernetes.default.svc.cluster.local" + +func TestIntegrationPolicyAllowsDNS(t *testing.T) { + runtime := policyRuntime(t) + + unrestricted := probePod(t, runtime, "policy-dns-open-"+uuid.NewString()[:8], + map[string]string{"agent-platform/probe": "unrestricted"}, dnsProbe) + if unrestricted != 0 { + t.Skipf("DNS does not work in this cluster even without policy (exit %d)", unrestricted) + } + + restricted := probePod(t, runtime, "policy-dns-"+uuid.NewString()[:8], + sessionProbeLabels(), dnsProbe) + if restricted != 0 { + t.Fatalf("an Agent pod could not resolve DNS (exit %d); "+ + "the egress proxy host would be unresolvable and no Session could start", restricted) + } +} diff --git a/test/e2e/kubernetes/kind-calico.yaml b/test/e2e/kubernetes/kind-calico.yaml new file mode 100644 index 0000000..b9f59de --- /dev/null +++ b/test/e2e/kubernetes/kind-calico.yaml @@ -0,0 +1,10 @@ +# kind with Calico instead of kindnet. +# +# kindnet accepts NetworkPolicy objects and ignores them, so the Agent egress +# policy cannot be verified on a default kind cluster. Calico enforces it. +kind: Cluster +apiVersion: kind.x-k8s.io/v1alpha4 +networking: + disableDefaultCNI: true + # Calico's default pool. + podSubnet: "192.168.0.0/16" diff --git a/test/e2e/kubernetes/policy-check.sh b/test/e2e/kubernetes/policy-check.sh new file mode 100755 index 0000000..55969ce --- /dev/null +++ b/test/e2e/kubernetes/policy-check.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +# Verify that the Agent egress NetworkPolicy is actually enforced. +# +# The proxy is only a boundary if an Agent cannot route around it. On a CNI that +# ignores NetworkPolicy the policy object exists and does nothing, so this +# builds a cluster with Calico and asserts the block, with a positive control so +# a cluster that simply has no route out cannot make the test pass for free. +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +CLUSTER="${CLUSTER:-agent-platform-calico}" +CALICO_VERSION="${CALICO_VERSION:-v3.29.1}" +NAMESPACE="${NAMESPACE:-agent-platform-runtime}" +KEEP="${KEEP:-false}" + +cd "$REPO_ROOT" + +if ! kind get clusters 2>/dev/null | grep -qx "$CLUSTER"; then + echo "==> Creating a kind cluster with no default CNI" + kind create cluster --name "$CLUSTER" --config test/e2e/kubernetes/kind-calico.yaml + + echo "==> Installing Calico ${CALICO_VERSION}" + kubectl apply -f "https://raw.githubusercontent.com/projectcalico/calico/${CALICO_VERSION}/manifests/calico.yaml" + kubectl -n kube-system rollout status daemonset/calico-node --timeout=600s + kubectl wait --for=condition=Ready nodes --all --timeout=300s +fi +if [ "$KEEP" != "true" ]; then + trap 'kind delete cluster --name "$CLUSTER"' EXIT +fi + +echo "==> Granting the worker its namespace permissions" +kubectl apply -f docs/examples/kubernetes-rbac.yaml + +echo "==> Running the enforcement tests" +KUBERNETES_API_URL="$(kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}')" \ +KUBERNETES_TOKEN="$(kubectl -n "$NAMESPACE" create token agent-platform-worker --duration=1h)" \ +KUBERNETES_CA_CERT="$(kubectl config view --minify --raw \ + -o jsonpath='{.clusters[0].cluster.certificate-authority-data}' | base64 -d)" \ +KUBERNETES_POLICY_ENFORCED=true \ +RUNTIME_NAMESPACE="$NAMESPACE" \ + go test ./internal/worker -run IntegrationPolicy -v -count=1 -timeout 15m