Scope boundary

Owns write authorization and enforcement at protected file-backed agent identity, primary-memory, and configuration surfaces through OS restrictions, a controlled writer, immutable baselines, reviewed diffs, rollback, and tamper evidence. Database-backed and vector-backed memory use the other AID-I-004 controls.

AID-I-004.003: AID-I-004.006 controls who may write protected file-backed identity and state; AID-I-004.003 proves cryptographic integrity of persistent-memory records after an authorized write. Write authorization and post-write integrity are separate claims and require separate enforcement and verification evidence.

AID-H-018.007: AID-I-004.006 enforces the protected file-write boundary; AID-H-018.007 determines whether an installed skill may request that capability under its permission manifest. A skill permission is an input to the file control, not proof that the exact write was allowed, mediated, or durably recorded.

Implementation guidance

• AID-I-004.006-G001: Default-deny writes to protected file-backed identity/state surfaces, and enforce updates through a controlled writer plus OS-level immutability.
• AID-I-004.006-G002: Verify baseline integrity of protected logical resources at startup before loading any file-backed identity/state content.
• AID-I-004.006-G003: Run diff checks on protected logical resources after skill install / update / uninstall, but only accept changes that originate from approved change context.
• AID-I-004.006-G004: Before a protected write, require an mTLS-authenticated signed scanner result and a one-time approval bound to the verified external principal, tenant, resource, action, and exact content digest.
• AID-I-004.006-G005: Restore only from a signed, version-addressed known-good snapshot under a trusted retention-protected root, using a one-time rollback approval bound to the verified principal and exact manifest.
• AID-I-004.006-G006: Emit structured tamper / write-attempt audit events for every protected-resource modification attempt, and correlate them to ticket, writer, and rollback context.

AID-I-004.006-G001

Concept

This sub-technique protects file-backed high-trust persistent identity/state surfaces, not every possible memory backend. Use reserved logical resource IDs as the canonical policy layer, and treat concrete files only as platform-specific backing implementations.

Note: The logical resource IDs and backing filenames below are representative examples. Different agent platforms will use different concrete files (e.g., persona configs, agent manifests, persistent instruction files). The principle — protect high-trust persistent identity/state surfaces — applies regardless of platform.

protected_resources:
  - agent.identity.soul           # commonly backed by SOUL.md or equivalent
  - agent.memory.primary          # commonly backed by MEMORY.md or equivalent
  - agent.identity.agents_config  # commonly backed by AGENTS.md or equivalent

Boundary with AID-H-018.007: a manifest may declare whether a skill is authorized to request protected-resource writes, but that manifest declaration alone must never be the write path. This sub-technique owns the persistence surface protection itself: immutable mounts, controlled-writer gates, baseline verification, diff checks, rollback, and tamper-audit.

Deployment Pattern A — Immutable baseline mode

Use this when the protected surface should almost never change in production.

• Store the backing files on a read-only mount or immutable path.
• Use a signed / approved baseline artifact for deployment.
• Changes happen only through formal rollout / replacement, not in-place writes.

Deployment Pattern B — Controlled mutable mode

Use this only when a protected file-backed surface truly requires approved updates.

• Do not allow direct skill file writes.
• Expose a dedicated controlled-writer API / service as the only legal update path.
• Require authenticated caller identity, authorization policy, audit logging, and pre-write snapshotting.

OS-level enforcement guidance

Linux-hosted VM / bare-metal environments may use chattr +i or equivalent immutable flag where supported. Containerized / orchestrated environments should instead rely primarily on read-only mounts, dedicated protected volumes, runtime file-write interception, and signed baseline artifacts. Do not assume chattr +i works uniformly across all container runtimes, overlay filesystems, or non-Linux platforms.

Controlled writer requirements

• Single responsibility: only updates protected identity/state resources.
• Strong authentication: accepts requests only from trusted workload identities.
• Strong authorization: requires approved policy / ticket / review context.
• Auditable: records requester, diff summary, signer / approver, and target logical resource.
• Recoverable: stores last-known-good snapshot before commit.

Action: Deny direct write syscalls to protected backing files by default. Use reserved logical resource IDs for policy, and map them to concrete backing files only inside the controlled-writer / enforcement layer.

AID-I-004.006-G002

Concept

Every agent startup must verify that each protected logical resource still matches a known-good approved baseline before the agent loads it into runtime state. This protects against out-of-band tampering that occurred while the agent was offline.

Recommended model

protected_resource_map:
  agent.identity.soul:
    backing_paths:
      - /var/lib/agent/SOUL.md
    approved_sha256: "sha256-from-signature-verified-release-manifest"
  agent.memory.primary:
    backing_paths:
      - /var/lib/agent/MEMORY.md
    approved_sha256: "sha256-from-signature-verified-release-manifest"
  agent.identity.agents_config:
    backing_paths:
      - /var/lib/agent/AGENTS.md
    approved_sha256: "sha256-from-signature-verified-release-manifest"

Startup logic

1. Resolve each logical resource to its platform-specific backing path.
2. Canonicalize the path and read the file in read-only mode.
3. Compute SHA-256.
4. Compare against the approved baseline hash stored in a secure metadata store or KMS-protected record.
5. Any mismatch should fail closed: refuse startup of the affected agent instance and emit a structured tamper event.

This complements AID-I-004.003: that sibling verifies signed entries / content integrity; this step verifies the whole protected file-backed surface against a known-good baseline state.

Action: Treat startup baseline verification as a mandatory boot gate for any agent using file-backed identity/state resources.

Verify safely

re-read files through an external verifier, replay signature and path resolution, and test symlink, bind-mount, replacement, missing-file, and stale-manifest cases.

AID-I-004.006-G003

Security boundary

Run this method around every skill install, update, and removal that can share a protected identity/state surface. The lifecycle event is only a trigger. It never authorizes a change. A non-empty allowed diff must come from a separately signed change manifest issued by the approved change service and bound to the exact lifecycle event, skill artifact, protected-resource map, asset, and policy version. Capture the complete protected population through a filesystem snapshot or controlled-writer freeze so concurrent writes cannot produce a torn comparison.

Reference implementation

# File: controls/protected_surface_diff.py
# Dependencies: Python 3.12+, PyJWT>=2.9,<3, cryptography>=43,<45
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from typing import Mapping, Sequence, Callable


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


@dataclass(frozen=True)
class AtomicSnapshotProvider:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    capture: Callable[[Sequence[Mapping[str, object]]], Sequence[Mapping[str, object]]]


@dataclass(frozen=True)
class DurableRunStore:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    create_before: Callable[[str, Mapping[str, object]], None]
    take_before_for_update: Callable[[str], Mapping[str, object]]
    finish: Callable[[str, Mapping[str, object]], None]


@dataclass(frozen=True)
class EvidenceClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append: Callable[[Mapping[str, object]], Mapping[str, object]]
    read: Callable[[str], Mapping[str, object]]


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


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


@dataclass(frozen=True)
class Dependencies:
    verifier: SignedDocumentVerifier
    snapshots: AtomicSnapshotProvider
    runs: DurableRunStore
    evidence: EvidenceClient
    startup_gate: StartupGate
    rollback: RollbackRequester


def canonical_digest(value: object) -> str:
    encoded = json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def normalize_snapshot(
    raw: Sequence[Mapping[str, object]],
    resource_map: Sequence[Mapping[str, object]],
) -> tuple[dict[str, object], ...]:
    expected_ids = {str(item["resource_id"]) for item in resource_map}
    if len(expected_ids) != len(resource_map):
        raise ValueError("resource map contains duplicate IDs")

    normalized: list[dict[str, object]] = []
    seen: set[str] = set()
    for item in raw:
        resource_id = str(item["resource_id"])
        if resource_id not in expected_ids or resource_id in seen:
            raise ValueError("snapshot has an extra or duplicate resource")
        seen.add(resource_id)
        present = bool(item["present"])
        row: dict[str, object] = {
            "resource_id": resource_id,
            "present": present,
            "canonical_path": str(item["canonical_path"]),
        }
        if present:
            required = {
                "content_sha256", "size_bytes", "device", "inode",
                "mode", "uid", "gid", "link_count",
            }
            if not required.issubset(item):
                raise ValueError("snapshot is missing file identity or digest")
            if int(item["link_count"]) != 1:
                raise ValueError("hard-linked protected file is not accepted")
            if int(item["mode"]) & 0o022:
                raise ValueError("protected file is group/world writable")
            row.update({key: item[key] for key in sorted(required)})
        normalized.append(row)

    if seen != expected_ids:
        raise ValueError("snapshot does not cover the complete resource map")
    return tuple(sorted(normalized, key=lambda row: str(row["resource_id"])))


def logical_diff(
    before: Sequence[Mapping[str, object]],
    after: Sequence[Mapping[str, object]],
) -> tuple[dict[str, object], ...]:
    before_by_id = {str(item["resource_id"]): item for item in before}
    after_by_id = {str(item["resource_id"]): item for item in after}
    if set(before_by_id) != set(after_by_id):
        raise ValueError("before/after populations differ")
    changed: list[dict[str, object]] = []
    compared = (
        "present", "content_sha256", "size_bytes", "mode", "uid", "gid",
        "device", "inode", "link_count", "canonical_path",
    )
    for resource_id in sorted(before_by_id):
        old = before_by_id[resource_id]
        new = after_by_id[resource_id]
        delta = {
            "resource_id": resource_id,
            "before": {key: old.get(key) for key in compared},
            "after": {key: new.get(key) for key in compared},
        }
        if delta["before"] != delta["after"]:
            changed.append(delta)
    return tuple(changed)


def begin(
    deps: Dependencies,
    *,
    lifecycle_event_jws: str,
    resource_map_jws: str,
) -> dict[str, object]:
    event = deps.verifier.verify(
        lifecycle_event_jws,
        typ="skill-lifecycle-event+jwt",
        audience="protected-surface-diff",
    )
    resource_doc = deps.verifier.verify(
        resource_map_jws,
        typ="protected-resource-map+jwt",
        audience="protected-surface-diff",
    )
    if event["resource_map_sha256"] != canonical_digest(resource_doc):
        raise ValueError("lifecycle event/resource-map binding mismatch")
    before = normalize_snapshot(
        deps.snapshots.capture(resource_doc["resources"]),
        resource_doc["resources"],
    )
    record = {
        "schema_version": 1,
        "event_id": event["event_id"],
        "asset_id": event["asset_id"],
        "operation": event["operation"],
        "skill_artifact_sha256": event["skill_artifact_sha256"],
        "policy_sha256": event["policy_sha256"],
        "resource_map_sha256": event["resource_map_sha256"],
        "resource_count": len(before),
        "before_population_sha256": canonical_digest(before),
        "before": before,
    }
    deps.runs.create_before(str(event["event_id"]), record)
    return record


def finish(
    deps: Dependencies,
    *,
    lifecycle_event_jws: str,
    resource_map_jws: str,
    approved_change_jws: str | None,
) -> dict[str, object]:
    event = deps.verifier.verify(
        lifecycle_event_jws,
        typ="skill-lifecycle-event+jwt",
        audience="protected-surface-diff",
    )
    resource_doc = deps.verifier.verify(
        resource_map_jws,
        typ="protected-resource-map+jwt",
        audience="protected-surface-diff",
    )
    before_record = deps.runs.take_before_for_update(str(event["event_id"]))
    bindings = (
        "asset_id", "operation", "skill_artifact_sha256",
        "policy_sha256", "resource_map_sha256",
    )
    if any(before_record[key] != event[key] for key in bindings):
        raise ValueError("finish event differs from recorded begin event")
    if event["resource_map_sha256"] != canonical_digest(resource_doc):
        raise ValueError("finish resource-map binding mismatch")

    after = normalize_snapshot(
        deps.snapshots.capture(resource_doc["resources"]),
        resource_doc["resources"],
    )
    observed = logical_diff(before_record["before"], after)

    if approved_change_jws is None:
        expected: tuple[Mapping[str, object], ...] = tuple()
        approval_digest = None
        approval_id = None
    else:
        approval = deps.verifier.verify(
            approved_change_jws,
            typ="protected-change-approval+jwt",
            audience="protected-surface-diff",
        )
        approval_bindings = (
            "event_id", "asset_id", "operation", "skill_artifact_sha256",
            "policy_sha256", "resource_map_sha256",
        )
        if any(approval[key] != event[key] for key in approval_bindings):
            raise ValueError("approved change is not bound to this event")
        expected = tuple(approval["changes"])
        approval_digest = hashlib.sha256(
            approved_change_jws.encode("ascii")
        ).hexdigest()
        approval_id = approval["approval_id"]

    matched = canonical_digest(observed) == canonical_digest(expected)
    preliminary = {
        "schema_version": 1,
        "control_id": "AID-I-004.006",
        "guidance_ref": "AID-I-004.006-G003",
        "event_id": event["event_id"],
        "asset_id": event["asset_id"],
        "policy_sha256": event["policy_sha256"],
        "resource_map_sha256": event["resource_map_sha256"],
        "resource_count": len(after),
        "before_population_sha256": before_record["before_population_sha256"],
        "after_population_sha256": canonical_digest(after),
        "observed_diff_sha256": canonical_digest(observed),
        "expected_diff_sha256": canonical_digest(expected),
        "approval_id": approval_id,
        "approval_jws_sha256": approval_digest,
        "outcome": "PASS" if matched else "FAIL",
    }
    receipt = deps.evidence.append(preliminary)
    readback = deps.evidence.read(str(receipt["receipt_id"]))
    if readback["event_sha256"] != receipt["event_sha256"]:
        raise RuntimeError("evidence readback mismatch")

    gate_receipt = None
    rollback_receipt = None
    if not matched:
        gate_receipt = deps.startup_gate.block(
            str(event["asset_id"]), str(receipt["receipt_id"])
        )
        rollback_receipt = deps.rollback.request_i004006_5(
            str(event["event_id"]), str(receipt["receipt_id"])
        )

    final = {
        **preliminary,
        "evidence_receipt_id": receipt["receipt_id"],
        "startup_gate_receipt": gate_receipt,
        "rollback_request_receipt": rollback_receipt,
    }
    deps.runs.finish(str(event["event_id"]), final)
    return final

Operational sequence

1. Verify the signed event and signed resource map, acquire an atomic filesystem snapshot or controlled-writer freeze, and call begin() before lifecycle execution.
2. Execute the skill lifecycle action without granting it protected-write authority.
3. Call finish() with no approval for the normal zero-diff case, or with a separately issued signed exact-change manifest.
4. Require the resulting receipt before agent startup/use. On mismatch, keep startup blocked and request AID-I-004.006-G005; this step does not claim that sibling rollback action or PASS.

Adapters must validate JWS issuer, audience, type, algorithm, key status, required claims, expiry, replay, and monotonic policy version; the snapshot adapter must expose a consistent, externally acquired population including path, device/inode, ownership, mode, link count, size, and content digest. A crash between begin() and finish() remains an incomplete run and blocks startup.

AID-I-004.006-G004

Keep the action safely bounded

Unknown resources, verification failure, stale results, replay, and service unavailability deny the write.

mTLS scanner and action-bound approval gate

# File: protected_writer/write_gate.py
# Dependencies: httpx[http2]>=0.28,<1, PyJWT>=2.9,<3,
# cryptography>=43,<46, boto3>=1.35,<2
from __future__ import annotations

import base64
import hashlib
import ipaddress
import json
import math
import os
import re
import ssl
import tempfile
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlsplit

import boto3
import httpx
import jwt
from botocore.exceptions import ClientError


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


def positive_policy_integer(name: str) -> int:
    raw = os.environ[name]
    if not raw.isascii() or not raw.isdecimal():
        raise RuntimeError(name + " must be an unsigned decimal integer")
    value = int(raw)
    if value <= 0:
        raise RuntimeError(name + " must be positive")
    return value


def require_policy_digest(value: str, field: str) -> None:
    if (
        SHA256_BINDING.fullmatch(value) is None
        or value == "sha256:" + "0" * 64
    ):
        raise RuntimeError(field + " must be a nonzero SHA-256 binding")


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


APPROVAL_KEY = Path(os.environ["APPROVAL_PUBLIC_KEY_PATH"]).read_bytes()
SCANNER_KEY = Path(os.environ["SCANNER_PUBLIC_KEY_PATH"]).read_bytes()
APPROVAL_ISSUER = os.environ["APPROVAL_ISSUER"]
SCANNER_ISSUER = os.environ["SCANNER_ISSUER"]
REPLAY_TABLE = os.environ["WRITE_APPROVAL_REPLAY_TABLE"]
SCANNER_ORIGIN = canonical_https_origin(
    os.environ["VERIFIED_PROTECTED_WRITE_SCANNER_ORIGIN"]
)
SCANNER_TOTAL_TIMEOUT_SECONDS = float(
    os.environ["VERIFIED_SCANNER_TOTAL_TIMEOUT_SECONDS"]
)
SCANNER_CONNECT_TIMEOUT_SECONDS = float(
    os.environ["VERIFIED_SCANNER_CONNECT_TIMEOUT_SECONDS"]
)
JWT_LEEWAY_SECONDS = float(
    os.environ["VERIFIED_PROTECTED_WRITE_JWT_LEEWAY_SECONDS"]
)
RUNTIME_POLICY_VERSION = os.environ[
    "VERIFIED_PROTECTED_WRITE_RUNTIME_POLICY_VERSION"
]
RUNTIME_POLICY_SHA256 = os.environ[
    "VERIFIED_PROTECTED_WRITE_RUNTIME_POLICY_SHA256"
]
RESPONSE_POLICY_VERSION = os.environ[
    "VERIFIED_PROTECTED_WRITE_RESPONSE_POLICY_VERSION"
]
RESPONSE_POLICY_SHA256 = os.environ[
    "VERIFIED_PROTECTED_WRITE_RESPONSE_POLICY_SHA256"
]
PROTECTED_WRITE_MAX_CANDIDATE_BYTES = positive_policy_integer(
    "VERIFIED_PROTECTED_WRITE_MAX_CANDIDATE_BYTES"
)
SCANNER_RESPONSE_MAX_BYTES = positive_policy_integer(
    "VERIFIED_SCANNER_RESPONSE_MAX_BYTES"
)
SCANNER_RESPONSE_CHUNK_BYTES = positive_policy_integer(
    "VERIFIED_SCANNER_RESPONSE_CHUNK_BYTES"
)
if (
    not math.isfinite(SCANNER_TOTAL_TIMEOUT_SECONDS)
    or SCANNER_TOTAL_TIMEOUT_SECONDS <= 0
    or not math.isfinite(SCANNER_CONNECT_TIMEOUT_SECONDS)
    or SCANNER_CONNECT_TIMEOUT_SECONDS <= 0
    or not math.isfinite(JWT_LEEWAY_SECONDS)
    or JWT_LEEWAY_SECONDS < 0
    or POLICY_VERSION.fullmatch(RUNTIME_POLICY_VERSION) is None
    or POLICY_VERSION.fullmatch(RESPONSE_POLICY_VERSION) is None
    or SCANNER_RESPONSE_CHUNK_BYTES > SCANNER_RESPONSE_MAX_BYTES
):
    raise RuntimeError("verified protected-write policy is invalid")
require_policy_digest(RUNTIME_POLICY_SHA256, "runtime_policy_sha256")
require_policy_digest(RESPONSE_POLICY_SHA256, "response_policy_sha256")

scanner_tls = ssl.create_default_context(
    ssl.Purpose.SERVER_AUTH,
    cafile=os.environ["SCANNER_CA_BUNDLE"],
)
scanner_tls.minimum_version = ssl.TLSVersion.TLSv1_2
scanner_tls.load_cert_chain(
    certfile=os.environ["WRITER_CLIENT_CERT"],
    keyfile=os.environ["WRITER_CLIENT_KEY"],
)

# Root-owned configuration; no request may supply a path.
RESOURCE_BINDINGS = {
    ("tenant-a", "agent.identity.soul"): Path("/var/lib/agent/SOUL.md"),
    ("tenant-a", "agent.memory.primary"): Path("/var/lib/agent/MEMORY.md"),
    ("tenant-a", "agent.identity.agents_config"): Path("/var/lib/agent/AGENTS.md"),
}
ddb = boto3.client("dynamodb")


@dataclass(frozen=True)
class VerifiedPrincipal:
    issuer: str
    tenant_id: str
    subject: str


@dataclass(frozen=True)
class ProtectedWriteRequest:
    resource_id: str
    content: bytes
    approval_jws: str


class ProtectedWriteRejected(Exception):
    pass


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


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


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

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

    payload = bytearray()
    read_size = min(
        SCANNER_RESPONSE_CHUNK_BYTES,
        SCANNER_RESPONSE_MAX_BYTES + 1,
    )
    for chunk in response.iter_raw(chunk_size=read_size):
        if time.monotonic() > deadline:
            raise ProtectedWriteRejected(
                "scanner response exceeded signed total deadline"
            )
        if not chunk:
            continue
        remaining = SCANNER_RESPONSE_MAX_BYTES + 1 - len(payload)
        payload.extend(chunk[:remaining])
        if len(chunk) > remaining or len(payload) > SCANNER_RESPONSE_MAX_BYTES:
            raise ProtectedWriteRejected(
                "scanner response exceeds signed response byte limit"
            )
    if time.monotonic() > deadline:
        raise ProtectedWriteRejected(
            "scanner response exceeded signed total deadline"
        )
    if not payload:
        raise ProtectedWriteRejected("scanner response is empty")
    if declared_bytes is not None and declared_bytes != len(payload):
        raise ProtectedWriteRejected(
            "scanner Content-Length differs from observed response bytes"
        )

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


def verify_approval(token, principal, resource_id, digest):
    header = jwt.get_unverified_header(token)
    if (
        header.get("alg") != "EdDSA"
        or header.get("typ") != "protected-write-approval+jwt"
    ):
        raise ProtectedWriteRejected("unexpected approval envelope")
    claims = jwt.decode(
        token,
        APPROVAL_KEY,
        algorithms=["EdDSA"],
        issuer=APPROVAL_ISSUER,
        audience="protected-writer",
        leeway=JWT_LEEWAY_SECONDS,
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti", "action",
            "tenant_id", "subject", "resource_id", "content_sha256",
            "change_id",
        ]},
    )
    expected = {
        "action": "WRITE_PROTECTED_RESOURCE",
        "tenant_id": principal.tenant_id,
        "subject": principal.subject,
        "resource_id": resource_id,
        "content_sha256": digest,
    }
    if any(claims.get(key) != value for key, value in expected.items()):
        raise ProtectedWriteRejected("approval is not bound to this action")
    return claims


def scan_candidate(content, digest):
    request_id = uuid.uuid4().hex
    deadline = time.monotonic() + SCANNER_TOTAL_TIMEOUT_SECONDS
    with httpx.Client(
        base_url=SCANNER_ORIGIN,
        verify=scanner_tls,
        timeout=httpx.Timeout(
            SCANNER_TOTAL_TIMEOUT_SECONDS,
            connect=SCANNER_CONNECT_TIMEOUT_SECONDS,
        ),
        follow_redirects=False,
        trust_env=False,
        http2=True,
    ) as scanner:
        with scanner.stream(
            "POST",
            "/v1/scan",
            json={
                "request_id": request_id,
                "mode": "protected_write",
                "content_sha256": digest,
                "content_b64": base64.b64encode(content).decode("ascii"),
                "scanner_origin": SCANNER_ORIGIN,
                "runtime_policy_version": RUNTIME_POLICY_VERSION,
                "runtime_policy_sha256": RUNTIME_POLICY_SHA256,
                "scanner_total_timeout_seconds":
                    SCANNER_TOTAL_TIMEOUT_SECONDS,
                "response_policy_version": RESPONSE_POLICY_VERSION,
                "response_policy_sha256": RESPONSE_POLICY_SHA256,
                "response_max_bytes": SCANNER_RESPONSE_MAX_BYTES,
                "response_chunk_bytes": SCANNER_RESPONSE_CHUNK_BYTES,
            },
            headers={
                "Idempotency-Key": request_id,
                "Accept": "application/json",
                "Accept-Encoding": "identity",
            },
        ) as response:
            if response.history or canonical_https_origin(
                str(response.url.copy_with(path="", query=None, fragment=None))
            ) != SCANNER_ORIGIN:
                raise ProtectedWriteRejected(
                    "scanner response redirected or changed origin"
                )
            response.raise_for_status()
            body, response_bytes = read_bounded_json_object(
                response,
                deadline=deadline,
            )
    if (
        set(body) != {"result_jws"}
        or not isinstance(body["result_jws"], str)
        or not body["result_jws"]
    ):
        raise ProtectedWriteRejected(
            "scanner response is not exactly one signed result"
        )

    token = body["result_jws"]
    header = jwt.get_unverified_header(token)
    if (
        header.get("alg") != "EdDSA"
        or header.get("typ") != "scan-result+jwt"
    ):
        raise ProtectedWriteRejected("unexpected scanner envelope")
    claims = jwt.decode(
        token,
        SCANNER_KEY,
        algorithms=["EdDSA"],
        issuer=SCANNER_ISSUER,
        audience="protected-writer",
        leeway=JWT_LEEWAY_SECONDS,
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti", "request_id",
            "mode", "content_sha256", "decision", "scanner_version",
            "rule_bundle_sha256", "scanner_origin",
            "runtime_policy_version", "runtime_policy_sha256",
            "scanner_total_timeout_seconds", "response_policy_version",
            "response_policy_sha256", "response_max_bytes",
            "response_chunk_bytes",
        ]},
    )
    if (
        isinstance(claims.get("scanner_total_timeout_seconds"), bool)
        or not isinstance(
            claims.get("scanner_total_timeout_seconds"),
            (int, float),
        )
        or not math.isfinite(claims["scanner_total_timeout_seconds"])
        or type(claims.get("response_max_bytes")) is not int
        or type(claims.get("response_chunk_bytes")) is not int
        or claims["request_id"] != request_id
        or claims["mode"] != "protected_write"
        or claims["content_sha256"] != digest
        or claims["decision"] != "ALLOW"
        or claims["scanner_origin"] != SCANNER_ORIGIN
        or claims["runtime_policy_version"] != RUNTIME_POLICY_VERSION
        or claims["runtime_policy_sha256"] != RUNTIME_POLICY_SHA256
        or claims["scanner_total_timeout_seconds"]
            != SCANNER_TOTAL_TIMEOUT_SECONDS
        or claims["response_policy_version"] != RESPONSE_POLICY_VERSION
        or claims["response_policy_sha256"] != RESPONSE_POLICY_SHA256
        or claims["response_max_bytes"] != SCANNER_RESPONSE_MAX_BYTES
        or claims["response_chunk_bytes"] != SCANNER_RESPONSE_CHUNK_BYTES
    ):
        raise ProtectedWriteRejected("scanner result is blocked or misbound")
    if (
        not isinstance(claims.get("scanner_version"), str)
        or not claims["scanner_version"]
    ):
        raise ProtectedWriteRejected("scanner version is missing")
    require_policy_digest(
        str(claims.get("rule_bundle_sha256", "")),
        "rule_bundle_sha256",
    )
    return claims, response_bytes


def consume_approval(claims, digest):
    try:
        ddb.put_item(
            TableName=REPLAY_TABLE,
            Item={
                "jti": {"S": claims["jti"]},
                "content_sha256": {"S": digest},
                "resource_id": {"S": claims["resource_id"]},
                "expires_at": {"N": str(int(claims["exp"]))},
            },
            ConditionExpression="attribute_not_exists(jti)",
        )
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            raise ProtectedWriteRejected("approval already consumed") from exc
        raise


def atomic_commit(target, content):
    fd, temp_name = tempfile.mkstemp(
        prefix=f".{target.name}.",
        dir=target.parent,
    )
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(temp_name, 0o600)
        os.replace(temp_name, target)
        dir_fd = os.open(target.parent, os.O_DIRECTORY)
        try:
            os.fsync(dir_fd)
        finally:
            os.close(dir_fd)
    finally:
        if os.path.exists(temp_name):
            os.unlink(temp_name)


def controlled_write(request, principal):
    # principal exists only after OIDC/SPIFFE verification and tenant mapping.
    target = RESOURCE_BINDINGS.get(
        (principal.tenant_id, request.resource_id)
    )
    if target is None:
        raise ProtectedWriteRejected(
            "resource is not registered for tenant"
        )
    if len(request.content) > PROTECTED_WRITE_MAX_CANDIDATE_BYTES:
        raise ProtectedWriteRejected("candidate exceeds signed byte limit")
    digest = hashlib.sha256(request.content).hexdigest()
    approval = verify_approval(
        request.approval_jws,
        principal,
        request.resource_id,
        digest,
    )
    scan, scanner_response_bytes = scan_candidate(
        request.content,
        digest,
    )
    consume_approval(approval, digest)
    atomic_commit(target, request.content)
    if hashlib.sha256(target.read_bytes()).hexdigest() != digest:
        raise ProtectedWriteRejected(
            "post-write digest verification failed"
        )
    return {
        "status": "APPLIED",
        "resource_id": request.resource_id,
        "content_sha256": digest,
        "change_id": approval["change_id"],
        "approval_jti": approval["jti"],
        "scanner_version": scan["scanner_version"],
        "rule_bundle_sha256": scan["rule_bundle_sha256"],
        "principal_issuer": principal.issuer,
        "principal_subject": principal.subject,
        "tenant_id": principal.tenant_id,
        "scanner_origin": SCANNER_ORIGIN,
        "runtime_policy_version": RUNTIME_POLICY_VERSION,
        "runtime_policy_sha256": RUNTIME_POLICY_SHA256,
        "scanner_total_timeout_seconds": SCANNER_TOTAL_TIMEOUT_SECONDS,
        "response_policy_version": RESPONSE_POLICY_VERSION,
        "response_policy_sha256": RESPONSE_POLICY_SHA256,
        "candidate_max_bytes": PROTECTED_WRITE_MAX_CANDIDATE_BYTES,
        "candidate_bytes": len(request.content),
        "scanner_response_max_bytes": SCANNER_RESPONSE_MAX_BYTES,
        "scanner_response_chunk_bytes": SCANNER_RESPONSE_CHUNK_BYTES,
        "scanner_response_bytes": scanner_response_bytes,
    }

Bound candidate admission before allocation

The API or message consumer that constructs ProtectedWriteRequest must enforce the same signed candidate limit on declared length and streamed bytes before buffering, base64 decoding, or scanner submission. Reject chunked overflow, declared/actual mismatch, duplicate JSON members, and ambiguous encodings before this function runs.

Verify and retain evidence

Write the returned record plus both compact JWS envelopes to append-only storage, excluding raw content. The record and signed scanner result bind the canonical scanner origin, versioned runtime and response-policy digests, total deadline, response byte/chunk caps, and observed response size. Independently replay both signatures and bindings, verify the one-time approval ledger, and read back the committed digest. Test caller-supplied identity fields, wrong tenant/resource/digest/action, expired or replayed approval, TLS/client-certificate failure, forged or stale scanner result, redirects, origin changes, non-identity encoding, invalid or mismatched Content-Length, empty or oversized chunked bodies, duplicate JSON members, nonstandard constants, extra envelope fields, scanner timeout, block verdict, and post-write mismatch. Only an APPLIED record with successful independent replay demonstrates the control.

Make approval consumption crash-safe

Do not deploy the earlier one-step consume_approval(); atomic_commit() sequence. Back the approval row with a durable state machine keyed by signed jti and change_id, and bind it to tenant, resource, candidate digest, scanner receipt, pre-write digest, and one operation ID. Under an exclusive per-resource writer lock, reserve or read that exact row, transition RESERVED to COMMITTING, perform the atomic replace, read back the bounded file through a no-follow descriptor, then conditionally transition to APPLIED with the observed digest. A retry of the same operation may reconcile COMMITTING: an exact candidate digest completes APPLIED; an exact recorded pre-write digest may retry while the approval and scan are still valid; any third digest is ERROR and requires investigation. A different binding using the same approval is always denied.

def reconcile_and_commit(*, ledger, lock, approval, scan, target, content, digest):
    operation_id = approval["change_id"]
    with lock.exclusive(resource_id=approval["resource_id"]):
        before = bounded_nofollow_sha256(target)
        binding = {
            "operation_id": operation_id,
            "approval_jti": approval["jti"],
            "tenant_id": approval["tenant_id"],
            "resource_id": approval["resource_id"],
            "content_sha256": digest,
            "scan_jti": scan["jti"],
        }
        row = ledger.reserve_or_read(binding, preimage_sha256=before)
        if row["binding"] != binding:
            raise ProtectedWriteRejected("approval operation binding differs")
        observed = bounded_nofollow_sha256(target)
        if row["state"] == "APPLIED":
            if observed != digest:
                raise ProtectedWriteRejected("applied operation readback drifted")
            return row["result"]
        if row["state"] not in {"RESERVED", "COMMITTING"}:
            raise ProtectedWriteRejected("approval operation is not retryable")
        if observed not in {row["preimage_sha256"], digest}:
            ledger.mark_error(operation_id, "unexpected target generation")
            raise ProtectedWriteRejected("protected resource changed concurrently")
        ledger.compare_and_set(operation_id, row["state"], "COMMITTING")
        if observed != digest:
            atomic_commit(target, content)
        if bounded_nofollow_sha256(target) != digest:
            ledger.mark_error(operation_id, "post-commit readback mismatch")
            raise ProtectedWriteRejected("post-commit readback mismatch")
        return ledger.finish_applied(operation_id, content_sha256=digest)

reserve_or_read, every transition, and finish_applied must be conditional durable writes with consistent readback. Keep unresolved RESERVED/COMMITTING rows for the reconciler; never delete them merely because the request process failed.

AID-I-004.006-G005

Keep the action safely bounded

Rollback must restore bytes from a version-addressed, retention-protected snapshot store, not from local mutable metadata. Require both a detached signature on the snapshot manifest and a short-lived rollback approval JWS bound to the authenticated external principal, tenant, resource, exact manifest digest, and ROLLBACK_PROTECTED_RESOURCE action.

Signed snapshot and rollback implementation

# File: protected_writer/rollback.py
# Dependencies: boto3>=1.35,<2, PyJWT>=2.9,<3, cryptography>=43,<45
from __future__ import annotations

import base64
import hashlib
import json
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path

import boto3
import jwt
from botocore.exceptions import ClientError
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

TRUSTED_BUCKET = os.environ["PROTECTED_SNAPSHOT_BUCKET"]
TRUSTED_PREFIX = os.environ.get("PROTECTED_SNAPSHOT_PREFIX", "approved/")
APPROVAL_ISSUER = os.environ["ROLLBACK_APPROVAL_ISSUER"]
APPROVAL_KEY = open(os.environ["ROLLBACK_APPROVAL_PUBLIC_KEY_PATH"], "rb").read()
MANIFEST_KEY = serialization.load_pem_public_key(
    open(os.environ["SNAPSHOT_MANIFEST_PUBLIC_KEY_PATH"], "rb").read()
)
if not isinstance(MANIFEST_KEY, Ed25519PublicKey):
    raise RuntimeError("snapshot manifest key must be Ed25519")

# Root-owned resource mapping; the caller never supplies a filesystem path.
RESOURCE_BINDINGS = {
    ("tenant-a", "agent.identity.soul"): Path("/var/lib/agent/SOUL.md"),
    ("tenant-a", "agent.memory.primary"): Path("/var/lib/agent/MEMORY.md"),
    ("tenant-a", "agent.identity.agents_config"): Path("/var/lib/agent/AGENTS.md"),
}
REPLAY_TABLE = os.environ["ROLLBACK_APPROVAL_REPLAY_TABLE"]
MANIFEST_MAX_BYTES = int(os.environ["ROLLBACK_MANIFEST_MAX_BYTES"])
SIGNATURE_MAX_BYTES = int(os.environ["ROLLBACK_SIGNATURE_MAX_BYTES"])
SNAPSHOT_MAX_BYTES = int(os.environ["ROLLBACK_SNAPSHOT_MAX_BYTES"])
READ_CHUNK_BYTES = int(os.environ["ROLLBACK_READ_CHUNK_BYTES"])
if any(
    isinstance(value, bool) or value < 1
    for value in (
        MANIFEST_MAX_BYTES, SIGNATURE_MAX_BYTES,
        SNAPSHOT_MAX_BYTES, READ_CHUNK_BYTES,
    )
):
    raise RuntimeError("verified rollback byte limits are invalid")
s3 = boto3.client("s3")
ddb = boto3.client("dynamodb")


@dataclass(frozen=True)
class VerifiedPrincipal:
    issuer: str
    tenant_id: str
    subject: str


class RollbackRejected(Exception):
    pass


def get_versioned_bytes(key, version_id, max_bytes):
    if not key.startswith(TRUSTED_PREFIX) or not version_id:
        raise RollbackRejected("object is outside trusted versioned root")
    response = s3.get_object(
        Bucket=TRUSTED_BUCKET,
        Key=key,
        VersionId=version_id,
        ChecksumMode="ENABLED",
    )
    declared = response.get("ContentLength")
    if (
        response.get("VersionId") != version_id
        or isinstance(declared, bool)
        or not isinstance(declared, int)
        or declared < 0
        or declared > max_bytes
    ):
        raise RollbackRejected("object generation or declared size mismatch")
    body = response["Body"]
    payload = bytearray()
    try:
        for chunk in body.iter_chunks(
            chunk_size=min(READ_CHUNK_BYTES, max_bytes + 1)
        ):
            if len(payload) + len(chunk) > max_bytes:
                raise RollbackRejected("object exceeds verified byte limit")
            payload.extend(chunk)
    finally:
        body.close()
    raw = bytes(payload)
    if len(raw) != declared:
        raise RollbackRejected("declared and actual object size differ")
    checksum = response.get("ChecksumSHA256")
    actual = base64.b64encode(hashlib.sha256(raw).digest()).decode("ascii")
    if not isinstance(checksum, str) or checksum != actual:
        raise RollbackRejected("returned object checksum is absent or invalid")
    return raw


def load_signed_manifest(
    manifest_key, manifest_version_id, signature_key, signature_version_id
):
    raw = get_versioned_bytes(
        manifest_key, manifest_version_id, MANIFEST_MAX_BYTES
    )
    signature = base64.b64decode(
        get_versioned_bytes(
            signature_key, signature_version_id, SIGNATURE_MAX_BYTES
        ),
        validate=True,
    )
    MANIFEST_KEY.verify(signature, raw)
    def reject_duplicate_keys(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise RollbackRejected("manifest contains duplicate JSON member")
            value[key] = item
        return value
    manifest = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=lambda value: (_ for _ in ()).throw(
            RollbackRejected("manifest contains non-finite value")
        ),
    )
    digest = hashlib.sha256(raw).hexdigest()
    required = {
        "schema_version", "tenant_id", "resource_id", "snapshot_key",
        "snapshot_version_id", "content_sha256", "approved_change_id",
        "created_at",
    }
    if set(manifest) != required or manifest["schema_version"] != 1:
        raise RollbackRejected("invalid signed manifest schema")
    if (
        not manifest["snapshot_key"].startswith(TRUSTED_PREFIX)
        or not manifest["snapshot_version_id"]
    ):
        raise RollbackRejected("manifest points outside trusted root")
    return manifest, digest


def verify_rollback_approval(token, principal, manifest, manifest_digest):
    header = jwt.get_unverified_header(token)
    if header.get("alg") != "EdDSA" or header.get("typ") != "rollback-approval+jwt":
        raise RollbackRejected("unexpected rollback approval envelope")
    claims = jwt.decode(
        token,
        APPROVAL_KEY,
        algorithms=["EdDSA"],
        issuer=APPROVAL_ISSUER,
        audience="protected-writer",
        leeway=10,
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti", "action",
            "tenant_id", "subject", "resource_id", "manifest_sha256",
            "incident_id",
        ]},
    )
    expected = {
        "action": "ROLLBACK_PROTECTED_RESOURCE",
        "tenant_id": principal.tenant_id,
        "subject": principal.subject,
        "resource_id": manifest["resource_id"],
        "manifest_sha256": manifest_digest,
    }
    if any(claims.get(k) != v for k, v in expected.items()):
        raise RollbackRejected("rollback approval is not action-bound")
    return claims


def consume_approval(claims):
    try:
        ddb.put_item(
            TableName=REPLAY_TABLE,
            Item={
                "jti": {"S": claims["jti"]},
                "resource_id": {"S": claims["resource_id"]},
                "manifest_sha256": {"S": claims["manifest_sha256"]},
                "expires_at": {"N": str(int(claims["exp"]))},
            },
            ConditionExpression="attribute_not_exists(jti)",
        )
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            raise RollbackRejected("rollback approval already consumed") from exc
        raise


def atomic_restore(target, content):
    fd, temp_name = tempfile.mkstemp(
        prefix=f".{target.name}.rollback.", dir=target.parent
    )
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(temp_name, 0o600)
        os.replace(temp_name, target)
        dir_fd = os.open(target.parent, os.O_DIRECTORY)
        try:
            os.fsync(dir_fd)
        finally:
            os.close(dir_fd)
    finally:
        if os.path.exists(temp_name):
            os.unlink(temp_name)


def rollback(
    principal,
    approval_jws,
    manifest_key,
    manifest_version_id,
    signature_key,
    signature_version_id,
):
    manifest, manifest_digest = load_signed_manifest(
        manifest_key, manifest_version_id, signature_key, signature_version_id
    )
    if manifest["tenant_id"] != principal.tenant_id:
        raise RollbackRejected("snapshot tenant mismatch")
    target = RESOURCE_BINDINGS.get(
        (principal.tenant_id, manifest["resource_id"])
    )
    if target is None:
        raise RollbackRejected("snapshot resource is not registered")
    approval = verify_rollback_approval(
        approval_jws, principal, manifest, manifest_digest
    )
    snapshot = get_versioned_bytes(
        manifest["snapshot_key"], manifest["snapshot_version_id"],
        SNAPSHOT_MAX_BYTES,
    )
    snapshot_digest = hashlib.sha256(snapshot).hexdigest()
    if snapshot_digest != manifest["content_sha256"]:
        raise RollbackRejected("snapshot digest mismatch")
    consume_approval(approval)
    atomic_restore(target, snapshot)
    if hashlib.sha256(target.read_bytes()).hexdigest() != snapshot_digest:
        raise RollbackRejected("post-restore digest mismatch")
    return {
        "status": "RESTORED",
        "resource_id": manifest["resource_id"],
        "content_sha256": snapshot_digest,
        "manifest_sha256": manifest_digest,
        "snapshot_version_id": manifest["snapshot_version_id"],
        "approved_change_id": manifest["approved_change_id"],
        "incident_id": approval["incident_id"],
        "approval_jti": approval["jti"],
        "principal_issuer": principal.issuer,
        "principal_subject": principal.subject,
        "tenant_id": principal.tenant_id,
    }

Verify safely

The verified rollback policy owns the manifest, signature, snapshot, chunk, and aggregate three-object byte budget plus client connect/read deadline. A missing checksum, generation mismatch, timeout, extra object request, declared/actual length mismatch, or exceeded aggregate budget denies restore before approval consumption.

Preserve the detached manifest signature, exact object version IDs, approval JWS, replay-ledger record, restore result, and post-restore readback in append-only evidence storage. Independently replay both signatures and all bindings.

AID-I-004.006-G006

Concept

Every attempted modification of a protected logical resource — whether allowed or blocked — must produce SIEM-grade structured telemetry. This is necessary for incident response, timeline reconstruction, and proving that only approved change paths were used.

Recommended event fields

{
  "agent_id": "agent-123",
  "skill_id": "skill-456",
  "protected_resource_id": "agent.identity.soul",
  "backing_path": "/var/lib/agent/SOUL.md",
  "operation": "write",
  "decision": "blocked",
  "reason": "no_approved_change_context",
  "writer_identity": "spiffe://corp/agent-writer",
  "approval_context": "CHG-2026-00421",
  "baseline_hash": "sha256-from-measured-before-state",
  "observed_hash": "sha256-from-measured-attempt-state",
  "rollback_candidate": true,
  "timestamp": "2026-03-30T12:00:00Z"
}

Minimum audit requirements:

• logical protected resource ID and concrete backing path
• requesting skill / writer identity
• operation type and decision
• approval / ticket context
• hash context before and after, when applicable
• rollback eligibility or recovery action if triggered

Action: Push these events to SIEM / SOAR and correlate them with controlled-writer logs, startup verification results, and rollback actions for end-to-end protected-surface forensics.

Verify safely

replay sink signatures, reconcile controlled-writer and OS audit populations, and alert on any expected event missing.

Open-source tools

• AIDE (integrity verification; not write enforcement)
• Wazuh (file-integrity and write-attempt telemetry; not write enforcement)
• OSSEC (file-integrity telemetry; not write enforcement)
• Falco (write-attempt telemetry; blocking requires an external responder)
• Cilium Tetragon
• SELinux (mandatory access control for protected state paths)
• AppArmor (mandatory access control for protected state paths)
• Landlock (process-scoped filesystem access control for protected state paths)
• Git (versioned baseline and diff evidence; not write enforcement)
• gVisor
• libseccomp
• OpenBao (signing and key service; not a content scanner)

Commercial / hosted tools

• CrowdStrike Falcon Insight XDR
• SentinelOne Singularity Endpoint
• Microsoft Defender for Endpoint
• Aqua Platform (container file immutability and runtime drift prevention)
• Palo Alto Networks Prisma Cloud
• HashiCorp Vault Enterprise
• HCP Vault Dedicated

Threat mappings

MITRE ATLAS

• AML.T0011 User Execution
• AML.T0081 Modify AI Agent Configuration
• AML.T0011.002 User Execution: Poisoned AI Agent Tool
• AML.T0080 AI Agent Context Poisoning
• AML.T0080.000 AI Agent Context Poisoning: Memory
• AML.T0121 AI Agent Environment Reconstruction (protected identity and state files cannot be rewritten to persist across restart or redeployment)

MAESTRO

• Agent Goal Manipulation (L7)
• Agent Identity Attack (L7)
• Backdoor Attacks (L3)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection
• LLM03:2026 Excessive Agency (controlled writers prevent agents from expanding identity or authority-bearing state)
• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI01:2026 Agent Goal Hijack
• ASI06:2026 Memory & Context Poisoning

NIST Adversarial Machine Learning 2025

• NISTAML.018 Prompt Injection
• NISTAML.039 Compromising connected resources (tampered agent identity/config files can redirect agent to compromise connected resources)

Cisco Integrated AI Security and Safety Framework

• AISubtech-14.1.2 Insufficient Access Controls
• AISubtech-5.1.1 Long-term / Short-term Memory Injection (protected primary-memory files reject unauthorized persistent writes)
• AISubtech-5.2.1 Agent Profile Tampering
• AISubtech-9.1.2 Unauthorized or Unsolicited System Access (protected state-file writes are confined to an approved controlled writer)
• AITech-5.1 Memory System Persistence (protected primary-memory files reject unauthorized persistent writes)
• AITech-5.2 Configuration Persistence
• AITech-7.2 Memory System Corruption (protected primary-memory files reject unauthorized persistent writes)
• AITech-9.1 Model or Agentic System Manipulation (protected state-file writes are confined to an approved controlled writer)

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (write protection on persistent identity/state files prevents persistent backdoors that enable rogue actions)
• PIJ: Prompt Injection (write protection prevents persistent injected instructions from entering identity/state files)

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning
• Agents - Core 13.6: Intent Breaking & Goal Manipulation