fix(egress): attribute subprocess proxy audit events to task/invocation (#338)

.claude/skills/forge.md

@@ -565,7 +565,12 @@ the in-process proxy isn't started — `NetworkPolicy` (generated by
 `forge package`) handles egress at the pod level.
 
 Emits `egress_allowed` / `egress_blocked` per request with `source`
-(in-process vs proxy) so consumers can attribute.
+(in-process vs proxy) so consumers can attribute. Both variants carry
+`correlation_id` + `task_id`: the in-process enforcer reads them from
+ctx; the subprocess proxy recovers them from the `Proxy-Authorization`
+creds the subprocess replays (runner stamps the IDs into the injected
+`HTTP_PROXY` userinfo). Creds-ignoring binaries still get enforced +
+audited, just without identity (#338).
 
 **Read**: `docs/security/egress-control.md`, `docs/security/overview.md`.
 

docs/security/audit-logging.md

@@ -48,12 +48,12 @@ All runtime security events are emitted as structured NDJSON to stderr with corr
 ```json
 {"ts":"2026-02-28T10:00:00Z","event":"session_start","correlation_id":"a1b2c3d4","task_id":"task-1"}
 {"ts":"2026-02-28T10:00:01Z","event":"tool_exec","correlation_id":"a1b2c3d4","fields":{"tool":"tavily_research","phase":"start"}}
-{"ts":"2026-02-28T10:00:01Z","event":"egress_allowed","correlation_id":"a1b2c3d4","fields":{"domain":"api.tavily.com","mode":"allowlist","source":"proxy"}}
+{"ts":"2026-02-28T10:00:01Z","event":"egress_allowed","correlation_id":"a1b2c3d4","task_id":"task-1","fields":{"domain":"api.tavily.com","mode":"allowlist","source":"proxy"}}
 {"ts":"2026-02-28T10:00:05Z","event":"tool_exec","correlation_id":"a1b2c3d4","fields":{"tool":"tavily_research","phase":"end"}}
 {"ts":"2026-02-28T10:00:06Z","event":"session_end","correlation_id":"a1b2c3d4","fields":{"state":"completed"}}
 ```
 
-The `source` field distinguishes in-process enforcer events from subprocess proxy events.
+The `source` field distinguishes in-process enforcer events from subprocess proxy events. Both carry `correlation_id` and `task_id`: the in-process enforcer reads them from the request context, while the subprocess proxy recovers them from the `Proxy-Authorization` credentials the subprocess replays (the runner stamps the task/invocation IDs into the `HTTP_PROXY` URL userinfo). A subprocess binary that ignores proxy credentials still produces an enforced, audited event — just without the identity fields (issue #338).
 
 ### Workflow correlation
 
@@ -581,7 +581,7 @@ documented inline:
 
 | Site | Why plain `Emit` |
 |---|---|
-| Egress proxy `OnAttempt` with `source=proxy` | Subprocess HTTP `CONNECT` has no Go ctx tying back to the A2A request |
+| Egress proxy `OnAttempt` with `source=proxy` | Subprocess HTTP `CONNECT` has no Go ctx tying back to the A2A request, so it can't use `EmitFromContext`. Since #338 it still recovers `task_id` + `correlation_id` out-of-band from the `Proxy-Authorization` creds and sets them on the event manually; seq + trace cross-link remain unavailable (no ctx). |
 | MCP server startup events (`mcp_server_started` / `_failed` / `_degraded`) | Pre-invocation; no scope |
 | Scheduler tick (`schedule_fire` / `schedule_complete` / `schedule_skip` / `schedule_modify`) | Runs on its own timer outside any A2A request |
 | Startup banners (`policy_loaded`, `agent_card_published`, `audit_export_status`) | Pre-invocation; no scope |

docs/security/egress-control.md

@@ -327,12 +327,12 @@ The `allow_private_ips` field controls whether RFC 1918 addresses are allowed th
 Both the enforcer and proxy emit structured audit events:
 
 ```json
-{"event":"egress_allowed","domain":"api.tavily.com","mode":"allowlist"}
-{"event":"egress_blocked","domain":"evil.com","mode":"allowlist"}
-{"event":"egress_allowed","domain":"api.tavily.com","mode":"allowlist","source":"proxy"}
+{"event":"egress_allowed","correlation_id":"a1b2c3d4","task_id":"task-1","fields":{"domain":"api.tavily.com","mode":"allowlist"}}
+{"event":"egress_blocked","correlation_id":"a1b2c3d4","task_id":"task-1","fields":{"domain":"evil.com","mode":"allowlist"}}
+{"event":"egress_allowed","correlation_id":"a1b2c3d4","task_id":"task-1","fields":{"domain":"api.tavily.com","mode":"allowlist","source":"proxy"}}
 ```
 
-Events without `"source"` come from the in-process enforcer; events with `"source": "proxy"` come from the subprocess proxy.
+Events without `"source"` come from the in-process enforcer; events with `"source": "proxy"` come from the subprocess proxy. Both carry `correlation_id` (the invocation ID) and `task_id`. The in-process enforcer reads them from the request context; the proxy recovers them from the `Proxy-Authorization` credentials the subprocess replays — the runner stamps the task/invocation IDs into the injected `HTTP_PROXY` URL as userinfo, and standard HTTP clients echo that back as a Basic proxy-auth header on every request and `CONNECT`. A binary that ignores proxy credentials is still enforced and audited, but its proxy events omit the identity fields (issue #338).
 
 ## Related Files
 

forge-cli/runtime/runner.go

@@ -721,14 +721,21 @@ func (r *Runner) Run(ctx context.Context) error {
 		if (!security.InContainer() && egressCfg.Mode != security.ModeDevOpen) || browserActive {
 			matcher := security.NewDomainMatcher(egressCfg.Mode, egressCfg.AllDomains)
 			egressProxy = security.NewEgressProxy(matcher, allowPrivateIPs)
-			egressProxy.OnAttempt = func(domain string, allowed bool) {
+			egressProxy.OnAttempt = func(a security.EgressAttempt) {
 				event := coreruntime.AuditEgressAllowed
-				if !allowed {
+				if !a.Allowed {
 					event = coreruntime.AuditEgressBlocked
 				}
+				// #338 — task_id/correlation_id are recovered from the
+				// Proxy-Authorization creds the subprocess replays (see
+				// SkillCommandExecutor / identityFromRequest). Empty when the
+				// proxied binary ignores proxy credentials, matching the
+				// pre-#338 domain-only event.
 				auditLogger.Emit(coreruntime.AuditEvent{
-					Event:  event,
-					Fields: map[string]any{"domain": domain, "mode": string(egressCfg.Mode), "source": "proxy"},
+					Event:         event,
+					TaskID:        a.TaskID,
+					CorrelationID: a.CorrelationID,
+					Fields:        map[string]any{"domain": a.Domain, "mode": string(egressCfg.Mode), "source": "proxy"},
 				})
 			}
 			var pErr error

forge-cli/tools/browser/manager_test.go

@@ -74,9 +74,9 @@ func startProxy(t *testing.T, matcher *security.DomainMatcher) (string, *sync.Mu
 	proxy := security.NewEgressProxy(matcher, false)
 	mu := &sync.Mutex{}
 	attempts := map[string]bool{}
-	proxy.OnAttempt = func(domain string, allowed bool) {
+	proxy.OnAttempt = func(a security.EgressAttempt) {
 		mu.Lock()
-		attempts[domain] = allowed
+		attempts[a.Domain] = a.Allowed
 		mu.Unlock()
 	}
 	proxyURL, err := proxy.Start(context.Background())

forge-cli/tools/browser/manual_demo_test.go

@@ -50,12 +50,12 @@ func TestManualBrowseDemo(t *testing.T) {
 	// demonstrates real egress enforcement, not dev-open.
 	matcher := security.NewDomainMatcher(security.ModeAllowlist, []string{u.Hostname()})
 	proxy := security.NewEgressProxy(matcher, false)
-	proxy.OnAttempt = func(domain string, allowed bool) {
+	proxy.OnAttempt = func(a security.EgressAttempt) {
 		verdict := "ALLOW"
-		if !allowed {
+		if !a.Allowed {
 			verdict = "BLOCK"
 		}
-		t.Logf("[egress] %-5s %s", verdict, domain)
+		t.Logf("[egress] %-5s %s", verdict, a.Domain)
 	}
 	proxyURL, err := proxy.Start(context.Background())
 	if err != nil {

forge-cli/tools/cli_execute.go

@@ -223,7 +223,7 @@ func (t *CLIExecuteTool) Execute(ctx context.Context, args json.RawMessage) (str
 	}
 
 	// Security check 6: Env isolation
-	cmd.Env = t.buildEnv(input.Binary)
+	cmd.Env = t.buildEnv(ctx, input.Binary)
 
 	// R9 JIT credentials: if the runner wired a credentials.Injector,
 	// mint fresh scoped-down creds now, merge them into the subprocess
@@ -332,7 +332,9 @@ func stripEnvKeys(env []string, override map[string]string) []string {
 // and explicitly configured passthrough variables. When workDir is set,
 // HOME is overridden to workDir so subprocess ~ expansion stays confined.
 // The binary parameter scopes credential env vars to only the binaries that need them.
-func (t *CLIExecuteTool) buildEnv(binary string) []string {
+// ctx carries the task/invocation identity that is stamped into the egress
+// proxy URL so proxied subprocess egress can be attributed in the audit log (#338).
+func (t *CLIExecuteTool) buildEnv(ctx context.Context, binary string) []string {
 	realHome := os.Getenv("HOME")
 	homeVal := realHome
 	if t.workDir != "" {
@@ -374,11 +376,15 @@ func (t *CLIExecuteTool) buildEnv(binary string) []string {
 		}
 	}
 	if t.proxyURL != "" {
+		// #338 — stamp the task/invocation IDs into the proxy URL userinfo so
+		// the egress proxy can attribute this subprocess's egress to its task
+		// in the audit log (same mechanism as SkillCommandExecutor).
+		proxyURL := proxyURLWithIdentity(ctx, t.proxyURL)
 		env = append(env,
-			"HTTP_PROXY="+t.proxyURL,
-			"HTTPS_PROXY="+t.proxyURL,
-			"http_proxy="+t.proxyURL,
-			"https_proxy="+t.proxyURL,
+			"HTTP_PROXY="+proxyURL,
+			"HTTPS_PROXY="+proxyURL,
+			"http_proxy="+proxyURL,
+			"https_proxy="+proxyURL,
 		)
 	}
 

forge-cli/tools/cli_execute_test.go

@@ -2,12 +2,15 @@ package tools
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strings"
 	"testing"
+
+	coreruntime "github.com/initializ/forge/forge-core/runtime"
 )
 
 func TestCLIExecute_Name(t *testing.T) {
@@ -583,7 +586,7 @@ func TestBuildEnv_GHConfigDirScopedToGh(t *testing.T) {
 	})
 
 	// gh binary should get GH_CONFIG_DIR
-	ghEnv := tool.buildEnv("gh")
+	ghEnv := tool.buildEnv(context.Background(), "gh")
 	found := false
 	for _, e := range ghEnv {
 		if strings.HasPrefix(e, "GH_CONFIG_DIR=") {
@@ -596,7 +599,7 @@ func TestBuildEnv_GHConfigDirScopedToGh(t *testing.T) {
 	}
 
 	// curl binary should NOT get GH_CONFIG_DIR
-	curlEnv := tool.buildEnv("curl")
+	curlEnv := tool.buildEnv(context.Background(), "curl")
 	for _, e := range curlEnv {
 		if strings.HasPrefix(e, "GH_CONFIG_DIR=") {
 			t.Error("GH_CONFIG_DIR should not be set for curl binary")
@@ -624,7 +627,7 @@ func TestBuildEnv_KubeconfigScopedToKubectl(t *testing.T) {
 	})
 
 	// kubectl should get KUBECONFIG
-	kubectlEnv := tool.buildEnv("kubectl")
+	kubectlEnv := tool.buildEnv(context.Background(), "kubectl")
 	found := false
 	for _, e := range kubectlEnv {
 		if strings.HasPrefix(e, "KUBECONFIG=") {
@@ -640,7 +643,7 @@ func TestBuildEnv_KubeconfigScopedToKubectl(t *testing.T) {
 	}
 
 	// curl should NOT get KUBECONFIG
-	curlEnv := tool.buildEnv("curl")
+	curlEnv := tool.buildEnv(context.Background(), "curl")
 	for _, e := range curlEnv {
 		if strings.HasPrefix(e, "KUBECONFIG=") {
 			t.Error("KUBECONFIG should not be set for curl binary")
@@ -669,7 +672,7 @@ func TestBuildEnv_KubectlNoProxy(t *testing.T) {
 	tool.proxyURL = "http://127.0.0.1:54321"
 
 	// kubectl should get NO_PROXY with the K8s API server host
-	kubectlEnv := tool.buildEnv("kubectl")
+	kubectlEnv := tool.buildEnv(context.Background(), "kubectl")
 	var noProxy string
 	for _, e := range kubectlEnv {
 		if strings.HasPrefix(e, "NO_PROXY=") {
@@ -688,10 +691,42 @@ func TestBuildEnv_KubectlNoProxy(t *testing.T) {
 	}
 
 	// curl should NOT get NO_PROXY
-	curlEnv := tool.buildEnv("curl")
+	curlEnv := tool.buildEnv(context.Background(), "curl")
 	for _, e := range curlEnv {
 		if strings.HasPrefix(e, "NO_PROXY=") {
 			t.Error("NO_PROXY should not be set for curl binary")
 		}
 	}
 }
+
+// TestBuildEnv_ProxyIdentityStamped pins that cli_execute (the tool the LLM
+// drives via curl/gh/etc.) stamps the task/invocation IDs into the injected
+// HTTP_PROXY, so the egress proxy can attribute its subprocess egress (#338).
+// This is the path that produced source=proxy events with no task_id before
+// the fix was extended beyond SkillCommandExecutor.
+func TestBuildEnv_ProxyIdentityStamped(t *testing.T) {
+	tool := NewCLIExecuteTool(CLIExecuteConfig{
+		AllowedBinaries: []string{"curl"},
+		WorkDir:         t.TempDir(),
+	})
+	tool.proxyURL = "http://127.0.0.1:54321"
+
+	ctx := coreruntime.WithCorrelationID(
+		coreruntime.WithTaskID(context.Background(), "aibuilderdemo-1784399795814070000"),
+		"7fabfb4f99467e8c",
+	)
+	want := "HTTP_PROXY=http://" +
+		base64.RawURLEncoding.EncodeToString([]byte("aibuilderdemo-1784399795814070000")) + ":" +
+		base64.RawURLEncoding.EncodeToString([]byte("7fabfb4f99467e8c")) + "@127.0.0.1:54321"
+
+	env := tool.buildEnv(ctx, "curl")
+	var found bool
+	for _, e := range env {
+		if e == want {
+			found = true
+		}
+	}
+	if !found {
+		t.Errorf("expected identity-stamped %q in cli_execute env; got:\n%s", want, strings.Join(env, "\n"))
+	}
+}

forge-cli/tools/exec.go

@@ -3,7 +3,9 @@ package tools
 import (
 	"bytes"
 	"context"
+	"encoding/base64"
 	"fmt"
+	"net/url"
 	"os"
 	"os/exec"
 	"time"
@@ -12,6 +14,7 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 
 	"github.com/initializ/forge/forge-core/llm/oauth"
+	coreruntime "github.com/initializ/forge/forge-core/runtime"
 )
 
 // otelEnvPassthroughPrefixes lists the OTel SDK env var prefixes / names
@@ -145,11 +148,18 @@ func (e *SkillCommandExecutor) Run(ctx context.Context, command string, args []s
 		env = append(env, "REVIEW_MODEL="+e.Model)
 	}
 	if e.ProxyURL != "" {
+		// #338 — stamp the task/invocation IDs into the proxy URL userinfo so
+		// the egress proxy can attribute each subprocess request back to its
+		// task in the audit log. HTTP clients replay userinfo as a Basic
+		// Proxy-Authorization header on every request/CONNECT; the proxy
+		// decodes it (identityFromRequest) and tags egress_allowed/blocked
+		// events. When ctx carries neither ID the base URL is used unchanged.
+		proxyURL := proxyURLWithIdentity(ctx, e.ProxyURL)
 		env = append(env,
-			"HTTP_PROXY="+e.ProxyURL,
-			"HTTPS_PROXY="+e.ProxyURL,
-			"http_proxy="+e.ProxyURL,
-			"https_proxy="+e.ProxyURL,
+			"HTTP_PROXY="+proxyURL,
+			"HTTPS_PROXY="+proxyURL,
+			"http_proxy="+proxyURL,
+			"https_proxy="+proxyURL,
 		)
 	}
 	// Issue #182 — propagate W3C trace context + curated OTel SDK env
@@ -198,6 +208,35 @@ func (e *SkillCommandExecutor) Run(ctx context.Context, command string, args []s
 	return stdout.String(), nil
 }
 
+// proxyURLWithIdentity embeds the request's task and correlation (invocation)
+// IDs as userinfo in the egress proxy URL. The subprocess's HTTP client replays
+// that userinfo as a Basic Proxy-Authorization header, letting the egress proxy
+// attribute each outbound request to its originating task/invocation in the
+// audit log (#338). Returns base unchanged when ctx carries neither ID, or when
+// base doesn't parse as a URL.
+//
+// The IDs are base64url-encoded before they go into the userinfo. The Basic
+// credential the client sends is base64(user:pass) with no internal escaping,
+// so a task_id containing a raw ':' (params.id is client-supplied) would
+// mis-split on the proxy side. base64url output is drawn from [A-Za-z0-9-_], so
+// it can never contain the ':' separator — identityFromRequest decodes each
+// half back to the verbatim ID. These are correlation identifiers, not secrets;
+// the encoding is for separator-safety, not confidentiality.
+func proxyURLWithIdentity(ctx context.Context, base string) string {
+	taskID := coreruntime.TaskIDFromContext(ctx)
+	corrID := coreruntime.CorrelationIDFromContext(ctx)
+	if taskID == "" && corrID == "" {
+		return base
+	}
+	u, err := url.Parse(base)
+	if err != nil {
+		return base
+	}
+	enc := base64.RawURLEncoding
+	u.User = url.UserPassword(enc.EncodeToString([]byte(taskID)), enc.EncodeToString([]byte(corrID)))
+	return u.String()
+}
+
 // resolveOAuthToken loads and refreshes the OpenAI OAuth token.
 // Returns nil if no valid token is available.
 func resolveOAuthToken() *oauth.Token {