fix(egress): make credential vault mutations immediately visible

components/egress/credential_vault_active.go

@@ -0,0 +1,74 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// 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 (
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/alibaba/opensandbox/egress/pkg/credentialvault"
+)
+
+func handleActiveVaultSnapshot(
+	w http.ResponseWriter,
+	r *http.Request,
+	store *credentialvault.Store,
+) {
+	knownTag, err := parseActiveVaultETag(r.Header.Get("If-None-Match"))
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	snapshot, tag, changed, err := store.ActiveSnapshotIfChanged(r.Context(), knownTag)
+	if err != nil {
+		credentialvault.WriteError(w, err)
+		return
+	}
+
+	w.Header().Set("ETag", formatActiveVaultETag(tag))
+	if !changed {
+		w.WriteHeader(http.StatusNotModified)
+		return
+	}
+	writeJSON(w, http.StatusOK, snapshot)
+}
+
+func parseActiveVaultETag(value string) (string, error) {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return "", nil
+	}
+	if len(value) < 3 || value[0] != '"' || value[len(value)-1] != '"' {
+		return "", fmt.Errorf("If-None-Match must be a quoted active-vault tag")
+	}
+	tag, err := strconv.Unquote(value)
+	if err != nil || len(tag) == 0 || len(tag) > 128 {
+		return "", fmt.Errorf("If-None-Match contains an invalid active-vault tag")
+	}
+	for _, char := range tag {
+		if !(char >= 'a' && char <= 'z') && !(char >= 'A' && char <= 'Z') &&
+			!(char >= '0' && char <= '9') && !strings.ContainsRune("-._~", char) {
+			return "", fmt.Errorf("If-None-Match contains an invalid active-vault tag")
+		}
+	}
+	return tag, nil
+}
+
+func formatActiveVaultETag(tag string) string {
+	return strconv.Quote(tag)
+}

components/egress/credential_vault_handler_test.go

@@ -103,7 +103,7 @@ func TestCredentialVaultActiveUnixSocketReturnsSnapshot(t *testing.T) {
 		require.NoError(t, os.RemoveAll(tmpDir))
 	})
 	socketPath := filepath.Join(tmpDir, "credential-proxy", "active.sock")
-	_, cleanup, err := credentialvault.StartActiveSocketServer(srv.handleCredentialVaultActive, socketPath, -1)
+	_, cleanup, err := credentialvault.StartActiveSocketServerRequestAware(srv.handleCredentialVaultActive, socketPath, -1)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
@@ -126,8 +126,47 @@ func TestCredentialVaultActiveUnixSocketReturnsSnapshot(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Equal(t, http.StatusOK, resp.StatusCode)
+	initialTag := resp.Header.Get("ETag")
+	require.NotEmpty(t, initialTag)
 	require.Contains(t, string(body), "secret-token")
 	require.Contains(t, string(body), "Private-Token")
+
+	req, err := http.NewRequest(http.MethodGet, "http://credential-proxy/credential-vault/_active", nil)
+	require.NoError(t, err)
+	req.Header.Set("If-None-Match", initialTag)
+	resp, err = client.Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	require.Equal(t, http.StatusNotModified, resp.StatusCode)
+	require.Equal(t, initialTag, resp.Header.Get("ETag"))
+
+	_, err = store.Patch(credentialvault.MutationRequest{
+		Credentials: &credentialvault.CredentialMutationSet{Replace: []credentialvault.Credential{{
+			Name:   "gitlab-token",
+			Source: json.RawMessage(`{"type":"inline","value":"new-secret-token"}`),
+		}}},
+	}, pol)
+	require.NoError(t, err)
+	req, err = http.NewRequest(http.MethodGet, "http://credential-proxy/credential-vault/_active", nil)
+	require.NoError(t, err)
+	req.Header.Set("If-None-Match", initialTag)
+	resp, err = client.Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	body, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.NotEqual(t, initialTag, resp.Header.Get("ETag"))
+	require.Contains(t, string(body), "new-secret-token")
+	require.NotContains(t, string(body), `"secret-token"`)
+
+	req, err = http.NewRequest(http.MethodGet, "http://credential-proxy/credential-vault/_active", nil)
+	require.NoError(t, err)
+	req.Header.Set("If-None-Match", "not-quoted")
+	resp, err = client.Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
 }
 
 func TestCredentialVaultActiveBindingBlocksEgressPolicyRemoval(t *testing.T) {

components/egress/docs/policy-traffic-vault-flow.md

@@ -139,10 +139,29 @@ flowchart LR
 Vault revisions are pushed over the proxy route and held **memory-only** per
 subject (OSEP-0012 model — no Secret volume, nothing written to egress disk).
 The shared mitmdump instance selects the subject's vault by the client's
-source IP (transparent REDIRECT/DNAT preserves it); a revision push rebinds
-in memory and new flows pick up the new credentials. See
+source IP (transparent REDIRECT/DNAT preserves it). It keeps an immutable
+snapshot per subject and conditionally checks the private Unix-socket endpoint
+for every new flow with its opaque `ETag`. An unchanged tag returns `304`
+without rendering or transferring credential material; a changed tag returns
+`200`, the full snapshot, its public revision, and a replacement `ETag`. The
+tag changes even when delete-then-create resets the public revision to `1`, so
+recreation cannot accidentally validate a pre-delete snapshot. Consequently, the
+first flow after a successful create, patch, or delete acknowledgement observes
+that mutation without a timer or cache-expiry sleep. See
 [fleet-mitm-data-plane](../../../docs/components/egress-fleet-mitm-data-plane.md).
 
+`404` has one explicit meaning for the addon: there is no active vault for the
+selected subject, so any older cached snapshot is removed and the flow remains
+ordinary non-credentialed egress. Transport timeout/refusal, `5xx`, malformed
+JSON/schema, an invalid `ETag`, or a non-advancing tag after a conditional
+request are lookup failures, not "no vault". Those failures discard the
+unconfirmed cached plaintext snapshot and fail closed for **all intercepted
+traffic**, including hosts that would not match any credential binding. The
+addon returns a local `503` when the request body is safely buffered, or kills
+a streamed/unknown-length flow before it can reach upstream. Operators should
+therefore treat the private credential-proxy socket as a hard availability
+dependency whenever transparent interception is enabled.
+
 ```mermaid
 sequenceDiagram
     autonumber
@@ -155,11 +174,22 @@ sequenceDiagram
 
     S->>P: PUT /v1/sandboxfleets/{sid}/egress/credential-vault (full revision)
     P->>E: forward (UID header -> subject)
-    E->>V: replace revision (memory-only, new flows rebind)
+    E->>V: atomically replace revision (memory-only)
+    V-->>E: mutation response acknowledges active revision
     C->>M: HTTP(S) flow (DNAT preserves source IP)
     M->>M: script: client source IP -> subject -> subject's vault
-    M->>V: resolve credential/binding for the flow
-    V-->>M: credential (active snapshot)
+    M->>V: GET _active + If-None-Match cached opaque ETag
+    alt snapshot tag unchanged
+        V-->>M: 304 + ETag (reuse immutable snapshot)
+    else snapshot tag changed
+        V-->>M: 200 + ETag + active snapshot
+    else no active vault
+        V-->>M: 404 (clear cached snapshot)
+    else lookup/protocol failure
+        V--xM: timeout/refused/5xx/invalid payload or revision
+        M--xC: 503 or connection termination (no upstream forwarding)
+    end
+    M->>M: resolve credential/binding from one flow-fixed snapshot
     M-->>C: proxied flow with credential applied
 ```
 
@@ -174,6 +204,10 @@ sequenceDiagram
 | Unload (REMOVE_BINDING) | chain + all sets removed in one transaction; stale fence ignored |
 | Egress restart | stale rules wiped (ApplyReset); new instanceId triggers Fastlet replay of SET_BINDING + reached Hooks |
 | Unregistered source | unmarked -> master-chain tail drop — denied before the binding is ever observed |
+| Vault snapshot unchanged | per-flow conditional UDS check returns `304`; cached immutable snapshot is reused without retransmitting secrets |
+| Vault snapshot changed | opaque-tag compare and snapshot render occur under one store read lock; `200` + replacement `ETag` atomically replaces the subject cache |
+| Vault deleted / no active vault | `404` clears any cached snapshot; the flow proceeds without credential injection |
+| Vault lookup or protocol failure | fail-closed before upstream: buffered request receives `503`; streamed, chunked, or HTTP/2 unknown-length request is killed |
 | Malformed action envelope | rejected (never silently ignored); the subject is never activated |
 | data-plane-ready without pending policy | failed (protocol violation) — the subject stays denying |
 

components/egress/fleet_server.go

@@ -284,9 +284,9 @@ func (s *fleetPolicyServer) Handler() http.Handler {
 // handleCredentialVaultActive is the fleet-profile active vault API: one
 // shared socket, dispatch inside. The addon carries the flow's client IP
 // (REDIRECT/DNAT preserves the source), and the handler resolves clientIp ->
-// subject -> that subject's vault snapshot. Unknown IPs 404 (the addon treats
-// that as no-vault, no injection). The sidecar's single-vault handler is
-// unchanged.
+// subject -> that subject's vault snapshot. Unknown IPs and subjects without a
+// vault return 404. Conditional requests return 304 when the subject's opaque
+// active-snapshot tag is unchanged.
 func (s *fleetPolicyServer) handleCredentialVaultActive(w http.ResponseWriter, r *http.Request) {
 	raw := strings.TrimSpace(r.URL.Query().Get("clientIp"))
 	if raw == "" {
@@ -303,12 +303,7 @@ func (s *fleetPolicyServer) handleCredentialVaultActive(w http.ResponseWriter, r
 		http.Error(w, "no subject for clientIp", http.StatusNotFound)
 		return
 	}
-	snapshot, err := s.vaultFor(subj).ActiveSnapshot()
-	if err != nil {
-		credentialvault.WriteError(w, err)
-		return
-	}
-	writeJSON(w, http.StatusOK, snapshot)
+	handleActiveVaultSnapshot(w, r, s.vaultFor(subj))
 }
 
 // subjectOf extracts and validates the routing header. The proxy is the only

components/egress/fleet_server_test.go

@@ -863,6 +863,18 @@ func TestFleetServerActiveVaultClientIPDispatch(t *testing.T) {
 	srv.handleCredentialVaultActive(rec, httptest.NewRequest(http.MethodGet, "/credential-vault/_active?clientIp=10.0.0.5", nil))
 	require.Equal(t, http.StatusOK, rec.Code)
 	require.Contains(t, rec.Body.String(), `"revision":1`)
+	initialTag := rec.Header().Get("ETag")
+	require.NotEmpty(t, initialTag)
+
+	// A conditional lookup for the active snapshot tag avoids rendering and
+	// retransmitting the subject's credentials.
+	req := httptest.NewRequest(http.MethodGet, "/credential-vault/_active?clientIp=10.0.0.5", nil)
+	req.Header.Set("If-None-Match", initialTag)
+	rec = httptest.NewRecorder()
+	srv.handleCredentialVaultActive(rec, req)
+	require.Equal(t, http.StatusNotModified, rec.Code)
+	require.Empty(t, rec.Body.String())
+	require.Equal(t, initialTag, rec.Header().Get("ETag"))
 
 	// subject B has no vault: 404 (the addon treats it as no-vault)
 	rec = httptest.NewRecorder()