diff --git a/benchmarking/atunnel/README.md b/benchmarking/atunnel/README.md index d1defc54ac..f22930b381 100644 --- a/benchmarking/atunnel/README.md +++ b/benchmarking/atunnel/README.md @@ -77,8 +77,8 @@ Which subcommand runs where: same order: - at startup: `tunnel.Start` with the worker's own flags, through `tunnel.RegisterFlags`; -- per actor: `tunnel.PrepareEgress`, `actornet.ServeSandbox`, and - `tunnel.Activate`; +- per actor: `tunnel.PrepareEgress`, `actornet.ServeSandbox`, + `tunnel.ActivateEgress`, and `tunnel.ActivateIngress`; - at teardown: `tunnel.Deactivate`, then closing the sandbox network. Each actor gets real network namespaces, the nftables redirect, an egress diff --git a/benchmarking/atunnel/internal/workerhost/workerhost.go b/benchmarking/atunnel/internal/workerhost/workerhost.go index ff7bc107f7..1080741a83 100644 --- a/benchmarking/atunnel/internal/workerhost/workerhost.go +++ b/benchmarking/atunnel/internal/workerhost/workerhost.go @@ -117,7 +117,10 @@ func (h *Host) host(ctx context.Context, attribution resources.ActorAttribution, h.mu.Lock() h.actors = append(h.actors, &actor{attribution: attribution, session: session}) h.mu.Unlock() - return h.tunnel.Activate(attribution, session.Dialer(), egress) + if err := h.tunnel.ActivateEgress(attribution, egress); err != nil { + return err + } + return h.tunnel.ActivateIngress(attribution, session.Dialer()) } // Info returns the hosted actors. diff --git a/cmd/atenet/internal/router/egress/egress.go b/cmd/atenet/internal/router/egress/egress.go index 207c0a6d03..4cceeecfdf 100644 --- a/cmd/atenet/internal/router/egress/egress.go +++ b/cmd/atenet/internal/router/egress/egress.go @@ -171,7 +171,7 @@ func (h *Handler) handleConnect(ctx context.Context, md *extproc.RequestMetadata "egress denied: invalid actor certificate") } - if err := h.validateActor(ctx, actorRef); err != nil { + if err := h.validateActor(ctx, actorRef, md.Host); err != nil { return extproc.Result{}, err } @@ -276,9 +276,10 @@ func (h *Handler) lookupPolicy(ctx context.Context, leg string, ref resources.Ac } // validateActor checks the actor certified by the certificate against the control -// plane's current view of that actor: it still exists and it is running. Every -// error it returns is already a client-facing ext_proc denial. -func (h *Handler) validateActor(ctx context.Context, actorRef resources.ActorRef) error { +// plane's current view of that actor: it still exists and it is placed on a +// worker. A refusal is logged with destination. Every error it returns is +// already a client-facing ext_proc denial. +func (h *Handler) validateActor(ctx context.Context, actorRef resources.ActorRef, destination string) error { // Confirm the certified actor still exists. // TODO: this can cause heavy load on ate api server. Change it based on https://github.com/agent-substrate/substrate/issues/592. actor, err := h.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ @@ -288,12 +289,19 @@ func (h *Handler) validateActor(ctx context.Context, actorRef resources.ActorRef return mapEgressIdentityError(actorRef.Atespace, actorRef.Name, err) } - // The actor performing egress must actually be running. - if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_RUNNING { + // The actor performing egress must be placed on a worker. RESUMING is + // saved together with the worker assignment, so a booting workload can + // reach the network before it answers its wakeup probe. Every other state + // has left its worker or is leaving it. + switch state := actor.GetStatus().GetState(); state { + case ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_RESUMING: + return nil + default: + slog.WarnContext(ctx, "egress denied: actor is not placed on a worker", + slog.Any("actor", actorRef), slog.String("state", state.String()), slog.String("destination", destination)) return extproc.NewReqError(envoy_type.StatusCode_Forbidden, - "egress denied: actor %q/%q is %s, not running", actorRef.Atespace, actorRef.Name, actor.GetStatus().GetState()) + "egress denied: actor %q/%q is %s, not placed on a worker", actorRef.Atespace, actorRef.Name, state) } - return nil } // authenticateActorCertificate turns the mTLS peer certificate Envoy recorded diff --git a/cmd/atenet/internal/router/egress/egress_test.go b/cmd/atenet/internal/router/egress/egress_test.go index 7f367cd3bf..62046045e3 100644 --- a/cmd/atenet/internal/router/egress/egress_test.go +++ b/cmd/atenet/internal/router/egress/egress_test.go @@ -721,7 +721,7 @@ func TestHandleRequestHeadersAuthorization(t *testing.T) { want envoy_type.StatusCode }{ { - name: "actor is not running", + name: "actor is suspended", actor: &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{ Atespace: testEgressAtespace, @@ -754,6 +754,37 @@ func TestHandleRequestHeadersAuthorization(t *testing.T) { } } +// Egress follows the actor's placement on a worker: a resuming actor is booting +// or restoring there and may fetch what it needs to become ready; every other +// state has left its worker or is leaving it. +func TestHandleRequestHeadersActorState(t *testing.T) { + ca := newTestCA(t, "actor-identity-ca") + allowed := map[ateapipb.ActorState]bool{ + ateapipb.ActorState_ACTOR_STATE_RESUMING: true, + ateapipb.ActorState_ACTOR_STATE_RUNNING: true, + } + for value := range ateapipb.ActorState_name { + state := ateapipb.ActorState(value) + if state == ateapipb.ActorState_ACTOR_STATE_UNSPECIFIED { + continue + } + t.Run(state.String(), func(t *testing.T) { + actor := runningActor() + actor.Status.State = state + h := egressHandler(ca.loadRoots, actor, nil) + leaf := ca.issueActorCert(t, "spiffe://substrate-actor.local/worker-for-actor/foo/bar", actorCertOptions{}) + _, err := h.HandleRequestHeaders(context.Background(), egressMetadata(xfccHeader(leaf))) + if allowed[state] { + if err != nil { + t.Fatalf("HandleRequestHeaders() error = %v, want nil", err) + } + return + } + wantStatus(t, err, envoy_type.StatusCode_Forbidden) + }) + } +} + // Without actor-identity roots the gateway must fail closed rather than tunnel // unauthenticated traffic. An ingress-only router has no CA configured, and a // configured bundle can become unreadable after startup. diff --git a/cmd/ateworker-gvisor/main.go b/cmd/ateworker-gvisor/main.go index 8481a61ff5..1816f424e3 100644 --- a/cmd/ateworker-gvisor/main.go +++ b/cmd/ateworker-gvisor/main.go @@ -671,6 +671,11 @@ func (s *AteWorkerService) RunWorkload(ctx context.Context, req *ateworkerpb.Run } } }() + // Egress is activated before the sandbox starts, ingress is activated after + // the wakeup probe (see tunnel.Tunnel.ActivateEgress). + if err := s.tunnel.ActivateEgress(attribution, egress); err != nil { + return nil, err + } containersToDelete = append(containersToDelete, ocispec.SandboxID) // Attribute the sandbox's runsc CLI output to the actor. pw, err := s.actorLogger.StartJSONLogPipe(attribution, ocispec.SandboxID) @@ -714,7 +719,7 @@ func (s *AteWorkerService) RunWorkload(ctx context.Context, req *ateworkerpb.Run if err := wakeupprobe.WaitAll(ctx, req.GetSpec().GetContainers(), actornet.ActorVethIP, wakeupprobe.DialFunc(s.sandboxDialer(req.GetActorUid()))); err != nil { return nil, fmt.Errorf("while waiting for container wakeup probe: %w", err) } - if err := s.tunnel.Activate(workerstats.ActorAttributionFromRequest(req), s.sandboxDialer(req.GetActorUid()), egress); err != nil { + if err := s.tunnel.ActivateIngress(attribution, s.sandboxDialer(req.GetActorUid())); err != nil { return nil, err } @@ -1084,6 +1089,11 @@ func (s *AteWorkerService) RestoreWorkload(ctx context.Context, req *ateworkerpb } } }() + err = s.tunnel.ActivateEgress(attribution, egress) + timing.activate = lap(&tLast) + if err != nil { + return nil, err + } checkpointDir := req.GetActorDirs().GetRestoreDir() restoreArgs, err := fsRestoreArgs(checkpointDir, fidelity) if err != nil { @@ -1170,8 +1180,8 @@ func (s *AteWorkerService) RestoreWorkload(ctx context.Context, req *ateworkerpb if err != nil { return nil, fmt.Errorf("while waiting for container wakeup probe: %w", err) } - err = s.tunnel.Activate(attribution, s.sandboxDialer(req.GetActorUid()), egress) - timing.activate = lap(&tLast) + err = s.tunnel.ActivateIngress(attribution, s.sandboxDialer(req.GetActorUid())) + timing.activate += lap(&tLast) if err != nil { return nil, err } diff --git a/cmd/ateworker-gvisor/phaselog.go b/cmd/ateworker-gvisor/phaselog.go index 1413099695..daa6abc6ea 100644 --- a/cmd/ateworker-gvisor/phaselog.go +++ b/cmd/ateworker-gvisor/phaselog.go @@ -57,7 +57,9 @@ const containerCountKey = "ate.actor.container.count" // restoreTiming is the elapsed time of each RestoreWorkload phase. A phase // left at zero never ran. Under VOLUMES and ROOTFS fidelity, sandboxRestore and -// appRestore time the cold start that stands in for the restore. +// appRestore time the cold start that stands in for the restore. activate is +// the total time spent activating the tunnel: egress is activated before the +// containers restore and ingress after the wakeup probe, and both are counted. type restoreTiming struct { prep, egressPrepare, netSetup time.Duration sandboxCreate, sandboxRestore time.Duration diff --git a/cmd/ateworker-microvm/restore.go b/cmd/ateworker-microvm/restore.go index 61a8bbc086..6c002c24e3 100644 --- a/cmd/ateworker-microvm/restore.go +++ b/cmd/ateworker-microvm/restore.go @@ -458,9 +458,18 @@ func (s *AteWorkerService) restoreMemoryFidelity(ctx context.Context, p actorBoo // and notifies the guest before unpausing the vCPUs, so a >=5.18 kernel reseeds its // CRNG on its own with no host round-trip, no per-restore RPC, and no post-Resume // race. At that point this agent-driven reseed can be dropped. + // + // Egress opens only once the reseed succeeds, so nothing leaves the host drawn + // from CRNG state the actor shares with the snapshot's other clones, and a + // failed reseed is still safe to retry (see restoreGuestReadyError). The + // restored workload is past its boot, so it needs egress only from the wakeup + // probe on. Ingress follows the wakeup probe, as in coldBootActor. if err := reseedGuestCRNG(ctx, guestAC); err != nil { return restoreGuestReadyError(ctx, "reseeding guest CRNG", err) } + if err := s.tunnel.ActivateEgress(p.attribution(), egress); err != nil { + return err + } tReseeded := time.Now() // Block until every wakeup-probe-enabled container reports 200. @@ -524,7 +533,7 @@ func (s *AteWorkerService) restoreMemoryFidelity(ctx context.Context, p actorBoo s.startActorLogForwarding(guestAC, attribution, c.id, c.name) } - if err := s.tunnel.Activate(p.attribution(), s.sandboxDialer(p.actorUID), egress); err != nil { + if err := s.tunnel.ActivateIngress(p.attribution(), s.sandboxDialer(p.actorUID)); err != nil { return err } s.setRunningVM(actorUID, ra) @@ -585,8 +594,9 @@ func (s *AteWorkerService) cleanupFailedRestoreRootfs(ctx context.Context, actor // the agent is dialed and the CRNG reseeded (up to about 15s plus // reseedTimeout). Nothing routes inbound traffic to it yet (not even the wakeup // probe), and atunnel's egress proxy closes its outbound TCP connections until -// tunnel.Activate, but a background task in the guest can query the DNS relay -// or write to its filesystems. The retry re-extracts the rootfs upper and +// tunnel.ActivateEgress, which restoreMemoryFidelity calls after the reseed for +// that reason, but a background task in the guest can query the DNS relay or +// write to its filesystems. The retry re-extracts the rootfs upper and // durable-dir volumes from the snapshot, whereas a CSI volume is outside the // snapshot and keeps any writes made during that window while guest memory and // its page cache resume from the snapshot; we accept that narrow window over diff --git a/cmd/ateworker-microvm/run.go b/cmd/ateworker-microvm/run.go index 33f5bd913e..5b01caf67b 100644 --- a/cmd/ateworker-microvm/run.go +++ b/cmd/ateworker-microvm/run.go @@ -441,6 +441,12 @@ func (s *AteWorkerService) coldBootActor(ctx context.Context, p actorBootParams) } }() + // Egress before the guest boots, ingress after the wakeup probe (see + // tunnel.Tunnel.ActivateEgress). + if err := s.tunnel.ActivateEgress(p.attribution(), egress); err != nil { + return err + } + kparams := s.kernelParams() // Size the VM to the actor's declared limits (see internal/sizing), which the @@ -604,7 +610,7 @@ func (s *AteWorkerService) coldBootActor(ctx context.Context, p actorBootParams) slog.Duration("since_boot", time.Since(tBooted))) ra := &runningActor{chProc: chProc, vfsdProc: vfsdProc, apiSocket: apiSocket, baseID: actorUID, guestAgent: ac, workloadIDs: workloadIDs(ctrs)} - if err := s.tunnel.Activate(p.attribution(), s.sandboxDialer(p.actorUID), egress); err != nil { + if err := s.tunnel.ActivateIngress(p.attribution(), s.sandboxDialer(p.actorUID)); err != nil { return err } s.setRunningVM(actorUID, ra) diff --git a/docs/api-guide.md b/docs/api-guide.md index 563a6a519f..e2584b2f8c 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -343,6 +343,7 @@ How it behaves: - **When it runs.** On every `ResumeActor` that actually wakes the actor. A `ResumeActor` on an actor that is already `RUNNING` is a no-op and does not probe. - **Block-until-ready semantics.** `ResumeActor` returns successfully only after every container with a `wakeupProbe` has returned HTTP 200. If any container has not done so within `timeoutSeconds` (30s by default), `ResumeActor` fails and the actor moves to `ACTOR_STATE_CRASHED`, as it does for any other error while waking. A crashed actor cannot be resumed directly: `ResumeActor` rejects it with `FAILED_PRECONDITION`. Call `RevertActor` first, which returns it to `ACTOR_STATE_SUSPENDED` with its last snapshot, then resume it again. - **Aggressive polling.** The poll loop is tuned for single-millisecond detection latency: a keep-alive HTTP client with a 1ms interval and 250ms per-request timeout. While the workload is still booting, kernel `RST`s return in microseconds, so the loop spends almost no time blocked; once the listener is up, the next attempt completes on veth-local latency. +- **Egress while waking.** The actor's egress is active before its containers start, so a workload can download what it needs to become ready (a model, a Git repository) before it answers the probe. A micro-VM restored from a memory snapshot is already past its boot, and gets egress just before the probe starts. Its EgressPolicy applies as it does once the actor is `RUNNING`; inbound traffic still waits for the probe. A wake that fails or is cut short after egress is active can be retried from the same snapshot, so outbound requests made while waking can be sent more than once. - **Golden snapshot warm-up shortcut.** When **every** container in a template declares `wakeupProbe`, the actor template controller skips its default ~20s "give the workload time to settle" delay before taking the golden snapshot — `ResumeActor` already blocked until the workload reported 200, so the workload is known to be initialized. Templates that omit `wakeupProbe` on any container keep the 20s warm-up as a safety net. - **Snapshot/restore interaction.** The TCP listener is part of the checkpointed RAM, so on resume `wakeupProbe` typically returns 200 on the first attempt, with no observable latency penalty. diff --git a/docs/network-egress.md b/docs/network-egress.md index 94344e9072..5400c1cede 100644 --- a/docs/network-egress.md +++ b/docs/network-egress.md @@ -24,9 +24,9 @@ In the default Kubernetes deployment, the PEP obtains its certificate and privat ## Client Authorization -`atunnel` MUST present an actor-specific client certificate when connecting to the PEP. At actor activation, `atunnel` generates a private key and requests a short-lived certificate from substrate through the node-local atelet. The private key remains in `atunnel`. The certificate identifies the actor by its atespace, name, and UID and is scoped to the `atunnel` purpose. +`atunnel` MUST present an actor-specific client certificate when connecting to the PEP. Before the actor's workload starts, `atunnel` generates a private key and requests a short-lived certificate from substrate through the node-local atelet, and egress is active from then on: a workload may need the network to become ready, so egress does not wait for the wakeup probe the way ingress does. The private key remains in `atunnel`. The certificate identifies the actor by its atespace, name, and UID and is scoped to the `atunnel` purpose. -The PEP MUST require and verify the client certificate before accepting a CONNECT request. It MUST verify the certificate chain and lifetime against the Actor Identity CA, require the client authentication usage, and require exactly one valid `ActorIdentity` extension whose atespace, actor name, actor UID, and purpose are present. The current extension OID is `1.3.6.1.4.1.11129.2.12.2`; it is allocated under Google's enterprise number and will change after the CNCF donation is complete. The purpose MUST be `atunnel`, and the certificate's actor URI SAN MUST identify the same actor as the extension. The PEP MUST then confirm with ate-api-server that the actor still exists, that its current UID matches the certificate, and that it is running. The PEP MUST reject the CONNECT request if any of these checks fail. +The PEP MUST require and verify the client certificate before accepting a CONNECT request. It MUST verify the certificate chain and lifetime against the Actor Identity CA, require the client authentication usage, and require exactly one valid `ActorIdentity` extension whose atespace, actor name, actor UID, and purpose are present. The current extension OID is `1.3.6.1.4.1.11129.2.12.2`; it is allocated under Google's enterprise number and will change after the CNCF donation is complete. The purpose MUST be `atunnel`, and the certificate's actor URI SAN MUST identify the same actor as the extension. The PEP MUST then confirm with ate-api-server that the actor still exists, that its current UID matches the certificate, and that it is placed on a worker: `RESUMING` (booting or restoring there) or `RUNNING`. The PEP MUST reject the CONNECT request if any of these checks fail. ```text CURRENT CONNECT EGRESS PATH (one tunnel per actor TCP connection except destination port 53) diff --git a/internal/atunnel/egress/egress.go b/internal/atunnel/egress/egress.go index 6f9c0f6d77..6c9d58c47c 100644 --- a/internal/atunnel/egress/egress.go +++ b/internal/atunnel/egress/egress.go @@ -57,6 +57,7 @@ type Proxy struct { } type activation struct { + actorUID string dialer tunnelDialer certificateSource actorCertificateSource expiresAt time.Time @@ -109,7 +110,7 @@ func (e *Proxy) activationLocked(actorUID string) *activation { active := e.active[actorUID] if active == nil { ctx, cancel := context.WithCancel(context.Background()) - active = &activation{ctx: ctx, cancel: cancel} + active = &activation{actorUID: actorUID, ctx: ctx, cancel: cancel} e.active[actorUID] = active } return active @@ -286,10 +287,23 @@ func (e *Proxy) handle(downstream net.Conn, active *activation) { _ = downstream.Close() return } + if active.dialer == nil { + e.mu.Unlock() + // The actor sees this as a connection that closes right away, like a + // network failure. This logs at Debug because an actor without an + // egress gateway hits this on every outbound connection. + slog.DebugContext(active.ctx, "atunnel dropped an egress connection: actor egress is not active", + slog.String("actorUID", active.actorUID)) + _ = downstream.Close() + return + } if time.Now().Compare(active.expiresAt) >= 0 { // Expiry blocks only new tunnels. Connections admitted with a valid // certificate have completed mTLS and are allowed to drain normally. e.mu.Unlock() + // renew logs the expiry or the denied renewal once at Warn. + slog.DebugContext(active.ctx, "atunnel dropped an egress connection: actor has no valid certificate", + slog.String("actorUID", active.actorUID)) _ = downstream.Close() return } diff --git a/internal/atunnel/tunnel.go b/internal/atunnel/tunnel.go index 484b288995..4372bbb125 100644 --- a/internal/atunnel/tunnel.go +++ b/internal/atunnel/tunnel.go @@ -236,12 +236,10 @@ func (t *Tunnel) PrepareEgress(ctx context.Context, actor resources.ActorAttribu return &ActorEgress{client: gatewayClient, certificateSource: certificateSource, expiresAt: expiresAt}, nil } -// Activate starts admitting the actor's traffic. Ingress reaches the actor -// through dial; egress is activated only when it was prepared. -func (t *Tunnel) Activate(actor resources.ActorAttribution, dial ingress.DialFunc, actorEgress *ActorEgress) error { - if err := t.Ingress.Activate(actor.Ref.Atespace, actor.Ref.Name, actor.UID, dial); err != nil { - return fmt.Errorf("while activating actor ingress: %w", err) - } +// ActivateEgress starts tunneling the actor's outbound traffic. Call it before +// the workload starts, since a workload may need the network to become ready. +// A nil egress is a no-op. +func (t *Tunnel) ActivateEgress(actor resources.ActorAttribution, actorEgress *ActorEgress) error { if actorEgress == nil { return nil } @@ -251,6 +249,15 @@ func (t *Tunnel) Activate(actor resources.ActorAttribution, dial ingress.DialFun return nil } +// ActivateIngress starts admitting the actor's inbound traffic through dial. +// Call it once the workload is ready to serve. +func (t *Tunnel) ActivateIngress(actor resources.ActorAttribution, dial ingress.DialFunc) error { + if err := t.Ingress.Activate(actor.Ref.Atespace, actor.Ref.Name, actor.UID, dial); err != nil { + return fmt.Errorf("while activating actor ingress: %w", err) + } + return nil +} + // Deactivate stops admitting the actor's traffic and drains its active streams, // before the actor network is torn down. It attempts both directions even if // one fails. diff --git a/internal/atunnel/tunnel_test.go b/internal/atunnel/tunnel_test.go index 8a0b7aa5e6..0a28956cac 100644 --- a/internal/atunnel/tunnel_test.go +++ b/internal/atunnel/tunnel_test.go @@ -327,8 +327,11 @@ func TestActivateAndDeactivate(t *testing.T) { actor := resources.ActorAttribution{Ref: resources.ActorRef{Atespace: "space", Name: "actor"}, UID: "uid"} dial := func(context.Context, string, string) (net.Conn, error) { return nil, net.ErrClosed } - if err := tunnel.Activate(actor, dial, nil); err != nil { - t.Fatalf("Activate without egress: %v", err) + if err := tunnel.ActivateEgress(actor, nil); err != nil { + t.Fatalf("ActivateEgress without a gateway: %v", err) + } + if err := tunnel.ActivateIngress(actor, dial); err != nil { + t.Fatalf("ActivateIngress: %v", err) } if err := tunnel.Deactivate(ctx, actor); err != nil { t.Fatalf("Deactivate: %v", err) @@ -338,8 +341,8 @@ func TestActivateAndDeactivate(t *testing.T) { t.Fatalf("second Deactivate: %v", err) } - err := tunnel.Activate(resources.ActorAttribution{UID: "uid"}, dial, nil) + err := tunnel.ActivateIngress(resources.ActorAttribution{UID: "uid"}, dial) if err == nil || !strings.Contains(err.Error(), "while activating actor ingress") { - t.Fatalf("Activate with no actor reference = %v, want an ingress error", err) + t.Fatalf("ActivateIngress with no actor reference = %v, want an ingress error", err) } } diff --git a/internal/e2e/fixture.go b/internal/e2e/fixture.go index 20b573146f..fb2540d49f 100644 --- a/internal/e2e/fixture.go +++ b/internal/e2e/fixture.go @@ -16,6 +16,7 @@ package e2e import ( "context" + "maps" "strings" "testing" "time" @@ -87,6 +88,24 @@ func decodeSubstrateTemplates(t *testing.T, rendered []byte) []*ateapipb.ActorTe return templates } +// FixtureOption adjusts what DeploySubstrateFixture installs. +type FixtureOption func(*fixtureConfig) + +type fixtureConfig struct { + templateValues map[string]string +} + +// WithTemplateValue replaces the inline placeholder ${name} in the fixture's +// ActorTemplate manifest with value. +func WithTemplateValue(name, value string) FixtureOption { + return func(c *fixtureConfig) { + if c.templateValues == nil { + c.templateValues = map[string]string{} + } + c.templateValues["${"+name+"}"] = value + } +} + // SubstrateFixtureManifests names the two manifest templates a substrate // fixture is built from (both repo-relative, under internal/e2e/fixtures). type SubstrateFixtureManifests struct { @@ -115,9 +134,14 @@ type SubstrateFixtureManifests struct { // // Returns the fixture's atespace (the same string that names the k8s // namespace holding the pool) and the created templates. -func DeploySubstrateFixture(t *testing.T, ctx context.Context, clients *Clients, manifests SubstrateFixtureManifests, bucket, name string, trustBundles []string) (string, []*ateapipb.ActorTemplate) { +func DeploySubstrateFixture(t *testing.T, ctx context.Context, clients *Clients, manifests SubstrateFixtureManifests, bucket, name string, trustBundles []string, opts ...FixtureOption) (string, []*ateapipb.ActorTemplate) { t.Helper() + var cfg fixtureConfig + for _, opt := range opts { + opt(&cfg) + } + // The pool manifest also carries the namespace. Its cleanup is registered // first so it runs last (t.Cleanup is LIFO), after the templates that // select the pool's workers are gone. @@ -133,6 +157,7 @@ func DeploySubstrateFixture(t *testing.T, ctx context.Context, clients *Clients, }) tmplInline, tmplBlocks := substrateTemplateSubstitutions(bucket, name, trustBundles) + maps.Copy(tmplInline, cfg.templateValues) rendered := koResolve(t, renderManifest(t, manifests.Template, tmplInline, tmplBlocks)) templates := decodeSubstrateTemplates(t, rendered) diff --git a/internal/e2e/fixtures/probe/main.go b/internal/e2e/fixtures/probe/main.go index 7a732bb4e2..2d0cc90888 100644 --- a/internal/e2e/fixtures/probe/main.go +++ b/internal/e2e/fixtures/probe/main.go @@ -35,6 +35,7 @@ import ( "runtime" "strconv" "strings" + "sync" "time" ) @@ -472,7 +473,120 @@ func writeJSON(w http.ResponseWriter, v any) { } } +// bootFetchURLEnv names a URL the probe must fetch, through the egress gateway +// and with the projected trust bundle, before it listens. A probe that cannot +// fetch it exits and never answers its wakeup probe, so a golden snapshot of +// it proves the actor had egress while it booted. +const bootFetchURLEnv = "PROBE_BOOT_FETCH_URL" + +// bootFetchTimeout bounds the boot fetch, inside the template's wakeupProbe +// timeout. +const bootFetchTimeout = 30 * time.Second + +// bootFetch GETs url once, trusting only trustBundle, and fails unless it +// answers 200. It does not retry: the actor's EgressPolicy exists before the +// actor resumes, so a failure is the finding. +func bootFetch(ctx context.Context, url, trustBundle string) error { + b, err := os.ReadFile(trustBundle) + if err != nil { + return fmt.Errorf("reading trust bundle: %w", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(b) { + return fmt.Errorf("no certificates parsed from %s", trustBundle) + } + client := &http.Client{ + Timeout: bootFetchTimeout, + Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool}}, + } + defer client.CloseIdleConnections() + if err := fetchOnce(ctx, client, url); err != nil { + return fmt.Errorf("boot fetch of %s: %w", url, err) + } + return nil +} + +// restoreFetch answers the wakeup probe once the actor named in its identity +// file has fetched url. A restore from the golden snapshot resumes past main's +// boot fetch with the restored actor's identity, so its wakeup probe passes +// only if the restored actor reaches url too. The prober times out each request +// after wakeupprobe.RequestTimeout, so the fetch runs in the background while +// /healthz answers 503. +type restoreFetch struct { + url, identityFile, trustBundle string + + mu sync.Mutex + fetched string + fetching string +} + +func (f *restoreFetch) healthz(w http.ResponseWriter, _ *http.Request) { + b, err := os.ReadFile(f.identityFile) + if err != nil { + http.Error(w, err.Error(), http.StatusServiceUnavailable) + return + } + id := string(b) + f.mu.Lock() + defer f.mu.Unlock() + if id == f.fetched { + w.WriteHeader(http.StatusOK) + return + } + if f.fetching != id { + f.fetching = id + go f.fetch(id) + } + http.Error(w, "fetching "+f.url+" as "+id, http.StatusServiceUnavailable) +} + +func (f *restoreFetch) fetch(id string) { + err := bootFetch(context.Background(), f.url, f.trustBundle) + f.mu.Lock() + defer f.mu.Unlock() + if f.fetching == id { + f.fetching = "" + } + if err != nil { + log.Printf("probe: restore fetch as %s: %v", id, err) + return + } + log.Printf("probe: restore fetch of %s as %s succeeded", f.url, id) + f.fetched = id +} + +func fetchOnce(ctx context.Context, client *http.Client, url string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return err + } + res, err := client.Do(req) + if err != nil { + return err + } + defer res.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(res.Body, maxFetchBody)) + if res.StatusCode != http.StatusOK { + return fmt.Errorf("status %d", res.StatusCode) + } + return nil +} + func main() { + healthz := func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) } + if url := os.Getenv(bootFetchURLEnv); url != "" { + booted, err := os.ReadFile(identityFile) + if err != nil { + log.Fatalf("probe: reading %s: %v", identityFile, err) + } + if err := bootFetch(context.Background(), url, trustFile); err != nil { + log.Fatalf("probe: %v", err) + } + log.Printf("probe: boot fetch of %s succeeded", url) + f := &restoreFetch{url: url, identityFile: identityFile, trustBundle: trustFile, fetched: string(booted)} + healthz = f.healthz + } + // Hold the identity file open before serving: every snapshot of this actor // then contains an open guest handle on a system-info file (see // heldIdentity). @@ -490,7 +604,7 @@ func main() { mux.HandleFunc("/lookup", lookup) mux.HandleFunc("/resources", resources) mux.HandleFunc("/capabilities", capabilities) - mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }) + mux.HandleFunc("/healthz", healthz) const addr = ":80" log.Printf("probe listening on %s", addr) diff --git a/internal/e2e/fixtures/probe/main_test.go b/internal/e2e/fixtures/probe/main_test.go index 94fff20588..99ac3bb6a7 100644 --- a/internal/e2e/fixtures/probe/main_test.go +++ b/internal/e2e/fixtures/probe/main_test.go @@ -21,9 +21,13 @@ import ( "net/http" "net/http/httptest" "net/url" + "os" + "path/filepath" "slices" "strings" + "sync/atomic" "testing" + "time" "github.com/google/go-cmp/cmp" @@ -363,3 +367,122 @@ func TestCapabilityNamesTable(t *testing.T) { } } } + +// writeServerTrustBundle writes the TLS test server's certificate as the only +// anchor of a trust bundle and returns its path. +func writeServerTrustBundle(t *testing.T, server *httptest.Server) string { + t.Helper() + path := filepath.Join(t.TempDir(), "trust-bundle.pem") + block := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw}) + if err := os.WriteFile(path, block, 0o600); err != nil { + t.Fatal(err) + } + return path +} + +// The actor's EgressPolicy exists before it resumes, so the boot fetch makes +// one request and reports its failure. +func TestBootFetch(t *testing.T) { + tests := []struct { + name string + status int + wantErr string + }{ + {name: "origin answers 200", status: http.StatusOK}, + {name: "origin denies", status: http.StatusForbidden, wantErr: "status 403"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var calls atomic.Int32 + origin := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + calls.Add(1) + w.WriteHeader(tt.status) + })) + defer origin.Close() + + err := bootFetch(t.Context(), origin.URL, writeServerTrustBundle(t, origin)) + if tt.wantErr == "" && err != nil { + t.Fatalf("bootFetch() error = %v, want nil", err) + } + if tt.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tt.wantErr)) { + t.Fatalf("bootFetch() error = %v, want %q", err, tt.wantErr) + } + if got := calls.Load(); got != 1 { + t.Errorf("origin saw %d requests, want 1", got) + } + }) + } +} + +// A restored actor resumes past the boot fetch with its own identity, so its +// wakeup probe fails until that identity has fetched. +func TestRestoreFetchGatesHealthzOnIdentity(t *testing.T) { + origin := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer origin.Close() + identity := filepath.Join(t.TempDir(), "actor-id") + writeIdentity := func(id string) { + t.Helper() + if err := os.WriteFile(identity, []byte(id), 0o600); err != nil { + t.Fatal(err) + } + } + f := &restoreFetch{url: origin.URL, identityFile: identity, trustBundle: writeServerTrustBundle(t, origin), fetched: "golden"} + healthz := func() int { + rec := httptest.NewRecorder() + f.healthz(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + return rec.Code + } + + writeIdentity("golden") + if got := healthz(); got != http.StatusOK { + t.Fatalf("healthz as the actor that booted = %d, want 200", got) + } + writeIdentity("restored") + if got := healthz(); got != http.StatusServiceUnavailable { + t.Fatalf("healthz as a restored actor before it fetched = %d, want 503", got) + } + f.fetch("restored") + if got := healthz(); got != http.StatusOK { + t.Fatalf("healthz as a restored actor after it fetched = %d, want 200", got) + } +} + +// A failed restore fetch leaves healthz at 503, and the next wakeup probe +// starts another fetch. +func TestRestoreFetchRetriesOnNextProbe(t *testing.T) { + var deny atomic.Bool + deny.Store(true) + origin := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if deny.Load() { + w.WriteHeader(http.StatusForbidden) + return + } + w.WriteHeader(http.StatusOK) + })) + defer origin.Close() + identity := filepath.Join(t.TempDir(), "actor-id") + if err := os.WriteFile(identity, []byte("restored"), 0o600); err != nil { + t.Fatal(err) + } + f := &restoreFetch{url: origin.URL, identityFile: identity, trustBundle: writeServerTrustBundle(t, origin), fetched: "golden"} + healthz := func() int { + rec := httptest.NewRecorder() + f.healthz(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + return rec.Code + } + + f.fetch("restored") + if got := healthz(); got != http.StatusServiceUnavailable { + t.Fatalf("healthz after a failed fetch = %d, want 503", got) + } + deny.Store(false) + deadline := time.Now().Add(10 * time.Second) + for healthz() != http.StatusOK { + if time.Now().After(deadline) { + t.Fatal("healthz did not answer 200 after the origin allowed the fetch") + } + time.Sleep(50 * time.Millisecond) + } +} diff --git a/internal/e2e/fixtures/probe/probe-bootfetch-template.yaml.tmpl b/internal/e2e/fixtures/probe/probe-bootfetch-template.yaml.tmpl new file mode 100644 index 0000000000..f365814183 --- /dev/null +++ b/internal/e2e/fixtures/probe/probe-bootfetch-template.yaml.tmpl @@ -0,0 +1,67 @@ +# 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. + +# Boot-fetch variant of the probe template: a protojson-shaped +# ateapipb.ActorTemplate whose probe fetches the suite's BOOT_FETCH_URL through the +# egress gateway before it listens, so its golden snapshot exists only if the +# actor had egress while it booted. Selects the probe pool of probe.yaml.tmpl; +# goldenEgressPolicy (preview gate GoldenEgressPolicy) allows the golden actor +# to reach BOOT_FETCH_HOST; the suite gives the template the trust bundle and +# BOOT_FETCH_HOST and BOOT_FETCH_URL (see e2e.WithTemplateValue). + +metadata: + atespace: ate-e2e-probe${FIXTURE_SUFFIX} + name: probe-bootfetch +workerSelector: + matchLabels: + workload: probe${FIXTURE_SUFFIX} +containers: +- name: probe + image: ko://go.ate.dev/substrate/internal/e2e/fixtures/probe + command: ["/ko-app/probe"] + env: + - name: PROBE_BOOT_FETCH_URL + value: ${BOOT_FETCH_URL} + volumeMounts: + - name: system-info + mountPath: /run/ate + # Longer than the probe's boot fetch timeout, so a failed fetch is the + # probe's logged error rather than a probe timeout. + wakeupProbe: + httpGet: + path: /healthz + port: 80 + timeoutSeconds: 60 +volumes: +- name: system-info + systemInfo: + dataSources: + - actorMetadata: + items: + - field: ACTOR_METADATA_FIELD_NAME + path: actor-id +${TEMPLATE_TRUST_BUNDLE} +# The actor resolves BOOT_FETCH_HOST through its search path, and the DNS relay +# answers only names the policy lists, so the policy also lists the name the +# search path expands it to. +goldenEgressPolicy: + rules: + - https: + hostnames: ["${BOOT_FETCH_HOST}", "${BOOT_FETCH_HOST}.cluster.local"] +resources: + limits: + cpuMillicores: 1000 + memoryBytes: 536870912 # 512Mi +snapshotConfig: + storageLocation: gs://${BUCKET_NAME}/ate-e2e-probe-bootfetch${FIXTURE_SUFFIX}/ diff --git a/internal/e2e/probe.go b/internal/e2e/probe.go index fbd99e339a..6e2cd7637b 100644 --- a/internal/e2e/probe.go +++ b/internal/e2e/probe.go @@ -40,7 +40,7 @@ type probeConfig struct{ trustBundles []string } // The allowlisted names atelet resolves trust bundles by. const ( - egressTrustBundleName = "egress-mitm.ate.dev" + EgressTrustBundleName = "egress-mitm.ate.dev" systemRootsBundleName = "system-roots.ate.dev" ) @@ -51,7 +51,7 @@ const ( // includes the bundle won't start if the bundle is missing, so a suite that // just needs a probe shouldn't depend on it. func WithTrustBundle() ProbeOption { - return func(c *probeConfig) { c.trustBundles = append(c.trustBundles, egressTrustBundleName) } + return func(c *probeConfig) { c.trustBundles = append(c.trustBundles, EgressTrustBundleName) } } // WithSystemRoots adds the WebPKI roots atelet ships to the probe's projected @@ -75,7 +75,7 @@ func DeployProbe(t *testing.T, bucket, name string, opts ...ProbeOption) (string for _, opt := range opts { opt(&cfg) } - if slices.Contains(cfg.trustBundles, egressTrustBundleName) { + if slices.Contains(cfg.trustBundles, EgressTrustBundleName) { // Every actor from this template — the fixture's golden boot included — // fails closed while the bundle is missing, so it has to exist first. EnsureEgressTrustBundle(t, context.Background(), GetClients()) diff --git a/internal/e2e/probe_test.go b/internal/e2e/probe_test.go index 66468bcf2d..d4cd75118f 100644 --- a/internal/e2e/probe_test.go +++ b/internal/e2e/probe_test.go @@ -82,7 +82,7 @@ func TestProbeTemplate_TrustBundle(t *testing.T) { } // The egress name must select the bundle atecontroller publishes, or // actors fail closed on a name atelet rejects. - if !strings.HasPrefix(EgressTrustBundleObjectName, egressTrustBundleName+":") { - t.Errorf("egress trust bundle name %q does not select %q", egressTrustBundleName, EgressTrustBundleObjectName) + if !strings.HasPrefix(EgressTrustBundleObjectName, EgressTrustBundleName+":") { + t.Errorf("egress trust bundle name %q does not select %q", EgressTrustBundleName, EgressTrustBundleObjectName) } } diff --git a/internal/e2e/sandbox_test.go b/internal/e2e/sandbox_test.go index 862d6a4ce2..71de42c504 100644 --- a/internal/e2e/sandbox_test.go +++ b/internal/e2e/sandbox_test.go @@ -15,6 +15,8 @@ package e2e import ( + "bytes" + "maps" "os" "strings" "testing" @@ -27,35 +29,44 @@ import ( // substrateFixtures are every fixture DeploySubstrateFixture is asked to // deploy: the pool half rendered by RenderFixtureManifest and the substrate // ActorTemplate half rendered with substrateTemplateSubstitutions, plus how -// many template documents the latter must yield. +// many template documents the latter must yield. values fills the template's +// own placeholders, as the suite deploying it does through WithTemplateValue. var substrateFixtures = []struct { manifests SubstrateFixtureManifests templates int + values map[string]string }{ {SubstrateFixtureManifests{ Pool: "internal/e2e/fixtures/probe/probe.yaml.tmpl", Template: "internal/e2e/fixtures/probe/probe-template.yaml.tmpl", - }, 1}, + }, 1, nil}, + {SubstrateFixtureManifests{ + Pool: "internal/e2e/fixtures/probe/probe.yaml.tmpl", + Template: "internal/e2e/fixtures/probe/probe-bootfetch-template.yaml.tmpl", + }, 1, map[string]string{ + "${BOOT_FETCH_HOST}": "origin.example", + "${BOOT_FETCH_URL}": "https://origin.example/healthz", + }}, {SubstrateFixtureManifests{ Pool: "internal/e2e/fixtures/probe/probe-sized.yaml.tmpl", Template: "internal/e2e/fixtures/probe/probe-sized-template.yaml.tmpl", - }, 1}, + }, 1, nil}, {SubstrateFixtureManifests{ Pool: "internal/e2e/fixtures/security/atespace-a-pool.yaml.tmpl", Template: "internal/e2e/fixtures/security/atespace-template.yaml.tmpl", - }, 1}, + }, 1, nil}, {SubstrateFixtureManifests{ Pool: "internal/e2e/fixtures/capabilities/capabilities.yaml.tmpl", Template: "internal/e2e/fixtures/capabilities/capabilities-templates.yaml.tmpl", - }, 2}, + }, 2, nil}, {SubstrateFixtureManifests{ Pool: "internal/e2e/fixtures/testserver/websocket.yaml.tmpl", Template: "internal/e2e/fixtures/testserver/websocket-template.yaml.tmpl", - }, 1}, + }, 1, nil}, {SubstrateFixtureManifests{ Pool: "internal/e2e/fixtures/testserver/grpcecho.yaml.tmpl", Template: "internal/e2e/fixtures/testserver/grpcecho-template.yaml.tmpl", - }, 1}, + }, 1, nil}, } // renderPool renders a fixture's pool manifest and strict-decodes its @@ -99,14 +110,21 @@ func renderPool(t *testing.T, relPath string) *v1alpha1.WorkerPool { // renderTemplates renders a fixture's substrate template manifest and // strict-decodes its ActorTemplate documents; the protojson decode plays the -// same misplaced-placeholder tripwire renderPool's strict mode does. -func renderTemplates(t *testing.T, relPath string) []*ateapipb.ActorTemplate { +// same misplaced-placeholder tripwire renderPool's strict mode does. A +// placeholder that values does not fill fails the test, since it would reach +// the cluster as literal text. +func renderTemplates(t *testing.T, relPath string, values map[string]string) []*ateapipb.ActorTemplate { t.Helper() inline, blocks := substrateTemplateSubstitutions("test-bucket", "render", nil) + maps.Copy(inline, values) rendered, err := os.ReadFile(renderManifest(t, relPath, inline, blocks)) if err != nil { t.Fatalf("reading the rendered %s: %v", relPath, err) } + if i := bytes.Index(rendered, []byte("${")); i >= 0 { + end := min(len(rendered), i+40) + t.Fatalf("rendered %s keeps a placeholder: %q", relPath, rendered[i:end]) + } return decodeSubstrateTemplates(t, rendered) } @@ -125,7 +143,7 @@ func TestRenderSubstrateFixtures_GVisor(t *testing.T) { t.Errorf("WorkerPool runtime = class %q, want gvisor", pool.Spec.DefaultSandboxClass()) } - templates := renderTemplates(t, fixture.manifests.Template) + templates := renderTemplates(t, fixture.manifests.Template, fixture.values) if len(templates) != fixture.templates { t.Fatalf("rendered %s yields %d templates, want %d", fixture.manifests.Template, len(templates), fixture.templates) } @@ -165,7 +183,7 @@ func TestRenderSubstrateFixtures_MicroVM(t *testing.T) { t.Errorf("WorkerPool runtime = class %q, want microvm", pool.Spec.DefaultSandboxClass()) } - templates := renderTemplates(t, fixture.manifests.Template) + templates := renderTemplates(t, fixture.manifests.Template, fixture.values) if len(templates) != fixture.templates { t.Fatalf("rendered %s yields %d templates, want %d", fixture.manifests.Template, len(templates), fixture.templates) } diff --git a/internal/e2e/suites/egressmitm/egressmitm_test.go b/internal/e2e/suites/egressmitm/egressmitm_test.go index 1cb96560de..d0cfbc9432 100644 --- a/internal/e2e/suites/egressmitm/egressmitm_test.go +++ b/internal/e2e/suites/egressmitm/egressmitm_test.go @@ -96,7 +96,7 @@ func TestActorEgressMITMTrust(t *testing.T) { const id = "probe-mitm" createAndResumeActor(t, ctx, clients, id, mitm, passthrough) - waitForActorState(t, ctx, clients, id, ateapipb.ActorState_ACTOR_STATE_RUNNING) + waitForActorState(t, ctx, clients, probeNamespace, id, ateapipb.ActorState_ACTOR_STATE_RUNNING) rc, err := e2e.NewRouterClient(t, ctx) if err != nil { @@ -371,12 +371,12 @@ func createAndResumeActor(t *testing.T, ctx context.Context, clients *e2e.Client } } -func waitForActorState(t *testing.T, ctx context.Context, clients *e2e.Clients, actorName string, want ateapipb.ActorState) { +func waitForActorState(t *testing.T, ctx context.Context, clients *e2e.Clients, atespace, actorName string, want ateapipb.ActorState) { t.Helper() deadline := time.Now().Add(60 * time.Second) for time.Now().Before(deadline) { resp, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{ - Actor: &ateapipb.ObjectRef{Atespace: probeNamespace, Name: actorName}, + Actor: &ateapipb.ObjectRef{Atespace: atespace, Name: actorName}, }) if err == nil && resp.GetStatus().GetState() == want { return diff --git a/internal/e2e/suites/egressmitm/goldenboot_test.go b/internal/e2e/suites/egressmitm/goldenboot_test.go new file mode 100644 index 0000000000..aa6f1975a5 --- /dev/null +++ b/internal/e2e/suites/egressmitm/goldenboot_test.go @@ -0,0 +1,85 @@ +// 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 egressmitm + +import ( + "context" + "testing" + + "go.ate.dev/substrate/internal/e2e" + "go.ate.dev/substrate/internal/preview" + "go.ate.dev/substrate/pkg/proto/ateapipb" +) + +var bootFetchManifests = e2e.SubstrateFixtureManifests{ + Pool: "internal/e2e/fixtures/probe/probe.yaml.tmpl", + Template: "internal/e2e/fixtures/probe/probe-bootfetch-template.yaml.tmpl", +} + +// TestGoldenBootEgress proves an actor has egress while it boots: the +// probe-bootfetch workload fetches an in-cluster origin through the MITM +// gateway before it listens, and exits if it cannot, so its template gets a +// golden snapshot only if the worker tunneled the golden actor's egress +// before the wakeup probe and the gateway admitted the actor while it was +// still resuming. The template's goldenEgressPolicy gives the golden actor its +// policy before the controller resumes it. An actor restored from that +// snapshot then has to fetch again as itself before its wakeup probe passes, +// which covers the restore paths. +// +// goldenEgressPolicy is a preview feature: install the cluster with its gate +// and pass the gate to the suite, which skips the test otherwise. +// +// ATE_PREVIEW=GoldenEgressPolicy hack/install-ate-kind.sh +// hack/run-e2e-kind.sh ./internal/e2e/suites/egressmitm -run TestGoldenBootEgress -v -args --no-color --preview=GoldenEgressPolicy +func TestGoldenBootEgress(t *testing.T) { + e2e.RequirePreview(t, preview.GateGoldenEgressPolicy) + env, err := e2e.CheckEnv("BUCKET_NAME", "KO_DOCKER_REPO") + if err != nil { + t.Fatalf("CheckEnv failed: %v", err) + } + ctx := context.Background() + clients := e2e.GetClients() + e2e.EnsureEgressTrustBundle(t, ctx, clients) + origin := e2e.DeployServerPod(t, ctx, originSpec("bootfetch-origin")) + + // DeploySubstrateFixture returns once every template has its golden + // snapshot, which is the assertion. + atespace, _ := e2e.DeploySubstrateFixture(t, ctx, clients, bootFetchManifests, env["BUCKET_NAME"], "egressmitm-bootfetch", []string{e2e.EgressTrustBundleName}, + e2e.WithTemplateValue("BOOT_FETCH_HOST", origin.SvcName()), + e2e.WithTemplateValue("BOOT_FETCH_URL", "https://"+origin.SvcName()+"/healthz")) + + t.Run("restore", func(t *testing.T) { + const id = "bootfetch-restore" + ref := &ateapipb.ObjectRef{Atespace: atespace, Name: id} + if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: id}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: atespace, Name: "probe-bootfetch"}, + }}); err != nil { + t.Fatalf("CreateActor %q: %v", id, err) + } + t.Cleanup(func() { + _, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: ref}) + if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{Actor: ref}); err != nil { + t.Logf("cleanup: DeleteActor %q failed: %v", id, err) + } + }) + // Before the resume, so the gateway never caches a missing policy. + e2e.EnsureEgressPolicy(t, ctx, clients, ref, e2e.EgressAllowHTTPS(origin.SvcName(), origin.SvcName()+".cluster.local")) + if _, err := e2e.ResumeActorAwaitCapacity(t, ctx, clients, &ateapipb.ResumeActorRequest{Actor: ref}); err != nil { + t.Fatalf("ResumeActor %q: %v", id, err) + } + waitForActorState(t, ctx, clients, atespace, id, ateapipb.ActorState_ACTOR_STATE_RUNNING) + }) +} diff --git a/manifests/ate-install/components/agentgateway/kustomization.yaml b/manifests/ate-install/components/agentgateway/kustomization.yaml index 17b2c17ed7..6765bb9491 100644 --- a/manifests/ate-install/components/agentgateway/kustomization.yaml +++ b/manifests/ate-install/components/agentgateway/kustomization.yaml @@ -42,7 +42,7 @@ patches: path: /spec/template/spec/containers/0 value: name: agentgateway - image: cr.agentgateway.dev/agentgateway:v0.0.0-alpha.168b05d6@sha256:5e8384d1fcfdc443a3ad5155ea6a64dc7953bde162df7f5f8e131a5c19ae08a3 + image: cr.agentgateway.dev/agentgateway:v0.0.0-alpha.9d36620d@sha256:bcfa0e5bfa8a4e8206ec4d2029f6b33a3599e55a9be81e1584ed4586298db3da args: - -f - /etc/agentgateway/config.yaml @@ -118,7 +118,7 @@ patches: path: /spec/template/spec/containers/0 value: name: agentgateway - image: cr.agentgateway.dev/agentgateway:v0.0.0-alpha.168b05d6@sha256:5e8384d1fcfdc443a3ad5155ea6a64dc7953bde162df7f5f8e131a5c19ae08a3 + image: cr.agentgateway.dev/agentgateway:v0.0.0-alpha.9d36620d@sha256:bcfa0e5bfa8a4e8206ec4d2029f6b33a3599e55a9be81e1584ed4586298db3da args: - -f - /etc/agentgateway/config.yaml