-
Notifications
You must be signed in to change notification settings - Fork 200
egress: add egress gateway support - atenet-egress #693
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -96,7 +96,16 @@ 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 | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Let's factor this so you could run it in this way, but in a real deployment, I think you would keep them as separate deployments as they will likely have different scaling parameters.
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Comment is confusing, they are currently separate deployments but egress and ingress is supported with the same atenet binary (different extproc handlers) in the same binary. I will reword the comment and will also make a mode to be able to launch the atenet in ingress only or egress only mode. |
||
| // CONNECT requests and ingress requests. Which one is decided by | ||
| // the accepting listener, not by anything in the request itself | ||
| // (see isEgressRequest). | ||
| // | ||
| 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) | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 ( | ||
| // 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. | ||
| FilterChainNameAttribute = "xds.filter_chain_name" | ||
| ) | ||
|
|
||
| // isEgressRequest reports whether an ext_proc RequestHeaders callback arrived on | ||
| // 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 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 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 filterChainName(req) == EgressFilterChainName | ||
| } | ||
|
|
||
| // 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 filterChainName(req *extprocv3.ProcessingRequest) string { | ||
| for _, attrs := range req.GetAttributes() { | ||
| if v, ok := attrs.GetFields()[FilterChainNameAttribute]; 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. | ||
| // | ||
| // 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, | ||
| ) (*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) | ||
| } | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Your comment and the help text are different.
As I understand it -- this sets a default egress if not specified on the Actor resource explicitly?
Can you fix it to match up?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
not directly. We are using this flag because we have not yet designed the egress API properly and we havent wired it up yet.
When we do, we dont need any egressgateway flag
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
BTW Eitan took some of it to his identity PR so if that one ends up in first, I will remove it from here and rebase