atenet: map HTTPS requests to actor port 443 with TLS re-origination

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,22 @@ 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 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))
 
 	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{

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,28 @@ 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.
+//
+// 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.
+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
 // "<actor_name>.<atespace>.actors.resources.substrate.ate.dev" (optionally with a

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{

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