Scope boundary

Owns fail-closed runtime allow or deny decisions for non-MCP model- or agent-triggered tool and service operations from verified subject, workload, tenant, run, delegation, tool, canonical object, type, version, action, environment, and policy attributes. It owns authoritative opaque-handle resolution and final-adapter enforcement for those paths, but not filesystem resolution, inference shared-memory topology, MCP-local object policy, or generic authorization for unrelated APIs.

AID-M-009.002: AID-H-018.002 evaluates the current request against active policy; AID-M-009.002 defines the approved authority envelope and action-risk classification supplied to that policy. An envelope is not a standing allow decision.

AID-M-009.003: AID-H-018.002 returns the authorization outcome; AID-M-009.003 proves the acting identity, original authority, delegation lineage, credential scope, and normalized authorization context. Identity and delegation evidence are policy inputs rather than authorization results.

AID-H-017.005: AID-H-018.002 enforces policy for the exact action; AID-H-017.005 publishes the signed agent-wide behavior contract. The contract constrains policy but cannot authorize execution by itself.

AID-D-003.003: AID-H-018.002 produces the canonical runtime authorization decision; AID-D-003.003 monitors those decisions, failures, and bursts for suspicious patterns. Enforcement and independent detection of enforcement outcomes must remain separate.

AID-H-017.004: AID-H-018.002 authorizes each read or write to a canonical mutable object for the verified tenant, principal, and run; AID-H-017.004 defines the ownership, isolation, retention, provenance, and expiry architecture for transient working state. Per-operation authorization does not make cross-run state architecture safe, and state isolation does not authorize an object operation.

AID-H-018.009: AID-H-018.002 authorizes the verified subject, action, and canonical object; AID-H-018.009 proves which filesystem object an authorized path actually opens or affects. Business authorization does not make pathname resolution race-safe.

AID-H-032.002: AID-H-018.002 authorizes application and service objects used by model- or agent-triggered operations; AID-H-032.002 isolates inference queues, caches, shared memory, IPC state, warm workers, and failover placement by tenant. A shared policy engine may support both controls, but their populations, enforcement points, evidence, and PASS results remain separate.

AID-H-034.003: AID-H-018.002 owns authorization at an external non-MCP application or service boundary; AID-H-034.003 owns the MCP handler's local tool, tenant, backend, and MCP-owned object checks. When an MCP handler calls an external service, it may consume that service's decision without awarding a second PASS for the same external-object authorization, but it still enforces its own MCP request boundary.

AID-H-034.006: AID-H-018.002 authorizes an application or service object whenever a non-MCP operation presents an opaque handle; AID-H-034.006 owns MCP-specific cross-call handle issuance, binding, expiry, revocation, and per-use lifecycle. Do not recreate MCP handle lifecycle in this protocol-neutral policy control.

Implementation guidance

• AID-H-018.002-G001: Use Open Policy Agent (OPA) or Casbin to make fail-closed decisions from verified identity and authoritative object attributes, then enforce the exact tenant, object, type, version, action, and run binding at the final adapter.

AID-H-018.002-G001

Concept:

Policy-based access control is only useful when one dispatcher always asks one policy decision point before a sensitive tool runs. Feed the PDP verified identity and normalized resource facts, return a structured decision object, and fail closed when the policy engine is unavailable or returns malformed data.

Step 1: Return a structured decision from policy

Do not make the PDP return only a bare boolean. Include rule ID and reason so enforcement and evidence stay explainable. Load the rule module, business-hours bounds, and policy timezone from one signature-verified OPA bundle; the trusted dispatcher supplies time normalized to that named timezone.

# File: policies/tools.rego
package agent.authz

import rego.v1

default decision := {
  "allow": false,
  "rule_id": "default_deny",
  "reason": "no matching authorization rule"
}

valid_business_hours_policy if {
  is_number(data.policy.trade_business_hours.start_hour)
  data.policy.trade_business_hours.start_hour >= 0
  data.policy.trade_business_hours.start_hour == floor(data.policy.trade_business_hours.start_hour)
  is_number(data.policy.trade_business_hours.end_hour)
  data.policy.trade_business_hours.end_hour <= 24
  data.policy.trade_business_hours.end_hour == floor(data.policy.trade_business_hours.end_hour)
  data.policy.trade_business_hours.start_hour < data.policy.trade_business_hours.end_hour
  is_string(data.policy.trade_business_hours.timezone)
  data.policy.trade_business_hours.timezone != ""
}

valid_time_context if {
  is_number(input.time.hour)
  input.time.hour >= 0
  input.time.hour < 24
  input.time.hour == floor(input.time.hour)
  is_string(input.time.timezone)
  input.time.timezone == data.policy.trade_business_hours.timezone
}

business_hours if {
  valid_business_hours_policy
  valid_time_context
  input.time.hour >= data.policy.trade_business_hours.start_hour
  input.time.hour < data.policy.trade_business_hours.end_hour
}

decision := {
  "allow": true,
  "rule_id": "trade.business_hours.finance_analyst",
  "reason": "finance analyst using approved account during business hours"
} if {
  input.tool_name == "execute_trade"
  input.user.role == "finance_analyst"
  is_array(input.user.allowed_account_ids)
  input.resource.canonical_id in input.user.allowed_account_ids
  input.resource.tenant_id == input.user.tenant_id
  input.resource.resource_type == "trading_account"
  input.resource.action == "execute_trade"
  is_string(input.resource.resource_version)
  input.resource.resource_version != ""
  business_hours
}

decision := {
  "allow": false,
  "rule_id": "trade.after_hours_block",
  "reason": "high-risk trading blocked outside business hours"
} if {
  input.tool_name == "execute_trade"
  valid_business_hours_policy
  valid_time_context
  not business_hours
}

Step 2: Build one fail-closed PDP client in the dispatcher path

# File: agent/policy_client.py
from __future__ import annotations

import json
import math
import os
from dataclasses import dataclass
from urllib.parse import urlsplit

import requests


def _required_policy_setting(name: str) -> str:
    value = os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"verified authorization-policy setting is absent: {name}")
    return value.strip()


OPA_URL = _required_policy_setting("VERIFIED_H018_AUTHZ_POLICY_URL")
OPA_CA_BUNDLE = _required_policy_setting("VERIFIED_H018_AUTHZ_POLICY_CA_BUNDLE")
OPA_CLIENT_CERT = _required_policy_setting("VERIFIED_H018_AUTHZ_POLICY_CLIENT_CERT")
OPA_CLIENT_KEY = _required_policy_setting("VERIFIED_H018_AUTHZ_POLICY_CLIENT_KEY")
OPA_TIMEOUT_SECONDS = float(_required_policy_setting("VERIFIED_H018_AUTHZ_POLICY_TIMEOUT_SECONDS"))
OPA_RESPONSE_MAX_BYTES = int(_required_policy_setting("VERIFIED_H018_AUTHZ_POLICY_RESPONSE_MAX_BYTES"))
if not math.isfinite(OPA_TIMEOUT_SECONDS) or OPA_TIMEOUT_SECONDS <= 0 or OPA_RESPONSE_MAX_BYTES <= 0:
    raise RuntimeError("verified authorization-policy timeout must be finite and positive")
try:
    _opa_endpoint = urlsplit(OPA_URL)
except ValueError as error:
    raise RuntimeError("authorization-policy URL is invalid") from error
if (
    _opa_endpoint.scheme.lower() != "https"
    or not _opa_endpoint.hostname
    or _opa_endpoint.username is not None
    or _opa_endpoint.password is not None
    or _opa_endpoint.query
    or _opa_endpoint.fragment
):
    raise RuntimeError("authorization-policy URL must be an authenticated HTTPS endpoint without embedded credentials")


@dataclass(frozen=True)
class PolicyDecision:
    allow: bool
    rule_id: str
    reason: str


class PolicyDenied(RuntimeError):
    pass


def authorize_tool_call(*, tool_name: str, verified_user: dict, resource: dict, time_context: dict) -> PolicyDecision:
    payload = {
        "input": {
            "tool_name": tool_name,
            "user": verified_user,
            "resource": resource,
            "time": time_context,
        }
    }

    try:
        with requests.post(
            OPA_URL,
            json=payload,
            timeout=OPA_TIMEOUT_SECONDS,
            verify=OPA_CA_BUNDLE,
            cert=(OPA_CLIENT_CERT, OPA_CLIENT_KEY),
            allow_redirects=False,
            stream=True,
        ) as response:
            response.raise_for_status()
            raw = bytearray()
            for chunk in response.iter_content(chunk_size=min(65536, OPA_RESPONSE_MAX_BYTES + 1)):
                if chunk:
                    raw.extend(chunk)
                if len(raw) > OPA_RESPONSE_MAX_BYTES:
                    raise ValueError("policy_engine_response_too_large")
        body = json.loads(bytes(raw))
    except requests.RequestException as exc:
        raise PolicyDenied("policy_engine_transport_or_http_failure") from exc
    except (TypeError, ValueError) as exc:
        raise PolicyDenied("policy_engine_invalid_json") from exc

    if not isinstance(body, dict):
        raise PolicyDenied("policy_engine_invalid_response_envelope")
    result = body.get("result")
    if not isinstance(result, dict) or set(result) != {"allow", "rule_id", "reason"}:
        raise PolicyDenied("policy_engine_invalid_result_schema")
    if type(result["allow"]) is not bool:
        raise PolicyDenied("policy_engine_allow_must_be_boolean")
    if not isinstance(result["rule_id"], str) or not result["rule_id"].strip():
        raise PolicyDenied("policy_engine_rule_id_invalid")
    if not isinstance(result["reason"], str) or not result["reason"].strip():
        raise PolicyDenied("policy_engine_reason_invalid")

    decision = PolicyDecision(
        allow=result["allow"],
        rule_id=result["rule_id"].strip(),
        reason=result["reason"].strip(),
    )

    print(
        json.dumps(
            {
                "event_type": "pbac_decision",
                "tool_name": tool_name,
                "rule_id": decision.rule_id,
                "allow": decision.allow,
                "reason": decision.reason,
            }
        )
    )
    if not decision.allow:
        raise PolicyDenied(f"{decision.rule_id}: {decision.reason}")
    return decision

Step 3: Resolve opaque handles authoritatively and enforce at the final adapter

# File: agent/dispatcher.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol

from agent.policy_client import PolicyDecision, authorize_tool_call


@dataclass(frozen=True)
class BoundResource:
    canonical_id: str
    tenant_id: str
    version: str
    resource_type: str


class ResourceRepository(Protocol):
    def resolve_authoritative(
        self, *, tenant_id: str, opaque_handle: str, expected_type: str
    ) -> BoundResource | None: ...


class ToolHandler(Protocol):
    resource_type: str
    action: str

    def execute_authorized(
        self, *, resource: BoundResource, arguments: dict,
        authorization: PolicyDecision,
    ): ...


def dispatch_tool(
    tool_name: str,
    args: dict,
    *,
    verified_context: dict,
    trusted_time_context: dict,
    handlers: dict[str, ToolHandler],
    resource_repository: ResourceRepository,
):
    handler = handlers.get(tool_name)
    if handler is None:
        raise PermissionError("unknown_or_unauthorized_tool")
    required_context = {
        "subject_id", "workload_id", "tenant_id", "run_id",
        "role", "allowed_account_ids",
    }
    string_context = required_context - {"allowed_account_ids"}
    allowed_account_ids = verified_context.get("allowed_account_ids")
    if (
        not required_context.issubset(verified_context)
        or any(
            not isinstance(verified_context[key], str)
            or not verified_context[key]
            for key in string_context
        )
        or not isinstance(allowed_account_ids, list)
        or not allowed_account_ids
        or any(not isinstance(item, str) or not item for item in allowed_account_ids)
        or len(allowed_account_ids) != len(set(allowed_account_ids))
    ):
        raise PermissionError("verified_identity_context_incomplete")

    opaque_handle = args.get("resource_handle")
    if not isinstance(opaque_handle, str) or not opaque_handle:
        raise PermissionError("resource_not_found_or_not_accessible")
    resource = resource_repository.resolve_authoritative(
        tenant_id=verified_context["tenant_id"],
        opaque_handle=opaque_handle,
        expected_type=handler.resource_type,
    )
    if (
        resource is None
        or resource.tenant_id != verified_context["tenant_id"]
        or resource.resource_type != handler.resource_type
        or not resource.canonical_id
        or not resource.version
    ):
        # Deliberately do not reveal whether another tenant owns the handle.
        raise PermissionError("resource_not_found_or_not_accessible")

    decision = authorize_tool_call(
        tool_name=tool_name,
        verified_user=verified_context,
        resource={
            "canonical_id": resource.canonical_id,
            "tenant_id": resource.tenant_id,
            "resource_type": resource.resource_type,
            "resource_version": resource.version,
            "action": handler.action,
        },
        time_context=trusted_time_context,
    )
    clean_arguments = {
        key: value for key, value in args.items() if key != "resource_handle"
    }
    return handler.execute_authorized(
        resource=resource,
        arguments=clean_arguments,
        authorization=decision,
    )

The final adapter receives the canonical object and version, not the caller's raw handle. It must enforce the same tenant, object, type, version, action, and policy binding in the storage transaction or downstream service call; a changed version, revoked object, stale delegation, or missing dependency denies the operation. A gateway or wrapper may transport this decision, but it cannot award another PASS for the same object authorization.

Bind mutable state to the verified run

Inventory mutable registries, tags, manifests, notes, queues, scratch objects, and agent metadata reachable by a model- or agent-triggered path. Bind every read and write to authoritative tenant, principal, and run identity. Prefer digest-addressed immutable objects and compare-and-swap promotion; deny cross-run writes and stale versions. Shared caches remain allowed only when entries are immutable, content-addressed, contain no tenant- or run-specific authority, and their lookup cannot disclose another tenant's object.

# File: agent/run_state.py
def update_run_state(
    *,
    repository,
    verified_context: dict,
    opaque_handle: str,
    expected_version: str,
    replacement: dict,
):
    current = repository.resolve_for_update(
        tenant_id=verified_context["tenant_id"],
        subject_id=verified_context["subject_id"],
        run_id=verified_context["run_id"],
        opaque_handle=opaque_handle,
    )
    if current is None:
        raise PermissionError("state_not_found_or_not_accessible")
    updated = repository.compare_and_swap(
        canonical_id=current.canonical_id,
        tenant_id=verified_context["tenant_id"],
        subject_id=verified_context["subject_id"],
        run_id=verified_context["run_id"],
        expected_version=expected_version,
        replacement=replacement,
    )
    if updated is None:
        raise RuntimeError("state_version_changed")
    return updated

Derive verified_context from the authenticated workload and trusted orchestration record, never from model output or caller-controlled arguments. Queue workers and failover workers must re-resolve the same binding rather than trusting serialized identity fields.

Operational notes

• The model must never author its own role, tenant, or approval status; pass those from your already verified identity layer.
• Normalize high-risk parameters into stable resource attributes before the OPA call so policy rules do not depend on free-form prompt text.
• Derive the current hour and named timezone from a trusted clock service or dispatcher runtime; never accept time context from model output or caller-controlled tool arguments.
• Fail closed on PDP timeout, malformed response, or missing context fields.

Action: Put one authenticated mTLS PDP client in front of every high-risk tool dispatch. Query a signature-verified policy bundle with verified identity, normalized resource facts, and trusted time context, then block execution unless the exact response schema contains a real Boolean true allow decision.

Keep the action safely bounded

At the fail-closed policy decision point called by the trusted dispatcher before execution, bind a signed, versioned policy (policy version and SHA-256 digest) to the exact asset_id and asset version for policy bundle, data/identity registry snapshots, policy engine, dispatcher, and tool/resource registry. Reconcile the complete population of every eligible tool call and every policy decision consumed by a dispatcher; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Verify safely

A separately credentialed verifier independently replays the policy bundle and authoritative attributes and exercises allow, deny, missing-attribute, stale-policy, revoked-object, wrong-type, forged-handle, cross-tenant, cross-principal, cross-run, stale-version, unknown-tool, and policy-engine outage cases. Enumerate REST, RPC, SDK, queue, worker, callback, retry, and internal-service paths that can reach each applicable object operation; direct or alternate paths must enforce the same final binding and produce no duplicate PASS. Read back the object version and effect from the authoritative service so a caller-supplied or handler-returned boolean cannot prove enforcement.

Open-source tools

• Open Policy Agent (OPA)
• Casbin
• Cedar
• Cerbos PDP
• OpenFGA

Commercial / hosted tools

• Cerbos Hub
• Permit.io
• Amazon Verified Permissions

Threat mappings

MITRE ATLAS

• AML.T0053 AI Agent Tool Invocation

MAESTRO

• Agent Tool Misuse (L7)
• Privilege Escalation (Cross-Layer)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (external policy enforcement keeps injected model output from authorizing tool use)
• LLM03:2026 Excessive Agency
• LLM08:2026 Hidden Context Exposure

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation
• ASI03:2026 Identity and Privilege Abuse
• ASI10:2026 Rogue Agents

NIST Adversarial Machine Learning 2025

• NISTAML.039 Compromising connected resources

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.1 Parameter Manipulation
• AISubtech-14.1.2 Insufficient Access Controls
• AISubtech-14.2.1 Permission Escalation via Delegation
• AISubtech-2.1.2 Capability and Permission Overreach
• AISubtech-4.2.2 Session Boundary Violation
• AISubtech-8.2.3 Data Exfiltration via Agent Tooling
• AITech-12.1 Tool Exploitation
• AITech-14.2 Abuse of Delegated Authority
• AITech-2.1 Excessive Agency
• AITech-4.2 Context Boundary Attacks
• AITech-8.2 Data Exfiltration / Exposure
• AITech-8.3 Information Disclosure

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (policy engine blocks unauthorized agent actions)

Databricks AI Security Framework 3.0

• Agents - Core 13.2: Tool Misuse (policy-based access control prevents tool misuse)
• Agents - Core 13.3: Privilege Compromise (policy engine prevents privilege escalation)
• Agents - Tools MCP Server 13.17: Confused Deputy (policy-based access control prevents confused deputy authorization bypass)