AID-H-020.003: Document/Chunk-Level Permission-Aware Retrieval

Version 1.20261015 · Actionable sub-technique

Enforce document- and chunk-level permissions at retrieval time so a RAG pipeline never assembles context from records the requesting principal is not entitled to read. The control mirrors source-system ACLs or relationship-based authorization metadata into chunk records during ingestion, evaluates those permissions on every query, and drops unauthorized chunks before context construction.

Pillars: data, app

Lifecycle: building, operation

Scope boundary

Owns query-time principal-to-document and principal-to-chunk entitlement in RAG, using source-system ACLs or relationship metadata mirrored into the index and enforced before context construction.

AID-H-020.001: AID-H-020.003 proves the caller may read the chunk; AID-H-020.001 proves the chunk and embedding are authentic and unchanged. Authorization and integrity must both pass.

AID-H-020.004: AID-H-020.003 enforces caller entitlement; AID-H-020.004 enforces source approval, release binding, and freshness under the current context-governance state. An entitled user may still be denied stale or unapproved context.

AID-I-004.002: AID-H-020.003 filters documents and chunks for a principal at retrieval time; AID-I-004.002 partitions persistent memory and knowledge namespaces by tenant and trust zone. Namespace isolation limits addressability, while entitlement applies record-level source permissions.

AID-H-029: AID-H-020.003 answers whether the principal may read the content; AID-H-029 answers whether the data may be used for this purpose, lifecycle stage, consent state, and retention context. Read access is not universal data-use authorization.

Implementation guidance

AID-H-020.003-G001: Mirror source-system ACLs or relationship tuples into chunk metadata at ingestion time, and fail ingestion when permission metadata is missing.
AID-H-020.003-G002: Apply query-time entitlement filters before context assembly and deny when the retriever cannot prove the caller may read the chunk.
AID-H-020.003-G003: Continuously test permission drift by replaying representative principals against the RAG index and comparing retrieval results with source-system access truth.

AID-H-020.003-G001

Mirror source-system ACLs or relationship tuples into chunk metadata at ingestion time, and fail ingestion when permission metadata is missing.

Concept:

A retriever cannot enforce permissions it never indexed. During ingestion, copy the source document ACL or relationship tuples into every chunk's metadata. Missing ACL metadata should fail closed instead of creating public chunks by accident.

Step 1: Normalize document permissions

# File: rag_authz/acl_ingestion.py
from __future__ import annotations

from dataclasses import dataclass, asdict
from typing import Iterable


@dataclass(frozen=True)
class DocumentAcl:
    source_doc_id: str
    owner_id: str
    allowed_principals: list[str]
    allowed_groups: list[str]
    acl_version: str


class MissingAclError(ValueError):
    pass


def normalize_acl(raw_acl: dict) -> DocumentAcl:
    required = {"source_doc_id", "owner_id", "allowed_principals", "allowed_groups", "acl_version"}
    if not isinstance(raw_acl, dict) or set(raw_acl) != required:
        raise MissingAclError("ACL schema mismatch")
    for field in ("source_doc_id", "owner_id", "acl_version"):
        if not isinstance(raw_acl[field], str) or not raw_acl[field].strip():
            raise MissingAclError(f"{field} must be a nonempty string")
    for field in ("allowed_principals", "allowed_groups"):
        values = raw_acl[field]
        if not isinstance(values, list) or any(
            not isinstance(value, str) or not value.strip() for value in values
        ):
            raise MissingAclError(f"{field} must be a list of nonempty strings")
    return DocumentAcl(
        source_doc_id=raw_acl["source_doc_id"],
        owner_id=raw_acl["owner_id"],
        allowed_principals=sorted(set(raw_acl["allowed_principals"])),
        allowed_groups=sorted(set(raw_acl["allowed_groups"])),
        acl_version=raw_acl["acl_version"],
    )


def attach_acl_to_chunks(chunks: Iterable[dict], acl: DocumentAcl) -> list[dict]:
    result = []
    seen_chunk_ids: set[str] = set()
    for chunk in chunks:
        if not isinstance(chunk, dict) or not isinstance(chunk.get("id"), str) or not chunk["id"]:
            raise MissingAclError("every chunk requires a nonempty string id")
        if chunk["id"] in seen_chunk_ids:
            raise MissingAclError("duplicate chunk id")
        seen_chunk_ids.add(chunk["id"])
        raw_metadata = chunk.get("metadata", {})
        if not isinstance(raw_metadata, dict):
            raise MissingAclError("chunk metadata must be an object")
        if set(raw_metadata).intersection({
            "source_doc_id", "acl_owner_id", "acl_allowed_principals",
            "acl_allowed_groups", "acl_version",
        }):
            raise MissingAclError("caller-supplied ACL metadata is forbidden")
        metadata = dict(raw_metadata)
        metadata.update({
            "source_doc_id": acl.source_doc_id,
            "acl_owner_id": acl.owner_id,
            "acl_allowed_principals": acl.allowed_principals,
            "acl_allowed_groups": acl.allowed_groups,
            "acl_version": acl.acl_version,
        })
        result.append({**chunk, "metadata": metadata})
    return result

Step 2: Reject chunks with incomplete ACL metadata

# File: rag_authz/verify_chunk_acl_metadata.py
from __future__ import annotations

import json
from pathlib import Path


REQUIRED_ACL_FIELDS = {
    "source_doc_id",
    "acl_owner_id",
    "acl_allowed_principals",
    "acl_allowed_groups",
    "acl_version",
}


def verify_chunks(chunks: list[dict]) -> dict:
    failures = []
    seen_ids = set()
    for chunk in chunks:
        if not isinstance(chunk, dict) or not isinstance(chunk.get("id"), str) or not chunk["id"]:
            failures.append({"chunk_id": None, "invalid": ["chunk_schema"]})
            continue
        if chunk["id"] in seen_ids:
            failures.append({"chunk_id": chunk["id"], "invalid": ["duplicate_chunk_id"]})
            continue
        seen_ids.add(chunk["id"])
        metadata = chunk.get("metadata", {})
        if not isinstance(metadata, dict):
            failures.append({"chunk_id": chunk["id"], "invalid": ["metadata_schema"]})
            continue
        missing = sorted(field for field in REQUIRED_ACL_FIELDS if field not in metadata)
        if missing:
            failures.append({"chunk_id": chunk.get("id"), "missing": missing})
            continue
        invalid = []
        for field in ("source_doc_id", "acl_owner_id", "acl_version"):
            if not isinstance(metadata[field], str) or not metadata[field]:
                invalid.append(field)
        for field in ("acl_allowed_principals", "acl_allowed_groups"):
            if not isinstance(metadata[field], list) or any(
                not isinstance(value, str) or not value for value in metadata[field]
            ):
                invalid.append(field)
        if invalid:
            failures.append({"chunk_id": chunk["id"], "invalid": invalid})
    return {
        "schema_version": "aidefend.rag_acl_ingestion.v1",
        "chunk_count": len(chunks),
        "failure_count": len(failures),
        "failures": failures,
        "status": "PASS" if not failures and chunks else "FAIL",
    }


chunks = json.loads(Path("out/rag_chunks_pending_index.json").read_text(encoding="utf-8"))
artifact = verify_chunks(chunks)
Path("artifacts").mkdir(exist_ok=True)
Path("artifacts/rag-acl-ingestion-evidence.json").write_text(
    json.dumps(artifact, indent=2, sort_keys=True),
    encoding="utf-8",
)
if artifact["status"] != "PASS":
    raise SystemExit("RAG ACL metadata verification failed")

Action: Make ACL metadata a mandatory ingestion field for every indexed chunk. The evidence artifact for this guidance is artifacts/rag-acl-ingestion-evidence.json, proving that no chunk entered the index without entitlement metadata.

AID-H-020.003-G002

Apply query-time entitlement filters before context assembly and deny when the retriever cannot prove the caller may read the chunk.

Concept:

Vector similarity is not authorization. Retrieve candidate chunks, then enforce document/chunk entitlements against the authenticated principal before any text reaches the model context. Fail closed on missing principal, missing ACL metadata, stale ACL version, or policy-engine errors.

Step 1: Implement a permission filter

# File: rag_authz/retrieval_filter.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class Principal:
    principal_id: str
    groups: set[str]


PolicyDecision = Callable[[str, set[str], str, str], bool]


@dataclass(frozen=True)
class PolicyEngine:
    decide: PolicyDecision

    def allowed(self, *, principal_id: str, groups: set[str], source_doc_id: str, action: str) -> bool:
        decision = self.decide(principal_id, set(groups), source_doc_id, action)
        if type(decision) is not bool:
            raise RuntimeError("policy adapter returned a non-boolean decision")
        return decision


class RetrievalDenied(PermissionError):
    pass


class RetrievalPolicyError(RuntimeError):
    pass


def acl_is_fresh(metadata: dict, current_acl_versions: dict[str, str]) -> bool:
    source_doc_id = metadata.get("source_doc_id")
    acl_version = metadata.get("acl_version")
    if any(not isinstance(value, str) or not value.strip()
           for value in (source_doc_id, acl_version)):
        return False
    if source_doc_id not in current_acl_versions:
        return False
    current_version = current_acl_versions[source_doc_id]
    if not isinstance(current_version, str) or not current_version.strip():
        return False
    return current_version == acl_version


def principal_can_read(
    metadata: dict,
    principal: Principal,
    current_acl_versions: dict[str, str],
    policy_engine: PolicyEngine | None = None,
) -> bool:
    source_doc_id = metadata.get("source_doc_id")
    if not source_doc_id or not acl_is_fresh(metadata, current_acl_versions):
        return False

    if policy_engine is not None:
        try:
            decision = policy_engine.allowed(
                principal_id=principal.principal_id,
                groups=principal.groups,
                source_doc_id=str(source_doc_id),
                action="read",
            )
        except Exception as exc:
            raise RetrievalPolicyError("policy_engine_error") from exc
        if type(decision) is not bool:
            raise RetrievalPolicyError("policy_engine_returned_non_boolean")
        return decision

    raw_principals = metadata.get("acl_allowed_principals")
    raw_groups = metadata.get("acl_allowed_groups")
    if not isinstance(raw_principals, list) or not isinstance(raw_groups, list):
        return False
    if any(not isinstance(value, str) for value in raw_principals + raw_groups):
        return False
    allowed_principals = set(raw_principals)
    allowed_groups = set(raw_groups)
    owner_id = metadata.get("acl_owner_id")
    if principal.principal_id == owner_id:
        return True
    if principal.principal_id in allowed_principals:
        return True
    if principal.groups.intersection(allowed_groups):
        return True
    return False


def filter_authorized_chunks(
    candidates: list[dict],
    principal: Principal,
    current_acl_versions: dict[str, str],
    policy_engine: PolicyEngine | None = None,
) -> tuple[list[dict], list[dict]]:
    allowed = []
    denied = []
    for chunk in candidates:
        metadata = chunk.get("metadata", {})
        if principal_can_read(metadata, principal, current_acl_versions, policy_engine):
            allowed.append(chunk)
        else:
            denied.append({
                "chunk_id": chunk.get("id"),
                "source_doc_id": metadata.get("source_doc_id"),
                "reason": "principal_not_entitled_acl_missing_or_stale",
            })
    return allowed, denied

Step 2: Enforce before model context construction

# File: rag_authz/secure_retriever.py
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable

from rag_authz.retrieval_filter import (
    PolicyEngine, Principal, RetrievalPolicyError, filter_authorized_chunks,
)


AppendAuditEvent = Callable[[dict], dict]


@dataclass(frozen=True)
class AuditWriter:
    append_event: AppendAuditEvent

    def write(self, event: dict) -> None:
        receipt = self.append_event(event)
        if (
            not isinstance(receipt, dict)
            or receipt.get("status") != "PASS"
            or not isinstance(receipt.get("event_sha256"), str)
            or len(receipt["event_sha256"]) != 64
        ):
            raise RuntimeError("append-only audit sink did not confirm exact event readback")


def retrieve_authorized_context(
    vector_store,
    query: str,
    principal: Principal,
    current_acl_versions: dict[str, str],
    audit_writer: AuditWriter,
    policy_engine: PolicyEngine | None = None,
    *,
    k: int = 10,
) -> list[dict]:
    if (
        not isinstance(principal, Principal)
        or not principal.principal_id
        or not isinstance(principal.groups, set)
        or any(not isinstance(group, str) or not group for group in principal.groups)
    ):
        raise PermissionError("missing_principal")
    if not isinstance(current_acl_versions, dict) or any(
        not isinstance(key, str) or not isinstance(value, str)
        for key, value in current_acl_versions.items()
    ):
        raise RetrievalPolicyError("invalid_acl_version_snapshot")

    candidates = vector_store.similarity_search(query, k=k)
    if not isinstance(candidates, list):
        raise RetrievalPolicyError("retriever_returned_invalid_population")
    try:
        allowed, denied = filter_authorized_chunks(
            candidates, principal, current_acl_versions, policy_engine
        )
    except RetrievalPolicyError as exc:
        audit_writer.write({
            "event_type": "rag_retrieval_authz_decision",
            "status": "ERROR",
            "reason": str(exc),
            "principal_id": principal.principal_id,
            "candidate_count": len(candidates),
        })
        raise
    candidate_ids = [str(candidate.get("id")) for candidate in candidates]
    evidence = {
        "event_type": "rag_retrieval_authz_decision",
        "status": "PASS",
        "observed_at": datetime.now(timezone.utc).isoformat(),
        "principal_id": principal.principal_id,
        "candidate_count": len(candidates),
        "candidate_population_sha256": hashlib.sha256(
            json.dumps(candidate_ids, separators=(",", ":")).encode()
        ).hexdigest(),
        "allowed_count": len(allowed),
        "allowed_chunk_ids": [str(chunk.get("id")) for chunk in allowed],
        "denied_count": len(denied),
        "acl_versions_checked": sorted(current_acl_versions),
        "denied": denied,
    }
    audit_writer.write(evidence)
    return allowed

Verification and evidence

# File: tests/test_rag_retrieval_filter.py
from rag_authz.retrieval_filter import Principal, filter_authorized_chunks


def test_unauthorized_chunk_is_removed():
    principal = Principal(principal_id="user:alice", groups={"finance"})
    current_acl_versions = {"legal-doc-1": "42", "finance-doc-1": "42"}
    candidates = [
        {"id": "chunk-1", "metadata": {"source_doc_id": "legal-doc-1", "acl_owner_id": "user:bob", "acl_allowed_principals": [], "acl_allowed_groups": ["legal"], "acl_version": "42"}},
        {"id": "chunk-2", "metadata": {"source_doc_id": "finance-doc-1", "acl_owner_id": "user:bob", "acl_allowed_principals": [], "acl_allowed_groups": ["finance"], "acl_version": "42"}},
    ]
    allowed, denied = filter_authorized_chunks(candidates, principal, current_acl_versions)
    assert [chunk["id"] for chunk in allowed] == ["chunk-2"]
    assert denied[0]["chunk_id"] == "chunk-1"


def test_stale_acl_version_is_removed():
    principal = Principal(principal_id="user:alice", groups={"finance"})
    candidates = [
        {"id": "chunk-3", "metadata": {"source_doc_id": "finance-doc-2", "acl_owner_id": "user:bob", "acl_allowed_principals": [], "acl_allowed_groups": ["finance"], "acl_version": "old"}}
    ]
    allowed, denied = filter_authorized_chunks(candidates, principal, {"finance-doc-2": "new"})
    assert allowed == []
    assert denied[0]["reason"] == "principal_not_entitled_acl_missing_or_stale"


def test_missing_current_acl_version_is_removed():
    principal = Principal(principal_id="user:alice", groups=set())
    candidates = [{"id": "missing-truth-chunk", "metadata": {
        "source_doc_id": "doc-with-no-current-truth",
        "acl_owner_id": "user:alice", "acl_allowed_principals": [],
        "acl_allowed_groups": [], "acl_version": "None",
    }}]
    allowed, denied = filter_authorized_chunks(candidates, principal, {})
    assert allowed == []
    assert denied[0]["chunk_id"] == "missing-truth-chunk"


def test_present_opaque_acl_version_is_not_a_missing_value():
    principal = Principal(principal_id="user:alice", groups=set())
    candidates = [{"id": "known-version-chunk", "metadata": {
        "source_doc_id": "known-doc", "acl_owner_id": "user:alice",
        "acl_allowed_principals": [], "acl_allowed_groups": [],
        "acl_version": "None",
    }}]
    allowed, denied = filter_authorized_chunks(candidates, principal, {"known-doc": "None"})
    assert [chunk["id"] for chunk in allowed] == ["known-version-chunk"]
    assert denied == []

Action: Put this filter after candidate retrieval and before context assembly. The evidence artifact for each query is the rag_retrieval_authz_decision event with candidate, allowed, denied, principal, and reason-code fields.

AID-H-020.003-G003

Continuously test permission drift by replaying representative principals against the RAG index and comparing retrieval results with source-system access truth.

Concept:

ACLs change after indexing. A user can leave a group, a document can become restricted, or a source system can revoke access while stale chunk metadata remains in the vector store. Run scheduled drift tests that compare the RAG index's stored ACL metadata with the source system's current authorization truth. Do not treat a static denylist as source truth; the replay job must call or import the current source authorization view for the tested documents.

Step 1: Define drift test cases

# File: rag_authz/drift_cases.json
[
  {
    "case_id": "finance-user-cannot-read-legal-doc",
    "principal_id": "user:alice",
    "groups": ["finance"],
    "query": "summarize acquisition legal strategy"
  }
]

Step 2: Export or query current source-system access truth

# File: rag_authz/source_access_truth.json
{
  "legal-doc-2026-001": {
    "acl_version": "2026-07-04T12:00:00Z",
    "allowed_principals": ["user:bob"],
    "allowed_groups": ["legal"]
  },
  "finance-doc-2026-002": {
    "acl_version": "2026-07-04T12:00:00Z",
    "allowed_principals": [],
    "allowed_groups": ["finance"]
  }
}

This JSON is a schema illustration and test fixture only. Production runs must inject a source-specific adapter that performs an authenticated, consistency-bounded read against the authoritative ACL service and returns its snapshot/version receipt. A static local file cannot establish current authorization truth.

Step 3: Execute replay checks against source truth

# File: rag_authz/run_permission_drift_tests.py
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

from rag_authz.retrieval_filter import Principal, filter_authorized_chunks


LoadAclVersions = Callable[[set[str]], dict[str, str]]
CheckSourceRead = Callable[[Principal, str], bool]


@dataclass(frozen=True)
class SourceAuthorizationProvider:
    load_acl_versions: LoadAclVersions
    check_source_read: CheckSourceRead

    def current_acl_versions(self, source_doc_ids: set[str]) -> dict[str, str]:
        result = self.load_acl_versions(set(source_doc_ids))
        if not isinstance(result, dict) or set(result) != set(source_doc_ids):
            raise RuntimeError("source ACL snapshot is incomplete")
        if any(not isinstance(key, str) or not isinstance(value, str) or not value for key, value in result.items()):
            raise RuntimeError("source ACL snapshot has invalid versions")
        return result

    def can_read(self, principal: Principal, source_doc_id: str) -> bool:
        result = self.check_source_read(principal, source_doc_id)
        if type(result) is not bool:
            raise RuntimeError("source authorization adapter returned a non-boolean decision")
        return result


def run_case(vector_store, source_authz: SourceAuthorizationProvider, case: dict) -> dict:
    required = {"case_id", "principal_id", "groups", "query"}
    if not isinstance(case, dict) or set(case) != required:
        return {"status": "ERROR", "reason": "drift_case_schema_mismatch"}
    if not isinstance(case["groups"], list) or any(
        not isinstance(group, str) or not group for group in case["groups"]
    ):
        return {"status": "ERROR", "reason": "drift_case_groups_invalid"}
    principal = Principal(case["principal_id"], set(case["groups"]))
    candidates = vector_store.similarity_search(case["query"], k=20)
    if not isinstance(candidates, list):
        return {"status": "ERROR", "reason": "retriever_population_invalid"}
    if any(
        not isinstance(chunk, dict) or not isinstance(chunk.get("metadata"), dict)
        for chunk in candidates
    ):
        return {"status": "ERROR", "reason": "candidate_schema_invalid"}
    candidate_doc_ids = {
        str(chunk.get("metadata", {}).get("source_doc_id"))
        for chunk in candidates
        if chunk.get("metadata", {}).get("source_doc_id")
    }
    current_acl_versions = source_authz.current_acl_versions(candidate_doc_ids)
    if not isinstance(current_acl_versions, dict) or any(
        not isinstance(key, str) or not isinstance(value, str)
        for key, value in current_acl_versions.items()
    ):
        return {"status": "ERROR", "reason": "source_authorization_snapshot_invalid"}
    missing_source_truth = sorted(candidate_doc_ids - set(current_acl_versions))
    stale_chunks = [
        {
            "chunk_id": chunk.get("id"),
            "source_doc_id": str(chunk.get("metadata", {}).get("source_doc_id")),
            "indexed_acl_version": chunk.get("metadata", {}).get("acl_version"),
            "current_acl_version": current_acl_versions.get(
                str(chunk.get("metadata", {}).get("source_doc_id"))
            ),
        }
        for chunk in candidates
        if chunk.get("metadata", {}).get("source_doc_id")
        and str(chunk.get("metadata", {}).get("source_doc_id")) in current_acl_versions
        and str(chunk.get("metadata", {}).get("acl_version"))
        != current_acl_versions[str(chunk.get("metadata", {}).get("source_doc_id"))]
    ]
    allowed, denied = filter_authorized_chunks(candidates, principal, current_acl_versions)
    try:
        violations = [
            {
                "chunk_id": chunk.get("id"),
                "source_doc_id": chunk.get("metadata", {}).get("source_doc_id"),
                "reason": "rag_returned_doc_not_allowed_by_source_truth",
            }
            for chunk in allowed
            if not source_authz.can_read(
                principal, str(chunk.get("metadata", {}).get("source_doc_id"))
            )
        ]
    except Exception as exc:
        return {"status": "ERROR", "reason": type(exc).__name__}
    if missing_source_truth:
        status = "INSUFFICIENT_DATA"
    elif stale_chunks or violations:
        status = "FAIL"
    else:
        status = "PASS"
    return {
        "case_id": case["case_id"],
        "allowed_count": len(allowed),
        "denied_count": len(denied),
        "missing_source_truth": missing_source_truth,
        "stale_chunks": stale_chunks,
        "violations": violations,
        "status": status,
    }


def run_all(vector_store, source_authz: SourceAuthorizationProvider) -> dict:
    cases = json.loads(Path("rag_authz/drift_cases.json").read_text(encoding="utf-8"))
    results = [run_case(vector_store, source_authz, case) for case in cases]
    states = {result["status"] for result in results}
    if not results:
        status = "INSUFFICIENT_DATA"
    elif "ERROR" in states:
        status = "ERROR"
    elif "INSUFFICIENT_DATA" in states:
        status = "INSUFFICIENT_DATA"
    elif "FAIL" in states:
        status = "FAIL"
    else:
        status = "PASS"
    report = {
        "schema_version": "aidefend.rag_permission_drift.v1",
        "case_count": len(results),
        "failures": [result for result in results if result["status"] != "PASS"],
        "results": results,
        "status": status,
    }
    Path("artifacts").mkdir(exist_ok=True)
    Path("artifacts/rag-permission-drift-report.json").write_text(
        json.dumps(report, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    if report["status"] != "PASS":
        raise SystemExit("RAG permission drift test failed")
    return report

Step 4: Store a dated drift report

Run the replay job after ACL-sync jobs, after source-system permission migrations, and before promoting a new RAG index. Store the report with the index version, ACL sync version, retriever version, case results, and source-truth comparison violations.

Action: Treat permission drift testing as a release and operations gate. The evidence artifact for this guidance is artifacts/rag-permission-drift-report.json, showing which principals, queries, source-system truth records, and index versions were tested.

Open-source tools

OpenFGA (relationship-based authorization)
Open Policy Agent (OPA)
Casbin
PostgreSQL row-level security
LangChain retriever hooks
LlamaIndex retriever hooks
SpiceDB

Commercial / hosted tools

Auth0 Fine-Grained Authorization (FGA)
Google Cloud IAM (identity/policy source only; not retrieval enforcement)
Google Cloud Identity (identity source only; not retrieval enforcement)
Microsoft Entra ID (identity source only; not retrieval enforcement)
Pinecone (metadata filtering only; external authorization truth required)
Weaviate Cloud (metadata filtering only; external authorization truth required)
Elastic Cloud (metadata filtering only; external authorization truth required)

Threat mappings

MITRE ATLAS

AML.T0085 Data from AI Services (retrieval authorization limits data returned by AI services)
AML.T0085.001 Data from AI Services: AI Agent Tools (authorized retrieval constrains tool-mediated context access)
AML.T0086 Exfiltration via AI Agent Tool Invocation (unauthorized chunks are removed before context assembly)
AML.T0126 Automated Collection (chunk-level entitlements deny automated retrieval of unauthorized documents)

MAESTRO

Data Leakage (Cross-Layer)
Privilege Escalation (Cross-Layer) (retrieval authorization prevents low-privilege principals receiving high-privilege chunks)

OWASP LLM Top 10 2026

LLM02:2026 Sensitive Information Disclosure

OWASP ML Top 10 2023

N/A (retrieval-time authorization is primarily an LLM/RAG application control)

OWASP Top 10 for Agentic Applications 2026

ASI03:2026 Identity and Privilege Abuse (retrieval authorization checks principal entitlements)
ASI02:2026 Tool Misuse and Exploitation (tool-driven retrieval cannot bypass ACLs)

NIST Adversarial Machine Learning 2025

NISTAML.039 Compromising connected resources (retrieval gate prevents unauthorized connected-resource reads)

Cisco Integrated AI Security and Safety Framework

AISubtech-14.1.2 Insufficient Access Controls
AISubtech-8.2.3 Data Exfiltration via Agent Tooling
AITech-12.1 Tool Exploitation
AITech-8.2 Data Exfiltration / Exposure
AITech-8.3 Information Disclosure

Google Secure AI Framework 2.0 - Risks

SDD: Sensitive Data Disclosure
EDH: Excessive Data Handling
IIC: Insecure Integrated Component (RAG retriever is an integrated component requiring authorization)

Databricks AI Security Framework 3.0

Raw Data 1.1: Insufficient access controls
Agents - Core 13.3: Privilege Compromise

AID-H-020.003: Document/Chunk-Level Permission-Aware Retrieval

Scope boundary

Implementation guidance

AID-H-020.003-G001

Concept:

Step 1: Normalize document permissions

Step 2: Reject chunks with incomplete ACL metadata

AID-H-020.003-G002

Concept:

Step 1: Implement a permission filter

Step 2: Enforce before model context construction

Verification and evidence

AID-H-020.003-G003

Concept:

Step 1: Define drift test cases

Step 2: Export or query current source-system access truth

Step 3: Execute replay checks against source truth

Step 4: Store a dated drift report

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-020.003: Document/Chunk-Level Permission-Aware Retrieval | AIDEFEND

AIDEFEND™