Scope boundary

Owns bounded read-only first-pass investigation of queued AI-security alerts and returns evidence, missing context, a structured disposition, and a recommended human action. It cannot suppress rules, change authoritative severity, close tickets, revoke credentials, quarantine workloads, or contain an incident.

AID-D-005.003: AID-D-005.008 triages a specific existing alert under a read-only query allowlist; AID-D-005.003 performs hypothesis-driven exploration for threats that may not have triggered an alert. Triage is queue-oriented evidence preparation, while hunting is open-ended investigation.

AID-D-008: AID-D-005.008 uses an agent to prepare a human-readable case disposition for one alert; AID-D-008 serves a controlled fusion model that emits a cross-detector security finding. A triage recommendation is not a new detector verdict, and a fusion score does not perform case investigation.

Implementation guidance

• AID-D-005.008-G001: Provision the triage agent with read-only identities and a query allowlist; do not give it write, suppression, ticket-closure, containment, or credential-management privileges.
• AID-D-005.008-G002: Run scheduled least-privilege attestation for the production triage identity and fail the gate if it can perform write, close, revoke, quarantine, or suppression actions.
• AID-D-005.008-G003: Require every triage result to satisfy a fixed disposition schema that separates evidence from recommendations and records that no containment action was invoked.
• AID-D-005.008-G004: Roll out one noisy rule at a time in shadow mode and require agreement, missed-critical, and time-saved gates before expanding to another rule family.

AID-D-005.008-G001

Concept:

The first-pass triage agent should enrich and summarize alerts, not become a response authority. Give it a distinct workload identity, read-only SIEM permissions, and a server-side query allowlist. If the model asks for a query outside the allowlist, the broker rejects the request before it reaches the SIEM.

Example OpenSearch role

# file: opensearch/roles/ai_triage_readonly.yml
ai_triage_readonly:
  cluster_permissions:
    - cluster_composite_ops_ro
  index_permissions:
    - index_patterns:
        - "security-alerts-*"
        - "agent-session-*"
        - "cloud-audit-*"
        - "edr-events-*"
      allowed_actions:
        - read
        - search
        - indices:data/read/search
        - indices:data/read/msearch

Example query broker

from __future__ import annotations

# file: triage/query_broker.py

import re
from copy import deepcopy
from dataclasses import dataclass
from typing import Any, Mapping

PARAM_PATTERN = re.compile(r"^[A-Za-z0-9._:@/-]{1,256}$")


@dataclass(frozen=True)
class QueryTemplate:
    name: str
    index: str
    body: dict


QUERY_ALLOWLIST: Mapping[str, QueryTemplate] = {
    "alert_context": QueryTemplate(
        name="alert_context",
        index="security-alerts-*",
        body={"query": {"term": {"alert.id": "{{alert_id}}"}}},
    ),
    "recent_agent_session": QueryTemplate(
        name="recent_agent_session",
        index="agent-session-*",
        body={"query": {"term": {"session.id": "{{session_id}}"}}},
    ),
}


def validate_params(params: dict[str, str]) -> dict[str, str]:
    for key, value in params.items():
        if not PARAM_PATTERN.fullmatch(value):
            raise ValueError("Invalid query parameter: " + key)
    return params


def render_value(value: Any, params: dict[str, str]) -> Any:
    if isinstance(value, str):
        rendered = value
        for key, replacement in params.items():
            rendered = rendered.replace("{{" + key + "}}", replacement)
        if "{{" in rendered or "}}" in rendered:
            raise ValueError("Template has unresolved parameters")
        return rendered
    if isinstance(value, list):
        return [render_value(item, params) for item in value]
    if isinstance(value, dict):
        return {key: render_value(item, params) for key, item in value.items()}
    return value


def render_template(template: QueryTemplate, params: dict[str, str]) -> tuple[str, dict]:
    if template.name not in QUERY_ALLOWLIST:
        raise PermissionError("Query is not allowlisted")
    params = validate_params(params)
    return template.index, render_value(deepcopy(template.body), params)


def execute_triage_query(opensearch_client, query_name: str, params: dict[str, str]) -> dict:
    template = QUERY_ALLOWLIST.get(query_name)
    if template is None:
        raise PermissionError("Query is not allowlisted: " + query_name)
    index, body = render_template(template, params)
    return opensearch_client.search(index=index, body=body)

Operational notes: Keep the broker outside the model runtime, authenticate it with a workload identity, and log every rejected query with alert_id, agent_id, query_name, and policy_version. The renderer performs bounded recursive replacement and fails closed on unresolved parameters; do not allow model-authored raw SIEM queries to bypass this broker.

Verify safely

An independent verifier must enumerate effective permissions under an independent identity, run negative write, suppression, closure, containment and credential-management probes, and reconcile every assigned alert with its reads and terminal result, read evidence back, verify signatures, digests, immutability and effective privileges, and compare the canonical result digest.

AID-D-005.008-G002

Concept:

A read-only triage design should produce fresh evidence that the deployed identity is still read-only. Run a scheduled attestation job against cloud IAM and SIEM-native roles, write a dated artifact, and fail deployment or promotion when the identity can acquire response authority.

Least-privilege attestation job

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()

def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value

def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value

RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
ROLE_READBACK_TIMEOUT_SECONDS = _aidefend_positive_float(
    "D005008_ROLE_READBACK_TIMEOUT_SECONDS"
)
ROLE_RESPONSE_MAXIMUM_BYTES = _aidefend_positive_int(
    "D005008_ROLE_RESPONSE_MAXIMUM_BYTES"
)

# File: triage/read_only_attestation.py
import json
import os
import ssl
import subprocess
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlsplit

FORBIDDEN_AWS_ACTIONS = (
    "iam:PassRole", "iam:AttachRolePolicy", "iam:UpdateAccessKey",
    "iam:DeleteAccessKey", "ec2:StopInstances", "ec2:TerminateInstances",
    "lambda:DeleteFunction", "secretsmanager:DeleteSecret",
    "secretsmanager:PutSecretValue", "es:ESHttpPut", "es:ESHttpDelete",
)
DENIED_AWS_DECISIONS = {"explicitDeny", "implicitDeny"}
AWS_DECISIONS = DENIED_AWS_DECISIONS | {"allowed"}

ROLE_REQUIRED_KEYS = {
    "cluster_permissions", "index_permissions", "tenant_permissions",
}
ROLE_OPTIONAL_KEYS = {"reserved", "hidden", "static", "description"}
INDEX_PERMISSION_KEYS = {
    "index_patterns", "dls", "fls", "masked_fields", "allowed_actions",
}
TENANT_PERMISSION_KEYS = {"tenant_patterns", "allowed_actions"}
READ_ONLY_ACTION_GROUPS = {
    "cluster_composite_ops_ro", "cluster_monitor", "get", "indices_monitor",
    "kibana_all_read", "read", "search", "suggest",
}
READ_ONLY_ACTION_PREFIXES = (
    "cluster:monitor/", "indices:data/read/", "indices:monitor/",
)
WRITE_ACTION_GROUPS = {
    "all_access", "cluster_all", "crud", "data_access", "delete", "index",
    "kibana_all_write", "manage", "manage_aliases", "unlimited", "write",
}
WRITE_ACTION_PREFIXES = (
    "cluster:admin/", "indices:admin/", "indices:data/write/",
)

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, request, file_pointer, code, message, headers, new_url):
        return None

def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate JSON key: {key}")
        value[key] = item
    return value

def reject_nonfinite_json(value: str) -> None:
    raise ValueError(f"non-finite JSON constant: {value}")

def strict_json(value: bytes | str, label: str) -> Any:
    try:
        text = value.decode("utf-8", errors="strict") if isinstance(value, bytes) else value
        return json.loads(
            text, object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_nonfinite_json,
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from error

def run_json(command: list[str]) -> dict[str, Any]:
    completed = subprocess.run(
        command, check=True, capture_output=True, encoding="utf-8",
        errors="strict", timeout=ROLE_READBACK_TIMEOUT_SECONDS,
        env={**os.environ, "AWS_PAGER": ""},
    )
    result = strict_json(completed.stdout, "AWS CLI output")
    if not isinstance(result, dict):
        raise ValueError("AWS CLI output is not an object")
    return result

def evaluate_aws_simulation(result: dict[str, Any]) -> list[dict[str, str]]:
    rows = result.get("EvaluationResults")
    if not isinstance(rows, list) or not rows:
        raise ValueError("AWS simulation result population is absent or empty")
    if result.get("IsTruncated", False) is not False:
        raise ValueError("AWS simulation result population is truncated")
    expected = {action.casefold(): action for action in FORBIDDEN_AWS_ACTIONS}
    observed: dict[str, str] = {}
    for row in rows:
        if not isinstance(row, dict):
            raise ValueError("AWS simulation result row is not an object")
        action = row.get("EvalActionName")
        decision = row.get("EvalDecision")
        if not isinstance(action, str) or not action:
            raise ValueError("AWS simulation result action is absent")
        canonical_action = action.casefold()
        if canonical_action not in expected:
            raise ValueError(f"AWS simulation returned an unexpected action: {action}")
        if canonical_action in observed:
            raise ValueError(f"AWS simulation duplicated an action: {action}")
        if decision not in AWS_DECISIONS:
            raise ValueError(f"AWS simulation returned an unknown decision for {action}")
        observed[canonical_action] = decision
    if set(observed) != set(expected):
        missing = sorted(expected[key] for key in set(expected) - set(observed))
        raise ValueError(f"AWS simulation omitted requested actions: {missing}")
    return [
        {"action": expected[key], "decision": observed[key]}
        for key in sorted(expected) if observed[key] not in DENIED_AWS_DECISIONS
    ]

def assert_aws_actions_denied(principal_arn: str) -> list[dict[str, str]]:
    result = run_json([
        "aws", "iam", "simulate-principal-policy", "--policy-source-arn",
        principal_arn, "--action-names", *FORBIDDEN_AWS_ACTIONS,
        "--output", "json", "--no-cli-pager",
    ])
    return evaluate_aws_simulation(result)

def require_string_list(value: Any, label: str, *, allow_empty: bool) -> list[str]:
    if (
        not isinstance(value, list)
        or (not allow_empty and not value)
        or any(not isinstance(item, str) or not item for item in value)
        or len(value) != len(set(value))
    ):
        raise ValueError(f"{label} must be a typed unique string list")
    return value

def validate_opensearch_role(role_doc: Any) -> dict[str, Any]:
    if not isinstance(role_doc, dict) or not role_doc:
        raise ValueError("named OpenSearch role is absent or empty")
    keys = set(role_doc)
    if not ROLE_REQUIRED_KEYS.issubset(keys) or keys - (ROLE_REQUIRED_KEYS | ROLE_OPTIONAL_KEYS):
        raise ValueError("OpenSearch role schema is missing required keys or contains unknown keys")
    for key in ("reserved", "hidden", "static"):
        if key in role_doc and not isinstance(role_doc[key], bool):
            raise ValueError(f"OpenSearch role metadata is not boolean: {key}")
    if "description" in role_doc and not isinstance(role_doc["description"], str):
        raise ValueError("OpenSearch role description is not a string")
    cluster = require_string_list(
        role_doc["cluster_permissions"], "cluster_permissions", allow_empty=True,
    )
    index_permissions = role_doc["index_permissions"]
    tenant_permissions = role_doc["tenant_permissions"]
    if not isinstance(index_permissions, list) or not isinstance(tenant_permissions, list):
        raise ValueError("OpenSearch permission populations must be lists")
    action_count = len(cluster)
    for permission in index_permissions:
        if not isinstance(permission, dict) or set(permission) != INDEX_PERMISSION_KEYS:
            raise ValueError("OpenSearch index permission schema differs")
        require_string_list(permission["index_patterns"], "index_patterns", allow_empty=False)
        if not isinstance(permission["dls"], str):
            raise ValueError("OpenSearch dls must be a string")
        require_string_list(permission["fls"], "fls", allow_empty=True)
        require_string_list(permission["masked_fields"], "masked_fields", allow_empty=True)
        actions = require_string_list(
            permission["allowed_actions"], "index allowed_actions", allow_empty=False,
        )
        action_count += len(actions)
    for permission in tenant_permissions:
        if not isinstance(permission, dict) or set(permission) != TENANT_PERMISSION_KEYS:
            raise ValueError("OpenSearch tenant permission schema differs")
        require_string_list(permission["tenant_patterns"], "tenant_patterns", allow_empty=False)
        actions = require_string_list(
            permission["allowed_actions"], "tenant allowed_actions", allow_empty=False,
        )
        action_count += len(actions)
    if action_count == 0:
        raise ValueError("OpenSearch role contains no permission actions")
    return role_doc

def parse_opensearch_role_response(raw: bytes, role_name: str) -> dict[str, Any]:
    body = strict_json(raw, "OpenSearch role response")
    if not isinstance(body, dict) or set(body) != {role_name}:
        raise ValueError("OpenSearch response does not contain exactly the named role")
    return validate_opensearch_role(body[role_name])

def fetch_opensearch_role(base_url: str, role_name: str, token: str) -> dict[str, Any]:
    parsed = urlsplit(base_url)
    if (
        parsed.scheme != "https" or not parsed.hostname
        or parsed.username or parsed.password or parsed.path not in {"", "/"}
        or parsed.query or parsed.fragment or not role_name or not token
    ):
        raise ValueError("OpenSearch role endpoint must be one authenticated HTTPS origin")
    origin = "https://" + parsed.netloc
    endpoint = origin + "/_plugins/_security/api/roles/" + quote(role_name, safe="")
    context = ssl.create_default_context(cafile=os.environ["D005008_OPENSEARCH_CA_FILE"])
    opener = urllib.request.build_opener(
        urllib.request.ProxyHandler({}),
        urllib.request.HTTPSHandler(context=context),
        NoRedirect(),
    )
    request = urllib.request.Request(
        endpoint,
        headers={
            "Authorization": "Bearer " + token,
            "Accept": "application/json",
            "Accept-Encoding": "identity",
        },
        method="GET",
    )
    with opener.open(request, timeout=ROLE_READBACK_TIMEOUT_SECONDS) as response:
        if response.getcode() != 200 or response.geturl() != endpoint:
            raise ValueError("OpenSearch role readback status or origin differs")
        if response.headers.get_content_type().lower() != "application/json":
            raise ValueError("OpenSearch role response media type differs")
        encoding = response.headers.get("Content-Encoding", "identity").strip().lower()
        if encoding not in {"", "identity"}:
            raise ValueError("encoded OpenSearch role response is forbidden")
        declared = response.headers.get("Content-Length")
        expected_length = None
        if declared is not None:
            if not declared.strip().isdecimal():
                raise ValueError("OpenSearch Content-Length is invalid")
            expected_length = int(declared)
            if expected_length > ROLE_RESPONSE_MAXIMUM_BYTES:
                raise ValueError("OpenSearch role response exceeds signed byte bound")
        raw = response.read(ROLE_RESPONSE_MAXIMUM_BYTES + 1)
    if len(raw) > ROLE_RESPONSE_MAXIMUM_BYTES:
        raise ValueError("OpenSearch role response exceeds signed byte bound")
    if expected_length is not None and len(raw) != expected_length:
        raise ValueError("OpenSearch role response length differs from Content-Length")
    return parse_opensearch_role_response(raw, role_name)

def action_is_read_only(action: str) -> bool:
    normalized = action.casefold()
    if normalized in WRITE_ACTION_GROUPS or normalized.startswith(WRITE_ACTION_PREFIXES):
        return False
    return normalized in READ_ONLY_ACTION_GROUPS or normalized.startswith(
        READ_ONLY_ACTION_PREFIXES
    )

def assert_opensearch_role_readonly(role_doc: dict[str, Any]) -> list[str]:
    scoped_actions: list[tuple[str, str]] = [
        ("cluster", action) for action in role_doc["cluster_permissions"]
    ]
    for index, permission in enumerate(role_doc["index_permissions"]):
        scoped_actions.extend(
            (f"index[{index}]", action) for action in permission["allowed_actions"]
        )
    for index, permission in enumerate(role_doc["tenant_permissions"]):
        scoped_actions.extend(
            (f"tenant[{index}]", action) for action in permission["allowed_actions"]
        )
    return [
        f"{scope}:{action}"
        for scope, action in scoped_actions
        if not action_is_read_only(action)
    ]

def build_attestation() -> dict[str, Any]:
    principal_arn = os.environ["TRIAGE_AWS_PRINCIPAL_ARN"]
    opensearch_url = os.environ["OPENSEARCH_URL"]
    opensearch_role = os.environ["OPENSEARCH_TRIAGE_ROLE"]
    opensearch_token = os.environ["OPENSEARCH_API_TOKEN"]
    aws_findings = assert_aws_actions_denied(principal_arn)
    role_doc = fetch_opensearch_role(opensearch_url, opensearch_role, opensearch_token)
    opensearch_findings = assert_opensearch_role_readonly(role_doc)
    status = "PASS" if not aws_findings and not opensearch_findings else "FAIL"
    return {
        "schema_version": "aidefend.triage_readonly_attestation.v2",
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "principal_arn": principal_arn,
        "opensearch_role": opensearch_role,
        "status": status,
        "aws_forbidden_actions_allowed": aws_findings,
        "opensearch_unapproved_actions": opensearch_findings,
    }

def main() -> None:
    artifact = build_attestation()
    Path("artifacts").mkdir(exist_ok=True)
    output_path = Path("artifacts") / (
        "triage-readonly-attestation-" + artifact["generated_at"][:10] + ".json"
    )
    output_path.write_text(
        json.dumps(artifact, sort_keys=True, indent=2, allow_nan=False) + "\n",
        encoding="utf-8",
    )
    if artifact["status"] != "PASS":
        raise SystemExit("Triage identity is not read-only; wrote " + str(output_path))

if __name__ == "__main__":
    main()

Fail-closed parser and population fixtures

Run these fixtures in CI whenever the AWS CLI, OpenSearch Security plugin, role schema, or read-only allowlist changes. Missing or ambiguous evidence is an error, never a clean attestation.

# File: tests/test_read_only_attestation.py
from __future__ import annotations

import copy
import json
import os
import unittest

os.environ.update({
    "AIDEFEND_RUNTIME_PROFILE_VERSION": "test-profile-v1",
    "AIDEFEND_RUNTIME_PROFILE_SHA256": "1" * 64,
    "D005008_ROLE_READBACK_TIMEOUT_SECONDS": "2",
    "D005008_ROLE_RESPONSE_MAXIMUM_BYTES": "65536",
})

from triage.read_only_attestation import (  # noqa: E402
    FORBIDDEN_AWS_ACTIONS,
    assert_opensearch_role_readonly,
    evaluate_aws_simulation,
    parse_opensearch_role_response,
    strict_json,
)

ROLE_NAME = "triage_readonly"
VALID_ROLE = {
    "reserved": False,
    "hidden": False,
    "cluster_permissions": ["cluster_monitor"],
    "index_permissions": [{
        "index_patterns": ["security-alerts-*"],
        "dls": "",
        "fls": [],
        "masked_fields": [],
        "allowed_actions": ["read"],
    }],
    "tenant_permissions": [],
    "static": False,
}

def denied_population() -> dict:
    return {
        "EvaluationResults": [
            {"EvalActionName": action, "EvalDecision": "implicitDeny"}
            for action in FORBIDDEN_AWS_ACTIONS
        ],
        "IsTruncated": False,
    }

class ReadOnlyAttestationTests(unittest.TestCase):
    def test_valid_complete_populations_pass(self):
        self.assertEqual(evaluate_aws_simulation(denied_population()), [])
        raw = json.dumps({ROLE_NAME: VALID_ROLE}).encode("utf-8")
        role = parse_opensearch_role_response(raw, ROLE_NAME)
        self.assertEqual(assert_opensearch_role_readonly(role), [])

    def test_strict_json_rejects_duplicate_and_nonfinite_values(self):
        for raw in (b'{"x":1,"x":2}', b'{"x":NaN}', b'{"x":Infinity}'):
            with self.subTest(raw=raw), self.assertRaises(ValueError):
                strict_json(raw, "fixture")

    def test_aws_rejects_missing_empty_duplicate_and_omitted_results(self):
        fixtures = [{}, {"EvaluationResults": [], "IsTruncated": False}]
        duplicate = denied_population()
        duplicate["EvaluationResults"].append(
            copy.deepcopy(duplicate["EvaluationResults"][0])
        )
        omitted = denied_population()
        omitted["EvaluationResults"].pop()
        fixtures.extend([duplicate, omitted])
        for fixture in fixtures:
            with self.subTest(fixture=fixture), self.assertRaises(ValueError):
                evaluate_aws_simulation(fixture)

    def test_opensearch_rejects_missing_empty_duplicate_and_nonfinite_roles(self):
        fixtures = [
            b'{}',
            json.dumps({ROLE_NAME: {
                "cluster_permissions": [],
                "index_permissions": [],
                "tenant_permissions": [],
            }}).encode("utf-8"),
            b'{"triage_readonly":{},"triage_readonly":{}}',
            b'{"triage_readonly":{"cluster_permissions":[NaN],"index_permissions":[],"tenant_permissions":[]}}',
        ]
        for raw in fixtures:
            with self.subTest(raw=raw), self.assertRaises(ValueError):
                parse_opensearch_role_response(raw, ROLE_NAME)

    def test_opensearch_rejects_unknown_permission_shape(self):
        role = copy.deepcopy(VALID_ROLE)
        role["index_permissions"][0]["unexpected"] = []
        raw = json.dumps({ROLE_NAME: role}).encode("utf-8")
        with self.assertRaises(ValueError):
            parse_opensearch_role_response(raw, ROLE_NAME)

    def test_opensearch_unknown_or_write_action_fails_closed(self):
        for action in ("future_read_like_group", "crud", "indices:data/write/index"):
            role = copy.deepcopy(VALID_ROLE)
            role["cluster_permissions"] = [action]
            with self.subTest(action=action):
                self.assertTrue(assert_opensearch_role_readonly(role))

if __name__ == "__main__":
    unittest.main()

Action: Run this job daily and before expanding the triage agent to a new rule family. Store the JSON artifact with the rollout report. The attestation should fail closed if the identity can write indexes, close incidents, suppress detections, revoke credentials, mutate cloud resources, or invoke containment APIs.

Verify complete coverage

Operate this detector at the independent privilege-attestation job over every production triage identity and reachable service.

Verify safely

An independent verifier must enumerate effective permissions from each authoritative system and execute non-mutating authorization probes for write, close, revoke, quarantine, suppress and credential actions, then compare all identities and services, read evidence back, verify signatures, digests, immutability and effective privileges, and compare the canonical result digest.

AID-D-005.008-G003

Concept:

Free-form alert summaries are hard to review and easy to over-trust. Force the model output through a schema with explicit evidence, uncertainty, and next-human-action fields. The schema should make containment impossible by contract: containment_performed must be false, and no downstream system should treat recommended_human_action as an execution command.

Pydantic disposition contract

from __future__ import annotations

# file: triage/disposition_schema.py

import hmac as _aidefend_policy_hmac
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import re as _aidefend_policy_re
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_write_private(path, payload):
    descriptor = _aidefend_policy_os.open(
        path,
        _aidefend_policy_os.O_WRONLY
        | _aidefend_policy_os.O_CREAT
        | _aidefend_policy_os.O_EXCL,
        0o600,
    )
    with _aidefend_policy_os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        _aidefend_policy_os.fsync(stream.fileno())
    _aidefend_policy_os.chmod(path, 0o400)


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

    return _aidefend_policy_json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    if policy_path == signature_path:
        raise RuntimeError("policy and signature paths must differ")
    policy_raw = policy_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = _aidefend_policy_os.environ[key_env]
    remote_key = bool(_aidefend_policy_re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*:(?://)?", key_reference
    )) and not key_reference.startswith("-----BEGIN ")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else _AidefendPolicyPath(key_reference).read_bytes()
    )
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout from the versioned runtime profile is invalid")
    with _aidefend_policy_tempfile.TemporaryDirectory(
        prefix="aidefend-signed-policy-"
    ) as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        policy_snapshot = root / "policy.json"
        signature_snapshot = root / "policy.sig"
        _aidefend_write_private(policy_snapshot, policy_raw)
        _aidefend_write_private(signature_snapshot, signature_raw)
        captured = [(policy_snapshot, policy_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            _aidefend_policy_subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(policy_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=verify_timeout,
            )
            if any(
                not _aidefend_policy_hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
            policy = _aidefend_strict_json(policy_raw)
        finally:
            for path, _ in captured:
                _aidefend_policy_os.chmod(path, 0o600)
    if (
        not isinstance(policy, dict)
        or policy.get("schema_version") != schema_version
        or not isinstance(policy.get("policy_version"), str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    return policy


from typing import Literal
from pydantic import BaseModel, Field, field_validator


TRIAGE_SCHEMA_POLICY = _aidefend_load_signed_policy(
    path_env="D005008_TRIAGE_SCHEMA_POLICY_PATH",
    signature_env="D005008_TRIAGE_SCHEMA_POLICY_SIGNATURE",
    key_env="D005008_TRIAGE_SCHEMA_POLICY_VERIFY_KEY",
    schema_version="aidefend.triage-disposition-schema-policy.v1",
)
_TRIAGE_BOUND_NAMES = (
    "source_min_length", "source_max_length",
    "query_name_min_length", "query_name_max_length",
    "observation_min_length", "observation_max_length",
    "summary_min_length", "summary_max_length",
    "evidence_min_count", "evidence_max_count",
    "affected_assets_max_count", "missing_context_max_count",
    "recommended_action_min_length", "recommended_action_max_length",
)
TRIAGE_BOUNDS = {name: TRIAGE_SCHEMA_POLICY.get(name) for name in _TRIAGE_BOUND_NAMES}
if any(
    isinstance(value, bool) or not isinstance(value, int) or value < 1
    for value in TRIAGE_BOUNDS.values()
):
    raise RuntimeError("signed triage schema bounds are invalid")
for minimum_name, maximum_name in (
    ("source_min_length", "source_max_length"),
    ("query_name_min_length", "query_name_max_length"),
    ("observation_min_length", "observation_max_length"),
    ("summary_min_length", "summary_max_length"),
    ("evidence_min_count", "evidence_max_count"),
    ("recommended_action_min_length", "recommended_action_max_length"),
):
    if TRIAGE_BOUNDS[minimum_name] > TRIAGE_BOUNDS[maximum_name]:
        raise RuntimeError("signed triage schema minimum exceeds maximum")


class EvidenceItem(BaseModel):
    source: str = Field(min_length=TRIAGE_BOUNDS["source_min_length"], max_length=TRIAGE_BOUNDS["source_max_length"])
    query_name: str = Field(min_length=TRIAGE_BOUNDS["query_name_min_length"], max_length=TRIAGE_BOUNDS["query_name_max_length"])
    observation: str = Field(min_length=TRIAGE_BOUNDS["observation_min_length"], max_length=TRIAGE_BOUNDS["observation_max_length"])
    event_time: str | None = None
    trace_id: str | None = None


class TriageDisposition(BaseModel):
    schema_version: Literal["aidefend.triage.v1"] = "aidefend.triage.v1"
    alert_id: str
    rule_id: str
    summary: str = Field(min_length=TRIAGE_BOUNDS["summary_min_length"], max_length=TRIAGE_BOUNDS["summary_max_length"])
    evidence: list[EvidenceItem] = Field(min_length=TRIAGE_BOUNDS["evidence_min_count"], max_length=TRIAGE_BOUNDS["evidence_max_count"])
    source_consistency: Literal["consistent", "conflicting", "single_source", "unknown"]
    likely_true_positive: Literal["yes", "no", "unknown"]
    affected_assets: list[str] = Field(default_factory=list, max_length=TRIAGE_BOUNDS["affected_assets_max_count"])
    missing_context: list[str] = Field(default_factory=list, max_length=TRIAGE_BOUNDS["missing_context_max_count"])
    recommended_human_action: str = Field(min_length=TRIAGE_BOUNDS["recommended_action_min_length"], max_length=TRIAGE_BOUNDS["recommended_action_max_length"])
    confidence: float = Field(ge=0.0, le=1.0)
    containment_performed: Literal[False] = False

    @field_validator("recommended_human_action")
    @classmethod
    def reject_execution_language(cls, value: str) -> str:
        blocked = ["I quarantined", "I disabled", "I revoked", "I deleted", "auto-containment complete"]
        if any(term.lower() in value.lower() for term in blocked):
            raise ValueError("Triage output must not claim response execution")
        return value


def validate_disposition(raw_json: str) -> TriageDisposition:
    disposition = TriageDisposition.model_validate_json(raw_json)
    if disposition.containment_performed is not False:
        raise ValueError("Read-only triage must not perform containment")
    return disposition

Operational notes: Persist both the raw model output and the validated disposition. Require the agent to mark source_consistency as single_source or conflicting when SIEM, EDR, cloud-audit, and agent-session evidence do not agree; this keeps observability poisoning or compromised logging tools visible to the human reviewer. The execution-language validator is only defense-in-depth because substring checks are easy to bypass; the real read-only guarantee comes from workload identity, the query broker, least-privilege attestation, and downstream systems treating recommended_human_action as non-executable text. Final disposition, severity changes, suppression, and response actions remain human or policy-gated decisions.

Verify complete coverage

The population is every triage result for every assigned alert in the signed rollout, including abstention, insufficient evidence, malformed, timeout, duplicate, error, and missing-result cases.

Verify safely

An independent verifier must revalidate every result with an independent strict parser, resolve every evidence reference read-only, verify no write-side events exist for the triage identity, and compare assignment and terminal-result populations, read evidence back, verify signatures, digests, immutability and effective privileges, and compare the canonical result digest.

AID-D-005.008-G004

Concept:

The safe deployment pattern is narrow and measurable: pick one high-volume rule, run the triage agent without automation authority, compare it to human reviewers for the evaluation window in signed rollout policy, and expand only if quality gates pass. This avoids automating the whole queue before you know where the model is reliable.

Signed-window agreement gate

from __future__ import annotations

# File: triage/shadow_rollout_gate.py
import hashlib
import json
import math
import os
import subprocess
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import hmac
import re
import tempfile




def _aidefend_write_private(path: Path, payload: bytes) -> None:
    descriptor = os.open(
        path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600,
    )
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _aidefend_strict_json(raw: bytes) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str,
    timeout_seconds: float,
    proof_flag: str,
) -> bytes:
    if payload_path == proof_path or proof_flag != "--bundle":
        raise RuntimeError("signed payload/proof inputs are invalid")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "proof"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(proof_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw
def parse_time(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def load_signed_rollout_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["D005008_ROLLOUT_POLICY_PATH"])
    bundle_path = Path(os.environ["D005008_ROLLOUT_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = _aidefend_capture_signed_bytes(
        policy_path, bundle_path, os.environ["D005008_ROLLOUT_POLICY_VERIFY_KEY"],
        verify_timeout, "--bundle",
    )
    policy = _aidefend_strict_json(raw)
    required = {
        "schema_version", "policy_version", "rule_id",
        "evaluation_window_start", "evaluation_window_end",
        "minimum_reviewed_alerts", "minimum_critical_alerts",
        "minimum_agreement_rate", "maximum_critical_miss_rate",
        "minimum_time_saved_ratio",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed rollout policy fields differ")
    if (
        policy["schema_version"] != "aidefend.triage-rollout-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
        or not isinstance(policy["rule_id"], str)
        or not policy["rule_id"]
    ):
        raise RuntimeError("signed rollout policy identity is invalid")
    start = parse_time(policy["evaluation_window_start"])
    end = parse_time(policy["evaluation_window_end"])
    if start.tzinfo is None or end.tzinfo is None or end <= start:
        raise RuntimeError("signed rollout evaluation window is invalid")
    for field in ("minimum_reviewed_alerts", "minimum_critical_alerts"):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise RuntimeError("signed rollout sample bound is invalid: " + field)
    for field in (
        "minimum_agreement_rate", "maximum_critical_miss_rate",
        "minimum_time_saved_ratio",
    ):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or not 0 <= float(value) <= 1
        ):
            raise RuntimeError("signed rollout rate is invalid: " + field)
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


POLICY, POLICY_SHA256 = load_signed_rollout_policy()


@dataclass(frozen=True)
class ReviewedAlert:
    alert_id: str
    rule_id: str
    reviewed_at: str
    ai_disposition: str
    human_disposition: str
    human_marked_critical: bool
    ai_marked_true_positive: bool
    human_minutes_without_ai: float
    human_minutes_with_ai: float


def evaluate_rollout(alerts: list[ReviewedAlert]) -> dict:
    if len({alert.alert_id for alert in alerts}) != len(alerts):
        raise ValueError("shadow alert population contains duplicate IDs")
    start = parse_time(POLICY["evaluation_window_start"])
    end = parse_time(POLICY["evaluation_window_end"])
    dispositions = {"yes", "no", "unknown"}
    for alert in alerts:
        reviewed_at = parse_time(alert.reviewed_at)
        if (
            not alert.alert_id
            or alert.rule_id != POLICY["rule_id"]
            or reviewed_at.tzinfo is None
            or not start <= reviewed_at < end
            or alert.ai_disposition not in dispositions
            or alert.human_disposition not in dispositions
            or any(
                isinstance(value, bool)
                or not isinstance(value, (int, float))
                or not math.isfinite(float(value))
                or float(value) < 0
                for value in (
                    alert.human_minutes_without_ai,
                    alert.human_minutes_with_ai,
                )
            )
        ):
            raise ValueError("shadow alert is malformed or outside signed scope")
    critical = [alert for alert in alerts if alert.human_marked_critical]
    if (
        len(alerts) < POLICY["minimum_reviewed_alerts"]
        or len(critical) < POLICY["minimum_critical_alerts"]
    ):
        return {
            "schema_version": "aidefend.triage-rollout-result.v1",
            "control": "AID-D-005.008",
            "policy_version": POLICY["policy_version"],
            "policy_sha256": POLICY_SHA256,
            "decision": "HOLD",
            "control_status": "INSUFFICIENT_DATA",
            "reviewed_alerts": len(alerts),
            "critical_alerts": len(critical),
        }
    agreement = sum(
        alert.ai_disposition == alert.human_disposition for alert in alerts
    ) / len(alerts)
    missed_critical_rate = sum(
        not alert.ai_marked_true_positive for alert in critical
    ) / len(critical)
    baseline_minutes = math.fsum(
        alert.human_minutes_without_ai for alert in alerts
    )
    assisted_minutes = math.fsum(
        alert.human_minutes_with_ai for alert in alerts
    )
    if baseline_minutes <= 0:
        raise ValueError("baseline review-time population has no positive duration")
    time_saved_ratio = (baseline_minutes - assisted_minutes) / baseline_minutes
    expand = (
        agreement >= float(POLICY["minimum_agreement_rate"])
        and missed_critical_rate <= float(POLICY["maximum_critical_miss_rate"])
        and time_saved_ratio >= float(POLICY["minimum_time_saved_ratio"])
    )
    return {
        "schema_version": "aidefend.triage-rollout-result.v1",
        "control": "AID-D-005.008",
        "policy_version": POLICY["policy_version"],
        "policy_sha256": POLICY_SHA256,
        "rule_id": POLICY["rule_id"],
        "decision": "EXPAND_READ_ONLY" if expand else "HOLD",
        "control_status": "PASS",
        "agreement_rate": agreement,
        "missed_critical_rate": missed_critical_rate,
        "time_saved_ratio": time_saved_ratio,
        "reviewed_alerts": len(alerts),
        "critical_alerts": len(critical),
    }

Action: Store the rollout report with the rule ID, review window, reviewer group, thresholds, and expansion decision. Do not enable autonomous containment as part of this gate; passing the gate only permits broader read-only triage coverage.

Verify safely

An independent verifier must rebuild the alert-label-result join independently, recompute agreement, missed-critical, abstention, error and time metrics with all unlabelled members visible, and compare the signed promotion decision, read evidence back, verify signatures, digests, immutability and effective privileges, and compare the canonical result digest.

Open-source tools

• OpenSearch
• Wazuh
• LangGraph
• LiteLLM
• Pydantic
• OpenTelemetry

Commercial / hosted tools

• StrangeBee TheHive
• Splunk Enterprise Security
• Microsoft Sentinel
• Google Security Operations
• Palo Alto Networks Cortex XSOAR
• ServiceNow Security Operations
• Datadog Cloud SIEM
• Dropzone AI (read-only alert investigation; response automation disabled)

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• N/A

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-2.1.2 Capability and Permission Overreach
• AITech-2.1 Excessive Agency

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Platform 12.3: Lack of incident response