Scope boundary

Owns non-executing AST, SAST, and policy analysis of exact executable bytes immediately before merge, build, or execution, plus the pre-execution block decision. It cannot prove arbitrary code safe.

AID-H-025.002: AID-H-025.001 blocks known dangerous constructs without running the code; AID-H-025.002 selects a restricted interpreter when dynamic evaluation is still required. A clean static result does not remove the need for runtime language restrictions.

AID-I-001: AID-H-025.001 is a pre-execution analysis gate; AID-I-001 owns containment of the permitted execution process, filesystem, network, and resources. Static analysis cannot establish runtime isolation.

AID-H-031.002: AID-H-025.001 provides generic dangerous-code analysis; AID-H-031.002 uses that result within provenance-aware promotion policy for AI-generated development artifacts. The scanner verdict is one input and does not authorize merge or promotion.

Implementation guidance

• AID-H-025.001-G001: Enforce a signed-policy AST admission gate on the exact Python artifact immediately before execution.
• AID-H-025.001-G002: Run a release-pinned Semgrep ruleset as a CI and just-in-time SAST gate with signed, replayable evidence.

AID-H-025.001-G001

Action: enforce the complete control unit

This is a complete Python execution-admission gate: verify a release-pinned signed policy, read one bounded UTF-8 regular file without following a symlink, parse without execution, resolve import aliases, apply exact/prefix call rules, sign a verdict bound to the execution request and bytes, then require the launcher to re-hash those bytes immediately before a separate sandbox starts.

# File: static_gate/python_ast_gate.py
from __future__ import annotations

import ast
import hashlib
import json
import math
import os
import re
import stat
import subprocess
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from tempfile import NamedTemporaryFile

COSIGN_TIMEOUT_SECONDS = float(os.environ["VERIFIED_H025_AST_COSIGN_TIMEOUT_SECONDS"])
if not math.isfinite(COSIGN_TIMEOUT_SECONDS) or COSIGN_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("verified H025 AST cosign timeout must be finite and positive")

POLICY_FIELDS = {
    "schema_version", "policy_version", "maximum_bytes",
    "forbidden_modules", "forbidden_calls", "forbidden_call_prefixes",
    "scanner_image_digest", "verdict_ttl_seconds",
}
SHA256 = re.compile(r"^[a-f0-9]{64}$")
IMAGE_DIGEST = re.compile(r"^.+@sha256:[a-f0-9]{64}$")


@dataclass(frozen=True)
class Finding:
    rule_id: str
    construct: str
    line: int
    column: int


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


def sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def load_policy() -> tuple[dict, str]:
    path = Path(os.environ["H025_AST_POLICY"])
    raw = path.read_bytes()
    actual = sha256(raw)
    if actual != os.environ["H025_AST_POLICY_SHA256"]:
        raise RuntimeError("policy_digest_mismatch")
    with NamedTemporaryFile(prefix="aidefend-h025-policy-", suffix=".json", delete=False) as handle:
        handle.write(raw)
        snapshot = Path(handle.name)
    os.chmod(snapshot, 0o600)
    try:
        subprocess.run(
            [
                "cosign", "verify-blob",
                "--key", os.environ["H025_AST_POLICY_VERIFY_KEY"],
                "--bundle", os.environ["H025_AST_POLICY_SIG"], str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
        )
    finally:
        snapshot.unlink(missing_ok=True)
    policy = json.loads(raw)
    if (
        not isinstance(policy, dict)
        or set(policy) != POLICY_FIELDS
        or policy["schema_version"] != "aidefend.h025001-ast-policy.v1"
        or not isinstance(policy["maximum_bytes"], int)
        or isinstance(policy["maximum_bytes"], bool)
        or policy["maximum_bytes"] < 1
        or not isinstance(policy["verdict_ttl_seconds"], int)
        or isinstance(policy["verdict_ttl_seconds"], bool)
        or not 1 <= policy["verdict_ttl_seconds"] <= 3600
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
        or not isinstance(policy["scanner_image_digest"], str)
        or not IMAGE_DIGEST.fullmatch(policy["scanner_image_digest"])
    ):
        raise ValueError("policy_schema_invalid")
    for field in ("forbidden_modules", "forbidden_calls", "forbidden_call_prefixes"):
        values = policy[field]
        if not isinstance(values, list) or values != sorted(set(values)):
            raise ValueError(f"{field}_must_be_sorted_unique_list")
    return policy, actual


def dotted(node: ast.AST, aliases: dict[str, str]) -> str | None:
    if isinstance(node, ast.Name):
        return aliases.get(node.id, node.id)
    if isinstance(node, ast.Attribute):
        parent = dotted(node.value, aliases)
        return f"{parent}.{node.attr}" if parent else None
    return None


def scan(source: str, policy: dict) -> list[Finding]:
    try:
        tree = ast.parse(source, mode="exec", type_comments=True)
    except SyntaxError as exc:
        return [Finding("syntax_error", str(exc.msg), exc.lineno or 0, exc.offset or 0)]

    aliases: dict[str, str] = {}
    findings: list[Finding] = []
    forbidden_modules = set(policy["forbidden_modules"])
    forbidden_calls = set(policy["forbidden_calls"])
    prefixes = tuple(policy["forbidden_call_prefixes"])

    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for item in node.names:
                aliases[item.asname or item.name.split(".", 1)[0]] = item.name
                if item.name.split(".", 1)[0] in forbidden_modules:
                    findings.append(Finding(
                        "forbidden_import", item.name,
                        getattr(node, "lineno", 0), getattr(node, "col_offset", 0),
                    ))
        elif isinstance(node, ast.ImportFrom):
            module = node.module or ""
            if module.split(".", 1)[0] in forbidden_modules:
                findings.append(Finding(
                    "forbidden_import", module,
                    getattr(node, "lineno", 0), getattr(node, "col_offset", 0),
                ))
            for item in node.names:
                aliases[item.asname or item.name] = f"{module}.{item.name}".strip(".")

    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        name = dotted(node.func, aliases)
        if name is None:
            findings.append(Finding(
                "dynamic_call_target", ast.dump(node.func, include_attributes=False),
                getattr(node, "lineno", 0), getattr(node, "col_offset", 0),
            ))
        elif name in forbidden_calls or any(
            name == prefix or name.startswith(prefix + ".") for prefix in prefixes
        ):
            findings.append(Finding(
                "forbidden_call", name,
                getattr(node, "lineno", 0), getattr(node, "col_offset", 0),
            ))
    return sorted(findings, key=lambda item: (item.line, item.column, item.rule_id, item.construct))


def read_bounded_regular(path: Path, maximum_bytes: int) -> bytes:
    if not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_CLOEXEC"):
        raise RuntimeError("secure_no_follow_open_is_unavailable")
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        before = os.fstat(fd)
        if not stat.S_ISREG(before.st_mode):
            raise PermissionError("artifact_is_not_a_regular_file")
        if before.st_size > maximum_bytes:
            raise PermissionError("artifact_exceeds_policy_limit")
        chunks: list[bytes] = []
        remaining = maximum_bytes + 1
        while remaining:
            chunk = os.read(fd, min(1024 * 1024, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        raw = b"".join(chunks)
        after = os.fstat(fd)
        stable = (
            before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns
        ) == (
            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns
        )
        if not stable or len(raw) != after.st_size:
            raise PermissionError("artifact_changed_while_scanning")
        if len(raw) > maximum_bytes:
            raise PermissionError("artifact_exceeds_policy_limit")
        return raw
    finally:
        os.close(fd)


def sign_verdict(output: Path, verdict: dict) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    payload = canonical(verdict)
    fd = os.open(
        output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600
    )
    try:
        os.write(fd, payload)
        os.fsync(fd)
    finally:
        os.close(fd)
    subprocess.run(
        [
            "cosign", "sign-blob", "--yes",
            "--key", "env://H025_EVIDENCE_SIGNING_KEY",
            "--bundle", str(output.with_suffix(".sig")), str(output),
        ],
        check=True, timeout=COSIGN_TIMEOUT_SECONDS,
    )


def gate(artifact: Path, *, request_id: str, output: Path) -> dict:
    if not isinstance(request_id, str) or not request_id:
        raise ValueError("request_id_is_required")
    path_sha = sha256(os.fsencode(str(artifact.absolute())))
    policy: dict | None = None
    policy_sha: str | None = os.environ.get("H025_AST_POLICY_SHA256")
    raw: bytes | None = None
    now = datetime.now(timezone.utc)
    try:
        policy, policy_sha = load_policy()
        if not SHA256.fullmatch(policy_sha):
            raise ValueError("policy_digest_is_invalid")
        raw = read_bounded_regular(artifact, policy["maximum_bytes"])
        findings = scan(raw.decode("utf-8", errors="strict"), policy)
        outcome = "FAIL" if findings else "PASS"
        reason = "policy_findings" if findings else "no_policy_findings"
    except Exception as exc:
        findings = []
        outcome = "ERROR"
        reason = type(exc).__name__

    ttl = policy["verdict_ttl_seconds"] if policy else 1
    verdict = {
        "schema_version": "aidefend.h025001-ast-verdict.v2",
        "control_id": "AID-H-025.001",
        "request_id": request_id,
        "artifact_path_sha256": path_sha,
        "artifact_sha256": sha256(raw) if raw is not None else None,
        "artifact_size": len(raw) if raw is not None else None,
        "policy_version": policy["policy_version"] if policy else None,
        "policy_sha256": policy_sha if isinstance(policy_sha, str) else None,
        "scanner_image_digest": policy["scanner_image_digest"] if policy else None,
        "issued_at": now.isoformat(),
        "expires_at": (now + timedelta(seconds=ttl)).isoformat(),
        "outcome": outcome,
        "reason": reason,
        "findings": [asdict(item) for item in findings],
    }
    sign_verdict(output, verdict)
    return verdict

Enforcement, verification, and evidence

Replay direct/aliased imports and calls, attribute chains, dynamic targets, syntax/Unicode/size failures, symlinks and special files, concurrent replacement, policy rollback, invalid signatures, expiry, and signer failure. Retain exact signed policy, artifact/policy/scanner digests, request/trace IDs, findings, signed verdict, sandbox launch receipt, complete launch population, and independent readback result. The AST rule coverage is explicit and bounded; PASS is not proof that arbitrary code is safe, so OS/container/WASM containment remains mandatory.

AID-H-025.001-G002

Action: enforce the complete control unit

Scan the complete signed target manifest in CI or immediately before interactive execution. Verify a signed policy and exact Semgrep/ruleset versions, reject target drift and symlinks, run without network access, classify every rule through policy rather than a universal severity threshold, sign the result, and block merge or execution unless the exact target population receives PASS.

Production Semgrep gate

# File: scanning/run_semgrep.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import stat
import subprocess
from pathlib import Path

COSIGN_TIMEOUT_SECONDS = float(os.environ["VERIFIED_H025_SEMGREP_COSIGN_TIMEOUT_SECONDS"])
VERSION_CHECK_TIMEOUT_SECONDS = float(os.environ["VERIFIED_H025_SEMGREP_VERSION_TIMEOUT_SECONDS"])
if any(
    not math.isfinite(value) or value <= 0
    for value in (COSIGN_TIMEOUT_SECONDS, VERSION_CHECK_TIMEOUT_SECONDS)
):
    raise RuntimeError("verified Semgrep gate timeouts are invalid")


def sha256_file(path: Path) -> str:
    if not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_CLOEXEC"):
        raise RuntimeError("secure_no_follow_open_is_unavailable")
    digest = hashlib.sha256()
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        before = os.fstat(fd)
        if not stat.S_ISREG(before.st_mode):
            raise ValueError("Semgrep population contains a non-regular file")
        with os.fdopen(fd, "rb", closefd=False) as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
        after = os.fstat(fd)
        if (
            before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns
        ) != (
            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns
        ):
            raise RuntimeError("Semgrep population changed while hashing")
    finally:
        os.close(fd)
    return digest.hexdigest()


def sha256_target(target: Path) -> str:
    if target.is_symlink():
        raise ValueError("Semgrep target may not be a symlink")
    if target.is_file():
        return sha256_file(target)
    if not target.is_dir():
        raise ValueError("Semgrep target must be a file or directory")
    digest = hashlib.sha256()
    entries = sorted(target.rglob("*"), key=lambda item: item.as_posix())
    if any(item.is_symlink() for item in entries):
        raise ValueError("Semgrep target contains a symlink")
    if any(not item.is_file() and not item.is_dir() for item in entries):
        raise ValueError("Semgrep target contains a special file")
    files = [item for item in entries if item.is_file()]
    if not files:
        raise ValueError("Semgrep target directory is empty")
    for file_path in files:
        relative = file_path.relative_to(target).as_posix().encode("utf-8")
        digest.update(len(relative).to_bytes(8, "big"))
        digest.update(relative)
        digest.update(bytes.fromhex(sha256_file(file_path)))
    return digest.hexdigest()

def write_evidence(path: Path, evidence: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(
        evidence, sort_keys=True, separators=(",", ":")
    ) + "\n"
    with path.open("x", encoding="utf-8", newline="") as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
    subprocess.run(
        [
            "cosign", "sign-blob", "--yes",
            "--key", "env://H025_EVIDENCE_SIGNING_KEY",
            "--bundle", str(path.with_suffix(".sig")), str(path),
        ],
        check=True, timeout=COSIGN_TIMEOUT_SECONDS,
    )


def run_semgrep(
    target: str, evidence_path: str, *, request_id: str,
) -> dict:
    if not isinstance(request_id, str) or not request_id:
        raise ValueError("request_id is required")
    target_path = Path(target)
    base_evidence = {
        "control_id": "AID-H-025.001",
        "request_id": request_id,
        "target_path_sha256": hashlib.sha256(
            os.fsencode(str(target_path.absolute()))
        ).hexdigest(),
        "expected_policy_sha256": os.environ.get("H025_SEMGREP_POLICY_SHA256"),
    }
    policy_path = Path(os.environ["H025_SEMGREP_POLICY_PATH"])
    try:
        policy_raw = policy_path.read_bytes()
        policy_sha256 = hashlib.sha256(policy_raw).hexdigest()
        if policy_sha256 != os.environ["H025_SEMGREP_POLICY_SHA256"]:
            raise RuntimeError("Semgrep policy digest mismatch")
        subprocess.run(
            [
                "cosign", "verify-blob",
                "--key", os.environ["H025_SEMGREP_POLICY_VERIFY_KEY"],
                "--bundle", os.environ["H025_SEMGREP_POLICY_SIG"],
                str(policy_path),
            ],
            check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
        )
        policy = json.loads(policy_raw)
        required = {
            "schema_version", "policy_version", "semgrep_version",
            "ruleset_sha256", "target_sha256", "maximum_scan_seconds",
            "rule_actions", "unknown_rule_action", "scanner_image_digest",
            "container_limits",
        }
        if not isinstance(policy, dict) or set(policy) != required:
            raise ValueError("Semgrep policy schema mismatch")
        if policy["schema_version"] != "aidefend.h025001-semgrep-policy.v1":
            raise ValueError("unsupported Semgrep policy schema")
        if policy["unknown_rule_action"] not in {"block", "record"}:
            raise ValueError("unknown rule action is invalid")
        if not isinstance(policy["rule_actions"], dict) or any(
            action not in {"block", "record"}
            for action in policy["rule_actions"].values()
        ):
            raise ValueError("Semgrep rule actions are invalid")
        maximum_scan_seconds = policy["maximum_scan_seconds"]
        if (
            isinstance(maximum_scan_seconds, bool)
            or not isinstance(maximum_scan_seconds, int)
            or maximum_scan_seconds < 1
        ):
            raise ValueError("maximum scan time is invalid")
        limits = policy["container_limits"]
        if (
            not isinstance(limits, dict)
            or set(limits) != {"memory", "cpus", "pids", "tmpfs_bytes"}
            or not isinstance(limits["memory"], str)
            or not re.fullmatch(r"[1-9][0-9]*(?:[mMgG])", limits["memory"])
            or not isinstance(limits["cpus"], str)
            or not re.fullmatch(r"[1-9][0-9]*(?:[.][0-9]+)?", limits["cpus"])
            or isinstance(limits["pids"], bool)
            or not isinstance(limits["pids"], int)
            or not 1 <= limits["pids"] <= 4096
            or isinstance(limits["tmpfs_bytes"], bool)
            or not isinstance(limits["tmpfs_bytes"], int)
            or not 1048576 <= limits["tmpfs_bytes"] <= 2147483648
        ):
            raise ValueError("signed container limits are invalid")
        scanner_image = policy["scanner_image_digest"]
        if (
            not isinstance(scanner_image, str)
            or not re.fullmatch(r".+@sha256:[a-f0-9]{64}", scanner_image)
        ):
            raise ValueError("scanner image must be digest-pinned")
        expected_version = policy["semgrep_version"]
        ruleset_path = Path(os.environ["SEMGREP_RULESET_PATH"])
        expected_ruleset_sha256 = policy["ruleset_sha256"].lower()
        expected_target_sha256 = policy["target_sha256"].lower()
        actual_target_sha256 = sha256_target(Path(target))
        if actual_target_sha256 != expected_target_sha256:
            raise RuntimeError("Semgrep target population digest mismatch")
        base_evidence.update({
            "policy_version": policy["policy_version"],
            "policy_sha256": policy_sha256,
            "target_sha256": actual_target_sha256,
            "ruleset_sha256": expected_ruleset_sha256,
            "scanner_image_digest": scanner_image,
            "container_limits": limits,
        })
    except Exception as exc:
        evidence = {
            "schema_version": "aidefend.h025001-semgrep-verdict.v2",
            **base_evidence,
            "outcome": "ERROR",
            "reason": "Semgrep signed policy or target verification failed",
            "error_class": type(exc).__name__,
        }
        write_evidence(Path(evidence_path), evidence)
        raise RuntimeError(evidence["reason"]) from exc
    try:
        actual_ruleset_sha256 = sha256_target(ruleset_path)
    except Exception as exc:
        evidence = {
            "schema_version": "aidefend.h025001-semgrep-verdict.v2",
            **base_evidence,
            "outcome": "ERROR",
            "reason": "Semgrep ruleset could not be measured",
            "error_class": type(exc).__name__,
        }
        write_evidence(Path(evidence_path), evidence)
        raise RuntimeError(evidence["reason"]) from exc
    if (
        len(expected_ruleset_sha256) != 64
        or actual_ruleset_sha256 != expected_ruleset_sha256
    ):
        evidence = {
            "schema_version": "aidefend.h025001-semgrep-verdict.v2",
            **base_evidence,
            "outcome": "ERROR",
            "reason": "Semgrep ruleset digest verification failed",
            "expected_ruleset_sha256": expected_ruleset_sha256,
            "actual_ruleset_sha256": actual_ruleset_sha256,
        }
        write_evidence(Path(evidence_path), evidence)
        raise RuntimeError(evidence["reason"])
    sandbox = [
        "docker", "run", "--rm", "--pull=never", "--network=none",
        "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges",
        "--user=65532:65532", "--pids-limit", str(limits["pids"]),
        "--memory", limits["memory"], "--cpus", limits["cpus"],
        "--tmpfs", f"/tmp:rw,noexec,nosuid,nodev,size={limits['tmpfs_bytes']}",
        "--mount", f"type=bind,src={target_path.absolute()},dst=/scan,readonly",
        "--mount", f"type=bind,src={ruleset_path.absolute()},dst=/rules,readonly",
        scanner_image,
    ]
    version_result = subprocess.run(
        sandbox + ["semgrep", "--version"],
        capture_output=True, text=True, timeout=VERSION_CHECK_TIMEOUT_SECONDS,
    )
    actual_version = version_result.stdout.strip()
    if version_result.returncode != 0 or actual_version != expected_version:
        evidence = {
            "schema_version": "aidefend.h025001-semgrep-verdict.v2",
            **base_evidence,
            "outcome": "ERROR",
            "reason": "Semgrep version verification failed",
            "expected_version": expected_version,
            "actual_version": actual_version or None,
            "version_exit_code": version_result.returncode,
        }
        write_evidence(Path(evidence_path), evidence)
        raise RuntimeError(evidence["reason"])

    command = sandbox + [
        "semgrep", "scan", "--config", "/rules", "--json",
        "--error", "--metrics=off", "/scan",
    ]
    try:
        result = subprocess.run(
            command, capture_output=True, text=True, timeout=maximum_scan_seconds
        )
    except subprocess.TimeoutExpired as exc:
        evidence = {
            "schema_version": "aidefend.h025001-semgrep-verdict.v2",
            **base_evidence,
            "outcome": "ERROR",
            "reason": "Semgrep timed out",
            "semgrep_version": actual_version,
            "timeout_seconds": maximum_scan_seconds,
        }
        write_evidence(Path(evidence_path), evidence)
        raise RuntimeError(evidence["reason"]) from exc

    parse_error = None
    try:
        report = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        report = {}
        parse_error = type(exc).__name__
    findings = report.get("results", [])
    errors = report.get("errors", [])
    schema_valid = isinstance(findings, list) and isinstance(errors, list)
    action_by_rule = policy["rule_actions"]
    default_action = policy["unknown_rule_action"]
    blocking = [
        item for item in findings
        if isinstance(item, dict)
        and action_by_rule.get(item.get("check_id"), default_action) == "block"
    ] if schema_valid else []

    execution_error = (
        result.returncode not in {0, 1}
        or parse_error is not None
        or not schema_valid
        or bool(errors)
        or (result.returncode == 0 and bool(findings))
        or (result.returncode == 1 and not findings)
    )
    outcome = "ERROR" if execution_error else ("FAIL" if blocking else "PASS")
    evidence = {
        "schema_version": "aidefend.h025001-semgrep-verdict.v2",
        **base_evidence,
        "outcome": outcome,
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "semgrep_version": actual_version,
        "ruleset_sha256": actual_ruleset_sha256,
        "target_sha256": sha256_target(Path(target)),
        "exit_code": result.returncode,
        "stdout_sha256": hashlib.sha256(result.stdout.encode()).hexdigest(),
        "stderr_sha256": hashlib.sha256(result.stderr.encode()).hexdigest(),
        "parse_error": parse_error,
        "errors": errors if schema_valid else [],
        "findings": findings if schema_valid else [],
        "blocking_finding_count": len(blocking),
    }
    write_evidence(Path(evidence_path), evidence)
    if execution_error:
        raise RuntimeError("Semgrep execution or evidence validation failed")
    if blocking:
        raise RuntimeError(f"Semgrep found {len(blocking)} blocking findings")
    return evidence

Enforcement, outcomes, and retained evidence

Retain the signed policy and ruleset/target/scanner digests, exact Semgrep version and immutable image, raw output digests, normalized findings and policy actions, execution/CI request, signed verdict, downstream merge/launch receipt, and complete-population reconciliation. Independently replay findings, unknown rules, scanner errors, malformed output, timeouts, target add/remove/mutation, symlink/path escape, policy/ruleset rollback, signature failure, and post-scan mutation. Never convert a missing scan or unsupported target into PASS.

Open-source tools

• Semgrep Community Edition
• Bandit (Python)
• ESLint
• eslint-plugin-security
• tree-sitter (multi-language parsing)
• Cosign
• LlamaFirewall CodeShield (static analysis of LLM-generated code)

Commercial / hosted tools

• GitHub Advanced Security
• Snyk Code
• Veracode
• Checkmarx One
• SonarQube Enterprise

Threat mappings

MITRE ATLAS

• AML.T0110 AI Agent Tool Poisoning
• AML.T0115 Publish Poisoned AI Artifacts
• AML.T0115.002 Publish Poisoned AI Artifacts: AI Agent Tools
• AML.T0110.001 AI Agent Tool Poisoning: Implementation
• AML.T0050 Command and Scripting Interpreter
• AML.T0072 Cyber Communication Channel (static policy blocks recognized reverse-shell and callback constructs in generated code before execution)
• AML.T0102 Generate Malicious Commands (analyzes generated commands and code before use)
• AML.T0105 Escape to Host (blocks recognized host-escape primitives before execution)

MAESTRO

• N/A

OWASP LLM Top 10 2026

• LLM10:2026 Improper Output Handling

OWASP ML Top 10 2023

• ML09:2023 Output Integrity Attack (validates executable model output before use)

OWASP Top 10 for Agentic Applications 2026

• ASI05:2026 Unexpected Code Execution (RCE)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.3 Unsafe System / Browser / File Execution
• AISubtech-12.2.1 Code Detection / Malicious Code Output
• AISubtech-9.1.1 Code Execution
• AITech-12.1 Tool Exploitation
• AITech-12.2 Insecure Output Handling
• AITech-9.1 Model or Agentic System Manipulation

Google Secure AI Framework 2.0 - Risks

• IMO: Insecure Model Output (analyzes executable model output before use)

Databricks AI Security Framework 3.0

• Agents - Core 13.11: Unexpected RCE and Code Attacks
• Platform 12.5: Poor security in the software development lifecycle