diff --git a/TRACKER.md b/TRACKER.md index ce507eb..fe0f82a 100644 --- a/TRACKER.md +++ b/TRACKER.md @@ -14,12 +14,13 @@ | | | |---|---| | **Last updated** | 2026-06-08 | -| **Current phase** | Phase 6 — Governance & Tenancy (**4 / 10 steps**) | -| **Overall** | **68 / 84 steps** — Phases 0–5 complete | -| **Next action** | **Step 6.5 — PII policies**: per-tenant PII enforcement at egress (block / redact / allow) through the PolicyEngine `egress_text` decision; `pii.egress_blocked` event. Builds on the per-tenant `pii_policy` already resolved onto `RequestContext` in 6.1. | +| **Current phase** | Phase 6 — Governance & Tenancy (**5 / 10 steps**) | +| **Overall** | **69 / 84 steps** — Phases 0–5 complete | +| **Next action** | **Step 6.6 — Immutable audit log**: hash-chain audit log with tamper-evident verification + WORM export; `GET /v1/audit`. Builds on the Step 0.7c `AuditStore` SPI + `AuditWriter` (SHA-256 hash chain) already in `rag-core`. | **Recently shipped** +- **6.5** ✅ PII egress policies — `PiiPolicyEngine` (`rag-pii`) answers `egress_text` over the context (`list[Chunk]`) + agent answer (`str`) the gateway already passes, applying the per-tenant `pii_policy` (allow / redact / mask / block); reuses the Step 1.7 detector + rewriters; opt-in `cfg.pii.enabled`; emits `pii.egress_blocked` — [#152](https://github.com/officialCodeWork/AgentContextOS/pull/152) - **6.4** ✅ ACL egress verifier — `AclEgressVerifier` re-checks returned `ChunkRef`s against the principal's labels at the gateway router boundary (defense-in-depth, same overlap semantics, independent of the PDP); `acl.egress_violation` event; red-team zero-violation-rate gate — [#151](https://github.com/officialCodeWork/AgentContextOS/pull/151) - **6.3** ✅ ACL push-down — opt-in `AclPolicyEngine` And-merges `any_in("acl_labels", principal.acl_labels)` into every `read_chunk` query (overlap, fail-closed); `acl.egress_denied` event — [#150](https://github.com/officialCodeWork/AgentContextOS/pull/150) - **6.2** ✅ Physical tenancy — per-tenant *dedicated* vector index/collection; `TenantConfig.dedicated_index` → `ctx.physical_index` → backend `-` (Noop/Pinecone/Qdrant); cross-tenant probe gate — [#149](https://github.com/officialCodeWork/AgentContextOS/pull/149) @@ -58,9 +59,9 @@ | 3 | Gateway & Agent Runtime | 11 | **11** | 0 | | 4 | Reliability | 6 | **6** | 0 | | 5 | Eval & Observability | 7 | **7** | 0 | -| 6 | Governance & Tenancy | 10 | **4** | 6 | +| 6 | Governance & Tenancy | 10 | **5** | 5 | | 7 | Pilot, Harden, GA | 10 | 0 | 10 | -| **Total** | | **84** | **68** | **16** | +| **Total** | | **84** | **69** | **15** | --- @@ -651,7 +652,7 @@ | 6.2 | Physical tenancy (dedicated index) | ✅ | [#149](https://github.com/officialCodeWork/AgentContextOS/pull/149) — `dedicated_index` → `ctx.physical_index` → backend `-` (Noop/Pinecone/Qdrant, lazy create); cross-tenant probe gate | | 6.3 | ACL push-down at retrieval | ✅ | [#150](https://github.com/officialCodeWork/AgentContextOS/pull/150) — opt-in `AclPolicyEngine` And-merges `any_in("acl_labels", …)` into every `read_chunk` push-down (overlap, fail-closed); `acl.egress_denied` | | 6.4 | ACL egress verifier | ✅ | [#151](https://github.com/officialCodeWork/AgentContextOS/pull/151) — `AclEgressVerifier` re-checks returned chunks at the gateway router boundary (defense-in-depth above the 6.3 push-down); `acl.egress_violation`; zero-violation-rate red-team gate | -| 6.5 | PII policies | ⏳ | Per-tenant PII enforcement (block / redact / allow); egress redaction; `pii.egress_blocked` event | +| 6.5 | PII policies | ✅ | [#152](https://github.com/officialCodeWork/AgentContextOS/pull/152) — `PiiPolicyEngine` egress_text decorator (allow / redact / mask / block per tenant) over answer + context; reuses Step 1.7 detector; `pii.egress_blocked` | | 6.6 | Immutable audit log | ⏳ | Hash-chain audit log; WORM export; tamper-evident verification; `GET /v1/audit` | | 6.7 | BYOK (Bring Your Own Key) | ⏳ | KMS integration (AWS KMS, GCP KMS, HashiCorp Vault); envelope encryption for embeddings | | 6.8 | SSO / SCIM | ⏳ | OIDC + SAML IdP federation; SCIM 2.0 user provisioning; per-tenant IdP config | @@ -695,6 +696,15 @@ - **`acl.egress_violation`** (pre-registered `EVT_ACL_EGRESS_VIOLATION`, `error` level, PII-free: tenant / principal / counts / dropped chunk **ids** only) fires once per call that drops ≥1 ref — the *unexpected* push-down failure, distinct from 6.3's *expected* `acl.egress_denied`; a clean pass is silent - **Scope:** gateway retrieval surfaces (everything reading `app.state.retrieval_router`); trusts the labels the backend reports on each `ChunkRef` (catching a *mislabelling* backend needs authoritative re-hydration — deferred), per-tenant/per-label violation metrics deferred to the 6.x governance dashboards. `AclConfig` → `rag.schema` regenerated; ~21 new tests incl. a **red-team zero-violation-rate gate** (`tests/redteam/test_acl_egress_verifier.py` — bypassed push-down + leaky backend → verifier drops every violation) + verifier unit + gateway wiring + event-schema; all gates green (ruff, mypy --strict 296 files, RAG001, schema-drift, policy-coverage, log-schema). [ADR-0036](docs/adr/ADR-0036-acl-egress-verifier.md), [architecture/policy-engine.md](docs/architecture/policy-engine.md), [reference/tenancy.md](docs/reference/tenancy.md) +### 6.5 — PII policies ✅ [#152](https://github.com/officialCodeWork/AgentContextOS/pull/152) + +- Step 1.7 detects + redacts PII at *ingest*; Step 6.1 resolved each tenant's `pii_policy` onto `RequestContext.pii_policy`; but nothing enforced it at *egress* — the `egress_text` PDP decision existed and the gateway already called it at every generation boundary, yet the default engine answered `allow`. New **`PiiPolicyEngine`** (`rag-pii`) — a decorator like `QuotaPolicyEngine` / `AclPolicyEngine` — answers `egress_text` and applies the per-tenant action: **allow** delegates, **redact / mask** → `transform`, **block** → `deny` +- **Handles both subject shapes the gateway already passes** — the retrieved **context** (`list[Chunk]` → the LLM on `/v1/query?generate` / `/v1/chat/completions` / gRPC `Converse`) and the agent's final **answer** (`str`) — so enforcement lands at **all four existing `egress_text` call sites with no route change and no coverage-linter entry** (the existing deny→drop / transform→substitute handling consumes the result unchanged) +- **Reuses Step 1.7 machinery**: detection via the injected `PIIDetector` SPI (default the dependency-free `RegexPIIDetector`; production injects Presidio), rewriting via the same `redact_spans` / `mask_spans`, and the **same `min_score` + `entities` filter** as the ingest `PiiProcessor` — so the two stages never disagree, and a clean scan is a **no-op**. `block` on a chunk list is all-or-nothing (any PII denies the whole context); redact/mask rewrite each affected chunk's `content` (clean chunks pass through, same object) +- Lives in **`rag-pii`** (which gains a `rag-policy` dep, exactly as `rag-quota` does for `QuotaPolicyEngine`; graph stays acyclic — `rag-policy` never imports `rag-pii`). `filter_pushdown` / `health` / non-`egress_text` decisions delegate inward, so it **composes with the ACL + quota PDP** +- **Opt-in** via new `cfg.pii.enabled` (default off — it can redact / withhold content); `build_app(pii_enabled=…, pii_detector=…)` wraps the engine after the acl wrap, `build_app_from_config` from config; `app.state.pii_enabled` reports the state. Events are **PII-free**: `pii.egress_blocked` (block, the alertable denial — the reserved event from Step 0.7b now emitted) / `pii.detected` (redact/mask) carry entity *types* + counts + surface, never the matched values +- **Scope:** enforces at the existing `egress_text` sites (context for query/OpenAI/gRPC, answer for the agent). **Deferred:** a post-generation answer re-check for the query/OpenAI/gRPC generate paths (they sanitise the *context* pre-LLM) and a PII check on retrieval-only citations — stored chunks are already ingest-sanitised (1.7). `PiiConfig` → `rag.schema` regenerated; ~24 new tests (engine unit: str/chunks × allow/redact/mask/block, min_score/entities, events, delegation; gateway wiring + composition with ACL + behavioral redact/block); all gates green (ruff, mypy --strict 297 files, RAG001, schema-drift, policy-coverage, log-schema/PII gate). [ADR-0037](docs/adr/ADR-0037-pii-egress-policies.md), [architecture/pii.md](docs/architecture/pii.md), [architecture/policy-engine.md](docs/architecture/policy-engine.md), [reference/pii.md](docs/reference/pii.md) + --- ## Phase 7 — Pilot, Harden, GA (Weeks 34–40) ⏳ @@ -850,6 +860,7 @@ Complete log of every PR. Routine Dependabot bumps are grouped; everything else | [#149](https://github.com/officialCodeWork/AgentContextOS/pull/149) | 2026-06-08 | feat(tenancy): physical tenancy — dedicated index per tenant (Step 6.2) | | [#150](https://github.com/officialCodeWork/AgentContextOS/pull/150) | 2026-06-08 | feat(policy): ACL push-down at retrieval — AclPolicyEngine (Step 6.3) | | [#151](https://github.com/officialCodeWork/AgentContextOS/pull/151) | 2026-06-08 | feat(policy): ACL egress verifier — defense-in-depth re-check (Step 6.4) | +| [#152](https://github.com/officialCodeWork/AgentContextOS/pull/152) | 2026-06-08 | feat(pii): PII egress policies — PiiPolicyEngine egress_text decorator (Step 6.5) | | #78–#80, #116–#118 | Open | Dependabot bumps — awaiting merge | | #81 | Closed | Dependabot bump — superseded | diff --git a/apps/gateway/src/rag_gateway/app.py b/apps/gateway/src/rag_gateway/app.py index 7908538..496382e 100644 --- a/apps/gateway/src/rag_gateway/app.py +++ b/apps/gateway/src/rag_gateway/app.py @@ -370,6 +370,8 @@ def build_app( tenant_resolver: Any | None = None, acl_enabled: bool = False, acl_verify_egress: bool = True, + pii_enabled: bool = False, + pii_detector: Any | None = None, enable_cors: bool = True, default_tenant_id: TenantId | None = None, ) -> FastAPI: @@ -518,8 +520,20 @@ def build_app( from rag_policy import AclPolicyEngine policy_engine = AclPolicyEngine(inner=policy_engine) + # Per-tenant PII enforcement at egress (Step 6.5) — when enabled, decorate the + # engine so the ``egress_text`` decision scans the context / answer leaving the + # system and applies the tenant's ``pii_policy`` (allow / redact / mask / block). + # Opt-in (it can redact or withhold content); the default detector is the + # dependency-free RegexPIIDetector — production injects Presidio. Plugs into the + # gateway's existing egress_text call sites, so no route change is needed. + if pii_enabled: + from rag_pii import PiiPolicyEngine, RegexPIIDetector + + detector = pii_detector if pii_detector is not None else RegexPIIDetector() + policy_engine = PiiPolicyEngine(inner=policy_engine, detector=detector) app.state.policy_engine = policy_engine app.state.acl_enabled = acl_enabled + app.state.pii_enabled = pii_enabled app.state.quota_enforcer = quota_enforcer # Hallucination guard (Step 4.3) — post-generation faithfulness check over diff --git a/apps/gateway/src/rag_gateway/wiring.py b/apps/gateway/src/rag_gateway/wiring.py index dce661d..b666463 100644 --- a/apps/gateway/src/rag_gateway/wiring.py +++ b/apps/gateway/src/rag_gateway/wiring.py @@ -707,6 +707,9 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: acl_enabled = overrides.pop("acl_enabled", cfg.acl.enabled) acl_verify_egress = overrides.pop("acl_verify_egress", cfg.acl.verify_egress) + # Per-tenant PII enforcement at egress (Step 6.5) — opt-in; off by default. + pii_enabled = overrides.pop("pii_enabled", cfg.pii.enabled) + return build_app( corpus_store=corpus_store, retrieval_router=retrieval_router, @@ -718,6 +721,7 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: quota_enforcer=quota_enforcer, acl_enabled=acl_enabled, acl_verify_egress=acl_verify_egress, + pii_enabled=pii_enabled, **overrides, ) diff --git a/apps/gateway/tests/test_pii.py b/apps/gateway/tests/test_pii.py new file mode 100644 index 0000000..3c18a79 --- /dev/null +++ b/apps/gateway/tests/test_pii.py @@ -0,0 +1,104 @@ +"""PII egress policy config + gateway wiring (Step 6.5).""" + +from __future__ import annotations + +from rag_config import RagConfig +from rag_config.schema import AclConfig, PiiConfig +from rag_core.types import ( + Chunk, + ChunkId, + CorpusId, + DocumentId, + PiiAction, + PiiPolicy, + Principal, + PrincipalId, + PrincipalKind, + RequestContext, + TenantId, +) +from rag_gateway import build_app +from rag_gateway.wiring import build_app_from_config +from rag_pii import PiiPolicyEngine +from rag_policy import AclPolicyEngine, PolicyDecision + + +def _ctx(action: PiiAction) -> RequestContext: + tid = TenantId("acme") + return RequestContext( + tenant_id=tid, + principal=Principal( + id=PrincipalId("p"), kind=PrincipalKind.user, display_name="p", tenant_id=tid + ), + pii_policy=PiiPolicy(action=action, min_score=0.0), + ) + + +# --------------------------------------------------------------------------- +# config + wiring +# --------------------------------------------------------------------------- +def test_pii_config_defaults_off() -> None: + assert RagConfig().pii.enabled is False + + +def test_inert_by_default() -> None: + app = build_app() + assert app.state.pii_enabled is False + assert not isinstance(app.state.policy_engine, PiiPolicyEngine) + + +def test_build_from_config_wraps_when_enabled() -> None: + app = build_app_from_config(RagConfig(pii=PiiConfig(enabled=True))) + assert app.state.pii_enabled is True + assert isinstance(app.state.policy_engine, PiiPolicyEngine) + + +def test_build_from_config_inert_when_disabled() -> None: + app = build_app_from_config(RagConfig()) + assert not isinstance(app.state.policy_engine, PiiPolicyEngine) + + +def test_pii_decorates_without_absorbing_inner() -> None: + app = build_app(pii_enabled=True) + engine = app.state.policy_engine + assert isinstance(engine, PiiPolicyEngine) + # the inner engine is preserved, so PII composes with — rather than replaces — + # the production PDP. + assert engine.inner is not None + + +def test_pii_composes_with_acl() -> None: + app = build_app_from_config(RagConfig(acl=AclConfig(enabled=True), pii=PiiConfig(enabled=True))) + engine = app.state.policy_engine + # PII is the outer decorator; ACL is preserved beneath it. + assert isinstance(engine, PiiPolicyEngine) + assert isinstance(engine.inner, AclPolicyEngine) + + +# --------------------------------------------------------------------------- +# behavioral — the wired engine enforces over egress_text with the default detector +# --------------------------------------------------------------------------- +async def test_wired_engine_blocks_pii_answer() -> None: + app = build_app(pii_enabled=True) + res = await app.state.policy_engine.evaluate( + _ctx(PiiAction.block), PolicyDecision.egress_text, "email me at bob@corp.com" + ) + assert res.is_deny() + + +async def test_wired_engine_redacts_pii_context() -> None: + app = build_app(pii_enabled=True) + tid = TenantId("acme") + chunk = Chunk( + id=ChunkId("c1"), + document_id=DocumentId("d1"), + tenant_id=tid, + corpus_id=CorpusId("c"), + content="ssn 123-45-6789 on file", + position=0, + ) + res = await app.state.policy_engine.evaluate( + _ctx(PiiAction.redact), PolicyDecision.egress_text, [chunk] + ) + assert res.is_transform() + assert "123-45-6789" not in res.transformed[0].content diff --git a/dist/rag.schema.json b/dist/rag.schema.json index 248af5f..0eb0527 100644 --- a/dist/rag.schema.json +++ b/dist/rag.schema.json @@ -965,6 +965,19 @@ "title": "PIIPolicy", "type": "string" }, + "PiiConfig": { + "additionalProperties": false, + "description": "PII enforcement at egress (Step 6.5).\n\nWhen ``enabled`` the gateway wraps its PolicyEngine in a ``PiiPolicyEngine``\nthat answers the ``egress_text`` decision: it scans the text about to leave the\nsystem \u2014 the retrieved context sent to the LLM (``/v1/query?generate`` /\n``/v1/chat/completions`` / gRPC ``Converse``) and the agent's final answer \u2014\nand applies the **per-tenant** action resolved onto ``RequestContext.pii_policy``\n(Step 6.1, from ``tenants[].pii_policy``): **allow** passes through, **redact** /\n**mask** sanitise the text (``transform``), and **block** withholds it (``deny``).\n\n**Disabled by default** \u2014 turning it on can redact or withhold answers. The\ndefault detector is the dependency-free ``RegexPIIDetector`` (email / phone /\nSSN / credit-card / IP); production injects Presidio. A **block** emits a\nPII-free ``pii.egress_blocked`` event; a redact/mask emits ``pii.detected``.", + "properties": { + "enabled": { + "default": false, + "title": "Enabled", + "type": "boolean" + } + }, + "title": "PiiConfig", + "type": "object" + }, "PlatformConfig": { "additionalProperties": false, "properties": { @@ -1621,6 +1634,9 @@ "acl": { "$ref": "#/$defs/AclConfig" }, + "pii": { + "$ref": "#/$defs/PiiConfig" + }, "webhooks": { "$ref": "#/$defs/WebhooksConfig" }, diff --git a/dist/rag.schema.yaml b/dist/rag.schema.yaml index d144141..9eb7af0 100644 --- a/dist/rag.schema.yaml +++ b/dist/rag.schema.yaml @@ -946,6 +946,44 @@ $defs: - allow title: PIIPolicy type: string + PiiConfig: + additionalProperties: false + description: 'PII enforcement at egress (Step 6.5). + + + When ``enabled`` the gateway wraps its PolicyEngine in a ``PiiPolicyEngine`` + + that answers the ``egress_text`` decision: it scans the text about to leave + the + + system — the retrieved context sent to the LLM (``/v1/query?generate`` / + + ``/v1/chat/completions`` / gRPC ``Converse``) and the agent''s final answer + — + + and applies the **per-tenant** action resolved onto ``RequestContext.pii_policy`` + + (Step 6.1, from ``tenants[].pii_policy``): **allow** passes through, **redact** + / + + **mask** sanitise the text (``transform``), and **block** withholds it (``deny``). + + + **Disabled by default** — turning it on can redact or withhold answers. The + + default detector is the dependency-free ``RegexPIIDetector`` (email / phone + / + + SSN / credit-card / IP); production injects Presidio. A **block** emits a + + PII-free ``pii.egress_blocked`` event; a redact/mask emits ``pii.detected``.' + properties: + enabled: + default: false + title: Enabled + type: boolean + title: PiiConfig + type: object PlatformConfig: additionalProperties: false properties: @@ -1497,6 +1535,8 @@ properties: $ref: '#/$defs/QuotaConfig' acl: $ref: '#/$defs/AclConfig' + pii: + $ref: '#/$defs/PiiConfig' webhooks: $ref: '#/$defs/WebhooksConfig' provenance: diff --git a/docs/README.md b/docs/README.md index a5172a2..b6208a5 100644 --- a/docs/README.md +++ b/docs/README.md @@ -20,7 +20,7 @@ | [ocr.md](architecture/ocr.md) | OCR pipeline: region-aware SPI evolution, Tesseract vs. PaddleOCR output normalisation, pipeline placement, policy boundary | | [chunker.md](architecture/chunker.md) | Structure-aware chunker: heading stack + parent-link semantics, overlap policy, token counting, OCR-region handling, pipeline placement | | [enricher.md](architecture/enricher.md) | Metadata enricher: what we tag and why, langdetect determinism, reading-level gating, section_path walker, composition contract, policy boundary | -| [pii.md](architecture/pii.md) | PII detection: detector / enforcer split, regex scoring + Luhn check, PolicyEngine composition, event protocol (no raw PII in logs), pipeline placement | +| [pii.md](architecture/pii.md) | PII detection: detector / enforcer split, regex scoring + Luhn check, PolicyEngine composition, event protocol (no raw PII in logs), pipeline placement; egress enforcement via `PiiPolicyEngine` (Step 6.5) | | [embedders.md](architecture/embedders.md) | Embedder pipeline: 3 plugins / 4 model families, shared BatchingEmbedder (batching + retry + dim normalize), server- vs client-side dim reduction, E5 passage prefix, Cohere input_type, retry scope | | [ingest.md](architecture/ingest.md) | Ingest pipeline orchestrator: why a dedicated package, stage order rationale, PolicyEngine consultation, per-document error isolation, gateway vs CLI split, what's deliberately deferred | | [retrieval-read-layer.md](architecture/retrieval-read-layer.md) | Read-layer contract & FilterExpr push-down (Step 2.1): allowed fields, per-backend translators (pgvector SQL, Qdrant), reference `evaluate()` semantics, reviewer checklist | @@ -85,7 +85,7 @@ | [ocr.md](reference/ocr.md) | `rag-ocr` reference — `OCR` SPI region-aware results, `TesseractOCR`, `PaddleOCRBackend`, `ragctl ocr`, extension points | | [chunker.md](reference/chunker.md) | `rag-chunker` reference — `Chunker` SPI, `HeadingAwareChunker`, `TokenCounter` / `TiktokenCounter`, `ocr_result_to_parsed_document`, `ragctl chunk` | | [enricher.md](reference/enricher.md) | `rag-enricher` reference — `Enricher` SPI, `DefaultEnricher`, `LanguageDetector` / `LangdetectDetector`, `doc_type_from_mime`, `section_path`, `ragctl enrich` | -| [pii.md](reference/pii.md) | `rag-pii` reference — `PIIDetector` SPI, `RegexPIIDetector`, `PresidioPIIDetector`, `PiiProcessor`, `redact_spans` / `mask_spans`, `ragctl pii` | +| [pii.md](reference/pii.md) | `rag-pii` reference — `PIIDetector` SPI, `RegexPIIDetector`, `PresidioPIIDetector`, `PiiProcessor`, `PiiPolicyEngine` (Step 6.5 egress: allow/redact/mask/block over answer + context via `egress_text`, `cfg.pii.enabled`), `redact_spans` / `mask_spans`, `ragctl pii` | | [embedders.md](reference/embedders.md) | `rag-embedders` reference — `OpenAIEmbedder`, `CohereEmbedder`, `SentenceTransformersEmbedder` + `bge_large_en()` / `e5_large_v2()`, `BatchingEmbedder` base, `RetryPolicy`, `normalize_dimension`, `ragctl embed` | | [ingest.md](reference/ingest.md) | `rag-ingest` reference — `IngestPipeline.ingest_document` / `ingest_connector`, `IngestResult` / `IngestRunSummary` / `IngestStatus`, `POST /v1/ingest/document`, `ragctl ingest` | | [retrieval.md](reference/retrieval.md) | `rag-retrieval` reference — `rrf_fuse`, `HybridRetriever`, `HybridWeights`, `GraphAdapter` / `default_graph_adapter`, `ragctl hybrid` | @@ -175,6 +175,7 @@ broken, and what to fix before committing to the next phase. | [ADR-0034-physical-multi-tenancy.md](adr/ADR-0034-physical-multi-tenancy.md) | Decision (Step 6.2): a *dedicated* vector index/collection per tenant. `TenantConfig.dedicated_index` resolves to a `physical_index` key on `TenantSettings`, threaded onto `RequestContext.physical_index`; backends read only `ctx` (graph is backends→core, never rag-config) and namespace their base under it (`-`), lazily creating it; one instance + per-tenant derivation (no per-tenant instances, no SPI change); Noop is the CI conformance oracle (keyed by `physical_index`) for a cross-tenant probe gate that proves isolation independent of the tenant filter; Noop + Pinecone + Qdrant this step, others later | | [ADR-0035-acl-pushdown.md](adr/ADR-0035-acl-pushdown.md) | Decision (Step 6.3): label-based ACL push-down at retrieval. `AclPolicyEngine` (a decorator like `QuotaPolicyEngine`) And-merges `any_in("acl_labels", principal.acl_labels)` into every `read_chunk` push-down at the canonical `HybridRetriever` PDP site — overlap semantics via the existing `AnyIn` predicate (zero backend/translator changes), **fail-closed** (label-less principal matches nothing; "public" = a shared label), **opt-in** via `cfg.acl.enabled`; emits `acl.egress_denied` on a request-level denial; graph edge ACLs + post-retrieval re-verification (6.4) deferred | | [ADR-0036-acl-egress-verifier.md](adr/ADR-0036-acl-egress-verifier.md) | Decision (Step 6.4): a post-retrieval ACL re-check as a **defense-in-depth second layer** behind the 6.3 push-down. `AclEgressVerifier.verify(ctx, refs)` drops any returned `ChunkRef` whose labels don't overlap the principal's — same overlap semantics (no-op on correct results), reading `ChunkRef.acl_labels` (no re-hydration), **independent of the PDP** (consults only `ctx.principal.acl_labels`) so a push-down bug/bypass can't disable both; wired at the gateway as a `SupportsRoute` wrapper around `app.state.retrieval_router` (covers query/retrieve/corpus/OpenAI/agent); `cfg.acl.verify_egress` default on but gated by `enabled`; emits `acl.egress_violation` on a caught leak; a red-team gate proves a zero escaped-violation rate when the push-down is bypassed; backend-mislabel re-hydration + per-tenant violation metrics deferred | +| [ADR-0037-pii-egress-policies.md](adr/ADR-0037-pii-egress-policies.md) | Decision (Step 6.5): per-tenant PII enforcement at egress via a `PiiPolicyEngine` `egress_text` decorator (mirrors `QuotaPolicyEngine` / `AclPolicyEngine`), living in `rag-pii` (gains a `rag-policy` dep, like `rag-quota`). Handles both subject shapes the gateway already passes — `list[Chunk]` context + `str` answer — so it plugs into the existing `egress_text` call sites with no route change; maps `ctx.pii_policy.action` allow→delegate / redact·mask→`transform` / block→`deny`, reusing the Step 1.7 detector + rewriters and the same `min_score`+`entities` filter (no-op on clean text); opt-in `cfg.pii.enabled` (injects `RegexPIIDetector`); PII-free `pii.egress_blocked` (block) / `pii.detected` (redact·mask); post-gen answer re-check for query/OpenAI/gRPC + citation egress deferred (stored chunks are ingest-sanitised) | | [ADR-0031-cost-anomaly.md](adr/ADR-0031-cost-anomaly.md) | Decision (Step 5.6c): detect per-tenant spend spikes with a rolling `CostTracker` (not the cumulative quota counter); detect scale-free on the token series (cost = tokens × a constant price) so detection is decoupled from quota pricing and works with quotas off; two gates (ratio + z-score, z relaxed on a flat baseline) → tri-state verdict; put it in `rag-observability` as a `dataclass` (gateway wraps it in a Pydantic `CostStatusResponse`) so there's **no `rag-core` type / `dist/schemas` churn**; feed O(1) from `record_request_usage` before the quota block; pull-based `GET /v1/status/cost` (no per-request span/event); rejected folding into the infra-scoped drift registry, a new package, a `cost.anomaly_detected` push event (deferred), per-model pricing, a time-series DB | | [ADR-0030-drift-monitors.md](adr/ADR-0030-drift-monitors.md) | Decision (Step 5.5): detect retrieval degradation with five drift monitors in a new `rag-drift` package (mirroring rag-feedback); two statistics — PSI (pure, binned, dependency-free) for the distribution monitors + mean-drop for the rate/score monitors — over one scalar-window `DriftMonitor`; infra-scoped registry (like breakers) fed via `observe` from the signals the gateway already computes (query length / retrieval score / HyDE-embedding norm / guard grounded-claim fraction / feedback citation clicks); detection on dashboard-poll `evaluate()` with transition-edge `drift.detected` (structured event + the Step 3.9 webhook, targeting `alert_tenant`); observe-only / inert-by-default / rebaseline; rejected per-tenant monitors, per-dimension embedding PSI, a stats library, a background scheduler, hot-path detection | | [ADR-0029-online-feedback.md](adr/ADR-0029-online-feedback.md) | Decision (Step 5.4): capture online feedback + implicit signals in a new `rag-feedback` package mirroring `rag-provenance` (SPI + types in rag-core; recorder + pure aggregator in the package); one polymorphic `POST /v1/feedback` (a `signal` enum spanning explicit thumbs/rating/comment + implicit citation-click/copy/regenerate/dwell, `kind` inferred); normalise every signal to a `[-1,1]` score so the dashboard has one satisfaction number; **redact-don't-hash** free-text comments via an injected `PIIDetector` (default `NoopPIIDetector` seam, `comment_redacted` flag, PII-free event) — opposite of provenance's hashing; body identity like `/v1/query`; degrade-open + inert-by-default; `GET /v1/status/feedback` dashboard (event-only, no per-call span); admin-UI card deferred to 5.6; rejected separate per-signal endpoints, header-auth, hashing/raw comments, an OTel span per submission, folding into provenance | diff --git a/docs/adr/ADR-0037-pii-egress-policies.md b/docs/adr/ADR-0037-pii-egress-policies.md new file mode 100644 index 0000000..26db0a7 --- /dev/null +++ b/docs/adr/ADR-0037-pii-egress-policies.md @@ -0,0 +1,88 @@ +# ADR-0037 — PII egress policies + +**Status:** Accepted +**Date:** 2026-06-08 +**Step:** 6.5 — PII policies (Phase 6 — Governance & Tenancy) +**Related:** [ADR-0005](ADR-0005-policy-engine.md) (PolicyEngine PDP), [ADR-0033](ADR-0033-logical-multi-tenancy.md) (logical tenancy), [ADR-0024](ADR-0024-quotas-rate-limiting.md) (QuotaPolicyEngine decorator), [architecture/policy-engine.md](../architecture/policy-engine.md), [architecture/pii.md](../architecture/pii.md), [reference/pii.md](../reference/pii.md) + +## Context + +Step 1.7 detects + redacts PII at **ingest** (`PiiProcessor`). Step 6.1 resolves +each tenant's `pii_policy` (block / redact / allow) onto +`RequestContext.pii_policy`. But nothing enforced it at **egress** — the +`egress_text` PolicyDecision existed and the gateway already called it at every +generation boundary, yet the default `NoopPolicyEngine` answered `allow`, so PII +in a generated answer or in retrieved context sent to an external LLM left the +system unchecked. + +Step 6.5 closes the loop: enforce the per-tenant PII policy on text leaving the +system. + +## Decision + +**1. A decorator engine — `PiiPolicyEngine`.** Mirroring `QuotaPolicyEngine` +(Step 4.5) and `AclPolicyEngine` (Step 6.3), a thin decorator wraps an inner +`PolicyEngine` and answers the `egress_text` decision; everything else +(`filter_pushdown`, other decisions, `health`) delegates. It composes with the +ACL / quota PDP rather than absorbing it. It lives in **`rag-pii`** (which gains a +`rag-policy` dependency), exactly as `QuotaPolicyEngine` lives in `rag-quota` — +so it can reuse the package's detector + rewriters. + +**2. No new call site.** The gateway already calls +`policy_engine.evaluate(ctx, egress_text, subject)` at all four generation +boundaries. Three pass the retrieved **context** (`list[Chunk]` → the LLM on +`/v1/query?generate` / `/v1/chat/completions` / gRPC `Converse`); the agent passes +the final **answer** (`str`). The engine handles **both subject shapes**, so +enabling it enforces PII everywhere those calls already are — no route change, no +coverage-linter entry. + +**3. Action maps to PolicyResult.** Per `ctx.pii_policy.action`: `allow` → +delegate (pass through); `redact` / `mask` → `transform` (the same `redact_spans` +/ `mask_spans` rewriters the ingest processor uses); `block` → `deny` (the caller +already drops the answer / context on `deny`). Spans are filtered by the tenant's +`min_score` + `entities` before any action — identical policy semantics to the +ingest `PiiProcessor`, so the two stages never disagree. A clean scan is a no-op. + +**4. Detection via the injected `PIIDetector` SPI.** The default wiring injects +the dependency-free `RegexPIIDetector` (so enabling the feature actually detects +something); production injects `PresidioPIIDetector` or a custom backend. The +engine reads only `ctx.pii_policy` + the detector, so it stays independent of how +the policy was resolved. + +**5. Opt-in.** Gated by `cfg.pii.enabled` (default **false**) because it can +redact or withhold content. When off, the pre-6.5 behaviour (egress_text resolves +to the inner engine's allow) is unchanged. + +**6. PII-free events.** A `block` emits the pre-registered `pii.egress_blocked` +(the alertable denial); a `redact` / `mask` emits `pii.detected` (action +recorded). Both carry entity *types* + counts + the egress surface +(`egress:answer` / `egress:context`) — never the matched values. A clean egress is +silent. + +## Consequences + +**Positive** +- Per-tenant PII policy is now enforced at egress, reusing the ingest detector + + rewriters and the existing `egress_text` call sites (minimal surface, one PDP). +- `block` keeps PII off the wire entirely (context never reaches the external LLM; + answer withheld); `redact` / `mask` sanitise while still answering. +- Opt-in + composes with ACL / quota; off by default keeps existing deployments + unchanged. + +**Negative / deferred** +- Enforced at the existing `egress_text` sites: the **context** for + query / OpenAI / gRPC (sanitised before the external model) and the **answer** + for the agent. A post-generation answer re-check for query / OpenAI / gRPC, and + a PII check on retrieval-only citations (`/v1/query` without `generate`, + `/v1/retrieve`), are **deferred** — stored chunks are already sanitised at + ingest (Step 1.7), so those paths are covered when ingest-time PII is on. +- `block` on context is all-or-nothing (any PII denies the whole context); a + "drop only the PII-bearing chunks" middle ground is deferred. +- Detector quality bounds enforcement (regex by default); production should pair + with Presidio for name / location coverage. + +## See also +- [architecture/pii.md](../architecture/pii.md) — detection / enforcement split +- [architecture/policy-engine.md](../architecture/policy-engine.md) — the PDP + decorators +- [reference/pii.md](../reference/pii.md) — `PiiPolicyEngine` usage +- [reference/tenancy.md](../reference/tenancy.md) — per-tenant `pii_policy` diff --git a/docs/architecture/pii.md b/docs/architecture/pii.md index 026e428..3b5bb42 100644 --- a/docs/architecture/pii.md +++ b/docs/architecture/pii.md @@ -20,13 +20,14 @@ class. Two reasons we split: set by the gateway from `rag.yaml`. Splitting means swapping the detector doesn't touch policy logic. 2. **Reusability across pipeline stages.** The same detector that - runs at ingest time also runs at egress (Step 6.5 PII egress - verifier). One implementation, two call sites with different - policies — easy when detect and enforce are separate. + runs at ingest time also runs at egress (Step 6.5 `PiiPolicyEngine`). + One implementation, two call sites with different policies — easy + when detect and enforce are separate. -The `PiiProcessor` in `rag-pii` is the canonical ingest-time -enforcer. Egress enforcement (Step 6.5) will be a thin re-use of the -same detector with `action=block` hardcoded. +The `PiiProcessor` in `rag-pii` is the canonical ingest-time enforcer; +the `PiiPolicyEngine` (below) is its egress sibling — same detector + +rewriters, driven by the same per-tenant `PiiPolicy` rather than a +hardcoded action. ## RegexPIIDetector design @@ -96,6 +97,45 @@ PolicyEngine still gets the final say at ingest commit time (`PolicyEngine.evaluate(decision=ingest_doc)`) — but PII filtering happens upstream so the PDP sees already-cleaned content. +## Egress enforcement: PiiPolicyEngine (Step 6.5) + +Where ingest-time PII is a pipeline *stage*, egress-time PII is a +`PolicyEngine` *decorator* — `PiiPolicyEngine`, mirroring +`QuotaPolicyEngine` (Step 4.5) and `AclPolicyEngine` (Step 6.3). It answers +the `egress_text` decision the gateway already raises at every generation +boundary and applies the per-tenant action from `ctx.pii_policy`: + +``` +gateway generation path → policy_engine.evaluate(ctx, egress_text, subject) + ↓ subject is str (agent answer) or list[Chunk] (retrieved context) + detector.detect over the text(s) + filter by policy.entities + policy.min_score + allow → delegate · redact/mask → transform · block → deny + emit pii.egress_blocked (block) / pii.detected (redact|mask), PII-free +``` + +Two design points worth calling out: + +1. **Why a decorator, not a second processor call site.** The gateway's + four generation paths already call `egress_text`; making PII a + PolicyEngine decorator means it plugs into all of them with no new call + site and no coverage-linter entry — and it composes with the ACL / quota + decorators (non-`egress_text` decisions, `filter_pushdown`, and `health` + all delegate inward). Reusing `PiiProcessor` directly would have meant + editing each route. +2. **Why it lives in `rag-pii` (gaining a `rag-policy` dep).** The engine + needs the detector + rewriters (rag-pii) *and* the `PolicyEngine` base / + `PolicyResult` (rag-policy). Putting it in rag-pii — which now depends on + rag-policy, exactly as `rag-quota` does for `QuotaPolicyEngine` — keeps + the detector and its egress enforcer in one package. The graph stays + acyclic (rag-policy never imports rag-pii). + +The same `min_score` + `entities` filtering as the ingest processor means the +two stages never disagree on what counts as PII for a tenant. Block on a +`list[Chunk]` is all-or-nothing (any PII denies the whole context); redact / +mask rewrite each affected chunk's `content` (clean chunks pass through +untouched, same object). See [ADR-0037](../adr/ADR-0037-pii-egress-policies.md). + ## Event protocol: never log raw PII The `pii.detected` event protocol is designed so that **even if a @@ -144,9 +184,12 @@ for `ingest_doc` policy evaluation. The PolicyEngine coverage linter allowlists the PII processor's chunk-iteration call site (it consults `ctx.pii_policy` directly, not -the PDP — that's the design intent for PII specifically). -`rag-pii` depends only on `rag-core` and `rag-observability`, not -on `rag-policy`. +the PDP — that's the design intent for PII specifically). The +ingest-direction `PiiProcessor` depends only on `rag-core` + +`rag-observability`; the egress-direction `PiiPolicyEngine` (Step 6.5) +additionally depends on `rag-policy` (it *is* a `PolicyEngine`), so +`rag-pii` now depends on `rag-policy` — acyclic, since `rag-policy` +never imports `rag-pii` (the same shape as `rag-quota`). ## Hot-path discipline @@ -189,12 +232,13 @@ against synthetic PII strings. - ADR-0008 (cost-aware planner) — Presidio detection cost feeds the ingest cost estimate; planner can swap to regex-only mode under budget pressure. +- ADR-0037 (PII egress policies) — the egress-direction + `PiiPolicyEngine` decorator (Step 6.5) built on this detector. Future work tracked separately: -- **`pii.egress_blocked` event** is wired (Step 0.7b registered it) - but the egress enforcer lands in Step 6.5 (PII egress verifier). -- **`PiiAction.allow`** isn't currently in the enum — the closest - semantic is "no detection happened" which produces no event / - outcome. If a policy needs an explicit `allow` log entry, add the - enum value and the matching processor branch. +- **Post-generation answer re-check** for `/v1/query?generate` / + `/v1/chat/completions` / gRPC `Converse` (Step 6.5 sanitises the + *context* sent to the LLM on those paths; the agent path already + checks the final answer) and a PII check on retrieval-only citations + — deferred (stored chunks are already ingest-sanitised). diff --git a/docs/architecture/policy-engine.md b/docs/architecture/policy-engine.md index e250ea4..fe33df2 100644 --- a/docs/architecture/policy-engine.md +++ b/docs/architecture/policy-engine.md @@ -113,6 +113,18 @@ class MyOrgPolicyEngine(PolicyEngine): > disable both. Wired at the gateway as a `SupportsRoute` wrapper around the > retrieval router (`cfg.acl.verify_egress`, default on, gated by `enabled`); emits > `acl.egress_violation` on a caught leak. See [ADR-0036](../adr/ADR-0036-acl-egress-verifier.md). +> +> **PII egress enforcement (Step 6.5).** `rag_pii.PiiPolicyEngine` is the +> `egress_text` decorator: it scans the text leaving the system — retrieved +> context (`list[Chunk]`) and the agent's final answer (`str`) — and applies the +> per-tenant `ctx.pii_policy` action: `allow` delegates, `redact` / `mask` → +> `transform`, `block` → `deny`. It reuses the Step 1.7 detector + rewriters, +> filters by the tenant's `min_score` + `entities`, and plugs into the gateway's +> existing `egress_text` call sites (no new call site). Opt-in via +> `cfg.pii.enabled` (default off; injects `RegexPIIDetector`); emits +> `pii.egress_blocked` (block) / `pii.detected` (redact|mask), PII-free. Lives in +> `rag-pii` (which gains a `rag-policy` dep, like `rag-quota`). See +> [ADR-0037](../adr/ADR-0037-pii-egress-policies.md). Register at the composition root (`apps/gateway/`): @@ -130,4 +142,4 @@ Future built-ins on the roadmap: `OpaPolicyEngine` (delegates to an OPA sidecar) - [ADR-0005](../adr/ADR-0005-policy-engine.md) — the decision. - [request-context.md](request-context.md) — the envelope every policy call receives. - TRACKER.md Step 1.1c — implementation step. -- Steps 1.7, 4.5, 6.3, 6.5 — consumers; Step 6.4 ships the egress verifier as an independent second layer ([ADR-0036](../adr/ADR-0036-acl-egress-verifier.md)). +- Steps 1.7, 4.5, 6.3 — consumers; Step 6.4 ships the egress verifier as an independent second layer ([ADR-0036](../adr/ADR-0036-acl-egress-verifier.md)); Step 6.5 ships PII egress enforcement as a `PiiPolicyEngine` `egress_text` decorator ([ADR-0037](../adr/ADR-0037-pii-egress-policies.md)). diff --git a/docs/reference/pii.md b/docs/reference/pii.md index a91aae3..5f75c73 100644 --- a/docs/reference/pii.md +++ b/docs/reference/pii.md @@ -88,6 +88,42 @@ entity type via `rag-observability`. Events carry only metadata (type, chunk_id, count) — never raw matched text — so log sinks are safe even when redaction is disabled. +## PiiPolicyEngine — egress enforcement (Step 6.5) + +`PiiProcessor` runs at **ingest**. `PiiPolicyEngine` is its **egress** sibling: +a `PolicyEngine` decorator (like `QuotaPolicyEngine` / `AclPolicyEngine`) that +answers the `egress_text` decision over text about to leave the system, applying +the per-tenant `ctx.pii_policy` action. + +```python +from rag_pii import PiiPolicyEngine, RegexPIIDetector +from rag_policy import NoopPolicyEngine + +engine = PiiPolicyEngine(inner=NoopPolicyEngine(), detector=RegexPIIDetector()) +# evaluate(ctx, PolicyDecision.egress_text, subject) where subject is a str +# (the generated answer) or a list[Chunk] (the retrieved context). +``` + +| `pii_policy.action` | Result | Notes | +|--------|--------|-------| +| `allow` | `allow` | passes through unchanged | +| `redact` | `transform` | spans → `` markers (str or each chunk's content) | +| `mask` | `transform` | spans → `*` of equal length | +| `block` | `deny` | text withheld; the caller drops the answer / context | + +The gateway already calls `policy_engine.evaluate(ctx, egress_text, …)` at every +generation boundary — the retrieved context (`list[Chunk]`) on +`/v1/query?generate` / `/v1/chat/completions` / gRPC `Converse`, and the final +answer (`str`) on the agent path — so enabling the engine enforces PII at all of +them **without any route change**. Spans are filtered by the tenant's `min_score` ++ `entities` before an action is taken; a clean scan is a no-op. A **block** emits +a PII-free `pii.egress_blocked` event (entity types + counts only); a redact/mask +emits `pii.detected`. Non-`egress_text` decisions, `filter_pushdown`, and +`health` delegate to the inner engine, so it composes with the ACL / quota PDP. + +Enable via `cfg.pii.enabled` (off by default; the gateway injects +`RegexPIIDetector` unless a `pii_detector` is supplied). + ## Rewriters ```python diff --git a/docs/reference/tenancy.md b/docs/reference/tenancy.md index 504fd56..a34ec9c 100644 --- a/docs/reference/tenancy.md +++ b/docs/reference/tenancy.md @@ -99,7 +99,7 @@ lands in later Phase-6 steps: - **Physical tenancy** (a dedicated index per tenant) — **6.2** (below). - **ACL push-down** at retrieval (inject the labels into every backend query) — **6.3** ✅: enable `cfg.acl.enabled` to wrap the PolicyEngine in an `AclPolicyEngine` that And-merges `any_in("acl_labels", principal.acl_labels)` into every `read_chunk` push-down. **Fail-closed** (a label-less principal retrieves nothing; model "public" as a shared label) and emits `acl.egress_denied`. See [ADR-0035](../adr/ADR-0035-acl-pushdown.md). - **ACL egress verifier** — a post-retrieval re-check that backstops the push-down — **6.4** ✅ (below). -- **PII egress** enforcement (block / redact answers per policy) — **6.5**. +- **PII egress** enforcement (block / redact / allow per tenant) — **6.5** ✅: enable `cfg.pii.enabled` to wrap the PolicyEngine in a `PiiPolicyEngine` that answers `egress_text` over the context + answer leaving the system, applying the tenant's `pii_policy`; emits `pii.egress_blocked`. See [ADR-0037](../adr/ADR-0037-pii-egress-policies.md) and [reference/pii.md](pii.md). ## Physical tenancy (Step 6.2) diff --git a/packages/config/src/rag_config/schema.py b/packages/config/src/rag_config/schema.py index 750ae45..07fafa1 100644 --- a/packages/config/src/rag_config/schema.py +++ b/packages/config/src/rag_config/schema.py @@ -563,6 +563,26 @@ class AclConfig(_StrictBase): verify_egress: bool = True +class PiiConfig(_StrictBase): + """PII enforcement at egress (Step 6.5). + + When ``enabled`` the gateway wraps its PolicyEngine in a ``PiiPolicyEngine`` + that answers the ``egress_text`` decision: it scans the text about to leave the + system — the retrieved context sent to the LLM (``/v1/query?generate`` / + ``/v1/chat/completions`` / gRPC ``Converse``) and the agent's final answer — + and applies the **per-tenant** action resolved onto ``RequestContext.pii_policy`` + (Step 6.1, from ``tenants[].pii_policy``): **allow** passes through, **redact** / + **mask** sanitise the text (``transform``), and **block** withholds it (``deny``). + + **Disabled by default** — turning it on can redact or withhold answers. The + default detector is the dependency-free ``RegexPIIDetector`` (email / phone / + SSN / credit-card / IP); production injects Presidio. A **block** emits a + PII-free ``pii.egress_blocked`` event; a redact/mask emits ``pii.detected``. + """ + + enabled: bool = False + + class QuotaConfig(_StrictBase): """Per-tenant quota & rate-limit enforcement knobs (Step 4.5). @@ -874,6 +894,7 @@ class RagConfig(_StrictBase): breakers: BreakerConfig = Field(default_factory=BreakerConfig) quotas: QuotaConfig = Field(default_factory=QuotaConfig) acl: AclConfig = Field(default_factory=AclConfig) + pii: PiiConfig = Field(default_factory=PiiConfig) webhooks: WebhooksConfig = Field(default_factory=WebhooksConfig) provenance: ProvenanceConfig = Field(default_factory=ProvenanceConfig) feedback: FeedbackConfig = Field(default_factory=FeedbackConfig) diff --git a/packages/pii/pyproject.toml b/packages/pii/pyproject.toml index 0ae4f72..9e75525 100644 --- a/packages/pii/pyproject.toml +++ b/packages/pii/pyproject.toml @@ -11,6 +11,7 @@ requires-python = ">=3.12" dependencies = [ "rag-core", "rag-observability", + "rag-policy", ] [project.optional-dependencies] @@ -38,3 +39,4 @@ allow-direct-references = true [tool.uv.sources] rag-core = { workspace = true } rag-observability = { workspace = true } +rag-policy = { workspace = true } diff --git a/packages/pii/src/rag_pii/__init__.py b/packages/pii/src/rag_pii/__init__.py index 12fb519..98b55a4 100644 --- a/packages/pii/src/rag_pii/__init__.py +++ b/packages/pii/src/rag_pii/__init__.py @@ -8,10 +8,14 @@ ``[presidio]`` extra. - ``PiiProcessor`` — applies the per-tenant ``PiiPolicy`` to a chunk list using a configured detector; emits ``pii.detected`` events. +- ``PiiPolicyEngine`` — per-tenant PII enforcement at *egress* (Step 6.5): + a ``PolicyEngine`` decorator answering ``egress_text`` (allow / redact / + mask / block) over the answer or context leaving the system. - ``redact_spans`` / ``mask_spans`` — pure helpers that take text + spans and return the rewritten text. """ +from rag_pii.policy import PiiPolicyEngine from rag_pii.presidio import PresidioPIIDetector from rag_pii.processor import PiiProcessor from rag_pii.regex_detector import RegexPIIDetector @@ -20,6 +24,7 @@ __version__ = "0.1.0" __all__ = [ + "PiiPolicyEngine", "PiiProcessor", "PresidioPIIDetector", "RegexPIIDetector", diff --git a/packages/pii/src/rag_pii/policy.py b/packages/pii/src/rag_pii/policy.py new file mode 100644 index 0000000..2993812 --- /dev/null +++ b/packages/pii/src/rag_pii/policy.py @@ -0,0 +1,190 @@ +"""PiiPolicyEngine — per-tenant PII enforcement at egress (Step 6.5). + +Step 1.7 detects + redacts PII at *ingest*; Step 6.1 resolves each tenant's +``pii_policy`` onto ``RequestContext.pii_policy``. This engine closes the loop at +*egress*: a thin **decorator** (mirroring ``QuotaPolicyEngine`` / ``AclPolicyEngine``) +that answers the ``egress_text`` :class:`~rag_policy.PolicyDecision` by scanning the +text about to leave the system and applying the tenant's action — **allow** (pass +through), **redact** / **mask** (sanitise → ``transform``), or **block** (withhold +→ ``deny``). Everything else delegates to the wrapped inner engine, so it composes +with any production PDP. + +The gateway already calls ``policy_engine.evaluate(ctx, egress_text, subject)`` at +every generation boundary — over the retrieved **context** (``list[Chunk]`` sent to +the LLM on ``/v1/query?generate`` / ``/v1/chat/completions`` / gRPC ``Converse``) +and over the final **answer** (``str`` on the agent path). This engine handles +both subject shapes, so enforcement lands at all of them with **no new call site**. + +Detection uses the injected :class:`~rag_core.spi.pii_detector.PIIDetector` (the +default wiring is the dependency-free ``RegexPIIDetector``; production injects +Presidio); spans are filtered by the tenant's ``min_score`` + ``entities`` before +an action is taken. Events are PII-free: a **block** emits ``pii.egress_blocked`` +(the alertable denial), a **redact/mask** emits ``pii.detected`` (action recorded) +— both carry entity *types* and counts, never the matched values. A clean scan is +silent. +""" + +from __future__ import annotations + +from typing import Any, cast + +from rag_core.events import EVT_PII_DETECTED, EVT_PII_EGRESS_BLOCKED, PiiEvent +from rag_core.logging import get_logger +from rag_core.spi.pii_detector import PIIDetector +from rag_core.types import Chunk, PiiAction, PIISpan, RequestContext +from rag_policy import FilterExpr, PolicyDecision, PolicyEngine, PolicyResult + +from rag_pii.rewriters import mask_spans, redact_spans + +_log = get_logger(__name__) + +__all__ = ["PiiPolicyEngine"] + + +class PiiPolicyEngine(PolicyEngine): + """Decorate an inner :class:`PolicyEngine`, enforcing PII policy at egress (Step 6.5).""" + + def __init__(self, *, inner: PolicyEngine, detector: PIIDetector) -> None: + self._inner = inner + self._detector = detector + + @property + def inner(self) -> PolicyEngine: + return self._inner + + async def evaluate( + self, + ctx: RequestContext, + decision: PolicyDecision, + subject: Any, + ) -> PolicyResult: + if decision is not PolicyDecision.egress_text: + return await self._inner.evaluate(ctx, decision, subject) + action = ctx.pii_policy.action + if action is PiiAction.allow: + return await self._inner.evaluate(ctx, decision, subject) + if isinstance(subject, str): + return await self._eval_text(ctx, subject, action) + if isinstance(subject, list) and all(isinstance(c, Chunk) for c in subject): + return await self._eval_chunks(ctx, cast("list[Chunk]", subject), action) + # Unknown subject shape — leave it to the inner engine so we never break + # an egress check this engine isn't responsible for. + return await self._inner.evaluate(ctx, decision, subject) + + async def filter_pushdown( + self, + ctx: RequestContext, + decision: PolicyDecision, + ) -> FilterExpr: + return await self._inner.filter_pushdown(ctx, decision) + + async def health(self) -> bool: + return await self._inner.health() and await self._detector.health() + + # ------------------------------------------------------------------ + # Internals + # ------------------------------------------------------------------ + async def _filtered_spans(self, ctx: RequestContext, text: str) -> list[PIISpan]: + """Detect PII, then keep only spans the tenant policy cares about.""" + policy = ctx.pii_policy + spans = await self._detector.detect(ctx, text) + return [ + s + for s in spans + if s.score >= policy.min_score + and (not policy.entities or s.entity_type in policy.entities) + ] + + async def _eval_text(self, ctx: RequestContext, text: str, action: PiiAction) -> PolicyResult: + if not text: + return PolicyResult.allow() + spans = await self._filtered_spans(ctx, text) + if not spans: + return PolicyResult.allow() + types = _types(spans) + if action is PiiAction.block: + _emit(ctx, EVT_PII_EGRESS_BLOCKED, surface="answer", types=types, span_count=len(spans)) + return PolicyResult.deny(f"pii egress blocked: {','.join(types)}") + _emit( + ctx, + EVT_PII_DETECTED, + surface="answer", + types=types, + span_count=len(spans), + action=action.value, + ) + return PolicyResult.transform(_rewrite(text, spans, action)) + + async def _eval_chunks( + self, ctx: RequestContext, chunks: list[Chunk], action: PiiAction + ) -> PolicyResult: + per_chunk: list[list[PIISpan]] = [] + all_types: set[str] = set() + total = 0 + for chunk in chunks: + spans = await self._filtered_spans(ctx, chunk.content) if chunk.content else [] + per_chunk.append(spans) + all_types.update(s.entity_type for s in spans) + total += len(spans) + if total == 0: + return PolicyResult.allow() + types = tuple(sorted(all_types)) + if action is PiiAction.block: + _emit(ctx, EVT_PII_EGRESS_BLOCKED, surface="context", types=types, span_count=total) + return PolicyResult.deny(f"pii egress blocked: {','.join(types)}") + rewritten: list[Chunk] = [ + chunk.model_copy(update={"content": _rewrite(chunk.content, spans, action)}) + if spans and chunk.content + else chunk + for chunk, spans in zip(chunks, per_chunk, strict=True) + ] + _emit( + ctx, + EVT_PII_DETECTED, + surface="context", + types=types, + span_count=total, + action=action.value, + ) + return PolicyResult.transform(rewritten) + + +def _types(spans: list[PIISpan]) -> tuple[str, ...]: + return tuple(sorted({s.entity_type for s in spans})) + + +def _rewrite(text: str, spans: list[PIISpan], action: PiiAction) -> str: + # mask preserves length with '*'; redact (and any non-block fallback) inserts + # ```` markers — the same rewriters the ingest PiiProcessor uses. + if action is PiiAction.mask: + return mask_spans(text, spans) + return redact_spans(text, spans) + + +def _emit( + ctx: RequestContext, + event_name: str, + *, + surface: str, + types: tuple[str, ...], + span_count: int, + action: str = "block", +) -> None: + """Emit one PII-free egress event — entity types + counts, never raw values.""" + event = PiiEvent( + event_name=event_name, + tenant_id=str(ctx.tenant_id), + field_name=f"egress:{surface}", + pii_type=",".join(types), + action_taken=action, + metadata={ + "principal_id": str(ctx.principal.id) if ctx.principal else None, + "span_count": span_count, + "entity_types": list(types), + "surface": surface, + }, + ) + if event_name == EVT_PII_EGRESS_BLOCKED: + _log.warning("pii.egress_blocked", extra={"event": event.model_dump()}) + else: + _log.info("pii.detected", extra={"event": event.model_dump()}) diff --git a/tests/pii/test_policy.py b/tests/pii/test_policy.py new file mode 100644 index 0000000..68472bf --- /dev/null +++ b/tests/pii/test_policy.py @@ -0,0 +1,190 @@ +"""Unit tests for PiiPolicyEngine — PII enforcement at egress (Step 6.5).""" + +from __future__ import annotations + +from typing import Any + +import pytest +from rag_core.types import Chunk, PiiAction +from rag_pii import PiiPolicyEngine, RegexPIIDetector +from rag_policy import AclPolicyEngine, NoopPolicyEngine, PolicyDecision, evaluate + +_EMAIL = "reach me at alice@example.com please" +_PHONE = "call me on 555-123-4567" +_CLEAN = "the quick brown fox jumps over the lazy dog" + + +def _engine(inner: Any | None = None) -> PiiPolicyEngine: + return PiiPolicyEngine(inner=inner or NoopPolicyEngine(), detector=RegexPIIDetector()) + + +# --------------------------------------------------------------------------- +# delegation — non-egress decisions + filter_pushdown + health +# --------------------------------------------------------------------------- +async def test_non_egress_decision_delegates(make_ctx: Any) -> None: + res = await _engine().evaluate(make_ctx(), PolicyDecision.read_chunk, "anything") + assert res.is_allow() + + +async def test_filter_pushdown_delegates_to_inner(make_ctx: Any) -> None: + # PII wraps ACL → the ACL read_chunk push-down must still flow through. + pii = _engine(inner=AclPolicyEngine(inner=NoopPolicyEngine())) + ctx = make_ctx() + ctx = ctx.model_copy( + update={"principal": ctx.principal.model_copy(update={"acl_labels": frozenset(["eng"])})} + ) + f = await pii.filter_pushdown(ctx, PolicyDecision.read_chunk) + assert evaluate(f, {"tenant_id": str(ctx.tenant_id), "acl_labels": frozenset(["eng"])}) is True + assert evaluate(f, {"tenant_id": str(ctx.tenant_id), "acl_labels": frozenset(["hr"])}) is False + + +async def test_health_delegates() -> None: + assert await _engine().health() is True + + +# --------------------------------------------------------------------------- +# allow + no-PII = passthrough +# --------------------------------------------------------------------------- +async def test_allow_action_passes_pii_through(make_ctx: Any) -> None: + res = await _engine().evaluate( + make_ctx(action=PiiAction.allow), PolicyDecision.egress_text, _EMAIL + ) + assert res.is_allow() + + +async def test_clean_text_passes_through(make_ctx: Any) -> None: + res = await _engine().evaluate( + make_ctx(action=PiiAction.redact), PolicyDecision.egress_text, _CLEAN + ) + assert res.is_allow() + + +# --------------------------------------------------------------------------- +# str subject (agent answer) +# --------------------------------------------------------------------------- +async def test_str_redact_transforms(make_ctx: Any) -> None: + res = await _engine().evaluate( + make_ctx(action=PiiAction.redact), PolicyDecision.egress_text, _EMAIL + ) + assert res.is_transform() + assert "alice@example.com" not in res.transformed + assert "" in res.transformed + + +async def test_str_mask_transforms(make_ctx: Any) -> None: + res = await _engine().evaluate( + make_ctx(action=PiiAction.mask), PolicyDecision.egress_text, _EMAIL + ) + assert res.is_transform() + assert "alice@example.com" not in res.transformed + assert "*" in res.transformed + + +async def test_str_block_denies(make_ctx: Any) -> None: + res = await _engine().evaluate( + make_ctx(action=PiiAction.block), PolicyDecision.egress_text, _EMAIL + ) + assert res.is_deny() + assert "EMAIL" in (res.reason or "") # type name only — never the raw value + assert "alice@example.com" not in (res.reason or "") + + +# --------------------------------------------------------------------------- +# list[Chunk] subject (retrieved context) +# --------------------------------------------------------------------------- +async def test_chunks_redact_transforms(make_ctx: Any, make_chunk: Any) -> None: + chunks = [make_chunk(_EMAIL, chunk_id="c1"), make_chunk(_CLEAN, chunk_id="c2")] + res = await _engine().evaluate( + make_ctx(action=PiiAction.redact), PolicyDecision.egress_text, chunks + ) + assert res.is_transform() + out: list[Chunk] = res.transformed + assert out[0].content is not None and "alice@example.com" not in out[0].content + assert "" in out[0].content + # the clean chunk is returned unchanged (same object, no needless copy) + assert out[1] is chunks[1] + + +async def test_chunks_block_denies_when_any_chunk_has_pii(make_ctx: Any, make_chunk: Any) -> None: + chunks = [make_chunk(_CLEAN), make_chunk(_PHONE)] + res = await _engine().evaluate( + make_ctx(action=PiiAction.block), PolicyDecision.egress_text, chunks + ) + assert res.is_deny() + + +async def test_chunks_clean_passes_through(make_ctx: Any, make_chunk: Any) -> None: + chunks = [make_chunk(_CLEAN), make_chunk("nothing sensitive here")] + res = await _engine().evaluate( + make_ctx(action=PiiAction.redact), PolicyDecision.egress_text, chunks + ) + assert res.is_allow() + + +# --------------------------------------------------------------------------- +# per-tenant policy knobs — min_score + entities filter +# --------------------------------------------------------------------------- +async def test_min_score_below_threshold_is_ignored(make_ctx: Any) -> None: + # PHONE detector score is 0.7; a 0.8 floor drops it → nothing to redact. + res = await _engine().evaluate( + make_ctx(action=PiiAction.redact, min_score=0.8), PolicyDecision.egress_text, _PHONE + ) + assert res.is_allow() + + +async def test_entities_filter_scopes_detection(make_ctx: Any) -> None: + # Policy only cares about EMAIL → a phone number passes through untouched. + res = await _engine().evaluate( + make_ctx(action=PiiAction.block, entities=frozenset({"EMAIL"})), + PolicyDecision.egress_text, + _PHONE, + ) + assert res.is_allow() + + +# --------------------------------------------------------------------------- +# events — PII-free +# --------------------------------------------------------------------------- +def _capture(monkeypatch: pytest.MonkeyPatch) -> dict[str, list[dict[str, Any]]]: + import rag_pii.policy as policy_mod + + calls: dict[str, list[dict[str, Any]]] = {"warning": [], "info": []} + monkeypatch.setattr( + policy_mod._log, "warning", lambda msg, **kw: calls["warning"].append({"msg": msg, **kw}) + ) + monkeypatch.setattr( + policy_mod._log, "info", lambda msg, **kw: calls["info"].append({"msg": msg, **kw}) + ) + return calls + + +async def test_block_emits_pii_egress_blocked( + make_ctx: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + calls = _capture(monkeypatch) + await _engine().evaluate(make_ctx(action=PiiAction.block), PolicyDecision.egress_text, _EMAIL) + blocked = [c for c in calls["warning"] if c["msg"] == "pii.egress_blocked"] + assert len(blocked) == 1 + event = blocked[0]["extra"]["event"] + assert event["event_name"] == "pii.egress_blocked" + assert event["action_taken"] == "block" + assert event["field_name"] == "egress:answer" + # PII-free: the dumped event carries entity types, never the raw value. + assert "alice@example.com" not in str(event) + assert "EMAIL" in event["pii_type"] + + +async def test_redact_emits_pii_detected(make_ctx: Any, monkeypatch: pytest.MonkeyPatch) -> None: + calls = _capture(monkeypatch) + await _engine().evaluate(make_ctx(action=PiiAction.redact), PolicyDecision.egress_text, _EMAIL) + detected = [c for c in calls["info"] if c["msg"] == "pii.detected"] + assert len(detected) == 1 + event = detected[0]["extra"]["event"] + assert event["action_taken"] == "redact" + assert "alice@example.com" not in str(event) + + +async def test_clean_egress_is_silent(make_ctx: Any, monkeypatch: pytest.MonkeyPatch) -> None: + calls = _capture(monkeypatch) + await _engine().evaluate(make_ctx(action=PiiAction.redact), PolicyDecision.egress_text, _CLEAN) + assert calls["warning"] == [] and calls["info"] == [] diff --git a/uv.lock b/uv.lock index 66e8fde..30d0e4f 100644 --- a/uv.lock +++ b/uv.lock @@ -7566,6 +7566,7 @@ source = { editable = "packages/pii" } dependencies = [ { name = "rag-core" }, { name = "rag-observability" }, + { name = "rag-policy" }, ] [package.optional-dependencies] @@ -7589,6 +7590,7 @@ requires-dist = [ { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=1.3" }, { name = "rag-core", editable = "packages/core" }, { name = "rag-observability", editable = "packages/observability" }, + { name = "rag-policy", editable = "packages/policy" }, { name = "spacy", marker = "extra == 'presidio'", specifier = ">=3.7" }, ] provides-extras = ["presidio", "dev"]