fix(cpex): address PR #493 review — multi-part redaction leak + hardening

Resolve Hai Huang's review findings on the CPEX plugin PR:

Must-fix (blocker):
- applyA2AResponseBodyMod: fail closed when >1 text-kind artifact parts
  exist, mirroring the inference response guard (cmf_inference.go:204).
  Previously only the first part was rewritten; parts[1..] forwarded
  the original unredacted content.

Security/correctness:
- Decision zero-value flipped from Allow to a fail-closed Unknown
  sentinel (DecisionUnknown = iota). An uninitialised Result never
  silently allows traffic.
- FakeManager default changed from Allow to Deny for test fidelity.
- Streaming (SSE) response gap detection: when the response body is
  non-empty but yields zero CMF parts (unparseable SSE), return
  DecisionError so fail_open governs rather than silently allowing.
- Dockerfile: cosign verify-blob of the FFI .a asset at build time
  (sha256 + signature verification). Runtime ABI assertion already
  exists in the Go bindings.

Nits:
- Stale "not yet implemented" comment removed (manager_cpex.go:269).
- IBM/... → contextforge-org/... import path comment fixed.
- realm-export.json: _WARNING field flagging demo-only credentials.
- 30-agent.yaml: model pinned from :latest to ollama/llama3.2:8b.
- README + cpex-plugin.md: clarify authbridge-cpex is a build variant
  (not a second sidecar); document OPA dual presence.

Verified: go test ./plugins/cpex/ passes (33 tests); go vet clean;
full hr-cpex demo 9/9 scenarios pass on kind cluster.

Signed-off-by: Frederico Araujo <frederico.araujo@ibm.com>

Author: araujof

authbridge/authlib/plugins/cpex/README.md

@@ -1,5 +1,10 @@
 # CPEX
 
+> **Deployment note:** `authbridge-cpex` is a **build variant** of the
+> AuthBridge proxy-sidecar (`authbridge-proxy`), deployed *in place of*
+> `authbridge-proxy` — not an additional sidecar. The operator selects
+> the image when CPEX policy enforcement is needed for a workload.
+
 The CPEX plugin routes AuthBridge pipeline hooks through the [CPEX](https://github.com/contextforge-org/cpex) framework,
 so operators can define authorization flows inline and declaratively, and turn decisions into ordered effects (including invoking CPEX plugins).
 
@@ -140,6 +145,16 @@ policy:
 Local structural checks run fast and first; external PDPs run when the local
 checks pass; their decisions feed the same `on_deny` / `on_allow` effect lists.
 
+> **OPA in two places.** AuthBridge ships a native `opa` pipeline plugin
+> (standalone, no CPEX dependency) for simple per-request OPA checks.
+> The CPEX `opa(...)` step shown above runs OPA as a CPEX sub-plugin —
+> within APL's effect framework, so its verdict feeds `on_deny`/`on_allow`
+> effects, session tainting, and ordered composition with other PDPs.
+> Use the native `opa` plugin for standalone binary OPA gates; use
+> CPEX's `opa(...)` when OPA participates in a multi-PDP orchestration
+> flow. The `authzen(...)` and `cedar` steps are doc-level BYOP examples
+> not exercised in this demo.
+
 ## CMF / extension mapping
 
 The AuthBridge pipeline context is mapped into the CMF Message plus

authbridge/authlib/plugins/cpex/cmf_a2a.go

@@ -139,18 +139,18 @@ func applyA2ARequestBodyMod(pctx *pipeline.Context, newTexts []string) (mutated
 }
 
 // applyA2AResponseBodyMod rewrites pctx.ResponseBody — a non-streaming
-// A2A JSON-RPC response — replacing the first artifact text part with
-// newArtifact (the single redacted text part CPEX returned on the
-// response phase).
+// A2A JSON-RPC response — replacing the single artifact text part with
+// newArtifact (the redacted text CPEX returned on the response phase).
 //
-// A2A responses are frequently SSE streams (message/stream); those won't
-// parse as one JSON object, so a requested redaction can't be applied and
-// we return an error to fail closed rather than forward unredacted output.
-// For the non-streaming JSON-RPC shape it rewrites the first text part
-// under result.artifacts[].parts[].
+// Because the write side receives only one redacted text while the read
+// side (a2aResponseParts) emits one text part per non-empty text-kind
+// artifact part, a response carrying more than one such part is an
+// ambiguous single-value rewrite — we fail closed rather than overwrite
+// only the first part and forward the rest unredacted. Streaming (SSE)
+// responses don't parse as one JSON object and also fail closed.
 //
-// Returns mutated=false (no error) when newArtifact is empty or there's no
-// artifact text part to replace.
+// Returns mutated=false (no error) when newArtifact is empty or there's
+// no artifact text part to replace.
 func applyA2AResponseBodyMod(pctx *pipeline.Context, newArtifact string) (mutated bool, err error) {
 	if len(pctx.ResponseBody) == 0 || newArtifact == "" {
 		return false, nil
@@ -168,6 +168,11 @@ func applyA2AResponseBodyMod(pctx *pipeline.Context, newArtifact string) (mutate
 		return false, nil
 	}
 
+	// Collect all non-empty text-kind artifact parts, matching what the
+	// read side surfaced. We only hold one redacted text, so >1 such
+	// part is ambiguous: fail closed instead of stamping the same value
+	// over distinct parts or rewriting only the first.
+	targets := make([]map[string]any, 0, 4)
 	for _, a := range artifacts {
 		artifact, ok := a.(map[string]any)
 		if !ok {
@@ -185,21 +190,31 @@ func applyA2AResponseBodyMod(pctx *pipeline.Context, newArtifact string) (mutate
 			if kind, _ := po["kind"].(string); kind != "text" {
 				continue
 			}
-			// Skip empty text parts so we rewrite the same part the read
-			// side surfaced (a2aResponseParts emits only non-empty text).
-			if t, ok := po["text"].(string); !ok || t == "" {
-				continue
-			}
-			po["text"] = newArtifact
-			newBody, err := json.Marshal(envelope)
-			if err != nil {
-				return false, fmt.Errorf("re-serialize A2A response body: %w", err)
+			if t, ok := po["text"].(string); ok && t != "" {
+				targets = append(targets, po)
 			}
-			pctx.SetResponseBody(newBody)
-			return true, nil
 		}
 	}
-	return false, nil
+
+	if len(targets) == 0 {
+		return false, nil
+	}
+	if len(targets) > 1 {
+		return false, fmt.Errorf(
+			"A2A response has %d text parts; single-value redaction rewrite is ambiguous",
+			len(targets))
+	}
+	if targets[0]["text"] == newArtifact {
+		return false, nil
+	}
+	targets[0]["text"] = newArtifact
+
+	newBody, err := json.Marshal(envelope)
+	if err != nil {
+		return false, fmt.Errorf("re-serialize A2A response body: %w", err)
+	}
+	pctx.SetResponseBody(newBody)
+	return true, nil
 }
 
 // a2aMessageParts navigates a decoded A2A JSON-RPC request envelope to

authbridge/authlib/plugins/cpex/cmf_a2a_test.go

@@ -122,6 +122,34 @@ func TestA2AResponseBodyMod_RewritesArtifact(t *testing.T) {
 	}
 }
 
+func TestA2AResponseBodyMod_MultiPartFailsClosed(t *testing.T) {
+	// Two text-kind artifact parts → ambiguous single-value rewrite → error.
+	pctx := &pipeline.Context{
+		ResponseBody: []byte(`{"result":{"artifacts":[{"parts":[{"kind":"text","text":"part one"},{"kind":"text","text":"part two"}]}]}}`),
+	}
+	mutated, err := applyA2AResponseBodyMod(pctx, "redacted")
+	if err == nil {
+		t.Fatal("expected fail-closed error for multi-part response, got nil")
+	}
+	if mutated {
+		t.Fatal("mutated=true on multi-part fail-closed")
+	}
+}
+
+func TestA2AResponseBodyMod_MultiArtifactMultiPartFailsClosed(t *testing.T) {
+	// Text parts spread across multiple artifacts still triggers the guard.
+	pctx := &pipeline.Context{
+		ResponseBody: []byte(`{"result":{"artifacts":[{"parts":[{"kind":"text","text":"first"}]},{"parts":[{"kind":"text","text":"second"}]}]}}`),
+	}
+	mutated, err := applyA2AResponseBodyMod(pctx, "redacted")
+	if err == nil {
+		t.Fatal("expected fail-closed error for multi-artifact text parts, got nil")
+	}
+	if mutated {
+		t.Fatal("mutated=true on multi-artifact fail-closed")
+	}
+}
+
 func TestA2AResponseBodyMod_StreamingFailsClosed(t *testing.T) {
 	pctx := &pipeline.Context{
 		ResponseBody: []byte("data: {\"result\":{}}\n\n"),

authbridge/authlib/plugins/cpex/cmf_body.go

@@ -10,6 +10,17 @@ import (
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 )
 
+// isStreamingResponseGap reports whether a response-phase invocation has a
+// non-empty body that yielded zero structured content parts — i.e. a
+// streaming (SSE) body the policy engine can't inspect. Callers use this
+// to fail closed rather than silently allowing unredacted content through.
+func isStreamingResponseGap(pctx *pipeline.Context, isResponse bool, cmfPartCount int) bool {
+	if !isResponse || cmfPartCount > 0 || len(pctx.ResponseBody) == 0 {
+		return false
+	}
+	return pctx.Extensions.MCP != nil || pctx.Extensions.Inference != nil || pctx.Extensions.A2A != nil
+}
+
 // sessionIDFromHeaders returns the X-Session-Id request header, the
 // session-correlation key AuthBridge threads into CPEX's session resolver
 // (tier-0 Agent.SessionID) for non-A2A traffic. MCP and inference requests

authbridge/authlib/plugins/cpex/cmf_body_test.go

@@ -573,3 +573,74 @@ func TestSessionIDFromHeaders(t *testing.T) {
 		t.Fatalf("got %q, want taint-demo-42", got)
 	}
 }
+
+// --- isStreamingResponseGap ---
+
+func TestIsStreamingResponseGap_SSEBodyWithProtocolExtension(t *testing.T) {
+	pctx := &pipeline.Context{
+		ResponseBody: []byte("data: {\"result\":{}}\n\n"),
+		Extensions: pipeline.Extensions{
+			A2A: &pipeline.A2AExtension{Method: "message/stream"},
+		},
+	}
+	if !isStreamingResponseGap(pctx, true, 0) {
+		t.Fatal("expected streaming gap: SSE body, A2A extension, zero parts")
+	}
+}
+
+func TestIsStreamingResponseGap_InferenceSSE(t *testing.T) {
+	pctx := &pipeline.Context{
+		ResponseBody: []byte("data: {\"choices\":[{\"delta\":{}}]}\n\n"),
+		Extensions: pipeline.Extensions{
+			Inference: &pipeline.InferenceExtension{Model: "gpt-4"},
+		},
+	}
+	if !isStreamingResponseGap(pctx, true, 0) {
+		t.Fatal("expected streaming gap for inference SSE")
+	}
+}
+
+func TestIsStreamingResponseGap_NotOnRequestPhase(t *testing.T) {
+	pctx := &pipeline.Context{
+		ResponseBody: []byte("data: something\n\n"),
+		Extensions: pipeline.Extensions{
+			A2A: &pipeline.A2AExtension{Method: "message/stream"},
+		},
+	}
+	if isStreamingResponseGap(pctx, false, 0) {
+		t.Fatal("request phase should never report a streaming gap")
+	}
+}
+
+func TestIsStreamingResponseGap_PartsPresent(t *testing.T) {
+	pctx := &pipeline.Context{
+		ResponseBody: []byte(`{"result":{"artifacts":[{"parts":[{"kind":"text","text":"ok"}]}]}}`),
+		Extensions: pipeline.Extensions{
+			A2A: &pipeline.A2AExtension{Method: "message/send"},
+		},
+	}
+	if isStreamingResponseGap(pctx, true, 1) {
+		t.Fatal("should not report gap when parts were successfully extracted")
+	}
+}
+
+func TestIsStreamingResponseGap_EmptyBodyNoGap(t *testing.T) {
+	pctx := &pipeline.Context{
+		ResponseBody: nil,
+		Extensions: pipeline.Extensions{
+			A2A: &pipeline.A2AExtension{Method: "message/send"},
+		},
+	}
+	if isStreamingResponseGap(pctx, true, 0) {
+		t.Fatal("empty body should not be a streaming gap")
+	}
+}
+
+func TestIsStreamingResponseGap_NoProtocolExtension(t *testing.T) {
+	pctx := &pipeline.Context{
+		ResponseBody: []byte("data: something\n\n"),
+	}
+	if isStreamingResponseGap(pctx, true, 0) {
+		t.Fatal("no protocol extension → no gap (unrecognised traffic)")
+	}
+}

authbridge/authlib/plugins/cpex/manager.go

@@ -12,7 +12,7 @@ import (
 //
 // The real implementation lives in manager_cpex.go (//go:build cpex)
 // and wraps the cpex.PluginManager from the
-// github.com/IBM/contextforge-plugins-framework/go/cpex package. The
+// github.com/contextforge-org/cpex/go/cpex package. The
 // stub in manager_stub.go (//go:build !cpex) makes NewManager error
 // out with a clear "build the cpex binary" message if anyone tries to
 // configure the plugin in a binary that wasn't built with -tags cpex.
@@ -88,9 +88,15 @@ type ManagerOptions struct {
 type Decision int
 
 const (
+	// DecisionUnknown is the zero-value sentinel. A Result whose
+	// Decision was never explicitly set fails closed in applyDecision
+	// (the default arm), so an uninitialised Result never silently
+	// allows traffic.
+	DecisionUnknown Decision = iota
+
 	// DecisionAllow: all sub-plugins continued. Request proceeds
 	// untouched.
-	DecisionAllow Decision = iota
+	DecisionAllow
 
 	// DecisionDeny: at least one sub-plugin returned a policy
 	// violation. Plugin emits pipeline.Deny.
@@ -113,6 +119,8 @@ const (
 // reasons and log fields.
 func (d Decision) String() string {
 	switch d {
+	case DecisionUnknown:
+		return "unknown"
 	case DecisionAllow:
 		return "allow"
 	case DecisionDeny:

authbridge/authlib/plugins/cpex/manager_cpex.go

@@ -114,6 +114,18 @@ func (c *cpexManager) Invoke(_ context.Context, hookName string, pctx *pipeline.
 	isResponse := pctx.CurrentPhase() == pipeline.InvocationPhaseResponse
 	payload, ext := buildCMF(pctx, isResponse)
 
+	// Streaming-response guard: when we're on the response phase, the
+	// body is non-empty, a protocol parser claimed this traffic, yet the
+	// CMF message has zero content parts — the body is an SSE stream
+	// that the policy engine can't inspect. Rather than silently allowing
+	// the unredacted stream through (the policy had nothing to evaluate),
+	// surface a DecisionError so the fail_open knob governs: the default
+	// fail-closed denies, fail_open=true logs and allows.
+	if isStreamingResponseGap(pctx, isResponse, len(payload.Message.Content)) {
+		return Result{Decision: DecisionError, Reason: "streaming response body not inspectable by policy"},
+			fmt.Errorf("cpex: response body present (%d bytes) but yielded zero content parts — likely SSE stream; failing closed", len(pctx.ResponseBody))
+	}
+
 	// Fused identity-resolve + hook invoke. cpex-core runs the identity
 	// resolvers (jwt-user / jwt-client) ONLY on the identity.resolve hook,
 	// never inside a tool/prompt/resource hook. An FFI host must therefore
@@ -265,8 +277,8 @@ func awaitBackground(hook, reqID string, bg *rcpex.BackgroundTasks) {
 }
 
 // applyModificationsToPctx writes CPEX's modified Extensions and body
-// back onto pctx. MCP body modifications are re-serialized; inference /
-// A2A body rewriting is not yet implemented and fails closed (see
+// back onto pctx. MCP, inference, and A2A body modifications are each
+// re-serialized via format-aware write-back functions (see
 // applyBodyModFromCMF).
 //
 // Extension changes applied:

authbridge/authlib/plugins/cpex/manager_test.go

@@ -25,7 +25,7 @@ type FakeManager struct {
 
 	// Hooks maps hook name → canned Result. Invoke returns the
 	// canned Result when the hook name matches; otherwise it
-	// returns DecisionAllow with a "no hook configured" reason.
+	// returns DecisionDeny (fail-closed default for test fidelity).
 	Hooks map[string]Result
 
 	// InvokeErr, if non-nil, is returned from every Invoke before
@@ -107,7 +107,10 @@ func (f *FakeManager) HasHook(name string) bool {
 }
 
 // Invoke records the call and returns InvokeErr (if set), the canned
-// Result for hookName (if in Hooks), or a default DecisionAllow.
+// Result for hookName (if in Hooks), or a default DecisionDeny. The
+// deny default mirrors production fail-closed semantics: tests must
+// explicitly wire Hooks with Allow/Modify results for the paths they
+// exercise, so an unconfigured hook is never silently permitted.
 func (f *FakeManager) Invoke(_ context.Context, hookName string, pctx *pipeline.Context) (Result, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -118,5 +121,5 @@ func (f *FakeManager) Invoke(_ context.Context, hookName string, pctx *pipeline.
 	if r, ok := f.Hooks[hookName]; ok {
 		return r, nil
 	}
-	return Result{Decision: DecisionAllow, Reason: "fake: no hook configured"}, nil
+	return Result{Decision: DecisionDeny, Reason: "fake: no hook configured (deny by default)"}, nil
 }

authbridge/authlib/plugins/cpex/plugin_test.go

@@ -529,6 +529,26 @@ func TestDispatch_ErrorDecisionFailOpen(t *testing.T) {
 	}
 }
 
+func TestDispatch_UnknownDecisionFailsClosed(t *testing.T) {
+	// A Decision value the switch doesn't recognise (including the
+	// zero-value DecisionUnknown) must fail closed via the default arm.
+	fake := &FakeManager{
+		KnownHooks: []string{HookToolPreInvoke},
+		Hooks: map[string]Result{
+			HookToolPreInvoke: {Decision: DecisionUnknown, Reason: "zero-value sentinel"},
+		},
+	}
+	cfg := `{"hooks":{"on_request":["cmf.tool_pre_invoke"]},"fail_open":false}`
+	p := setupAndInit(t, fake, cfg)
+	a := p.OnRequest(context.Background(), &pipeline.Context{})
+	if a.Type != pipeline.Reject {
+		t.Fatalf("Type = %d, want Reject (DecisionUnknown must fail closed)", a.Type)
+	}
+	if a.Violation == nil || a.Violation.Code != "cpex.error" {
+		t.Fatalf("want Violation code=cpex.error, got %#v", a.Violation)
+	}
+}
+
 func TestSanitizeReason(t *testing.T) {
 	cases := []struct {
 		in   string