Scope boundary

Owns runtime provenance and sensitivity metadata for values and blocks unauthorized flows into outbound, write, model-call, retrieval, storage, and other sensitive sinks. It answers whether this value may reach this destination, not merely whether the parameter is well formed.

AID-H-018.001: AID-H-018.005 enforces provenance-aware value-to-sink policy; AID-H-018.001 validates parameter type, shape, range, and format. A structurally valid value can still violate data-use or exfiltration policy.

AID-H-029: AID-H-018.005 enforces value flows at agent action and sink boundaries; AID-H-029 governs purpose, consent, retention, lifecycle stage, and authorized use of data across the broader AI lifecycle. The sink gate consumes data-use policy but does not define the complete lifecycle authorization contract.

Implementation guidance

• AID-H-018.005-G001: Application-layer value tagging (taint) plus sensitive-sink checks in the tool dispatcher, backed by a versioned policy engine (OPA) and session-isolated metadata storage.
• AID-H-018.005-G002: Network-layer egress filtering (service mesh / firewall) to enforce outbound constraints even if application-layer checks fail, with default-deny posture and centrally managed allowlists.
• AID-H-018.005-G003: Enforce prompt-body and model-call sink policies before sensitive runtime values are sent to an LLM or external model provider.

AID-H-018.005-G001

Delivery level: reusable dispatcher enforcement module

OPA owns the organization-specific data-flow rules, Redis owns session-scoped label storage, and the audit service owns durable evidence. The dispatcher owns the security-critical integration: every registered tool has a policy-owned class and closed Draft 2020-12 argument schema, unknown tools and argument keys are denied, sensitive values are resolved server-side, and OPA authorizes the same canonical argument snapshot the handler receives under the active policy revision.

OPA decision contract

Load data.aidefend_config.policy_revision from the same signature-verified policy bundle as the Rego module. A missing or invalid revision keeps the decision at DENY.

package aidefend.dataflow

policy_revision := value if {
  value := data.aidefend_config.policy_revision
  is_string(value)
  value != ""
} else := "MISSING"

approved_internal_email if {
  policy_revision != "MISSING"
  input.tool_name == "send_email"
  input.tool_class == "sensitive_sink"
  input.labels.sensitivity == "internal"
  endswith(lower(input.arguments.recipient), "@corp.example")
}

result := {
  "decision": "ALLOW",
  "rule_id": "internal_email",
  "reason": "internal content to approved internal recipient",
  "policy_revision": policy_revision,
} if {
  approved_internal_email
}

result := {
  "decision": "DENY",
  "rule_id": "deny_by_default",
  "reason": "no matching rule",
  "policy_revision": policy_revision,
} if {
  not approved_internal_email
}

Fail-closed registry, label, OPA, and audit adapters

# File: agent_runtime/dataflow_dispatcher.py
from __future__ import annotations

import hashlib
import json
import math
import time
from dataclasses import asdict, dataclass
from typing import Any, Callable, Literal, Mapping, Protocol, Sequence
from urllib.parse import urlsplit

import jsonschema
import requests

ToolClass = Literal["pure_read", "internal_write", "sensitive_sink"]
Decision = Literal["ALLOW", "DENY", "REQUIRE_APPROVAL"]
Handler = Callable[[dict[str, Any]], Any]


class EnforcementError(RuntimeError):
    pass


@dataclass(frozen=True)
class ValueLabels:
    source: str
    sensitivity: str
    expires_at_epoch_seconds: int
    allowed_domains: Sequence[str] = ()
    allowed_recipients: Sequence[str] = ()
    requires_user_approval: bool = False


@dataclass(frozen=True)
class ToolSpec:
    tool_class: ToolClass
    handler: Handler
    argument_validator: Any
    argument_schema_sha256: str


@dataclass(frozen=True)
class RequestContext:
    tenant_id: str
    session_id: str
    user_id: str
    request_id: str


class MetadataStore(Protocol):
    def get_labels(
        self, tenant_id: str, session_id: str, content_id: str
    ) -> ValueLabels | None: ...


class AuditSink(Protocol):
    def write_and_ack(self, event: Mapping[str, Any]) -> str: ...


def _strict_json(raw: bytes) -> dict:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key: {key}")
            value[key] = item
        return value
    def reject_constant(value):
        raise ValueError(f"non-finite JSON number: {value}")
    document = json.loads(raw, object_pairs_hook=unique, parse_constant=reject_constant)
    if not isinstance(document, dict):
        raise ValueError("JSON root must be an object")
    return document


class PolicyClient:
    def __init__(
        self, *, url: str, expected_revision: str, timeout_seconds: float,
        maximum_response_bytes: int,
    ) -> None:
        parsed = urlsplit(url)
        local_http = parsed.scheme == "http" and parsed.hostname == "127.0.0.1"
        remote_https = parsed.scheme == "https" and bool(parsed.hostname)
        if (not (local_http or remote_https) or parsed.username is not None or
                parsed.password is not None or parsed.fragment):
            raise ValueError("OPA URL must be an exact loopback HTTP route or canonical HTTPS endpoint")
        if not expected_revision:
            raise ValueError("active OPA policy revision is required")
        if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
            raise ValueError("OPA timeout must be finite and positive")
        if maximum_response_bytes < 1:
            raise ValueError("OPA response cap must be positive")
        self.url = url
        self.expected_revision = expected_revision
        self.timeout_seconds = timeout_seconds
        self.maximum_response_bytes = maximum_response_bytes

    def evaluate(self, input_document: dict[str, Any]) -> tuple[Decision, str, str]:
        try:
            with requests.post(
                self.url, json={"input": input_document},
                timeout=self.timeout_seconds, allow_redirects=False, stream=True,
            ) as response:
                response.raise_for_status()
                raw = bytearray()
                for chunk in response.iter_content(
                    chunk_size=min(65536, self.maximum_response_bytes + 1)
                ):
                    if chunk:
                        raw.extend(chunk)
                    if len(raw) > self.maximum_response_bytes:
                        raise EnforcementError("OPA response exceeds the policy cap")
        except requests.RequestException as error:
            raise EnforcementError("OPA dependency failed") from error
        body = _strict_json(bytes(raw))
        if not isinstance(body, dict) or not isinstance(body.get("result"), dict):
            raise EnforcementError("OPA returned no decision object")
        result = body["result"]
        if set(result) != {"decision", "rule_id", "reason", "policy_revision"}:
            raise EnforcementError("OPA decision schema mismatch")
        decision = result["decision"]
        if decision not in {"ALLOW", "DENY", "REQUIRE_APPROVAL"}:
            raise EnforcementError("OPA returned an unsupported decision")
        if result["policy_revision"] != self.expected_revision:
            raise EnforcementError("OPA policy revision differs from the admitted revision")
        decision_id = body.get("decision_id")
        if not isinstance(decision_id, str) or not decision_id:
            raise EnforcementError("OPA decision_id is required for correlation")
        if not isinstance(result["rule_id"], str) or not result["rule_id"]:
            raise EnforcementError("OPA rule_id is required")
        return decision, result["rule_id"], decision_id


def canonical_document(value: Any) -> tuple[bytes, Any]:
    def validate(item: Any) -> None:
        if item is None or isinstance(item, (bool, int, str)):
            return
        if isinstance(item, float):
            if not math.isfinite(item):
                raise EnforcementError("non-finite tool argument")
            return
        if isinstance(item, (list, tuple)):
            for child in item:
                validate(child)
            return
        if isinstance(item, dict):
            if any(not isinstance(key, str) for key in item):
                raise EnforcementError("tool argument keys must be strings")
            for child in item.values():
                validate(child)
            return
        raise EnforcementError("tool arguments must be JSON values")

    validate(value)
    raw = json.dumps(
        value, ensure_ascii=False, separators=(",", ":"),
        sort_keys=True, allow_nan=False,
    ).encode("utf-8")
    return raw, _strict_json(raw)


def reject_external_schema_references(value: Any) -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            if key in {"$ref", "$dynamicRef"}:
                if not isinstance(child, str) or not child.startswith("#"):
                    raise ValueError("external argument-schema references are disabled")
            reject_external_schema_references(child)
    elif isinstance(value, list):
        for child in value:
            reject_external_schema_references(child)


def compile_argument_schema(schema_value: Mapping[str, Any]) -> tuple[Any, str]:
    raw, schema = canonical_document(dict(schema_value))
    if (
        not isinstance(schema, dict)
        or schema.get("type") != "object"
        or schema.get("additionalProperties") is not False
        or not isinstance(schema.get("properties"), dict)
    ):
        raise ValueError("tool argument schema must be a closed object schema")
    reject_external_schema_references(schema)
    jsonschema.Draft202012Validator.check_schema(schema)
    return jsonschema.Draft202012Validator(schema), hashlib.sha256(raw).hexdigest()


def digest_value(value: Any) -> str:
    canonical, _ = canonical_document(value)
    return hashlib.sha256(canonical).hexdigest()


class DataFlowDispatcher:
    def __init__(
        self, *, handlers: Mapping[str, Handler],
        signed_tool_classes: Mapping[str, ToolClass],
        signed_argument_schemas: Mapping[str, Mapping[str, Any]],
        metadata_store: MetadataStore, policy: PolicyClient, audit: AuditSink,
    ) -> None:
        populations = (
            set(handlers), set(signed_tool_classes), set(signed_argument_schemas)
        )
        if not handlers or any(population != populations[0] for population in populations[1:]):
            raise ValueError("handler, class, and argument-schema populations differ")
        allowed_classes = {"pure_read", "internal_write", "sensitive_sink"}
        if any(value not in allowed_classes for value in signed_tool_classes.values()):
            raise ValueError("signed registry contains an unsupported tool class")
        tools = {}
        for name, handler in handlers.items():
            validator, schema_sha256 = compile_argument_schema(
                signed_argument_schemas[name]
            )
            tools[name] = ToolSpec(
                signed_tool_classes[name], handler, validator, schema_sha256
            )
        self.tools = tools
        self.metadata_store = metadata_store
        self.policy = policy
        self.audit = audit

    def dispatch(
        self, context: RequestContext, tool_name: str, arguments: dict[str, Any]
    ) -> Any:
        spec = self.tools.get(tool_name)
        if spec is None:
            raise EnforcementError("unknown or unclassified tool")
        if not all(
            isinstance(value, str) and value
            for value in (
                context.tenant_id, context.session_id,
                context.user_id, context.request_id,
            )
        ):
            raise EnforcementError("request identity is incomplete")

        if not isinstance(arguments, dict):
            raise EnforcementError("tool arguments must be an object")
        argument_bytes, argument_snapshot = canonical_document(arguments)
        errors = sorted(
            spec.argument_validator.iter_errors(argument_snapshot),
            key=lambda error: tuple(str(part) for part in error.absolute_path),
        )
        if errors:
            raise EnforcementError("tool arguments violate the signed closed schema")

        content_id = argument_snapshot.get("content_id")
        labels = None
        if spec.tool_class in {"internal_write", "sensitive_sink"}:
            if not isinstance(content_id, str) or not content_id:
                raise EnforcementError("side-effecting tools require a server-side content_id")
            labels = self.metadata_store.get_labels(
                context.tenant_id, context.session_id, content_id
            )
            if labels is None:
                raise EnforcementError("content labels are missing")
            if labels.expires_at_epoch_seconds <= int(time.time()):
                raise EnforcementError("content labels are expired")

        policy_input = {
            "tenant_id": context.tenant_id,
            "session_id": context.session_id,
            "user_id": context.user_id,
            "tool_name": tool_name,
            "tool_class": spec.tool_class,
            "arguments": argument_snapshot,
            "labels": asdict(labels) if labels is not None else None,
        }
        decision, rule_id, decision_id = self.policy.evaluate(policy_input)
        event = {
            "schema_version": "aidefend.dataflow-dispatch.v1",
            "request_id": context.request_id,
            "tenant_id": context.tenant_id,
            "session_id": context.session_id,
            "user_id": context.user_id,
            "tool_name": tool_name,
            "tool_class": spec.tool_class,
            "content_id_sha256": digest_value(content_id) if content_id else None,
            "argument_sha256": hashlib.sha256(argument_bytes).hexdigest(),
            "argument_schema_sha256": spec.argument_schema_sha256,
            "policy_revision": self.policy.expected_revision,
            "policy_rule_id": rule_id,
            "opa_decision_id": decision_id,
            "decision": decision,
        }
        decision_receipt = self.audit.write_and_ack(event)
        if not isinstance(decision_receipt, str) or not decision_receipt:
            raise EnforcementError("audit sink did not acknowledge the decision")
        if decision != "ALLOW":
            raise EnforcementError(f"tool dispatch denied: {decision}:{rule_id}")

        result = spec.handler(argument_snapshot)
        completion = {
            **event,
            "decision_receipt": decision_receipt,
            "execution": "COMPLETED",
            "result_sha256": digest_value(result),
        }
        completion_receipt = self.audit.write_and_ack(completion)
        if not isinstance(completion_receipt, str) or not completion_receipt:
            raise EnforcementError("audit sink did not acknowledge execution completion")
        return result

Integration and verification

The deployment adapter must load signed_tool_classes, one top-level additionalProperties: false argument schema per tool, and expected_revision from the same signature-verified runtime profile, then reconcile both signed populations with the actual handler registry at startup. The Redis adapter must scope keys by tenant, session, and opaque content ID and return expiry-bearing labels; the audit adapter must acknowledge a durable externally controlled write. Test a registered allow, registered deny, unknown and wrong-typed argument keys, post-authorization caller mutation, missing/expired labels, OPA undefined result, stale revision, audit outage, and a newly registered handler or schema absent from policy. The last case must fail at startup, so plugin or tool growth cannot silently create an ungoverned sink.

AID-H-018.005-G002

Concept

Application-layer checks can fail due to bugs, alternative code paths, or compromised runtimes. High-assurance environments require an infrastructure-layer control that restricts where agent workloads can send traffic, regardless of what the LLM decides.

Goals

• Default-deny outbound from agent workloads.
• Force all egress through a choke point (egress gateway / firewall / proxy) for allowlisting + logging.
• Block bypass paths (direct IP egress, raw SMTP, arbitrary DNS).
• Change-managed allowlists (GitOps / ticketed approvals), versioned and auditable.

Step-by-step

1. Segment workloads: run agents in dedicated namespaces/workloads by risk (e.g., agent-readonly, agent-highrisk). Attach distinct egress policies.
2. Enforce baseline default-deny with Kubernetes NetworkPolicy (or CiliumNetworkPolicy) so agent pods cannot talk to the Internet directly.
3. Force egress via a gateway: allow only traffic to an Istio/Envoy egress gateway (or an enterprise firewall/proxy). Block all other external destinations.
4. Allowlist destinations centrally: create explicit allowlists for required external SaaS endpoints using ServiceEntry (Istio) or firewall policy objects. Prefer allowlisting by domain + SNI where possible and pin to known IP ranges when feasible.
5. Use an internal proxy pattern: tools should call an internal proxy service ("tool proxy") that validates destination + request schema. The proxy then makes outbound calls via the egress gateway.
6. Logging + detection: export egress gateway logs (SNI/host/path/status) to SIEM and correlate with tool-call logs (same request_id/trace_id).

Kubernetes NetworkPolicy (default-deny + allow only egress gateway)

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agent-egress-default-deny
  namespace: agent
spec:
  podSelector: {}
  policyTypes:
  - Egress
  egress:
  # Allow DNS only to cluster DNS (adjust to your DNS setup)
  - to:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/metadata.name: kube-system
      podSelector:
        matchLabels:
          k8s-app: kube-dns
    ports:
    - protocol: UDP
      port: 53
    - protocol: TCP
      port: 53
  # Allow egress only to Istio egress gateway (or your proxy/firewall)
  - to:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/metadata.name: istio-system
      podSelector:
        matchLabels:
          istio: egressgateway
    ports:
    - protocol: TCP
      port: 443

Istio: allowlisted external hosts via ServiceEntry + route through EgressGateway

apiVersion: networking.istio.io/v1beta1
kind: ServiceEntry
metadata:
  name: allow-approved-saas
  namespace: agent
spec:
  hosts:
  - api.approved-saas.com
  - webhook.approved-saas.com
  location: MESH_EXTERNAL
  ports:
  - number: 443
    name: tls
    protocol: TLS
  resolution: DNS
---
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: route-egress-approved-saas
  namespace: agent
spec:
  hosts:
  - api.approved-saas.com
  - webhook.approved-saas.com
  gateways:
  - mesh
  - istio-system/istio-egressgateway
  tls:
  - match:
    - gateways: ["mesh"]
      sniHosts: ["api.approved-saas.com", "webhook.approved-saas.com"]
    route:
    - destination:
        host: istio-egressgateway.istio-system.svc.cluster.local
        port:
          number: 443

Practical hardening tips

• Block raw SMTP from agent runtimes. Require a controlled email service that enforces approval gates, DLP checks, and audit logs.
• Prevent DNS bypass: consider DNS policy/allowlists and restrict egress by both domain and IP ranges. In high-assurance environments, disallow direct IP egress entirely.
• Use workload identity: bind egress permissions to workload identity (service account, SPIFFE/SPIRE) so only approved runtimes can reach the gateway.
• Centralize allowlist changes: manage allowlists via GitOps (pull requests + approvals) and tag with policy_version to align with app-layer logs.

What this does NOT replace

Network-layer controls cannot determine whether the content contains secrets. They must be paired with Strategy 1 (value tagging + sink checks) to prevent exfiltration via allowed destinations (e.g., approved SaaS webhooks).

Keep the action safely bounded

Reconcile the complete population of every agent pod/workload instance and all attempted outbound connections; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Verify safely

A separately credentialed verifier independently replays effective network and mesh policy from each workload with allowed, denied, DNS, direct-IP, redirect, SNI/Host mismatch, IPv6, alternate-port, sidecar-bypass, and gateway-outage fixtures, reads back effective state from the enforcement system, exercises positive, negative, boundary, stale-evidence, and dependency-failure fixtures, and reconciles every eligible item.

AID-H-018.005-G003

Concept:

Tool sinks are not the only exfiltration path. The model call itself is a sink: confidential tool output, retrieved documents, customer data, secrets, or regulated fields can be copied into the prompt body and sent to an internal or external model provider. This guidance extends value-level taint enforcement to the model.invoke() boundary. It does not replace output redaction in AID-D-003.002 or purpose/lifecycle authorization in AID-H-029.

Step 1: Represent prompt parts as labeled values

# File: llm_gateway/prompt_labels.py
from __future__ import annotations

from dataclasses import dataclass, asdict
from typing import Literal


Sensitivity = Literal["public", "internal", "confidential", "pii", "secret"]
ProviderTrust = Literal["internal_model", "approved_external_model", "unapproved_external_model"]


@dataclass(frozen=True)
class PromptPart:
    tenant_id: str
    content_id: str
    content: str
    source: str
    sensitivity: Sensitivity
    allowed_model_providers: list[str]
    purpose: str

    def to_audit_safe(self) -> dict:
        safe = asdict(self)
        safe.pop("content", None)
        safe["content_length"] = len(self.content)
        return safe

Step 2: Enforce policy at the model gateway

# File: llm_gateway/model_call_gate.py
from __future__ import annotations

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

from llm_gateway.prompt_labels import PromptPart, ProviderTrust


class PromptSinkDenied(PermissionError):
    pass


@dataclass(frozen=True)
class ModelTarget:
    provider: str
    model: str
    trust: ProviderTrust
    tenant_id: str


def hash_content(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def enforce_prompt_sink(parts: Iterable[PromptPart], target: ModelTarget, *, policy_version: str) -> dict:
    decisions = []
    denied_reasons = []

    for part in parts:
        if part.tenant_id != target.tenant_id:
            denied_reasons.append({
                "content_id": part.content_id,
                "reason": "cross_tenant_prompt_value",
            })
        if target.provider not in set(part.allowed_model_providers):
            denied_reasons.append({
                "content_id": part.content_id,
                "reason": "provider_not_allowed_for_value",
                "provider": target.provider,
            })
        if target.trust == "unapproved_external_model" and part.sensitivity in {"confidential", "pii", "secret"}:
            denied_reasons.append({
                "content_id": part.content_id,
                "reason": "sensitive_value_to_unapproved_external_model",
                "sensitivity": part.sensitivity,
            })
        decisions.append({
            **part.to_audit_safe(),
            "content_sha256": hash_content(part.content),
        })

    audit_event = {
        "event_type": "prompt_sink_decision",
        "observed_at": datetime.now(timezone.utc).isoformat(),
        "target": target.__dict__,
        "policy_version": policy_version,
        "parts": decisions,
        "decision": "deny" if denied_reasons else "allow",
        "denied_reasons": denied_reasons,
    }
    print(json.dumps(audit_event, sort_keys=True))

    if denied_reasons:
        raise PromptSinkDenied("Prompt sink policy denied model call")
    return audit_event

Step 3: Call the LLM only after the gate allows it

# File: llm_gateway/safe_invoke.py
from __future__ import annotations

import hashlib

from llm_gateway.model_call_gate import ModelTarget, enforce_prompt_sink
from llm_gateway.prompt_labels import PromptPart


from typing import Protocol


class ProviderClient(Protocol):
    def invoke(self, provider: str, model: str, prompt: str) -> str:
        ...


class EvidenceWriter(Protocol):
    def write(self, event: dict) -> None:
        ...


def safe_model_invoke(
    parts: list[PromptPart],
    target: ModelTarget,
    *,
    policy_version: str,
    provider_client: ProviderClient,
    evidence_writer: EvidenceWriter,
) -> str:
    evidence = enforce_prompt_sink(parts, target, policy_version=policy_version)
    prompt = "\n\n".join(part.content for part in parts)
    prompt_sha256 = hashlib.sha256(prompt.encode("utf-8")).hexdigest()
    evidence_writer.write({**evidence, "phase": "authorized", "prompt_sha256": prompt_sha256})
    try:
        response = provider_client.invoke(target.provider, target.model, prompt)
    except Exception as exc:
        evidence_writer.write({**evidence, "phase": "provider_error", "prompt_sha256": prompt_sha256, "error_type": type(exc).__name__})
        raise
    evidence_writer.write({**evidence, "phase": "completed", "prompt_sha256": prompt_sha256, "response_sha256": hashlib.sha256(response.encode("utf-8")).hexdigest()})
    return response

Verification and evidence

# File: tests/test_prompt_sink_gate.py
from llm_gateway.model_call_gate import ModelTarget, PromptSinkDenied, enforce_prompt_sink
from llm_gateway.prompt_labels import PromptPart


def test_confidential_value_denied_to_unapproved_external_model():
    part = PromptPart(
        tenant_id="prod",
        content_id="ctx-1",
        content="customer secret",
        source="crm",
        sensitivity="confidential",
        allowed_model_providers=["internal-llm"],
        purpose="support_summary",
    )
    target = ModelTarget(
        provider="consumer-ai.example",
        model="chat",
        trust="unapproved_external_model",
        tenant_id="prod",
    )
    try:
        enforce_prompt_sink([part], target, policy_version="2026.07.1")
    except PromptSinkDenied:
        return
    raise AssertionError("confidential prompt part was not denied")

Action: Put this gate immediately before every model provider call. Resolve ModelTarget from the trusted gateway route rather than caller fields, and reject any prompt part whose tenant differs from that effective route. The evidence artifact for each call is the prompt_sink_decision event containing content IDs, hashes, sensitivity labels, provider identity, policy version, and allow/deny outcome.

Keep the action safely bounded

Reconcile the complete population of every prompt part in every eligible model call and every provider/model destination; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Verify safely

A separately credentialed verifier independently replays part-to-prompt propagation and provider decisions with missing labels, mixed sensitivity, provider spoofing, model drift, cross-tenant values, serialization changes, and provider-error fixtures, reads back effective state from the enforcement system, exercises positive, negative, boundary, stale-evidence, and dependency-failure fixtures, and reconciles every eligible item.

Open-source tools

• Open Policy Agent (OPA)
• Envoy Proxy
• Istio
• Cilium
• Kubernetes
• LlamaFirewall (agent-action alignment and custom sink-risk scanning; value provenance, taint propagation, and sink authorization remain external)

Commercial / hosted tools

• Palo Alto Networks Next-Generation Firewall
• Prisma Access
• Zscaler Internet Access
• Cloudflare Gateway
• AWS Network Firewall
• Azure Firewall

Threat mappings

MITRE ATLAS

• AML.T0110 AI Agent Tool Poisoning
• AML.T0110.002 AI Agent Tool Poisoning: Runtime Response (taint tracking contains poisoned-response propagation to sensitive sinks)
• AML.T0051 LLM Prompt Injection
• AML.T0051.001 LLM Prompt Injection: Indirect (prevents indirect injection from diverting data flows to attacker-controlled sinks)
• AML.T0086 Exfiltration via AI Agent Tool Invocation
• AML.T0053 AI Agent Tool Invocation (sink enforcement controls what tools can receive sensitive data)
• AML.T0085 Data from AI Services (sink enforcement prevents exfiltration of data collected via AI services)
• AML.T0085.001 Data from AI Services: AI Agent Tools (prevents agent tools from sending collected data to unauthorized sinks)
• AML.T0114 AI Service Web Interface (sink enforcement blocks sensitive data sent through AI-service prompts or URLs)
• AML.T0127 Data Staged (sensitivity-tagged values cannot be written into unapproved storage sinks)

MAESTRO

• Data Exfiltration (L2)
• Data Leakage (Cross-Layer)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (provenance-aware value-to-sink policy blocks injected values from driving exfiltration or durable writes)
• LLM02:2026 Sensitive Information Disclosure
• LLM10:2026 Improper Output Handling

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation (prevents data exfiltration via legitimate tool chains)
• ASI01:2026 Agent Goal Hijack (sink enforcement prevents hijacked goals from exfiltrating data)

NIST Adversarial Machine Learning 2025

• NISTAML.036 Leaking information from user interactions
• NISTAML.039 Compromising connected resources (sink enforcement prevents reaching unauthorized resources)
• NISTAML.015 Indirect Prompt Injection (prevents injection from diverting data to attacker sinks)

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.1 Parameter Manipulation
• AISubtech-14.1.2 Insufficient Access Controls
• AISubtech-8.2.3 Data Exfiltration via Agent Tooling
• AISubtech-8.3.3 Personally Identifiable Information Exposure
• AISubtech-9.1.3 Unauthorized or Unsolicited Network Access
• AITech-12.1 Tool Exploitation
• AITech-12.2 Insecure Output Handling
• AITech-8.2 Data Exfiltration / Exposure
• AITech-8.3 Information Disclosure
• AITech-9.1 Model or Agentic System Manipulation

Google Secure AI Framework 2.0 - Risks

• SDD: Sensitive Data Disclosure (sink enforcement prevents sensitive data from flowing to unauthorized destinations)
• PIJ: Prompt Injection (sink enforcement prevents injection-driven data exfiltration)
• RA: Rogue Actions (data flow enforcement prevents rogue exfiltration actions)
• EDH: Excessive Data Handling (sink enforcement controls data handling boundaries)

Databricks AI Security Framework 3.0

• Model Serving - Inference response 10.6: Sensitive data output from a model
• Agents - Core 13.2: Tool Misuse (sink enforcement prevents data exfiltration via tool misuse)