Scope boundary

Owns disablement of the exact compromised AI-agent or workload principal at every authoritative identity, workload, federation, and authorization plane and denial of every fresh credential or token issuance path. A separately attested replacement principal is a recovery object, not proof that the original principal was disabled.

AID-E-001.001: AID-E-001.003 disables the principal and stops new issuance; AID-E-001.001 revokes or rotates individual root and long-lived credential objects. Principal disablement does not independently rotate every exposed password, key, secret, signing key, or certificate.

AID-E-001.002: AID-E-001.003 stops the principal from obtaining new authentication artifacts; AID-E-001.002 revokes tokens, sessions, certificates, and leases that were already issued. An issuer-side disablement does not prove that every relying party rejects previously issued artifacts.

AID-E-001.004: AID-E-001.003 disables an AI-agent or workload principal and its issuance paths; AID-E-001.004 removes delegated grants and connected-app authorizations. Delegated authorization can remain independently effective or associated with another principal, so its authoritative grant population requires separate eviction.

Implementation guidance

• AID-E-001.003-G001: Disable every exact SPIFFE/SPIRE workload registration entry for the compromised principal and prove fresh SVID issuance is denied.
• AID-E-001.003-G002: Use the prebuilt token-exchange deny-bundle control to block compromised agent actors and verify fresh downstream token minting fails.
• AID-E-001.003-G003: Disable the compromised cloud workload-identity binding and prove fresh credential minting is denied.
• AID-E-001.003-G004: Disable and, when authorized, remove exact attacker-created AI-agent or workload principal objects.
• AID-E-001.003-G005: Reconcile every applicable identity-plane disablement target against authoritative principal, issuance, federation, and authorization state.

AID-E-001.003-G001

Concept and owner boundary

In SPIFFE/SPIRE, a registration entry is an authoritative issuance object that allows a workload principal to obtain an SVID. Multiple entries can bind the same SPIFFE ID, so disabling only the first printed entry does not stop issuance. This method enumerates and deletes or disables every exact registration entry for the signed principal and proves that a fresh SVID cannot be minted. Already-issued SVIDs and established mTLS sessions remain AID-E-001.002 objects; their residual lifetime or teardown evidence is not a PASS for this method.

Signed target

The identity-response authority signs the exact canonical manifest below. The action script verifies that Cosign bundle under an image-pinned key before deriving either the SPIFFE ID or evidence path.

{
  "control": "AID-E-001.003",
  "incident_id": "INC-2026-0718-0042",
  "schema_version": "aidefend.e001003-spire-entry-eviction/v1",
  "spiffe_id": "spiffe://example.org/agent/compromised-agent"
}

Step 1: Enumerate and delete all entries for the exact SPIFFE ID

# File: eviction/delete_spire_entries.sh
#!/usr/bin/env bash
set -euo pipefail

if [[ $# -ne 2 ]]; then
  echo "usage: delete_spire_entries.sh TARGET.json TARGET.json.sigstore.json" >&2
  exit 2
fi
SOURCE_MANIFEST=$1
SOURCE_BUNDLE=$2
TRUST_KEY="/opt/aidefend/trust/e001003-spire-authority.pub"
: "${E001003_COMMAND_TIMEOUT_SECONDS:?required by the versioned runtime profile}"
if [[ ! "$E001003_COMMAND_TIMEOUT_SECONDS" =~ ^[0-9]+([.][0-9]+)?$ ]]; then
  echo "command timeout must be numeric" >&2
  exit 2
fi
awk -v value="$E001003_COMMAND_TIMEOUT_SECONDS"   'BEGIN { exit !(value > 0) }'
SNAPSHOT_ROOT="$(mktemp -d)"
chmod 700 "$SNAPSHOT_ROOT"
trap 'rm -rf -- "$SNAPSHOT_ROOT"' EXIT
MANIFEST="$SNAPSHOT_ROOT/target.json"
BUNDLE="$SNAPSHOT_ROOT/target.json.sigstore.json"
CANONICAL_FILE="$SNAPSHOT_ROOT/canonical.json"
ENTRY_IDS_FILE="$SNAPSHOT_ROOT/entry-ids.txt"
REMAINING_IDS_FILE="$SNAPSHOT_ROOT/remaining-ids.txt"
install -m 0400 -- "$SOURCE_MANIFEST" "$MANIFEST"
install -m 0400 -- "$SOURCE_BUNDLE" "$BUNDLE"

timeout "$E001003_COMMAND_TIMEOUT_SECONDS"   cosign verify-blob --key "$TRUST_KEY"   --bundle "$BUNDLE" "$MANIFEST"
jq -S -c . "$MANIFEST" > "$CANONICAL_FILE"
if ! cmp -s -- "$MANIFEST" "$CANONICAL_FILE"; then
  echo "target manifest is not canonical JSON" >&2
  exit 2
fi
jq -e '
  type == "object"
  and (keys == ["control","incident_id","schema_version","spiffe_id"])
  and .schema_version == "aidefend.e001003-spire-entry-eviction/v1"
  and .control == "AID-E-001.003"
  and (.incident_id | type == "string" and length > 0)
  and (.spiffe_id | type == "string" and length > 0)
' "$MANIFEST" >/dev/null
SPIFFE_ID=$(jq -er '.spiffe_id' "$MANIFEST")
INCIDENT_ID=$(jq -er '.incident_id' "$MANIFEST")
if [[ ! "$SPIFFE_ID" =~ ^spiffe://[A-Za-z0-9.-]+/[A-Za-z0-9._~:/@%-]+$ ]]; then
  echo "SPIFFE ID is not a bounded absolute URI" >&2
  exit 2
fi
if [[ ! "$INCIDENT_ID" =~ ^INC-[A-Z0-9][A-Z0-9._-]{2,63}$ ]]; then
  echo "incident ID is not bounded" >&2
  exit 2
fi
EVIDENCE_DIR="artifacts/$INCIDENT_ID"
mkdir -p -- "$EVIDENCE_DIR"
cp -- "$MANIFEST" "$EVIDENCE_DIR/spire-target.json"
cp -- "$BUNDLE" "$EVIDENCE_DIR/spire-target.json.sigstore.json"

# SPIRE supports machine-readable JSON for entry show. Accept only an array
# or the entries wrapper, then select exact IDs for this SPIFFE ID.
extract_entry_ids() {
  jq -r --arg spiffe_id "$SPIFFE_ID" '
    (if type == "array" then .
     elif type == "object" and (.entries | type) == "array" then .entries
     else error("unexpected SPIRE entry-show JSON schema") end)
    | map(select((.spiffe_id // .spiffeId) == $spiffe_id) | .id)
    | if any(.[]; type != "string" or length == 0)
      then error("SPIRE entry ID is missing or invalid") else .[] end
  '
}

BEFORE=$(timeout "$E001003_COMMAND_TIMEOUT_SECONDS" \
  spire-server entry show -spiffeID "$SPIFFE_ID" -output json)
printf '%s\n' "$BEFORE" > "$EVIDENCE_DIR/spire-entries-before.txt"
printf '%s\n' "$BEFORE" | extract_entry_ids > "$ENTRY_IDS_FILE"

if ! test -s "$ENTRY_IDS_FILE"; then
  echo "No registration entries found for exact SPIFFE ID: $SPIFFE_ID" >&2
  exit 1
fi

while IFS= read -r ENTRY_ID; do
  case "$ENTRY_ID" in
    *[!A-Za-z0-9._:-]*)
      echo "Refusing unexpected entry ID syntax: $ENTRY_ID" >&2
      exit 1
      ;;
  esac
  timeout "$E001003_COMMAND_TIMEOUT_SECONDS" \
    spire-server entry delete -entryID "$ENTRY_ID" |
    tee -a "$EVIDENCE_DIR/spire-entry-delete-results.txt"
done < "$ENTRY_IDS_FILE"

AFTER=$(timeout "$E001003_COMMAND_TIMEOUT_SECONDS" \
  spire-server entry show -spiffeID "$SPIFFE_ID" -output json)
printf '%s\n' "$AFTER" > "$EVIDENCE_DIR/spire-entries-after.txt"
printf '%s\n' "$AFTER" | extract_entry_ids > "$REMAINING_IDS_FILE"
if test -s "$REMAINING_IDS_FILE"; then
  echo "Eviction incomplete; matching registration entries remain" >&2
  cat "$REMAINING_IDS_FILE" >&2
  exit 1
fi

cp "$ENTRY_IDS_FILE" "$EVIDENCE_DIR/deleted-entry-ids.txt"

Reference: SPIRE documents entry show -spiffeID, confirms that more than one entry may share a SPIFFE ID, and requires deletion by individual entry ID at https://spiffe.io/docs/latest/deploying/registering/.

Step 2: Prove principal registration and issuance disablement

Preserve the verified signed target, before/after entry listings, and every changed entry ID, then re-enumerate through a separately credentialed SPIRE control-plane identity and require zero matching registration entries across the complete declared trust-domain population. From the exact compromised workload selectors, request a fresh SVID and require issuance denial. Publish the principal to separately owned authorization or token-exchange deny paths where applicable, but route old SVID presentation and connection teardown to AID-E-001.002. Restore registration only through reviewed re-enrollment under a separately attested replacement principal.

Action: Delete or disable every exact registration entry for the compromised SPIFFE principal and accept AID-E-001.003 only after independent enumeration proves zero issuer entries and a fresh issuance attempt is denied. Do not infer issued-SVID revocation or mTLS session termination from registration deletion.

AID-E-001.003-G002

Runtime policy: Load deny-bundle byte/entry/text/list ceilings, freshness, lifetime, and verifier deadline from the signed, versioned broker policy bound to this trust domain and revision, bind its version/digest to admission and negative-probe evidence, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Build-ahead prerequisite:

This is a prebuilt control-plane capability, not an incident-time software project. Before an incident, the token broker must already enforce Cosign-bundled deny payloads, pinned verification keys, durable high-water revision storage, fail-closed bundle retrieval, and negative minting tests. During an incident, operators only prepare the scoped bundle, sign and publish it through the approved channel, wait for broker high-water acceptance, and run the denial probes.

Incident runbook:

1. Freeze the compromised actor IDs, credential IDs, issuer, audience, trust domain, incident ID, and release ID from evidence.
2. Publish one short-lived deny bundle signed by the identity-deny authority.
3. Confirm every broker replica reports the same accepted revision and digest.
4. Attempt token exchange from a separately credentialed probe and require denial for the old actor plus success for the clean replacement.

Required broker capability:

If the broker does not already implement this contract, record INSUFFICIENT_DATA for immediate delegated-issuance eviction and fall back to provider-native grant or workload-identity revocation paths while the missing control is built for future incidents.

Concept:

If an agent can authenticate to a token broker, revoking only its current access token is not enough: it may exchange again. Every broker replica must consume the same centrally published, detached-signed deny bundle and enforce it before any token-exchange or on-behalf-of call. The bundle is bound to one system or tenant, issuer, audience, and trust domain; it has a short validity window and a monotonically increasing revision. The broker durably records the highest accepted revision and fails closed when retrieval, Cosign-bundle verification, parsing, freshness, scope validation, or high-water storage fails.

Step 1: Publish a scope-bound deny bundle from the incident-response authority

{
  "schema_version": "aidefend.identity-deny-bundle/v1",
  "system_or_tenant": "tenant:customer-support-prod",
  "incident_id": "INC-2026-0417",
  "release_id": "support-agent-prod-2026-07-10.1",
  "revision": 42,
  "issued_at": "2026-04-17T14:22:10Z",
  "expires_at": "2026-04-17T14:27:10Z",
  "issuer": "https://id.company.com/",
  "audience": "urn:company:ai-token-broker",
  "trust_domain": "spiffe://company.com",
  "revoked_identity_ids": ["spiffe://company.com/agent/support/prod-001"],
  "revoked_credential_ids": ["jwt-jti:9d61b3c2", "svid-serial:70a39f2e"]
}

set -euo pipefail
jq -S . identity-deny-bundle.json > identity-deny-bundle.canonical.json
mv identity-deny-bundle.canonical.json identity-deny-bundle.json
cosign sign-blob --yes \
  --key "awskms:///alias/aidefend-identity-deny-authority" \
  --bundle identity-deny-bundle.json.sigstore.json identity-deny-bundle.json
cosign verify-blob --key keys/identity-deny-authority.pub \
  --bundle identity-deny-bundle.json.sigstore.json identity-deny-bundle.json

Step 2: Verify and consume the bundle before delegated issuance

# File: token_broker/actor_revocation_guard.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import subprocess
import tempfile
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path


DENY_PUBLIC_KEY = Path("/opt/token-broker/keys/identity-deny-authority.pub")
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
MAX_PAYLOAD_BYTES = int(os.environ["IDENTITY_DENY_MAX_PAYLOAD_BYTES"])
MAX_COSIGN_BUNDLE_BYTES = int(os.environ["IDENTITY_DENY_MAX_COSIGN_BUNDLE_BYTES"])
MAXIMUM_IDENTITY_COUNT = int(os.environ["IDENTITY_DENY_MAXIMUM_IDENTITY_COUNT"])
COMMAND_TIMEOUT_SECONDS = float(os.environ["IDENTITY_DENY_COMMAND_TIMEOUT_SECONDS"])
MAXIMUM_FUTURE_SKEW_SECONDS = float(os.environ["IDENTITY_DENY_MAXIMUM_FUTURE_SKEW_SECONDS"])
MAXIMUM_BUNDLE_LIFETIME_SECONDS = float(os.environ["IDENTITY_DENY_MAXIMUM_BUNDLE_LIFETIME_SECONDS"])
if (not RUNTIME_PROFILE_VERSION or len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
        or RUNTIME_PROFILE_SHA256 == "0" * 64
        or MAX_PAYLOAD_BYTES < 1 or MAX_COSIGN_BUNDLE_BYTES < 1 or MAXIMUM_IDENTITY_COUNT < 1
        or any(not math.isfinite(value) or value <= 0 for value in
               (COMMAND_TIMEOUT_SECONDS, MAXIMUM_BUNDLE_LIFETIME_SECONDS))
        or not math.isfinite(MAXIMUM_FUTURE_SKEW_SECONDS)
        or MAXIMUM_FUTURE_SKEW_SECONDS < 0):
    raise RuntimeError("versioned identity-deny runtime profile is invalid")
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9:/.@_-]{0,511}$")
INCIDENT_RE = re.compile(r"^INC-[A-Z0-9][A-Z0-9._-]{2,63}$")
BUNDLE_FIELDS = {
    "schema_version", "system_or_tenant", "incident_id", "release_id", "revision",
    "issued_at", "expires_at", "issuer", "audience", "trust_domain",
    "revoked_identity_ids", "revoked_credential_ids",
}
HIGH_WATER_LUA = """
local revision_key = KEYS[1]
local digest_key = KEYS[2]
local proposed_revision = tonumber(ARGV[1])
local proposed_digest = ARGV[2]
local current_revision = tonumber(redis.call('GET', revision_key) or '-1')
local current_digest = redis.call('GET', digest_key)
if proposed_revision < current_revision then return {-1, current_revision} end
if proposed_revision == current_revision then
  if current_digest ~= proposed_digest then return {-2, current_revision} end
  return {0, current_revision}
end
redis.call('SET', revision_key, proposed_revision)
redis.call('SET', digest_key, proposed_digest)
return {1, proposed_revision}
"""


class ActorRevoked(Exception):
    pass


class IdentityDenyUnavailable(Exception):
    pass


@dataclass(frozen=True)
class VerifiedActorIdentity:
    # Authentication middleware constructs this only after signature, issuer,
    # audience, expiry, subject, and credential binding have passed.
    actor_id: str
    credential_id: str
    issuer: str
    audience: str
    trust_domain: str
    system_or_tenant: str
    incident_id: str
    release_id: str


def parse_time(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("timestamp must be a string")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise ValueError("timestamp must include an offset")
    return parsed.astimezone(timezone.utc)


def reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate deny-payload JSON key: {key}")
        value[key] = item
    return value


def reject_nonfinite(value):
    raise ValueError(f"non-finite deny-payload JSON value: {value}")


def strict_json(raw: bytes) -> dict:
    try:
        value = json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_nonfinite,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("deny payload is not strict UTF-8 JSON") from exc
    if not isinstance(value, dict):
        raise ValueError("deny payload must be one JSON object")
    return value


def verify_cosign_bundle(payload: bytes, cosign_bundle: bytes) -> None:
    if not payload or len(payload) > MAX_PAYLOAD_BYTES:
        raise ValueError("deny payload size is invalid")
    if not cosign_bundle or len(cosign_bundle) > MAX_COSIGN_BUNDLE_BYTES:
        raise ValueError("Cosign bundle size is invalid")
    with tempfile.TemporaryDirectory(prefix="aidefend-deny-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_path = root / "identity-deny-bundle.json"
        bundle_path = root / "identity-deny-bundle.json.sigstore.json"
        payload_path.write_bytes(payload)
        bundle_path.write_bytes(cosign_bundle)
        os.chmod(payload_path, 0o400)
        os.chmod(bundle_path, 0o400)
        subprocess.run([
            "cosign", "verify-blob", "--key", str(DENY_PUBLIC_KEY),
            "--bundle", str(bundle_path), str(payload_path),
        ], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT_SECONDS)
        verified_payload = payload_path.read_bytes()
        verified_bundle = bundle_path.read_bytes()
        if verified_payload != payload or verified_bundle != cosign_bundle:
            raise OSError("verified deny payload or Cosign bundle snapshot changed")


def require_id_list(value: object, field: str) -> list[str]:
    if (
        not isinstance(value, list) or not value or len(value) > MAXIMUM_IDENTITY_COUNT
        or any(not isinstance(item, str) or ID_RE.fullmatch(item) is None for item in value)
        or len(value) != len(set(value)) or value != sorted(value)
    ):
        raise ValueError(f"{field} must be a nonempty sorted unique ID list")
    return value


def verify_bundle(
    payload: bytes,
    cosign_bundle: bytes,
    authenticated_actor: VerifiedActorIdentity,
    high_water_store,
) -> dict:
    verify_cosign_bundle(payload, cosign_bundle)
    bundle = strict_json(payload)
    if not isinstance(bundle, dict) or set(bundle) != BUNDLE_FIELDS:
        raise ValueError("deny bundle schema differs")
    if bundle["schema_version"] != "aidefend.identity-deny-bundle/v1":
        raise ValueError("unsupported deny bundle schema")
    if INCIDENT_RE.fullmatch(bundle["incident_id"]) is None:
        raise ValueError("invalid incident ID")
    expected_scope = {
        "system_or_tenant": authenticated_actor.system_or_tenant,
        "incident_id": authenticated_actor.incident_id,
        "release_id": authenticated_actor.release_id,
        "issuer": authenticated_actor.issuer,
        "audience": authenticated_actor.audience,
        "trust_domain": authenticated_actor.trust_domain,
    }
    if any(bundle.get(field) != expected for field, expected in expected_scope.items()):
        raise ValueError("deny bundle is not bound to this broker scope")
    revision = bundle["revision"]
    if isinstance(revision, bool) or not isinstance(revision, int) or not 0 <= revision <= 2**63 - 1:
        raise ValueError("deny bundle revision is invalid")
    issued_at = parse_time(bundle["issued_at"])
    expires_at = parse_time(bundle["expires_at"])
    now = datetime.now(timezone.utc)
    if expires_at <= issued_at:
        raise ValueError("deny bundle lifetime is not positive")
    if issued_at > now + timedelta(seconds=MAXIMUM_FUTURE_SKEW_SECONDS) or expires_at <= now:
        raise ValueError("deny bundle is not currently valid")
    if expires_at - issued_at > timedelta(seconds=MAXIMUM_BUNDLE_LIFETIME_SECONDS):
        raise ValueError("deny bundle lifetime exceeds the signed policy maximum")
    require_id_list(bundle["revoked_identity_ids"], "revoked_identity_ids")
    require_id_list(bundle["revoked_credential_ids"], "revoked_credential_ids")

    scope_digest = hashlib.sha256(
        json.dumps(expected_scope, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()
    bundle_digest = hashlib.sha256(payload).hexdigest()
    result = high_water_store.eval(
        HIGH_WATER_LUA,
        2,
        f"identity-deny:{scope_digest}:revision",
        f"identity-deny:{scope_digest}:digest",
        revision,
        bundle_digest,
    )
    if not isinstance(result, (list, tuple)) or not result or int(result[0]) < 0:
        raise ValueError("deny bundle rollback or equivocation detected")
    return bundle


def issue_downstream_token(
    exchange_request: dict,
    authenticated_actor: VerifiedActorIdentity,
    central_deny_client,
    durable_high_water_store,
    token_exchange_client,
) -> dict:
    # The central client is configured with a pinned URL, server identity, mTLS
    # client identity, response-size cap, and redirects disabled.
    try:
        payload, cosign_bundle = central_deny_client.fetch_signed_current_bundle(
            authenticated_actor.system_or_tenant
        )
        bundle = verify_bundle(
            payload, cosign_bundle, authenticated_actor, durable_high_water_store
        )
    except Exception as exc:
        # Do not mint a token from stale local state when the central deny path,
        # Cosign bundle verifier, parser, or durable monotonic store is unavailable.
        raise IdentityDenyUnavailable("identity deny decision unavailable") from exc

    claimed_actor_id = exchange_request.get("actor_id")
    if claimed_actor_id is not None and claimed_actor_id != authenticated_actor.actor_id:
        raise IdentityDenyUnavailable("request actor differs from authenticated actor")
    if (
        authenticated_actor.actor_id in bundle["revoked_identity_ids"]
        or authenticated_actor.credential_id in bundle["revoked_credential_ids"]
    ):
        raise ActorRevoked("revoked identity or credential cannot mint a downstream token")

    canonical_request = dict(exchange_request)
    canonical_request["actor_id"] = authenticated_actor.actor_id
    canonical_request["upstream_credential_id"] = authenticated_actor.credential_id
    canonical_request["identity_deny_revision"] = bundle["revision"]
    canonical_request["identity_deny_bundle_sha256"] = hashlib.sha256(payload).hexdigest()
    return token_exchange_client.exchange_if_deny_generation_current(
        canonical_request,
        expected_revision=bundle["revision"],
        expected_bundle_sha256=hashlib.sha256(payload).hexdigest(),
    )

Action: Make the incident-response authority the only writer of the centrally hosted bundle. Require canonical JSON and a Cosign bundle produced with the pinned KMS key, a pinned mTLS retrieval channel with redirects disabled, exact scope and freshness validation, and a durable Redis/Valkey high-water record shared by broker replicas. An unavailable or invalid deny decision must stop issuance; do not fall back to a cached allow decision. The token service's exchange_if_deny_generation_current operation must compare the supplied revision and bundle digest with its current deny generation in the same authorization transaction that mints the token; a generation change returns a retryable denial rather than issuing from the stale pre-check. Run negative tests for both the revoked actor ID and every revoked credential ID, and verify that an independently attested replacement identity is allowed only after it is absent from the current signed deny set.

AID-E-001.003-G003

Production implementation

A prebuilt provider adapter re-reads the exact workload or service-principal record, subject selectors, federation provider, role association, and policy revision, rejects any UID, subject, issuer, audience, or resource-version mismatch, then removes or explicitly denies only the signed principal-to-issuer binding through a provider conditional API. It writes a durable tombstone or deny record where supported and reconciles infrastructure-as-code through the approved emergency path. Individual keys or long-lived certificates are handed to AID-E-001.001; already-issued tokens, certificates, and sessions are handed to AID-E-001.002.

Independent verification

A separate read-only provider identity enumerates the complete declared principal and federation population, exports effective trust policies and bindings, and requires the exact compromised edge absent or denied. From a pod or workload bound to the compromised immutable service-account or workload UID, an isolated probe attempts fresh credential minting and must be denied; a separately attested replacement principal must mint only the intended least-privilege credential. A deleted local annotation without authoritative provider readback is insufficient.

AID-E-001.003-G004

Act on principal objects, not their child credentials

A provider-native adapter selected from trusted configuration re-enumerates each exact AI-agent, workload, service-principal, or provider-native agent identity, compares immutable principal ID, subject, owner, tenant, privilege, issuer bindings, and resource version, then disables the principal and every path that can issue new credentials to it. After the signed forensic hold, delete only principal objects explicitly marked for removal and retain a provider-native deny or tombstone where supported. Attached API keys, client secrets, signing keys, and long-lived certificates remain AID-E-001.001; issued tokens, certificates, sessions, and leases remain AID-E-001.002; delegated grants remain AID-E-001.004.

Independent verification

A separately credentialed read-only adapter enumerates the complete affected principal and issuer population, requires every signed principal disabled or absent as authorized, and proves fresh authentication, federation, token exchange, and credential issuance attempts fail at every applicable boundary. It confirms non-target principals and bindings are unchanged and a separately attested replacement principal succeeds only with the intended least privilege. The action set must exactly equal the signed manifest set.

AID-E-001.003-G005

When identity-plane reconciliation applies

Use this path when the compromised AI agent or workload principal is represented at more than one authoritative identity, workload-registration, federation, token-issuance, or authorization boundary. Build and sign the canonical expected target set from the incident-bound principal IDs and each provider's authoritative inventory. Already-issued tokens, certificates, leases, and sessions remain under AID-E-001.002; delegated grants remain under AID-E-001.004.

Reconcile the exact target population

For every expected boundary, record a stable target ID, target class, authority ID, incident ID, and the disable or deny revision applied by G001-G004. A separately credentialed read-only adapter must query that same authority and report whether the exact principal or selector is disabled or absent and whether a fresh issuance or authentication attempt is denied. A replacement-principal positive test is required only when the recovery plan creates one. Missing boundaries, duplicate aliases, stale revisions, unsupported readback, or an adapter error leave the identity-plane reconciliation incomplete.

Executable population check

# File: eviction/reconcile_identity_plane.py
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
import sys
from pathlib import Path

EXPECTED_TRUST_KEY = Path("/opt/aidefend/trust/e001003-identity-scope.pub")
OBSERVED_TRUST_KEY = Path("/opt/aidefend/trust/e001003-identity-verifier.pub")
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(os.environ["E001003_COMMAND_TIMEOUT_SECONDS"])
if (
    not RUNTIME_PROFILE_VERSION
    or len(RUNTIME_PROFILE_SHA256) != 64
    or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
    or RUNTIME_PROFILE_SHA256 == "0" * 64
    or not math.isfinite(COMMAND_TIMEOUT_SECONDS)
    or COMMAND_TIMEOUT_SECONDS <= 0
):
    raise RuntimeError("versioned E-001.003 runtime profile is invalid")


def reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate signed JSON key: {key}")
        value[key] = item
    return value


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


def strict_json(raw: bytes, label: str):
    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_nonfinite,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from exc


def verified_signed_bytes(
    payload_path: Path,
    bundle_path: Path,
    trust_key: str | Path,
    timeout_seconds: float,
    label: str,
) -> bytes:
    payload = payload_path.read_bytes()
    bundle = bundle_path.read_bytes()
    if not payload or not bundle:
        raise ValueError(f"{label} payload or bundle is empty")
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        bundle_snapshot = root / "payload.sigstore.json"
        payload_snapshot.write_bytes(payload)
        bundle_snapshot.write_bytes(bundle)
        os.chmod(payload_snapshot, 0o400)
        os.chmod(bundle_snapshot, 0o400)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(trust_key),
             "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        verified = payload_snapshot.read_bytes()
        verified_bundle = bundle_snapshot.read_bytes()
        if verified != payload or verified_bundle != bundle:
            raise OSError(f"{label} verified payload or bundle snapshot changed")
    return verified


def load_jsonl(
    path: Path, bundle: Path, trust_key: Path,
) -> tuple[list[dict], bytes]:
    raw = verified_signed_bytes(
        path, bundle, trust_key, COMMAND_TIMEOUT_SECONDS, str(path)
    )
    text = raw.decode("utf-8", errors="strict")
    lines = text.splitlines()
    if not lines or not text.endswith("\n") or any(not line for line in lines):
        raise ValueError(f"{path}: canonical nonempty JSONL population required")
    rows = []
    for number, line in enumerate(lines, 1):
        value = strict_json(line.encode("utf-8"), f"{path}:{number}")
        if (
            not isinstance(value, dict)
            or line != json.dumps(
                value, sort_keys=True, separators=(",", ":"), allow_nan=False
            )
        ):
            raise ValueError(f"{path}:{number}: canonical JSON object required")
        rows.append(value)
    return rows, raw


def key(row: dict) -> tuple[str, str, str]:
    required = {"incident_id", "authority_id", "target_class", "target_id", "disable_revision"}
    if not required.issubset(row):
        raise ValueError(f"identity-plane row is missing fields: {required - set(row)}")
    values = (row["authority_id"], row["target_class"], row["target_id"])
    if any(not isinstance(value, str) or not value for value in values):
        raise ValueError("identity-plane key fields must be nonempty strings")
    return values


def main(
    expected_path: Path,
    expected_bundle: Path,
    observed_path: Path,
    observed_bundle: Path,
) -> None:
    expected, expected_bytes = load_jsonl(
        expected_path, expected_bundle, EXPECTED_TRUST_KEY
    )
    observed, observed_bytes = load_jsonl(
        observed_path, observed_bundle, OBSERVED_TRUST_KEY
    )
    expected_by_key = {key(row): row for row in expected}
    observed_by_key = {key(row): row for row in observed}
    if len(expected_by_key) != len(expected) or len(observed_by_key) != len(observed):
        raise ValueError("duplicate identity-plane target")
    if set(expected_by_key) != set(observed_by_key):
        raise RuntimeError("observed identity-plane population differs from expected")
    for target, wanted in expected_by_key.items():
        actual = observed_by_key[target]
        if actual.get("incident_id") != wanted.get("incident_id"):
            raise RuntimeError(f"incident binding differs: {target}")
        if actual.get("disable_revision") != wanted.get("disable_revision"):
            raise RuntimeError(f"disable revision differs: {target}")
        if actual.get("authoritative_state") not in {"DISABLED", "ABSENT"}:
            raise RuntimeError(f"principal remains enabled: {target}")
        for field in (
            "authority_revision", "observation_source", "observed_at",
            "raw_evidence_sha256",
        ):
            if not isinstance(actual.get(field), str) or not actual[field]:
                raise RuntimeError(f"identity observation lacks {field}: {target}")
        digest = actual["raw_evidence_sha256"]
        if (
            len(digest) != 64
            or set(digest) - set("0123456789abcdef")
            or digest == "0" * 64
        ):
            raise RuntimeError(f"identity evidence digest is invalid: {target}")
        if actual.get("fresh_issuance_denied") is not True:
            raise RuntimeError(f"fresh issuance or authentication was not denied: {target}")
        if actual.get("observation_complete") is not True:
            raise RuntimeError(f"authoritative observation incomplete: {target}")
    print(json.dumps({
        "schema_version": "aidefend.e001003-identity-reconciliation/v1",
        "control": "AID-E-001.003",
        "target_count": len(expected),
        "expected_manifest_sha256": hashlib.sha256(expected_bytes).hexdigest(),
        "observation_receipt_sha256": hashlib.sha256(observed_bytes).hexdigest(),
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
        "result": "reconciled",
    }, sort_keys=True, separators=(",", ":")))


if __name__ == "__main__":
    if len(sys.argv) != 5:
        raise SystemExit(
            "usage: reconcile_identity_plane.py "
            "EXPECTED.jsonl EXPECTED.jsonl.sigstore.json OBSERVED.jsonl OBSERVED.jsonl.sigstore.json"
        )
    main(
        Path(sys.argv[1]), Path(sys.argv[2]),
        Path(sys.argv[3]), Path(sys.argv[4]),
    )

Action: run provider-native read adapters under identities that cannot re-enable or modify the principal, sign their canonical JSONL observations under the pinned verifier key, and reconcile every expected boundary exactly once, and retain the authoritative revision plus negative issuance result. Do not infer success from a control-plane write response or from another E-001 sub-technique.

Open-source tools

• SPIRE (workload registration disablement and SVID issuance control)
• Keycloak
• ZITADEL

Commercial / hosted tools

• Amazon EKS Pod Identity (workload principal and association disablement)
• Google Cloud Workload Identity Federation (workload principal and provider disablement)
• Microsoft Entra Agent ID (agent principal disablement)
• Amazon Bedrock AgentCore Identity (agent identity and issuance control)

Threat mappings

MITRE ATLAS

• AML.T0012 Valid Accounts (authoritative principal disablement blocks fresh authentication and issuance for the compromised AI-agent or workload identity)
• AML.T0121 AI Agent Environment Reconstruction (disabled principals cannot re-obtain credentials to rebuild their access after eviction)
• AML.T0125 Create Account (rogue agent or workload principals are disabled at every authoritative control plane)

MAESTRO

• Agent Identity Attack (L7)
• Lateral Movement (Cross-Layer)
• Privilege Escalation (Cross-Layer)

OWASP LLM Top 10 2026

• LLM03:2026 Excessive Agency (principal disablement stops a compromised agent from obtaining fresh authority)

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI03:2026 Identity and Privilege Abuse
• ASI10:2026 Rogue Agents (revoking rogue agent identities)

NIST Adversarial Machine Learning 2025

• NISTAML.039 Compromising connected resources

Cisco Integrated AI Security and Safety Framework

• AISubtech-14.1.2 Insufficient Access Controls
• AISubtech-14.2.1 Permission Escalation via Delegation
• AISubtech-2.1.2 Capability and Permission Overreach (stops the compromised agent from acquiring fresh authority; issued artifacts require separate revocation)
• AISubtech-4.1.1 Rogue Agent Introduction (evicts attacker-created agent principals and their fresh credential-issuance paths)
• AITech-14.1 Unauthorized Access (disables workload issuance paths into AI hosting and federation infrastructure)
• AITech-14.2 Abuse of Delegated Authority
• AITech-2.1 Excessive Agency (stops the compromised agent from acquiring fresh authority; issued artifacts require separate revocation)
• AITech-4.1 Agent Injection (evicts attacker-created agent principals and their fresh credential-issuance paths)

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (identity revocation stops rogue agent actions)
• MXF: Model Exfiltration (workload identity revocation prevents model theft by compromised agents)

Databricks AI Security Framework 3.0

• Agents - Core 13.9: Identity Spoofing & Impersonation
• Agents - Core 13.3: Privilege Compromise
• Agents - Core 13.13: Rogue Agents in Multi-Agent Systems