atenet: Router checks the routed worker's SPIFFE ID on atunnel's cert (#2678)

Part of #2078 (Envoy dataplane, @EItanya @howardjohn for agw)

The router accepted any podidentity cert in the trust domain as atunnel,
so any pod in the cluster could pose as a worker. It now requires the
exact SPIFFE ID of the worker each request is routed to.

- atecontroller's syncer records the worker pod's SPIFFE ID on the
Worker (new `worker_spiffe_id`, built from the pod's namespace and
ServiceAccount), and ateapi copies it into the actor's worker
assignment. ateapi rejects a value that isn't a valid SPIFFE ID.
- ext_proc passes that ID to Envoy as is and refuses to route without
it. Envoy requires that exact SAN, on both the :443 and CONNECT paths.
`--upstream-spiffe-prefix` is removed; the cluster still checks the
trust domain.
- The Worker carries the SPIFFE ID rather than the ServiceAccount, so
the router doesn't need to know how the ID is built. A different ID
scheme only changes atecontroller.
- WorkerPools that don't set serviceAccountName now run as an
`ateworker` ServiceAccount the controller creates, instead of `default`,
so the worker's ID isn't shared with every other pod in the namespace.

Remaining gap: another pod running as the same ServiceAccount in the
same namespace can still pose as the worker. Closing it needs a signer
that puts the pod in the worker's SPIFFE ID; atecontroller would then
record that ID, with no router change.

#### Breaking change

- The router rejects workers registered before this change until their
pods are replaced. Pools on the default ServiceAccount roll
automatically; pools that set `spec.template.serviceAccountName` need a
restart. Upgrade atecontroller and ate-api-server before atenet-router.
- Workers now run as an `ateworker` ServiceAccount by default instead of
`default`. RBAC, Workload Identity bindings and imagePullSecrets
attached to `default` no longer apply to them; set `serviceAccountName:
default` or move those to `ateworker`.
- atenet-router no longer has `--upstream-spiffe-prefix`.
- atecontroller now needs create and patch on serviceaccounts (included
in the generated role).

> It's a good idea to open an issue first for discussion.

- [x] Tests pass
- [x] Appropriate changes to documentation are included in the PR

Author: LiorLieberman

cmd/ateapi/internal/apivalidation/actor.go

@@ -138,6 +138,11 @@ func ValidateCustom_WorkerAssignment_WorkerPodIps(_ context.Context, _ operation
 	return validateDualStackIPs(fldPath, value)
 }
 
+// This is needed because DV doesn't have a format for SPIFFE IDs.
+func ValidateCustom_WorkerAssignment_WorkerSpiffeId(_ context.Context, _ operation.Operation, fldPath *field.Path, value, _ *string) field.ErrorList {
+	return validateSPIFFEID(fldPath, *value)
+}
+
 func ValidateCustom_ExternalVolume_VolumeType(_ context.Context, _ operation.Operation, fldPath *field.Path, value, _ *string) field.ErrorList {
 	return resources.ValidateVolumeType(fldPath, *value)
 }

cmd/ateapi/internal/apivalidation/actor_test.go

@@ -354,6 +354,21 @@ func TestValidateActorUpdate(t *testing.T) {
 		want: field.ErrorList{
 			field.TooMany(field.NewPath("status", "worker_assignment", "worker_pod_ips"), 3, 2).WithOrigin("maxItems"),
 		},
+	}, {
+		name:   "set valid actor.status.worker_assignment.worker_spiffe_id",
+		oldVal: validInput(withStatus()),
+		newVal: validOutput(withStatus(withWorkerAssignment(func(wa *ateapipb.WorkerAssignment) {
+			wa.WorkerSpiffeId = "spiffe://cluster.local/ns/ns/sa/ateworker"
+		}))),
+	}, {
+		name:   "invalid actor.status.worker_assignment.worker_spiffe_id",
+		oldVal: validInput(withStatus()),
+		newVal: validOutput(withStatus(withWorkerAssignment(func(wa *ateapipb.WorkerAssignment) {
+			wa.WorkerSpiffeId = "ns/ns/sa/ateworker"
+		}))),
+		want: field.ErrorList{
+			field.Invalid(field.NewPath("status", "worker_assignment", "worker_spiffe_id"), nil, ""),
+		},
 	}, {
 		name:   "clear actor.status.worker_assignment",
 		oldVal: validInput(withStatus(withWorkerAssignment())),

cmd/ateapi/internal/apivalidation/common.go

@@ -22,6 +22,7 @@ import (
 	"github.com/agent-substrate/substrate/internal/preview"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"github.com/spiffe/go-spiffe/v2/spiffeid"
 	"k8s.io/apimachinery/pkg/api/operation"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
@@ -119,3 +120,12 @@ func validateDualStackIPs(fldPath *field.Path, ips []string) field.ErrorList {
 	}
 	return errs
 }
+
+// validateSPIFFEID checks that id is a SPIFFE ID. Presence and length are
+// enforced by tags.
+func validateSPIFFEID(fldPath *field.Path, id string) field.ErrorList {
+	if _, err := spiffeid.FromString(id); err != nil {
+		return field.ErrorList{field.Invalid(fldPath, id, err.Error())}
+	}
+	return nil
+}

cmd/ateapi/internal/apivalidation/worker.go

@@ -134,6 +134,11 @@ func ValidateCustom_Worker_Ips(_ context.Context, _ operation.Operation, fldPath
 	return validateDualStackIPs(fldPath, value)
 }
 
+// This is needed because DV doesn't have a format for SPIFFE IDs.
+func ValidateCustom_Worker_WorkerSpiffeId(_ context.Context, _ operation.Operation, fldPath *field.Path, value, _ *string) field.ErrorList {
+	return validateSPIFFEID(fldPath, *value)
+}
+
 // epoch only moves forward: a lower value would make Actors placed during the
 // current run look older than it, and crash them.
 func ValidateCustom_Worker_Epoch(_ context.Context, op operation.Operation, fldPath *field.Path, value, oldValue *int64) field.ErrorList {

cmd/ateapi/internal/apivalidation/worker_test.go

@@ -142,6 +142,23 @@ func TestValidateCreateWorkerRequest(t *testing.T) {
 			w.Ips = []string{"10.1.2.3", "fd00::1", "10.1.2.4"}
 		})),
 		want: field.ErrorList{field.TooMany(field.NewPath("worker", "ips"), 3, 2).WithOrigin("maxItems")},
+	}, {
+		name: "valid worker_spiffe_id",
+		req: validReq(validWorker(apiWorkerName, func(w *ateapipb.Worker) {
+			w.WorkerSpiffeId = "spiffe://cluster.local/ns/ate-system/sa/ateworker"
+		})),
+	}, {
+		name: "invalid worker_spiffe_id",
+		req: validReq(validWorker(apiWorkerName, func(w *ateapipb.Worker) {
+			w.WorkerSpiffeId = "https://cluster.local/ns/ate-system/sa/ateworker"
+		})),
+		want: field.ErrorList{field.Invalid(field.NewPath("worker", "worker_spiffe_id"), nil, "")},
+	}, {
+		name: "worker_spiffe_id too long",
+		req: validReq(validWorker(apiWorkerName, func(w *ateapipb.Worker) {
+			w.WorkerSpiffeId = "spiffe://cluster.local/" + strings.Repeat("a", 2048)
+		})),
+		want: field.ErrorList{field.TooLong(field.NewPath("worker", "worker_spiffe_id"), nil, 2048).WithOrigin("maxLength")},
 	}, {
 		name: "sandbox_class too long",
 		req:  validReq(validWorker(apiWorkerName, func(w *ateapipb.Worker) { w.SandboxClass = strings.Repeat("x", 64) })),

cmd/ateapi/internal/apivalidation/zz_generated.validation.go

@@ -610,6 +610,7 @@ func workerAssignmentFrom(w *ateapipb.Worker, snapRuntime *ateapipb.SandboxRunti
 		WorkerPodUid:    w.GetWorkerPodUid(),
 		WorkerPodIps:    slices.Clone(w.GetIps()),
 		NodeName:        w.GetNodeName(),
+		WorkerSpiffeId:  w.GetWorkerSpiffeId(),
 		// The runtime the actor is started or restored under. Snapshots taken
 		// during this assignment are stamped with it, so a worker that later
 		// changes its default does not relabel state captured on the old one.

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -2689,3 +2689,13 @@ func TestCanRelaxRuntime(t *testing.T) {
 		})
 	}
 }
+
+// atenet-router requires the assignment's SPIFFE ID on atunnel's certificate,
+// so the worker's has to reach it.
+func TestWorkerAssignmentFromCopiesSPIFFEID(t *testing.T) {
+	const id = "spiffe://cluster.local/ns/workers/sa/substrate-worker"
+	got := workerAssignmentFrom(&ateapipb.Worker{WorkerSpiffeId: id}, nil)
+	if got.GetWorkerSpiffeId() != id {
+		t.Errorf("worker SPIFFE ID = %q, want %q", got.GetWorkerSpiffeId(), id)
+	}
+}

cmd/ateapi/internal/controlapi/workflow_resume_test.go

@@ -551,7 +551,7 @@ func applyWorkerPoolPodTemplate(
 	podSpecAC.NodeSelector = map[string]string{}
 	podSpecAC.Tolerations = []corev1ac.TolerationApplyConfiguration{}
 	podSpecAC.WithPriorityClassName("")
-	podSpecAC.WithServiceAccountName("default")
+	podSpecAC.WithServiceAccountName(installdefaults.WorkerServiceAccount)
 	podSpecAC.WithAffinity(corev1ac.Affinity())
 	resourcesAC := corev1ac.ResourceRequirements()
 	containerAC.WithResources(resourcesAC)

cmd/atecontroller/internal/controllers/workerpool_apply.go

@@ -905,7 +905,7 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 		sandboxClassTolerationAC(atev1alpha1.SandboxClassGvisor),
 	}
 	podSpecAC.WithPriorityClassName("")
-	podSpecAC.WithServiceAccountName("default")
+	podSpecAC.WithServiceAccountName("ateworker")
 	podSpecAC.WithAffinity(corev1ac.Affinity())
 	podSpecAC.WithTerminationGracePeriodSeconds(workerTerminationGracePeriodSeconds)
 	if mutatePodSpec != nil {