Scope boundary

Owns runtime verification and fail-closed enforcement of the signed context-source governance release, binding every candidate chunk to approved source identity, content digest, release, and current freshness before context assembly.

AID-H-021.003: AID-H-020.004 enforces the current signed context-governance release on retrieval; AID-H-021.003 defines source ownership, trust tier, precedence, freshness policy, code-to-document synchronization, and release eligibility. The retrieval gate consumes the policy rather than authoring it.

AID-H-020.001: AID-H-020.004 verifies source approval and freshness; AID-H-020.001 verifies chunk and embedding byte integrity. Fresh approved context can be tampered, and intact context can be stale.

AID-H-020.002: AID-H-020.004 fails closed on unlisted, mismatched, or stale sources; AID-H-020.002 adjusts ranking according to calibrated source reputation. Reputation is not an admission override.

AID-H-020.003: AID-H-020.004 enforces context-source policy and freshness; AID-H-020.003 enforces principal-to-chunk entitlement. Both policy dimensions must pass independently.

Implementation guidance

• AID-H-020.004-G001: Verify and enforce the signed context governance release against the complete retrieval candidate population before context assembly.

AID-H-020.004-G001

Verify policy state and filter candidates

# File: rag_policy/enforce_context_release.py
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path


MANIFEST = Path(os.environ["CONTEXT_GOVERNANCE_MANIFEST"])
SIGNATURE = Path(os.environ["CONTEXT_GOVERNANCE_SIGNATURE"])
VERIFY_KEY = os.environ["CONTEXT_GOVERNANCE_VERIFY_KEY"]
COSIGN_TIMEOUT_SECONDS = float(os.environ["VERIFIED_CONTEXT_GOVERNANCE_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(COSIGN_TIMEOUT_SECONDS) or COSIGN_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("verified context-governance timeout must be finite and positive")
REQUIRED_MANIFEST = {
    "schema_version", "release_id", "sources_sha256",
    "freshness_report_sha256", "precedence_policy_sha256",
    "source_count", "source_states",
}
REQUIRED_STATE = {"source_id", "content_sha256", "expires_at"}


def parse_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("source expiry lacks timezone")
    return parsed.astimezone(timezone.utc)


def strict_json_object(raw: bytes) -> dict:
    def unique_pairs(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate signed-JSON key: {key}")
            value[key] = item
        return value

    def reject_constant(value):
        raise ValueError(f"non-finite signed-JSON number: {value}")

    document = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_pairs, parse_constant=reject_constant,
    )
    if not isinstance(document, dict):
        raise ValueError("signed JSON root must be an object")
    return document


def load_verified_manifest() -> tuple[dict, str]:
    if not MANIFEST.is_file() or not SIGNATURE.is_file():
        raise LookupError("governance artifact is missing")
    raw = MANIFEST.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h020-manifest-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run([
            "cosign", "verify-blob", "--key", VERIFY_KEY,
            "--bundle", str(SIGNATURE), str(snapshot),
        ], check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS)
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    manifest = strict_json_object(raw)
    if not isinstance(manifest, dict) or set(manifest) != REQUIRED_MANIFEST:
        raise ValueError("context governance manifest schema mismatch")
    if manifest["schema_version"] != "aidefend.context-governance-release.v1":
        raise ValueError("unsupported context governance manifest")
    states = manifest["source_states"]
    if not isinstance(states, list) or len(states) != manifest["source_count"]:
        raise ValueError("context source population is incomplete")
    if any(not isinstance(state, dict) or set(state) != REQUIRED_STATE for state in states):
        raise ValueError("context source state schema mismatch")
    source_ids = [state["source_id"] for state in states]
    if source_ids != sorted(set(source_ids)):
        raise ValueError("context source state IDs must be sorted and unique")
    return manifest, hashlib.sha256(raw).hexdigest()


def enforce(request_id: str, candidates: list[dict]) -> dict:
    try:
        manifest, manifest_sha256 = load_verified_manifest()
        states = {item["source_id"]: item for item in manifest["source_states"]}
        now = datetime.now(timezone.utc)
        admitted: list[str] = []
        denied: list[dict] = []
        seen: set[str] = set()
        for candidate in candidates:
            required = {
                "chunk_id", "source_id", "source_content_sha256",
                "context_release_id",
            }
            if not isinstance(candidate, dict) or set(candidate) != required:
                return {"status": "ERROR", "reason": "candidate schema mismatch"}
            if candidate["chunk_id"] in seen:
                return {"status": "ERROR", "reason": "duplicate chunk_id"}
            seen.add(candidate["chunk_id"])
            state = states.get(candidate["source_id"])
            reasons: list[str] = []
            if state is None:
                reasons.append("source_not_in_signed_release")
            else:
                if candidate["context_release_id"] != manifest["release_id"]:
                    reasons.append("context_release_mismatch")
                if candidate["source_content_sha256"] != state["content_sha256"]:
                    reasons.append("source_digest_mismatch")
                if now >= parse_time(state["expires_at"]):
                    reasons.append("source_state_expired")
            if reasons:
                denied.append({"chunk_id": candidate["chunk_id"], "reasons": reasons})
            else:
                admitted.append(candidate["chunk_id"])
        status = "PASS" if not denied else "FAIL"
        return {
            "schema_version": "aidefend.h020004-decision.v1",
            "status": status,
            "request_id": request_id,
            "context_release_id": manifest["release_id"],
            "manifest_sha256": manifest_sha256,
            "candidate_count": len(candidates),
            "admitted_chunk_ids": admitted,
            "denied": denied,
        }
    except LookupError as exc:
        return {"status": "INSUFFICIENT_DATA", "reason": str(exc)}
    except (FileNotFoundError, json.JSONDecodeError, subprocess.SubprocessError, TypeError, ValueError) as exc:
        return {"status": "ERROR", "reason": str(exc)}

Enforcement and evidence

Call this function after candidate retrieval and before context assembly. The decision is request-atomic: if any candidate is denied, the result is FAIL and none of the request's chunks may proceed; admitted_chunk_ids in a FAIL result is diagnostic only and must not be assembled. Only a PASS result proceeds to the separate H-020.001 integrity and H-020.003 entitlement gates; INSUFFICIENT_DATA and ERROR do not enter model context. Write the exact manifest/signature digests, request ID, complete candidate population, source-state comparisons, denied reasons, and final context-assembly digest to append-only evidence storage outside the retriever identity.

A read-only verifier replays the same candidate set and signed manifest at the recorded decision time. It must reproduce the admitted/denied sets and confirm that the actual assembled context contains only admitted chunk IDs. Missing candidates or unavailable evidence cannot be interpreted as a pass.

Action: Call the enforcement function on the complete candidate set before context assembly and reject every non-PASS result.

Open-source tools

• Cosign
• python-jsonschema
• Open Policy Agent (OPA)
• OpenTelemetry
• LlamaIndex

Threat mappings

MITRE ATLAS

• AML.T0099 AI Agent Tool Data Poisoning
• AML.T0070 RAG Poisoning
• AML.T0071 False RAG Entry Injection
• AML.T0066 Retrieval Content Crafting
• AML.T0051.001 LLM Prompt Injection: Indirect (stale or unapproved context is rejected before assembly)

MAESTRO

• Compromised RAG Pipelines (L2)
• Data Tampering (L2)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (approved-source and freshness enforcement reduces exposure to injected retrieved content but does not inspect instruction semantics)
• LLM05:2026 Data and Model Poisoning
• LLM07:2026 Misinformation

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning
• ASI01:2026 Agent Goal Hijack

NIST Adversarial Machine Learning 2025

• NISTAML.015 Indirect Prompt Injection

Cisco Integrated AI Security and Safety Framework

• AISubtech-6.1.1 Knowledge Base Poisoning
• AITech-1.2 Indirect Prompt Injection
• AITech-7.3 Data Source Abuse and Manipulation

Google Secure AI Framework 2.0 - Risks

• PIJ: Prompt Injection
• IIC: Insecure Integrated Component

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning
• Agents - Core 13.6: Intent Breaking & Goal Manipulation