AID-E-003.002: Manifest-Bound Poisoned Dataset & Graph Eviction
Pillars: data
Lifecycle: response, improvement
Scope boundary
Owns manifest-bound eviction or isolation of exact poisoned source dataset rows, graph nodes, and incident edges, together with independent proof of the authorized set difference. Candidate detection is an input, not this control's result. It does not restore source data or operate the derived vector-index serving layer.
AID-R-002: AID-E-003.002 removes the exact poisoned source-record or graph population; AID-R-002 restores or deterministically repairs trusted source data after eviction. An eviction manifest proves what was removed, while recovery proves the resulting source dataset is complete and trustworthy enough for reuse.
AID-R-005: AID-E-003.002 evicts poisoned source rows, nodes, and edges; AID-R-005 quarantines, rebuilds, validates, and cuts over the derived vector index. Source eviction does not remove already indexed vectors or caches, and vector rollback does not prove that the malicious source records cannot be reingested.
Implementation guidance
- AID-E-003.002-G001: Remove only exact row IDs authorized by a signed analyst-reviewed eviction manifest.
- AID-E-003.002-G002: Apply a signed exact-node/edge eviction manifest to a versioned graph snapshot, preserving directed and multigraph identity, then require an independent full-population set-difference and topology replay before release.
AID-E-003.002-G001
Concept:
Removal is a separate governed action from detection. The analyst-reviewed manifest must be signed by an authorized decision identity, bind the immutable input dataset and separately signed candidate report by digest, enumerate exact row IDs, and forbid wildcard or inferred removal. The removal worker creates a new dataset; it never overwrites the forensic source.
Signed eviction manifest
{
"schema_version": "aidefend.e003002-eviction-manifest/v1",
"control": "AID-E-003.002",
"incident_id": "INC-2026-0718-0042",
"authorization_id": "change://CHG-2026-0718-031",
"eviction_policy_version": "poison-row-eviction/2026.07",
"retention_policy_version": "dataset-retention/2026.07",
"dataset_sha256": "c61e73c8f22a997daf0f9a4acbb7e0c45bc7184a2b81cb184034a3e879b330e2",
"candidate_report_sha256": "b83a6e1f95d3442cb32dcc023139fd75e17d54b062eacb4f6762a7ed662d7791",
"remove_row_ids": ["row-001", "row-014"],
"reviewer_identity": "urn:aidefend:reviewer:security-reviewer",
"decision_evidence_refs": ["case://INC-2026-0718-0042/finding-7"]
}cosign verify-blob --key keys/e003002-candidate-authority.pub --bundle candidate-report.json.sigstore.json candidate-report.json
cosign verify-blob --key keys/e003002-analyst.pub --bundle eviction-manifest.json.sigstore.json eviction-manifest.jsonApply only the verified exact-ID decision
# File: data_cleansing/apply_eviction_manifest.py
from __future__ import annotations
import hashlib
import io
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path
import pandas as pd
CANDIDATE_TRUST_KEY = Path("/opt/aidefend/trust/e003002-candidate-authority.pub")
MANIFEST_TRUST_KEY = Path("/opt/aidefend/trust/e003002-analyst.pub")
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(os.environ["E003002_COMMAND_TIMEOUT_SECONDS"])
MANIFEST_FIELDS = {
"schema_version", "control", "incident_id", "authorization_id",
"eviction_policy_version", "retention_policy_version", "dataset_sha256",
"candidate_report_sha256", "remove_row_ids", "reviewer_identity",
"decision_evidence_refs",
}
if (
not RUNTIME_PROFILE_VERSION
or len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
or RUNTIME_PROFILE_SHA256 == "0" * 64
or not math.isfinite(COMMAND_TIMEOUT_SECONDS)
or COMMAND_TIMEOUT_SECONDS <= 0
):
raise RuntimeError("versioned E-003.002 runtime profile is invalid")
def verify_blob(path: Path, bundle: Path, trust_key: Path) -> None:
subprocess.run(
[
"cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle), str(path),
],
check=True,
capture_output=True,
text=True,
timeout=COMMAND_TIMEOUT_SECONDS,
)
def reject_duplicate_keys(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_nonfinite(value):
raise ValueError(f"non-finite JSON constant is forbidden: {value}")
def verified_snapshot(path: Path, bundle: Path, trust_key: Path) -> bytes:
raw = path.read_bytes()
bundle_raw = bundle.read_bytes()
if not raw or not bundle_raw:
raise ValueError("signed payload or Cosign bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-e003002-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload.json"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(raw)
bundle_snapshot.write_bytes(bundle_raw)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
verify_blob(payload_snapshot, bundle_snapshot, trust_key)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != raw or verified_bundle != bundle_raw:
raise OSError("verified payload or Cosign bundle snapshot changed")
return verified
def require_sha256(value: object, field: str) -> str:
if (
not isinstance(value, str)
or len(value) != 64
or set(value) - set("0123456789abcdef")
or value == "0" * 64
):
raise ValueError(f"{field} must be a nonzero lowercase SHA-256 digest")
return value
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def apply_manifest(
input_csv: Path,
candidate_report_path: Path,
candidate_report_bundle: Path,
manifest_path: Path,
manifest_bundle: Path,
output_csv: Path,
) -> dict:
candidate_raw = verified_snapshot(
candidate_report_path, candidate_report_bundle, CANDIDATE_TRUST_KEY
)
manifest_raw = verified_snapshot(manifest_path, manifest_bundle, MANIFEST_TRUST_KEY)
manifest = json.loads(
manifest_raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys, parse_constant=reject_nonfinite,
)
if (
not isinstance(manifest, dict)
or set(manifest) != MANIFEST_FIELDS
or manifest.get("schema_version") != "aidefend.e003002-eviction-manifest/v1"
):
raise ValueError("unexpected eviction-manifest schema")
if manifest.get("control") != "AID-E-003.002":
raise ValueError("wrong control binding")
for field in ("authorization_id", "eviction_policy_version", "retention_policy_version", "reviewer_identity"):
if not isinstance(manifest.get(field), str) or not manifest[field]:
raise ValueError(field + " is required")
dataset_sha256 = require_sha256(
manifest.get("dataset_sha256"), "manifest.dataset_sha256"
)
candidate_sha256 = require_sha256(
manifest.get("candidate_report_sha256"),
"manifest.candidate_report_sha256",
)
dataset_raw = input_csv.read_bytes()
if hashlib.sha256(dataset_raw).hexdigest() != dataset_sha256:
raise ValueError("input dataset digest differs")
if hashlib.sha256(candidate_raw).hexdigest() != candidate_sha256:
raise ValueError("candidate-report digest differs")
row_ids = manifest.get("remove_row_ids")
if (
not isinstance(row_ids, list)
or not row_ids
or row_ids != sorted(set(row_ids))
or any(not isinstance(item, str) or not item or "*" in item for item in row_ids)
):
raise ValueError("remove_row_ids must be a sorted unique exact-ID list")
candidates = json.loads(
candidate_raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys, parse_constant=reject_nonfinite,
)
candidate_ids = candidates.get("candidate_row_ids") if isinstance(candidates, dict) else None
if (not isinstance(candidate_ids, list) or candidate_ids != sorted(set(candidate_ids))
or any(not isinstance(item, str) or not item for item in candidate_ids)):
raise ValueError("signed candidate-row population is invalid")
if not set(row_ids).issubset(set(candidate_ids)):
raise ValueError("manifest includes rows outside the signed candidate report")
frame = pd.read_csv(io.BytesIO(dataset_raw), dtype={"row_id": "string"})
if "row_id" not in frame or not frame["row_id"].is_unique:
raise ValueError("dataset requires unique row_id values")
selected = frame["row_id"].astype(str).isin(row_ids)
if int(selected.sum()) != len(row_ids):
raise ValueError("not every authorized row exists exactly once")
cleaned = frame.loc[~selected]
with output_csv.open("x", encoding="utf-8", newline="") as handle:
cleaned.to_csv(handle, index=False)
return {
"schema_version": "aidefend.e003002-owner-receipt/v1",
"control": "AID-E-003.002",
"incident_id": manifest["incident_id"],
"authorization_id": manifest["authorization_id"],
"eviction_policy_version": manifest["eviction_policy_version"],
"retention_policy_version": manifest["retention_policy_version"],
"dataset_before_sha256": manifest["dataset_sha256"],
"candidate_report_sha256": manifest["candidate_report_sha256"],
"eviction_manifest_sha256": hashlib.sha256(manifest_raw).hexdigest(),
"removed_row_ids": row_ids,
"dataset_after_sha256": sha256(output_csv),
"runtime_profile_version": RUNTIME_PROFILE_VERSION,
"runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
}
Independent verification: The writer verifies both signatures under image-pinned keys before reading either document. A read-only verifier with no dataset-write or approval permission must independently verify both signatures, recompute all four digests, replay the exact set difference between the immutable input and new output, confirm no undeclared row changed, and sign aidefend.e003002-verification/v1. A missing signature, digest mismatch, absent row, extra changed row, or incomplete replay fails closed.
Action: Remove only exact analyst-authorized rows into a new artifact, preserve the source snapshot, and require the owner receipt plus independent replay receipt before Restore consumes the dataset.
AID-E-003.002-G002
When graph eviction applies
Use this method when the incident-bound affected asset is a governed graph dataset and the authorized eviction scope identifies exact node IDs or exact incident edge identities. If the incident scope contains no governed graph asset, use the applicable non-graph method. When a graph asset exists, the writer and verifier require a complete versioned export, immutable source digest, exact node and edge population, direction flag, multigraph flag, and every multigraph edge key. An unsupported or partial export, missing node or edge identity, absent multigraph key, or incomplete snapshot blocks graph eviction and cannot be treated as not applicable.
Bind the exact graph difference
The recovery authority signs an incident-specific manifest containing the immutable source digest, operation, exact node IDs, and exact incident edge identities. delete_nodes removes the authorized nodes and all of their incoming and outgoing incident edges. isolate_edges retains the nodes but removes every incoming and outgoing incident edge authorized for isolation. The manifest edge set must equal the complete incident-edge set for the selected nodes; this prevents a one-direction-only cleanup or a parallel-edge omission from passing.
{
"schema_version": "aidefend.graph-eviction/v1",
"incident_id": "INC-2026-0718-0042",
"asset_id": "recommendation-graph-prod",
"source_sha256": "4fb7f4d9f2ce7cb28bb7c5311a5f3f0b3ea99c1f8a95638997276e48e97c5fa1",
"operation": "delete_nodes",
"node_ids": ["poison-node-17"],
"edge_ids": [
{"source": "customer-4", "target": "poison-node-17", "key": "edge-93"},
{"source": "poison-node-17", "target": "item-8", "key": "edge-94"}
]
}Apply and independently verify the complete set difference
# File: eviction/evict_graph_members.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Any
MANIFEST_TRUST_KEY = Path("/opt/aidefend/trust/e003002-graph-authority.pub")
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(os.environ["E003002_COMMAND_TIMEOUT_SECONDS"])
if (
not RUNTIME_PROFILE_VERSION
or len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
or RUNTIME_PROFILE_SHA256 == "0" * 64
or not math.isfinite(COMMAND_TIMEOUT_SECONDS)
or COMMAND_TIMEOUT_SECONDS <= 0
):
raise RuntimeError("versioned graph-eviction runtime profile is invalid")
def reject_duplicate_keys(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_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def verified_signed_bytes(
payload_path: Path,
bundle_path: Path,
trust_key: str | Path,
timeout_seconds: float,
label: str,
) -> bytes:
payload = payload_path.read_bytes()
bundle = bundle_path.read_bytes()
if not payload or not bundle:
raise ValueError(f"{label} payload or bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(payload)
bundle_snapshot.write_bytes(bundle)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != payload or verified_bundle != bundle:
raise OSError(f"{label} verified payload or bundle snapshot changed")
return verified
def verify_manifest_bundle(path: Path, bundle: Path) -> bytes:
return verified_signed_bytes(
path, bundle, MANIFEST_TRUST_KEY, COMMAND_TIMEOUT_SECONDS,
"graph-eviction manifest",
)
def canonical(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def digest(value: Any) -> str:
return hashlib.sha256(canonical(value)).hexdigest()
def load_object_bytes(raw: bytes, label: str) -> dict:
value = strict_json(raw, label)
if not isinstance(value, dict):
raise ValueError(f"{label}: JSON object required")
return value
def load_object(path: Path) -> dict:
return load_object_bytes(path.read_bytes(), str(path))
def require_id(value: object, field: str) -> str:
if not isinstance(value, str) or not value or len(value) > 512:
raise ValueError(f"{field}: nonempty bounded string required")
return value
def require_sha256(value: object, field: str) -> str:
if (
not isinstance(value, str)
or len(value) != 64
or set(value) - set("0123456789abcdef")
or value == "0" * 64
):
raise ValueError(f"{field} must be a nonzero lowercase SHA-256 digest")
return value
def node_map(graph: dict) -> dict[str, dict]:
nodes = graph.get("nodes")
if not isinstance(nodes, list):
raise ValueError("complete node list required")
result = {}
for node in nodes:
if not isinstance(node, dict) or "id" not in node:
raise ValueError("node object with id required")
node_id = require_id(node["id"], "node.id")
if node_id in result:
raise ValueError(f"duplicate node: {node_id}")
result[node_id] = node
return result
def edge_identity(edge: dict, *, directed: bool, multigraph: bool) -> tuple[str, str, str | None]:
if not isinstance(edge, dict):
raise ValueError("edge object required")
source = require_id(edge.get("source"), "edge.source")
target = require_id(edge.get("target"), "edge.target")
key = edge.get("key")
if multigraph:
key = require_id(key, "edge.key")
elif key is not None:
raise ValueError("simple graph edge key must be null or absent")
if not directed and target < source:
source, target = target, source
return source, target, key
def edge_map(graph: dict) -> dict[tuple[str, str, str | None], dict]:
directed = graph.get("directed")
multigraph = graph.get("multigraph")
edges = graph.get("edges")
if type(directed) is not bool or type(multigraph) is not bool or not isinstance(edges, list):
raise ValueError("direction, multigraph flag, and complete edge list are required")
result = {}
for edge in edges:
identity = edge_identity(edge, directed=directed, multigraph=multigraph)
if identity in result:
raise ValueError(f"duplicate edge: {identity}")
result[identity] = edge
return result
def validate_graph(graph: dict) -> tuple[dict[str, dict], dict[tuple[str, str, str | None], dict]]:
if graph.get("schema_version") != "aidefend.graph-dataset/v1":
raise ValueError("unsupported graph schema")
if graph.get("population_complete") is not True:
raise ValueError("partial graph export cannot be evicted")
nodes = node_map(graph)
edges = edge_map(graph)
for source, target, _key in edges:
if source not in nodes or target not in nodes:
raise ValueError("edge references a missing node")
return nodes, edges
def manifest_sets(manifest: dict, graph: dict) -> tuple[set[str], set[tuple[str, str, str | None]]]:
required = {
"schema_version", "incident_id", "asset_id", "source_sha256",
"operation", "node_ids", "edge_ids",
}
if set(manifest) != required or manifest["schema_version"] != "aidefend.graph-eviction/v1":
raise ValueError("eviction manifest schema differs")
if manifest["operation"] not in {"delete_nodes", "isolate_edges"}:
raise ValueError("unsupported graph eviction operation")
if not isinstance(manifest["node_ids"], list) or not manifest["node_ids"]:
raise ValueError("nonempty node population required")
if not isinstance(manifest["edge_ids"], list):
raise ValueError("edge population must be a list")
nodes = {require_id(value, "node_ids") for value in manifest["node_ids"]}
if len(nodes) != len(manifest["node_ids"]):
raise ValueError("duplicate manifest node")
edges = {
edge_identity(value, directed=graph["directed"], multigraph=graph["multigraph"])
for value in manifest["edge_ids"]
}
if len(edges) != len(manifest["edge_ids"]):
raise ValueError("duplicate manifest edge")
return nodes, edges
def expected_difference(graph: dict, manifest: dict) -> tuple[set[str], set[tuple[str, str, str | None]]]:
nodes, edges = validate_graph(graph)
selected_nodes, selected_edges = manifest_sets(manifest, graph)
if not selected_nodes.issubset(nodes):
raise ValueError("manifest references an absent node")
incident_edges = {edge for edge in edges if edge[0] in selected_nodes or edge[1] in selected_nodes}
if selected_edges != incident_edges:
raise ValueError("manifest does not enumerate the complete incident-edge population")
removed_nodes = selected_nodes if manifest["operation"] == "delete_nodes" else set()
return removed_nodes, selected_edges
def apply(source: dict, manifest: dict) -> dict:
source_sha256 = require_sha256(
manifest.get("source_sha256"), "manifest.source_sha256"
)
if digest(source) != source_sha256:
raise ValueError("source graph digest differs from signed manifest")
removed_nodes, removed_edges = expected_difference(source, manifest)
result = dict(source)
result["nodes"] = [node for node in source["nodes"] if node["id"] not in removed_nodes]
result["edges"] = [
edge for edge in source["edges"]
if edge_identity(edge, directed=source["directed"], multigraph=source["multigraph"]) not in removed_edges
]
result["source_sha256"] = manifest["source_sha256"]
result["eviction_incident_id"] = manifest["incident_id"]
result["population_complete"] = True
validate_graph(result)
return result
def verify(source: dict, candidate: dict, manifest: dict) -> None:
source_sha256 = require_sha256(
manifest.get("source_sha256"), "manifest.source_sha256"
)
if digest(source) != source_sha256:
raise ValueError("source graph digest differs from signed manifest")
if (
candidate.get("source_sha256") != manifest["source_sha256"]
or candidate.get("eviction_incident_id") != manifest["incident_id"]
):
raise ValueError("candidate source or incident binding differs")
source_nodes, source_edges = validate_graph(source)
candidate_nodes, candidate_edges = validate_graph(candidate)
# Eviction may change membership only, never graph semantics or metadata.
mutable = {"nodes", "edges", "source_sha256", "eviction_incident_id", "population_complete"}
if canonical({k: v for k, v in source.items() if k not in mutable}) != canonical(
{k: v for k, v in candidate.items() if k not in mutable}
):
raise RuntimeError("candidate changed graph semantics or metadata")
removed_nodes, removed_edges = expected_difference(source, manifest)
if set(source_nodes) - set(candidate_nodes) != removed_nodes:
raise RuntimeError("node set difference differs from manifest")
if set(source_edges) - set(candidate_edges) != removed_edges:
raise RuntimeError("edge set difference differs from manifest")
if set(candidate_nodes) - set(source_nodes) or set(candidate_edges) - set(source_edges):
raise RuntimeError("candidate introduced undeclared graph members")
for identity in set(candidate_nodes):
if canonical(candidate_nodes[identity]) != canonical(source_nodes[identity]):
raise RuntimeError(f"retained node changed: {identity}")
for identity in set(candidate_edges):
if canonical(candidate_edges[identity]) != canonical(source_edges[identity]):
raise RuntimeError(f"retained edge changed: {identity}")
selected_nodes, _ = manifest_sets(manifest, source)
if manifest["operation"] == "isolate_edges" and any(
source_id in selected_nodes or target_id in selected_nodes
for source_id, target_id, _key in candidate_edges
):
raise RuntimeError("isolated node retains incoming or outgoing influence")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("mode", choices=("apply", "verify"))
parser.add_argument("source", type=Path)
parser.add_argument("manifest", type=Path)
parser.add_argument("manifest_bundle", type=Path)
parser.add_argument("candidate", type=Path)
args = parser.parse_args()
manifest_raw = verify_manifest_bundle(args.manifest, args.manifest_bundle)
source = load_object(args.source)
manifest = load_object_bytes(manifest_raw, "graph-eviction manifest")
if args.mode == "apply":
candidate = apply(source, manifest)
args.candidate.write_bytes(canonical(candidate) + b"\n")
else:
candidate = load_object(args.candidate)
verify(source, candidate, manifest)
print(json.dumps({"control": "AID-E-003.002", "candidate_sha256": digest(candidate), "result": "verified"}, sort_keys=True))
if __name__ == "__main__":
main()
Separate writer and verifier
Verify the manifest bundle before either execution. Run apply only under the governed dataset writer and write to a new immutable version; never edit the serving graph in place. A different read-only identity fetches the source and candidate from their authoritative immutable locations, reruns verify, compares the exact full-population difference, and replays directed reachability for isolated nodes before release. Any partial export, digest mismatch, extra removal, retained incident edge, direction change, multigraph-key loss, or unreadable member keeps the candidate quarantined.
Open-source tools
- DVC
- lakeFS (versioned object-store publication of externally computed manifest set differences)
- Delta Lake (row-level eviction and versioned table publication)
- Apache Spark
- Dask
- NetworkX
Commercial / hosted tools
- Databricks Data Intelligence Platform
Threat mappings
MITRE ATLAS
- AML.T0010 AI Supply Chain Compromise (exact source-record or graph eviction removes the signed compromised data population)
- AML.T0020 Training Data Poisoning
- AML.T0059 Erode Dataset Integrity
- AML.T0070 RAG Poisoning
- AML.T0071 False RAG Entry Injection (manifest-bound graph eviction removes the exact injected false entry before governed index rebuild)
- AML.T0010.002 AI Supply Chain Compromise: Data
MAESTRO
- Data Poisoning (L2)
- Data Tampering (L2)
- Compromised RAG Pipelines (L2)
- Data Poisoning (Training Phase) (L1)
- Supply Chain Attacks (Cross-Layer)
OWASP LLM Top 10 2026
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- ML02:2023 Data Poisoning Attack
OWASP Top 10 for Agentic Applications 2026
- ASI06:2026 Memory & Context Poisoning (evicting poisoned RAG source and durable graph records prevents their reuse)
- ASI04:2026 Agentic Supply Chain Vulnerabilities (poisoned datasets in supply chain)
NIST Adversarial Machine Learning 2025
- NISTAML.013 Data Poisoning
- NISTAML.024 Targeted Poisoning
- NISTAML.023 Backdoor Poisoning
- NISTAML.012 Clean-label Poisoning (cleansing removes clean-label poisoned samples)
- NISTAML.021 Clean-label Backdoor
Cisco Integrated AI Security and Safety Framework
- AISubtech-6.1.1 Knowledge Base Poisoning
- AITech-7.3 Data Source Abuse and Manipulation (Evicts the identified poisoned source records; downstream retrieval publication is separate)
- AITech-6.1 Training Data Poisoning (evicts incident-bound poisoned training or fine-tuning dataset records; trained-model recovery remains separate)
Google Secure AI Framework 2.0 - Risks
- DP: Data Poisoning
- UTD: Unauthorized Training Data (cleansing removes unauthorized data from training sets)
- PIJ: Prompt Injection (eviction removes injected content from the governed RAG source corpus before AID-R-005 rebuilds its index)
Databricks AI Security Framework 3.0
- Datasets 3.1: Data poisoning
- Datasets 3.3: Label flipping
- Raw Data 1.7: Lack of data trustworthiness
- Raw Data 1.11: Compromised 3rd-party datasets
- Agents - Core 13.1: Memory Poisoning