Adding atunnel - networking component within ateom

cmd/atecontroller/internal/controllers/workerpool_apply.go

@@ -24,6 +24,13 @@ import (
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 )
 
+const (
+	atunnelIdentityVolume       = "atunnel-identity"
+	atunnelIdentityMountPath    = "/run/podidentity.podcert.ate.dev"
+	atunnelEgressTrustVolume    = "atunnel-egress-trust"
+	atunnelEgressTrustMountPath = "/run/servicedns.podcert.ate.dev"
+)
+
 // buildDeploymentApplyConfig constructs the SSA apply configuration for the
 // Deployment managed by a WorkerPool. Only fields owned by this controller
 // are declared here.
@@ -33,7 +40,16 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool) *appsv1ac.Deployment
 		WithImage(wp.Spec.AteomImage).
 		WithArgs(
 			"--pod-uid=$(POD_UID)",
+			"--atunnel-listen-address=0.0.0.0:443",
+			"--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem",
+			"--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem",
+			"--atunnel-egress-listen-address=0.0.0.0:15001",
+			"--atunnel-egress-trust-bundle="+atunnelEgressTrustMountPath+"/trust-bundle.pem",
 		).
+		WithPorts(corev1ac.ContainerPort().
+			WithName("https").
+			WithContainerPort(443).
+			WithProtocol(corev1.ProtocolTCP)).
 		WithSecurityContext(ateomSecurityContext(wp.Spec.SandboxClass)).
 		WithEnv(
 			corev1ac.EnvVar().
@@ -42,20 +58,61 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool) *appsv1ac.Deployment
 					WithFieldRef(corev1ac.ObjectFieldSelector().
 						WithFieldPath("metadata.uid"))),
 		).
-		WithVolumeMounts(corev1ac.VolumeMount().
-			WithName("run-ateom").
-			WithMountPath(ateompath.BasePath).
-			WithMountPropagation(corev1.MountPropagationHostToContainer))
+		WithVolumeMounts(
+			corev1ac.VolumeMount().
+				WithName("run-ateom").
+				WithMountPath(ateompath.BasePath).
+				WithMountPropagation(corev1.MountPropagationHostToContainer),
+			corev1ac.VolumeMount().
+				WithName(atunnelIdentityVolume).
+				WithMountPath(atunnelIdentityMountPath).
+				WithReadOnly(true),
+			corev1ac.VolumeMount().
+				WithName(atunnelEgressTrustVolume).
+				WithMountPath(atunnelEgressTrustMountPath).
+				WithReadOnly(true),
+		)
 
 	podSpecAC := corev1ac.PodSpec().
 		WithSecurityContext(corev1ac.PodSecurityContext().
 			WithRunAsUser(0).
 			WithRunAsGroup(0)).
-		WithVolumes(corev1ac.Volume().
-			WithName("run-ateom").
-			WithHostPath(corev1ac.HostPathVolumeSource().
-				WithPath(ateompath.BasePath).
-				WithType(corev1.HostPathDirectoryOrCreate)))
+		WithVolumes(
+			corev1ac.Volume().
+				WithName("run-ateom").
+				WithHostPath(corev1ac.HostPathVolumeSource().
+					WithPath(ateompath.BasePath).
+					WithType(corev1.HostPathDirectoryOrCreate)),
+			corev1ac.Volume().
+				WithName(atunnelIdentityVolume).
+				WithProjected(corev1ac.ProjectedVolumeSource().
+					WithSources(
+						corev1ac.VolumeProjection().
+							WithPodCertificate(corev1ac.PodCertificateProjection().
+								WithSignerName("podidentity.podcert.ate.dev/identity").
+								WithKeyType("ECDSAP256").
+								WithCredentialBundlePath("credential-bundle.pem")),
+						corev1ac.VolumeProjection().
+							WithClusterTrustBundle(corev1ac.ClusterTrustBundleProjection().
+								WithSignerName("podidentity.podcert.ate.dev/identity").
+								WithLabelSelector(metav1ac.LabelSelector().
+									WithMatchLabels(map[string]string{"podcert.ate.dev/canarying": "live"})).
+								WithPath("trust-bundle.pem")),
+					),
+				),
+			corev1ac.Volume().
+				WithName(atunnelEgressTrustVolume).
+				WithProjected(corev1ac.ProjectedVolumeSource().
+					WithSources(
+						corev1ac.VolumeProjection().
+							WithClusterTrustBundle(corev1ac.ClusterTrustBundleProjection().
+								WithSignerName("servicedns.podcert.ate.dev/identity").
+								WithLabelSelector(metav1ac.LabelSelector().
+									WithMatchLabels(map[string]string{"podcert.ate.dev/canarying": "live"})).
+								WithPath("trust-bundle.pem")),
+					),
+				),
+		)
 
 	applyWorkerPoolPodTemplate(podSpecAC, containerAC, wp.Spec.Template)
 	maybeApplyMicroVMPodShape(podSpecAC, containerAC, wp.Spec.SandboxClass)

cmd/atecontroller/internal/controllers/workerpool_apply_test.go

@@ -327,15 +327,57 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 		WithSecurityContext(corev1ac.PodSecurityContext().
 			WithRunAsUser(0).
 			WithRunAsGroup(0)).
-		WithVolumes(corev1ac.Volume().
-			WithName("run-ateom").
-			WithHostPath(corev1ac.HostPathVolumeSource().
-				WithPath(ateompath.BasePath).
-				WithType(corev1.HostPathDirectoryOrCreate))).
+		WithVolumes(
+			corev1ac.Volume().
+				WithName("run-ateom").
+				WithHostPath(corev1ac.HostPathVolumeSource().
+					WithPath(ateompath.BasePath).
+					WithType(corev1.HostPathDirectoryOrCreate)),
+			corev1ac.Volume().
+				WithName(atunnelIdentityVolume).
+				WithProjected(corev1ac.ProjectedVolumeSource().
+					WithSources(
+						corev1ac.VolumeProjection().
+							WithPodCertificate(corev1ac.PodCertificateProjection().
+								WithSignerName("podidentity.podcert.ate.dev/identity").
+								WithKeyType("ECDSAP256").
+								WithCredentialBundlePath("credential-bundle.pem")),
+						corev1ac.VolumeProjection().
+							WithClusterTrustBundle(corev1ac.ClusterTrustBundleProjection().
+								WithSignerName("podidentity.podcert.ate.dev/identity").
+								WithLabelSelector(metav1ac.LabelSelector().
+									WithMatchLabels(map[string]string{"podcert.ate.dev/canarying": "live"})).
+								WithPath("trust-bundle.pem")),
+					),
+				),
+			corev1ac.Volume().
+				WithName(atunnelEgressTrustVolume).
+				WithProjected(corev1ac.ProjectedVolumeSource().
+					WithSources(
+						corev1ac.VolumeProjection().
+							WithClusterTrustBundle(corev1ac.ClusterTrustBundleProjection().
+								WithSignerName("servicedns.podcert.ate.dev/identity").
+								WithLabelSelector(metav1ac.LabelSelector().
+									WithMatchLabels(map[string]string{"podcert.ate.dev/canarying": "live"})).
+								WithPath("trust-bundle.pem")),
+					),
+				),
+		).
 		WithContainers(corev1ac.Container().
 			WithName("ateom").
 			WithImage(wp.Spec.AteomImage).
-			WithArgs("--pod-uid=$(POD_UID)").
+			WithArgs(
+				"--pod-uid=$(POD_UID)",
+				"--atunnel-listen-address=0.0.0.0:443",
+				"--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem",
+				"--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem",
+				"--atunnel-egress-listen-address=0.0.0.0:15001",
+				"--atunnel-egress-trust-bundle="+atunnelEgressTrustMountPath+"/trust-bundle.pem",
+			).
+			WithPorts(corev1ac.ContainerPort().
+				WithName("https").
+				WithContainerPort(443).
+				WithProtocol(corev1.ProtocolTCP)).
 			WithSecurityContext(corev1ac.SecurityContext().
 				WithRunAsUser(0).
 				WithRunAsGroup(0).
@@ -350,10 +392,20 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 				WithValueFrom(corev1ac.EnvVarSource().
 					WithFieldRef(corev1ac.ObjectFieldSelector().
 						WithFieldPath("metadata.uid")))).
-			WithVolumeMounts(corev1ac.VolumeMount().
-				WithName("run-ateom").
-				WithMountPath(ateompath.BasePath).
-				WithMountPropagation(corev1.MountPropagationHostToContainer)).
+			WithVolumeMounts(
+				corev1ac.VolumeMount().
+					WithName("run-ateom").
+					WithMountPath(ateompath.BasePath).
+					WithMountPropagation(corev1.MountPropagationHostToContainer),
+				corev1ac.VolumeMount().
+					WithName(atunnelIdentityVolume).
+					WithMountPath(atunnelIdentityMountPath).
+					WithReadOnly(true),
+				corev1ac.VolumeMount().
+					WithName(atunnelEgressTrustVolume).
+					WithMountPath(atunnelEgressTrustMountPath).
+					WithReadOnly(true),
+			).
 			WithResources(corev1ac.ResourceRequirements()))
 
 	podSpecAC.NodeSelector = map[string]string{}

cmd/atecontroller/internal/controllers/workerpool_controller_test.go

@@ -138,8 +138,10 @@ func TestWorkerPoolCreatesDeployment(t *testing.T) {
 		if len(dep.OwnerReferences) == 0 || dep.OwnerReferences[0].Name != wp.Name {
 			return false, nil
 		}
-		return len(dep.Spec.Template.Spec.Volumes) == 1 &&
-			dep.Spec.Template.Spec.Volumes[0].Name == "run-ateom", nil
+		return len(dep.Spec.Template.Spec.Volumes) == 3 &&
+			dep.Spec.Template.Spec.Volumes[0].Name == "run-ateom" &&
+			dep.Spec.Template.Spec.Volumes[1].Name == atunnelIdentityVolume &&
+			dep.Spec.Template.Spec.Volumes[2].Name == atunnelEgressTrustVolume, nil
 	})
 }
 

cmd/atenet/README.md

@@ -9,8 +9,6 @@ This is built as a single binary for convenience in the prototyping.
 
 ## Cluster deployment
 
-![atenet diagram](atenet-diagram.png)
-
 ### router
 
 (Note: this deployment model combines Envoy dataplane with the router. This will
@@ -22,6 +20,9 @@ likely be split in the future for better scalability.)
   * atenet router
 * Service will expose:
   * Envoy port 80 and 443.
+* Upstream: Envoy's `ORIGINAL_DST` actor cluster dials the actor's in-worker
+  `atunnel` ingress server on the worker pod's port 443 over mTLS, using the
+  address `atenet router`'s ext_proc resolved into `x-ate-original-dst`.
 
 RBAC permissions:
 * read, list on ActorTemplate

cmd/atenet/internal/dns/README.md

@@ -4,7 +4,7 @@ The DNS Controller orchestrates the configuration needed to setup the ATE routin
 
 We want to resolve requests for <actor-name>.<atespace>.actors.resources.substrate.ate.dev to the router service address.
 
-* Stub resolver mode: orchestrate running a CoreDNS instance with the actor name mapped to the router service address.
+* Stub resolver mode: orchestrate running a CoreDNS instance with the actor name mapped to the atenet-router service address.
 
 Cluster resources:
 

cmd/atenet/internal/router/cmd.go

@@ -54,6 +54,9 @@ func NewRouterCmd() *cobra.Command {
 	cmd.Flags().DurationVar(&cfg.HealthInterval, "health-interval", 1*time.Second, "Interval for checking health of dependent services")
 	cmd.Flags().IntVar(&cfg.HttpsPort, "port-https", 8443, "TCP port for HTTPS workload traffic entering through the Envoy Router")
 	cmd.Flags().StringVar(&cfg.EnvoyCertPath, "envoy-cert-path", "", "Path to the Envoy certificate file.")
+	cmd.Flags().StringVar(&cfg.UpstreamClientCertPath, "upstream-client-cert-path", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "PEM credential bundle (cert+key) the router presents as the client cert when dialing the actor's atunnel ingress server over mTLS. Empty disables upstream mTLS (legacy plaintext pod-IP:80).")
+	cmd.Flags().StringVar(&cfg.UpstreamTrustPath, "upstream-trust-bundle-path", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "PEM trust bundle used to validate the actor's atunnel ingress server certificate.")
+	cmd.Flags().StringVar(&cfg.UpstreamSpiffePrefix, "upstream-spiffe-prefix", "spiffe://cluster.local/", "SPIFFE URI SAN prefix (trust domain) the actor's atunnel server cert must match. Empty falls back to default SAN check against the dialed pod IP (which SPIFFE-only certs never match).")
 	cmd.Flags().StringVar(&cfg.OtlpCollectorAddress, "otlp-collector-address", "", "host:port of the OTLP gRPC collector that Envoy reports tracing spans to (empty disables Envoy tracing)")
 	cmd.Flags().StringVar(&cfg.Auth.AteapiCAFile, "ateapi-ca-file", "", "PEM file with CAs trusted to verify the ateapi server cert. Required.")
 	cmd.Flags().StringVar(&cfg.Auth.AteapiClientCertPath, "ateapi-client-cert", "", "Credential bundle presented as the client certificate when dialing ateapi. Required unless --ateapi-use-token-auth is set, ignored otherwise.")

cmd/atenet/internal/router/config.go

@@ -48,8 +48,17 @@ type routerConfig struct {
 	HealthInterval time.Duration
 	HttpsPort      int
 	EnvoyCertPath  string
-	LogLevel       string
-	MetricsAddr    string
+	// UpstreamClientCertPath is the router's podidentity credential bundle
+	// (cert+key) presented as the client cert when dialing the actor's atunnel
+	// ingress server over mTLS. UpstreamTrustPath is the CA bundle used to
+	// validate that server. Empty UpstreamClientCertPath disables upstream mTLS.
+	UpstreamClientCertPath string
+	UpstreamTrustPath      string
+	// UpstreamSpiffePrefix validates the actor's atunnel server cert by its
+	// SPIFFE URI SAN prefix (trust domain) instead of the dialed pod IP.
+	UpstreamSpiffePrefix string
+	LogLevel             string
+	MetricsAddr          string
 	// OtlpCollectorAddress is the host:port of the OTLP gRPC collector that
 	// Envoy reports tracing spans to. Empty disables Envoy-side tracing.
 	OtlpCollectorAddress string

cmd/atenet/internal/router/extproc.go

@@ -170,14 +170,22 @@ func (s *ExtProcServer) handleRequestHeaders(
 			"actor %s routing failed", actorRef)
 	}
 
+	// The actor is reached through the in-worker atunnel ingress server, which
+	// listens on :443 (mTLS) and forwards to the actor's :80. The worker no
+	// longer DNATs pod-IP:80 to the actor, so the router dials :443 and the
+	// ORIGINAL_DST cluster's upstream TLS context presents the router's
+	// podidentity client cert (see buildOriginalDstCluster and
+	// buildUpstreamTransportSocket).
 	// TODO(bowei) -- handle more than port 80 on the actor.
-	targetAddr := net.JoinHostPort(workerIP, "80")
+	targetAddr := net.JoinHostPort(workerIP, "443")
 
 	slog.InfoContext(ctx, "Route ok", slog.Any("actor", actorRef), slog.String("targetAddr", targetAddr))
 
-	// Route by rewriting the :authority header.
+	// Route by telling the ORIGINAL_DST cluster which worker atunnel address to
+	// dial, without touching :authority — atunnel authorizes the actor by the
+	// original Host (actor DNS name).
 	mutation := &extprocv3.HeaderMutation{}
-	addAuthorityMutation(targetAddr, mutation)
+	addOriginalDstMutation(targetAddr, mutation)
 
 	return &extprocv3.HeadersResponse{
 		Response: &extprocv3.CommonResponse{

cmd/atenet/internal/router/extproc_out.go

@@ -32,12 +32,22 @@ type reqError struct {
 func (e *reqError) Error() string { return e.msg }
 func (e *reqError) Unwrap() error { return e.cause }
 
-func addAuthorityMutation(auth string, mut *extproc.HeaderMutation) {
+// addOriginalDstMutation sets the header the ORIGINAL_DST cluster reads to pick
+// the upstream address (the worker atunnel IP:443). Unlike an :authority
+// rewrite it leaves the request Host intact, so atunnel still sees the actor
+// DNS name and can authorize the active actor.
+//
+// Nothing strips this header from the incoming request, so overwrite rather
+// than append: a client-supplied value must never influence the address Envoy
+// dials. ext_proc mutations already default to replace, but the default is
+// split across the deprecated append field and append_action — pin it.
+func addOriginalDstMutation(dst string, mut *extproc.HeaderMutation) {
 	mut.SetHeaders = append(mut.SetHeaders,
 		&corev3.HeaderValueOption{
+			AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD,
 			Header: &corev3.HeaderValue{
-				Key:      ":authority",
-				RawValue: []byte(auth),
+				Key:      OriginalDstHeader,
+				RawValue: []byte(dst),
 			},
 		},
 	)

cmd/atenet/internal/router/extproc_test.go

@@ -171,7 +171,7 @@ func TestExtProcHeadersEvaluation(t *testing.T) {
 				},
 			},
 			expectErr:      false,
-			expectedTarget: "10.0.0.52:80",
+			expectedTarget: "10.0.0.52:443",
 		},
 	}
 
@@ -235,7 +235,7 @@ func TestExtProcHeadersEvaluation(t *testing.T) {
 			}
 
 			headerOption := mutation.GetSetHeaders()[0]
-			if strings.ToLower(headerOption.Header.Key) != ":authority" {
+			if strings.ToLower(headerOption.Header.Key) != OriginalDstHeader {
 				t.Errorf("invalid resulting dynamic parameter key: %s", headerOption.Header.Key)
 			}
 

cmd/atenet/internal/router/resumer_test.go

@@ -53,7 +53,6 @@ func TestActorResumer_ResumeActor(t *testing.T) {
 				resumeCalled++
 				return &ateapipb.ResumeActorResponse{
 					Actor: &ateapipb.Actor{
-						Metadata:   &ateapipb.ResourceMetadata{Name: testActorName},
 						Status:     ateapipb.Actor_STATUS_RUNNING,
 						AteomPodIp: expectedIP,
 					},
@@ -84,7 +83,6 @@ func TestActorResumer_ResumeActor(t *testing.T) {
 				}
 				return &ateapipb.ResumeActorResponse{
 					Actor: &ateapipb.Actor{
-						Metadata:   &ateapipb.ResourceMetadata{Name: testActorName},
 						Status:     ateapipb.Actor_STATUS_RUNNING,
 						AteomPodIp: expectedIP,
 					},
@@ -131,7 +129,6 @@ func TestActorResumer_ResumeActor(t *testing.T) {
 				time.Sleep(20 * time.Millisecond)
 				return &ateapipb.ResumeActorResponse{
 					Actor: &ateapipb.Actor{
-						Metadata:   &ateapipb.ResourceMetadata{Name: testActorName},
 						Status:     ateapipb.Actor_STATUS_RUNNING,
 						AteomPodIp: expectedIP,
 					},

cmd/atenet/internal/router/router.go

@@ -190,6 +190,7 @@ func (s *RouterServer) Run(ctx context.Context) error {
 	}
 
 	xdsSrv.SetTlsConfig(s.cfg.HttpsPort, s.cfg.EnvoyCertPath)
+	xdsSrv.SetUpstreamTls(s.cfg.UpstreamClientCertPath, s.cfg.UpstreamTrustPath, s.cfg.UpstreamSpiffePrefix)
 	if s.extprocSrv == nil {
 		routeDuration, err := newRouteDurationHistogram()
 		if err != nil {