From 7a02250ac0f54a89c8cfda542ebd268873ad6688 Mon Sep 17 00:00:00 2001 From: Lior Lieberman Date: Mon, 27 Jul 2026 16:23:31 -0700 Subject: [PATCH 1/3] egress: add an Envoy egress gateway. The gateway terminates actor's CONNECT request. It requires downstream mTLS, so only a worker's atunnel can reach it. The gateway consists of an Envoy and an `atenet router --standalone` ext_proc sidecar. --- .../internal/controlapi/functional_test.go | 10 + .../internal/controlapi/workflow_resume.go | 3 + cmd/atelet/main.go | 39 ++ cmd/atelet/main_test.go | 4 + cmd/atenet/internal/router/extproc.go | 19 +- cmd/atenet/internal/router/extproc_egress.go | 176 +++++++++ .../internal/router/extproc_egress_test.go | 127 +++++++ cmd/atenet/internal/router/extproc_in.go | 6 + hack/install-ate.sh | 13 +- internal/proto/ateletpb/atelet.pb.go | 98 +++-- internal/proto/ateletpb/atelet.proto | 11 + manifests/ate-install/atelet.yaml | 7 + manifests/ate-install/ateway-egress.yaml | 338 ++++++++++++++++++ .../kind/atelet/kustomization.yaml | 5 + manifests/ate-install/kind/kustomization.yaml | 1 + .../token-client/kustomization.yaml | 1 + 16 files changed, 828 insertions(+), 30 deletions(-) create mode 100644 cmd/atenet/internal/router/extproc_egress.go create mode 100644 cmd/atenet/internal/router/extproc_egress_test.go create mode 100644 manifests/ate-install/ateway-egress.yaml diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 646da16a0..13c273241 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -1653,6 +1653,13 @@ func TestResumeActor(t *testing.T) { if !tc.fakeAtelet.RestoreCalled { t.Errorf("expected Restore to be called") } + restoreReq := tc.fakeAtelet.lastRestoreRequest() + if restoreReq == nil || restoreReq.GetActorVersion() < 1 { + t.Fatalf("Restore actor_version = %v, want a positive version", restoreReq) + } + if restoreReq.EgressGatewayAddress != nil { + t.Fatalf("Restore egress_gateway_address = %q, want absent until the Actor egress API is available", restoreReq.GetEgressGatewayAddress()) + } getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, @@ -1660,6 +1667,9 @@ func TestResumeActor(t *testing.T) { if err != nil { t.Fatalf("GetActor failed: %v", err) } + if restoreReq.GetActorVersion() >= getResp.GetMetadata().GetVersion() { + t.Errorf("Restore actor_version = %d, final Actor version = %d; want the assignment version to precede finalization", restoreReq.GetActorVersion(), getResp.GetMetadata().GetVersion()) + } want := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: name, Atespace: testAtespace}, ActorTemplateNamespace: ns, diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index f24e91a49..897ecd34d 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -526,6 +526,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, ActorName: state.Actor.GetMetadata().GetName(), ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(), ActorTemplateName: state.Actor.GetActorTemplateName(), + ActorVersion: state.Actor.GetMetadata().GetVersion(), Spec: workloadSpec, ActorUid: state.Actor.GetMetadata().Uid, } @@ -563,6 +564,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, ActorName: state.Actor.GetMetadata().GetName(), ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(), ActorTemplateName: state.Actor.GetActorTemplateName(), + ActorVersion: state.Actor.GetMetadata().GetVersion(), Spec: workloadSpec, Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL, Config: &ateletpb.RestoreRequest_ExternalConfig{ @@ -594,6 +596,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, ActorName: state.Actor.GetMetadata().GetName(), ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(), ActorTemplateName: state.Actor.GetActorTemplateName(), + ActorVersion: state.Actor.GetMetadata().GetVersion(), SandboxAssets: sandboxAssets, Spec: workloadSpec, ActorUid: state.Actor.GetMetadata().Uid, diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index b3f0241c9..d7780db40 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -76,10 +76,39 @@ var ( localhostRegistryReplacement = pflag.String("localhost-registry-replacement", "", "The replacement registry endpoint for localhost and/or loopback IP addresses, useful for local development. for example kind-registry:5000") imageCacheDir = pflag.String("image-cache-dir", ateompath.ImageCacheDir, "Directory for the node-local OCI image layer cache. Must be on the volume shared with the ateom pods (the cached layers are their overlay lowerdirs), and on a disk sized for both capacity and IOPS: unpack throughput is gated by the volume's IOPS.") + // SEE(lior): both sides kept — main added --log-level here while this branch + // added --egress-gateway-address. Independent flags, no interaction. + // + // egressGatewayAddress turns on pluggable actor egress cluster-wide (POC). + // When set, actors whose Run/Restore request does not already carry an egress + // gateway address have this address injected, causing ateom to redirect actor + // TCP egress through atunnel to the egress gateway. Empty keeps egress off. + egressGatewayAddress = pflag.String("egress-gateway-address", "", "Address (host:port) of the egress gateway. When set, actor TCP egress is transparently tunneled through atunnel to this gateway. Empty disables egress.") + showVersion = pflag.Bool("version", false, "Print version and exit.") logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.") ) +// egressAddrRequest is satisfied by the atelet Run/Restore request protos, both +// of which carry an optional egress gateway address. +type egressAddrRequest interface { + GetEgressGatewayAddress() string +} + +// effectiveEgressGatewayAddress prefers the per-request address (reserved for a +// future per-actor egress API) and otherwise falls back to the cluster-wide +// atelet flag. +func effectiveEgressGatewayAddress(req egressAddrRequest) *string { + addr := req.GetEgressGatewayAddress() + if addr == "" { + addr = *egressGatewayAddress + } + if addr == "" { + return nil + } + return &addr +} + func main() { pflag.Parse() if *showVersion { @@ -288,6 +317,8 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp * ActorName: actorRef.Name, ActorTemplateNamespace: req.GetActorTemplateNamespace(), ActorTemplateName: req.GetActorTemplateName(), + ActorVersion: req.GetActorVersion(), + EgressGatewayAddress: effectiveEgressGatewayAddress(req), RunscPath: runscPathFor(assetPaths), RuntimeAssetPaths: assetPaths, Spec: buildAteomWorkloadSpec(req.GetSpec()), @@ -662,6 +693,8 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) ActorName: actorRef.Name, ActorTemplateNamespace: req.GetActorTemplateNamespace(), ActorTemplateName: req.GetActorTemplateName(), + ActorVersion: req.GetActorVersion(), + EgressGatewayAddress: effectiveEgressGatewayAddress(req), RunscPath: runscPathFor(assetPaths), RuntimeAssetPaths: assetPaths, Spec: buildAteomWorkloadSpec(req.GetSpec()), @@ -990,6 +1023,9 @@ func (d *AteomDialer) DialAteomPod(ctx context.Context, podUID string) (*grpc.Cl // internal/resources so other components can apply them at their boundaries. func validateRunRequest(req *ateletpb.RunRequest) error { var errs field.ErrorList + if req.GetActorVersion() < 1 { + errs = append(errs, field.Invalid(field.NewPath("actor_version"), req.GetActorVersion(), "must be positive")) + } errs = append(errs, resources.ValidateResourceName(req.GetAtespace(), field.NewPath("atespace"))...) errs = append(errs, resources.ValidateResourceName(req.GetActorName(), field.NewPath("actor_name"))...) errs = append(errs, resources.ValidateResourceName(req.GetActorUid(), field.NewPath("actor_uid"))...) @@ -1067,6 +1103,9 @@ func validateCheckpointRequest(req *ateletpb.CheckpointRequest) error { func validateRestoreRequest(req *ateletpb.RestoreRequest) error { var errs field.ErrorList + if req.GetActorVersion() < 1 { + errs = append(errs, field.Invalid(field.NewPath("actor_version"), req.GetActorVersion(), "must be positive")) + } errs = append(errs, resources.ValidateResourceName(req.GetAtespace(), field.NewPath("atespace"))...) errs = append(errs, resources.ValidateResourceName(req.GetActorName(), field.NewPath("actor_name"))...) errs = append(errs, resources.ValidateResourceName(req.GetActorUid(), field.NewPath("actor_uid"))...) diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index 50fe8a82d..37d4fed0a 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -175,6 +175,7 @@ func validRunRequest() *ateletpb.RunRequest { ActorTemplateName: "counter", TargetAteomUid: "422938ba-8860-4983-a25d-d6bcb0a69d4e", ActorUid: "123e4567-e89b-12d3-a456-426614174000", + ActorVersion: 7, Spec: &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "worker"}}}, } } @@ -206,6 +207,7 @@ func validRestoreRequest() *ateletpb.RestoreRequest { ActorTemplateName: "counter", TargetAteomUid: "422938ba-8860-4983-a25d-d6bcb0a69d4e", ActorUid: "123e4567-e89b-12d3-a456-426614174000", + ActorVersion: 7, Spec: &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "worker"}}}, Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL, Config: &ateletpb.RestoreRequest_ExternalConfig{ @@ -230,6 +232,7 @@ func TestValidateRunRequest(t *testing.T) { {"invalid actor uid", func(r *ateletpb.RunRequest) { r.ActorUid = "../escape" }, true}, {"invalid actor template namespace", func(r *ateletpb.RunRequest) { r.ActorTemplateNamespace = "Not_Valid" }, true}, {"invalid actor template name", func(r *ateletpb.RunRequest) { r.ActorTemplateName = "Not_Valid" }, true}, + {"invalid actor version", func(r *ateletpb.RunRequest) { r.ActorVersion = 0 }, true}, {"invalid container name", func(r *ateletpb.RunRequest) { r.Spec.Containers = []*ateletpb.Container{{Name: "../escape"}} }, true}, @@ -316,6 +319,7 @@ func TestValidateRestoreRequest(t *testing.T) { {"invalid actor uid", makeReq(func(r *ateletpb.RestoreRequest) { r.ActorUid = "../escape" }), true}, {"invalid actor template namespace", makeReq(func(r *ateletpb.RestoreRequest) { r.ActorTemplateNamespace = "Not_Valid" }), true}, {"invalid actor template name", makeReq(func(r *ateletpb.RestoreRequest) { r.ActorTemplateName = "Not_Valid" }), true}, + {"invalid actor version", makeReq(func(r *ateletpb.RestoreRequest) { r.ActorVersion = 0 }), true}, {"invalid container name", makeReq(func(r *ateletpb.RestoreRequest) { r.Spec.Containers = []*ateletpb.Container{{Name: "../escape"}} }), true}, diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index 795afd583..cba79c9af 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -96,7 +96,24 @@ 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, resumeOutcome, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders) + // One ext_proc server handles both directions: actor egress + // CONNECT requests and ingress requests. Which one is decided by + // the accepting listener, not by anything in the request itself + // (see isEgressRequest). + // + // SEE(lior): main grew a ResumeOutcome return on + // handleRequestHeaders while this branch was out. Rather than + // splitting the dispatch into two separately-typed call sites, the + // egress handler was widened to the same signature and returns + // ResumeOutcomeNone — egress requires an already-RUNNING actor and + // never resumes one, so "none" is accurate, and it keeps the + // route-duration metric's resume label a closed set (no new empty + // value) across both directions. + handle := s.handleRequestHeaders + if isEgressRequest(req) { + handle = s.handleEgressRequestHeaders + } + hResponse, rqm, target, tmplNs, tmplName, resumeOutcome, err := handle(stream.Context(), reqType.RequestHeaders) elapsed := time.Since(start) outcomeStr := classifyOutcome(err) resumeStr := string(resumeOutcome) diff --git a/cmd/atenet/internal/router/extproc_egress.go b/cmd/atenet/internal/router/extproc_egress.go new file mode 100644 index 000000000..69f3411be --- /dev/null +++ b/cmd/atenet/internal/router/extproc_egress.go @@ -0,0 +1,176 @@ +// 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 router + +import ( + "context" + "log/slog" + "strconv" + "strings" + + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/internal/atunnel" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +const ( + // EgressListenerName is the Envoy listener that terminates actor egress + // CONNECTs. It must stay in sync with the listener name in + // manifests/ate-install/ateway-egress.yaml. + EgressListenerName = "egress" + // ListenerNameAttribute is the CEL attribute carrying the name of the + // listener that accepted the request. The egress Envoy asks for it via + // request_attributes on its ext_proc filter. + ListenerNameAttribute = "xds.listener_name" +) + +// isEgressRequest reports whether an ext_proc RequestHeaders callback arrived on +// the egress gateway's listener rather than on an ingress listener. This lets +// one ext_proc server handle both directions off the same stream. +// +// Dispatch is by listener, not by :method, because the two handlers apply +// opposite trust models: on egress the X-Ate-* identity headers are asserted by +// atunnel over mTLS, while on ingress the same headers are unauthenticated +// client input. Keying on :method would let any external client sending CONNECT +// select the egress handler and use its denial messages as an actor-existence +// and status oracle. Envoy asserts the listener name; the request cannot +// influence it. +// +// An unrecognized or absent attribute means ingress, the fail-safe direction: an +// egress request misrouted to the ingress handler fails to parse as an actor DNS +// name and 404s, whereas the reverse leaks control-plane state. +func isEgressRequest(req *extprocv3.ProcessingRequest) bool { + return listenerName(req) == EgressListenerName +} + +// listenerName returns the xds.listener_name attribute Envoy attached to the +// request, or "" when the listener did not request the attribute. The +// attributes map is keyed by the ext_proc filter's name within the HCM chain, +// which we do not want to hardcode here, so scan every entry. +func listenerName(req *extprocv3.ProcessingRequest) string { + for _, attrs := range req.GetAttributes() { + if v, ok := attrs.GetFields()[ListenerNameAttribute]; ok { + return v.GetStringValue() + } + } + return "" +} + +// handleEgressRequestHeaders authenticates the actor identity that atunnel +// asserts on an egress CONNECT, before the gateway tunnels it out. This turns the +// worker-asserted X-Ate-* headers into a control-plane-verified identity. +// +// Authorization by destination and credential/token injection are deliberately +// a TODO once we have SessionIdentity RPC service figured out. +// +// The signature mirrors handleRequestHeaders so Process can dispatch to either +// with a single branch. The (target, tmplNs, tmplName) results are unused for +// egress and returned empty. +// +// SEE(lior): the trailing ResumeOutcome exists only to match +// handleRequestHeaders after main added it. Egress never resumes an actor — it +// requires one already RUNNING — so every path returns ResumeOutcomeNone. +func (s *ExtProcServer) handleEgressRequestHeaders( + ctx context.Context, + reqHeaders *extprocv3.HttpHeaders, +) (*extprocv3.HeadersResponse, *requestMetadata, string, string, string, ResumeOutcome, error) { + metadata := newRequestMetadata(reqHeaders.Headers.GetHeaders()) + + // Dispatch is by listener, so reaching here means the egress listener + // accepted the request. That listener only routes CONNECT (its sole route + // is a connect_matcher), so anything else is a config drift rather than a + // client the gateway should tunnel for. + if !strings.EqualFold(metadata.method, "CONNECT") { + return nil, metadata, "", "", "", ResumeOutcomeNone, newReqError(envoy_type.StatusCode_MethodNotAllowed, + "egress denied: expected CONNECT, got %q", metadata.method) + } + + atespace := metadata.headers[strings.ToLower(atunnel.ActorAtespaceHeader)] + actorName := metadata.headers[strings.ToLower(atunnel.ActorNameHeader)] + assertedVersion := metadata.headers[strings.ToLower(atunnel.ActorVersionHeader)] + // For a CONNECT the :authority is the actor's original destination (IP:port). + destination := metadata.host + + if atespace == "" || actorName == "" { + return nil, metadata, "", "", "", ResumeOutcomeNone, newReqError(envoy_type.StatusCode_Forbidden, + "egress denied: missing actor identity headers") + } + if !resources.IsValidResourceName(atespace) || !resources.IsValidResourceName(actorName) { + return nil, metadata, "", "", "", ResumeOutcomeNone, newReqError(envoy_type.StatusCode_Forbidden, + "egress denied: invalid actor identity %q/%q", atespace, actorName) + } + + // Authenticate the worker-asserted identity against the control plane. A + // claimed-but-nonexistent actor (a spoofed identity) surfaces as NotFound. + actor, err := s.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: atespace, Name: actorName}, + }) + if err != nil { + return nil, metadata, "", "", "", ResumeOutcomeNone, mapEgressIdentityError(atespace, actorName, err) + } + + // The actor performing egress must actually be running. + if actor.GetStatus() != ateapipb.Actor_STATUS_RUNNING { + return nil, metadata, "", "", "", ResumeOutcomeNone, newReqError(envoy_type.StatusCode_Forbidden, + "egress denied: actor %q/%q is %s, not running", atespace, actorName, actor.GetStatus()) + } + + // X-Ate-Actor-Version is the Actor version the worker observed when it was + // assigned, and atunnel documents it as a lower bound on trustworthy actor + // metadata. If our authoritative view is older than what the worker asserts, + // we cannot yet vouch for the identity, so reject rather than allow blindly. + if assertedVersion != "" { + if want, perr := strconv.ParseInt(assertedVersion, 10, 64); perr == nil && actor.GetMetadata().GetVersion() < want { + return nil, metadata, "", "", "", ResumeOutcomeNone, newReqError(envoy_type.StatusCode_Forbidden, + "egress denied: actor %q/%q metadata stale (known v%d < asserted v%d)", + atespace, actorName, actor.GetMetadata().GetVersion(), want) + } + } + + slog.InfoContext(ctx, "egress identity authenticated", + slog.String("atespace", atespace), + slog.String("actor", actorName), + slog.String("destination", destination), + slog.String("status", actor.GetStatus().String())) + + // Identity is authenticated; let the CONNECT proceed unchanged. Milestone 2 + // would additionally authorize `destination` and inject upstream credentials + // here by returning a HeaderMutation. + return &extprocv3.HeadersResponse{ + Response: &extprocv3.CommonResponse{}, + }, metadata, "", "", "", ResumeOutcomeNone, nil +} + +// mapEgressIdentityError converts a GetActor failure into a client-facing +// ext_proc denial. An unknown actor is treated as a forbidden (spoofed) +// identity; transient control-plane failures fail closed with 503. +func mapEgressIdentityError(atespace, actorName string, err error) error { + switch status.Code(err) { + case codes.NotFound: + return newReqError(envoy_type.StatusCode_Forbidden, + "egress denied: unknown actor %q/%q", atespace, actorName) + case codes.Unavailable, codes.DeadlineExceeded: + return newReqError(envoy_type.StatusCode_ServiceUnavailable, + "egress identity check unavailable for %q/%q: %v", atespace, actorName, err) + default: + return newReqError(envoy_type.StatusCode_Forbidden, + "egress denied for %q/%q: %v", atespace, actorName, err) + } +} diff --git a/cmd/atenet/internal/router/extproc_egress_test.go b/cmd/atenet/internal/router/extproc_egress_test.go new file mode 100644 index 000000000..f9d449f28 --- /dev/null +++ b/cmd/atenet/internal/router/extproc_egress_test.go @@ -0,0 +1,127 @@ +// 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 router + +import ( + "testing" + + corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + "google.golang.org/protobuf/types/known/structpb" +) + +// connectRequest builds a RequestHeaders ProcessingRequest for a CONNECT +// carrying worker-asserted actor identity, optionally attributed to a listener. +// filterKey is the ext_proc filter name Envoy keys the attributes map by. +func connectRequest(filterKey, listener string) *extprocv3.ProcessingRequest { + req := &extprocv3.ProcessingRequest{ + Request: &extprocv3.ProcessingRequest_RequestHeaders{ + RequestHeaders: &extprocv3.HttpHeaders{ + Headers: &corev3.HeaderMap{ + Headers: []*corev3.HeaderValue{ + {Key: ":method", RawValue: []byte("CONNECT")}, + {Key: ":authority", RawValue: []byte("10.0.0.9:443")}, + {Key: "x-ate-atespace", RawValue: []byte("default")}, + {Key: "x-ate-actor", RawValue: []byte("my-actor")}, + }, + }, + }, + }, + } + if listener != "" { + req.Attributes = map[string]*structpb.Struct{ + filterKey: { + Fields: map[string]*structpb.Value{ + ListenerNameAttribute: structpb.NewStringValue(listener), + }, + }, + } + } + return req +} + +func TestIsEgressRequest(t *testing.T) { + tests := []struct { + name string + filterKey string + listener string + want bool + }{ + { + name: "egress listener", + filterKey: "envoy.filters.http.ext_proc", + listener: EgressListenerName, + want: true, + }, + { + name: "egress listener under a renamed filter", + filterKey: "some.custom.ext_proc.name", + listener: EgressListenerName, + want: true, + }, + { + // The pre-listener-dispatch hole: an external client sending + // CONNECT to the ingress gateway must not reach the egress handler, + // whose denials would otherwise report whether an arbitrary actor + // exists and is running. + name: "CONNECT on the ingress HTTP listener", + filterKey: "envoy.filters.http.ext_proc", + listener: IngressHTTPListener, + want: false, + }, + { + name: "CONNECT on the ingress HTTPS listener", + filterKey: "envoy.filters.http.ext_proc", + listener: IngressHTTPSListener, + want: false, + }, + { + // A listener that never requested the attribute falls back to + // ingress, the fail-safe direction. + name: "no attributes at all", + listener: "", + want: false, + }, + { + name: "unrecognised listener", + filterKey: "envoy.filters.http.ext_proc", + listener: "some-other-listener", + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := isEgressRequest(connectRequest(tc.filterKey, tc.listener)); got != tc.want { + t.Errorf("isEgressRequest() = %v, want %v", got, tc.want) + } + }) + } +} + +// A request the client dresses up to look like egress must not be enough: only +// the Envoy-asserted listener name selects the egress handler. +func TestIsEgressRequestIgnoresClientSuppliedAttributeHeader(t *testing.T) { + req := connectRequest("envoy.filters.http.ext_proc", IngressHTTPListener) + rh := req.GetRequestHeaders().GetHeaders() + rh.Headers = append(rh.Headers, + &corev3.HeaderValue{Key: ListenerNameAttribute, RawValue: []byte(EgressListenerName)}, + &corev3.HeaderValue{Key: "x-envoy-listener-name", RawValue: []byte(EgressListenerName)}, + ) + + if isEgressRequest(req) { + t.Error("isEgressRequest() = true for a client-forged listener header, want false") + } +} diff --git a/cmd/atenet/internal/router/extproc_in.go b/cmd/atenet/internal/router/extproc_in.go index 459b26b92..dfa0325da 100644 --- a/cmd/atenet/internal/router/extproc_in.go +++ b/cmd/atenet/internal/router/extproc_in.go @@ -26,12 +26,14 @@ type requestMetadata struct { headers map[string]string path string host string + method string } func newRequestMetadata(headers []*corev3.HeaderValue) *requestMetadata { headersMap := make(map[string]string) var path string var host string + var method string for _, h := range headers { k := strings.ToLower(h.Key) @@ -47,12 +49,16 @@ func newRequestMetadata(headers []*corev3.HeaderValue) *requestMetadata { if k == ":authority" || k == "host" { host = val } + if k == ":method" { + method = val + } } return &requestMetadata{ headers: headersMap, path: path, host: host, + method: method, } } diff --git a/hack/install-ate.sh b/hack/install-ate.sh index 871350a0e..6db16f54b 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -327,6 +327,7 @@ deploy_ate_system() { run_kubectl rollout status deployment/ate-api-server -n ate-system --timeout=120s run_kubectl rollout status deployment/ate-controller -n ate-system --timeout=120s run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s + run_kubectl rollout status deployment/ateway-egress -n ate-system --timeout=120s run_kubectl rollout status statefulset/valkey-cluster -n ate-system --timeout=120s run_kubectl rollout status daemonset/atelet -n ate-system --timeout=120s } @@ -390,11 +391,15 @@ deploy_atenet() { && run_kubectl wait --for=jsonpath='{.status.phase}'=Active namespace/ate-system --timeout=60s run_ko apply -f manifests/ate-install/atenet-router.yaml + run_ko apply -f manifests/ate-install/ateway-egress.yaml run_ko apply -f manifests/ate-install/atenet-dns.yaml run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s - # The Deployment in atenet-dns.yaml is named "dns"; every other resource in - # that file is "atenet-dns". Waiting on the filename rather than the actual - # Deployment made this step fail with NotFound on every successful deploy. + run_kubectl rollout status deployment/ateway-egress -n ate-system --timeout=120s + # SEE(lior): this branch also added a `deployment/atenet-dns` wait here, which + # main has since fixed to `deployment/dns` (the Deployment in atenet-dns.yaml + # is named "dns"; every other resource in that file is "atenet-dns", so the + # old name was NotFound on every successful deploy). Dropped this branch's + # duplicate in favour of main's corrected line. run_kubectl rollout status deployment/dns -n ate-system --timeout=120s } @@ -509,6 +514,8 @@ delete_ate_system() { delete_atenet() { log_step "delete_atenet" run_kubectl delete --ignore-not-found -f manifests/ate-install/atenet-router.yaml + run_kubectl delete --ignore-not-found -f manifests/ate-install/ateway-egress.yaml + run_kubectl delete --ignore-not-found -f manifests/ate-install/atenet-dns.yaml } deploy_benchmarks() { diff --git a/internal/proto/ateletpb/atelet.pb.go b/internal/proto/ateletpb/atelet.pb.go index 021fc8dad..9a7cdf311 100644 --- a/internal/proto/ateletpb/atelet.pb.go +++ b/internal/proto/ateletpb/atelet.pb.go @@ -201,20 +201,25 @@ func (SnapshotScope) EnumDescriptor() ([]byte, []int) { } type RunRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` - Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` - ActorName string `protobuf:"bytes,3,opt,name=actor_name,json=actorName,proto3" json:"actor_name,omitempty"` - ActorUid string `protobuf:"bytes,4,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` - ActorTemplateNamespace string `protobuf:"bytes,5,opt,name=actor_template_namespace,json=actorTemplateNamespace,proto3" json:"actor_template_namespace,omitempty"` - ActorTemplateName string `protobuf:"bytes,6,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"` - Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` + Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` + ActorName string `protobuf:"bytes,3,opt,name=actor_name,json=actorName,proto3" json:"actor_name,omitempty"` + ActorUid string `protobuf:"bytes,4,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + // Actor resource version observed by ate-api when assigning this worker. + ActorVersion int64 `protobuf:"varint,9,opt,name=actor_version,json=actorVersion,proto3" json:"actor_version,omitempty"` + ActorTemplateNamespace string `protobuf:"bytes,5,opt,name=actor_template_namespace,json=actorTemplateNamespace,proto3" json:"actor_template_namespace,omitempty"` + ActorTemplateName string `protobuf:"bytes,6,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"` + Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"` // The sandbox binaries to use for booting this actor from scratch. atelet // fetches the relevant assets and records them with the actor's on-node state // so a later Checkpoint can pin the same version into the snapshot manifest. SandboxAssets *SandboxAssets `protobuf:"bytes,8,opt,name=sandbox_assets,json=sandboxAssets,proto3" json:"sandbox_assets,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Remote egress gateway selected for this activation. When absent, actor + // traffic uses direct egress instead of being redirected through atunnel. + EgressGatewayAddress *string `protobuf:"bytes,10,opt,name=egress_gateway_address,json=egressGatewayAddress,proto3,oneof" json:"egress_gateway_address,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *RunRequest) Reset() { @@ -275,6 +280,13 @@ func (x *RunRequest) GetActorUid() string { return "" } +func (x *RunRequest) GetActorVersion() int64 { + if x != nil { + return x.ActorVersion + } + return 0 +} + func (x *RunRequest) GetActorTemplateNamespace() string { if x != nil { return x.ActorTemplateNamespace @@ -303,6 +315,13 @@ func (x *RunRequest) GetSandboxAssets() *SandboxAssets { return nil } +func (x *RunRequest) GetEgressGatewayAddress() string { + if x != nil && x.EgressGatewayAddress != nil { + return *x.EgressGatewayAddress + } + return "" +} + // AssetFile is one content-addressed file atelet fetches for a sandbox runtime // (e.g. the gVisor runsc binary). type AssetFile struct { @@ -1338,13 +1357,15 @@ func (*CheckpointResponse) Descriptor() ([]byte, []int) { } type RestoreRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` - Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` - ActorName string `protobuf:"bytes,3,opt,name=actor_name,json=actorName,proto3" json:"actor_name,omitempty"` - ActorUid string `protobuf:"bytes,4,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` - ActorTemplateNamespace string `protobuf:"bytes,5,opt,name=actor_template_namespace,json=actorTemplateNamespace,proto3" json:"actor_template_namespace,omitempty"` - ActorTemplateName string `protobuf:"bytes,6,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` + Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` + ActorName string `protobuf:"bytes,3,opt,name=actor_name,json=actorName,proto3" json:"actor_name,omitempty"` + ActorUid string `protobuf:"bytes,4,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + // Actor resource version observed by ate-api when assigning this worker. + ActorVersion int64 `protobuf:"varint,13,opt,name=actor_version,json=actorVersion,proto3" json:"actor_version,omitempty"` + ActorTemplateNamespace string `protobuf:"bytes,5,opt,name=actor_template_namespace,json=actorTemplateNamespace,proto3" json:"actor_template_namespace,omitempty"` + ActorTemplateName string `protobuf:"bytes,6,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"` // Sandbox binary config is not sent on restore: the snapshot is // self-describing. atelet reads the snapshot manifest to recover the pinned // sandbox version that created it. @@ -1366,8 +1387,11 @@ type RestoreRequest struct { // of the `config` oneof: the actor's snapshot may be local (a pause // checkpoint) while the golden snapshot is always external. GoldenSnapshotUriPrefix string `protobuf:"bytes,12,opt,name=golden_snapshot_uri_prefix,json=goldenSnapshotUriPrefix,proto3" json:"golden_snapshot_uri_prefix,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Remote egress gateway selected for this activation. When absent, actor + // traffic uses direct egress instead of being redirected through atunnel. + EgressGatewayAddress *string `protobuf:"bytes,14,opt,name=egress_gateway_address,json=egressGatewayAddress,proto3,oneof" json:"egress_gateway_address,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *RestoreRequest) Reset() { @@ -1428,6 +1452,13 @@ func (x *RestoreRequest) GetActorUid() string { return "" } +func (x *RestoreRequest) GetActorVersion() int64 { + if x != nil { + return x.ActorVersion + } + return 0 +} + func (x *RestoreRequest) GetActorTemplateNamespace() string { if x != nil { return x.ActorTemplateNamespace @@ -1495,6 +1526,13 @@ func (x *RestoreRequest) GetGoldenSnapshotUriPrefix() string { return "" } +func (x *RestoreRequest) GetEgressGatewayAddress() string { + if x != nil && x.EgressGatewayAddress != nil { + return *x.EgressGatewayAddress + } + return "" +} + type isRestoreRequest_Config interface { isRestoreRequest_Config() } @@ -1551,18 +1589,22 @@ var File_atelet_proto protoreflect.FileDescriptor const file_atelet_proto_rawDesc = "" + "\n" + - "\fatelet.proto\x12\x06atelet\"\xe0\x02\n" + + "\fatelet.proto\x12\x06atelet\"\xdb\x03\n" + "\n" + "RunRequest\x12(\n" + "\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" + "\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" + "\n" + "actor_name\x18\x03 \x01(\tR\tactorName\x12\x1b\n" + - "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x128\n" + + "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x12#\n" + + "\ractor_version\x18\t \x01(\x03R\factorVersion\x128\n" + "\x18actor_template_namespace\x18\x05 \x01(\tR\x16actorTemplateNamespace\x12.\n" + "\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" + "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\x12<\n" + - "\x0esandbox_assets\x18\b \x01(\v2\x15.atelet.SandboxAssetsR\rsandboxAssets\"5\n" + + "\x0esandbox_assets\x18\b \x01(\v2\x15.atelet.SandboxAssetsR\rsandboxAssets\x129\n" + + "\x16egress_gateway_address\x18\n" + + " \x01(\tH\x00R\x14egressGatewayAddress\x88\x01\x01B\x19\n" + + "\x17_egress_gateway_address\"5\n" + "\tAssetFile\x12\x10\n" + "\x03url\x18\x01 \x01(\tR\x03url\x12\x16\n" + "\x06sha256\x18\x02 \x01(\tR\x06sha256\"\x8e\x01\n" + @@ -1638,13 +1680,14 @@ const file_atelet_proto_rawDesc = "" + " \x01(\v2'.atelet.ExternalCheckpointConfigurationH\x00R\x0eexternalConfig\x12+\n" + "\x05scope\x18\v \x01(\x0e2\x15.atelet.SnapshotScopeR\x05scopeB\b\n" + "\x06config\"\x14\n" + - "\x12CheckpointResponse\"\xe5\x04\n" + + "\x12CheckpointResponse\"\xe0\x05\n" + "\x0eRestoreRequest\x12(\n" + "\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" + "\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" + "\n" + "actor_name\x18\x03 \x01(\tR\tactorName\x12\x1b\n" + - "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x128\n" + + "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x12#\n" + + "\ractor_version\x18\r \x01(\x03R\factorVersion\x128\n" + "\x18actor_template_namespace\x18\x05 \x01(\tR\x16actorTemplateNamespace\x12.\n" + "\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" + "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\x12*\n" + @@ -1653,8 +1696,10 @@ const file_atelet_proto_rawDesc = "" + "\x0fexternal_config\x18\n" + " \x01(\v2'.atelet.ExternalCheckpointConfigurationH\x00R\x0eexternalConfig\x12+\n" + "\x05scope\x18\v \x01(\x0e2\x15.atelet.SnapshotScopeR\x05scope\x12;\n" + - "\x1agolden_snapshot_uri_prefix\x18\f \x01(\tR\x17goldenSnapshotUriPrefixB\b\n" + - "\x06config\"\x11\n" + + "\x1agolden_snapshot_uri_prefix\x18\f \x01(\tR\x17goldenSnapshotUriPrefix\x129\n" + + "\x16egress_gateway_address\x18\x0e \x01(\tH\x01R\x14egressGatewayAddress\x88\x01\x01B\b\n" + + "\x06configB\x19\n" + + "\x17_egress_gateway_address\"\x11\n" + "\x0fRestoreResponse*`\n" + "\n" + "VolumeType\x12\x1b\n" + @@ -1761,6 +1806,7 @@ func file_atelet_proto_init() { if File_atelet_proto != nil { return } + file_atelet_proto_msgTypes[0].OneofWrappers = []any{} file_atelet_proto_msgTypes[7].OneofWrappers = []any{ (*Volume_DurableDir)(nil), (*Volume_External)(nil), diff --git a/internal/proto/ateletpb/atelet.proto b/internal/proto/ateletpb/atelet.proto index cee9e6a89..c2e1d5aa3 100644 --- a/internal/proto/ateletpb/atelet.proto +++ b/internal/proto/ateletpb/atelet.proto @@ -38,6 +38,8 @@ message RunRequest { string atespace = 2; string actor_name = 3; string actor_uid = 4; + // Actor resource version observed by ate-api when assigning this worker. + int64 actor_version = 9; string actor_template_namespace = 5; string actor_template_name = 6; @@ -48,6 +50,10 @@ message RunRequest { // fetches the relevant assets and records them with the actor's on-node state // so a later Checkpoint can pin the same version into the snapshot manifest. SandboxAssets sandbox_assets = 8; + + // Remote egress gateway selected for this activation. When absent, actor + // traffic uses direct egress instead of being redirected through atunnel. + optional string egress_gateway_address = 10; } // AssetFile is one content-addressed file atelet fetches for a sandbox runtime @@ -227,6 +233,8 @@ message RestoreRequest { string atespace = 2; string actor_name = 3; string actor_uid = 4; + // Actor resource version observed by ate-api when assigning this worker. + int64 actor_version = 13; string actor_template_namespace = 5; string actor_template_name = 6; @@ -254,6 +262,9 @@ message RestoreRequest { // of the `config` oneof: the actor's snapshot may be local (a pause // checkpoint) while the golden snapshot is always external. string golden_snapshot_uri_prefix = 12; + // Remote egress gateway selected for this activation. When absent, actor + // traffic uses direct egress instead of being redirected through atunnel. + optional string egress_gateway_address = 14; } message RestoreResponse { diff --git a/manifests/ate-install/atelet.yaml b/manifests/ate-install/atelet.yaml index b8e057a86..43aa1e44d 100644 --- a/manifests/ate-install/atelet.yaml +++ b/manifests/ate-install/atelet.yaml @@ -71,6 +71,13 @@ spec: - --gcp-auth-for-image-pulls=true - --grpc-server-cred-bundle=/run/podidentity.podcert.ate.dev/credential-bundle.pem - --client-ca-certs=/run/podidentity.podcert.ate.dev/trust-bundle.pem + # SEE(lior): purely additive against main's new mTLS flags — both sides + # kept. This is the one line that arms the nftables egress REDIRECT that + # shipped dormant with atunnel. + # ONLY UNTIL API IS DECIDED, FOR POC: turn on pluggable actor egress cluster-wide. Actor TCP egress is + # transparently redirected (nftables) into atunnel, which wraps it in + # mTLS + HTTP CONNECT to the an egress gateway. + - --egress-gateway-address=ateway-egress.ate-system.svc:443 # atelet does no mounts, netlink, device, or namespace operations (those # live in the ateom worker pod) — it only reads/writes the # /var/lib/ateom-gvisor hostPath as root, so it needs no Linux diff --git a/manifests/ate-install/ateway-egress.yaml b/manifests/ate-install/ateway-egress.yaml new file mode 100644 index 000000000..1c6a4a382 --- /dev/null +++ b/manifests/ate-install/ateway-egress.yaml @@ -0,0 +1,338 @@ +# 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. + +# contract expected by atunnel's egress client: +# * downstream mTLS on :443 (present servicedns identity, require + verify a +# podidentity client cert), +# * terminate the actor's HTTP CONNECT and tunnel raw TCP to the CONNECT +# authority (the actor's original destination, always sent as IP:port). +apiVersion: v1 +kind: ServiceAccount +metadata: + name: ateway-egress + namespace: ate-system +--- +# RBAC for the co-located ext_proc sidecar (atenet router), which watches +# ActorTemplates as part of its normal operation. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: ateway-egress +rules: +- apiGroups: + - "ate.dev" + resources: + - actortemplates + verbs: + - get + - watch + - list +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: ateway-egress +subjects: +- kind: ServiceAccount + name: ateway-egress + namespace: ate-system +roleRef: + kind: ClusterRole + name: ateway-egress + apiGroup: rbac.authorization.k8s.io +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: ateway-egress + namespace: ate-system +data: + envoy.yaml: | + admin: + address: + socket_address: { address: 0.0.0.0, port_value: 15000 } + static_resources: + listeners: + - name: egress + address: + socket_address: { address: 0.0.0.0, port_value: 443 } + filter_chains: + - transport_socket: + name: envoy.transport_sockets.tls + typed_config: + "@type": type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.DownstreamTlsContext + require_client_certificate: true + common_tls_context: + tls_certificates: + # Gateway server identity (servicedns signer). credential-bundle.pem + # holds the leaf cert and key concatenated. watched_directory picks + # up kubelet's projected certificate rotation without a restart. + - certificate_chain: { filename: /run/servicedns.podcert.ate.dev/credential-bundle.pem } + private_key: { filename: /run/servicedns.podcert.ate.dev/credential-bundle.pem } + watched_directory: { path: /run/servicedns.podcert.ate.dev } + validation_context: + # Actors/workers authenticate with a podidentity client cert. + trusted_ca: { filename: /run/podidentity.podcert.ate.dev/trust-bundle.pem } + filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: egress_connect + codec_type: HTTP1 + upgrade_configs: + - upgrade_type: CONNECT + # Emit the access log as soon as the CONNECT tunnel is established + # (atunnel keeps the tunnel open, so the default log-on-close would + # not fire during the demo). + access_log_options: + flush_log_on_tunnel_successfully_established: true + access_log: + - name: envoy.access_loggers.stdout + typed_config: + "@type": type.googleapis.com/envoy.extensions.access_loggers.stream.v3.StdoutAccessLog + log_format: + text_format_source: + # SEE(lior): actor= reads X-ATE-ACTOR-NAME, not X-ATE-ACTOR. + # atunnel's ActorNameHeader was renamed during PR review + # after this manifest was written; the Go handlers use the + # constant so they followed the rename, but this literal + # did not and was logging an empty actor. + inline_string: "[egress] authority=%REQ(:AUTHORITY)% atespace=%REQ(X-ATE-ATESPACE)% actor=%REQ(X-ATE-ACTOR-NAME)% ver=%REQ(X-ATE-ACTOR-VERSION)% auth_present=%REQ(AUTHORIZATION)% code=%RESPONSE_CODE% flags=%RESPONSE_FLAGS% up_bytes=%BYTES_RECEIVED% down_bytes=%BYTES_SENT%\n" + route_config: + name: connect_route + virtual_hosts: + - name: connect + domains: ["*"] + routes: + - match: { connect_matcher: {} } + route: + cluster: egress_forward_proxy + upgrade_configs: + - upgrade_type: CONNECT + connect_config: {} + http_filters: + # Actor-identity authentication: on every egress CONNECT, ext_proc + # asks the atenet router (which calls the ate API) whether the + # worker-asserted X-Ate-* identity is a real, running actor. Denials + # come back as an immediate 403. Fails closed if the router is down. + - name: envoy.filters.http.ext_proc + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor + grpc_service: + envoy_grpc: + cluster_name: ext_proc_server + timeout: 2s + failure_mode_allow: false + # How the ext_proc server tells egress from ingress. It applies + # opposite trust models to the two directions, and dispatches on + # this Envoy-asserted listener name so that no client can select + # the egress path by crafting a request. The value must match + # EgressListenerName in cmd/atenet/internal/router/extproc_egress.go; + # renaming the listener above without updating it fails closed + # (egress requests take the ingress path and 404). + request_attributes: + - xds.listener_name + processing_mode: + request_header_mode: SEND + response_header_mode: SKIP + request_body_mode: NONE + response_body_mode: NONE + request_trailer_mode: SKIP + response_trailer_mode: SKIP + - name: envoy.filters.http.dynamic_forward_proxy + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_forward_proxy.v3.FilterConfig + dns_cache_config: + name: egress_dns_cache + dns_lookup_family: V4_ONLY + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + clusters: + # ext_proc gRPC server = the atenet router, co-located in this pod as a + # sidecar and called over localhost (same topology the ingress gateway uses + # for its Envoy + ext_proc). + - name: ext_proc_server + type: STATIC + lb_policy: ROUND_ROBIN + connect_timeout: 1s + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: ext_proc_server + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 50051 + # Dials the terminated-CONNECT target (IP:port from the authority). atunnel + # always sends an IP:port, so DNS resolution is effectively a passthrough. + - name: egress_forward_proxy + lb_policy: CLUSTER_PROVIDED + connect_timeout: 5s + cluster_type: + name: envoy.clusters.dynamic_forward_proxy + typed_config: + "@type": type.googleapis.com/envoy.extensions.clusters.dynamic_forward_proxy.v3.ClusterConfig + dns_cache_config: + name: egress_dns_cache + dns_lookup_family: V4_ONLY +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: ateway-egress + namespace: ate-system + labels: + app: ateway-egress +spec: + replicas: 1 + selector: + matchLabels: + app: ateway-egress + template: + metadata: + labels: + app: ateway-egress + spec: + serviceAccountName: ateway-egress + securityContext: + # Allow the non-root envoy user to bind :443. + sysctls: + - name: net.ipv4.ip_unprivileged_port_start + value: "0" + containers: + - name: envoy + image: envoyproxy/envoy:v1.34-latest + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + runAsUser: 65532 + args: + - -c + - /etc/envoy/envoy.yaml + - --service-node + - ateway-egress + - --service-cluster + - ateway-egress + ports: + - name: https + containerPort: 443 + - name: admin + containerPort: 15000 + readinessProbe: + httpGet: + path: /ready + port: admin + periodSeconds: 10 + startupProbe: + failureThreshold: 60 + httpGet: + path: /ready + port: admin + periodSeconds: 1 + volumeMounts: + - name: config + mountPath: /etc/envoy + readOnly: true + - name: servicedns + mountPath: /run/servicedns.podcert.ate.dev + readOnly: true + - name: podidentity + mountPath: /run/podidentity.podcert.ate.dev + readOnly: true + # Co-located ext_proc server (the atenet router, ext_proc-only). The egress + # Envoy calls it over localhost to authenticate actor identity against the + # ate API on every CONNECT. This mirrors the ingress gateway topology + # (Envoy + ext_proc in one pod); a shared/standalone ext_proc is a future step. + - name: ext-proc + image: ko://github.com/agent-substrate/substrate/cmd/atenet + args: + - router + - --standalone + - --namespace=ate-system + - --port-extproc=50051 + - --extproc-address=127.0.0.1 + - --ateapi-address=api.ate-system.svc:443 + - --ateapi-auth=mtls + - --otlp-collector-address= + env: + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + ports: + - name: extproc + containerPort: 50051 + readinessProbe: + tcpSocket: + port: extproc + periodSeconds: 10 + volumes: + - name: config + configMap: + name: ateway-egress + - name: servicedns + projected: + sources: + - podCertificate: + signerName: servicedns.podcert.ate.dev/identity + keyType: ECDSAP256 + credentialBundlePath: credential-bundle.pem + - clusterTrustBundle: + signerName: servicedns.podcert.ate.dev/identity + labelSelector: + matchLabels: + podcert.ate.dev/canarying: live + path: trust-bundle.pem + - name: podidentity + projected: + sources: + - podCertificate: + signerName: podidentity.podcert.ate.dev/identity + keyType: ECDSAP256 + credentialBundlePath: credential-bundle.pem + - clusterTrustBundle: + signerName: podidentity.podcert.ate.dev/identity + labelSelector: + matchLabels: + podcert.ate.dev/canarying: live + path: trust-bundle.pem +--- +apiVersion: v1 +kind: Service +metadata: + name: ateway-egress + namespace: ate-system +spec: + type: ClusterIP + selector: + app: ateway-egress + ports: + - name: https + port: 443 + targetPort: https + protocol: TCP diff --git a/manifests/ate-install/kind/atelet/kustomization.yaml b/manifests/ate-install/kind/atelet/kustomization.yaml index c70724fac..c7c87d3d1 100644 --- a/manifests/ate-install/kind/atelet/kustomization.yaml +++ b/manifests/ate-install/kind/atelet/kustomization.yaml @@ -38,6 +38,11 @@ patches: # Kind clusters are dev/CI: run atelet at debug so e2e suites can # assert on per-item log lines. Production installs default to info. - --log-level=debug + # SEE(lior): both sides kept. A strategic-merge patch replaces the + # whole args list rather than merging it, so dropping either side's + # flags here would silently unset them on the kind DaemonSet. + # POC: turn on pluggable actor egress cluster-wide. + - --egress-gateway-address=ateway-egress.ate-system.svc:443 env: - name: OTEL_EXPORTER_OTLP_ENDPOINT value: http://opentelemetry-collector.otel-system.svc:4317 diff --git a/manifests/ate-install/kind/kustomization.yaml b/manifests/ate-install/kind/kustomization.yaml index adc7858f4..1e264e670 100644 --- a/manifests/ate-install/kind/kustomization.yaml +++ b/manifests/ate-install/kind/kustomization.yaml @@ -20,6 +20,7 @@ resources: - ../ate-controller.yaml - ./atelet - ../atenet-dns.yaml + - ../ateway-egress.yaml - ../atenet-router.yaml - ../valkey.yaml - ../pod-certificate-controller.yaml diff --git a/manifests/ate-install/token-client/kustomization.yaml b/manifests/ate-install/token-client/kustomization.yaml index fc004748d..d35743f81 100644 --- a/manifests/ate-install/token-client/kustomization.yaml +++ b/manifests/ate-install/token-client/kustomization.yaml @@ -23,6 +23,7 @@ resources: - ../ate-controller.yaml - ../atelet.yaml - ../atenet-dns.yaml + - ../ateway-egress.yaml - ../atenet-router.yaml - ../valkey.yaml - ../pod-certificate-controller.yaml From 598a5487cf5c82cba5bb8e2f2f1c3dd82ff4e648 Mon Sep 17 00:00:00 2001 From: Lior Lieberman Date: Mon, 27 Jul 2026 16:32:39 -0700 Subject: [PATCH 2/3] egress: add the egress demo and e2e coverage demos/egress is a small Actor that fetches a URL it is given and echoes the upstream status and body back, which makes the egress path observable from outside the sandbox. hack/install-demo-egress.sh registers it as a --deploy-demo-egress fixture and hack/verify-egress-demo.sh drives it and checks the atenet-egress logs for the corresponding authorized CONNECT. TestActorEgress in the networking suite covers the same path automatically: it creates an Actor from the demo template, POSTs a fetch request through atenet-router, and asserts 200. The suite's actor helper is parameterised by template so the ingress test keeps using the counter fixture. --- .github/workflows/pr-workflow.yaml | 4 + Makefile | 2 +- cmd/atenet/internal/router/extproc.go | 8 - cmd/atenet/internal/router/extproc_egress.go | 47 +++-- .../internal/router/extproc_egress_test.go | 44 ++--- demos/egress/README.md | 133 +++++++++++++ demos/egress/egress.yaml.tmpl | 56 ++++++ demos/egress/main.go | 125 ++++++++++++ demos/egress/main_test.go | 109 ++++++++++ demos/egress/test-egress.sh | 186 ++++++++++++++++++ hack/install-ate.sh | 14 +- hack/install-demo-egress.sh | 51 +++++ hack/verify-egress-demo.sh | 65 ++++++ internal/e2e/router_client.go | 21 +- internal/e2e/router_client_test.go | 78 ++++++++ .../e2e/suites/networking/networking_test.go | 56 +++++- manifests/ate-install/atelet.yaml | 7 +- ...{ateway-egress.yaml => atenet-egress.yaml} | 70 ++++--- .../kind/atelet/kustomization.yaml | 6 +- manifests/ate-install/kind/kustomization.yaml | 2 +- .../token-client/kustomization.yaml | 2 +- 21 files changed, 977 insertions(+), 109 deletions(-) create mode 100644 demos/egress/README.md create mode 100644 demos/egress/egress.yaml.tmpl create mode 100644 demos/egress/main.go create mode 100644 demos/egress/main_test.go create mode 100755 demos/egress/test-egress.sh create mode 100644 hack/install-demo-egress.sh create mode 100644 hack/verify-egress-demo.sh create mode 100644 internal/e2e/router_client_test.go rename manifests/ate-install/{ateway-egress.yaml => atenet-egress.yaml} (87%) diff --git a/.github/workflows/pr-workflow.yaml b/.github/workflows/pr-workflow.yaml index 47a14340c..708a8aa3a 100644 --- a/.github/workflows/pr-workflow.yaml +++ b/.github/workflows/pr-workflow.yaml @@ -96,6 +96,10 @@ jobs: run: hack/run-microvm-demo-kind.sh --ateapi-client-auth=${{ matrix.ateapi-client-auth }} - name: Deploy gVisor counter demo run: hack/install-ate-kind.sh --deploy-demo-counter + - name: Deploy egress demo + # TestActorEgress in the networking suite builds its Actor from the + # ate-demo-egress/egress ActorTemplate, so the fixture has to exist before. + run: hack/install-ate-kind.sh --deploy-demo-egress - name: Wait for micro-VM golden snapshot run: | kubectl --context kind-kind wait --for=condition=Ready \ diff --git a/Makefile b/Makefile index c68dbd3ea..9f424b239 100644 --- a/Makefile +++ b/Makefile @@ -57,7 +57,7 @@ build-atenet: .PHONY: build-demos build-demos: - $(KO) build --ldflags="$(LDFLAGS)" ./demos/counter + $(KO) build --ldflags="$(LDFLAGS)" ./demos/counter ./demos/egress .PHONY: test test: diff --git a/cmd/atenet/internal/router/extproc.go b/cmd/atenet/internal/router/extproc.go index cba79c9af..55d280322 100644 --- a/cmd/atenet/internal/router/extproc.go +++ b/cmd/atenet/internal/router/extproc.go @@ -101,14 +101,6 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer // the accepting listener, not by anything in the request itself // (see isEgressRequest). // - // SEE(lior): main grew a ResumeOutcome return on - // handleRequestHeaders while this branch was out. Rather than - // splitting the dispatch into two separately-typed call sites, the - // egress handler was widened to the same signature and returns - // ResumeOutcomeNone — egress requires an already-RUNNING actor and - // never resumes one, so "none" is accurate, and it keeps the - // route-duration metric's resume label a closed set (no new empty - // value) across both directions. handle := s.handleRequestHeaders if isEgressRequest(req) { handle = s.handleEgressRequestHeaders diff --git a/cmd/atenet/internal/router/extproc_egress.go b/cmd/atenet/internal/router/extproc_egress.go index 69f3411be..2a19a347e 100644 --- a/cmd/atenet/internal/router/extproc_egress.go +++ b/cmd/atenet/internal/router/extproc_egress.go @@ -31,42 +31,51 @@ import ( ) const ( - // EgressListenerName is the Envoy listener that terminates actor egress - // CONNECTs. It must stay in sync with the listener name in - // manifests/ate-install/ateway-egress.yaml. - EgressListenerName = "egress" - // ListenerNameAttribute is the CEL attribute carrying the name of the - // listener that accepted the request. The egress Envoy asks for it via + // EgressFilterChainName is the Envoy filter chain that terminates actor + // egress CONNECTs. It must stay in sync with the filter chain name in + // manifests/ate-install/atenet-egress.yaml. + EgressFilterChainName = "egress" + // FilterChainNameAttribute is the CEL attribute carrying the name of the + // filter chain that accepted the request. The egress Envoy asks for it via // request_attributes on its ext_proc filter. - ListenerNameAttribute = "xds.listener_name" + // + // SEE(lior): this was xds.listener_name, which reads more naturally but + // which Envoy 1.34 cannot parse: it logs "error parsing cel expression + // xds.listener_name" at trace level, then sends the ProcessingRequest with + // an empty attributes map rather than failing config load. Because an + // absent attribute means "ingress" (the fail-safe direction), every egress + // CONNECT silently took the ingress path and 404'd on the actor DNS name + // parse. xds.filter_chain_name parses on the same Envoy build and is + // equally Envoy-asserted, so the trust model is unchanged. + FilterChainNameAttribute = "xds.filter_chain_name" ) // isEgressRequest reports whether an ext_proc RequestHeaders callback arrived on -// the egress gateway's listener rather than on an ingress listener. This lets -// one ext_proc server handle both directions off the same stream. +// the egress gateway's filter chain rather than on an ingress one. This lets one +// ext_proc server handle both directions off the same stream. // -// Dispatch is by listener, not by :method, because the two handlers apply +// Dispatch is by filter chain, not by :method, because the two handlers apply // opposite trust models: on egress the X-Ate-* identity headers are asserted by // atunnel over mTLS, while on ingress the same headers are unauthenticated // client input. Keying on :method would let any external client sending CONNECT // select the egress handler and use its denial messages as an actor-existence -// and status oracle. Envoy asserts the listener name; the request cannot +// and status oracle. Envoy asserts the filter chain name; the request cannot // influence it. // // An unrecognized or absent attribute means ingress, the fail-safe direction: an // egress request misrouted to the ingress handler fails to parse as an actor DNS // name and 404s, whereas the reverse leaks control-plane state. func isEgressRequest(req *extprocv3.ProcessingRequest) bool { - return listenerName(req) == EgressListenerName + return filterChainName(req) == EgressFilterChainName } -// listenerName returns the xds.listener_name attribute Envoy attached to the -// request, or "" when the listener did not request the attribute. The +// filterChainName returns the xds.filter_chain_name attribute Envoy attached to +// the request, or "" when the listener did not request the attribute. The // attributes map is keyed by the ext_proc filter's name within the HCM chain, // which we do not want to hardcode here, so scan every entry. -func listenerName(req *extprocv3.ProcessingRequest) string { +func filterChainName(req *extprocv3.ProcessingRequest) string { for _, attrs := range req.GetAttributes() { - if v, ok := attrs.GetFields()[ListenerNameAttribute]; ok { + if v, ok := attrs.GetFields()[FilterChainNameAttribute]; ok { return v.GetStringValue() } } @@ -84,9 +93,9 @@ func listenerName(req *extprocv3.ProcessingRequest) string { // with a single branch. The (target, tmplNs, tmplName) results are unused for // egress and returned empty. // -// SEE(lior): the trailing ResumeOutcome exists only to match -// handleRequestHeaders after main added it. Egress never resumes an actor — it -// requires one already RUNNING — so every path returns ResumeOutcomeNone. +// The trailing ResumeOutcome exists only to match handleRequestHeaders. +// Egress never resumes an actor — it requires one already RUNNING - +// so every path returns ResumeOutcomeNone. func (s *ExtProcServer) handleEgressRequestHeaders( ctx context.Context, reqHeaders *extprocv3.HttpHeaders, diff --git a/cmd/atenet/internal/router/extproc_egress_test.go b/cmd/atenet/internal/router/extproc_egress_test.go index f9d449f28..90a4a3854 100644 --- a/cmd/atenet/internal/router/extproc_egress_test.go +++ b/cmd/atenet/internal/router/extproc_egress_test.go @@ -23,9 +23,9 @@ import ( ) // connectRequest builds a RequestHeaders ProcessingRequest for a CONNECT -// carrying worker-asserted actor identity, optionally attributed to a listener. -// filterKey is the ext_proc filter name Envoy keys the attributes map by. -func connectRequest(filterKey, listener string) *extprocv3.ProcessingRequest { +// carrying worker-asserted actor identity, optionally attributed to a filter +// chain. filterKey is the ext_proc filter name Envoy keys the attributes map by. +func connectRequest(filterKey, chain string) *extprocv3.ProcessingRequest { req := &extprocv3.ProcessingRequest{ Request: &extprocv3.ProcessingRequest_RequestHeaders{ RequestHeaders: &extprocv3.HttpHeaders{ @@ -40,11 +40,11 @@ func connectRequest(filterKey, listener string) *extprocv3.ProcessingRequest { }, }, } - if listener != "" { + if chain != "" { req.Attributes = map[string]*structpb.Struct{ filterKey: { Fields: map[string]*structpb.Value{ - ListenerNameAttribute: structpb.NewStringValue(listener), + FilterChainNameAttribute: structpb.NewStringValue(chain), }, }, } @@ -56,19 +56,19 @@ func TestIsEgressRequest(t *testing.T) { tests := []struct { name string filterKey string - listener string + chain string want bool }{ { - name: "egress listener", + name: "egress filter chain", filterKey: "envoy.filters.http.ext_proc", - listener: EgressListenerName, + chain: EgressFilterChainName, want: true, }, { - name: "egress listener under a renamed filter", + name: "egress filter chain under a renamed filter", filterKey: "some.custom.ext_proc.name", - listener: EgressListenerName, + chain: EgressFilterChainName, want: true, }, { @@ -78,33 +78,33 @@ func TestIsEgressRequest(t *testing.T) { // exists and is running. name: "CONNECT on the ingress HTTP listener", filterKey: "envoy.filters.http.ext_proc", - listener: IngressHTTPListener, + chain: IngressHTTPListener, want: false, }, { name: "CONNECT on the ingress HTTPS listener", filterKey: "envoy.filters.http.ext_proc", - listener: IngressHTTPSListener, + chain: IngressHTTPSListener, want: false, }, { // A listener that never requested the attribute falls back to // ingress, the fail-safe direction. - name: "no attributes at all", - listener: "", - want: false, + name: "no attributes at all", + chain: "", + want: false, }, { - name: "unrecognised listener", + name: "unrecognised filter chain", filterKey: "envoy.filters.http.ext_proc", - listener: "some-other-listener", + chain: "some-other-chain", want: false, }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - if got := isEgressRequest(connectRequest(tc.filterKey, tc.listener)); got != tc.want { + if got := isEgressRequest(connectRequest(tc.filterKey, tc.chain)); got != tc.want { t.Errorf("isEgressRequest() = %v, want %v", got, tc.want) } }) @@ -112,16 +112,16 @@ func TestIsEgressRequest(t *testing.T) { } // A request the client dresses up to look like egress must not be enough: only -// the Envoy-asserted listener name selects the egress handler. +// the Envoy-asserted filter chain name selects the egress handler. func TestIsEgressRequestIgnoresClientSuppliedAttributeHeader(t *testing.T) { req := connectRequest("envoy.filters.http.ext_proc", IngressHTTPListener) rh := req.GetRequestHeaders().GetHeaders() rh.Headers = append(rh.Headers, - &corev3.HeaderValue{Key: ListenerNameAttribute, RawValue: []byte(EgressListenerName)}, - &corev3.HeaderValue{Key: "x-envoy-listener-name", RawValue: []byte(EgressListenerName)}, + &corev3.HeaderValue{Key: FilterChainNameAttribute, RawValue: []byte(EgressFilterChainName)}, + &corev3.HeaderValue{Key: "x-envoy-filter-chain-name", RawValue: []byte(EgressFilterChainName)}, ) if isEgressRequest(req) { - t.Error("isEgressRequest() = true for a client-forged listener header, want false") + t.Error("isEgressRequest() = true for a client-forged filter chain header, want false") } } diff --git a/demos/egress/README.md b/demos/egress/README.md new file mode 100644 index 000000000..4517be99a --- /dev/null +++ b/demos/egress/README.md @@ -0,0 +1,133 @@ +# Egress Demo — Pluggable Egress Networking + +This demo shows an Actor's outbound traffic being **transparently tunneled through an +egress gateway** and **authenticated by actor identity**, end to end. + +The Actor is a tiny service that accepts `{"url":"..."}`, performs an HTTP `GET`, and returns +the upstream response. The Actor believes it is dialing plain HTTP directly — but its egress is +intercepted and carried over mTLS to a gateway that verifies who is making the request. + +## What it demonstrates + +``` + ┌──────────────── ateom worker pod ─────────────────┐ + │ Actor (gVisor) │ + │ GET http://:80/ (plain HTTP) │ + │ │ │ + │ ▼ nftables REDIRECT │ + │ atunnel egress ──(mTLS + HTTP CONNECT, │ + │ │ X-Ate-Atespace/Actor/Version) │ + └────────┼────────────────────────────────────────────┘ + ▼ + ┌──────────── atenet-egress pod ───────────────────┐ + │ Envoy egress gateway │ + │ • terminates downstream mTLS │ + │ • terminates HTTP CONNECT │ + │ • ext_proc ──(localhost)──► atenet router (ext_proc sidecar) + │ • dynamic_forward_proxy │ GetActor(atespace, actor) → ate API + │ │ │ allow RUNNING actor / deny 403 + └───────────┼───────────────────────────────────────┘ + ▼ + real destination (the CONNECT authority, an IP:port) +``` + +1. **Guide 1 — gateway accepts CONNECT + mTLS.** `atenet-egress` is an Envoy dynamic-forward-proxy + that terminates the actor's mTLS `CONNECT` and tunnels to the requested destination. +2. **Guide 2 — transparent interception.** `nftables` REDIRECTs actor TCP egress into `atunnel`, + which wraps it in mTLS + `CONNECT`. +3. **Guide 3 — HTTP-only actors, identity injected.** The Actor only dials plain HTTP; atunnel + still injects `X-Ate-Atespace`/`X-Ate-Actor`/`X-Ate-Actor-Version` identity headers. +4. **Identity authentication.** The gateway calls `ext_proc` on every CONNECT; the **atenet router** + (co-located in the gateway pod as an ext_proc sidecar, the same ext_proc code used for ingress) + validates the asserted identity against the ate API (`GetActor`) and returns **403** unless it is + a real, `RUNNING` actor. This mirrors the ingress gateway's Envoy + ext_proc co-location; a + standalone/shared ext_proc is a future step. + +## Components + +- **Egress app (`main.go`)** — the Actor: `POST /` with `{"url":"..."}` → fetches it → returns + status + body. +- **Egress gateway** — `manifests/ate-install/atenet-egress.yaml`. One pod, two containers: + an Envoy (`envoy`) and the atenet router ext_proc (`ext-proc`), called over localhost. +- **Egress opt-in** — `atelet --egress-gateway-address=atenet-egress.ate-system.svc:443` + (set in the kind overlay), which turns on tunneled egress cluster-wide. + +## Prerequisites + +- A kind cluster with Agent Substrate installed (`hack/create-kind-cluster.sh` then + `hack/install-ate-kind.sh --deploy-ate-system`). Egress is enabled by the atelet flag above. +- `ko`, `kubectl`, and `kubectl-ate` (`go install ./cmd/kubectl-ate`). + +## Deploy the demo fixture + +```bash +./hack/install-ate.sh --deploy-demo-egress +kubectl wait --for=condition=Ready actortemplate/egress -n ate-demo-egress --timeout=5m +``` + +## Run the automated test (easiest) + +```bash +./demos/egress/test-egress.sh +``` + +It deploys an in-cluster HTTP target, creates & resumes an Actor, then asserts: + +- **positive** — a real Actor's egress reaches the target (`HTTP 200`) *through the gateway* + (the target sees the gateway's IP as its client), and the gateway logs the CONNECT with the + actor identity; +- **negative** — a spoofed/unknown actor identity is rejected by ext_proc with **`HTTP 403`**. + +Add `--cleanup` to remove everything the script created. + +## Manual walkthrough + +```bash +# 1. An in-cluster target the Actor will fetch (any HTTP server works). +kubectl create namespace egress-target +kubectl -n egress-target create deployment whoami --image=traefik/whoami +kubectl -n egress-target expose deployment whoami --port=80 +TARGET_IP=$(kubectl -n egress-target get svc whoami -o jsonpath='{.spec.clusterIP}') + +# 2. Create and resume an Actor. +kubectl ate create atespace demo +kubectl ate create actor egress-demo -a demo --template ate-demo-egress/egress +kubectl ate resume actor egress-demo -a demo # wait for STATUS_RUNNING + +# 3. Drive the Actor's egress through the ingress gateway. +kubectl -n ate-system port-forward service/atenet-router 8000:80 & +curl -s -X POST http://localhost:8000/ \ + -H 'Host: egress-demo.demo.actors.resources.substrate.ate.dev' \ + -H 'Content-Type: application/json' \ + -d "{\"url\":\"http://${TARGET_IP}:80/\"}" +``` + +### What to observe + +```bash +# The egress gateway logs each tunneled CONNECT with the actor identity + result: +kubectl -n ate-system logs deploy/atenet-egress | grep '\[egress\]' +# [egress] authority=:80 atespace=demo actor=egress-demo ver=… code=200 … + +# The co-located ext_proc sidecar logs the identity decision: +kubectl -n ate-system logs deploy/atenet-egress -c ext-proc | grep -i 'egress identity\|egress denied' +# egress identity authenticated atespace=demo actor=egress-demo status=STATUS_RUNNING +``` + +The `whoami` body shows `RemoteAddr: ` — proof the request egressed +*through* the gateway rather than directly. + +## Notes / limitations + +- This milestone **authenticates** identity (is this a real, running actor?). **Authorizing** + egress by destination and injecting upstream credentials/tokens is a follow-up, implemented in + the same `ext_proc` (policy API TBD). +- The gateway currently trusts any pod-identity holder's *asserted* `X-Ate-*` headers; binding the + mTLS worker identity to the actor's assigned worker is the next hardening step. + +## Cleanup + +```bash +./demos/egress/test-egress.sh --cleanup +./hack/install-ate.sh --delete-demo-egress +``` diff --git a/demos/egress/egress.yaml.tmpl b/demos/egress/egress.yaml.tmpl new file mode 100644 index 000000000..e6f9b080f --- /dev/null +++ b/demos/egress/egress.yaml.tmpl @@ -0,0 +1,56 @@ +# 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. + +apiVersion: v1 +kind: Namespace +metadata: + name: ate-demo-egress + +--- + +apiVersion: ate.dev/v1alpha1 +kind: WorkerPool +metadata: + name: egress + namespace: ate-demo-egress + labels: + workload: egress +spec: + replicas: 2 + ateomImage: ko://github.com/agent-substrate/substrate/cmd/ateom-gvisor + +--- + +apiVersion: ate.dev/v1alpha1 +kind: ActorTemplate +metadata: + name: egress + namespace: ate-demo-egress +spec: + pauseImage: "registry.k8s.io/pause:3.10.2@sha256:f548e0e8e3dc1896ca956272154dde3314e8cc4fde0a57577ee9fa1c63f5baf4" + containers: + - name: egress + image: ko://github.com/agent-substrate/substrate/demos/egress + command: ["/ko-app/egress"] + readyz: + httpGet: + path: /readyz + port: 80 + workerSelector: + matchLabels: + workload: egress + snapshotsConfig: + onPause: Full + onCommit: Full + location: gs://${BUCKET_NAME}/ate-demo-egress/ diff --git a/demos/egress/main.go b/demos/egress/main.go new file mode 100644 index 000000000..4c0288e12 --- /dev/null +++ b/demos/egress/main.go @@ -0,0 +1,125 @@ +// 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. + +// Command egress is a small HTTP service for demonstrating per-Actor egress +// policy. It accepts a URL, fetches it, and returns the upstream response. +package main + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "os" + "time" +) + +const ( + listenAddress = ":80" + maxRequestBody = 64 << 10 + maxResponseBody = 1 << 20 + requestTimeout = 15 * time.Second +) + +type fetchRequest struct { + URL string `json:"url"` +} + +type fetchResponse struct { + StatusCode int `json:"statusCode,omitempty"` + Body string `json:"body,omitempty"` + Error string `json:"error,omitempty"` +} + +func main() { + slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil))) + + client := &http.Client{Timeout: requestTimeout} + slog.Info("starting egress demo", "address", listenAddress) + if err := http.ListenAndServe(listenAddress, newHandler(client)); err != nil { + slog.Error("egress demo stopped", "error", err) + os.Exit(1) + } +} + +func newHandler(client *http.Client) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "ok\n") + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + writeJSON(w, http.StatusMethodNotAllowed, fetchResponse{Error: "method must be POST"}) + return + } + + var input fetchRequest + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxRequestBody)) + if err := decoder.Decode(&input); err != nil { + writeJSON(w, http.StatusBadRequest, fetchResponse{Error: fmt.Sprintf("invalid JSON payload: %v", err)}) + return + } + if err := validateURL(input.URL); err != nil { + writeJSON(w, http.StatusBadRequest, fetchResponse{Error: err.Error()}) + return + } + + outbound, err := http.NewRequestWithContext(r.Context(), http.MethodGet, input.URL, nil) + if err != nil { + writeJSON(w, http.StatusBadRequest, fetchResponse{Error: fmt.Sprintf("invalid URL: %v", err)}) + return + } + if traceparent := r.Header.Get("traceparent"); traceparent != "" { + outbound.Header.Set("traceparent", traceparent) + } + response, err := client.Do(outbound) + if err != nil { + writeJSON(w, http.StatusBadGateway, fetchResponse{Error: fmt.Sprintf("request failed: %v", err)}) + return + } + defer response.Body.Close() + + body, err := io.ReadAll(io.LimitReader(response.Body, maxResponseBody)) + if err != nil { + writeJSON(w, http.StatusBadGateway, fetchResponse{Error: fmt.Sprintf("reading response: %v", err)}) + return + } + writeJSON(w, response.StatusCode, fetchResponse{StatusCode: response.StatusCode, Body: string(body)}) + }) + return mux +} + +func validateURL(raw string) error { + parsed, err := url.Parse(raw) + if err != nil { + return fmt.Errorf("invalid URL: %w", err) + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return fmt.Errorf("URL scheme must be http or https") + } + if parsed.Hostname() == "" { + return fmt.Errorf("URL must include a hostname") + } + return nil +} + +func writeJSON(w http.ResponseWriter, status int, response fetchResponse) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(response) +} diff --git a/demos/egress/main_test.go b/demos/egress/main_test.go new file mode 100644 index 000000000..6cd203732 --- /dev/null +++ b/demos/egress/main_test.go @@ -0,0 +1,109 @@ +// 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 main + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestFetch(t *testing.T) { + const traceparent = "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01" + client := &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + if r.Method != http.MethodGet { + t.Errorf("upstream method = %s, want GET", r.Method) + } + if got := r.Header.Get("traceparent"); got != traceparent { + t.Errorf("upstream traceparent = %q, want %q", got, traceparent) + } + return &http.Response{ + StatusCode: http.StatusTeapot, + Body: io.NopCloser(strings.NewReader("hello from upstream")), + Header: make(http.Header), + }, nil + })} + + payload, err := json.Marshal(fetchRequest{URL: "https://allowed.example/"}) + if err != nil { + t.Fatal(err) + } + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(string(payload))) + request.Header.Set("traceparent", traceparent) + newHandler(client).ServeHTTP(recorder, request) + + if recorder.Code != http.StatusTeapot { + t.Fatalf("status = %d, want %d", recorder.Code, http.StatusTeapot) + } + var got fetchResponse + if err := json.NewDecoder(recorder.Body).Decode(&got); err != nil { + t.Fatalf("decoding response: %v", err) + } + if got.StatusCode != http.StatusTeapot || got.Body != "hello from upstream" { + t.Errorf("response = %+v", got) + } +} + +func TestInvalidRequests(t *testing.T) { + tests := []struct { + name string + method string + body string + status int + }{ + {name: "method", method: http.MethodGet, body: `{}`, status: http.StatusMethodNotAllowed}, + {name: "malformed JSON", method: http.MethodPost, body: `{`, status: http.StatusBadRequest}, + {name: "missing hostname", method: http.MethodPost, body: `{"url":"https:///path"}`, status: http.StatusBadRequest}, + {name: "unsupported scheme", method: http.MethodPost, body: `{"url":"file:///etc/passwd"}`, status: http.StatusBadRequest}, + } + + handler := newHandler(http.DefaultClient) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(test.method, "/", strings.NewReader(test.body)) + handler.ServeHTTP(recorder, request) + if recorder.Code != test.status { + t.Errorf("status = %d, want %d; body = %s", recorder.Code, test.status, recorder.Body.String()) + } + }) + } +} + +func TestOutboundFailure(t *testing.T) { + client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("blocked") + })} + handler := newHandler(client) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"url":"https://example.com/"}`)) + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusBadGateway { + t.Errorf("status = %d, want %d; body = %s", recorder.Code, http.StatusBadGateway, recorder.Body.String()) + } +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return f(request) +} diff --git a/demos/egress/test-egress.sh b/demos/egress/test-egress.sh new file mode 100755 index 000000000..b24045953 --- /dev/null +++ b/demos/egress/test-egress.sh @@ -0,0 +1,186 @@ +#!/usr/bin/env bash + +# 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. + +# End-to-end test for pluggable actor egress. Reproduces: +# * POSITIVE — a real, running Actor's plain-HTTP egress is transparently +# tunneled (nftables -> atunnel -> mTLS + CONNECT) through the Envoy egress +# gateway to an in-cluster target, and the gateway's ext_proc authenticates +# the actor identity against the ate API (allowed, HTTP 200). +# * NEGATIVE — a spoofed / unknown actor identity presented directly to the +# gateway is rejected by ext_proc (HTTP 403). +# +# Prerequisites: a substrate cluster with `--deploy-demo-egress` applied, plus +# kubectl and kubectl-ate on PATH. See demos/egress/README.md. +# +# Usage: +# demos/egress/test-egress.sh # run the tests +# demos/egress/test-egress.sh --cleanup # remove everything this script created + +set -o errexit -o nounset -o pipefail + +CTX="${KUBECTL_CONTEXT:-kind-kind}" +ATESPACE="${ATESPACE:-demo}" +ACTOR="${ACTOR:-egress-demo}" +TEMPLATE="${TEMPLATE:-ate-demo-egress/egress}" +TARGET_NS="${TARGET_NS:-egress-target}" +GHOST_ACTOR="ghost-actor-does-not-exist" +PROBE_POD="egress-identity-probe" + +K="kubectl --context ${CTX}" +KATE="kubectl-ate --context ${CTX}" + +log() { printf '\n\033[1;36m== %s\033[0m\n' "$*"; } +info() { printf ' %s\n' "$*"; } +pass() { printf '\033[1;32mPASS\033[0m %s\n' "$*"; } +fail() { printf '\033[1;31mFAIL\033[0m %s\n' "$*"; FAILED=1; } +FAILED=0 + +require() { command -v "$1" >/dev/null 2>&1 || { echo "missing required tool: $1"; exit 1; }; } + +cleanup() { + log "cleanup" + ${K} -n ate-system delete pod "${PROBE_POD}" --ignore-not-found --wait=false >/dev/null 2>&1 || true + ${KATE} suspend actor "${ACTOR}" -a "${ATESPACE}" >/dev/null 2>&1 || true + ${KATE} delete actor "${ACTOR}" -a "${ATESPACE}" >/dev/null 2>&1 || true + ${K} delete namespace "${TARGET_NS}" --ignore-not-found --wait=false >/dev/null 2>&1 || true + info "done" +} + +if [[ "${1:-}" == "--cleanup" ]]; then require kubectl; require kubectl-ate; cleanup; exit 0; fi + +require kubectl +require kubectl-ate +trap '[[ "${KEEP:-}" == "1" ]] || cleanup' EXIT + +log "preflight: egress gateway (Envoy + co-located ext_proc) is running" +${K} -n ate-system rollout status deployment/atenet-egress --timeout=120s + +log "deploy an in-cluster HTTP target (whoami)" +${K} create namespace "${TARGET_NS}" >/dev/null 2>&1 || true +${K} -n "${TARGET_NS}" create deployment whoami --image=traefik/whoami >/dev/null 2>&1 || true +${K} -n "${TARGET_NS}" expose deployment whoami --port=80 >/dev/null 2>&1 || true +${K} -n "${TARGET_NS}" rollout status deployment/whoami --timeout=120s +TARGET_IP=$(${K} -n "${TARGET_NS}" get svc whoami -o jsonpath='{.spec.clusterIP}') +info "target ClusterIP = ${TARGET_IP}" + +log "create + resume Actor ${ATESPACE}/${ACTOR}" +${KATE} create atespace "${ATESPACE}" >/dev/null 2>&1 || true +${KATE} create actor "${ACTOR}" -a "${ATESPACE}" --template "${TEMPLATE}" >/dev/null 2>&1 || true +${KATE} resume actor "${ACTOR}" -a "${ATESPACE}" >/dev/null 2>&1 || true +for _ in $(seq 1 30); do + ${KATE} get actors -a "${ATESPACE}" 2>/dev/null | grep -q "STATUS_RUNNING" && break + sleep 3 +done +${KATE} get actors -a "${ATESPACE}" 2>/dev/null | grep "${ACTOR}" || true +${KATE} get actors -a "${ATESPACE}" 2>/dev/null | grep -q "STATUS_RUNNING" || { echo "actor did not reach RUNNING"; exit 1; } + +egress_log_since() { ${K} -n ate-system logs deployment/atenet-egress --tail=-1 2>/dev/null | grep '\[egress\]' | tail -n +"$(( $1 + 1 ))"; } +egress_log_count() { ${K} -n ate-system logs deployment/atenet-egress --tail=-1 2>/dev/null | grep -c '\[egress\]' || true; } +# wait_egress_log : retry for log-shipping lag. +wait_egress_log() { for _ in $(seq 1 10); do if egress_log_since "$1" | grep -qE "$2"; then egress_log_since "$1" | grep -E "$2" | tail -1; return 0; fi; sleep 1; done; return 1; } + +############################################################################## +log "POSITIVE — real Actor egress is tunneled through the gateway (expect 200)" +############################################################################## +BEFORE=$(egress_log_count) +${K} -n ate-system port-forward service/atenet-router 18099:80 >/tmp/egress-pf.log 2>&1 & +PF=$!; sleep 4 +CODE=$(curl -s -o /tmp/egress-body.txt -w '%{http_code}' -X POST http://localhost:18099/ \ + -H "Host: ${ACTOR}.${ATESPACE}.actors.resources.substrate.ate.dev" \ + -H 'Content-Type: application/json' \ + -d "{\"url\":\"http://${TARGET_IP}:80/\"}" || true) +kill "${PF}" >/dev/null 2>&1 || true +sleep 1 +info "actor round-trip HTTP ${CODE}" +GW_IP=$(${K} -n ate-system get pod -l app=atenet-egress -o jsonpath='{.items[0].status.podIP}') +if [[ "${CODE}" == "200" ]]; then pass "actor fetched the target (HTTP 200)"; else fail "expected HTTP 200, got ${CODE}"; fi +if grep -q "RemoteAddr: ${GW_IP}" /tmp/egress-body.txt 2>/dev/null; then + pass "target saw the egress gateway (${GW_IP}) as its client — traffic went through the gateway" +else + info "target body RemoteAddr: $(grep -o 'RemoteAddr: [0-9.]*' /tmp/egress-body.txt 2>/dev/null || echo '?') (gateway IP ${GW_IP})" +fi +if LINE=$(wait_egress_log "${BEFORE}" "actor=${ACTOR}.*code=200"); then + pass "gateway logged the CONNECT: ${LINE}" +else + fail "gateway did not log an allowed CONNECT for ${ACTOR}" +fi + +############################################################################## +log "NEGATIVE — spoofed/unknown actor identity is rejected by ext_proc (expect 403)" +############################################################################## +# A probe pod with a real pod-identity client cert (so mTLS succeeds) but a +# bogus X-Ate-Actor header. ext_proc's GetActor lookup fails -> 403. +${K} apply -f - >/dev/null <<'YAML' +apiVersion: v1 +kind: Pod +metadata: + name: egress-identity-probe + namespace: ate-system +spec: + containers: + - name: curl + image: curlimages/curl:latest + command: ["sleep", "600"] + volumeMounts: + - { name: podidentity, mountPath: /run/podidentity.podcert.ate.dev, readOnly: true } + - { name: servicedns, mountPath: /run/servicedns.podcert.ate.dev, readOnly: true } + volumes: + - name: podidentity + projected: + sources: + - podCertificate: + signerName: podidentity.podcert.ate.dev/identity + keyType: ECDSAP256 + credentialBundlePath: credential-bundle.pem + - name: servicedns + projected: + sources: + - clusterTrustBundle: + signerName: servicedns.podcert.ate.dev/identity + labelSelector: { matchLabels: { podcert.ate.dev/canarying: live } } + path: trust-bundle.pem +YAML +${K} -n ate-system wait --for=condition=Ready pod/${PROBE_POD} --timeout=60s >/dev/null + +BEFORE=$(egress_log_count) +# A denied CONNECT does not populate %{http_code}; %{http_connect} carries the +# proxy's CONNECT response code (403). curl exits non-zero on a failed tunnel. +CODE=$(${K} -n ate-system exec ${PROBE_POD} -- sh -c "curl -s -o /dev/null -w '%{http_connect}' \ + --proxy-cacert /run/servicedns.podcert.ate.dev/trust-bundle.pem \ + --proxy-cert /run/podidentity.podcert.ate.dev/credential-bundle.pem \ + --proxy-key /run/podidentity.podcert.ate.dev/credential-bundle.pem \ + --proxy-header 'X-Ate-Atespace: ${ATESPACE}' \ + --proxy-header 'X-Ate-Actor: ${GHOST_ACTOR}' \ + --proxy-header 'X-Ate-Actor-Version: 1' \ + --proxytunnel -x https://atenet-egress.ate-system.svc:443 http://${TARGET_IP}:80/ || true") +info "spoofed-identity CONNECT returned HTTP ${CODE:-000}" +if [[ "${CODE}" == "403" ]]; then + pass "spoofed identity rejected with 403" +else + fail "expected HTTP 403 for spoofed identity, got ${CODE:-000}" +fi +if LINE=$(wait_egress_log "${BEFORE}" "actor=${GHOST_ACTOR}.*code=403"); then + pass "gateway logged the denial: ${LINE}" +else + info "gateway egress log (recent): $(egress_log_since "${BEFORE}" | tail -1)" +fi + +echo +if [[ "${FAILED}" == "0" ]]; then + printf '\033[1;32mALL CHECKS PASSED\033[0m — pluggable egress + identity authentication working.\n' +else + printf '\033[1;31mSOME CHECKS FAILED\033[0m\n'; exit 1 +fi diff --git a/hack/install-ate.sh b/hack/install-ate.sh index 6db16f54b..b405878f9 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -40,6 +40,7 @@ ATE_DEMOS=() # Include demos. source "${ROOT}"/hack/install-demo-counter.sh +source "${ROOT}"/hack/install-demo-egress.sh source "${ROOT}"/hack/install-demo-sandbox.sh source "${ROOT}"/hack/install-demo-claude-code-multiplex.sh source "${ROOT}"/hack/install-demo-multi-template.sh @@ -327,7 +328,7 @@ deploy_ate_system() { run_kubectl rollout status deployment/ate-api-server -n ate-system --timeout=120s run_kubectl rollout status deployment/ate-controller -n ate-system --timeout=120s run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s - run_kubectl rollout status deployment/ateway-egress -n ate-system --timeout=120s + run_kubectl rollout status deployment/atenet-egress -n ate-system --timeout=120s run_kubectl rollout status statefulset/valkey-cluster -n ate-system --timeout=120s run_kubectl rollout status daemonset/atelet -n ate-system --timeout=120s } @@ -391,15 +392,10 @@ deploy_atenet() { && run_kubectl wait --for=jsonpath='{.status.phase}'=Active namespace/ate-system --timeout=60s run_ko apply -f manifests/ate-install/atenet-router.yaml - run_ko apply -f manifests/ate-install/ateway-egress.yaml + run_ko apply -f manifests/ate-install/atenet-egress.yaml run_ko apply -f manifests/ate-install/atenet-dns.yaml run_kubectl rollout status deployment/atenet-router -n ate-system --timeout=120s - run_kubectl rollout status deployment/ateway-egress -n ate-system --timeout=120s - # SEE(lior): this branch also added a `deployment/atenet-dns` wait here, which - # main has since fixed to `deployment/dns` (the Deployment in atenet-dns.yaml - # is named "dns"; every other resource in that file is "atenet-dns", so the - # old name was NotFound on every successful deploy). Dropped this branch's - # duplicate in favour of main's corrected line. + run_kubectl rollout status deployment/atenet-egress -n ate-system --timeout=120s run_kubectl rollout status deployment/dns -n ate-system --timeout=120s } @@ -514,7 +510,7 @@ delete_ate_system() { delete_atenet() { log_step "delete_atenet" run_kubectl delete --ignore-not-found -f manifests/ate-install/atenet-router.yaml - run_kubectl delete --ignore-not-found -f manifests/ate-install/ateway-egress.yaml + run_kubectl delete --ignore-not-found -f manifests/ate-install/atenet-egress.yaml run_kubectl delete --ignore-not-found -f manifests/ate-install/atenet-dns.yaml } diff --git a/hack/install-demo-egress.sh b/hack/install-demo-egress.sh new file mode 100644 index 000000000..e4e0074a9 --- /dev/null +++ b/hack/install-demo-egress.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash + +# 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. +# +# This is sourced as part of install-ate.sh. Do not run directly. + +ATE_DEMOS+=(demo-egress) # register demo-egress + +demo-egress_cmdline() { + case "${1}" in + --deploy-demo-egress) demo-egress_deploy ;; + --delete-demo-egress) demo-egress_delete ;; + *) + return 1 + ;; + esac + return 0 +} + +demo-egress_deploy() { + log_step "demo-egress_deploy" + ensure_crds + sed "s|\${BUCKET_NAME}|${BUCKET_NAME}|g" demos/egress/egress.yaml.tmpl \ + | run_ko apply -f - + + log_step "Waiting for egress demo to be ready..." + # The WorkerPool controller names the Deployment after the WorkerPool + # ("egress"), the same way demo-counter gets "deployment/counter". The old + # "egress-deployment" name was NotFound on every successful deploy. + run_kubectl rollout status deployment/egress -n ate-demo-egress --timeout=300s + run_kubectl wait --for=condition=Ready actortemplate/egress -n ate-demo-egress --timeout=300s +} + +demo-egress_delete() { + log_step "demo-egress_delete" + delete_demo_actors ate-demo-egress egress + sed "s|\${BUCKET_NAME}|${BUCKET_NAME}|g" demos/egress/egress.yaml.tmpl \ + | run_kubectl delete --ignore-not-found -f - +} diff --git a/hack/verify-egress-demo.sh b/hack/verify-egress-demo.sh new file mode 100644 index 000000000..b2211988f --- /dev/null +++ b/hack/verify-egress-demo.sh @@ -0,0 +1,65 @@ +#!/usr/bin/env bash + +# 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. +# +# Preconditions: +# hack/create-kind-cluster.sh +# hack/install-ate-kind.sh --deploy-demo-egress +# +# The egress demo Actor accepts {"url":"..."} and performs an HTTP GET. With +# egress turned on (atelet --egress-gateway-address), the Actor's outbound TCP +# is nftables-REDIRECTed into atunnel, wrapped in mTLS + HTTP CONNECT, and sent +# to the Envoy egress gateway, which terminates CONNECT and tunnels to the real +# destination. This script drives that path and shows the gateway's access log +# proving the actor identity + CONNECT authority were seen. +set -o errexit -o nounset -o pipefail +ROOT="$(git rev-parse --show-toplevel)"; cd "${ROOT}" + +CTX="${KUBECTL_CONTEXT:-kind-kind}" +K="kubectl --context ${CTX}" +ATESPACE="${ATESPACE:-demo}" +ACTOR="${ACTOR:-egress-demo}" +TARGET_URL="${TARGET_URL:-http://example.com/}" + +echo "== gateway should be running ==" +${K} -n ate-system rollout status deployment/atenet-egress --timeout=120s + +echo "== create atespace + actor ==" +kubectl-ate --context "${CTX}" create atespace "${ATESPACE}" 2>/dev/null || true +kubectl-ate --context "${CTX}" create actor "${ACTOR}" \ + --atespace "${ATESPACE}" --template ate-demo-egress/egress 2>/dev/null || true +${K} -n ate-system wait --for=condition=Ready "actor/${ACTOR}" 2>/dev/null || sleep 10 + +echo "== snapshot gateway log offset ==" +BEFORE=$(${K} -n ate-system logs deployment/atenet-egress --tail=-1 2>/dev/null | wc -l | tr -d ' ') + +echo "== drive actor egress: GET ${TARGET_URL} via the actor ==" +${K} -n ate-system port-forward service/atenet-router 18000:80 >/tmp/pf.log 2>&1 & +PF=$!; trap 'kill ${PF} 2>/dev/null || true' EXIT +sleep 3 +RESP=$(curl -s -o /dev/null -w "%{http_code}" -X POST http://localhost:18000/ \ + -H "Host: ${ACTOR}.${ATESPACE}.actors.resources.substrate.ate.dev" \ + -H 'Content-Type: application/json' \ + -d "{\"url\":\"${TARGET_URL}\"}") || true +echo "actor round-trip HTTP ${RESP} (200 = the actor fetched ${TARGET_URL} through egress)" + +echo "== NEW egress gateway access log lines (proof of CONNECT+mTLS+identity) ==" +${K} -n ate-system logs deployment/atenet-egress --tail=-1 2>/dev/null \ + | tail -n +"$((BEFORE + 1))" | grep '\[egress\]' || { + echo "!! no [egress] lines — dumping recent gateway logs:" + ${K} -n ate-system logs deployment/atenet-egress --tail=20 + exit 1 + } +echo "== PASS: actor egress traversed the Envoy egress gateway ==" diff --git a/internal/e2e/router_client.go b/internal/e2e/router_client.go index 058e58852..f168e397e 100644 --- a/internal/e2e/router_client.go +++ b/internal/e2e/router_client.go @@ -15,8 +15,10 @@ package e2e import ( + "bytes" "context" "fmt" + "io" "net/http" "time" @@ -31,7 +33,7 @@ const ( routerService = "atenet-router" ) -// RouterClient sends HTTP requests to actors through the atenet router, the +// RouterClient sends HTTP requests to actors through the ingress atenet-router, the // same way real traffic arrives (so the request is routed and, if needed, the // actor is resumed). It port-forwards the router Service, mirroring the // approach in internal/ateclient. @@ -41,7 +43,7 @@ type RouterClient struct { stop func() } -// NewRouterClient establishes a port-forward to the atenet router. Call Close +// NewRouterClient establishes a port-forward to the ingress atenet-router. Call Close // to tear it down. func NewRouterClient(ctx context.Context) (*RouterClient, error) { config, err := ateclient.LoadConfig(KubeConfig, KubeContext) @@ -73,10 +75,23 @@ func (c *RouterClient) Close() { // Get issues GET path to actor through the router, setting the actor's mesh Host // so the router routes (and resumes) it. The caller must close the body. func (c *RouterClient) Get(ctx context.Context, actorRef resources.ActorRef, path string) (*http.Response, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil) + return c.request(ctx, http.MethodGet, actorRef, path, nil) +} + +// PostJSON issues a POST with a JSON body to an Actor through the router. The +// caller must close the response body. +func (c *RouterClient) PostJSON(ctx context.Context, actorRef resources.ActorRef, path string, body []byte) (*http.Response, error) { + return c.request(ctx, http.MethodPost, actorRef, path, bytes.NewReader(body)) +} + +func (c *RouterClient) request(ctx context.Context, method string, actorRef resources.ActorRef, path string, body io.Reader) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, body) if err != nil { return nil, err } + if method == http.MethodPost { + req.Header.Set("Content-Type", "application/json") + } // The router routes on the Host/:authority, not a header. req.Host = actorRef.DNSName() return c.http.Do(req) diff --git a/internal/e2e/router_client_test.go b/internal/e2e/router_client_test.go new file mode 100644 index 000000000..d9be88241 --- /dev/null +++ b/internal/e2e/router_client_test.go @@ -0,0 +1,78 @@ +// 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 e2e + +import ( + "context" + "io" + "net/http" + "strings" + "testing" + + "github.com/agent-substrate/substrate/internal/resources" +) + +// SEE(lior): this file used to also hold TestResolveHTTPTargetPort and +// TestIsPodReady. Main moved both into internal/portforward/portforward_test.go +// (upstream 4453b5e7, "Prefactoring: Consolidate logic to port-forward to a +// service Pod") and deleted this file, so git rename-matched this branch's +// version of it against the new portforward test and reported the whole thing as +// a conflict. Took main's move as-is and re-added only the net-new PostJSON test +// here, rather than resurrecting the port-forward tests in a package that no +// longer owns that code. +func TestRouterClientPostJSON(t *testing.T) { + client := &RouterClient{ + baseURL: "http://router.test", + http: &http.Client{Transport: testRoundTripper(func(request *http.Request) (*http.Response, error) { + if request.Method != http.MethodPost { + t.Errorf("method = %q, want POST", request.Method) + } + if request.Host != "fetcher.demo.actors.resources.substrate.ate.dev" { + t.Errorf("host = %q", request.Host) + } + if request.URL.Path != "/fetch" { + t.Errorf("path = %q, want /fetch", request.URL.Path) + } + if request.Header.Get("Content-Type") != "application/json" { + t.Errorf("content type = %q, want application/json", request.Header.Get("Content-Type")) + } + body, err := io.ReadAll(request.Body) + if err != nil { + t.Fatalf("reading body: %v", err) + } + if string(body) != `{"url":"https://example.com/"}` { + t.Errorf("body = %q", body) + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("ok")), + Header: make(http.Header), + }, nil + })}, + } + + actorRef := resources.ActorRef{Atespace: "demo", Name: "fetcher"} + response, err := client.PostJSON(context.Background(), actorRef, "/fetch", []byte(`{"url":"https://example.com/"}`)) + if err != nil { + t.Fatalf("PostJSON: %v", err) + } + response.Body.Close() +} + +type testRoundTripper func(*http.Request) (*http.Response, error) + +func (f testRoundTripper) RoundTrip(request *http.Request) (*http.Response, error) { + return f(request) +} diff --git a/internal/e2e/suites/networking/networking_test.go b/internal/e2e/suites/networking/networking_test.go index 5f3c16866..e853d09f6 100644 --- a/internal/e2e/suites/networking/networking_test.go +++ b/internal/e2e/suites/networking/networking_test.go @@ -29,9 +29,24 @@ import ( const networkingAtespace = "networking-e2e" +// actorTemplate identifies a demo ActorTemplate to build test Actors from, +// along with the hack/install-ate.sh flag that deploys it. +type actorTemplate struct { + namespace string + name string + // deployFlag names the install flag that creates the template, so a + // missing fixture reports how to fix it rather than just failing. + deployFlag string +} + +var ( + counterTemplate = actorTemplate{namespace: "ate-demo-counter", name: "counter", deployFlag: "--deploy-demo-counter"} + egressTemplate = actorTemplate{namespace: "ate-demo-egress", name: "egress", deployFlag: "--deploy-demo-egress"} +) + func TestActorDirectAccess(t *testing.T) { ctx := context.Background() - actorName, actor := createAndResumeActor(t, ctx, "direct") + actorName, actor := createAndResumeActor(t, ctx, "direct", counterTemplate) router := mustRouterClient(t, ctx) defer router.Close() @@ -56,7 +71,38 @@ func TestActorDirectAccess(t *testing.T) { }) } -func createAndResumeActor(t *testing.T, ctx context.Context, prefix string) (string, *ateapipb.Actor) { +// TestActorEgress exercises the full egress path. The Actor's outbound TCP +// connection is transparently redirected by nftables into atunnel, wrapped in +// mTLS + HTTP CONNECT to atenet-egress, authorized there against the Actor's +// asserted identity, and only then dialed out. A masqueraded (pre-gateway) +// egress would also return 200, so this asserts the gateway is deployed and +// that it did not reject the Actor. +func TestActorEgress(t *testing.T) { + ctx := context.Background() + actorName, _ := createAndResumeActor(t, ctx, "egress", egressTemplate) + router := mustRouterClient(t, ctx) + defer router.Close() + + // The egress demo fetches the URL it is given and echoes the upstream + // status and body back. + payload := []byte(`{"url":"http://example.com/"}`) + actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} + response, err := router.PostJSON(ctx, actorRef, "/", payload) + if err != nil { + t.Fatalf("POST to egress Actor through ingress: %v", err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("reading egress response body (HTTP %d): %v", response.StatusCode, err) + } + if response.StatusCode != http.StatusOK { + t.Fatalf("Actor egress fetch returned HTTP %d, want 200; body: %s", response.StatusCode, body) + } + t.Logf("Actor egress fetch succeeded; body: %s", body) +} + +func createAndResumeActor(t *testing.T, ctx context.Context, prefix string, template actorTemplate) (string, *ateapipb.Actor) { t.Helper() clients := e2e.GetClients() actorName := fmt.Sprintf("%s-%d", prefix, time.Now().UnixNano()) @@ -68,10 +114,10 @@ func createAndResumeActor(t *testing.T, ctx context.Context, prefix string) (str }) if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: networkingAtespace, Name: actorName}, - ActorTemplateNamespace: "ate-demo-counter", - ActorTemplateName: "counter", + ActorTemplateNamespace: template.namespace, + ActorTemplateName: template.name, }}); err != nil { - t.Fatalf("CreateActor: %v (deploy the fixture with --deploy-demo-counter)", err) + t.Fatalf("CreateActor from %s/%s: %v (deploy the fixture with %s)", template.namespace, template.name, err, template.deployFlag) } t.Cleanup(func() { _, _ = clients.SubstrateAPI.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{Actor: actorRef}) diff --git a/manifests/ate-install/atelet.yaml b/manifests/ate-install/atelet.yaml index 43aa1e44d..bc8a93a54 100644 --- a/manifests/ate-install/atelet.yaml +++ b/manifests/ate-install/atelet.yaml @@ -71,13 +71,10 @@ spec: - --gcp-auth-for-image-pulls=true - --grpc-server-cred-bundle=/run/podidentity.podcert.ate.dev/credential-bundle.pem - --client-ca-certs=/run/podidentity.podcert.ate.dev/trust-bundle.pem - # SEE(lior): purely additive against main's new mTLS flags — both sides - # kept. This is the one line that arms the nftables egress REDIRECT that - # shipped dormant with atunnel. - # ONLY UNTIL API IS DECIDED, FOR POC: turn on pluggable actor egress cluster-wide. Actor TCP egress is + # ONLY UNTIL EGRESS API IS DECIDED, FOR POC: turn on pluggable actor egress cluster-wide. Actor TCP egress is # transparently redirected (nftables) into atunnel, which wraps it in # mTLS + HTTP CONNECT to the an egress gateway. - - --egress-gateway-address=ateway-egress.ate-system.svc:443 + - --egress-gateway-address=atenet-egress.ate-system.svc:443 # atelet does no mounts, netlink, device, or namespace operations (those # live in the ateom worker pod) — it only reads/writes the # /var/lib/ateom-gvisor hostPath as root, so it needs no Linux diff --git a/manifests/ate-install/ateway-egress.yaml b/manifests/ate-install/atenet-egress.yaml similarity index 87% rename from manifests/ate-install/ateway-egress.yaml rename to manifests/ate-install/atenet-egress.yaml index 1c6a4a382..7fa9f4e03 100644 --- a/manifests/ate-install/ateway-egress.yaml +++ b/manifests/ate-install/atenet-egress.yaml @@ -20,7 +20,7 @@ apiVersion: v1 kind: ServiceAccount metadata: - name: ateway-egress + name: atenet-egress namespace: ate-system --- # RBAC for the co-located ext_proc sidecar (atenet router), which watches @@ -28,7 +28,7 @@ metadata: apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: - name: ateway-egress + name: atenet-egress rules: - apiGroups: - "ate.dev" @@ -42,20 +42,20 @@ rules: apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: - name: ateway-egress + name: atenet-egress subjects: - kind: ServiceAccount - name: ateway-egress + name: atenet-egress namespace: ate-system roleRef: kind: ClusterRole - name: ateway-egress + name: atenet-egress apiGroup: rbac.authorization.k8s.io --- apiVersion: v1 kind: ConfigMap metadata: - name: ateway-egress + name: atenet-egress namespace: ate-system data: envoy.yaml: | @@ -68,7 +68,11 @@ data: address: socket_address: { address: 0.0.0.0, port_value: 443 } filter_chains: - - transport_socket: + # Named so ext_proc can read it back as xds.filter_chain_name. Must + # match EgressFilterChainName in + # cmd/atenet/internal/router/extproc_egress.go. + - name: egress + transport_socket: name: envoy.transport_sockets.tls typed_config: "@type": type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.DownstreamTlsContext @@ -103,11 +107,6 @@ data: "@type": type.googleapis.com/envoy.extensions.access_loggers.stream.v3.StdoutAccessLog log_format: text_format_source: - # SEE(lior): actor= reads X-ATE-ACTOR-NAME, not X-ATE-ACTOR. - # atunnel's ActorNameHeader was renamed during PR review - # after this manifest was written; the Go handlers use the - # constant so they followed the rename, but this literal - # did not and was logging an empty actor. inline_string: "[egress] authority=%REQ(:AUTHORITY)% atespace=%REQ(X-ATE-ATESPACE)% actor=%REQ(X-ATE-ACTOR-NAME)% ver=%REQ(X-ATE-ACTOR-VERSION)% auth_present=%REQ(AUTHORIZATION)% code=%RESPONSE_CODE% flags=%RESPONSE_FLAGS% up_bytes=%BYTES_RECEIVED% down_bytes=%BYTES_SENT%\n" route_config: name: connect_route @@ -136,13 +135,14 @@ data: failure_mode_allow: false # How the ext_proc server tells egress from ingress. It applies # opposite trust models to the two directions, and dispatches on - # this Envoy-asserted listener name so that no client can select - # the egress path by crafting a request. The value must match - # EgressListenerName in cmd/atenet/internal/router/extproc_egress.go; - # renaming the listener above without updating it fails closed - # (egress requests take the ingress path and 404). + # this Envoy-asserted filter chain name so that no client can + # select the egress path by crafting a request. The value must + # match EgressFilterChainName in + # cmd/atenet/internal/router/extproc_egress.go; renaming the + # filter chain above without updating it fails closed (egress + # requests take the ingress path and 404). request_attributes: - - xds.listener_name + - xds.filter_chain_name processing_mode: request_header_mode: SEND response_header_mode: SKIP @@ -197,21 +197,21 @@ data: apiVersion: apps/v1 kind: Deployment metadata: - name: ateway-egress + name: atenet-egress namespace: ate-system labels: - app: ateway-egress + app: atenet-egress spec: replicas: 1 selector: matchLabels: - app: ateway-egress + app: atenet-egress template: metadata: labels: - app: ateway-egress + app: atenet-egress spec: - serviceAccountName: ateway-egress + serviceAccountName: atenet-egress securityContext: # Allow the non-root envoy user to bind :443. sysctls: @@ -231,9 +231,9 @@ spec: - -c - /etc/envoy/envoy.yaml - --service-node - - ateway-egress + - atenet-egress - --service-cluster - - ateway-egress + - atenet-egress ports: - name: https containerPort: 443 @@ -272,8 +272,9 @@ spec: - --namespace=ate-system - --port-extproc=50051 - --extproc-address=127.0.0.1 - - --ateapi-address=api.ate-system.svc:443 - - --ateapi-auth=mtls + - --ateapi-address=dns:///api.ate-system.svc:443 + - --ateapi-ca-file=/run/servicedns.podcert.ate.dev/trust-bundle.pem + - --ateapi-client-cert=/run/podidentity.podcert.ate.dev/credential-bundle.pem - --otlp-collector-address= env: - name: POD_NAME @@ -291,10 +292,19 @@ spec: tcpSocket: port: extproc periodSeconds: 10 + volumeMounts: + # Trust bundle used to verify ateapi's servicedns serving cert. + - name: servicedns + mountPath: /run/servicedns.podcert.ate.dev + readOnly: true + # ext-proc's own client identity presented to ateapi. + - name: podidentity + mountPath: /run/podidentity.podcert.ate.dev + readOnly: true volumes: - name: config configMap: - name: ateway-egress + name: atenet-egress - name: servicedns projected: sources: @@ -325,12 +335,12 @@ spec: apiVersion: v1 kind: Service metadata: - name: ateway-egress + name: atenet-egress namespace: ate-system spec: type: ClusterIP selector: - app: ateway-egress + app: atenet-egress ports: - name: https port: 443 diff --git a/manifests/ate-install/kind/atelet/kustomization.yaml b/manifests/ate-install/kind/atelet/kustomization.yaml index c7c87d3d1..3d2e7d450 100644 --- a/manifests/ate-install/kind/atelet/kustomization.yaml +++ b/manifests/ate-install/kind/atelet/kustomization.yaml @@ -38,11 +38,7 @@ patches: # Kind clusters are dev/CI: run atelet at debug so e2e suites can # assert on per-item log lines. Production installs default to info. - --log-level=debug - # SEE(lior): both sides kept. A strategic-merge patch replaces the - # whole args list rather than merging it, so dropping either side's - # flags here would silently unset them on the kind DaemonSet. - # POC: turn on pluggable actor egress cluster-wide. - - --egress-gateway-address=ateway-egress.ate-system.svc:443 + - --egress-gateway-address=atenet-egress.ate-system.svc:443 env: - name: OTEL_EXPORTER_OTLP_ENDPOINT value: http://opentelemetry-collector.otel-system.svc:4317 diff --git a/manifests/ate-install/kind/kustomization.yaml b/manifests/ate-install/kind/kustomization.yaml index 1e264e670..f50d3a2b2 100644 --- a/manifests/ate-install/kind/kustomization.yaml +++ b/manifests/ate-install/kind/kustomization.yaml @@ -20,7 +20,7 @@ resources: - ../ate-controller.yaml - ./atelet - ../atenet-dns.yaml - - ../ateway-egress.yaml + - ../atenet-egress.yaml - ../atenet-router.yaml - ../valkey.yaml - ../pod-certificate-controller.yaml diff --git a/manifests/ate-install/token-client/kustomization.yaml b/manifests/ate-install/token-client/kustomization.yaml index d35743f81..37fd9a5a1 100644 --- a/manifests/ate-install/token-client/kustomization.yaml +++ b/manifests/ate-install/token-client/kustomization.yaml @@ -23,7 +23,7 @@ resources: - ../ate-controller.yaml - ../atelet.yaml - ../atenet-dns.yaml - - ../ateway-egress.yaml + - ../atenet-egress.yaml - ../atenet-router.yaml - ../valkey.yaml - ../pod-certificate-controller.yaml From c133167631114b2f87cd78da2ac953919baa6409 Mon Sep 17 00:00:00 2001 From: Lior Lieberman Date: Wed, 5 Aug 2026 11:42:24 -0700 Subject: [PATCH 3/3] fixes --- cmd/atelet/main.go | 5 +---- cmd/atenet/internal/router/extproc_egress.go | 9 --------- 2 files changed, 1 insertion(+), 13 deletions(-) diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index d7780db40..517ef8c27 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -76,10 +76,7 @@ var ( localhostRegistryReplacement = pflag.String("localhost-registry-replacement", "", "The replacement registry endpoint for localhost and/or loopback IP addresses, useful for local development. for example kind-registry:5000") imageCacheDir = pflag.String("image-cache-dir", ateompath.ImageCacheDir, "Directory for the node-local OCI image layer cache. Must be on the volume shared with the ateom pods (the cached layers are their overlay lowerdirs), and on a disk sized for both capacity and IOPS: unpack throughput is gated by the volume's IOPS.") - // SEE(lior): both sides kept — main added --log-level here while this branch - // added --egress-gateway-address. Independent flags, no interaction. - // - // egressGatewayAddress turns on pluggable actor egress cluster-wide (POC). + // egressGatewayAddress turns on pluggable actor egress cluster-wide. // When set, actors whose Run/Restore request does not already carry an egress // gateway address have this address injected, causing ateom to redirect actor // TCP egress through atunnel to the egress gateway. Empty keeps egress off. diff --git a/cmd/atenet/internal/router/extproc_egress.go b/cmd/atenet/internal/router/extproc_egress.go index 2a19a347e..50d1ae0ad 100644 --- a/cmd/atenet/internal/router/extproc_egress.go +++ b/cmd/atenet/internal/router/extproc_egress.go @@ -38,15 +38,6 @@ const ( // FilterChainNameAttribute is the CEL attribute carrying the name of the // filter chain that accepted the request. The egress Envoy asks for it via // request_attributes on its ext_proc filter. - // - // SEE(lior): this was xds.listener_name, which reads more naturally but - // which Envoy 1.34 cannot parse: it logs "error parsing cel expression - // xds.listener_name" at trace level, then sends the ProcessingRequest with - // an empty attributes map rather than failing config load. Because an - // absent attribute means "ingress" (the fail-safe direction), every egress - // CONNECT silently took the ingress path and 404'd on the actor DNS name - // parse. xds.filter_chain_name parses on the same Envoy build and is - // equally Envoy-asserted, so the trust model is unchanged. FilterChainNameAttribute = "xds.filter_chain_name" )