Scope boundary

Owns per-invocation sandbox freshness, one-shot execution, deterministic teardown, lease revocation, and evidence that no workspace, process, or residue survived the tool call. It delegates isolation strength and egress-policy definition to their canonical controls.

AID-I-001.002: AID-I-001.003 owns the single-use lifecycle; AID-I-001.002 owns the microVM, user-space-kernel, WASM, or OS-level isolation profile. This control binds an approved profile to one invocation but does not independently establish the strength of that profile.

AID-I-001.004: AID-I-001.003 creates and destroys the one-shot sandbox; AID-I-001.004 defines and independently enforces its outbound network policy. A teardown receipt does not prove default-deny egress, and an egress policy does not prove that the invocation used a fresh workspace or was destroyed.

AID-I-001.005: AID-I-001.003 provides one-shot execution and teardown for a tool invocation; AID-I-001.005 detonates an artifact across a declared case population and emits attested raw behavioral observations. Detonation consumes sandbox lifecycle evidence but has a separate artifact-analysis population and output; ordinary ephemeral tool execution is not a behavioral-analysis result.

Implementation guidance

• AID-I-001.003-G001: Provision a fresh, single-use sandbox (microVM / gVisor / Kata) for every tool invocation, execute once, then destroy it.
• AID-I-001.003-G002: Broker managed agent sandboxes through a lifecycle-controlled interface that enforces one-shot execution, delegated runtime-profile selection, delegated egress-policy binding, and teardown proof.

AID-I-001.003-G001

Concept:

High-risk tool execution should never happen in a reused runtime. The control creates a new isolated workload for each invocation, passes only an opaque signed-manifest reference, executes exactly one approved profile, captures a signed result reference, and destroys every runtime-owned artifact. The teardown path must run even when pod creation, execution, receipt collection, or polling fails.

Step 1: Create a one-shot sandbox pod with policy-owned limits

Use a runtime class such as Kata or gVisor so the tool runs in a fresh microVM-backed or syscall-intercepted boundary. Resolve the runtime class, image digest, resource bounds, and deadlines from a signed policy; activeDeadlineSeconds protects the cluster if the client dies and the independent poll deadline protects the caller.

# File: isolate/single_use_sandbox.py
from __future__ import annotations

import json
import re
import time
from dataclasses import dataclass
from typing import Callable

from kubernetes import client, config
from kubernetes.client.exceptions import ApiException


INVOCATION_ID = re.compile(r"^[a-z0-9][a-z0-9-]{7,62}$")


@dataclass(frozen=True)
class VerifiedSandboxPolicy:
    policy_version: str
    policy_sha256: str
    namespace: str
    runtime_class_name: str
    runner_image_digest: str
    max_runtime_seconds: int
    teardown_timeout_seconds: int
    poll_interval_seconds: float
    work_volume_size: str
    cpu_request: str
    cpu_limit: str
    memory_request: str
    memory_limit: str
    termination_grace_period_seconds: int
    run_as_user: int
    run_as_group: int
    fs_group: int


@dataclass(frozen=True)
class VerifiedInvocation:
    invocation_id: str
    manifest_sha256: str
    tenant_id: str
    task_id: str


@dataclass(frozen=True)
class ReceiptVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_execution_receipt: Callable[[str, VerifiedInvocation], dict[str, object]]
    verify_absence: Callable[[VerifiedInvocation, str], dict[str, object]]


@dataclass(frozen=True)
class EvidenceWriter:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict[str, object]], dict[str, str]]


def validate_inputs(policy: VerifiedSandboxPolicy, invocation: VerifiedInvocation) -> None:
    if not INVOCATION_ID.fullmatch(invocation.invocation_id):
        raise ValueError("invalid invocation_id")
    if "@sha256:" not in policy.runner_image_digest:
        raise ValueError("runner image must be an immutable digest reference")
    if not policy.policy_version or not re.fullmatch(r"sha256:[0-9a-f]{64}", policy.policy_sha256):
        raise ValueError("verified sandbox policy identity is invalid")
    if (
        isinstance(policy.termination_grace_period_seconds, bool)
        or not isinstance(policy.termination_grace_period_seconds, int)
        or policy.termination_grace_period_seconds < 0
    ):
        raise ValueError("signed termination grace period is invalid")
    for field in ("run_as_user", "run_as_group", "fs_group"):
        value = getattr(policy, field)
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError("signed sandbox identity is invalid: " + field)
    if min(
        policy.max_runtime_seconds,
        policy.teardown_timeout_seconds,
        policy.poll_interval_seconds,
    ) <= 0:
        raise ValueError("signed lifecycle timeouts must be positive")


def build_sandbox_pod(
    policy: VerifiedSandboxPolicy,
    invocation: VerifiedInvocation,
) -> client.V1Pod:
    return client.V1Pod(
        metadata=client.V1ObjectMeta(
            name=invocation.invocation_id,
            labels={
                "app": "tool-sandbox",
                "sandbox-mode": "single-use",
                "aidefend-invocation": invocation.invocation_id,
            },
            annotations={
                "aidefend.io/manifest-sha256": invocation.manifest_sha256,
                "aidefend.io/policy-version": policy.policy_version,
                "aidefend.io/policy-sha256": policy.policy_sha256,
            },
        ),
        spec=client.V1PodSpec(
            runtime_class_name=policy.runtime_class_name,
            restart_policy="Never",
            active_deadline_seconds=policy.max_runtime_seconds,
            termination_grace_period_seconds=policy.termination_grace_period_seconds,
            automount_service_account_token=False,
            security_context=client.V1PodSecurityContext(
                run_as_non_root=True, run_as_user=policy.run_as_user, run_as_group=policy.run_as_group,
                fs_group=policy.fs_group, fs_group_change_policy="OnRootMismatch",
                seccomp_profile=client.V1SeccompProfile(type="RuntimeDefault"),
            ),
            containers=[
                client.V1Container(
                    name="runner",
                    image=policy.runner_image_digest,
                    command=["/opt/aidefend/run-once"],
                    args=[
                        "--invocation-id", invocation.invocation_id,
                        "--manifest-sha256", invocation.manifest_sha256,
                    ],
                    volume_mounts=[client.V1VolumeMount(name="work", mount_path="/work")],
                    resources=client.V1ResourceRequirements(
                        requests={
                            "cpu": policy.cpu_request,
                            "memory": policy.memory_request,
                        },
                        limits={
                            "cpu": policy.cpu_limit,
                            "memory": policy.memory_limit,
                        },
                    ),
                    security_context=client.V1SecurityContext(
                        run_as_non_root=True,
                        run_as_user=policy.run_as_user,
                        run_as_group=policy.run_as_group,
                        allow_privilege_escalation=False,
                        read_only_root_filesystem=True,
                        capabilities=client.V1Capabilities(drop=["ALL"]),
                    ),
                )
            ],
            volumes=[client.V1Volume(
                name="work",
                empty_dir=client.V1EmptyDirVolumeSource(
                    size_limit=policy.work_volume_size,
                ),
            )],
        ),
    )


def read_terminal_state(
    api: client.CoreV1Api,
    policy: VerifiedSandboxPolicy,
    invocation: VerifiedInvocation,
) -> tuple[str, int, str, str]:
    deadline = time.monotonic() + policy.max_runtime_seconds
    while time.monotonic() < deadline:
        pod = api.read_namespaced_pod(
            name=invocation.invocation_id,
            namespace=policy.namespace,
        )
        if pod.status.phase in {"Succeeded", "Failed"}:
            statuses = pod.status.container_statuses or []
            if len(statuses) != 1 or statuses[0].state.terminated is None:
                raise RuntimeError("terminal Pod lacks one terminated runner status")
            terminated = statuses[0].state.terminated
            message = json.loads(terminated.message or "{}")
            if set(message) != {"execution_receipt_ref"}:
                raise PermissionError("termination message has invalid receipt schema")
            return (
                pod.status.phase,
                int(terminated.exit_code),
                str(message["execution_receipt_ref"]),
                str(pod.metadata.uid),
            )
        time.sleep(policy.poll_interval_seconds)
    raise TimeoutError("single-use sandbox exceeded signed runtime policy")


def delete_and_verify_absent(
    api: client.CoreV1Api,
    policy: VerifiedSandboxPolicy,
    invocation_id: str,
    pod_uid: str,
) -> None:
    try:
        api.delete_namespaced_pod(
            name=invocation_id,
            namespace=policy.namespace,
            grace_period_seconds=policy.termination_grace_period_seconds,
            propagation_policy="Foreground",
            body=client.V1DeleteOptions(preconditions=client.V1Preconditions(uid=pod_uid)),
        )
    except ApiException as exc:
        if exc.status != 404:
            raise
    deadline = time.monotonic() + policy.teardown_timeout_seconds
    while time.monotonic() < deadline:
        try:
            api.read_namespaced_pod(
                name=invocation_id,
                namespace=policy.namespace,
            )
        except ApiException as exc:
            if exc.status == 404:
                return
            raise
        time.sleep(policy.poll_interval_seconds)
    raise TimeoutError("sandbox Pod deletion was not independently observed")


def run_tool_once(
    *,
    policy: VerifiedSandboxPolicy,
    invocation: VerifiedInvocation,
    receipt_verifier: ReceiptVerifier,
    evidence: EvidenceWriter,
) -> dict[str, object]:
    validate_inputs(policy, invocation)
    config.load_incluster_config()
    api = client.CoreV1Api()
    pod = build_sandbox_pod(policy, invocation)
    create_attempted = False
    pod_uid: str | None = None
    failure: Exception | None = None
    terminal: tuple[str, int, str, str] | None = None
    execution_receipt: dict[str, object] | None = None
    try:
        create_attempted = True
        created_pod = api.create_namespaced_pod(namespace=policy.namespace, body=pod)
        pod_uid = str(created_pod.metadata.uid or "")
        if not pod_uid:
            raise RuntimeError("Kubernetes create response omitted Pod UID")
        terminal = read_terminal_state(api, policy, invocation)
        if terminal[3] != pod_uid:
            raise PermissionError("terminal Pod UID changed during invocation")
        execution_receipt = receipt_verifier.verify_execution_receipt(
            terminal[2], invocation
        )
    except Exception as exc:
        failure = exc

    cleanup_failure: Exception | None = None
    absence_receipt: dict[str, object] | None = None
    if create_attempted:
        try:
            if not pod_uid:
                # A timeout may follow accepted creation. Resolve only our exact
                # invocation before deleting; a 404 cannot prove a timed-out create
                # will never materialize and therefore remains unresolved.
                observed = api.read_namespaced_pod(
                    name=invocation.invocation_id, namespace=policy.namespace,
                )
                labels = observed.metadata.labels or {}
                annotations = observed.metadata.annotations or {}
                if (labels.get("aidefend-invocation") != invocation.invocation_id
                    or annotations.get("aidefend.io/manifest-sha256") != invocation.manifest_sha256
                    or annotations.get("aidefend.io/policy-sha256") != policy.policy_sha256):
                    raise PermissionError("ambiguous create resolved to another invocation")
                pod_uid = str(observed.metadata.uid or "")
            if not pod_uid:
                raise RuntimeError("created Pod UID is unavailable for absence proof")
            delete_and_verify_absent(api, policy, invocation.invocation_id, pod_uid)
            absence_receipt = receipt_verifier.verify_absence(invocation, pod_uid)
        except Exception as exc:
            cleanup_failure = exc

    if failure is not None or cleanup_failure is not None:
        outcome = "FAIL" if isinstance(failure, PermissionError) else "ERROR"
        evidence.append_and_verify({
            "schema_version": "aidefend.single-use-sandbox.v1",
            "control": "AID-I-001.003",
            "invocation_id": invocation.invocation_id,
            "tenant_id": invocation.tenant_id,
            "task_id": invocation.task_id,
            "manifest_sha256": invocation.manifest_sha256,
            "policy_version": policy.policy_version,
            "policy_sha256": policy.policy_sha256,
            "error_stage": "teardown" if cleanup_failure else "execution",
            "pod_absent": absence_receipt is not None,
            "absence_receipt_sha256": (
                absence_receipt["receipt_sha256"]
                if absence_receipt is not None else None
            ),
            "outcome": outcome,
        })
        raise cleanup_failure or failure

    if terminal is None or execution_receipt is None or absence_receipt is None:
        raise RuntimeError("terminal, execution, or absence receipt is missing")
    final = evidence.append_and_verify({
        "schema_version": "aidefend.single-use-sandbox.v1",
        "control": "AID-I-001.003",
        "invocation_id": invocation.invocation_id,
        "tenant_id": invocation.tenant_id,
        "task_id": invocation.task_id,
        "manifest_sha256": invocation.manifest_sha256,
        "policy_sha256": policy.policy_sha256,
        "pod_uid": terminal[3],
        "terminal_phase": terminal[0],
        "tool_exit_code": terminal[1],
        "execution_receipt_sha256": execution_receipt["receipt_sha256"],
        "absence_receipt_sha256": absence_receipt["receipt_sha256"],
        "pod_absent": True,
        "outcome": "PASS",
    })
    return {"tool_exit_code": terminal[1], "evidence": final}

Runner and manifest contract

The digest-pinned runner authenticates to an invocation broker over mTLS, fetches a signature-verified manifest and payload by opaque invocation ID, rehashes payload bytes, executes exactly the named profile once, stores bounded output outside the Pod, and writes only one small receipt reference to the Kubernetes termination message. The Pod never receives arbitrary script bytes, credentials, or Kubernetes API access through environment variables. Runtime strength is delegated to the signed AID-I-001.002 profile and egress is delegated to AID-I-001.004; this step binds both sibling receipts while leaving their runtime-isolation and egress results separately verified.

An ambiguous create with no resolvable UID remains ERROR with pod_absent=false. The controller must retain the invocation for reconciliation until any late Pod is identified by the invocation and manifest, deleted with a UID precondition, and independently attested absent; the active runtime deadline alone is not teardown evidence. Never reuse an invocation ID or retry it as a second execution.

AID-I-001.003-G002

When to use the execution broker

Use this method when tool execution is delegated to E2B, Daytona, Modal Sandboxes, Cloudflare Sandboxes, microsandbox, or another managed sandbox API. The internal broker is the only principal allowed to call provider APIs. It accepts an opaque, immutable invocation-manifest reference; it must not accept caller-supplied shell commands, inline files, provider credentials, runtime selectors, or lifecycle flags. Provider choice, runtime profile, egress policy, credential handles, output bound, deadlines, and teardown requirements come from a signature-verified, monotonic policy. A self-hosted Kubernetes one-shot runtime uses the preceding guidance instead.

Strict broker client

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

import ipaddress
import json
import re
from dataclasses import dataclass
from typing import Callable
from urllib.parse import urlsplit

import requests


IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,255}$")
POLICY_VERSION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
SHA256 = re.compile(r"^[0-9a-f]{64}$")


@dataclass(frozen=True)
class VerifiedBrokerPolicy:
    policy_version: str
    policy_sha256: str
    response_policy_version: str
    response_policy_sha256: str
    broker_url: str
    client_cert_path: str
    client_key_path: str
    ca_bundle_path: str
    runtime_profile_ref: str
    egress_policy_ref: str
    request_timeout_seconds: int
    connect_timeout_seconds: int
    response_max_bytes: int
    response_chunk_bytes: int


@dataclass(frozen=True)
class VerifiedSandboxInvocation:
    request_id: str
    tenant_id: str
    task_id: str
    manifest_ref: str
    manifest_sha256: str


@dataclass(frozen=True)
class ReceiptVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_run_receipt: Callable[[str, VerifiedBrokerPolicy, VerifiedSandboxInvocation], dict[str, object]]


@dataclass(frozen=True)
class AbsenceVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_provider_absence: Callable[[str, str, list[str], VerifiedSandboxInvocation], dict[str, object]]


@dataclass(frozen=True)
class EvidenceWriter:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict[str, object]], dict[str, str]]


def require_sha256(value: str, field: str) -> None:
    if SHA256.fullmatch(value) is None or value == "0" * 64:
        raise ValueError(field + " must be a nonzero lowercase SHA-256 digest")


def canonical_https_origin(value: str) -> str:
    parsed = urlsplit(value)
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or parsed.path not in {"", "/"}
    ):
        raise ValueError("broker_url must be one credential-free HTTPS origin")
    try:
        port = parsed.port
    except ValueError as exc:
        raise ValueError("broker_url has an invalid port") from exc
    host_value = parsed.hostname.rstrip(".")
    try:
        host = ipaddress.ip_address(host_value).compressed.lower()
    except ValueError:
        host = host_value.encode("idna").decode("ascii").lower()
    if not host:
        raise ValueError("broker_url hostname is empty")
    display_host = "[" + host + "]" if ":" in host else host
    effective_port = port or 443
    return "https://" + display_host + (
        "" if effective_port == 443 else ":" + str(effective_port)
    )


def validate(
    policy: VerifiedBrokerPolicy,
    invocation: VerifiedSandboxInvocation,
) -> None:
    for value in (
        invocation.request_id,
        invocation.tenant_id,
        invocation.task_id,
        invocation.manifest_ref,
        policy.runtime_profile_ref,
        policy.egress_policy_ref,
    ):
        if not IDENTIFIER.fullmatch(value):
            raise ValueError("invalid identifier in sandbox request")
    for value in (policy.policy_version, policy.response_policy_version):
        if POLICY_VERSION.fullmatch(value) is None:
            raise ValueError("signed policy version is invalid")
    require_sha256(invocation.manifest_sha256, "manifest_sha256")
    require_sha256(policy.policy_sha256, "policy_sha256")
    require_sha256(policy.response_policy_sha256, "response_policy_sha256")
    canonical_https_origin(policy.broker_url)
    for field in (
        "request_timeout_seconds",
        "connect_timeout_seconds",
        "response_max_bytes",
        "response_chunk_bytes",
    ):
        value = getattr(policy, field)
        if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
            raise ValueError(
                "signed broker policy field must be a positive integer: " + field
            )
    if policy.response_chunk_bytes > policy.response_max_bytes:
        raise ValueError("response_chunk_bytes exceeds response_max_bytes")


def reject_duplicate_keys(
    pairs: list[tuple[str, object]],
) -> dict[str, object]:
    value: dict[str, object] = {}
    for key, item in pairs:
        if key in value:
            raise PermissionError(
                "broker response contains duplicate JSON member: " + key
            )
        value[key] = item
    return value


def reject_non_json_constant(value: str) -> None:
    raise PermissionError("broker response contains non-JSON constant: " + value)


def read_bounded_json_object(
    response: requests.Response,
    *,
    max_bytes: int,
    chunk_bytes: int,
) -> tuple[dict[str, object], int]:
    content_encoding = response.headers.get("Content-Encoding")
    if content_encoding is not None and content_encoding.strip().lower() != "identity":
        raise PermissionError("broker response Content-Encoding is not identity")
    media_type = response.headers.get("Content-Type", "").split(";", 1)[0]
    if media_type.strip().lower() != "application/json":
        raise PermissionError("broker response Content-Type is not application/json")

    declared = response.headers.get("Content-Length")
    declared_bytes: int | None = None
    if declared is not None:
        if not declared.isascii() or not declared.isdecimal():
            raise PermissionError("broker Content-Length is invalid")
        declared_bytes = int(declared)
        if declared_bytes <= 0 or declared_bytes > max_bytes:
            raise PermissionError(
                "broker Content-Length violates signed response policy"
            )

    payload = bytearray()
    read_size = min(chunk_bytes, max_bytes + 1)
    for chunk in response.iter_content(chunk_size=read_size):
        if not chunk:
            continue
        remaining = max_bytes + 1 - len(payload)
        payload.extend(chunk[:remaining])
        if len(chunk) > remaining or len(payload) > max_bytes:
            raise PermissionError(
                "broker response exceeds signed response_max_bytes"
            )
    if not payload:
        raise PermissionError("broker response is empty")
    if declared_bytes is not None and declared_bytes != len(payload):
        raise PermissionError(
            "broker Content-Length differs from observed response bytes"
        )

    try:
        parsed = json.loads(
            payload.decode("utf-8"),
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_non_json_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise PermissionError(
            "broker response is not strict UTF-8 JSON"
        ) from exc
    if not isinstance(parsed, dict):
        raise PermissionError("broker response must be a JSON object")
    return parsed, len(payload)


def run_once(
    *,
    policy: VerifiedBrokerPolicy,
    invocation: VerifiedSandboxInvocation,
    receipts: ReceiptVerifier,
    absence: AbsenceVerifier,
    evidence: EvidenceWriter,
) -> dict[str, object]:
    validate(policy, invocation)
    broker_origin = canonical_https_origin(policy.broker_url)
    endpoint = broker_origin + "/v1/sandbox-invocations"
    with requests.Session() as session:
        session.trust_env = False
        session.cert = (policy.client_cert_path, policy.client_key_path)
        session.verify = policy.ca_bundle_path
        with session.post(
            endpoint,
            json={
                "schema_version": "aidefend.managed-sandbox-request.v1",
                "request_id": invocation.request_id,
                "tenant_id": invocation.tenant_id,
                "task_id": invocation.task_id,
                "manifest_ref": invocation.manifest_ref,
                "manifest_sha256": invocation.manifest_sha256,
                "policy_version": policy.policy_version,
                "policy_sha256": policy.policy_sha256,
                "response_policy_version": policy.response_policy_version,
                "response_policy_sha256": policy.response_policy_sha256,
                "response_max_bytes": policy.response_max_bytes,
                "response_chunk_bytes": policy.response_chunk_bytes,
            },
            headers={
                "Accept": "application/json",
                "Accept-Encoding": "identity",
            },
            timeout=(
                policy.connect_timeout_seconds,
                policy.request_timeout_seconds,
            ),
            allow_redirects=False,
            stream=True,
        ) as response:
            if response.history or response.url != endpoint:
                raise PermissionError("broker response redirected or changed origin")
            response.raise_for_status()
            body, response_bytes = read_bounded_json_object(
                response,
                max_bytes=policy.response_max_bytes,
                chunk_bytes=policy.response_chunk_bytes,
            )
    if (
        set(body) != {"receipt_jws"}
        or not isinstance(body["receipt_jws"], str)
        or not body["receipt_jws"]
    ):
        raise PermissionError("broker response is not exactly one signed receipt")

    receipt = receipts.verify_run_receipt(
        body["receipt_jws"], policy=policy, invocation=invocation
    )
    required = {
        "receipt_sha256", "sandbox_id", "workspace_id", "provider_backend",
        "runtime_profile_ref", "egress_policy_ref", "manifest_sha256",
        "policy_version", "policy_sha256", "response_policy_version",
        "response_policy_sha256", "response_max_bytes", "response_chunk_bytes",
        "credential_lease_ids", "terminal_state", "tool_exit_code",
        "output_receipt_ref", "destroyed_at",
    }
    if set(receipt) != required:
        raise PermissionError("verified run receipt has an unexpected schema")
    expected_policy = {
        "policy_version": policy.policy_version,
        "policy_sha256": policy.policy_sha256,
        "response_policy_version": policy.response_policy_version,
        "response_policy_sha256": policy.response_policy_sha256,
        "response_max_bytes": policy.response_max_bytes,
        "response_chunk_bytes": policy.response_chunk_bytes,
        "runtime_profile_ref": policy.runtime_profile_ref,
        "egress_policy_ref": policy.egress_policy_ref,
        "manifest_sha256": invocation.manifest_sha256,
    }
    if any(receipt.get(key) != value for key, value in expected_policy.items()):
        raise PermissionError(
            "verified run receipt has a policy or invocation mismatch"
        )
    require_sha256(str(receipt["receipt_sha256"]), "receipt_sha256")
    if receipt["terminal_state"] not in {"SUCCEEDED", "FAILED", "TIMED_OUT"}:
        raise PermissionError("invalid terminal state")
    if (
        isinstance(receipt["tool_exit_code"], bool)
        or not isinstance(receipt["tool_exit_code"], int)
    ):
        raise PermissionError("tool_exit_code is invalid")
    lease_ids = receipt["credential_lease_ids"]
    if (
        not isinstance(lease_ids, list)
        or len(set(lease_ids)) != len(lease_ids)
        or any(not isinstance(item, str) or not item for item in lease_ids)
    ):
        raise PermissionError("invalid credential-lease population")

    absence_receipt = absence.verify_provider_absence(
        sandbox_id=str(receipt["sandbox_id"]),
        workspace_id=str(receipt["workspace_id"]),
        credential_lease_ids=lease_ids,
        invocation=invocation,
    )
    if absence_receipt.get("outcome") != "PASS":
        raise PermissionError("provider resource or credential lease remains")
    require_sha256(
        str(absence_receipt.get("receipt_sha256", "")),
        "absence receipt_sha256",
    )

    final = evidence.append_and_verify({
        "schema_version": "aidefend.managed-sandbox-evidence.v1",
        "control": "AID-I-001.003",
        "request_id": invocation.request_id,
        "tenant_id": invocation.tenant_id,
        "task_id": invocation.task_id,
        "manifest_sha256": invocation.manifest_sha256,
        "broker_origin": broker_origin,
        "policy_version": policy.policy_version,
        "policy_sha256": policy.policy_sha256,
        "response_policy_version": policy.response_policy_version,
        "response_policy_sha256": policy.response_policy_sha256,
        "response_max_bytes": policy.response_max_bytes,
        "response_chunk_bytes": policy.response_chunk_bytes,
        "broker_response_bytes": response_bytes,
        "provider_backend": receipt["provider_backend"],
        "sandbox_id": receipt["sandbox_id"],
        "runtime_profile_ref": receipt["runtime_profile_ref"],
        "egress_policy_ref": receipt["egress_policy_ref"],
        "run_receipt_sha256": receipt["receipt_sha256"],
        "absence_receipt_sha256": absence_receipt["receipt_sha256"],
        "tool_exit_code": receipt["tool_exit_code"],
        "outcome": "PASS",
    })
    return {
        "tool_exit_code": receipt["tool_exit_code"],
        "output_receipt_ref": receipt["output_receipt_ref"],
        "policy_sha256": policy.policy_sha256,
        "response_policy_sha256": policy.response_policy_sha256,
        "response_max_bytes": policy.response_max_bytes,
        "response_chunk_bytes": policy.response_chunk_bytes,
        "broker_response_bytes": response_bytes,
        "evidence": final,
    }

Broker prerequisites and teardown

The broker authenticates the workload over mTLS, verifies the signed manifest and policy before provider selection, creates a provider workspace with a unique generation ID, injects only short-lived credential handles, executes one manifest-defined profile, bounds output, destroys the sandbox on success, failure, timeout, cancellation, client disconnect, and evidence failure, then issues a signed receipt. A separately credentialed read-only verifier inventories the provider control plane and credential broker after teardown. Runtime isolation is independently evaluated by AID-I-001.002 and outbound enforcement by AID-I-001.004; their receipts are bound here as prerequisites, while runtime-isolation and egress results remain separately verified.

Open-source tools

• gVisor
• Kata Containers
• Firecracker
• E2B
• Wasmtime (per-store WebAssembly resource isolation for single-use tool execution)
• Cloudflare Sandbox SDK (Apache-2.0 SDK; requires the managed Cloudflare Workers and Containers runtime)
• microsandbox
• Daytona

Commercial / hosted tools

• AWS Fargate
• Google Kubernetes Engine Sandbox
• Modal Sandboxes
• E2B Enterprise

Threat mappings

MITRE ATLAS

• AML.T0011 User Execution
• AML.T0011.002 User Execution: Poisoned AI Agent Tool
• AML.T0050 Command and Scripting Interpreter
• AML.T0072 Cyber Communication Channel (guaranteed teardown ends any channel a compromised tool opened once the invocation completes)
• AML.T0053 AI Agent Tool Invocation

MAESTRO

• Agent Tool Misuse (L7)
• Resource Hijacking (L4) (ephemeral teardown prevents sustained hijacking)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation
• ASI05:2026 Unexpected Code Execution (RCE)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.3 Unsafe System / Browser / File Execution (single-use tool workspaces limit persistence after coerced execution)
• AISubtech-9.1.1 Code Execution (single-use tool workspaces limit persistence after coerced execution)
• AITech-12.1 Tool Exploitation (single-use tool workspaces limit persistence after coerced execution)
• AITech-9.1 Model or Agentic System Manipulation (single-use tool workspaces limit persistence after coerced execution)

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (ephemeral teardown destroys compromised component state)
• RA: Rogue Actions (ephemeral execution prevents persistent rogue behavior)

Databricks AI Security Framework 3.0

• Agents - Core 13.2: Tool Misuse (ephemeral teardown prevents tool misuse persistence)
• Agents - Core 13.11: Unexpected RCE and Code Attacks