From 60abb78cc61c23efc3c742e725df7f8b08ca184f Mon Sep 17 00:00:00 2001 From: Yufan Su Date: Mon, 27 Jul 2026 16:32:38 -0700 Subject: [PATCH 1/3] re-originate HTTPS request from router port 443 to actor port 443 --- cmd/atenet/internal/router/extproc.go | 20 ++- cmd/atenet/internal/router/extproc_in.go | 42 ++++++ cmd/atenet/internal/router/extproc_out.go | 16 +++ cmd/atenet/internal/router/extproc_test.go | 134 +++++++++++++++--- cmd/atenet/internal/router/xds.go | 103 ++++++++++++-- cmd/atenet/internal/router/xds_test.go | 150 +++++++++++++++++++++ cmd/ateom-gvisor/main.go | 59 ++++---- cmd/ateom-microvm/net.go | 61 +++++---- demos/counter/counter.go | 66 +++++++++ 9 files changed, 565 insertions(+), 86 deletions(-) diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 203c668d1..2d16d3480 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -31,7 +31,9 @@ import ( "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/propagation" "google.golang.org/grpc" + "google.golang.org/protobuf/types/known/structpb" + "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -92,7 +94,7 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer switch reqType := req.Request.(type) { case *extprocv3.ProcessingRequest_RequestHeaders: start := time.Now() - hResponse, rqm, target, tmplNs, tmplName, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders) + hResponse, rqm, target, tmplNs, tmplName, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders, req.GetAttributes()) elapsed := time.Since(start) if err != nil { slog.ErrorContext(stream.Context(), "Error during ext_proc RequestHeaders processing", slog.String("err", err.Error())) @@ -130,9 +132,12 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer func (s *ExtProcServer) handleRequestHeaders( ctx context.Context, reqHeaders *extprocv3.HttpHeaders, + attrs map[string]*structpb.Struct, ) (*extprocv3.HeadersResponse, *requestMetadata, string, string, string, error) { metadata := newRequestMetadata(reqHeaders.Headers.GetHeaders()) - slog.InfoContext(ctx, "Request", slog.String("host", metadata.host)) + tlsVersion := requestTLSVersion(attrs) + tlsIngress := tlsVersion != "" + slog.InfoContext(ctx, "Request", slog.String("host", metadata.host), slog.String("tlsVersion", tlsVersion)) // Envoy doesn't propagate trace context into the ext_proc gRPC // stream's metadata — the per-request traceparent arrives in the @@ -171,14 +176,21 @@ func (s *ExtProcServer) handleRequestHeaders( "actor %q routing failed", actorName) } - // TODO(bowei) -- handle more than port 80 on the actor. - targetAddr := net.JoinHostPort(workerIP, "80") + // Actors serve plaintext on 80. A request that arrived over TLS is + // re-originated as TLS to 443, so the actor sees the scheme the client used. + // TODO(bowei) -- handle actor ports beyond this fixed 80/443 pair. + targetAddr := net.JoinHostPort(workerIP, actorPortForIngress(tlsIngress)) slog.InfoContext(ctx, "Route ok", slog.String("actor", actorName), slog.String("targetAddr", targetAddr)) // Route by rewriting the :authority header. mutation := &extprocv3.HeaderMutation{} addAuthorityMutation(targetAddr, mutation) + if tlsIngress { + // :authority is now a bare pod IP, so the upstream TLS handshake takes + // its SNI from this header instead (see SNIHeader). + addSniMutation(resources.ActorDNSName(atespace, actorName), mutation) + } return &extprocv3.HeadersResponse{ Response: &extprocv3.CommonResponse{ diff --git a/cmd/atenet/internal/router/extproc_in.go b/cmd/atenet/internal/router/extproc_in.go index 5bfdb4974..d61389d16 100644 --- a/cmd/atenet/internal/router/extproc_in.go +++ b/cmd/atenet/internal/router/extproc_in.go @@ -20,6 +20,23 @@ import ( "github.com/agent-substrate/substrate/internal/resources" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + "google.golang.org/protobuf/types/known/structpb" +) + +const ( + // extProcFilterName keys this filter's entry in a ProcessingRequest's + // attributes map. + extProcFilterName = "envoy.filters.http.ext_proc" + // tlsVersionAttribute is the Envoy attribute holding the TLS version + // negotiated on the downstream connection. It is requested via + // ExternalProcessor.request_attributes (see buildHcm). Envoy omits it + // entirely for connections that did not complete a TLS handshake. + tlsVersionAttribute = "connection.tls_version" + + // Ports the actor serves on. Plaintext ingress is forwarded to 80; + // TLS ingress is re-originated as TLS to 443. + actorHTTPPort = "80" + actorHTTPSPort = "443" ) type requestMetadata struct { @@ -56,6 +73,31 @@ func newRequestMetadata(headers []*corev3.HeaderValue) *requestMetadata { } } +// requestTLSVersion reports the TLS version the request's downstream connection +// negotiated, from the attributes Envoy attaches to the ProcessingRequest. It is +// "" for a plaintext connection. +// +// The connection's own TLS state decides whether ingress was encrypted, rather +// than request.scheme, because Envoy derives :scheme from x-forwarded-proto and +// preserves a caller-supplied one unless configured as an edge proxy — reading +// the scheme would let a caller pick which actor port this routes to. Nothing on +// the wire can influence connection state. +// +// An absent attribute also yields "", which is read as plaintext — the safe +// direction, since it cannot turn a plaintext request into an upstream TLS one. +func requestTLSVersion(attrs map[string]*structpb.Struct) string { + return attrs[extProcFilterName].GetFields()[tlsVersionAttribute].GetStringValue() +} + +// actorPortForIngress returns the actor port a request is forwarded to. TLS +// ingress is re-originated as TLS to 443; plaintext ingress goes to 80. +func actorPortForIngress(tls bool) string { + if tls { + return actorHTTPSPort + } + return actorHTTPPort +} + // parseActorRef extracts the (atespace, actor name) an incoming request is // addressed to from its Host/:authority, which has the form // "..actors.resources.substrate.ate.dev" (optionally with a diff --git a/cmd/atenet/internal/router/extproc_out.go b/cmd/atenet/internal/router/extproc_out.go index 98d21d93e..45563c2bc 100644 --- a/cmd/atenet/internal/router/extproc_out.go +++ b/cmd/atenet/internal/router/extproc_out.go @@ -43,6 +43,22 @@ func addAuthorityMutation(auth string, mut *extproc.HeaderMutation) { ) } +// addSniMutation sets the SNI the upstream TLS handshake should use. +// OVERWRITE_IF_EXISTS_OR_ADD is required rather than the default append action: +// a caller that sends its own SNIHeader would otherwise have its value retained +// alongside ours, letting it influence the upstream handshake. +func addSniMutation(sni string, mut *extproc.HeaderMutation) { + mut.SetHeaders = append(mut.SetHeaders, + &corev3.HeaderValueOption{ + Header: &corev3.HeaderValue{ + Key: SNIHeader, + RawValue: []byte(sni), + }, + AppendAction: corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD, + }, + ) +} + func immediateResponse(statusCode envoy_type.StatusCode, message string) *extproc.ProcessingResponse { return &extproc.ProcessingResponse{ Response: &extproc.ProcessingResponse_ImmediateResponse{ diff --git a/cmd/atenet/internal/router/extproc_test.go b/cmd/atenet/internal/router/extproc_test.go index 81f97a944..a6cc1e3ad 100644 --- a/cmd/atenet/internal/router/extproc_test.go +++ b/cmd/atenet/internal/router/extproc_test.go @@ -31,6 +31,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/structpb" ) type mockClient struct { @@ -69,7 +70,7 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { }, } - _, metadata, target, _, _, err := s.handleRequestHeaders(context.Background(), reqHeaders) + _, metadata, target, _, _, err := s.handleRequestHeaders(context.Background(), reqHeaders, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -90,18 +91,44 @@ func TestHandleRequestHeadersDoesNotLogSensitiveData(t *testing.T) { } } +// tlsAttrs builds the attributes Envoy attaches to a ProcessingRequest when the +// downstream TLS version is requested. An empty version yields no attributes at +// all, which is what Envoy sends for a plaintext connection. +func tlsAttrs(version string) map[string]*structpb.Struct { + if version == "" { + return nil + } + return map[string]*structpb.Struct{ + extProcFilterName: { + Fields: map[string]*structpb.Value{ + tlsVersionAttribute: structpb.NewStringValue(version), + }, + }, + } +} + func TestExtProcHeadersEvaluation(t *testing.T) { const testUUID = "123e4567-e89b-12d3-a456-426614174000" tests := []struct { - name string - authority string + name string + authority string + // tlsVersion is the value of the connection.tls_version attribute Envoy + // reports; "" stands for the attribute being absent altogether, which is + // what a plaintext connection produces. + tlsVersion string + // extraHeaders are merged into the request on top of the base set, to + // check that caller-supplied headers cannot influence routing. + extraHeaders map[string]string resumeResp *ateapipb.ResumeActorResponse resumeErr error expectErr bool expectedErrStr string expectedStatus envoy_type.StatusCode expectedTarget string + // expectedSNI is the value expected in SNIHeader; "" means the header + // must not be set at all. + expectedSNI string }{ { name: "invalid host returns 404 identifying the host", @@ -163,8 +190,9 @@ func TestExtProcHeadersEvaluation(t *testing.T) { expectedStatus: envoy_type.StatusCode_InternalServerError, }, { - name: "Successful resume", - authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", + name: "Successful resume", + authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", + tlsVersion: "", resumeResp: &ateapipb.ResumeActorResponse{ Actor: &ateapipb.Actor{ AteomPodIp: "10.0.0.52", @@ -173,6 +201,43 @@ func TestExtProcHeadersEvaluation(t *testing.T) { expectErr: false, expectedTarget: "10.0.0.52:80", }, + { + name: "https ingress is re-originated to port 443 with the actor mesh name as SNI", + authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", + tlsVersion: "TLSv1.3", + resumeResp: &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{ + AteomPodIp: "10.0.0.52", + }, + }, + expectedTarget: "10.0.0.52:443", + expectedSNI: testUUID + ".team-a.actors.resources.substrate.ate.dev", + }, + { + name: "a missing tls version attribute falls back to plaintext", + authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", + resumeResp: &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{ + AteomPodIp: "10.0.0.52", + }, + }, + expectedTarget: "10.0.0.52:80", + }, + { + name: "caller-supplied proto and SNI headers cannot upgrade a plaintext request", + authority: testUUID + ".team-a.actors.resources.substrate.ate.dev", + tlsVersion: "", + extraHeaders: map[string]string{ + "x-forwarded-proto": "https", + SNIHeader: "evil.example", + }, + resumeResp: &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{ + AteomPodIp: "10.0.0.52", + }, + }, + expectedTarget: "10.0.0.52:80", + }, } for _, tc := range tests { @@ -191,17 +256,19 @@ func TestExtProcHeadersEvaluation(t *testing.T) { s := NewExtProcServer(50051, clientMock, nil) + headers := []*corev3.HeaderValue{ + {Key: ":path", Value: "/v1/actors/invoke"}, + {Key: ":authority", Value: tc.authority}, + {Key: ":method", Value: "POST"}, + } + for k, v := range tc.extraHeaders { + headers = append(headers, &corev3.HeaderValue{Key: k, Value: v}) + } reqHeaders := &extprocv3.HttpHeaders{ - Headers: &corev3.HeaderMap{ - Headers: []*corev3.HeaderValue{ - {Key: ":path", Value: "/v1/actors/invoke"}, - {Key: ":authority", Value: tc.authority}, - {Key: ":method", Value: "POST"}, - }, - }, + Headers: &corev3.HeaderMap{Headers: headers}, } - res, metadata, target, _, _, err := s.handleRequestHeaders(context.Background(), reqHeaders) + res, metadata, target, _, _, err := s.handleRequestHeaders(context.Background(), reqHeaders, tlsAttrs(tc.tlsVersion)) if tc.expectErr { if err == nil { t.Fatalf("expected error but got nil") @@ -230,17 +297,44 @@ func TestExtProcHeadersEvaluation(t *testing.T) { } mutation := res.Response.GetHeaderMutation() - if len(mutation.GetSetHeaders()) != 1 { - t.Fatalf("expected exactly one Header option set, found: %v", mutation.GetSetHeaders()) + set := map[string]*corev3.HeaderValueOption{} + for _, h := range mutation.GetSetHeaders() { + set[strings.ToLower(h.GetHeader().GetKey())] = h } - headerOption := mutation.GetSetHeaders()[0] - if strings.ToLower(headerOption.Header.Key) != ":authority" { - t.Errorf("invalid resulting dynamic parameter key: %s", headerOption.Header.Key) + wantHeaders := 1 + if tc.expectedSNI != "" { + wantHeaders = 2 + } + if len(set) != wantHeaders { + t.Fatalf("expected %d Header options set, found: %v", wantHeaders, mutation.GetSetHeaders()) } - if string(headerOption.Header.RawValue) != tc.expectedTarget { - t.Errorf("invalid destination mapping found: %s, expected: %s", headerOption.Header.RawValue, tc.expectedTarget) + authority, ok := set[":authority"] + if !ok { + t.Fatalf("no :authority mutation found in: %v", mutation.GetSetHeaders()) + } + if got := string(authority.GetHeader().GetRawValue()); got != tc.expectedTarget { + t.Errorf("invalid destination mapping found: %s, expected: %s", got, tc.expectedTarget) + } + + sni, ok := set[SNIHeader] + if tc.expectedSNI == "" { + if ok { + t.Errorf("plaintext request must not set %s, got %q", SNIHeader, sni.GetHeader().GetRawValue()) + } + } else { + if !ok { + t.Fatalf("no %s mutation found in: %v", SNIHeader, mutation.GetSetHeaders()) + } + if got := string(sni.GetHeader().GetRawValue()); got != tc.expectedSNI { + t.Errorf("SNI = %q, want %q", got, tc.expectedSNI) + } + // Anything but an overwrite would keep a caller-supplied value + // alongside ours, letting it reach the upstream handshake. + if got := sni.GetAppendAction(); got != corev3.HeaderValueOption_OVERWRITE_IF_EXISTS_OR_ADD { + t.Errorf("SNI append action = %v, want OVERWRITE_IF_EXISTS_OR_ADD", got) + } } // Confirm that query logs recorded metric trace details diff --git a/cmd/atenet/internal/router/xds.go b/cmd/atenet/internal/router/xds.go index 3d52657af..78f2af19f 100644 --- a/cmd/atenet/internal/router/xds.go +++ b/cmd/atenet/internal/router/xds.go @@ -64,8 +64,18 @@ const ( IngressHTTPListener = "ingress_http_listener" IngressHTTPSListener = "ingress_https_listener" RouteName = "substrate_routes" + RouteNameHTTPS = "substrate_routes_https" ClusterName = "ate-cluster" OtlpClusterName = "otel_collector_cluster" + DFPClusterName = "dynamic_forward_proxy_cluster" + DFPTLSClusterName = "dynamic_forward_proxy_tls_cluster" + + // SNIHeader carries the actor's mesh DNS name for the upstream TLS + // handshake. ext_proc rewrites :authority to a bare pod IP, which would + // make Envoy's auto-SNI derive the SNI from an IP literal, so the TLS + // cluster reads the SNI from this header instead (see + // UpstreamHttpProtocolOptions.override_auto_sni_header). + SNIHeader = "x-substrate-sni" ) // XdsServer implements an aggregated discovery service server for dynamic Envoy router nodes. @@ -151,7 +161,7 @@ func (x *XdsServer) UpdateSnapshot() error { // Clusters clusters := []types.Resource{ x.buildCluster(), - x.buildDynamicForwardProxyCluster(), + x.buildDynamicForwardProxyCluster(DFPClusterName, false), } if x.otlpHost != "" { clusters = append(clusters, x.buildOtlpCollectorCluster()) @@ -159,14 +169,21 @@ func (x *XdsServer) UpdateSnapshot() error { // Routes routes := []types.Resource{ - x.buildRoutes(), + x.buildRoutes(RouteName, DFPClusterName), } // Listeners listeners := []types.Resource{ x.buildListener(), } + + // The HTTPS ingress gets its own cluster and route configuration so that + // TLS is re-originated upstream only for traffic that arrived over TLS. + // All three resources are gated together to keep the snapshot internally + // consistent (no dangling RDS or CDS reference). if x.httpsPort > 0 { + clusters = append(clusters, x.buildDynamicForwardProxyCluster(DFPTLSClusterName, true)) + routes = append(routes, x.buildRoutes(RouteNameHTTPS, DFPTLSClusterName)) listeners = append(listeners, x.buildHttpsListener()) } @@ -329,17 +346,32 @@ func (x *XdsServer) buildOtlpCollectorCluster() *clusterv3.Cluster { } } -func (x *XdsServer) buildDynamicForwardProxyCluster() *clusterv3.Cluster { +// buildDynamicForwardProxyCluster builds a dynamic forward proxy cluster that +// takes its upstream address from the (ext_proc-rewritten) :authority. With +// useTLS the connection to the actor is re-originated as TLS, for traffic that +// arrived on the router's HTTPS listener. +// +// All dynamic forward proxy resources share one DNS cache, which Envoy keys by +// the cache config's name — the configs must stay byte-identical, so both +// clusters and the DFP filter use buildDnsCacheConfig unmodified. +func (x *XdsServer) buildDynamicForwardProxyCluster(name string, useTLS bool) *clusterv3.Cluster { dfpClusterConfig := &dfpclusterv3.ClusterConfig{ ClusterImplementationSpecifier: &dfpclusterv3.ClusterConfig_DnsCacheConfig{ DnsCacheConfig: buildDnsCacheConfig(), }, + // A dynamic forward proxy cluster with a TLS transport socket demands + // both auto_sni and auto_san_validation, and rejects the whole CDS + // update otherwise. SAN validation is meaningless here because the + // upstream context below carries no validation context at all, so the + // requirement has to be waived explicitly. Removing this waiver is part + // of the same change that adds a trusted_ca. + AllowInsecureClusterOptions: useTLS, } clusterConfigAny, _ := anypb.New(dfpClusterConfig) - return &clusterv3.Cluster{ - Name: "dynamic_forward_proxy_cluster", + cluster := &clusterv3.Cluster{ + Name: name, LbPolicy: clusterv3.Cluster_CLUSTER_PROVIDED, ClusterDiscoveryType: &clusterv3.Cluster_ClusterType{ ClusterType: &clusterv3.Cluster_CustomClusterType{ @@ -348,11 +380,53 @@ func (x *XdsServer) buildDynamicForwardProxyCluster() *clusterv3.Cluster { }, }, } + + if !useTLS { + return cluster + } + + // The actor's certificate is deliberately not verified: actors have no + // platform-issued identity yet, so they serve certs they mint themselves. + // This buys encryption in transit, not upstream authentication. The SNI + // below is still set correctly, so enabling verification later is a matter + // of adding a validation context here. + tlsAny, _ := anypb.New(&tlsv3.UpstreamTlsContext{ + CommonTlsContext: &tlsv3.CommonTlsContext{ + AlpnProtocols: []string{"http/1.1"}, + }, + }) + cluster.TransportSocket = &corev3.TransportSocket{ + Name: "envoy.transport_sockets.tls", + ConfigType: &corev3.TransportSocket_TypedConfig{ + TypedConfig: tlsAny, + }, + } + + // ExplicitHttpConfig pins HTTP/1.1, matching the protocol Envoy would pick + // implicitly for the plaintext cluster; it is required because + // UpstreamHttpProtocolOptions can only be carried inside HttpProtocolOptions, + // whose upstream_protocol_options oneof must be set. + protoOpts, _ := anypb.New(&httpv3.HttpProtocolOptions{ + UpstreamProtocolOptions: &httpv3.HttpProtocolOptions_ExplicitHttpConfig_{ + ExplicitHttpConfig: &httpv3.HttpProtocolOptions_ExplicitHttpConfig{ + ProtocolConfig: &httpv3.HttpProtocolOptions_ExplicitHttpConfig_HttpProtocolOptions{}, + }, + }, + UpstreamHttpProtocolOptions: &corev3.UpstreamHttpProtocolOptions{ + AutoSni: true, + OverrideAutoSniHeader: SNIHeader, + }, + }) + cluster.TypedExtensionProtocolOptions = map[string]*anypb.Any{ + "envoy.extensions.upstreams.http.v3.HttpProtocolOptions": protoOpts, + } + + return cluster } -func (x *XdsServer) buildRoutes() *routev3.RouteConfiguration { +func (x *XdsServer) buildRoutes(routeName, clusterName string) *routev3.RouteConfiguration { return &routev3.RouteConfiguration{ - Name: RouteName, + Name: routeName, VirtualHosts: []*routev3.VirtualHost{ { Name: "local_service", @@ -367,7 +441,7 @@ func (x *XdsServer) buildRoutes() *routev3.RouteConfiguration { Action: &routev3.Route_Route{ Route: &routev3.RouteAction{ ClusterSpecifier: &routev3.RouteAction_Cluster{ - Cluster: "dynamic_forward_proxy_cluster", + Cluster: clusterName, }, Timeout: durationpb.New(10 * time.Second), }, @@ -379,7 +453,7 @@ func (x *XdsServer) buildRoutes() *routev3.RouteConfiguration { } } -func (x *XdsServer) buildHcm(statPrefix string) *anypb.Any { +func (x *XdsServer) buildHcm(statPrefix, routeName string) *anypb.Any { extProcConfig, _ := anypb.New(&extprocv3filter.ExternalProcessor{ GrpcService: &corev3.GrpcService{ TargetSpecifier: &corev3.GrpcService_EnvoyGrpc_{ @@ -392,6 +466,11 @@ func (x *XdsServer) buildHcm(statPrefix string) *anypb.Any { MutationRules: &mutationrulesv3.HeaderMutationRules{ AllowAllRouting: &wrapperspb.BoolValue{Value: true}, }, + // The downstream connection's TLS version tells ext_proc which port to + // route the actor on. Unlike the request scheme it is connection state, + // so it is not derivable from anything the caller sends. See + // requestTLSVersion. + RequestAttributes: []string{tlsVersionAttribute}, // Explicitly configure the message timeout to avoid the 200ms default MessageTimeout: durationpb.New(5 * time.Second), ProcessingMode: &extprocv3filter.ProcessingMode{ @@ -448,7 +527,7 @@ func (x *XdsServer) buildHcm(statPrefix string) *anypb.Any { }, RouteSpecifier: &hcmv3.HttpConnectionManager_Rds{ Rds: &hcmv3.Rds{ - RouteConfigName: RouteName, + RouteConfigName: routeName, ConfigSource: &corev3.ConfigSource{ ResourceApiVersion: corev3.ApiVersion_V3, ConfigSourceSpecifier: &corev3.ConfigSource_Ads{ @@ -497,7 +576,7 @@ func (x *XdsServer) buildTracing() *hcmv3.HttpConnectionManager_Tracing { } func (x *XdsServer) buildListener() *listenerv3.Listener { - hcm := x.buildHcm("ingress_http") + hcm := x.buildHcm("ingress_http", RouteName) return &listenerv3.Listener{ Name: IngressHTTPListener, @@ -527,7 +606,7 @@ func (x *XdsServer) buildListener() *listenerv3.Listener { } func (x *XdsServer) buildHttpsListener() *listenerv3.Listener { - hcm := x.buildHcm("ingress_https") + hcm := x.buildHcm("ingress_https", RouteNameHTTPS) tlsConfig := &tlsv3.DownstreamTlsContext{ CommonTlsContext: &tlsv3.CommonTlsContext{ diff --git a/cmd/atenet/internal/router/xds_test.go b/cmd/atenet/internal/router/xds_test.go index 9a2c2460f..3bd3d40c2 100644 --- a/cmd/atenet/internal/router/xds_test.go +++ b/cmd/atenet/internal/router/xds_test.go @@ -17,6 +17,7 @@ package router import ( "context" "net" + "slices" "strings" "testing" "time" @@ -24,10 +25,64 @@ import ( clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3" listenerv3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" routev3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" + dfpclusterv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dynamic_forward_proxy/v3" + extprocv3filter "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3" + hcmv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3" + httpv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/upstreams/http/v3" cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" ) +// extProcConfig unpacks the ext_proc filter config from a listener's sole filter +// chain. +func extProcConfig(t *testing.T, l *listenerv3.Listener) *extprocv3filter.ExternalProcessor { + t.Helper() + + hcm := &hcmv3.HttpConnectionManager{} + if err := l.GetFilterChains()[0].GetFilters()[0].GetTypedConfig().UnmarshalTo(hcm); err != nil { + t.Fatalf("Failed to unmarshal connection manager for '%s': %v", l.GetName(), err) + } + + for _, f := range hcm.GetHttpFilters() { + if f.GetName() != extProcFilterName { + continue + } + cfg := &extprocv3filter.ExternalProcessor{} + if err := f.GetTypedConfig().UnmarshalTo(cfg); err != nil { + t.Fatalf("Failed to unmarshal ext_proc config for '%s': %v", l.GetName(), err) + } + return cfg + } + + t.Fatalf("Listener '%s' has no '%s' filter", l.GetName(), extProcFilterName) + return nil +} + +// assertRequestsTLSVersion checks that ext_proc is sent the downstream TLS +// version. Envoy silently drops attribute names it does not recognise, and an +// undelivered attribute reads as plaintext — which strands the HTTPS ingress on +// the actor's port 80 with no error anywhere in the control plane. +func assertRequestsTLSVersion(t *testing.T, l *listenerv3.Listener) { + t.Helper() + + if got := extProcConfig(t, l).GetRequestAttributes(); !slices.Contains(got, tlsVersionAttribute) { + t.Errorf("Listener '%s' requests attributes %v, want to include '%s'", l.GetName(), got, tlsVersionAttribute) + } +} + +// dfpClusterConfig unpacks the dynamic forward proxy config from a cluster's +// custom cluster type. +func dfpClusterConfig(t *testing.T, c *clusterv3.Cluster) *dfpclusterv3.ClusterConfig { + t.Helper() + + cfg := &dfpclusterv3.ClusterConfig{} + raw := c.GetClusterType().GetTypedConfig() + if err := raw.UnmarshalTo(cfg); err != nil { + t.Fatalf("Failed to unmarshal dynamic forward proxy config for '%s': %v", c.GetName(), err) + } + return cfg +} + func TestXdsServer_UpdateSnapshot(t *testing.T) { server := NewXdsServer(18000) server.SetConfig(8081, 50052, "10.0.0.1") @@ -135,6 +190,7 @@ func TestXdsServer_UpdateSnapshot(t *testing.T) { if sa.GetAddress() != "0.0.0.0" { t.Errorf("Expected address '0.0.0.0', got %s", sa.GetAddress()) } + assertRequestsTLSVersion(t, l) } } @@ -158,6 +214,12 @@ func TestXdsServer_UpdateSnapshot_WithHttps(t *testing.T) { t.Fatalf("Snapshot doesn't conform to type *cachev3.Snapshot, got %T", res) } + // The HTTPS listener, its route configuration and the TLS upstream cluster + // are gated together, so enabling TLS adds exactly one of each. + if err := snap.Consistent(); err != nil { + t.Fatalf("Integrity check failed on snapshot: %v", err) + } + listenersMap := snap.GetResources(resourcev3.ListenerType) if len(listenersMap) != 2 { t.Fatalf("Expected 2 listener definitions, got %d", len(listenersMap)) @@ -178,6 +240,94 @@ func TestXdsServer_UpdateSnapshot_WithHttps(t *testing.T) { if ts.GetName() != "envoy.transport_sockets.tls" { t.Errorf("Expected transport socket 'envoy.transport_sockets.tls', got '%s'", ts.GetName()) } + assertRequestsTLSVersion(t, l) + } + + // Both ingresses read the same attribute; the plaintext one relies on Envoy + // omitting it for connections with no TLS handshake. + if raw, exists := listenersMap[IngressHTTPListener]; !exists { + t.Errorf("Listener name '%s' is missing from snapshot listeners", IngressHTTPListener) + } else { + assertRequestsTLSVersion(t, raw.(*listenerv3.Listener)) + } + + clustersMap := snap.GetResources(resourcev3.ClusterType) + if len(clustersMap) != 3 { + t.Fatalf("Expected 3 cluster definitions, got %d", len(clustersMap)) + } + + if raw, exists := clustersMap[DFPClusterName]; !exists { + t.Errorf("Cluster '%s' is missing from clusters", DFPClusterName) + } else { + c := raw.(*clusterv3.Cluster) + // The plaintext cluster must stay plaintext. + if ts := c.GetTransportSocket(); ts != nil { + t.Errorf("Expected no transport socket on '%s', got '%s'", DFPClusterName, ts.GetName()) + } + // Nothing is being waived on the plaintext cluster: with no transport + // socket there is no SNI or SAN requirement to waive. + if dfpClusterConfig(t, c).GetAllowInsecureClusterOptions() { + t.Errorf("Expected AllowInsecureClusterOptions to be false on '%s'", DFPClusterName) + } + } + + if raw, exists := clustersMap[DFPTLSClusterName]; !exists { + t.Errorf("Cluster '%s' is missing from clusters", DFPTLSClusterName) + } else { + c := raw.(*clusterv3.Cluster) + if ts := c.GetTransportSocket(); ts.GetName() != "envoy.transport_sockets.tls" { + t.Errorf("Expected upstream transport socket 'envoy.transport_sockets.tls', got '%s'", ts.GetName()) + } + + // Without this waiver Envoy rejects the entire CDS update, because a + // dynamic forward proxy cluster with a TLS transport socket requires + // auto_san_validation, which this cluster deliberately does not do. + if !dfpClusterConfig(t, c).GetAllowInsecureClusterOptions() { + t.Errorf("Expected AllowInsecureClusterOptions on '%s'; Envoy rejects the CDS update without it", DFPTLSClusterName) + } + + // The SNI has to come from SNIHeader: by the time the request reaches + // the upstream, ext_proc has rewritten :authority to a bare pod IP. + raw, exists := c.GetTypedExtensionProtocolOptions()["envoy.extensions.upstreams.http.v3.HttpProtocolOptions"] + if !exists { + t.Fatalf("Cluster '%s' is missing HttpProtocolOptions", DFPTLSClusterName) + } + protoOpts := &httpv3.HttpProtocolOptions{} + if err := raw.UnmarshalTo(protoOpts); err != nil { + t.Fatalf("Failed to unmarshal HttpProtocolOptions: %v", err) + } + if got := protoOpts.GetUpstreamHttpProtocolOptions().GetOverrideAutoSniHeader(); got != SNIHeader { + t.Errorf("Expected SNI sourced from header '%s', got '%s'", SNIHeader, got) + } + if !protoOpts.GetUpstreamHttpProtocolOptions().GetAutoSni() { + t.Error("Expected AutoSni to be enabled on the TLS cluster") + } + } + + // Each listener routes to its own cluster: only HTTPS ingress is + // re-originated as TLS. + routesMap := snap.GetResources(resourcev3.RouteType) + if len(routesMap) != 2 { + t.Fatalf("Expected 2 route configuration objects, got %d", len(routesMap)) + } + + for routeName, wantCluster := range map[string]string{ + RouteName: DFPClusterName, + RouteNameHTTPS: DFPTLSClusterName, + } { + raw, exists := routesMap[routeName] + if !exists { + t.Errorf("Route name '%s' is missing from snapshot routes configuration", routeName) + continue + } + rc := raw.(*routev3.RouteConfiguration) + if len(rc.GetVirtualHosts()) != 1 || len(rc.GetVirtualHosts()[0].GetRoutes()) != 1 { + t.Errorf("Expected 1 VirtualHost with 1 route in '%s'", routeName) + continue + } + if got := rc.GetVirtualHosts()[0].GetRoutes()[0].GetRoute().GetCluster(); got != wantCluster { + t.Errorf("Expected route '%s' to target cluster '%s', got '%s'", routeName, wantCluster, got) + } } } diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 1f1a907c8..ac7684943 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -71,6 +71,11 @@ const ( actorNftTableName = "ateom_actor" ) +// actorInboundPorts are the TCP ports DNATed from the worker pod IP to the +// actor. 80 carries plaintext ingress; 443 carries ingress the router +// re-originated as TLS. +var actorInboundPorts = []uint16{80, 443} + func main() { pflag.Parse() if *showVersion { @@ -872,8 +877,8 @@ func installActorNftablesRules(podIP net.IP) error { // // * postrouting: masquerade actor egress from 169.254.17.2 behind the worker // pod IP so replies route back to the pod. - // * prerouting: DNAT traffic sent to the worker pod IP on TCP/80 to the - // actor veth IP on TCP/80, preserving existing inbound behavior. + // * prerouting: DNAT traffic sent to the worker pod IP on each actor + // inbound port to the same port on the actor veth IP. // * forward: accept forwarded packets between the actor veth and pod eth0. // // This is not the final egress policy path. The later AgentGateway phase @@ -899,30 +904,32 @@ func installActorNftablesRules(podIP net.IP) error { }) // TODO: Support inbound UDP DNAT for actors that expose UDP protocols such // as QUIC. - // TODO: Replace the hard-coded HTTP port with the actor's configured - // inbound ports, either by adding one rule per port or by matching a set. - preroutingExprs := append(ipDestinationEqual(podIP.String()), tcpDestinationPortEqual(80)...) - preroutingExprs = append(preroutingExprs, - &expr.Immediate{ - Register: 1, - Data: net.ParseIP(actorVethIP).To4(), - }, - &expr.Immediate{ - Register: 2, - Data: binaryutil.BigEndian.PutUint16(80), - }, - &expr.NAT{ - Type: expr.NATTypeDestNAT, - Family: unix.NFPROTO_IPV4, - RegAddrMin: 1, - RegProtoMin: 2, - }, - ) - c.AddRule(&nftables.Rule{ - Table: table, - Chain: prerouting, - Exprs: preroutingExprs, - }) + // TODO: Replace this fixed port pair with the actor's configured inbound + // ports, either by keeping one rule per port or by matching a set. + for _, port := range actorInboundPorts { + preroutingExprs := append(ipDestinationEqual(podIP.String()), tcpDestinationPortEqual(port)...) + preroutingExprs = append(preroutingExprs, + &expr.Immediate{ + Register: 1, + Data: net.ParseIP(actorVethIP).To4(), + }, + &expr.Immediate{ + Register: 2, + Data: binaryutil.BigEndian.PutUint16(port), + }, + &expr.NAT{ + Type: expr.NATTypeDestNAT, + Family: unix.NFPROTO_IPV4, + RegAddrMin: 1, + RegProtoMin: 2, + }, + ) + c.AddRule(&nftables.Rule{ + Table: table, + Chain: prerouting, + Exprs: preroutingExprs, + }) + } postrouting := c.AddChain(&nftables.Chain{ Name: "postrouting", diff --git a/cmd/ateom-microvm/net.go b/cmd/ateom-microvm/net.go index cc047d4f2..8671c0b6f 100644 --- a/cmd/ateom-microvm/net.go +++ b/cmd/ateom-microvm/net.go @@ -21,7 +21,8 @@ package main // 169.254.17.1/30) staying in the pod netns next to the pod's real eth0, and // the peer moved into the interior netns, renamed eth0, and given the stable // actor address 169.254.17.2/30. nftables rules in the pod netns masquerade -// actor egress behind the pod IP and DNAT inbound pod-IP:80 to the actor. +// actor egress behind the pod IP and DNAT the actor's inbound pod-IP ports to +// the actor. // // kata consumes the interior netns exactly like a CNI-provisioned container // netns: its tcfilter network model builds a tap cross-connected to eth0 (the @@ -82,6 +83,11 @@ const ( actorVethSubnet = "169.254.17.0/30" ) +// actorInboundPorts are the TCP ports DNATed from the worker pod IP to the +// actor. 80 carries plaintext ingress; 443 carries ingress the router +// re-originated as TLS. +var actorInboundPorts = []uint16{80, 443} + // Parsed forms of the fixed network constants above, cooked once at package init // (a malformed constant is a programmer error, so these panic). Callers use them // directly instead of re-parsing on every activation. @@ -336,7 +342,8 @@ func enableIPv4Forwarding() error { func installActorNftablesRules(podIP net.IP) error { // Dedicated ateom-owned IPv4 table (cheap cleanup, no CNI chain mutation): // * postrouting: masquerade actor egress (169.254.17.2) behind the pod IP. - // * prerouting: DNAT pod-IP:80/tcp to the actor veth IP. + // * prerouting: DNAT each actor inbound port on the pod IP to the same + // port on the actor veth IP. // * forward: accept forwarded packets between the actor veth and pod eth0. // Mirrors cmd/ateom-gvisor (same compatibility-bridge caveats and TODOs). if err := removeActorNftablesRules(); err != nil { @@ -357,28 +364,34 @@ func installActorNftablesRules(podIP net.IP) error { Hooknum: nftables.ChainHookPrerouting, Priority: nftables.ChainPriorityNATDest, }) - preroutingExprs := append(ipDestinationEqual(podIP.String()), tcpDestinationPortEqual(80)...) - preroutingExprs = append(preroutingExprs, - &expr.Immediate{ - Register: 1, - Data: net.ParseIP(actorVethIP).To4(), - }, - &expr.Immediate{ - Register: 2, - Data: binaryutil.BigEndian.PutUint16(80), - }, - &expr.NAT{ - Type: expr.NATTypeDestNAT, - Family: unix.NFPROTO_IPV4, - RegAddrMin: 1, - RegProtoMin: 2, - }, - ) - c.AddRule(&nftables.Rule{ - Table: table, - Chain: prerouting, - Exprs: preroutingExprs, - }) + // TODO: Support inbound UDP DNAT for actors that expose UDP protocols such + // as QUIC. + // TODO: Replace this fixed port pair with the actor's configured inbound + // ports, either by keeping one rule per port or by matching a set. + for _, port := range actorInboundPorts { + preroutingExprs := append(ipDestinationEqual(podIP.String()), tcpDestinationPortEqual(port)...) + preroutingExprs = append(preroutingExprs, + &expr.Immediate{ + Register: 1, + Data: net.ParseIP(actorVethIP).To4(), + }, + &expr.Immediate{ + Register: 2, + Data: binaryutil.BigEndian.PutUint16(port), + }, + &expr.NAT{ + Type: expr.NATTypeDestNAT, + Family: unix.NFPROTO_IPV4, + RegAddrMin: 1, + RegProtoMin: 2, + }, + ) + c.AddRule(&nftables.Rule{ + Table: table, + Chain: prerouting, + Exprs: preroutingExprs, + }) + } postrouting := c.AddChain(&nftables.Chain{ Name: "postrouting", diff --git a/demos/counter/counter.go b/demos/counter/counter.go index 0b18f0ff3..34c354248 100644 --- a/demos/counter/counter.go +++ b/demos/counter/counter.go @@ -18,12 +18,18 @@ package main import ( "context" + "crypto/ecdsa" + "crypto/elliptic" "crypto/rand" "crypto/sha256" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" "encoding/base64" "fmt" "io" "log/slog" + "math/big" "net" "net/http" "os" @@ -114,6 +120,8 @@ func main() { } }() + go serveTLS(ctx, defaultMux) + // Write some random data to a file in the root filesystem, to test // filesystem checkpoint/restore. if err := writeRandomFile(); err != nil { @@ -136,6 +144,64 @@ func main() { } } +// serveTLS mirrors the plaintext handlers on port 443 behind a self-signed +// certificate, so the actor can be reached over the router's HTTPS ingress +// (which re-originates TLS to 443 rather than forwarding cleartext to 80). +// +// The certificate is minted here rather than provisioned by the platform: +// actors have no platform-issued identity, and the router deliberately does not +// verify what the actor presents. Each handshake logs the SNI the router +// offered, which is what makes the upstream TLS wiring observable end to end. +func serveTLS(ctx context.Context, mux *http.ServeMux) { + cert, err := selfSignedCert() + if err != nil { + slog.ErrorContext(ctx, "Error generating TLS certificate", slog.Any("err", err)) + return + } + + srv := &http.Server{ + Addr: ":443", + Handler: mux, + TLSConfig: &tls.Config{ + GetCertificate: func(hello *tls.ClientHelloInfo) (*tls.Certificate, error) { + slog.InfoContext(ctx, "TLS handshake", slog.String("sni", hello.ServerName)) + return cert, nil + }, + }, + } + + slog.InfoContext(ctx, "Starting counter TLS server on port 443") + // Both arguments are empty because the certificate is served from + // TLSConfig.GetCertificate rather than read from disk. + if err := srv.ListenAndServeTLS("", ""); err != nil { + slog.ErrorContext(ctx, "Error starting TLS server", slog.Any("err", err)) + } +} + +// selfSignedCert mints a throwaway certificate covering any actor mesh DNS +// name. Nothing verifies it, so the SANs only matter for legibility when +// inspecting the handshake. +func selfSignedCert() (*tls.Certificate, error) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, fmt.Errorf("while generating key: %w", err) + } + + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "counter"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(365 * 24 * time.Hour), + DNSNames: []string{"*.actors.resources.substrate.ate.dev"}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + return nil, fmt.Errorf("while creating certificate: %w", err) + } + + return &tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}, nil +} + func writeRandomFile() error { rf, err := os.Create("/random-content-file") if err != nil { From 3e7e84fa7f5e86ff4a0486ec88fa08248bd2dc74 Mon Sep 17 00:00:00 2001 From: Yufan Su Date: Mon, 27 Jul 2026 16:40:51 -0700 Subject: [PATCH 2/3] minor comment fix --- cmd/atenet/internal/router/extproc.go | 5 +++-- cmd/atenet/internal/router/extproc_in.go | 7 ++----- cmd/atenet/internal/router/xds.go | 4 +--- 3 files changed, 6 insertions(+), 10 deletions(-) diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 2d16d3480..e2dd9d2f8 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -176,8 +176,9 @@ func (s *ExtProcServer) handleRequestHeaders( "actor %q routing failed", actorName) } - // Actors serve plaintext on 80. A request that arrived over TLS is - // re-originated as TLS to 443, so the actor sees the scheme the client used. + // Actors serve plaintext on 80. A request whose downstream connection was + // TLS is re-originated as TLS to 443, so the actor is reached over the same + // transport the client used. // TODO(bowei) -- handle actor ports beyond this fixed 80/443 pair. targetAddr := net.JoinHostPort(workerIP, actorPortForIngress(tlsIngress)) diff --git a/cmd/atenet/internal/router/extproc_in.go b/cmd/atenet/internal/router/extproc_in.go index d61389d16..d89622c20 100644 --- a/cmd/atenet/internal/router/extproc_in.go +++ b/cmd/atenet/internal/router/extproc_in.go @@ -77,11 +77,8 @@ func newRequestMetadata(headers []*corev3.HeaderValue) *requestMetadata { // negotiated, from the attributes Envoy attaches to the ProcessingRequest. It is // "" for a plaintext connection. // -// The connection's own TLS state decides whether ingress was encrypted, rather -// than request.scheme, because Envoy derives :scheme from x-forwarded-proto and -// preserves a caller-supplied one unless configured as an edge proxy — reading -// the scheme would let a caller pick which actor port this routes to. Nothing on -// the wire can influence connection state. +// Whether ingress was encrypted is read from connection state so that nothing +// the caller sends can influence which actor port the request is routed to. // // An absent attribute also yields "", which is read as plaintext — the safe // direction, since it cannot turn a plaintext request into an upstream TLS one. diff --git a/cmd/atenet/internal/router/xds.go b/cmd/atenet/internal/router/xds.go index 78f2af19f..9dde76faf 100644 --- a/cmd/atenet/internal/router/xds.go +++ b/cmd/atenet/internal/router/xds.go @@ -467,9 +467,7 @@ func (x *XdsServer) buildHcm(statPrefix, routeName string) *anypb.Any { AllowAllRouting: &wrapperspb.BoolValue{Value: true}, }, // The downstream connection's TLS version tells ext_proc which port to - // route the actor on. Unlike the request scheme it is connection state, - // so it is not derivable from anything the caller sends. See - // requestTLSVersion. + // route the actor on. See requestTLSVersion. RequestAttributes: []string{tlsVersionAttribute}, // Explicitly configure the message timeout to avoid the 200ms default MessageTimeout: durationpb.New(5 * time.Second), From 8724c33d3778f6e5a09bb6524881005b2814b093 Mon Sep 17 00:00:00 2001 From: Yufan Su Date: Mon, 27 Jul 2026 17:04:35 -0700 Subject: [PATCH 3/3] fix typo --- cmd/atenet/internal/router/xds_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/atenet/internal/router/xds_test.go b/cmd/atenet/internal/router/xds_test.go index 3bd3d40c2..bb3c6351c 100644 --- a/cmd/atenet/internal/router/xds_test.go +++ b/cmd/atenet/internal/router/xds_test.go @@ -59,7 +59,7 @@ func extProcConfig(t *testing.T, l *listenerv3.Listener) *extprocv3filter.Extern } // assertRequestsTLSVersion checks that ext_proc is sent the downstream TLS -// version. Envoy silently drops attribute names it does not recognise, and an +// version. Envoy silently drops attribute names it does not recognize, and an // undelivered attribute reads as plaintext — which strands the HTTPS ingress on // the actor's port 80 with no error anywhere in the control plane. func assertRequestsTLSVersion(t *testing.T, l *listenerv3.Listener) {