feat: Guardrail prerequisites — outbound deny recording + on_error policy

authbridge/authlib/config/config.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 	"gopkg.in/yaml.v3"
 )
 
@@ -67,19 +68,25 @@ type PipelineStageConfig struct {
 // PluginEntry names a plugin and optionally carries per-instance config.
 //
 // The YAML accepts both the bare-name form ("jwt-validation") and the
-// full form ({name, id, config}). The short form keeps existing pipeline
-// configs parsing unchanged; the long form is what plugins that
-// implement pipeline.Configurable actually need. See
+// full form ({name, id, on_error, config}). The short form keeps
+// existing pipeline configs parsing unchanged; the long form is what
+// plugins that implement pipeline.Configurable actually need. See
 // authbridge/docs/plugin-reference.md for the convention plugins
 // follow when decoding Config.
 //
 // Config is captured as a raw subtree via json.RawMessage so the plugin
 // can do its own DisallowUnknownFields decode against a typed struct —
 // the framework does not interpret it.
+//
+// OnError is the framework-owned wrapper policy (see ErrorPolicy).
+// Plugin authors do not consume it — it lives outside the plugin's
+// own config block so all plugins get the same rollout story without
+// each one re-implementing shadow mode.
 type PluginEntry struct {
-	Name   string          `yaml:"name" json:"name"`
-	ID     string          `yaml:"id,omitempty" json:"id,omitempty"`
-	Config json.RawMessage `yaml:"-" json:"config,omitempty"`
+	Name    string               `yaml:"name" json:"name"`
+	ID      string               `yaml:"id,omitempty" json:"id,omitempty"`
+	OnError pipeline.ErrorPolicy `yaml:"on_error,omitempty" json:"on_error,omitempty"`
+	Config  json.RawMessage      `yaml:"-" json:"config,omitempty"`
 }
 
 // UnmarshalYAML accepts either a bare string or a map. The string form
@@ -108,6 +115,16 @@ func (p *PluginEntry) UnmarshalYAML(node *yaml.Node) error {
 				if err := val.Decode(&p.ID); err != nil {
 					return fmt.Errorf("plugin entry id: %w", err)
 				}
+			case "on_error":
+				var raw string
+				if err := val.Decode(&raw); err != nil {
+					return fmt.Errorf("plugin entry on_error: %w", err)
+				}
+				policy := pipeline.ErrorPolicy(raw)
+				if !policy.Valid() {
+					return fmt.Errorf("plugin entry on_error: %q is not a valid policy (expected: enforce, observe, off)", raw)
+				}
+				p.OnError = policy
 			case "config":
 				// Explicit `config: null` (or `config:` with no value)
 				// decodes to a null-tagged scalar node. Normalize to

authbridge/authlib/config/config_test.go

@@ -8,6 +8,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 )
 
 // --- Preset Tests ---
@@ -436,3 +438,78 @@ func TestSessionConfig_SessionEnabled(t *testing.T) {
 		})
 	}
 }
+
+// TestPluginEntry_OnError covers the three accepted policy values plus
+// the omitted case. An invalid policy must fail loud at YAML parse
+// time so operators catch typos before the pod boots.
+func TestPluginEntry_OnError(t *testing.T) {
+	cases := []struct {
+		name     string
+		yaml     string
+		want     pipeline.ErrorPolicy
+		wantFail bool
+	}{
+		{"enforce explicit", "enforce", pipeline.ErrorPolicyEnforce, false},
+		{"observe", "observe", pipeline.ErrorPolicyObserve, false},
+		{"off", "off", pipeline.ErrorPolicyOff, false},
+		{"typo rejected", "observer", "", true},
+		{"upper rejected", "ENFORCE", "", true},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			dir := t.TempDir()
+			path := filepath.Join(dir, "config.yaml")
+			content := "mode: envoy-sidecar\n" +
+				"pipeline:\n" +
+				"  inbound:\n" +
+				"    plugins:\n" +
+				"      - name: custom-guardrail\n" +
+				"        on_error: " + c.yaml + "\n"
+			if err := os.WriteFile(path, []byte(content), 0600); err != nil {
+				t.Fatal(err)
+			}
+			cfg, err := Load(path)
+			if c.wantFail {
+				if err == nil {
+					t.Fatalf("expected parse error for on_error=%q", c.yaml)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Load: %v", err)
+			}
+			got := cfg.Pipeline.Inbound.Plugins[0].OnError
+			if got != c.want {
+				t.Errorf("OnError = %q, want %q", got, c.want)
+			}
+		})
+	}
+}
+
+// TestPluginEntry_OnError_Omitted verifies the empty-string default.
+// Resolved() should treat an absent on_error as enforce; the parsed
+// field itself stays empty so round-tripping YAML doesn't invent a
+// value the operator didn't write.
+func TestPluginEntry_OnError_Omitted(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "config.yaml")
+	content := "mode: envoy-sidecar\n" +
+		"pipeline:\n" +
+		"  inbound:\n" +
+		"    plugins:\n" +
+		"      - name: custom-guardrail\n"
+	if err := os.WriteFile(path, []byte(content), 0600); err != nil {
+		t.Fatal(err)
+	}
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	got := cfg.Pipeline.Inbound.Plugins[0].OnError
+	if got != "" {
+		t.Errorf("omitted OnError parsed as %q, want empty", got)
+	}
+	if got.Resolved() != pipeline.ErrorPolicyEnforce {
+		t.Errorf("Resolved() of empty = %q, want enforce", got.Resolved())
+	}
+}

authbridge/authlib/pipeline/context.go

@@ -104,14 +104,18 @@ type Context struct {
 
 	Extensions Extensions
 
-	// currentPlugin and currentPhase are framework-owned fields set by
-	// Pipeline.Run / RunResponse around each plugin dispatch. They feed
-	// the Record / Allow / Skip / Observe / Modify / DenyAndRecord helper
-	// methods so plugin code doesn't have to repeat its own Name(),
-	// the phase it's in, or the direction on every Invocation literal.
-	// Unexported so plugins can only set them indirectly (via the framework).
+	// currentPlugin, currentPhase, and currentPolicy are framework-owned
+	// fields set by Pipeline.Run / RunResponse around each plugin
+	// dispatch. They feed the Record / Allow / Skip / Observe / Modify /
+	// DenyAndRecord helper methods so plugin code doesn't have to repeat
+	// its own Name(), the phase it's in, or the direction on every
+	// Invocation literal. currentPolicy is consulted by SetBody /
+	// SetResponseBody to suppress mutation propagation when the current
+	// plugin runs under ErrorPolicyObserve. Unexported so plugins can
+	// only set them indirectly (via the framework).
 	currentPlugin string
 	currentPhase  InvocationPhase
+	currentPolicy ErrorPolicy
 
 	// bodyMutated / responseBodyMutated flag that a plugin called
 	// SetBody / SetResponseBody on this context. Listeners read the flag
@@ -135,16 +139,35 @@ type Context struct {
 // pctx in their own dispatch loops (not strictly via Pipeline.Run) need
 // to set the same fields to get consistent Invocation attribution.
 // Production plugins should never call this directly.
+//
+// Sets the policy to the default (enforce). Call setCurrent (unexported)
+// to stamp a non-default policy — only Pipeline.Run/RunResponse should
+// supply a non-default, and they use the unexported path.
 func (c *Context) SetCurrentPlugin(name string, phase InvocationPhase) {
-	c.currentPlugin = name
-	c.currentPhase = phase
+	c.setCurrent(name, phase, ErrorPolicyEnforce)
 }
 
 // ClearCurrentPlugin resets the framework-owned attribution fields.
 // Paired with SetCurrentPlugin.
 func (c *Context) ClearCurrentPlugin() {
+	c.clearCurrent()
+}
+
+// setCurrent stamps the per-dispatch attribution including the current
+// plugin's on_error policy. Internal to the pipeline package; exported
+// SetCurrentPlugin is the listener-facing entry point and defaults
+// policy to enforce.
+func (c *Context) setCurrent(name string, phase InvocationPhase, policy ErrorPolicy) {
+	c.currentPlugin = name
+	c.currentPhase = phase
+	c.currentPolicy = policy.Resolved()
+}
+
+// clearCurrent zeroes the per-dispatch attribution fields.
+func (c *Context) clearCurrent() {
 	c.currentPlugin = ""
 	c.currentPhase = ""
+	c.currentPolicy = ""
 }
 
 // Record appends an Invocation to pctx under the current pipeline
@@ -224,6 +247,13 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action {
 // SetBody mutate the in-memory Context (readers downstream see the
 // change), but the wire is unchanged.
 //
+// Under ErrorPolicyObserve (shadow mode) SetBody is a NO-OP on bytes:
+// the in-memory body is not replaced, bodyMutated stays false, and
+// downstream plugins continue to see the original. A modify
+// Invocation is still recorded — with Shadow=true — so operators can
+// count "would have redacted" on the rollout dashboard. Plugin code
+// therefore looks identical under enforce and observe.
+//
 // SetBody auto-emits a modify-action Invocation with Reason
 // "body_rewritten" and publishes a plugin-public event under
 // "body-mutation/event" carrying the before/after length and sha256 —
@@ -233,6 +263,10 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action {
 // Callers should NOT assign pctx.Body directly — the listener wouldn't
 // know to propagate the change, and the Invocation wouldn't be emitted.
 func (c *Context) SetBody(newBody []byte) {
+	if c.currentPolicy == ErrorPolicyObserve {
+		c.recordShadowBodyMutation("request", c.Body, newBody)
+		return
+	}
 	old := c.Body
 	c.Body = newBody
 	c.bodyMutated = true
@@ -242,14 +276,44 @@ func (c *Context) SetBody(newBody []byte) {
 // SetResponseBody is the response-side analogue of SetBody. Used by
 // plugins that redact or rewrite the upstream response (prompt-safety
 // guardrails on LLM output, content filters, DLP). Same contract —
-// Invocation + body-mutation/event emitted; never logs the body.
+// Invocation + body-mutation/event emitted; never logs the body —
+// and the same observe-mode suppression: under ErrorPolicyObserve the
+// response body is untouched and the Invocation is marked Shadow=true.
 func (c *Context) SetResponseBody(newBody []byte) {
+	if c.currentPolicy == ErrorPolicyObserve {
+		c.recordShadowBodyMutation("response", c.ResponseBody, newBody)
+		return
+	}
 	old := c.ResponseBody
 	c.ResponseBody = newBody
 	c.responseBodyMutated = true
 	c.emitBodyMutation("response", old, newBody)
 }
 
+// recordShadowBodyMutation emits the would-mutate Invocation for a
+// SetBody / SetResponseBody call that was suppressed under observe
+// mode. Mirrors emitBodyMutation's telemetry (length + sha256 delta)
+// so dashboards get the same shape they see under enforce, just with
+// Shadow=true and no wire-level effect.
+func (c *Context) recordShadowBodyMutation(phase string, oldBody, newBody []byte) {
+	c.Record(Invocation{
+		Action: ActionModify,
+		Reason: "body_rewritten",
+		Shadow: true,
+	})
+	if c.Extensions.Custom == nil {
+		c.Extensions.Custom = map[string]any{}
+	}
+	c.Extensions.Custom["body-mutation"+PluginEventSuffix] = bodyMutationEvent{
+		Phase:        phase,
+		Plugin:       c.currentPlugin,
+		LengthBefore: len(oldBody),
+		LengthAfter:  len(newBody),
+		SHA256Before: hashHex(oldBody),
+		SHA256After:  hashHex(newBody),
+	}
+}
+
 // BodyMutated reports whether a plugin called SetBody during this
 // request. Listeners check this after Run to decide whether to emit a
 // body mutation on the wire. Stream-scoped — a new Context starts with
@@ -353,3 +417,37 @@ type AgentIdentity struct {
 	WorkloadID  string
 	TrustDomain string
 }
+
+// markLastInvocationShadow finds the most recent Invocation authored
+// by pluginName in the given phase (within the current direction
+// bucket) and flips its Shadow flag to true. Returns true when a
+// matching record was found. Used by the pipeline to retroactively
+// tag a plugin's deny record as shadow once Pipeline.Run has observed
+// that the plugin returned Reject under ErrorPolicyObserve.
+//
+// Walking backwards and matching on Plugin+Phase is O(N) in the
+// worst case but typically short-circuits at index -1 — plugins
+// almost always Record right before returning Reject, so the last
+// entry matches.
+func (c *Context) markLastInvocationShadow(pluginName string, phase InvocationPhase) bool {
+	if c.Extensions.Invocations == nil {
+		return false
+	}
+	var list []Invocation
+	switch c.Direction {
+	case Inbound:
+		list = c.Extensions.Invocations.Inbound
+	case Outbound:
+		list = c.Extensions.Invocations.Outbound
+	default:
+		return false
+	}
+	for i := len(list) - 1; i >= 0; i-- {
+		if list[i].Plugin != pluginName || list[i].Phase != phase {
+			continue
+		}
+		list[i].Shadow = true
+		return true
+	}
+	return false
+}

authbridge/authlib/pipeline/extensions.go

@@ -231,6 +231,46 @@ type Invocations struct {
 	Outbound []Invocation `json:"outbound,omitempty"`
 }
 
+// FilteredByPhase returns a new *Invocations containing only entries
+// whose Phase matches the argument. Strict match — untagged entries
+// (Phase == "") are dropped because the framework always populates
+// Phase via Context.Record; an untagged entry is a plugin bug and
+// including it in the wrong phase would double-report in one event
+// and be missing from the other.
+//
+// The underlying Invocation values are copied shallowly; mutating a
+// returned entry's Details map mutates the source. Acceptable for
+// the per-request flow where the original lives only on pctx and is
+// discarded after recording.
+//
+// Returns nil when no entries match so callers can drop the field
+// from the SessionEvent without a null check.
+//
+// Intended for reject-event recording in listeners and for the
+// accept-path phase split (one SessionEvent per phase). Listeners
+// that need full independence from pctx's lifecycle layer their own
+// snapshot on top.
+func (in *Invocations) FilteredByPhase(phase InvocationPhase) *Invocations {
+	if in == nil {
+		return nil
+	}
+	out := &Invocations{}
+	for _, inv := range in.Inbound {
+		if inv.Phase == phase {
+			out.Inbound = append(out.Inbound, inv)
+		}
+	}
+	for _, inv := range in.Outbound {
+		if inv.Phase == phase {
+			out.Outbound = append(out.Outbound, inv)
+		}
+	}
+	if out.Inbound == nil && out.Outbound == nil {
+		return nil
+	}
+	return out
+}
+
 // Invocation records one plugin's action on one pipeline pass. Plugin is
 // the plugin's Name() for traceability. Action is the universal 5-value
 // verb (see Action). Reason is a stable machine-readable label paired
@@ -295,6 +335,15 @@ type Invocation struct {
 	// The session API has no auth on it — only safe-to-log data
 	// belongs in Invocation.Details.
 	Details map[string]string `json:"details,omitempty"`
+
+	// Shadow reports that the plugin ran under on_error: observe and
+	// its decision (deny or modify) was NOT applied to the request.
+	// An operator reading a would-have-blocked timeline filters on
+	// Shadow=true to count rollout-candidate events; a dashboard that
+	// aggregates "effective denies" filters Shadow=false. The
+	// framework, not the plugin, sets this — plugin code looks
+	// identical under enforce and observe.
+	Shadow bool `json:"shadow,omitempty"`
 }
 
 // DelegationExtension tracks the token delegation chain across hops.