AID-H-018.004: Intent-Based Dynamic Capability Scoping

Version 1.20261015 · Actionable sub-technique

Dynamically restrict an agent's tool capabilities per request or session based on the authenticated subject, workload, initial intent, and the exact versioned tool registry enforced by the dispatcher, instead of granting a broad static role allowlist.

How it works: The grant service intersects role, workload, and request policy, selects the smallest allowed tool set and action budget, and issues a tamper-evident signed scope whose subject, workload, registry ID/version, expiry, and budget the server verifies on every action.

Pillars: app, infra

Lifecycle: building, operation

Scope boundary

Owns per-request or per-session capability minimization by intersecting authenticated subject, workload, intent, role, tool-registry version, expiry, and action-budget policy, then issuing and atomically enforcing a signed scope at the dispatcher. Alternate paths outside that dispatcher are not protected.

AID-H-017.002: AID-H-018.004 chooses the smallest currently allowed subset from the tool registry; AID-H-017.002 designs the registry itself around narrow, single-purpose, least-privilege tools. Dynamic scoping cannot make an overpowered primitive safe.

AID-H-018.007: AID-H-018.004 scopes capabilities for the current request or session; AID-H-018.007 enforces the declared resource permissions of a specific installed skill. A session grant and a skill manifest are intersecting but distinct authorization layers.

AID-M-009.001: AID-H-018.004 issues the concrete runtime tool set and budget; AID-M-009.001 defines the approved degree of autonomy for Observe, Analyze, Decide, Execute, Delegate, and Persist. Autonomy profile informs the grant but does not enumerate the exact current capabilities.

Implementation guidance

AID-H-018.004-G001

Intersect registry-bound tools, action budgets, and TTL ceilings, then enforce the signed scope with an atomic server-side counter.
# File: capability_scope.py
from __future__ import annotations

import hashlib
import os
import time
import uuid
from dataclasses import dataclass
from pathlib import Path

import jwt
import redis


ISSUER = "aidefend-control-plane"
AUDIENCE = "aidefend-tool-dispatcher"
PRIVATE_KEY = Path("/run/secrets/capability_scope_ed25519.pem").read_text(encoding="utf-8")
PUBLIC_KEY = Path("/etc/aidefend/capability_scope_ed25519.pub").read_text(encoding="utf-8")
KEY_ID = os.environ["CAPABILITY_SCOPE_KEY_ID"]


@dataclass(frozen=True)
class GrantEvidence:
    subject_id: str
    workload_identity: str
    subject_registry_id: str
    subject_registry_version: str
    role_policy_version: str
    workload_grant_version: str
    intent_policy_version: str
    role_tools: frozenset[str]
    workload_tools: frozenset[str]
    requested_tools: frozenset[str]
    role_max_actions: int
    workload_max_actions: int
    requested_max_actions: int
    role_max_ttl_seconds: int
    workload_max_ttl_seconds: int
    requested_ttl_seconds: int


@dataclass(frozen=True)
class EffectiveScope:
    tools: tuple[str, ...]
    max_actions: int
    ttl_seconds: int


def normalized_tools(values: frozenset[str]) -> set[str]:
    tools = {value.strip() for value in values if isinstance(value, str) and value.strip()}
    if not tools or any(" " in value or "/" in value for value in tools):
        raise PermissionError("invalid_or_empty_tool_grant")
    return tools


def derive_scope(evidence: GrantEvidence) -> EffectiveScope:
    if not all([
        evidence.subject_id,
        evidence.workload_identity,
        evidence.subject_registry_id,
        evidence.subject_registry_version,
    ]):
        raise PermissionError("subject_or_registry_binding_missing")
    role = normalized_tools(evidence.role_tools)
    workload = normalized_tools(evidence.workload_tools)
    requested = normalized_tools(evidence.requested_tools)
    durable_ceiling = role & workload
    if requested - durable_ceiling:
        raise PermissionError("request_attempted_tool_scope_expansion")
    tools = tuple(sorted(durable_ceiling & requested))
    actions = min(
        evidence.role_max_actions,
        evidence.workload_max_actions,
        evidence.requested_max_actions,
    )
    ttl = min(
        evidence.role_max_ttl_seconds,
        evidence.workload_max_ttl_seconds,
        evidence.requested_ttl_seconds,
    )
    if not tools or actions <= 0 or ttl <= 0:
        raise PermissionError("effective_scope_is_empty")
    return EffectiveScope(tools=tools, max_actions=actions, ttl_seconds=ttl)


def mint_scope(session_id: str, evidence: GrantEvidence) -> str:
    if not session_id:
        raise PermissionError("session_id_missing")
    effective = derive_scope(evidence)
    now = int(time.time())
    claims = {
        "iss": ISSUER,
        "aud": AUDIENCE,
        "sub": evidence.subject_id,
        "workload_identity": evidence.workload_identity,
        "subject_registry_id": evidence.subject_registry_id,
        "subject_registry_version": evidence.subject_registry_version,
        "jti": str(uuid.uuid4()),
        "iat": now,
        "nbf": now,
        "exp": now + effective.ttl_seconds,
        "session_id": session_id,
        "allowed_tools": list(effective.tools),
        "max_actions": effective.max_actions,
        "role_policy_version": evidence.role_policy_version,
        "workload_grant_version": evidence.workload_grant_version,
        "intent_policy_version": evidence.intent_policy_version,
    }
    return jwt.encode(
        claims,
        PRIVATE_KEY,
        algorithm="EdDSA",
        headers={"kid": KEY_ID, "typ": "JWT"},
    )


class RedisAtomicBudget:
    SCRIPT = """
local key = KEYS[1]
local expires_at = tonumber(ARGV[1])
local maximum = tonumber(ARGV[2])
local count = redis.call('INCR', key)
if count == 1 then
  redis.call('EXPIREAT', key, expires_at)
elseif redis.call('TTL', key) < 0 then
  return redis.error_reply('budget key lost expiry')
end
if count > maximum then
  return {0, count}
end
return {1, count}
"""

    def __init__(self, client: redis.Redis) -> None:
        self.client = client
        self.script = client.register_script(self.SCRIPT)

    def consume(self, scope_id: str, expires_at: int, maximum: int) -> tuple[bool, int]:
        try:
            allowed, count = self.script(
                keys=[f"capability_budget:{scope_id}"],
                args=[expires_at, maximum],
            )
        except redis.RedisError as exc:
            raise PermissionError("atomic_budget_store_unavailable") from exc
        return bool(allowed), int(count)


def enforce_tool_call(
    token: str,
    expected_session_id: str,
    expected_subject_id: str,
    expected_workload_identity: str,
    expected_registry_id: str,
    expected_registry_version: str,
    tool_name: str,
    budget: RedisAtomicBudget,
) -> dict:
    claims = jwt.decode(
        token,
        PUBLIC_KEY,
        algorithms=["EdDSA"],
        issuer=ISSUER,
        audience=AUDIENCE,
        options={"require": [
            "iss", "aud", "sub", "workload_identity", "subject_registry_id",
            "subject_registry_version", "jti", "iat", "nbf", "exp", "session_id",
            "allowed_tools", "max_actions",
        ]},
    )
    expected = {
        "sub": expected_subject_id,
        "workload_identity": expected_workload_identity,
        "subject_registry_id": expected_registry_id,
        "subject_registry_version": expected_registry_version,
        "session_id": expected_session_id,
    }
    if any(claims.get(field) != value for field, value in expected.items()):
        raise PermissionError("capability_scope_binding_mismatch")
    tools = claims["allowed_tools"]
    if not isinstance(tools, list) or tools != sorted(set(tools)) or tool_name not in tools:
        raise PermissionError("tool_not_in_effective_scope")
    maximum = claims["max_actions"]
    if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum <= 0:
        raise PermissionError("invalid_action_budget")
    allowed, count = budget.consume(claims["jti"], int(claims["exp"]), maximum)
    if not allowed:
        raise PermissionError("scope_action_budget_exceeded")
    return {
        "schema_version": "aidefend.capability-dispatch-decision.v1",
        "decision": "allow",
        "scope_id": claims["jti"],
        "scope_token_sha256": hashlib.sha256(token.encode()).hexdigest(),
        "subject_id": claims["sub"],
        "workload_identity": claims["workload_identity"],
        "subject_registry_id": claims["subject_registry_id"],
        "subject_registry_version": claims["subject_registry_version"],
        "session_id": claims["session_id"],
        "tool_name": tool_name,
        "action_count": count,
        "max_actions": maximum,
        "role_policy_version": claims["role_policy_version"],
        "workload_grant_version": claims["workload_grant_version"],
        "intent_policy_version": claims["intent_policy_version"],
    }
Keep the action safely bounded

Reconcile the complete population of every requested tool, scope dimension, action budget, and TTL constraint in a grant request; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Intersect rather than union subject, workload, registry, intent, and policy bounds; reject empty or widened scopes; mint a signed audience/task/tenant-bound grant with policy-derived TTL/actions and unique ID; initialize an atomic server-side counter.

Verify safely

A separately credentialed verifier independently replays grant derivation, signature/claims, and counter with widening, mixed-tenant, unknown-tool, zero/negative, replay, expiry, and concurrent-use 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.004-G002

Load grants and registry bindings from a pinned signed snapshot and require a verified external audit receipt before the dispatcher invokes any scoped tool.
Mandatory production trust boundary

The intersection code above is accepted only behind this wrapper. Authenticated subject and workload identities come from middleware; durable grants come from an exact-schema registry snapshot verified under a key pinned in the issuer and dispatcher images. Request intent may narrow those grants but cannot supply registry versions or ceilings. The dispatcher re-reads the same registry and obtains a separately signed audit receipt before returning an executable allow decision.

# File: trusted_capability_dispatch.py
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Any

import jwt
import requests

from capability_scope import (
    GrantEvidence, RedisAtomicBudget, enforce_tool_call, mint_scope,
)


AUDIT_TIMEOUT_SECONDS = float(os.environ["VERIFIED_H018_CAPABILITY_AUDIT_TIMEOUT_SECONDS"])
REGISTRY_VERIFY_TIMEOUT_SECONDS = float(os.environ["VERIFIED_H018_CAPABILITY_REGISTRY_VERIFY_TIMEOUT_SECONDS"])
AUDIT_RESPONSE_MAX_BYTES = int(os.environ["VERIFIED_H018_CAPABILITY_AUDIT_RESPONSE_MAX_BYTES"])
if (
    not math.isfinite(AUDIT_TIMEOUT_SECONDS) or AUDIT_TIMEOUT_SECONDS <= 0
    or not math.isfinite(REGISTRY_VERIFY_TIMEOUT_SECONDS)
    or REGISTRY_VERIFY_TIMEOUT_SECONDS <= 0
    or AUDIT_RESPONSE_MAX_BYTES <= 0
):
    raise RuntimeError("verified capability timeouts must be finite and positive")

REGISTRY_PATH = Path("/etc/aidefend/capabilities/registry.json")
REGISTRY_SIGNATURE = Path("/etc/aidefend/capabilities/registry.sig")
REGISTRY_TRUST_KEY = Path("/opt/aidefend/trust/capability-registry-authority.pub")
AUDIT_RECEIPT_KEY = Path("/opt/aidefend/trust/capability-audit-receipt.pub").read_text()
AUDIT_URL = "https://capability-audit.internal.example/v1/decisions"
AUDIT_CA = "/opt/aidefend/trust/capability-audit-ca.pem"
AUDIT_CLIENT_CERT = "/var/run/secrets/capability-audit/client.crt"
AUDIT_CLIENT_KEY = "/var/run/secrets/capability-audit/client.key"
REGISTRY_FIELDS = {
    "schema_version", "registry_id", "registry_version",
    "role_policy_version", "subjects", "workloads",
}
GRANT_FIELDS = {"tools", "max_actions", "max_ttl_seconds", "grant_version"}
INTENT_FIELDS = {
    "intent_policy_version", "requested_tools",
    "requested_max_actions", "requested_ttl_seconds",
}


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

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

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


def load_registry() -> tuple[dict[str, Any], str]:
    raw = REGISTRY_PATH.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h018-registry-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run([
            "cosign", "verify-blob", "--key", str(REGISTRY_TRUST_KEY),
            "--bundle", str(REGISTRY_SIGNATURE), str(snapshot),
        ], check=True, capture_output=True, text=True, timeout=REGISTRY_VERIFY_TIMEOUT_SECONDS)
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    registry = strict_json_object(raw)
    if not isinstance(registry, dict) or set(registry) != REGISTRY_FIELDS:
        raise PermissionError("capability_registry_schema_mismatch")
    if registry["schema_version"] != "aidefend.capability-registry.v1":
        raise PermissionError("capability_registry_version_unsupported")
    for field in ("registry_id", "registry_version", "role_policy_version"):
        if not isinstance(registry[field], str) or not registry[field]:
            raise PermissionError("capability_registry_binding_missing")
    if not isinstance(registry["subjects"], dict) or not isinstance(registry["workloads"], dict):
        raise PermissionError("capability_registry_grants_missing")
    return registry, hashlib.sha256(raw).hexdigest()


def validated_grant(grant: object, kind: str) -> dict:
    if not isinstance(grant, dict) or set(grant) != GRANT_FIELDS:
        raise PermissionError(f"{kind}_grant_schema_mismatch")
    tools = grant["tools"]
    if (
        not isinstance(tools, list) or not tools or tools != sorted(set(tools))
        or any(not isinstance(tool, str) or not tool for tool in tools)
    ):
        raise PermissionError(f"{kind}_tool_grant_invalid")
    for field in ("max_actions", "max_ttl_seconds"):
        value = grant[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise PermissionError(f"{kind}_{field}_invalid")
    if not isinstance(grant["grant_version"], str) or not grant["grant_version"]:
        raise PermissionError(f"{kind}_grant_version_missing")
    return grant


def trusted_grant_evidence(
    authenticated_subject: str,
    authenticated_workload: str,
    intent: dict[str, Any],
) -> tuple[GrantEvidence, dict, str]:
    if not isinstance(intent, dict) or set(intent) != INTENT_FIELDS:
        raise PermissionError("intent_scope_schema_mismatch")
    registry, registry_sha256 = load_registry()
    try:
        subject = validated_grant(
            registry["subjects"][authenticated_subject], "subject"
        )
        workload = validated_grant(
            registry["workloads"][authenticated_workload], "workload"
        )
    except KeyError as exc:
        raise PermissionError("authenticated_identity_has_no_registry_grant") from exc
    requested_tools = intent["requested_tools"]
    if (
        not isinstance(requested_tools, list) or not requested_tools
        or requested_tools != sorted(set(requested_tools))
    ):
        raise PermissionError("requested_tool_scope_invalid")
    for field in ("requested_max_actions", "requested_ttl_seconds"):
        value = intent[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise PermissionError("requested_ceiling_invalid")
    evidence = GrantEvidence(
        subject_id=authenticated_subject,
        workload_identity=authenticated_workload,
        subject_registry_id=registry["registry_id"],
        subject_registry_version=registry["registry_version"],
        role_policy_version=registry["role_policy_version"],
        workload_grant_version=workload["grant_version"],
        intent_policy_version=intent["intent_policy_version"],
        role_tools=frozenset(subject["tools"]),
        workload_tools=frozenset(workload["tools"]),
        requested_tools=frozenset(requested_tools),
        role_max_actions=subject["max_actions"],
        workload_max_actions=workload["max_actions"],
        requested_max_actions=intent["requested_max_actions"],
        role_max_ttl_seconds=subject["max_ttl_seconds"],
        workload_max_ttl_seconds=workload["max_ttl_seconds"],
        requested_ttl_seconds=intent["requested_ttl_seconds"],
    )
    return evidence, registry, registry_sha256


def canonical_digest(value: dict) -> str:
    return hashlib.sha256(json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode()).hexdigest()


def write_and_verify_audit_receipt(decision: dict) -> tuple[str, str]:
    digest = canonical_digest(decision)
    with requests.post(
        AUDIT_URL,
        json={"decision": decision, "decision_sha256": digest},
        cert=(AUDIT_CLIENT_CERT, AUDIT_CLIENT_KEY),
        verify=AUDIT_CA,
        timeout=AUDIT_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, AUDIT_RESPONSE_MAX_BYTES + 1)):
            if chunk:
                raw.extend(chunk)
            if len(raw) > AUDIT_RESPONSE_MAX_BYTES:
                raise PermissionError("audit response exceeds the signed decoded-byte cap")
    body = json.loads(bytes(raw))
    if not isinstance(body, dict) or set(body) != {"receipt_jws"}:
        raise PermissionError("audit receipt response schema differs")
    token = body["receipt_jws"]
    if not isinstance(token, str) or not token:
        raise PermissionError("audit receipt is missing")
    claims = jwt.decode(
        token,
        AUDIT_RECEIPT_KEY,
        algorithms=["EdDSA"],
        issuer="aidefend-capability-audit",
        audience="aidefend-tool-dispatcher",
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti",
            "decision_sha256", "scope_id", "tool_name", "action_count",
        ]},
    )
    expected = {
        "decision_sha256": digest,
        "scope_id": decision["scope_id"],
        "tool_name": decision["tool_name"],
        "action_count": decision["action_count"],
    }
    if any(claims.get(field) != value for field, value in expected.items()):
        raise PermissionError("audit_receipt_binding_mismatch")
    return claims["jti"], hashlib.sha256(token.encode()).hexdigest()


def mint_from_trusted_registry(
    authenticated_subject: str,
    authenticated_workload: str,
    session_id: str,
    intent: dict[str, Any],
) -> str:
    evidence, _, _ = trusted_grant_evidence(
        authenticated_subject, authenticated_workload, intent
    )
    return mint_scope(session_id, evidence)


def authorize_before_tool(
    *, token: str, authenticated_subject: str, authenticated_workload: str,
    session_id: str, tool_name: str, budget: RedisAtomicBudget,
) -> dict:
    registry, registry_sha256 = load_registry()
    # Re-read and validate both live grants immediately before dispatch.
    validated_grant(registry["subjects"].get(authenticated_subject), "subject")
    validated_grant(registry["workloads"].get(authenticated_workload), "workload")
    decision = enforce_tool_call(
        token=token,
        expected_session_id=session_id,
        expected_subject_id=authenticated_subject,
        expected_workload_identity=authenticated_workload,
        expected_registry_id=registry["registry_id"],
        expected_registry_version=registry["registry_version"],
        tool_name=tool_name,
        budget=budget,
    )
    decision["registry_sha256"] = registry_sha256
    receipt_id, receipt_sha256 = write_and_verify_audit_receipt(decision)
    decision["audit_receipt_id"] = receipt_id
    decision["audit_receipt_jws_sha256"] = receipt_sha256
    return decision

Fail semantics: registry signature/schema failure, absent live grant, registry drift, invalid scope, budget-store failure, audit-service failure, or receipt mismatch denies execution. The dispatcher invokes the tool only after authorize_before_tool returns a verified receipt-bound decision. Persist the registry/signature, scope digest, budget event, decision, and receipt in the external append-only audit store; independently replay the intersection and receipt verification.

Keep the action safely bounded

Reconcile the complete population of every eligible grant use and all active dispatcher instances; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Scope externally initiated tasks without laundering their origin

When the initiating context includes a link-prefill or external-event origin, resolve the corresponding ingress/event receipt from the trusted registry. Preserve that ancestry across turns and sub-agents. The receipt may prove submission or event admission; it does not itself authorize private reads, memory writes, exports, or arbitrary tools. Intersect capabilities with a separately authorized task scope and the caller/workflow's current entitlements. Sensitive read tools such as mailbox and internal-document search need the same task binding as writes. A natural-language instruction to “trust this source” cannot widen the capability set. Reconfirmation must bind the specific requested expansion; later turns cannot silently remove the restriction. Test a link-derived private read, a follow-up write, and a sub-agent relay against the same task scope.

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-018.004: Intent-Based Dynamic Capability Scoping | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service