diff --git a/internal/atenet/headers.go b/internal/atenet/headers.go new file mode 100644 index 0000000000..ee1d640876 --- /dev/null +++ b/internal/atenet/headers.go @@ -0,0 +1,40 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package atenet defines the shared contract for Substrate actor networking. +package atenet + +import ( + "fmt" + "strings" + + "github.com/agent-substrate/substrate/internal/resources" +) + +const ( + // TargetActorHeader identifies the actor selected for ingress routing as + // "/". HTTP field names are case-insensitive; this uses its + // HTTP/2 wire form so dataplane configuration and metadata are native. + TargetActorHeader = "ate-target-actor" +) + +// ParseTargetActor parses and validates a TargetActorHeader value. +func ParseTargetActor(value string) (resources.ActorRef, error) { + atespace, actorName, ok := strings.Cut(value, "/") + if !ok || strings.Contains(actorName, "/") || + !resources.IsValidResourceName(atespace) || !resources.IsValidResourceName(actorName) { + return resources.ActorRef{}, fmt.Errorf("invalid actor reference %q", value) + } + return resources.ActorRef{Atespace: atespace, Name: actorName}, nil +} diff --git a/internal/atenet/headers_test.go b/internal/atenet/headers_test.go new file mode 100644 index 0000000000..b640ac8d1d --- /dev/null +++ b/internal/atenet/headers_test.go @@ -0,0 +1,47 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package atenet + +import "testing" + +func TestParseTargetActor(t *testing.T) { + tests := []struct { + name string + value string + wantAtespace string + wantActorName string + wantErr bool + }{ + {name: "valid", value: "team-a/actor-1", wantAtespace: "team-a", wantActorName: "actor-1"}, + {name: "missing separator", value: "team-a", wantErr: true}, + {name: "extra separator", value: "team-a/actor-1/extra", wantErr: true}, + {name: "empty atespace", value: "/actor-1", wantErr: true}, + {name: "empty actor", value: "team-a/", wantErr: true}, + {name: "invalid atespace", value: "TEAM-A/actor-1", wantErr: true}, + {name: "invalid actor", value: "team-a/ACTOR-1", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := ParseTargetActor(tt.value) + if (err != nil) != tt.wantErr { + t.Fatalf("ParseTargetActor(%q) error = %v, wantErr %v", tt.value, err, tt.wantErr) + } + if got.Atespace != tt.wantAtespace || got.Name != tt.wantActorName { + t.Errorf("ParseTargetActor(%q) = %q, want %q/%q", tt.value, got.String(), tt.wantAtespace, tt.wantActorName) + } + }) + } +} diff --git a/internal/e2e/router_client.go b/internal/e2e/router_client.go index ba29765bec..987692557e 100644 --- a/internal/e2e/router_client.go +++ b/internal/e2e/router_client.go @@ -29,6 +29,7 @@ import ( "time" "github.com/agent-substrate/substrate/internal/ateclient" + "github.com/agent-substrate/substrate/internal/atenet" "github.com/agent-substrate/substrate/internal/portforward" "github.com/agent-substrate/substrate/internal/resources" "k8s.io/client-go/kubernetes" @@ -106,8 +107,7 @@ func (c *RouterClient) BaseURL() string { return c.baseURL } -// Get issues GET path to actor through the router, setting the actor's DNS Host -// so the router routes (and resumes) it. The caller must close the body. +// Get issues GET path to actor through the router. The caller must close the body. func (c *RouterClient) Get(ctx context.Context, actorRef resources.ActorRef, path string) (*http.Response, error) { return c.request(ctx, http.MethodGet, actorRef, path, nil) } @@ -126,8 +126,7 @@ func (c *RouterClient) request(ctx context.Context, method string, actorRef reso if method == http.MethodPost { req.Header.Set("Content-Type", "application/json") } - // The router routes on the Host/:authority, not a header. - req.Host = resources.ActorDNSName(actorRef) + req.Header.Set(atenet.TargetActorHeader, actorRef.String()) return c.http.Do(req) } @@ -149,11 +148,14 @@ func (c *RouterClient) Connect(ctx context.Context, actorRef resources.ActorRef, return nil, fmt.Errorf("connecting to router's CONNECT listener: %w", err) } - destination := net.JoinHostPort(resources.ActorDNSName(actorRef), strconv.Itoa(port)) + destination := net.JoinHostPort(actorRef.Name, strconv.Itoa(port)) req := &http.Request{ Method: http.MethodConnect, URL: &url.URL{Host: destination}, Host: destination, + Header: http.Header{ + atenet.TargetActorHeader: []string{actorRef.String()}, + }, } if err := req.Write(rawConn); err != nil { _ = rawConn.Close() diff --git a/internal/e2e/suites/demo/demo_test.go b/internal/e2e/suites/demo/demo_test.go index 9e112814df..ca2b69b50e 100644 --- a/internal/e2e/suites/demo/demo_test.go +++ b/internal/e2e/suites/demo/demo_test.go @@ -27,6 +27,7 @@ import ( "time" "github.com/agent-substrate/substrate/internal/ateclient" + "github.com/agent-substrate/substrate/internal/atenet" "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" @@ -1330,7 +1331,7 @@ func callActorPathOnce(t *testing.T, actorRef resources.ActorRef, method, path s if err != nil { return "", fmt.Errorf("failed to create request: %w", err) } - reqHttp.Host = resources.ActorDNSName(actorRef) + reqHttp.Header.Set(atenet.TargetActorHeader, actorRef.String()) httpClient := &http.Client{Timeout: 15 * time.Second} resp, err := httpClient.Do(reqHttp) diff --git a/internal/e2e/suites/networking/arbitraryport_test.go b/internal/e2e/suites/networking/arbitraryport_test.go index 9217b8fb51..679f5bc80a 100644 --- a/internal/e2e/suites/networking/arbitraryport_test.go +++ b/internal/e2e/suites/networking/arbitraryport_test.go @@ -25,6 +25,7 @@ import ( "testing" "time" + "github.com/agent-substrate/substrate/internal/atenet" "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" ) @@ -106,7 +107,8 @@ func TestActorArbitraryPortAccess(t *testing.T) { defer conn.Close() conn.SetDeadline(time.Now().Add(10 * time.Second)) - if _, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: " + resources.ActorDNSName(actorRef) + "\r\nConnection: close\r\n\r\n")); err != nil { + if _, err := fmt.Fprintf(conn, "GET / HTTP/1.1\r\nHost: %s\r\n%s: %s\r\nConnection: close\r\n\r\n", + actorRef.Name, atenet.TargetActorHeader, actorRef.String()); err != nil { t.Fatalf("writing tunneled request: %v", err) } resp, err := http.ReadResponse(bufio.NewReader(conn), nil) @@ -140,7 +142,7 @@ func waitForTunneledRouteReady(t *testing.T, ctx context.Context, router *e2e.Ro for { conn, err := router.Connect(ctx, actorRef, port) if err == nil { - resp, body, requestErr := requestTunneled(conn, resources.ActorDNSName(actorRef)) + resp, body, requestErr := requestTunneled(conn, actorRef.Name) _ = conn.Close() if requestErr == nil && resp.StatusCode == http.StatusOK { return body diff --git a/internal/e2e/suites/networking/grpcingress_test.go b/internal/e2e/suites/networking/grpcingress_test.go index 210cf0484f..0d3202920d 100644 --- a/internal/e2e/suites/networking/grpcingress_test.go +++ b/internal/e2e/suites/networking/grpcingress_test.go @@ -26,8 +26,10 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" "github.com/agent-substrate/substrate/internal/ateclient" + "github.com/agent-substrate/substrate/internal/atenet" "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/portforward" "github.com/agent-substrate/substrate/internal/proto/grpcechopb" @@ -72,7 +74,7 @@ func TestIngressProtocolDowngrade(t *testing.T) { if err != nil { return nil, err } - req.Host = resources.ActorDNSName(actorRef) + req.Header.Set(atenet.TargetActorHeader, actorRef.String()) if contentType != "" { req.Header.Set("Content-Type", contentType) } @@ -150,17 +152,18 @@ func TestIngressGRPC(t *testing.T) { fixture := deployGRPCEchoTemplate(t, ctx, env["BUCKET_NAME"]) actorName, _ := createAndResumeSubstrateActor(t, ctx, "grpcingress", fixture) actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} + ctx = metadata.AppendToOutgoingContext(ctx, + atenet.TargetActorHeader, actorRef.String(), + ) - // Cleartext h2c to the router's HTTP port, with the Actor's DNS name as the - // :authority — the same routing key every other ingress test in this suite - // uses, just carried by a gRPC client instead of an HTTP one. The h2 ALPN - // offer is about the *TLS* listener; nothing here needs it. + // Cleartext h2c to the router's HTTP port. Explicit metadata identifies the + // Actor; the conventional actor authority remains application metadata. The + // h2 ALPN offer is about the *TLS* listener; nothing here needs it. conn, err := grpc.NewClient(routerAddress(t, ctx), grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithAuthority(resources.ActorDNSName(actorRef)), ) if err != nil { - t.Fatalf("creating the gRPC client for %s: %v", resources.ActorDNSName(actorRef), err) + t.Fatalf("creating the gRPC client for %s: %v", actorRef, err) } defer conn.Close() client := grpcechopb.NewEchoClient(conn) diff --git a/internal/e2e/suites/networking/websocketingress_test.go b/internal/e2e/suites/networking/websocketingress_test.go index d0afb3b54a..085f03146e 100644 --- a/internal/e2e/suites/networking/websocketingress_test.go +++ b/internal/e2e/suites/networking/websocketingress_test.go @@ -22,6 +22,7 @@ import ( "testing" "time" + "github.com/agent-substrate/substrate/internal/atenet" "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" "github.com/gorilla/websocket" @@ -67,7 +68,7 @@ func TestWebsocketIngressPing(t *testing.T) { actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} header := http.Header{} - header.Set("Host", resources.ActorDNSName(actorRef)) + header.Set(atenet.TargetActorHeader, actorRef.String()) var c *websocket.Conn