AID-E-001.002: Issued Token, Authentication Session & Lease Revocation

Version 1.20261015 · Actionable sub-technique

Revoke or invalidate the complete incident-scoped population of already-issued access tokens, refresh tokens, session cookies, authorization-server sessions, gateway authentication sessions, and agent-to-agent leases at every authoritative issuer, verifier, cache, and enforcement point.

Pillars: infra

Lifecycle: response

Scope boundary

Owns revocation or invalidation of the complete incident-scoped population of already-issued access and refresh tokens, session cookies, authorization-server sessions, gateway authentication sessions, workload certificates, established authenticated channels, and agent-to-agent leases. Automated response integration is only an implementation method, not a separate control outcome.

AID-E-001.001: AID-E-001.002 revokes already-issued authentication artifacts; AID-E-001.001 evicts their root or long-lived credential objects. Revoking current tokens and sessions does not rotate a compromised password, key, client secret, signing key, or long-lived certificate that can issue future artifacts.

AID-E-001.003: AID-E-001.002 invalidates artifacts already issued to a principal; AID-E-001.003 disables the principal and stops new issuance. Existing tokens can remain effective after principal disablement, and revoking the current token set does not necessarily prevent the principal from obtaining replacements.

AID-E-001.004: AID-E-001.002 revokes issued tokens and authentication sessions; AID-E-001.004 removes the delegated grants or connected-app authorizations that can mint fresh tokens or sustain SaaS access. Both populations require authoritative verification when a delegated authorization was involved.

AID-E-005: AID-E-001.002 revokes identity-provider and gateway authentication sessions; AID-E-005 removes durable application session and agent-state records. A durable conversation, webhook, queue, schedule, or tool registration can survive authentication-session revocation and reload compromised state later.

Implementation guidance

AID-E-001.002-G001

Revoke the complete incident-scoped population of access tokens, refresh tokens, cookies, authorization-server sessions, gateway sessions, and agent leases at their authoritative issuers and enforcement points.
Signed target population

This example owns already-issued Keycloak online user sessions. It does not disable AWS access keys or other long-lived credentials; those belong to AID-E-001.001. It does not delete delegated grants, consents, or app-role assignments that can authorize future issuance; those belong to AID-E-001.004. The incident scope must declare every applicable issuer and artifact class. Offline sessions, broker sessions, provider-native refresh families, gateway sessions, and agent leases require their own authoritative adapters; an applicable class without a working enumerator and revoker leaves E-001.002 incomplete.

The recovery authority signs the exact immutable admin API base URL, issuer, realm, subject, session, incident, and policy bindings. Never place raw access or refresh tokens in the manifest or ordinary logs.

{
  "schema_version": "aidefend.e001002.keycloak-session-eviction/v1",
  "control": "AID-E-001.002",
  "admin_base_url": "https://id.example.com",
  "issuer": "https://id.example.com/realms/ai-production",
  "realm": "ai-production",
  "subject_id": "6a105f68-26d7-45f2-9567-7b3ee5bff4a7",
  "session_ids": [
    "8f715b78-8075-4e98-8690-1ed91f7c90ef"
  ],
  "incident_id": "INC-2026-0718-0042",
  "policy_sha256": "6d6f9cb8164f379df6e0d4dc97e7c8797806f7657fd44a4c44a9ada0c1c6c8b3"
}
Revoke only the exact signed Keycloak session population
# File: incident_response/revoke_keycloak_sessions.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import stat
import subprocess
import tempfile
import sys
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import quote, urlsplit

import ssl

import httpx


EXPECTED_FIELDS = {
    "schema_version", "control", "admin_base_url", "issuer", "realm",
    "subject_id", "session_ids", "incident_id", "policy_sha256",
}
SAFE_SEGMENT = re.compile(r"^[A-Za-z0-9._-]+$")
TRUST_KEY = Path("/opt/aidefend/trust/e001002-session-authority.pub")
OWNER_TOKEN_FILE = Path(os.environ["KEYCLOAK_OWNER_TOKEN_FILE"])
CA_FILE = os.environ["KEYCLOAK_CA_FILE"]
TLS_CONTEXT = ssl.create_default_context(cafile=CA_FILE)
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["E001002_COMMAND_TIMEOUT_SECONDS"])
CONNECT_TIMEOUT_SECONDS = float(os.environ["E001002_CONNECT_TIMEOUT_SECONDS"])
READ_TIMEOUT_SECONDS = float(os.environ["E001002_READ_TIMEOUT_SECONDS"])
MAXIMUM_RESPONSE_BYTES = int(os.environ["E001002_MAXIMUM_RESPONSE_BYTES"])
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 any(
        not math.isfinite(value) or value <= 0
        for value in (
            COMMAND_TIMEOUT_SECONDS,
            CONNECT_TIMEOUT_SECONDS,
            READ_TIMEOUT_SECONDS,
        )
    )
    or MAXIMUM_RESPONSE_BYTES < 1
):
    raise RuntimeError("versioned E-001.002 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 canonical_json_bytes(value: dict) -> bytes:
    return (
        json.dumps(
            value,
            sort_keys=True,
            separators=(",", ":"),
            ensure_ascii=False,
            allow_nan=False,
        )
        + "\n"
    ).encode("utf-8")


def require_sha256(value: object, field: str) -> str:
    if not isinstance(value, str) or len(value) != 64:
        raise ValueError(f"{field} must be a SHA-256 digest")
    try:
        int(value, 16)
    except ValueError as exc:
        raise ValueError(f"{field} must be a SHA-256 digest") from exc
    if value.lower() == "0" * 64:
        raise ValueError(f"{field} must not be the unresolved zero digest")
    return value.lower()


def verify_manifest(path: Path, bundle: Path) -> tuple[dict, bytes]:
    raw = verified_signed_bytes(
        path, bundle, TRUST_KEY, COMMAND_TIMEOUT_SECONDS,
        "Keycloak session-eviction manifest",
    )
    value = strict_json(raw, "Keycloak session-eviction manifest")
    if (
        not isinstance(value, dict)
        or set(value) != EXPECTED_FIELDS
        or raw != canonical_json_bytes(value)
        or value["schema_version"]
        != "aidefend.e001002.keycloak-session-eviction/v1"
        or value["control"] != "AID-E-001.002"
    ):
        raise ValueError("manifest schema, canonical bytes, or control differs")
    for field in (
        "admin_base_url", "issuer", "realm", "subject_id", "incident_id",
    ):
        if not isinstance(value[field], str) or not value[field]:
            raise ValueError(f"manifest.{field} is required")
    if (
        not SAFE_SEGMENT.fullmatch(value["realm"])
        or not SAFE_SEGMENT.fullmatch(value["subject_id"])
    ):
        raise ValueError("realm or subject ID is not a bounded path segment")
    sessions = value["session_ids"]
    if (
        not isinstance(sessions, list)
        or not sessions
        or sessions != sorted(set(sessions))
        or any(
            not isinstance(item, str) or not SAFE_SEGMENT.fullmatch(item)
            for item in sessions
        )
    ):
        raise ValueError("session_ids must be a sorted unique exact-ID list")
    value["policy_sha256"] = require_sha256(
        value["policy_sha256"], "manifest.policy_sha256"
    )
    return value, raw


def read_owner_token() -> str:
    flags = os.O_RDONLY
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(OWNER_TOKEN_FILE, flags)
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode):
            raise PermissionError("owner token must be a regular file")
        if stat.S_IMODE(info.st_mode) & 0o077:
            raise PermissionError(
                "owner token file must not be group/world accessible"
            )
        with os.fdopen(descriptor, "r", encoding="utf-8") as handle:
            descriptor = -1
            token = handle.read().strip()
    finally:
        if descriptor != -1:
            os.close(descriptor)
    if not token or "\n" in token or "\r" in token:
        raise PermissionError("owner token file is empty or malformed")
    return token


def bounded_json(client: httpx.Client, path: str) -> object:
    with client.stream("GET", path) as response:
        response.raise_for_status()
        chunks = []
        size = 0
        for chunk in response.iter_bytes():
            size += len(chunk)
            if size > MAXIMUM_RESPONSE_BYTES:
                raise RuntimeError("Keycloak response exceeds signed bound")
            chunks.append(chunk)
    return strict_json(b"".join(chunks), "Keycloak response")


def enumerate_session_ids(
    client: httpx.Client,
    realm: str,
    subject_id: str,
) -> list[str]:
    path = (
        f"/admin/realms/{quote(realm, safe='')}/users/"
        f"{quote(subject_id, safe='')}/sessions"
    )
    value = bounded_json(client, path)
    if not isinstance(value, list):
        raise RuntimeError("Keycloak session response must be an array")
    identifiers = []
    for item in value:
        if not isinstance(item, dict) or not isinstance(item.get("id"), str):
            raise RuntimeError("Keycloak session response contains an invalid row")
        identifiers.append(item["id"])
    if len(identifiers) != len(set(identifiers)):
        raise RuntimeError("Keycloak returned duplicate session IDs")
    return sorted(identifiers)


def main() -> None:
    manifest, manifest_bytes = verify_manifest(
        Path(sys.argv[1]), Path(sys.argv[2])
    )
    admin_base_url = manifest["admin_base_url"].rstrip("/")
    issuer = manifest["issuer"].rstrip("/")
    admin_parts = urlsplit(admin_base_url)
    issuer_parts = urlsplit(issuer)
    for label, parts in (
        ("admin_base_url", admin_parts),
        ("issuer", issuer_parts),
    ):
        if (
            parts.scheme != "https"
            or not parts.netloc
            or parts.username is not None
            or parts.password is not None
            or parts.query
            or parts.fragment
        ):
            raise ValueError(f"signed {label} must be an absolute HTTPS URL")
    if admin_parts.path not in ("", "/"):
        raise ValueError("signed admin_base_url must be an HTTPS origin")
    if (
        admin_parts.scheme,
        admin_parts.hostname,
        admin_parts.port,
    ) != (
        issuer_parts.scheme,
        issuer_parts.hostname,
        issuer_parts.port,
    ):
        raise ValueError("Keycloak admin base and issuer origins differ")
    expected_issuer = (
        f"{admin_base_url}/realms/{quote(manifest['realm'], safe='')}"
    )
    if issuer != expected_issuer:
        raise ValueError("signed issuer is not the declared Keycloak realm issuer")
    discovery_url = issuer + "/.well-known/openid-configuration"

    timeout = httpx.Timeout(
        READ_TIMEOUT_SECONDS,
        connect=CONNECT_TIMEOUT_SECONDS,
        write=READ_TIMEOUT_SECONDS,
        pool=CONNECT_TIMEOUT_SECONDS,
    )
    with httpx.Client(
        verify=TLS_CONTEXT,
        timeout=timeout,
        follow_redirects=False,
        trust_env=False,
    ) as discovery_client:
        discovery = bounded_json(discovery_client, discovery_url)
    if (
        not isinstance(discovery, dict)
        or not isinstance(discovery.get("issuer"), str)
        or discovery["issuer"].rstrip("/") != issuer
    ):
        raise RuntimeError("OIDC discovery issuer differs from signed issuer")

    headers = {"Authorization": f"Bearer {read_owner_token()}"}
    targets = manifest["session_ids"]
    actions = []
    with httpx.Client(
        base_url=admin_base_url,
        headers=headers,
        verify=TLS_CONTEXT,
        timeout=timeout,
        follow_redirects=False,
        trust_env=False,
    ) as client:
        before = enumerate_session_ids(
            client, manifest["realm"], manifest["subject_id"]
        )
        before_set = set(before)
        for session_id in targets:
            if session_id not in before_set:
                actions.append(
                    {
                        "session_id": session_id,
                        "present_before": False,
                        "action": "already_absent",
                        "delete_status": None,
                    }
                )
                continue
            path = (
                f"/admin/realms/{quote(manifest['realm'], safe='')}/sessions/"
                f"{quote(session_id, safe='')}"
            )
            with client.stream("DELETE", path) as response:
                delete_status = response.status_code
                response_size = 0
                for chunk in response.iter_bytes():
                    response_size += len(chunk)
                    if response_size > MAXIMUM_RESPONSE_BYTES:
                        raise RuntimeError(
                            "Keycloak deletion response exceeds signed bound"
                        )
            if delete_status not in (204, 404):
                raise RuntimeError(
                    f"Keycloak session deletion failed: {delete_status}"
                )
            actions.append(
                {
                    "session_id": session_id,
                    "present_before": True,
                    "action": "delete",
                    "delete_status": delete_status,
                }
            )
        after = enumerate_session_ids(
            client, manifest["realm"], manifest["subject_id"]
        )
    remaining = sorted(set(targets) & set(after))
    if remaining:
        raise RuntimeError(
            "target sessions remain after revocation: " + ",".join(remaining)
        )
    receipt = {
        "schema_version": "aidefend.e001002.session-owner-receipt/v1",
        "control": "AID-E-001.002",
        "admin_base_url": admin_base_url,
        "issuer": issuer,
        "realm": manifest["realm"],
        "subject_id": manifest["subject_id"],
        "incident_id": manifest["incident_id"],
        "policy_sha256": manifest["policy_sha256"],
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
        "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
        "target_session_ids": targets,
        "actions": actions,
        "remaining_target_session_ids": remaining,
        "completed_at": datetime.now(timezone.utc).isoformat(),
    }
    sys.stdout.buffer.write(canonical_json_bytes(receipt))


if __name__ == "__main__":
    main()
Run and sign the owner receipt
#!/usr/bin/env bash
set -euo pipefail
python incident_response/revoke_keycloak_sessions.py   evidence/e001002-keycloak-sessions.json   evidence/e001002-keycloak-sessions.json.sigstore.json   > evidence/e001002-session-owner-receipt.json
cosign sign-blob --yes   --key env://E001002_OWNER_SIGNING_KEY   --bundle evidence/e001002-session-owner-receipt.json.sigstore.json   evidence/e001002-session-owner-receipt.json
Independent verification

A separate read-only Keycloak identity re-verifies the authority manifest and owner receipt, enumerates the same subject and exact target population, and requires every targeted session ID to be absent. It also checks every declared provider/session-class adapter, confirms non-target sessions were not revoked, and performs a provider-approved negative access or refresh test for each safely retained incident token. For a self-contained JWT that a relying party could still accept after issuer-session deletion, require the deny-state propagation and relying-party probes in the next guidance before E-001.002 passes. Preserve canonical manifests and receipts, signatures, IdP readbacks, negative probes, adapter versions, policy digests, and errors in immutable incident evidence.

Action: revoke only the exact signed issued-session population at its authoritative issuer, then accept E-001.002 only after independent absence and end-to-end denial evidence covers every applicable token and session class.

AID-E-001.002-G002

Ensure prompt propagation of revocation for stateless tokens.

Runtime policy: Load permitted clock skew, Redis connect/socket/health-check bounds, and JWKS cache lifetime from the signed, versioned JWT-revocation profile bound to this issuer, audience, gateway release, and namespace, bind its version/digest to enforcement evidence, and return ERROR if the profile or any value is absent, invalid, or unverifiable.

Concept:

A stateless token is not accepted merely because it parses. The enforcement point must verify the signature with an issuer-controlled key, pin the allowed algorithm, require the expected issuer and audience, enforce exp/iat/nbf time claims, require a non-empty jti, and then consult a shared revocation store on every request. A missing key, unavailable JWKS endpoint, unavailable revocation store, malformed claim, or stale token must fail closed.

Step 1: Verify the token and create a scoped revocation record
# File: api/jwt_revocation.py
from __future__ import annotations

import json
import math
import os
import ssl
import time
from urllib.parse import urlsplit

import httpx
import jwt
import redis
from jwt import InvalidTokenError

ISSUER = os.environ["JWT_ISSUER"].rstrip("/")
AUDIENCE = os.environ["JWT_AUDIENCE"]
JWKS_URL = os.environ["JWT_JWKS_URL"]
REVOCATION_REDIS_URL = os.environ["REVOCATION_REDIS_URL"]
ALLOWED_ALGORITHMS = ["RS256"]
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
CLOCK_SKEW_SECONDS = float(os.environ["JWT_CLOCK_SKEW_SECONDS"])
REDIS_CONNECT_TIMEOUT_SECONDS = float(os.environ["JWT_REDIS_CONNECT_TIMEOUT_SECONDS"])
REDIS_SOCKET_TIMEOUT_SECONDS = float(os.environ["JWT_REDIS_SOCKET_TIMEOUT_SECONDS"])
REDIS_HEALTH_CHECK_INTERVAL_SECONDS = float(os.environ["JWT_REDIS_HEALTH_CHECK_INTERVAL_SECONDS"])
JWKS_CACHE_LIFESPAN_SECONDS = int(os.environ["JWT_JWKS_CACHE_LIFESPAN_SECONDS"])
JWKS_TIMEOUT_SECONDS = float(os.environ["JWT_JWKS_TIMEOUT_SECONDS"])
JWKS_CONNECT_TIMEOUT_SECONDS = float(os.environ["JWT_JWKS_CONNECT_TIMEOUT_SECONDS"])
JWKS_MAX_RESPONSE_BYTES = int(os.environ["JWT_JWKS_MAX_RESPONSE_BYTES"])
JWKS_RESPONSE_CHUNK_BYTES = int(os.environ["JWT_JWKS_RESPONSE_CHUNK_BYTES"])
JWKS_CA_FILE = os.environ["JWT_JWKS_CA_FILE"]
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(CLOCK_SKEW_SECONDS) or CLOCK_SKEW_SECONDS < 0
        or any(not math.isfinite(value) or value <= 0 for value in (
            REDIS_CONNECT_TIMEOUT_SECONDS, REDIS_SOCKET_TIMEOUT_SECONDS,
            REDIS_HEALTH_CHECK_INTERVAL_SECONDS, JWKS_TIMEOUT_SECONDS,
            JWKS_CONNECT_TIMEOUT_SECONDS,
        ))
        or JWKS_CACHE_LIFESPAN_SECONDS < 1
        or JWKS_MAX_RESPONSE_BYTES < 2
        or JWKS_RESPONSE_CHUNK_BYTES < 1
        or JWKS_RESPONSE_CHUNK_BYTES > JWKS_MAX_RESPONSE_BYTES):
    raise RuntimeError("versioned JWT-revocation runtime profile is invalid")

redis_client = redis.Redis.from_url(
    REVOCATION_REDIS_URL,
    decode_responses=True,
    socket_connect_timeout=REDIS_CONNECT_TIMEOUT_SECONDS,
    socket_timeout=REDIS_SOCKET_TIMEOUT_SECONDS,
    health_check_interval=REDIS_HEALTH_CHECK_INTERVAL_SECONDS,
)
_jwks_document = None
_jwks_expires_at = 0.0


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


def https_origin(value: str) -> tuple[str, str, int]:
    parsed = urlsplit(value)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
        raise ValueError("issuer/JWKS URL is not canonical HTTPS")
    return parsed.scheme, parsed.hostname.lower(), parsed.port or 443


def load_jwks() -> dict:
    global _jwks_document, _jwks_expires_at
    now = time.monotonic()
    if _jwks_document is not None and now < _jwks_expires_at:
        return _jwks_document
    if https_origin(JWKS_URL) != https_origin(ISSUER):
        raise ValueError("JWKS origin differs from the signed issuer origin")
    tls = ssl.create_default_context(cafile=JWKS_CA_FILE)
    timeout = httpx.Timeout(JWKS_TIMEOUT_SECONDS, connect=JWKS_CONNECT_TIMEOUT_SECONDS)
    with httpx.Client(verify=tls, timeout=timeout, follow_redirects=False) as client:
        with client.stream("GET", JWKS_URL, headers={"Accept-Encoding": "identity"}) as response:
            if 300 <= response.status_code < 400:
                raise RuntimeError("JWKS redirect is forbidden")
            response.raise_for_status()
            chunks, observed = [], 0
            for chunk in response.iter_bytes(JWKS_RESPONSE_CHUNK_BYTES):
                observed += len(chunk)
                if observed > JWKS_MAX_RESPONSE_BYTES:
                    raise RuntimeError("JWKS response exceeds signed bound")
                chunks.append(chunk)
    document = json.loads(b"".join(chunks).decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
    if not isinstance(document, dict) or set(document) != {"keys"} or not isinstance(document["keys"], list):
        raise ValueError("JWKS schema differs")
    _jwks_document = document
    _jwks_expires_at = now + JWKS_CACHE_LIFESPAN_SECONDS
    return document


def verify_token(token: str) -> dict:
    if not isinstance(token, str) or not token:
        raise PermissionError("bearer token is missing")
    try:
        header = jwt.get_unverified_header(token)
        if header.get("alg") != "RS256" or not isinstance(header.get("kid"), str) or not header["kid"]:
            raise ValueError("JWT header is invalid")
        matches = [item for item in load_jwks()["keys"]
                   if isinstance(item, dict) and item.get("kid") == header["kid"] and item.get("alg") == "RS256"]
        if len(matches) != 1:
            raise ValueError("JWT signing key is absent or ambiguous")
        signing_key = jwt.PyJWK.from_dict(matches[0], algorithm="RS256").key
        claims = jwt.decode(
            token, signing_key, algorithms=ALLOWED_ALGORITHMS,
            audience=AUDIENCE, issuer=ISSUER, leeway=CLOCK_SKEW_SECONDS,
            options={"require": ["sub", "jti", "iat", "nbf", "exp"],
                     "verify_signature": True, "verify_aud": True, "verify_iss": True,
                     "verify_iat": True, "verify_nbf": True, "verify_exp": True},
        )
    except (InvalidTokenError, httpx.HTTPError, OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
        raise PermissionError("JWT cryptographic or claim validation failed") from exc
    jti = claims.get("jti")
    exp = claims.get("exp")
    if not isinstance(jti, str) or not jti or isinstance(exp, bool) or not isinstance(exp, int):
        raise PermissionError("JWT jti or exp claim is invalid")
    return claims


def revocation_key(jti: str) -> str:
    return f"jwt_revoked:{ISSUER}:{AUDIENCE}:{jti}"


def revoke_verified_token(token: str, incident_id: str) -> dict:
    claims = verify_token(token)
    jti = claims["jti"]
    exp = claims["exp"]
    # Retain through the verifier's complete accepted lifetime. Ceil avoids
    # truncating a fractional skew; one-second retention margin is arithmetic
    # precision, not a universal operational policy.
    deny_until = math.ceil(exp + CLOCK_SKEW_SECONDS)
    ttl = max(1, math.ceil(deny_until - time.time()) + 1)
    if ttl <= 0:
        raise PermissionError("cannot create a revocation record for an expired token")
    evidence = json.dumps(
        {"incident_id": incident_id, "sub": claims["sub"], "jti": jti, "exp": exp},
        sort_keys=True,
    )
    key = revocation_key(jti)
    try:
        stored = redis_client.set(key, evidence, ex=ttl)
        observed = redis_client.get(key)
    except redis.RedisError as exc:
        raise RuntimeError("revocation store unavailable; token remains denied") from exc
    if stored is not True or observed != evidence:
        raise RuntimeError("revocation record write/readback differs")
    return {
        "jti": jti,
        "expires_at": exp,
        "deny_until": deny_until,
        "revoked": True,
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
    }
Step 2: Enforce the deny decision on every authenticated request
# File: api/auth_middleware.py
import redis
from fastapi import HTTPException

from api.jwt_revocation import redis_client, revocation_key, verify_token


def validate_token_with_revocation_check(token: str) -> dict:
    try:
        claims = verify_token(token)
        revoked = redis_client.exists(revocation_key(claims["jti"]))
    except PermissionError as exc:
        raise HTTPException(status_code=401, detail="invalid bearer token") from exc
    except redis.RedisError as exc:
        raise HTTPException(status_code=503, detail="token revocation check unavailable") from exc

    if revoked:
        raise HTTPException(status_code=401, detail="bearer token has been revoked")
    return claims
Step 3: Verify propagation independently

From a separately credentialed probe, present the revoked token to every gateway or relying-party class and require an authentication denial. Store the signed probe result with issuer, audience, jti, policy revision, probe timestamp, and token expiry and deny_until in append-only incident evidence. Repeat the denial probe after nominal exp but before exp plus the configured verifier leeway; include fractional-clock boundaries and revocation during that leeway window. Alert if a relying party accepts the token or if the expected denial record is absent.

Action: Cryptographically validate before deriving jti/exp, write a TTL-bound scoped deny record to the shared store, fail closed when verification or deny-state lookup is unavailable, and close the response only after a fresh independent negative test proves the revoked token is denied at every in-scope relying party.

AID-E-001.002-G003

Invalidate already-issued workload certificates and terminate established mTLS sessions at every relying-party boundary.

Runtime policy: Load replica convergence deadline, poll cadence, probe timeout, and receipt age from the signed, versioned rollout policy bound to this broker release, bundle revision, and replica population, bind its version/digest to observations and probe replay, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Concept and owner boundary

This method owns already-issued workload certificates or SVIDs and the established mTLS sessions that can continue using them. Disabling a SPIRE registration, cloud workload binding, agent principal, or token-exchange subject stops new issuance under AID-E-001.003 but does not invalidate these artifacts. Rotating an issuer signing key or trust anchor belongs to AID-E-001.001 and is required only when that long-lived issuer credential is itself affected. Expiry is a bounded residual-risk deadline, not proof of active revocation.

Step 1: Revoke the exact certificate in a PKI that supports CRL or OCSP

The following commands are appropriate only for an OpenSSL-backed internal CA whose database is serialized and protected by an HSM-backed operator workflow. For an enterprise CA, invoke its supported revocation API and export the equivalent signed CRL or OCSP evidence.

#!/usr/bin/env bash
set -euo pipefail

if [ -z "$E001_ROLLOUT_TIMEOUT" ] || [ -z "$E001_TLS_PROBE_TIMEOUT_SECONDS" ]; then
  echo "required versioned runtime-profile timeout is missing" >&2; exit 2
fi
case "$E001_TLS_PROBE_TIMEOUT_SECONDS" in
  *[!0-9.]*|'') echo "TLS probe timeout must be a required positive numeric profile value" >&2; exit 2 ;;
esac

CA_CONFIG="/etc/pki/aidefend-openssl.cnf"
COMPROMISED_CERT="/incident/INC-2026-0417/compromised-svid.pem"
CA_CHAIN="/etc/pki/aidefend-ca-chain.pem"
CRL_STAGING="/srv/pki/crl/aidefend-current.crl.pem.new"
CRL_ACTIVE="/srv/pki/crl/aidefend-current.crl.pem"

# Serialize access to the OpenSSL CA database in the protected CA job.
flock /run/lock/aidefend-ca.lock \
  openssl ca -batch -config "$CA_CONFIG" \
    -revoke "$COMPROMISED_CERT" -crl_reason keyCompromise
flock /run/lock/aidefend-ca.lock \
  openssl ca -batch -config "$CA_CONFIG" -gencrl -out "$CRL_STAGING"

openssl crl -in "$CRL_STAGING" -noout -verify -CAfile "$CA_CHAIN"
install -o root -g root -m 0444 "$CRL_STAGING" "$CRL_ACTIVE"
sha256sum "$COMPROMISED_CERT" "$CRL_ACTIVE"

If the CA publishes OCSP, query the signed responder after revocation and configure each relying party to reject revoked or unverifiable status according to incident policy:

openssl ocsp \
  -issuer /etc/pki/aidefend-issuer.pem \
  -cert /incident/INC-2026-0417/compromised-svid.pem \
  -url https://ocsp.pki.example.org \
  -CAfile /etc/pki/aidefend-ca-chain.pem \
  -resp_text
Step 2: Make relying parties enforce the revocation state

Publish the signed deny bundle at every authorization and federation boundary. For Envoy-based relying parties, deliver the trusted CA and complete PEM CRL set through the protected SDS path. Envoy requires a CRL for each relevant CA in the chain unless the reviewed policy intentionally enables leaf-only checking.

common_tls_context:
  validation_context:
    trusted_ca:
      filename: /etc/envoy/tls/aidefend-ca-chain.pem
    crl:
      filename: /etc/envoy/tls/aidefend-current.crl.pem
    only_verify_leaf_cert_crl: false

Apply a principal-specific service-mesh deny before draining sessions:

apiVersion: security.istio.io/v1
kind: AuthorizationPolicy
metadata:
  name: deny-compromised-agent
  namespace: ai-production
spec:
  action: DENY
  rules:
    - from:
        - source:
            principals:
              - spiffe://example.org/agent/compromised-agent
kubectl apply -f deny-compromised-agent.yaml

# Reload trust/CRL material and terminate pooled mTLS sessions on the exact
# relying deployments identified by the incident connection inventory.
kubectl rollout restart deployment/payments-api deployment/model-gateway \
  -n ai-production
kubectl rollout status deployment/payments-api -n ai-production --timeout="$E001_ROLLOUT_TIMEOUT"
kubectl rollout status deployment/model-gateway -n ai-production --timeout="$E001_ROLLOUT_TIMEOUT"
Step 3: Prove issued-artifact denial and clean session establishment

From every reachable relying-party and federation boundary, present each signed affected certificate or credential and require a fresh handshake to fail. Enumerate the incident-defined preexisting connection population and require every affected connection ID to disappear after the authorized drain or restart. Then prove a separately attested clean certificate establishes only the intended least-privilege session. Archive exact certificate serials, CRL or OCSP responses where supported, relying-party deny or trust-state revisions, rollout receipts, and connection-drain observations.

set -euo pipefail
TARGET="payments-api.ai-production.svc.cluster.local:8443"

if timeout "$E001_TLS_PROBE_TIMEOUT_SECONDS" openssl s_client -brief -verify_return_error \
  -connect "$TARGET" \
  -cert /incident/INC-2026-0417/compromised-svid.pem \
  -key /incident/INC-2026-0417/compromised-svid-key.pem \
  -CAfile /etc/pki/aidefend-ca-chain.pem </dev/null; then
  echo "old credential was still accepted" >&2
  exit 1
fi

timeout "$E001_TLS_PROBE_TIMEOUT_SECONDS" openssl s_client -brief -verify_return_error \
  -connect "$TARGET" \
  -cert /run/clean-identity/replacement-svid.pem \
  -key /run/clean-identity/replacement-svid-key.pem \
  -CAfile /etc/pki/aidefend-ca-chain.pem </dev/null

Action: Accept this method only when the complete issued-certificate population is denied at every declared relying party, every affected established session is gone, and the clean replacement session succeeds. Missing serials, unenforced CRL or OCSP state, incomplete relying-party rollout, unknown connection state, or an old credential accepted anywhere blocks AID-E-001.002; it does not become a principal-disablement result.

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.002: Issued Token, Authentication Session & Lease Revocation | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service