AID-E-001.001: Root & Long-Lived Credential Object Eviction

Version 1.20261015 · Actionable sub-technique

Revoke, disable, rotate, or invalidate the exact incident-scoped population of root and long-lived credential objects, including passwords, API keys, client secrets, signing keys, and long-lived certificates, at their authoritative issuer or verifier regardless of whether the owning principal is a human, conventional service, AI agent, or workload.

Pillars: infra

Lifecycle: response

Scope boundary

Owns incident-time revocation, disablement, rotation, or invalidation of the exact root and long-lived credential-object population, including passwords, API keys, client secrets, signing keys, and long-lived certificates. Routine non-incident secret rotation is outside this control.

AID-E-001.002: AID-E-001.001 evicts root and long-lived credential objects; AID-E-001.002 revokes already-issued tokens, authentication sessions, cookies, certificates, and leases. Rotating a root secret does not prove that every artifact already issued from it is unusable, so both controls may be required for the same incident.

AID-E-001.003: AID-E-001.001 evicts individual long-lived credentials; AID-E-001.003 disables the AI-agent or workload principal and its ability to obtain new credentials. A principal can retain other issuance paths after one secret is rotated, while disabling the principal does not independently rotate every exposed credential object.

AID-E-001.004: AID-E-001.001 evicts credential objects; AID-E-001.004 removes delegated grants and connected-app authorizations. A grant can sustain or recreate access after credential rotation, and grant revocation does not invalidate an unrelated password, key, secret, or certificate.

Implementation guidance

AID-E-001.001-G001

Rotate or invalidate the exact incident-scoped population of root and long-lived secret credential objects across every declared consumer and verifier.
Rotate the credential object, not merely its stored copy

This executable path is intentionally scoped to one IAM-user access key stored in one Secrets Manager secret and consumed by an exact signed population of ECS services. Other credential and consumer classes require their own authoritative adapters. The incident authority signs the IAM user ARN, old key ID, full secret ARN, sorted ECS cluster/service population, incident ID, action nonce, and policy digest. The action service verifies that manifest locally; it does not accept a caller-provided probe verdict.

Phase-aware rotation and readback

The adapter creates the replacement key, advances AWSCURRENT, forces every signed ECS service to a stable new deployment, verifies every running task started after rotation and references the exact secret, and calls STS with the replacement credential. Only then does it deactivate the old key and prove an old-key STS call is denied. If a pre-deactivation step fails, it rolls AWSCURRENT back, redeploys and verifies every consumer on the old version, and deletes the new key only after that rollback succeeds. After old-key deactivation, an error never deletes the replacement key.

# File: response/rotate_iam_key.py
from __future__ import annotations

import argparse
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import boto3
from botocore.exceptions import ClientError

TRUST_KEY = Path("/opt/aidefend/trust/e001001-credential-authority.pub")
PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT = float(os.environ["E001001_COMMAND_TIMEOUT_SECONDS"])
WAITER_DELAY = int(os.environ["E001001_ECS_WAITER_DELAY_SECONDS"])
WAITER_ATTEMPTS = int(os.environ["E001001_ECS_WAITER_MAX_ATTEMPTS"])
FIELDS = {
    "schema_version", "control", "iam_user_name", "iam_user_arn",
    "secret_arn", "compromised_key_id", "ecs_services",
    "incident_id", "action_nonce", "policy_sha256",
}
SERVICE_FIELDS = {"cluster_arn", "service_arn"}


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 require_sha256(value: object, label: str) -> str:
    if (
        not isinstance(value, str)
        or len(value) != 64
        or set(value) - set("0123456789abcdef")
        or value == "0" * 64
    ):
        raise ValueError(f"{label} must be a nonzero lowercase SHA-256 digest")
    return value


if (
    not PROFILE_VERSION
    or not math.isfinite(COMMAND_TIMEOUT)
    or COMMAND_TIMEOUT <= 0
    or WAITER_DELAY <= 0
    or WAITER_ATTEMPTS <= 0
):
    raise RuntimeError("versioned rotation runtime profile is invalid")
require_sha256(PROFILE_SHA256, "runtime profile digest")


def canonical(value: dict[str, Any]) -> bytes:
    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()


def load_manifest(path: Path, bundle: Path) -> dict[str, Any]:
    raw = verified_signed_bytes(
        path, bundle, TRUST_KEY, COMMAND_TIMEOUT, "rotation manifest"
    )
    value = strict_json(raw, "rotation manifest")
    if (
        not isinstance(value, dict)
        or set(value) != FIELDS
        or raw != canonical(value)
        or value["schema_version"] != "aidefend.e001001.ecs-key-rotation/v1"
        or value["control"] != "AID-E-001.001"
    ):
        raise ValueError("rotation manifest schema or canonical bytes differ")
    for field in (
        "iam_user_name", "iam_user_arn", "secret_arn", "compromised_key_id",
        "incident_id", "action_nonce",
    ):
        if not isinstance(value[field], str) or not value[field] or "*" in value[field]:
            raise ValueError(f"manifest.{field} must be an exact nonempty value")
    if not 32 <= len(value["action_nonce"]) <= 64:
        raise ValueError("action_nonce must satisfy Secrets Manager idempotency bounds")
    require_sha256(value["policy_sha256"], "manifest policy digest")
    services = value["ecs_services"]
    if (
        not isinstance(services, list) or not services
        or any(not isinstance(item, dict) or set(item) != SERVICE_FIELDS for item in services)
        or services != sorted(
            services, key=lambda item: (item["cluster_arn"], item["service_arn"])
        )
        or len({(item["cluster_arn"], item["service_arn"]) for item in services})
           != len(services)
        or any(
            not item["cluster_arn"] or not item["service_arn"]
            or "*" in item["cluster_arn"] + item["service_arn"]
            for item in services
        )
    ):
        raise ValueError("ecs_services must be a sorted unique exact population")
    return value


def credentials(secret_value: str) -> dict[str, str]:
    value = strict_json(secret_value.encode("utf-8"), "credential secret")
    if (
        not isinstance(value, dict)
        or set(value) != {"aws_access_key_id", "aws_secret_access_key"}
        or any(not isinstance(item, str) or not item for item in value.values())
    ):
        raise ValueError("credential secret schema differs")
    return value


def sts_identity(creds: dict[str, str]) -> dict[str, str]:
    return boto3.client(
        "sts",
        aws_access_key_id=creds["aws_access_key_id"],
        aws_secret_access_key=creds["aws_secret_access_key"],
    ).get_caller_identity()


def verify_consumers(
    ecs, services: list[dict[str, str]], secret_arn: str, not_before: datetime,
) -> None:
    for item in services:
        cluster, service = item["cluster_arn"], item["service_arn"]
        ecs.get_waiter("services_stable").wait(
            cluster=cluster,
            services=[service],
            WaiterConfig={"Delay": WAITER_DELAY, "MaxAttempts": WAITER_ATTEMPTS},
        )
        rows = ecs.describe_services(cluster=cluster, services=[service])["services"]
        if (
            len(rows) != 1 or len(rows[0]["deployments"]) != 1
            or rows[0]["pendingCount"] != 0
        ):
            raise RuntimeError(f"ECS service did not converge: {service}")
        task_arns, token = [], None
        while True:
            request = {
                "cluster": cluster, "serviceName": service,
                "desiredStatus": "RUNNING",
            }
            if token is not None:
                request["nextToken"] = token
            page = ecs.list_tasks(**request)
            task_arns.extend(page["taskArns"])
            token = page.get("nextToken")
            if token is None:
                break
        if rows[0]["desiredCount"] < 1:
            raise RuntimeError(f"ECS service has no measurable running population: {service}")
        if len(task_arns) != rows[0]["desiredCount"]:
            raise RuntimeError(f"ECS running-task population differs: {service}")
        tasks = []
        for offset in range(0, len(task_arns), 100):
            tasks.extend(
                ecs.describe_tasks(
                    cluster=cluster, tasks=task_arns[offset:offset + 100]
                )["tasks"]
            )
        for task in tasks:
            if task.get("startedAt") is None or task["startedAt"] < not_before:
                raise RuntimeError(f"ECS task predates rotation: {task['taskArn']}")
            definition = ecs.describe_task_definition(
                taskDefinition=task["taskDefinitionArn"]
            )["taskDefinition"]
            refs = [
                secret["valueFrom"]
                for container in definition["containerDefinitions"]
                for secret in container.get("secrets", [])
            ]
            bound_refs = [
                ref for ref in refs
                if ref == secret_arn or ref.startswith(secret_arn + ":")
            ]
            if not bound_refs or any(ref != secret_arn for ref in bound_refs):
                raise RuntimeError(
                    f"ECS task must consume only the AWSCURRENT secret ARN: {task['taskArn']}"
                )


def rotate(manifest: dict[str, Any]) -> dict[str, Any]:
    iam = boto3.client("iam")
    secrets = boto3.client("secretsmanager")
    ecs = boto3.client("ecs")
    if (
        iam.get_user(UserName=manifest["iam_user_name"])["User"]["Arn"]
        != manifest["iam_user_arn"]
    ):
        raise RuntimeError("IAM user ARN differs from signed target")
    inventory = {
        item["AccessKeyId"]: item["Status"]
        for item in iam.list_access_keys(
            UserName=manifest["iam_user_name"]
        )["AccessKeyMetadata"]
    }
    if inventory.get(manifest["compromised_key_id"]) != "Active":
        raise RuntimeError("signed compromised key is not active")
    old_secret = secrets.get_secret_value(
        SecretId=manifest["secret_arn"], VersionStage="AWSCURRENT"
    )
    old_creds = credentials(old_secret["SecretString"])
    old_version = old_secret["VersionId"]
    if old_creds["aws_access_key_id"] != manifest["compromised_key_id"]:
        raise RuntimeError("AWSCURRENT does not contain the signed old key")
    created = iam.create_access_key(UserName=manifest["iam_user_name"])["AccessKey"]
    new_version, old_inactive = None, False
    try:
        rotated_at = datetime.now(timezone.utc)
        new_value = json.dumps(
            {
                "aws_access_key_id": created["AccessKeyId"],
                "aws_secret_access_key": created["SecretAccessKey"],
            },
            separators=(",", ":"),
        )
        new_version = secrets.put_secret_value(
            SecretId=manifest["secret_arn"],
            ClientRequestToken=manifest["action_nonce"],
            SecretString=new_value,
        )["VersionId"]
        for item in manifest["ecs_services"]:
            ecs.update_service(
                cluster=item["cluster_arn"], service=item["service_arn"],
                forceNewDeployment=True,
            )
        verify_consumers(
            ecs, manifest["ecs_services"], manifest["secret_arn"], rotated_at
        )
        if sts_identity(credentials(new_value))["Arn"] != manifest["iam_user_arn"]:
            raise RuntimeError("replacement credential identity differs")
        iam.update_access_key(
            UserName=manifest["iam_user_name"],
            AccessKeyId=manifest["compromised_key_id"], Status="Inactive",
        )
        old_inactive = True
        readback = {
            item["AccessKeyId"]: item["Status"]
            for item in iam.list_access_keys(
                UserName=manifest["iam_user_name"]
            )["AccessKeyMetadata"]
        }
        if readback.get(manifest["compromised_key_id"]) != "Inactive":
            raise RuntimeError("old-key status readback differs")
        try:
            sts_identity(old_creds)
        except ClientError as exc:
            denial_code = exc.response.get("Error", {}).get("Code")
            if denial_code not in {"InvalidClientTokenId", "SignatureDoesNotMatch"}:
                raise RuntimeError("old-key probe did not return a typed invalid-credential denial") from exc
        else:
            raise RuntimeError("old key remains usable")
    except Exception:
        if not old_inactive and new_version is not None:
            secrets.update_secret_version_stage(
                SecretId=manifest["secret_arn"], VersionStage="AWSCURRENT",
                MoveToVersionId=old_version, RemoveFromVersionId=new_version,
            )
            rollback_at = datetime.now(timezone.utc)
            for item in manifest["ecs_services"]:
                ecs.update_service(
                    cluster=item["cluster_arn"], service=item["service_arn"],
                    forceNewDeployment=True,
                )
            verify_consumers(
                ecs, manifest["ecs_services"], manifest["secret_arn"], rollback_at
            )
            if sts_identity(old_creds)["Arn"] != manifest["iam_user_arn"]:
                raise RuntimeError("rollback credential identity differs")
            iam.delete_access_key(
                UserName=manifest["iam_user_name"],
                AccessKeyId=created["AccessKeyId"],
            )
        raise
    return {
        "schema_version": "aidefend.e001001.ecs-key-rotation-receipt/v1",
        "control": "AID-E-001.001",
        "incident_id": manifest["incident_id"],
        "retired_key_id": manifest["compromised_key_id"],
        "replacement_key_id": created["AccessKeyId"],
        "secret_version_id": new_version,
        "ecs_services": manifest["ecs_services"],
        "runtime_profile_version": PROFILE_VERSION,
        "runtime_profile_sha256": PROFILE_SHA256,
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("manifest", type=Path)
    parser.add_argument("bundle", type=Path)
    args = parser.parse_args()
    print(json.dumps(
        rotate(load_manifest(args.manifest, args.bundle)),
        sort_keys=True, separators=(",", ":"),
    ))


if __name__ == "__main__":
    main()

A separately credentialed verifier repeats the IAM status readback, enumerates the exact signed ECS service/task population, checks the consumed secret version from deployment evidence, and runs both old-key denial and clean replacement STS probes. Root-account credentials and HSM-backed keys require provider-native break-glass adapters; this IAM-user example must not claim those populations.

AID-E-001.001-G002

Reset compromised human password credential objects without claiming revocation of issued sessions.
Reset only the exact password credential

Through a pre-registered identity-provider administration adapter using workload identity and least-privilege scopes, re-read and compare the exact human principal, set a cryptographically random recovery-only credential or invoke the provider's secure administrative reset primitive, increment the credential epoch where supported, and require verified account recovery plus phishing-resistant MFA before the next normal authentication. Keep temporary values out of logs, tickets, and evidence. Existing tokens and identity-provider, gateway, or application sessions are enumerated and handed to AID-E-001.002; their revocation is a dependency and is never claimed by the password-reset receipt.

Independent verification

A separate read-only identity confirms the reset or credential epoch at the authoritative verifier without reading the new secret. When a securely captured old password exists, an isolated probe must receive an authentication denial; otherwise preserve the missing negative probe as insufficient evidence. A controlled recovery account proves the configured reset, phishing-resistant MFA, and clean authentication path still work.

Executable human-password reset path
# File: response/reset_cognito_password.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import subprocess
import tempfile
import time
from pathlib import Path

import boto3


TARGET = Path("/run/incident/e001001-password-reset-target.json")
BUNDLE = Path(str(TARGET) + ".sigstore.json")
VERIFY_KEY = Path("/opt/aidefend/trust/e001001-password-reset-authority.pub")
RECEIPT_ROOT = Path("/var/lib/aidefend/e001001")
COMMAND_TIMEOUT_SECONDS = float(os.environ["E001001_RESET_VERIFY_TIMEOUT_SECONDS"])
TARGET_FIELDS = {
    "schema_version", "control", "user_pool_id", "username", "subject_id",
    "temporary_password_secret_id", "expected_user_modified_epoch",
    "incident_id", "action_nonce", "expires_at_epoch",
}


def canonical(value: dict) -> bytes:
    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()


def strict_object(raw: bytes) -> dict:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate target key: {key}")
            value[key] = item
        return value
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique,
        parse_constant=lambda item: (_ for _ in ()).throw(
            ValueError("non-finite target value: " + item)
        ),
    )
    if not isinstance(value, dict) or set(value) != TARGET_FIELDS:
        raise ValueError("password-reset target schema differs")
    return value


def verified_target() -> tuple[dict, str]:
    if (
        not math.isfinite(COMMAND_TIMEOUT_SECONDS)
        or COMMAND_TIMEOUT_SECONDS <= 0
    ):
        raise RuntimeError("password-reset verifier timeout is invalid")
    raw, bundle = TARGET.read_bytes(), BUNDLE.read_bytes()
    if not 1 <= len(raw) <= 65_536 or not 1 <= len(bundle) <= 1_048_576:
        raise ValueError("password-reset target or bundle is outside bootstrap bounds")
    with tempfile.TemporaryDirectory(prefix="e001001-password-reset-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_path, bundle_path = root / "target.json", root / "target.sigstore.json"
        payload_path.write_bytes(raw)
        bundle_path.write_bytes(bundle)
        os.chmod(payload_path, 0o400)
        os.chmod(bundle_path, 0o400)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(VERIFY_KEY),
                "--bundle", str(bundle_path), str(payload_path),
            ],
            check=True, capture_output=True, timeout=COMMAND_TIMEOUT_SECONDS,
        )
        if payload_path.read_bytes() != raw or bundle_path.read_bytes() != bundle:
            raise OSError("verified password-reset target changed")
    target = strict_object(raw)
    if (
        raw != canonical(target)
        or target["schema_version"] != "aidefend.e001001.password-reset/v1"
        or target["control"] != "AID-E-001.001"
    ):
        raise ValueError("password-reset target is not canonical or has wrong control")
    for field in (
        "user_pool_id", "username", "subject_id",
        "temporary_password_secret_id", "incident_id",
    ):
        if not isinstance(target[field], str) or not target[field] or "*" in target[field]:
            raise ValueError(f"target.{field} must be an exact nonempty value")
    if (
        not isinstance(target["action_nonce"], str)
        or re.fullmatch(r"[A-Za-z0-9._-]{16,96}", target["action_nonce"]) is None
        or type(target["expected_user_modified_epoch"]) is not int
        or target["expected_user_modified_epoch"] < 0
        or type(target["expires_at_epoch"]) is not int
        or target["expires_at_epoch"] <= int(time.time())
    ):
        raise ValueError("password-reset target freshness or version fence is invalid")
    return target, hashlib.sha256(raw).hexdigest()


def attributes(user: dict) -> dict[str, str]:
    return {item["Name"]: item["Value"] for item in user.get("UserAttributes", [])}


def main() -> None:
    target, target_sha256 = verified_target()
    RECEIPT_ROOT.mkdir(mode=0o700, parents=True, exist_ok=True)
    receipt_path = RECEIPT_ROOT / f"password-reset-{target['action_nonce']}.json"
    if receipt_path.exists():
        prior = json.loads(receipt_path.read_text(encoding="utf-8"))
        if (
            not isinstance(prior, dict)
            or prior.get("action_nonce") != target["action_nonce"]
            or prior.get("target_sha256") != target_sha256
        ):
            raise RuntimeError("action nonce was reused for a different target")
        print(json.dumps(prior, sort_keys=True, separators=(",", ":")))
        return

    idp = boto3.client("cognito-idp")
    secrets = boto3.client("secretsmanager")
    before = idp.admin_get_user(
        UserPoolId=target["user_pool_id"], Username=target["username"]
    )
    before_attributes = attributes(before)
    if (
        before.get("Username") != target["username"]
        or before_attributes.get("sub") != target["subject_id"]
        or before.get("Enabled") is not True
        or int(before["UserLastModifiedDate"].timestamp())
        != target["expected_user_modified_epoch"]
    ):
        raise RuntimeError("authoritative password-reset pre-state differs")

    temporary_password = secrets.get_secret_value(
        SecretId=target["temporary_password_secret_id"]
    ).get("SecretString")
    if not isinstance(temporary_password, str) or not temporary_password:
        raise RuntimeError("recovery-only password secret is empty")
    idp.admin_set_user_password(
        UserPoolId=target["user_pool_id"],
        Username=target["username"],
        Password=temporary_password,
        Permanent=False,
    )
    del temporary_password

    after = idp.admin_get_user(
        UserPoolId=target["user_pool_id"], Username=target["username"]
    )
    if (
        after.get("Username") != target["username"]
        or attributes(after).get("sub") != target["subject_id"]
        or after.get("UserStatus") != "FORCE_CHANGE_PASSWORD"
        or after.get("Enabled") is not True
    ):
        raise RuntimeError("authoritative password-reset readback differs")
    receipt = {
        "schema_version": "aidefend.e001001.password-reset-receipt/v1",
        "control": "AID-E-001.001",
        "incident_id": target["incident_id"],
        "action_nonce": target["action_nonce"],
        "target_sha256": target_sha256,
        "user_pool_id": target["user_pool_id"],
        "username": target["username"],
        "subject_id": target["subject_id"],
        "before_user_modified_epoch": int(before["UserLastModifiedDate"].timestamp()),
        "after_user_modified_epoch": int(after["UserLastModifiedDate"].timestamp()),
        "after_status": after["UserStatus"],
    }
    with receipt_path.open("x", encoding="utf-8", newline="\n") as handle:
        json.dump(receipt, handle, sort_keys=True, separators=(",", ":"))
        handle.write("\n")
    print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))


if __name__ == "__main__":
    main()

Use the equivalent provider-native reset API for the affected identity system and require MFA/identity re-proofing under incident policy. This changes the password credential object only; terminate issued sessions and tokens separately under AID-E-001.002.

AID-E-001.001-G003

Disable and remove exact attacker-created root or long-lived credential objects.
Act only on signed credential-object identities

A provider-native adapter from trusted configuration re-enumerates each exact API key, client secret, signing key, certificate, or comparable long-lived credential, compares immutable object ID, fingerprint, owner, privilege, issuer, and resource version, then first disables the object. After the policy-defined forensic hold, delete only objects explicitly authorized for deletion and retain a provider-native deny or tombstone where supported. Attacker-created human, service, agent, or workload principal objects are handed to AID-E-001.003; issued tokens and sessions are handed to AID-E-001.002.

Independent verification

A separately credentialed read-only adapter enumerates the complete affected provider population and requires every signed credential object to be disabled or absent as authorized, every retained object unchanged, and controlled authentication with each safely captured unauthorized credential to be denied. It also proves the action set exactly equals the signed manifest set; an unmeasured key class or removed account record cannot substitute for credential-object evidence.

Executable attacker-created key removal
#!/usr/bin/env bash
set -euo pipefail
umask 077

TARGET=/run/incident/e001001-api-key-eviction.json
BUNDLE=/run/incident/e001001-api-key-eviction.json.sigstore.json
VERIFY_KEY=/opt/aidefend/trust/e001001-credential-object-authority.pub
: "${E001001_IAM_COMMAND_TIMEOUT_SECONDS:?required by the versioned runtime profile}"
[[ "$E001001_IAM_COMMAND_TIMEOUT_SECONDS" =~ ^[0-9]+([.][0-9]+)?$ ]]
awk -v value="$E001001_IAM_COMMAND_TIMEOUT_SECONDS" 'BEGIN { exit !(value > 0) }'

SNAPSHOT_ROOT="$(mktemp -d)"
chmod 0700 "$SNAPSHOT_ROOT"
trap 'rm -rf -- "$SNAPSHOT_ROOT"' EXIT
install -m 0400 -- "$TARGET" "$SNAPSHOT_ROOT/target.json"
install -m 0400 -- "$BUNDLE" "$SNAPSHOT_ROOT/target.sigstore.json"
timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS" cosign verify-blob   --key "$VERIFY_KEY"   --bundle "$SNAPSHOT_ROOT/target.sigstore.json"   "$SNAPSHOT_ROOT/target.json" >/dev/null
jq -e '
  type == "object"
  and (keys == [
    "action_nonce","control","delete_not_before_epoch","expected_account_id",
    "expected_create_date","expected_status","iam_user_arn","iam_user_name",
    "incident_id","key_id","schema_version","terminal_action"
  ])
  and .schema_version == "aidefend.e001001.api-key-eviction/v1"
  and .control == "AID-E-001.001"
  and (.action_nonce | type == "string" and test("^[A-Za-z0-9._-]{16,96}$"))
  and (.expected_account_id | type == "string" and test("^[0-9]{12}$"))
  and (.iam_user_name | (type == "string" and length > 0 and (contains("*") | not)))
  and (.iam_user_arn | type == "string" and startswith("arn:"))
  and (.key_id | type == "string" and test("^[A-Z0-9]{16,128}$"))
  and (.expected_create_date | type == "string" and length > 0)
  and .expected_status == "Active"
  and (.incident_id | type == "string" and test("^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"))
  and (.terminal_action == "disable_only" or .terminal_action == "disable_then_delete")
  and (.delete_not_before_epoch | type == "number" and floor == .)
' "$SNAPSHOT_ROOT/target.json" >/dev/null

IAM_USER="$(jq -er '.iam_user_name' "$SNAPSHOT_ROOT/target.json")"
IAM_USER_ARN="$(jq -er '.iam_user_arn' "$SNAPSHOT_ROOT/target.json")"
KEY_ID="$(jq -er '.key_id' "$SNAPSHOT_ROOT/target.json")"
EXPECTED_CREATE_DATE="$(jq -er '.expected_create_date' "$SNAPSHOT_ROOT/target.json")"
EXPECTED_ACCOUNT_ID="$(jq -er '.expected_account_id' "$SNAPSHOT_ROOT/target.json")"
TERMINAL_ACTION="$(jq -er '.terminal_action' "$SNAPSHOT_ROOT/target.json")"
DELETE_NOT_BEFORE="$(jq -er '.delete_not_before_epoch' "$SNAPSHOT_ROOT/target.json")"
INCIDENT_ID="$(jq -er '.incident_id' "$SNAPSHOT_ROOT/target.json")"
EVIDENCE_DIR="evidence/$INCIDENT_ID"
mkdir -p -- "$EVIDENCE_DIR"

ACCOUNT_ID="$(timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"   aws sts get-caller-identity --query Account --output text --no-cli-pager)"
[[ "$ACCOUNT_ID" == "$EXPECTED_ACCOUNT_ID" ]]
ACTUAL_USER_ARN="$(timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"   aws iam get-user --user-name "$IAM_USER" --query 'User.Arn' --output text --no-cli-pager)"
[[ "$ACTUAL_USER_ARN" == "$IAM_USER_ARN" ]]

timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"   aws iam list-access-keys --user-name "$IAM_USER" --no-cli-pager   > "$EVIDENCE_DIR/access-keys.before.json"
jq -e --arg key "$KEY_ID" --arg created "$EXPECTED_CREATE_DATE" '
  [.AccessKeyMetadata[]
   | select(.AccessKeyId == $key and .Status == "Active" and .CreateDate == $created)]
  | length == 1
' "$EVIDENCE_DIR/access-keys.before.json" >/dev/null

timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"   aws iam update-access-key --user-name "$IAM_USER"   --access-key-id "$KEY_ID" --status Inactive --no-cli-pager
timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"   aws iam list-access-keys --user-name "$IAM_USER" --no-cli-pager   > "$EVIDENCE_DIR/access-keys.disabled.json"
jq -e --arg key "$KEY_ID" --arg created "$EXPECTED_CREATE_DATE" '
  [.AccessKeyMetadata[]
   | select(.AccessKeyId == $key and .Status == "Inactive" and .CreateDate == $created)]
  | length == 1
' "$EVIDENCE_DIR/access-keys.disabled.json" >/dev/null

if [[ "$TERMINAL_ACTION" == "disable_then_delete" ]]; then
  if [[ "$(date +%s)" -lt "$DELETE_NOT_BEFORE" ]]; then
    echo "signed forensic hold has not elapsed; key remains disabled" >&2
    exit 3
  fi
  timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"     aws iam delete-access-key --user-name "$IAM_USER"     --access-key-id "$KEY_ID" --no-cli-pager
fi

timeout "$E001001_IAM_COMMAND_TIMEOUT_SECONDS"   aws iam list-access-keys --user-name "$IAM_USER" --no-cli-pager   > "$EVIDENCE_DIR/access-keys.after.json"
if [[ "$TERMINAL_ACTION" == "disable_then_delete" ]]; then
  jq -e --arg key "$KEY_ID" '
    [.AccessKeyMetadata[] | select(.AccessKeyId == $key)] | length == 0
  ' "$EVIDENCE_DIR/access-keys.after.json" >/dev/null
else
  jq -e --arg key "$KEY_ID" '
    [.AccessKeyMetadata[] | select(.AccessKeyId == $key and .Status == "Inactive")]
    | length == 1
  ' "$EVIDENCE_DIR/access-keys.after.json" >/dev/null
fi

Bind the target key ID to incident evidence before deletion and retain the before/after authoritative inventories. Do not remove a key merely because its creation time looks unusual; confirm ownership and preserve required forensic records first.

AID-E-001.001-G004

Rotate compromised issuer signing or trust material only when the issuer credential boundary itself is affected.
When issuer material is in scope

Use this method only when forensic and cryptographic evidence shows that an issuer signing key, CA key, trust anchor, or equivalent long-lived issuer credential may be compromised. A single compromised workload certificate or token does not authorize trust-domain-wide rotation. Before changing trust, bind the exact old and replacement key IDs, bundle digests, relying-party population, activation order, rollback prohibition, and old-anchor removal condition to the signed incident plan.

Step 3: Rotate issuer material only for issuer or trust-anchor compromise

Create a new signing key inside the approved HSM or managed CA, then publish a signed rotation manifest that binds the incident, affected trust domain, old and new bundle digests, activation time, coverage evidence, old-anchor removal deadline, and rollback owner. Distribute the staged new bundle to clean relying parties first, switch issuance to the new signer, verify the replacement identity, deny and drain the old issuer path, and remove the compromised anchor only after coverage is proven. Never reuse the compromised signing key as a rollback mechanism.

{
  "schema_version": "aidefend.identity-trust-rotation/v1",
  "incident_id": "INC-2026-0417",
  "trust_domain": "spiffe://example.org",
  "old_bundle_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "new_bundle_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
  "new_issuer_key_id": "hsm:aidefend-issuer-2026-04-b",
  "activation_at": "2026-04-17T15:00:00Z",
  "remove_old_anchor_after": "2026-04-17T16:00:00Z",
  "coverage_receipt": "evidence://INC-2026-0417/trust-rollout.json",
  "rollback_owner": "identity-incident-commander"
}
Independent verification

A separately credentialed verifier reads the effective trust bundle and signer identity at every declared issuer and relying party, requires the replacement signer and bundle digest to be active, proves the compromised signer can no longer issue an accepted credential, and proves the old anchor is absent after the signed coverage condition. A clean replacement credential must authenticate under least privilege. Missing relying parties, a staged bundle that was not consumed, retained old-anchor acceptance, or reuse of the compromised key blocks this credential-object eviction.

Open-source tools

Source-available / open-weight 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-E-001.001: Root & Long-Lived Credential Object Eviction | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service