AID-E-002: AI Runtime Process, Lease & Pod Eviction

Version 1.20261015 · Actionable technique

Revoke compromised execution leases and terminate confirmed-compromised AI model processes, agent runs, supervisor jobs, pods, or containerized runtime instances so active malicious execution stops and cannot reacquire work.

Pillars: infra, app

Lifecycle: response

Scope boundary

Owns revocation of active execution leases and termination of confirmed-compromised processes, agent runs, supervisor jobs, pods, containers, and run-local ephemeral execution. It does not remove the credentials, durable application state, or deployable artifacts that could recreate execution.

AID-E-001: AID-E-002 terminates active execution; AID-E-001 revokes the credentials, sessions, principals, and grants that authorize execution. A killed process can restart while authority remains, and revoked authority does not prove that the currently running workload stopped.

AID-E-005: AID-E-002 terminates active workers, processes, runs, jobs, pods, and containers; AID-E-005 removes durable application session, memory, registration, queue, and schedule records. Runtime termination does not delete durable records that can relaunch work, while deleting a queued or scheduled record does not terminate work that is already executing.

AID-E-003: AID-E-002 stops live execution instances; AID-E-003 quarantines or evicts malicious model, dataset, adapter, code, and configuration artifacts. A terminated workload may be recreated from the same malicious artifact, and artifact eviction does not itself stop an already running in-memory process.

Implementation guidance

AID-E-002-G001

Identify and terminate malicious AI model or inference server processes.
Prerequisites and target selection

Terminate one confirmed-compromised Linux process without allowing PID reuse or a host/PID-namespace mix-up to redirect the signal. The incident approver must sign a canonical JSON target containing host_machine_id, host_boot_id, orchestrator_node_uid, pid_namespace_inode, pid, start_time_ticks, uid, executable_device, executable_inode, executable_sha256, cgroup_sha256, incident_id, action_nonce, policy_sha256, and signed-policy-derived term_grace_seconds and kill_wait_seconds. Wildcards and caller-selected host paths are forbidden. Run the action through a root-owned local service on that bound host; do not accept a PID over a general-purpose remote shell.

Atomic Linux pidfd action
# File: /opt/aidefend/e002/evict_exact_process.py
from __future__ import annotations

import hashlib
import json
import math
import os
import select
import signal
import subprocess
import tempfile
import sys
import time
from pathlib import Path

TRUST_KEY = Path("/opt/aidefend/trust/e002-process-approver.pub")
NODE_UID_FILE = Path("/etc/aidefend/orchestrator-node-uid")
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["E002_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-002 runtime profile is invalid")
EXPECTED_FIELDS = {
    "schema_version", "host_machine_id", "host_boot_id",
    "orchestrator_node_uid", "pid_namespace_inode", "pid",
    "start_time_ticks", "uid", "executable_device", "executable_inode",
    "executable_sha256", "cgroup_sha256", "incident_id", "action_nonce",
    "policy_sha256", "term_grace_seconds", "kill_wait_seconds",
}

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 verify_signed_blob(path: Path, bundle: Path) -> dict:
    raw = verified_signed_bytes(
        path, bundle, TRUST_KEY, COMMAND_TIMEOUT_SECONDS, "process target"
    )
    value = strict_json(raw, "process target")
    canonical = (json.dumps(value, sort_keys=True, separators=(",", ":"),
                            allow_nan=False) + "\n").encode()
    if (
        not isinstance(value, dict)
        or raw != canonical
        or set(value) != EXPECTED_FIELDS
        or value.get("schema_version") != "aidefend.e002-process-target/v1"
    ):
        raise ValueError("target is not the signed canonical schema")
    return value

def sha256_file(path: str) -> str:
    digest = hashlib.sha256()
    with open(path, "rb", buffering=0) as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()

def process_identity(pid: int) -> dict:
    proc = Path("/proc") / str(pid)
    stat_text = (proc / "stat").read_text(encoding="utf-8")
    fields_after_comm = stat_text.rsplit(")", 1)[1].strip().split()
    exe_stat = os.stat(proc / "exe", follow_symlinks=True)
    return {
        "host_machine_id": Path("/etc/machine-id").read_text().strip(),
        "host_boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip(),
        "orchestrator_node_uid": NODE_UID_FILE.read_text().strip(),
        "pid_namespace_inode": os.stat(proc / "ns/pid").st_ino,
        "pid": pid,
        "start_time_ticks": int(fields_after_comm[19]),
        "uid": os.stat(proc).st_uid,
        "executable_device": exe_stat.st_dev,
        "executable_inode": exe_stat.st_ino,
        "executable_sha256": sha256_file(str(proc / "exe")),
        "cgroup_sha256": hashlib.sha256((proc / "cgroup").read_bytes()).hexdigest(),
    }

def evict(target: dict) -> dict:
    if type(target["pid"]) is not int or target["pid"] <= 0:
        raise ValueError("pid must be a positive integer")
    grace = target["term_grace_seconds"]
    kill_wait = target["kill_wait_seconds"]
    if (
        type(grace) not in (int, float)
        or type(kill_wait) not in (int, float)
        or not math.isfinite(grace)
        or not math.isfinite(kill_wait)
        or grace < 0
        or kill_wait <= 0
    ):
        raise ValueError("signed signal timing policy is invalid")
    identity_fields = set(process_identity(int(target["pid"])))
    expected = {key: target[key] for key in identity_fields}
    if any(target.get(key) in (None, "") for key in identity_fields):
        raise ValueError("target identity is incomplete")
    pid = int(target["pid"])
    before = process_identity(pid)
    if before != expected:
        raise RuntimeError("pre-action process identity differs from signed target")
    pidfd = os.pidfd_open(pid, 0)
    try:
        after_open = process_identity(pid)
        if after_open != before:
            raise RuntimeError("process identity changed while acquiring pidfd")
        poller = select.poll()
        poller.register(pidfd, select.POLLIN)
        signals_sent = []
        try:
            signal.pidfd_send_signal(pidfd, signal.SIGTERM)
            signals_sent.append("SIGTERM")
        except ProcessLookupError:
            pass
        if signals_sent and not poller.poll(int(grace * 1000)):
            try:
                signal.pidfd_send_signal(pidfd, signal.SIGKILL)
                signals_sent.append("SIGKILL")
            except ProcessLookupError:
                pass
            if not poller.poll(int(kill_wait * 1000)):
                raise RuntimeError("pidfd did not report process exit")
    finally:
        os.close(pidfd)
    return {"schema_version": "aidefend.e002-process-owner-receipt/v1",
            "control": "AID-E-002", "incident_id": target["incident_id"],
            "target_identity": expected, "action": "pidfd_term_then_kill",
            "signals_sent": signals_sent, "action_nonce": target["action_nonce"],
            "policy_sha256": target["policy_sha256"],
            "runtime_profile_version": RUNTIME_PROFILE_VERSION,
            "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
            "completed_at_unix_ns": time.time_ns()}

if __name__ == "__main__":
    manifest = verify_signed_blob(Path(sys.argv[1]), Path(sys.argv[2]))
    print(json.dumps(evict(manifest), sort_keys=True, separators=(",", ":")))

The service must write the owner receipt to append-only evidence storage and sign it with the action-service identity. kill_wait_seconds and the grace period come from the signed policy; a framework example must not supply universal incident timing.

Independent verification

A different read-only identity on the same bound host rechecks machine ID, boot ID, node UID, and PID namespace. It proves that no live /proc entry has the signed tuple of start ticks, executable device/inode/digest, UID, and cgroup digest. If the numeric PID now belongs to another identity, report that fact but never signal it. Preserve the signed target, cosign verification output, before/after identities, pidfd action receipt, verifier receipt, tool versions, and incident correlation in immutable storage.

Action: authorize and terminate only the exact signed process identity through pidfd, then accept the control only after the independent identity-bound absence readback succeeds.

AID-E-002-G002

Terminate hijacked AI agent runtime instances immediately.
Runtime scope

Stop one hijacked agent runtime through its authenticated supervisor. This step handles the live worker instance and termination acknowledgement only. Durable lease tombstoning and anti-reacquisition belong to the next guidance; pod/process fallback is separate evidence and cannot make this supervisor action pass.

Production implementation
  1. The incident approver signs canonical JSON binding tenant, agent ID, run ID, worker-instance ID, supervisor-cluster ID, runtime image digest, incident ID, action nonce, policy digest, and manifest expiry. Exact IDs are required; URLs, credentials, and wildcard selectors are forbidden.
  2. The SOAR worker re-fetches the live run through a pinned supervisor endpoint selected from trusted configuration, authenticates with workload mTLS, and compares every identity field and resource version with the signed target.
  3. Call the supervisor's idempotent termination API with the signed action nonce and an If-Match resource version. The supervisor must durably change the exact worker to TERMINATING, stop new tool/model calls, cancel child subprocesses, and return a signed acknowledgement containing the same identity and monotonically increasing event sequence.
  4. A separate read-only verifier queries the supervisor quorum and worker-heartbeat store after the policy-defined convergence window. It requires the exact worker instance to be terminal and requires zero fresh heartbeats or tool/model invocations with an event sequence after the termination acknowledgement.

The endpoint in this illustrative request contract comes from trusted service configuration.

POST /v1/runtime-evictions HTTP/1.1
Host: runtime-control.example.internal
If-Match: "rv-01J2Y5KH7VYQ7K4X6N3E9T2M8A"
Idempotency-Key: "e002-01J2Y5M7H8R3C9ZK4TQ6V1NW2P"
Content-Type: application/json

{"agent_id":"support-agent-prod-017","agent_run_id":"run-01J2Y4ZWV8C6M0BQ9F7S3K2H1D",
 "worker_instance_id":"worker-4f9c7a12","incident_id":"INC-2026-0718-0042",
 "expected_image_digest":"sha256:9e7d3a4c1b28f506d87211c685f59ac73851ee3c309f82de4c76f0ab1d29e546","command":"TERMINATE_NOW"}

Store the signed authorization, pre-action read, mTLS peer identity, request/response transcript, supervisor event-log segment, heartbeat/tool-call readback, and verifier signature in immutable evidence storage. A publish-only message, unacknowledged queue write, or client timeout is not proof of termination.

Action: issue one authenticated, idempotent, resource-version-guarded termination to the trusted supervisor for the exact signed worker instance and independently prove its terminal state and absence of post-acknowledgement activity.

AID-E-002-G003

Tombstone the evicted agent run and prevent its runtime lease from being reacquired.
Lease-fencing goal

Make an evicted agent run unable to acquire or renew runtime work. A delete of the current lease is insufficient: every acquisition, renewal, heartbeat, and result commit must be fenced by a durable tombstone and a monotonically increasing epoch.

Atomic lease-store design

The incident approver signs exact tenant, run ID, current lease owner, current resource version, incident ID, and policy digest. The runtime-control service verifies that signature, re-reads the current record, and executes a transactional store procedure that increments fence_epoch, sets evicted=true, records the incident and action nonce, deletes the current lease, and appends an audit event. The procedure returns the prior result for the same nonce and exact incident/new-version tuple without incrementing the epoch or appending a second event; reuse with changed bytes fails closed. A different nonce cannot re-evict an already tombstoned run. The procedure aborts on any target/version mismatch. Do not use Redis Pub/Sub as the durable decision record.

-- Redis Lua contract; KEYS are derived from trusted prefixes plus the signed exact run ID.
local current = redis.call('HGET', KEYS[1], 'resource_version')
local prior_nonce = redis.call('HGET', KEYS[1], 'action_nonce')
if prior_nonce then
  if prior_nonce == ARGV[3] then
    local prior_incident = redis.call('HGET', KEYS[1], 'incident_id')
    local prior_evicted = redis.call('HGET', KEYS[1], 'evicted')
    local prior_epoch = redis.call('HGET', KEYS[1], 'fence_epoch')
    if prior_incident ~= ARGV[2] or prior_evicted ~= 'true'
        or current ~= ARGV[4] or not prior_epoch then
      return redis.error_reply('NONCE_REUSE_MISMATCH')
    end
    return {tonumber(prior_epoch), current, 'ALREADY_APPLIED'}
  end
  if redis.call('HGET', KEYS[1], 'evicted') == 'true' then
    return redis.error_reply('RUN_ALREADY_EVICTED')
  end
end
if current ~= ARGV[1] then
  return redis.error_reply('RESOURCE_VERSION_MISMATCH')
end
local epoch = redis.call(
  'HINCRBY',
  KEYS[1],
  'fence_epoch',
  1
)
redis.call(
  'HSET',
  KEYS[1],
  'evicted',
  'true',
  'incident_id',
  ARGV[2],
  'action_nonce',
  ARGV[3],
  'resource_version',
  ARGV[4]
)
redis.call('DEL', KEYS[2])
redis.call(
  'XADD',
  KEYS[3],
  '*',
  'event',
  'RUN_EVICTED',
  'run_id',
  ARGV[5],
  'fence_epoch',
  tostring(epoch),
  'incident_id',
  ARGV[2],
  'action_nonce',
  ARGV[3]
)
return {
  epoch,
  ARGV[4],
  'APPLIED'
}
Enforce the fence on every work path

The lease-acquire/renew procedure must atomically read the same run record and reject when evicted=true; otherwise it returns the current fence epoch. Workers include tenant, run ID, worker ID, and epoch on heartbeat, tool/model call authorization, and result commit. Each sink rejects a stale epoch or tombstoned run. A notification may accelerate shutdown but is not enforcement.

Independent verification

A separately credentialed verifier reads the primary/quorum, proves the signed exact run is tombstoned, the old lease is absent, the epoch is greater than the pre-action epoch, and a controlled acquire/renew/commit attempt with both the old and a newly requested lease is denied. Preserve signed input, pre/post records, transaction result, append-only stream entry, negative-test transcripts, store topology/version, and verifier receipt.

Action: atomically tombstone and fence the exact run, revoke its current lease, enforce that state in every acquisition and commit path, and independently replay denial before accepting the control.

AID-E-002-G004

Forcefully delete or quarantine compromised Kubernetes pods/containers.

Runtime policy: Load termination grace, deletion-observation deadline, and poll cadence from the signed, versioned workload policy bound to this cluster UID, pod UID, controller disposition, and incident, bind its version/digest to Kubernetes action/readback evidence, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Correct controller semantics

Deleting a compromised pod does not prove that a clean pod will replace it. A Deployment, ReplicaSet, StatefulSet, DaemonSet, Job, or custom controller normally recreates from its current template, which may be the same compromised template. Before deletion, bind the action to a signed exact pod identity and separately dispose of its owner controller.

Authorized exact target

The signed canonical manifest must contain cluster UID, namespace UID/name, pod name/UID/resourceVersion, node UID, service-account UID, runtime class, every container and init-container image ID/digest, owner-reference kind/name/UID, incident ID, action nonce, and controller disposition. The disposition is either HELD with a signed controller receipt proving reconciliation is suspended or replicas are safely zero, or VERIFIED_CLEAN_TEMPLATE with an independent receipt binding the controller UID/generation/template digest and every approved immutable image digest. A missing owner reference or unmanaged pod is recorded explicitly.

Render signed_grace_period_seconds from the signature-verified response policy and validate it as a non-negative integer before constructing the UID/resourceVersion-preconditioned request. The concrete request below is a rendered fixture; production values must be copied from the verified manifest and pre-action GET, not reused from the fixture.

# Kubernetes API request; use an mTLS service account scoped to this namespace and verb.
DELETE /api/v1/namespaces/ai-prod/pods/support-agent-7d4c6b8d9f-k2m8q
Content-Type: application/json
Idempotency-Key: e002-01J2Y5M7H8R3C9ZK4TQ6V1NW2P

{"apiVersion":"v1","kind":"DeleteOptions","gracePeriodSeconds":15,
 "propagationPolicy":"Foreground",
 "preconditions":{"uid":"4e04b920-9fd7-4f2f-8b93-5f8d7f39cb60","resourceVersion":"184732991"}}

The responder first performs a quorum-backed GET and compares all signed fields. The delete must use both UID and resourceVersion preconditions; a name-only or label-wide delete is prohibited. Capture policy-defined volatile forensics before deletion when doing so does not extend harmful execution beyond the signed response policy.

Independent verification

A different read-only service account proves the exact pod UID is absent from the API and node runtime. For HELD, it proves no replacement owned by that controller UID becomes Ready. For VERIFIED_CLEAN_TEMPLATE, it proves any replacement is owned by the exact controller UID/generation and runs only the approved template and immutable image digests; mere pod recreation is not evidence of cleanliness. Preserve authorization, pre-action object, controller receipt, deletion audit event, API/node readbacks, replacement population, and signed verifier result.

Action: hold or independently validate the exact owner controller, delete only the signed pod UID with API preconditions, and verify both target absence and the authorized replacement condition.

AID-E-002-G005

Independently prove runtime absence, requiring PID checks to run on the explicitly bound host, node, and PID namespace.
Independent runtime-absence verifier

Run this verification with a read-only identity independent from the eviction action. Its signed verification manifest enumerates every applicable target class: exact process identity, agent run/worker/fence epoch, and pod UID/controller disposition. An absent class is marked NOT_APPLICABLE with its architectural prerequisite; it is never silently omitted.

Process verification without PID ambiguity

Bind the verifier to host machine ID, boot ID, orchestrator node UID, and PID namespace inode. Search the complete visible /proc population for the signed process tuple: start-time ticks, UID, executable device/inode/SHA-256, and cgroup SHA-256. The original numeric PID being empty is insufficient, and a reused PID must be reported as a different identity rather than killed or counted as the original target.

Run and pod verification

Query the lease-store primary/quorum for the exact tenant/run. Require the signed tombstone, increased fence epoch, zero lease, zero fresh worker heartbeats, and controlled rejection of acquisition/commit. Query both Kubernetes API and bound node runtime for the exact pod UID. Apply the signed HELD or VERIFIED_CLEAN_TEMPLATE controller condition from the pod-eviction guidance across the complete replacement population.

Evidence and replay

Produce canonical JSON containing signed manifest digest, verifier workload identity, tool versions, trusted endpoints, observation times, source revisions, per-target observations, population counts, and raw-evidence digests. Sign the receipt and retain raw API/store/proc snapshots in immutable storage. The action service cannot sign this verifier result.

Action: independently enumerate and verify every applicable exact runtime target and accept E-002 only when each applicable target reaches its own absence/fence/controller condition.

AID-E-002-G006

Reconcile the exact evicted runtime population against independent process, pod, supervisor, and lease observations.
When the local runtime audit applies

Use this final E-002 path after G001-G004 have targeted one or more model processes, agent runs, supervisor jobs, execution leases, pods, or containers. It audits only the runtime objects in the incident-bound E-002 population. Durable application state, credentials, artifacts, and broader fleet recovery remain with their owning controls.

Observe absence at each authoritative runtime plane

Resolve every target to its stable host and PID namespace, supervisor run ID, lease key, scheduler job identity, or Kubernetes pod UID. Query those authorities with read-only identities independent of the eviction actor. Each observation must bind the incident, target class and ID, authority revision, observation source, and whether execution is still active or a lease can be reacquired. A reused PID, recreated pod name, scaled replacement, stale supervisor cache, unavailable node, missing namespace binding, or incomplete lease readback cannot be treated as absence.

Executable exact-population audit
# File: eviction/audit_runtime_absence.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/e002-runtime-scope.pub")
OBSERVED_TRUST_KEY = Path("/opt/aidefend/trust/e002-runtime-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["E002_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-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: object) -> bytes:
    return (
        json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False)
        + "\n"
    ).encode("utf-8")


def load(path: Path, bundle: Path, trust_key: Path) -> tuple[list[dict], bytes]:
    raw = verified_signed_bytes(
        path, bundle, trust_key, COMMAND_TIMEOUT_SECONDS, str(path)
    )
    value = strict_json(raw, str(path))
    if (
        not isinstance(value, list)
        or not value
        or raw != canonical_json_bytes(value)
        or any(not isinstance(row, dict) for row in value)
    ):
        raise ValueError(f"{path}: signed canonical nonempty target list required")
    return value, raw


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


def main(
    expected_path: Path,
    expected_bundle: Path,
    observed_path: Path,
    observed_bundle: Path,
) -> None:
    expected, expected_bytes = load(
        expected_path, expected_bundle, EXPECTED_TRUST_KEY
    )
    observed, observed_bytes = load(
        observed_path, observed_bundle, OBSERVED_TRUST_KEY
    )
    expected_by_id = {identity(row): row for row in expected}
    observed_by_id = {identity(row): row for row in observed}
    if len(expected_by_id) != len(expected) or len(observed_by_id) != len(observed):
        raise ValueError("duplicate runtime target")
    if set(expected_by_id) != set(observed_by_id):
        raise RuntimeError("observed runtime population differs from expected")
    for target, wanted in expected_by_id.items():
        actual = observed_by_id[target]
        if actual.get("incident_id") != wanted.get("incident_id"):
            raise RuntimeError(f"incident binding differs: {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"runtime 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"runtime evidence digest is invalid: {target}")
        if actual.get("observation_complete") is not True:
            raise RuntimeError(f"runtime observation incomplete: {target}")
        if actual.get("active") is not False:
            raise RuntimeError(f"runtime target remains active: {target}")
        if actual.get("lease_applicable") is True and actual.get("lease_reacquirable") is not False:
            raise RuntimeError(f"runtime lease can be reacquired: {target}")
    print(json.dumps({
        "schema_version": "aidefend.e002-runtime-reconciliation/v1",
        "control": "AID-E-002",
        "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": "absent",
    }, sort_keys=True, separators=(",", ":")))


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

Action: create and sign the canonical expected population before eviction, collect and separately sign each canonical authoritative observation after eviction under the pinned read-only verifier key, and require an exact set match with no active or reacquirable member. Store this audit event with the incident record; it is not a cross-control receipt aggregator.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-E-002: AI Runtime Process, Lease & Pod Eviction | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service