AID-H-031.003: Dynamic Promotion Validation with Ephemeral Sandboxes

Version 1.20261015 · Actionable sub-technique

Apply provenance-aware, risk-tiered dynamic validation to high-risk AI-generated development artifacts before merge, build, or deployment. This stage binds the exact generated change, static-stage result, behavioral-analysis evidence, selected risk tier, and signed validation policy, then emits a dynamic-stage decision for the promotion orchestrator.

Pillars: infra, app

Lifecycle: validation

Scope boundary

Owns provenance-aware routing of AI-generated changes into required behavioral profiles, exact change and report binding, risk-tier sufficiency, and the signed dynamic-stage decision. It consumes rather than reimplements generic sandbox execution.

AID-I-001.005: AID-H-031.003 selects the AI-change behavioral profile and interprets its attested result; AID-I-001.005 owns isolated execution, complete observation, anti-evasion coverage, teardown proof, and report issuance. The dynamic stage must not claim a second sandbox implementation.

AID-H-031.004: AID-H-031.003 emits the routed dynamic-stage result; AID-H-031.004 combines it with provenance, static evidence, policy, and approval to authorize promotion. A clean dynamic result alone never permits merge or deployment.

Implementation guidance

AID-H-031.003-G001

Consume provenance, static-stage, subject-binding, and AID-I-001.005 behavioral evidence under the selected risk tier; emit a signed dynamic-stage result without authorizing promotion.
Boundary and required inputs

This stage does not rerun or re-score a generic sandbox control. It verifies a signed AID-I-001.005 behavioral-analysis evidence object and binds it to the exact AI-generated change routed by AID-H-031.001. It also consumes the signed AID-H-031.002 static-stage result and a build-issued subject manifest that binds the routed change set to the executable artifact analyzed. A PASS here means only that the required dynamic stage completed without a finding under the selected profile. AID-H-031.004 must independently verify this receipt before any merge, build, or deployment authorization.

Sign the dynamic-stage policy

Store this policy on a root-owned control-plane mount and sign it with the admission-policy identity. The values below are an example organizational profile, not universal thresholds.

{
  "schema_version": "aidefend.h031.dynamic-policy.v1",
  "policy_version": "2026.07.17.1",
  "valid_until": "2026-10-01T00:00:00Z",
  "tiers": {
    "generated-privileged": {
      "behavior_profile_id": "generated-privileged-v3",
      "require_static_pass": true,
      "allowed_finding_states": ["NO_FINDING"]
    }
  }
}
Verify upstream evidence and emit only the dynamic-stage decision
# File: validation/evaluate_dynamic_stage.py
from __future__ import annotations

import hashlib
import json
import os
import re
import secrets
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


CONTROL_DIR = Path("/etc/aidefend/h031")
INPUT_DIR = Path("/var/lib/aidefend/h031/input")
OUTPUT_DIR = Path("/var/lib/aidefend/h031/evidence")
ARTIFACT = INPUT_DIR / "dynamic-subject.bin"
FILES = {
    "policy": CONTROL_DIR / "dynamic-policy.json",
    "routing": INPUT_DIR / "routing.json",
    "static": INPUT_DIR / "static-stage.json",
    "subject": INPUT_DIR / "dynamic-subject.json",
    "behavior": INPUT_DIR / "behavioral-analysis.json",
}
VERIFY_KEYS = {
    "policy": CONTROL_DIR / "keys/policy.pub",
    "routing": CONTROL_DIR / "keys/routing.pub",
    "static": CONTROL_DIR / "keys/static-stage.pub",
    "subject": CONTROL_DIR / "keys/build-subject.pub",
    "behavior": CONTROL_DIR / "keys/behavioral-analysis.pub",
}
SIGNING_KEY = Path("/run/secrets/aidefend/h031-dynamic-stage.key")
VERIFICATION_KEY = CONTROL_DIR / "keys/dynamic-stage.pub"
COSIGN = "/usr/local/bin/cosign"
SHA256 = re.compile(r"^[a-f0-9]{64}$")
COMMAND_TIMEOUT_SECONDS = float(os.environ["VERIFIED_H031_COMMAND_TIMEOUT_SECONDS"])
if not 0 < COMMAND_TIMEOUT_SECONDS <= 300:
    raise RuntimeError("verified command timeout is invalid")


class StageStop(Exception):
    def __init__(self, outcome: str, reason: str):
        super().__init__(reason)
        self.outcome = outcome
        self.reason = reason


def canonical(document: dict[str, Any]) -> bytes:
    return json.dumps(
        document, sort_keys=True, separators=(",", ":"),
        ensure_ascii=False, allow_nan=False,
    ).encode("utf-8")


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


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


def verify_blob(path: Path, key: Path) -> bytes:
    signature = Path(str(path) + ".sig")
    if path.is_symlink() or signature.is_symlink() or not path.is_file() or not signature.is_file():
        raise StageStop("INSUFFICIENT_DATA", "required_signed_input_missing")
    raw = path.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h031-signed-input-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [COSIGN, "verify-blob", "--key", str(key),
             "--bundle", str(signature), str(snapshot)],
            check=True, capture_output=True, text=True,
            timeout=COMMAND_TIMEOUT_SECONDS,
        )
        return raw
    except subprocess.CalledProcessError as error:
        raise StageStop("FAIL", "input_signature_invalid") from error
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)


def load_signed(name: str, required_fields: set[str]) -> tuple[dict[str, Any], str]:
    path = FILES[name]
    raw = verify_blob(path, VERIFY_KEYS[name])
    try:
        def unique_pairs(pairs):
            value = {}
            for key, item in pairs:
                if key in value:
                    raise ValueError(f"duplicate {name} key: {key}")
                value[key] = item
            return value

        def reject_constant(value):
            raise ValueError(f"non-finite {name} number: {value}")

        document = json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=unique_pairs, parse_constant=reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
        raise StageStop("FAIL", f"{name}_json_invalid") from error
    if not isinstance(document, dict) or set(document) != required_fields:
        raise StageStop("FAIL", f"{name}_schema_mismatch")
    if raw not in {canonical(document), canonical(document) + b"\n"}:
        raise StageStop("FAIL", f"{name}_not_canonical")
    return document, sha256(canonical(document))


def parse_time(value: Any) -> datetime:
    if not isinstance(value, str):
        raise StageStop("FAIL", "policy_expiry_invalid")
    try:
        result = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error:
        raise StageStop("FAIL", "policy_expiry_invalid") from error
    if result.utcoffset() is None:
        raise StageStop("FAIL", "policy_expiry_timezone_missing")
    return result.astimezone(timezone.utc)


def require_sha256(value: Any, reason: str) -> str:
    if not isinstance(value, str) or not SHA256.fullmatch(value):
        raise StageStop("FAIL", reason)
    return value


def evaluate(context: dict[str, Any]) -> None:
    policy, policy_sha = load_signed(
        "policy", {"schema_version", "policy_version", "valid_until", "tiers"}
    )
    routing, routing_sha = load_signed(
        "routing", {"schema_version", "provenance_sha256", "policy_version",
        "selected_tier", "required_checks", "required_review_roles",
        "base_commit", "head_commit", "change_set_sha256"}
    )
    static, static_sha = load_signed(
        "static", {"schema_version", "outcome", "provenance_sha256",
        "routing_sha256", "change_set_sha256", "check_report_sha256",
        "check_outcomes", "derived_risk_paths"}
    )
    subject, subject_sha = load_signed(
        "subject", {"schema_version", "change_set_sha256", "routing_sha256",
        "artifact_sha256", "artifact_media_type", "build_provenance_sha256",
        "release_id", "issued_at"}
    )
    behavior, behavior_sha = load_signed(
        "behavior", {"schema_version", "control", "tenant_id", "release_id",
        "artifact_sha256", "policy_sha256", "profile_id",
        "analyzer_image_digest", "runtime_profile_ref", "egress_policy_ref",
        "execution_receipt_sha256", "analysis_report_sha256",
        "absence_receipt_sha256", "finding_state", "outcome"}
    )
    context.update(policy_sha256=policy_sha, routing_sha256=routing_sha,
                   static_stage_sha256=static_sha, subject_sha256=subject_sha,
                   behavioral_evidence_sha256=behavior_sha)

    if policy["schema_version"] != "aidefend.h031.dynamic-policy.v1":
        raise StageStop("FAIL", "dynamic_policy_version_unsupported")
    if parse_time(policy["valid_until"]) <= datetime.now(timezone.utc):
        raise StageStop("FAIL", "dynamic_policy_expired")
    if routing["schema_version"] != "aidefend.generated-change-routing.v2":
        raise StageStop("FAIL", "routing_version_unsupported")
    if static["schema_version"] != "aidefend.generated-static-admission.v2":
        raise StageStop("FAIL", "static_stage_version_unsupported")
    if subject["schema_version"] != "aidefend.h031.dynamic-subject.v1":
        raise StageStop("FAIL", "dynamic_subject_version_unsupported")
    if (behavior["schema_version"] != "aidefend.behavioral-analysis-evidence.v1"
            or behavior["control"] != "AID-I-001.005"):
        raise StageStop("FAIL", "behavioral_evidence_version_unsupported")

    if ARTIFACT.is_symlink() or not ARTIFACT.is_file():
        raise StageStop("INSUFFICIENT_DATA", "dynamic_subject_artifact_missing")
    artifact_sha = sha256_file(ARTIFACT)
    require_sha256(subject["artifact_sha256"], "subject_artifact_digest_invalid")
    if artifact_sha != subject["artifact_sha256"] or artifact_sha != behavior["artifact_sha256"]:
        raise StageStop("FAIL", "artifact_behavior_binding_mismatch")
    if (subject["change_set_sha256"] != routing["change_set_sha256"]
            or subject["routing_sha256"] != routing_sha
            or static["routing_sha256"] != routing_sha
            or static["change_set_sha256"] != routing["change_set_sha256"]
            or static["provenance_sha256"] != routing["provenance_sha256"]
            or behavior["release_id"] != subject["release_id"]):
        raise StageStop("FAIL", "upstream_evidence_binding_mismatch")

    tiers = policy["tiers"]
    tier = tiers.get(routing["selected_tier"]) if isinstance(tiers, dict) else None
    required_tier_fields = {
        "behavior_profile_id", "require_static_pass", "allowed_finding_states"
    }
    if not isinstance(tier, dict) or set(tier) != required_tier_fields:
        raise StageStop("FAIL", "selected_dynamic_tier_missing")
    allowed = tier["allowed_finding_states"]
    if allowed != sorted(set(allowed)) or not allowed or any(
        value not in {"NO_FINDING", "FINDING", "INCONCLUSIVE", "ERROR"}
        for value in allowed
    ):
        raise StageStop("FAIL", "allowed_finding_states_invalid")
    if behavior["profile_id"] != tier["behavior_profile_id"]:
        raise StageStop("FAIL", "behavior_profile_mismatch")
    if tier["require_static_pass"] is True and static["outcome"] != "pass":
        raise StageStop("FAIL", "static_stage_did_not_pass")
    if behavior["finding_state"] not in allowed or behavior["outcome"] != "PASS":
        raise StageStop("FAIL", "dynamic_behavior_did_not_pass")


def publish(record: dict[str, Any]) -> tuple[Path, str]:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True, mode=0o750)
    name = "dynamic-stage-" + secrets.token_urlsafe(18)
    temporary = OUTPUT_DIR / ("." + name + ".json.tmp")
    temporary_signature = Path(str(temporary) + ".sig")
    final = OUTPUT_DIR / (name + ".json")
    final_signature = Path(str(final) + ".sig")
    payload = canonical(record) + b"\n"
    with temporary.open("xb") as handle:
        os.fchmod(handle.fileno(), 0o640)
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
    subprocess.run(
        [COSIGN, "sign-blob", "--yes", "--key", str(SIGNING_KEY),
         "--bundle", str(temporary_signature), str(temporary)],
        check=True, capture_output=True, text=True,
        timeout=COMMAND_TIMEOUT_SECONDS,
    )
    verify_blob(temporary, VERIFICATION_KEY)
    os.replace(temporary, final)
    os.replace(temporary_signature, final_signature)
    if final.read_bytes() != payload:
        raise OSError("dynamic-stage receipt readback mismatch")
    verify_blob(final, VERIFICATION_KEY)
    return final, sha256(payload)


def main() -> int:
    context = {
        "policy_sha256": None, "routing_sha256": None,
        "static_stage_sha256": None, "subject_sha256": None,
        "behavioral_evidence_sha256": None,
    }
    outcome, reason = "PASS", "required_dynamic_stage_verified"
    try:
        evaluate(context)
    except StageStop as error:
        outcome, reason = error.outcome, error.reason
    except (OSError, subprocess.SubprocessError):
        outcome, reason = "ERROR", "trusted_verifier_dependency_error"
    except Exception:
        outcome, reason = "ERROR", "unexpected_dynamic_stage_error"
    record = {
        "schema_version": "aidefend.h031.dynamic-stage-decision.v1",
        "control": "AID-H-031.003",
        "outcome": outcome,
        "reason_code": reason,
        **context,
        "issued_at": datetime.now(timezone.utc).isoformat(),
    }
    path, digest = publish(record)
    print(json.dumps({"outcome": outcome, "receipt": str(path), "sha256": digest}))
    return 0 if outcome == "PASS" else 2


if __name__ == "__main__":
    raise SystemExit(main())
Promotion integration and independent verification

The dynamic-stage signer must be unavailable to the sandbox runner, build worker, AI agent, and final promotion service. AID-H-031.004 verifies the signed receipt, recomputes every referenced digest, and requires the exact outcome PASS; no report or receipt from this stage is itself an ALLOW credential. Test a benign generated artifact plus forbidden file, process, network, delayed-execution, incomplete-case, wrong-profile, changed-artifact, forged-signature, stale-policy, missing-static-stage, residue, and verifier-outage fixtures. Each negative fixture must either produce a non-PASS receipt or prevent the final promotion gate from authorizing the change.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-031.003: Dynamic Promotion Validation with Ephemeral Sandboxes | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service