AID-D-012.001: Controlled GNN Backdoor Trigger-Response Testing

Version 1.20261015 · Actionable sub-technique

Evaluate a candidate GNN with paired clean and boundedly perturbed graph cases generated from an explicit threat model. Measure targeted attack success rate, prediction flips, target-class margin shift, clean utility, and bounded edge-drop instability against a trusted reference model and integrity-protected test artifacts.

Technical applicability: this control applies to an exact candidate that consumes graph-structured inputs and can be executed with the preregistered perturbation operators. Non-graph models are outside its technical scope. For a graph candidate, a missing contract, reference, corpus, eligible-case manifest, runner, or result must remain distinct from a completed measurement.

Pillars: model, data

Lifecycle: validation

Scope boundary

Owns validation-time paired clean and bounded perturbation testing of an exact candidate GNN against preregistered trigger operators, graph corpus, seeds, targets, budgets, reference model, and thresholds. It does not discover arbitrary unknown triggers or prove the model clean.

AID-D-012.002: AID-D-012.001 measures candidate-model response to preregistered bounded trigger and perturbation cases; AID-D-012.002 detects anomalous relationships between node features across observed edges. Trigger-response testing evaluates model behavior, while structure-feature detection evaluates graph data consistency.

AID-D-012.003: AID-D-012.001 measures prediction and utility changes under controlled perturbations; AID-D-012.003 detects degree and betweenness-centrality outliers from graph topology alone. A topology outlier is not a backdoor trigger-response result, and controlled trigger testing does not exhaust unknown topology anomalies.

AID-M-003.006: AID-D-012.001 executes the signed trigger-test contract against the candidate and reference; AID-M-003.006 creates and approves the graph-specific reference artifacts. The test consumes the baseline but does not author or recalibrate it.

Implementation guidance

AID-D-012.001-G001: Verify a signed, expiring GNN trigger-test contract that binds the candidate, reference, held-out graph corpus, bounded perturbations, runner image, seeds, eligibility rules, and preregistered thresholds.
AID-D-012.001-G002: Execute paired clean and boundedly perturbed candidate/reference cases, report the measured decision for the exact graph population, and preserve excluded, unmeasured, and execution-failure cases.
AID-D-012.001-G003: For an applicable graph-input candidate, replay the exact test in a separately credentialed, digest-pinned isolated runner and retain signed raw and aggregate results in append-only storage.
AID-D-012.001-G004: Apply a pinned-trust production envelope that binds directed graph semantics and enforces the signed cap on unmeasured eligible rows before accepting the evaluation for promotion.

AID-D-012.001-G001

Verify a signed, expiring GNN trigger-test contract that binds the candidate, reference, held-out graph corpus, bounded perturbations, runner image, seeds, eligibility rules, and preregistered thresholds.

Applicability and one-generation boundary

Use this implementation only for a graph-input candidate identified by the deployment asset inventory. A non-graph asset is NOT_APPLICABLE before invocation; a graph candidate with missing or unverifiable inputs is ERROR, never N/A. The signed contract binds the exact candidate, trusted reference, held-out graph corpus, eligible-case set, retention policy, perturbation operators, thresholds, seeds, and digest-pinned runner.

Capture and verify the complete contract once

from __future__ import annotations

# File: gnn_security/contract.py

import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from types import MappingProxyType

OPERATORS = {"feature_patch", "edge_injection", "random_edge_drop"}
UNITS = {
    "feature_patch": "linf",
    "edge_injection": "edge_count",
    "random_edge_drop": "fraction",
}
THRESHOLDS = {
    "max_targeted_asr_delta",
    "max_flip_rate_delta",
    "max_target_margin_shift_delta",
    "max_clean_accuracy_drop",
    "max_edge_drop_instability_delta",
}
REQUIRED = {
    "schema_version", "policy_version", "candidate_sha256",
    "reference_sha256", "graph_corpus_sha256", "runner_image_digest",
    "eligible_case_ids_sha256", "operators", "budgets_by_operator",
    "budget_units_by_operator", "max_budget_by_operator", "seeds",
    "target_class", "thresholds", "min_eligible_count",
    "min_eligible_per_operator", "min_non_target_eligible_per_operator",
    "issued_at", "expires_at", "retention_policy_sha256",
}
SHA256 = re.compile(r"^[0-9a-f]{64}$")
OCI_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
VERIFY_TIMEOUT_SECONDS = float(os.environ["D012_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(VERIFY_TIMEOUT_SECONDS) or VERIFY_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("signature-verification timeout is invalid")


class ContractError(ValueError):
    pass


def strict_json(raw: bytes) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ContractError("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 token: (_ for _ in ()).throw(
            ContractError("non-finite JSON constant: " + token)
        ),
    )


def _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 _local_key_bytes(reference: str) -> bytes | None:
    if reference.startswith("-----BEGIN "):
        return reference.encode("utf-8")
    if bool(re.match(r"^[A-Za-z][A-Za-z0-9+.-]*://", reference)) \
            or reference.startswith("pkcs11:"):
        return None
    candidate = Path(reference)
    if candidate.is_symlink():
        raise ContractError("verification key must not be a symlink")
    resolved = candidate.resolve(strict=True)
    if not resolved.is_file() or resolved.stat().st_mode & 0o022:
        raise ContractError("verification key must be a non-writable regular file")
    return resolved.read_bytes()


def capture_signed_bytes(
    payload_path: Path,
    bundle_path: Path,
    key_reference: str | Path,
) -> bytes:
    payload_path = Path(payload_path)
    bundle_path = Path(bundle_path)
    if payload_path.resolve(strict=True) == bundle_path.resolve(strict=True):
        raise ContractError("signed payload and bundle paths collide")
    payload_raw = payload_path.read_bytes()
    bundle_raw = bundle_path.read_bytes()
    key_text = str(key_reference)
    key_raw = _local_key_bytes(key_text)
    with tempfile.TemporaryDirectory(prefix="aidefend-d012-contract-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        bundle_snapshot = root / "payload.bundle"
        _write_private(payload_snapshot, payload_raw)
        _write_private(bundle_snapshot, bundle_raw)
        captured = [(payload_snapshot, payload_raw), (bundle_snapshot, bundle_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _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(bundle_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=VERIFY_TIMEOUT_SECONDS,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise ContractError("private verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw


def parse_time(value: object) -> datetime:
    if not isinstance(value, str):
        raise ContractError("timestamp is not text")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error:
        raise ContractError("timestamp is invalid") from error
    if parsed.utcoffset() is None:
        raise ContractError("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def positive_int(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ContractError(f"{field} must be a positive integer")
    return value


def finite_number(value: object, field: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ContractError(f"{field} must be numeric")
    number = float(value)
    if not math.isfinite(number):
        raise ContractError(f"{field} must be finite")
    return number


@dataclass(frozen=True)
class VerifiedContractInputs:
    manifest_raw: bytes
    manifest_sha256: str
    eligible_case_ids: tuple[str, ...]
    artifacts: Mapping[str, bytes]

    @property
    def manifest(self) -> dict:
        document = strict_json(self.manifest_raw)
        if not isinstance(document, dict):
            raise ContractError("verified manifest stopped being an object")
        return document

    def artifact(self, digest_field: str) -> bytes:
        try:
            return self.artifacts[digest_field]
        except KeyError as error:
            raise ContractError("unknown verified artifact: " + digest_field) from error

    def assert_fresh(self, now: datetime | None = None) -> None:
        manifest = self.manifest
        issued_at = parse_time(manifest["issued_at"])
        expires_at = parse_time(manifest["expires_at"])
        current = now or datetime.now(timezone.utc)
        if issued_at > current or expires_at <= issued_at or current >= expires_at:
            raise ContractError("contract is not currently valid")


def verify_contract(
    manifest_path: Path,
    bundle_path: Path,
    public_key_path: Path,
    *,
    candidate_path: Path,
    reference_path: Path,
    corpus_path: Path,
    eligible_case_ids_path: Path,
    retention_policy_path: Path,
    now: datetime | None = None,
) -> VerifiedContractInputs:
    manifest_raw = capture_signed_bytes(
        manifest_path, bundle_path, public_key_path
    )
    manifest = strict_json(manifest_raw)
    if not isinstance(manifest, dict) or set(manifest) != REQUIRED:
        raise ContractError("contract schema mismatch")
    if manifest["schema_version"] != "aidefend.gnn_trigger_contract.v1":
        raise ContractError("unsupported contract schema version")
    if not isinstance(manifest["policy_version"], str) or not manifest["policy_version"]:
        raise ContractError("missing contract policy version")

    artifact_paths = {
        "candidate_sha256": candidate_path,
        "reference_sha256": reference_path,
        "graph_corpus_sha256": corpus_path,
        "eligible_case_ids_sha256": eligible_case_ids_path,
        "retention_policy_sha256": retention_policy_path,
    }
    artifacts = {
        field: Path(path).read_bytes() for field, path in artifact_paths.items()
    }
    for field, raw in artifacts.items():
        expected = manifest[field]
        if not isinstance(expected, str) or not SHA256.fullmatch(expected):
            raise ContractError(f"{field} is not a SHA-256 digest")
        if hashlib.sha256(raw).hexdigest() != expected:
            raise ContractError(f"{field} artifact mismatch")
    if not OCI_DIGEST.fullmatch(str(manifest["runner_image_digest"])):
        raise ContractError("runner image is not digest-pinned")

    operators = manifest["operators"]
    if (
        not isinstance(operators, list)
        or len(operators) != len(set(operators))
        or set(operators) != OPERATORS
    ):
        raise ContractError("operators must contain each supported operator exactly once")
    budgets = manifest["budgets_by_operator"]
    units = manifest["budget_units_by_operator"]
    maxima = manifest["max_budget_by_operator"]
    if not all(
        isinstance(value, dict) and set(value) == OPERATORS
        for value in (budgets, units, maxima)
    ) or units != UNITS:
        raise ContractError("operator policy keys or units differ")
    for operator in operators:
        maximum = finite_number(maxima[operator], f"max_budget_by_operator.{operator}")
        if maximum <= 0 or (operator == "random_edge_drop" and maximum > 1):
            raise ContractError("operator maximum is outside its valid range")
        values = budgets[operator]
        if not isinstance(values, list) or not values:
            raise ContractError(f"{operator} has no budgets")
        canonical = []
        for value in values:
            number = finite_number(value, f"budgets_by_operator.{operator}")
            if number <= 0 or number > maximum:
                raise ContractError(f"{operator} budget is outside signed bound")
            if operator == "edge_injection" and number != int(number):
                raise ContractError("edge-injection budgets must be integral")
            canonical.append(number)
        if len(canonical) != len(set(canonical)):
            raise ContractError(f"{operator} budgets are duplicated")

    seeds = manifest["seeds"]
    if (
        not isinstance(seeds, list) or not seeds
        or any(isinstance(seed, bool) or not isinstance(seed, int) or seed < 0 for seed in seeds)
        or len(seeds) != len(set(seeds))
    ):
        raise ContractError("seeds must be unique non-negative integers")
    target_class = manifest["target_class"]
    if isinstance(target_class, bool) or not isinstance(target_class, int) or target_class < 0:
        raise ContractError("target_class must be a non-negative integer")
    positive_int(manifest["min_eligible_count"], "min_eligible_count")
    positive_int(manifest["min_eligible_per_operator"], "min_eligible_per_operator")
    positive_int(
        manifest["min_non_target_eligible_per_operator"],
        "min_non_target_eligible_per_operator",
    )

    thresholds = manifest["thresholds"]
    if not isinstance(thresholds, dict) or set(thresholds) != THRESHOLDS:
        raise ContractError("threshold schema mismatch")
    for field, value in thresholds.items():
        number = finite_number(value, f"thresholds.{field}")
        if number < 0 or number > 1:
            raise ContractError(f"{field} must be within [0,1]")

    eligible_document = strict_json(artifacts["eligible_case_ids_sha256"])
    if (
        not isinstance(eligible_document, list) or not eligible_document
        or any(not isinstance(case_id, str) or not case_id for case_id in eligible_document)
        or len(eligible_document) != len(set(eligible_document))
    ):
        raise ContractError("eligible case ID set is invalid")
    verified = VerifiedContractInputs(
        manifest_raw=manifest_raw,
        manifest_sha256=hashlib.sha256(manifest_raw).hexdigest(),
        eligible_case_ids=tuple(eligible_document),
        artifacts=MappingProxyType(dict(artifacts)),
    )
    verified.assert_fresh(now)
    return verified

Use and verify the returned snapshot

The verifier returns immutable bytes for every digest-bound artifact plus the exact manifest digest and eligible population. The executor and scorer must consume VerifiedContractInputs; reopening any source path creates a different, unverified generation. Keep the verification key and timeout in the root-owned service definition, retain the bundle and manifest digest, and verify contract freshness again immediately before a final PASS.

Exercise valid directed and undirected fixtures plus an expired or future contract, duplicate JSON keys, NaN/Infinity, empty or duplicate eligible IDs, unknown or duplicate operators, excessive or duplicate budgets, corrupt bundle/key, changed candidate/reference/corpus/retention bytes, and same-path generation replacement inside one worker. Every malformed or substituted input must fail closed, while a second invocation at the same path must capture the new signed generation rather than reuse stale process-global bytes.

AID-D-012.001-G002

Execute paired clean and boundedly perturbed candidate/reference cases, report the measured decision for the exact graph population, and preserve excluded, unmeasured, and execution-failure cases.

Score one captured generation

The runner must produce one row for every signed corpus case, operator, budget, and seed; non-eligible cases remain explicit EXCLUDED rows and eligible execution failures remain UNMEASURED. The scorer accepts the invocation-scoped contract snapshot from the previous guidance and one captured result byte string. It never reopens a corpus, manifest, eligible-case, or result path while computing evidence.

from __future__ import annotations

# File: gnn_security/score_trigger_response.py

import hashlib
import math
from pathlib import Path
from statistics import fmean

from gnn_security.contract import (
    ContractError,
    VerifiedContractInputs,
    strict_json,
    verify_contract,
)

IDENTITY_FIELDS = {"case_id", "operator", "budget", "seed", "status"}
METRIC_FIELDS = {
    "ground_truth_label", "candidate_clean_label", "reference_clean_label",
    "candidate_perturbed_label", "reference_perturbed_label",
    "candidate_clean_target_margin", "reference_clean_target_margin",
    "candidate_perturbed_target_margin", "reference_perturbed_target_margin",
}
ROW_FIELDS = IDENTITY_FIELDS | METRIC_FIELDS
STATUSES = {"MEASURED", "UNMEASURED", "EXCLUDED"}


def finite(value, field):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{field} must be numeric")
    number = float(value)
    if not math.isfinite(number):
        raise ValueError(f"{field} must be finite")
    return number


def label(value, field):
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{field} must be a non-negative integer")
    return value


def normalized_budget(value):
    number = finite(value, "budget")
    return int(number) if number.is_integer() else number


def parse_result_rows(raw: bytes) -> list[dict]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("result JSONL is empty or contains a blank record")
    rows = []
    for line_number, line in enumerate(lines, 1):
        row = strict_json(line.encode("utf-8"))
        if not isinstance(row, dict) or set(row) != ROW_FIELDS:
            raise ValueError(f"row {line_number} schema mismatch")
        if (
            not isinstance(row["case_id"], str) or not row["case_id"]
            or not isinstance(row["operator"], str) or not row["operator"]
            or isinstance(row["seed"], bool) or not isinstance(row["seed"], int)
            or row["seed"] < 0 or row["status"] not in STATUSES
        ):
            raise ValueError(f"row {line_number} identity field invalid")
        row["budget"] = normalized_budget(row["budget"])
        if row["status"] == "MEASURED":
            for field in (
                "ground_truth_label", "candidate_clean_label",
                "reference_clean_label", "candidate_perturbed_label",
                "reference_perturbed_label",
            ):
                row[field] = label(row[field], field)
            for field in (
                "candidate_clean_target_margin", "reference_clean_target_margin",
                "candidate_perturbed_target_margin",
                "reference_perturbed_target_margin",
            ):
                row[field] = finite(row[field], field)
                if not -1 <= row[field] <= 1:
                    raise ValueError(f"{field} is outside normalized margin range")
        elif any(row[field] is not None for field in METRIC_FIELDS):
            raise ValueError("unmeasured/excluded row must contain null metrics")
        rows.append(row)
    return rows


def corpus_case_ids(raw: bytes) -> list[str]:
    corpus = strict_json(raw)
    case_ids = corpus.get("case_ids") if isinstance(corpus, dict) else None
    if (
        not isinstance(case_ids, list) or not case_ids
        or any(not isinstance(case_id, str) or not case_id for case_id in case_ids)
        or len(case_ids) != len(set(case_ids))
    ):
        raise ContractError("verified corpus manifest has invalid case_ids")
    return case_ids


def _base_result(
    verified: VerifiedContractInputs,
    results_raw: bytes,
) -> dict:
    manifest = verified.manifest
    return {
        "schema_version": "aidefend.gnn_trigger_result.v1",
        "policy_version": manifest["policy_version"],
        "manifest_sha256": verified.manifest_sha256,
        "raw_results_sha256": hashlib.sha256(results_raw).hexdigest(),
        "eligible_case_count": len(verified.eligible_case_ids),
    }


def score_verified(
    verified: VerifiedContractInputs,
    results_raw: bytes,
) -> dict:
    manifest = verified.manifest
    all_cases = corpus_case_ids(verified.artifact("graph_corpus_sha256"))
    eligible = set(verified.eligible_case_ids)
    if not eligible.issubset(all_cases):
        raise ContractError("eligible cases are not a subset of verified corpus")

    expected = {
        (case_id, operator, normalized_budget(budget), seed)
        for case_id in all_cases
        for operator in manifest["operators"]
        for budget in manifest["budgets_by_operator"][operator]
        for seed in manifest["seeds"]
    }
    observed = {}
    for row in parse_result_rows(results_raw):
        key = (row["case_id"], row["operator"], row["budget"], row["seed"])
        if key not in expected:
            raise ValueError(f"unknown matrix row: {key}")
        if key in observed:
            raise ValueError(f"duplicate matrix row: {key}")
        should_measure = row["case_id"] in eligible
        if should_measure and row["status"] == "EXCLUDED":
            raise ValueError("verified eligible case was marked excluded")
        if not should_measure and row["status"] != "EXCLUDED":
            raise ValueError("non-eligible case was not marked excluded")
        if row["status"] == "MEASURED" and not should_measure:
            raise ValueError("non-eligible case contains measurements")
        observed[key] = row

    missing = sorted(expected - set(observed))
    if missing:
        return {
            **_base_result(verified, results_raw),
            "status": "INSUFFICIENT_DATA",
            "reason": "expected matrix rows missing",
            "missing_row_count": len(missing),
            "first_missing_rows": missing[:20],
        }

    measured = [row for row in observed.values() if row["status"] == "MEASURED"]
    measured_case_ids = {row["case_id"] for row in measured}
    target_class = manifest["target_class"]
    per_operator_case_ids = {
        operator: {
            row["case_id"] for row in measured if row["operator"] == operator
        }
        for operator in manifest["operators"]
    }
    per_operator_non_target_case_ids = {
        operator: {
            row["case_id"] for row in measured
            if row["operator"] == operator
            and row["ground_truth_label"] != target_class
        }
        for operator in manifest["operators"]
    }
    if (
        len(measured_case_ids) < manifest["min_eligible_count"]
        or any(
            len(case_ids) < manifest["min_eligible_per_operator"]
            for case_ids in per_operator_case_ids.values()
        )
        or any(
            len(case_ids) < manifest["min_non_target_eligible_per_operator"]
            for case_ids in per_operator_non_target_case_ids.values()
        )
    ):
        return {
            **_base_result(verified, results_raw),
            "status": "INSUFFICIENT_DATA",
            "reason": "eligible/non-target coverage below signed minimum",
            "measured_case_count": len(measured_case_ids),
            "measured_case_count_by_operator": {
                key: len(value) for key, value in per_operator_case_ids.items()
            },
            "non_target_case_count_by_operator": {
                key: len(value)
                for key, value in per_operator_non_target_case_ids.items()
            },
        }

    clean_by_case = {}
    for row in measured:
        clean = (
            row["ground_truth_label"], row["candidate_clean_label"],
            row["reference_clean_label"], row["candidate_clean_target_margin"],
            row["reference_clean_target_margin"],
        )
        previous = clean_by_case.setdefault(row["case_id"], clean)
        if previous != clean:
            raise ValueError("clean measurement changed across paired rows")

    operator_metrics = {}
    for operator in manifest["operators"]:
        rows = [row for row in measured if row["operator"] == operator]
        non_target_rows = [
            row for row in rows if row["ground_truth_label"] != target_class
        ]
        candidate_asr = fmean(
            row["candidate_perturbed_label"] == target_class
            for row in non_target_rows
        )
        reference_asr = fmean(
            row["reference_perturbed_label"] == target_class
            for row in non_target_rows
        )
        candidate_flip = fmean(
            row["candidate_perturbed_label"] != row["candidate_clean_label"]
            for row in rows
        )
        reference_flip = fmean(
            row["reference_perturbed_label"] != row["reference_clean_label"]
            for row in rows
        )
        candidate_margin_shift = fmean(
            row["candidate_perturbed_target_margin"]
            - row["candidate_clean_target_margin"]
            for row in non_target_rows
        )
        reference_margin_shift = fmean(
            row["reference_perturbed_target_margin"]
            - row["reference_clean_target_margin"]
            for row in non_target_rows
        )
        operator_metrics[operator] = {
            "measured_row_count": len(rows),
            "non_target_measured_row_count": len(non_target_rows),
            "candidate_targeted_asr": candidate_asr,
            "reference_targeted_asr": reference_asr,
            "targeted_asr_delta": max(0.0, candidate_asr - reference_asr),
            "candidate_flip_rate": candidate_flip,
            "reference_flip_rate": reference_flip,
            "flip_rate_delta": max(0.0, candidate_flip - reference_flip),
            "candidate_target_margin_shift": candidate_margin_shift,
            "reference_target_margin_shift": reference_margin_shift,
            "target_margin_shift_delta": max(
                0.0, candidate_margin_shift - reference_margin_shift
            ),
        }

    clean_values = list(clean_by_case.values())
    candidate_accuracy = fmean(value[1] == value[0] for value in clean_values)
    reference_accuracy = fmean(value[2] == value[0] for value in clean_values)
    clean_accuracy_drop = max(0.0, reference_accuracy - candidate_accuracy)
    thresholds = manifest["thresholds"]
    breaches = []
    for operator, metrics in operator_metrics.items():
        for metric, threshold_name in (
            ("targeted_asr_delta", "max_targeted_asr_delta"),
            ("flip_rate_delta", "max_flip_rate_delta"),
            ("target_margin_shift_delta", "max_target_margin_shift_delta"),
        ):
            if metrics[metric] > thresholds[threshold_name]:
                breaches.append(
                    {"operator": operator, "metric": metric, "value": metrics[metric]}
                )
    edge_drop_delta = operator_metrics["random_edge_drop"]["flip_rate_delta"]
    if edge_drop_delta > thresholds["max_edge_drop_instability_delta"]:
        breaches.append({
            "operator": "random_edge_drop",
            "metric": "edge_drop_instability_delta",
            "value": edge_drop_delta,
        })
    if clean_accuracy_drop > thresholds["max_clean_accuracy_drop"]:
        breaches.append({
            "operator": "clean",
            "metric": "clean_accuracy_drop",
            "value": clean_accuracy_drop,
        })

    status = "FAIL" if breaches else "PASS"
    if status == "PASS":
        verified.assert_fresh()
    return {
        **_base_result(verified, results_raw),
        "status": status,
        "measured_case_count": len(measured_case_ids),
        "excluded_row_count": sum(
            row["status"] == "EXCLUDED" for row in observed.values()
        ),
        "unmeasured_eligible_row_count": sum(
            row["status"] == "UNMEASURED" for row in observed.values()
        ),
        "operator_metrics": operator_metrics,
        "clean_accuracy_drop": clean_accuracy_drop,
        "breaches": breaches,
    }


def safe_score_verified(
    verified: VerifiedContractInputs,
    results_raw: bytes,
) -> dict:
    try:
        return score_verified(verified, results_raw)
    except Exception as error:
        return {
            "schema_version": "aidefend.gnn_trigger_result.v1",
            "status": "ERROR",
            "reason": "contract, artifact, or result verification failed",
            "manifest_sha256": verified.manifest_sha256,
            "raw_results_sha256": hashlib.sha256(results_raw).hexdigest(),
            "error_class": type(error).__name__,
        }


def score(
    *,
    manifest_path: Path,
    bundle_path: Path,
    public_key_path: Path,
    candidate_path: Path,
    reference_path: Path,
    corpus_path: Path,
    eligible_case_ids_path: Path,
    retention_policy_path: Path,
    results_path: Path,
) -> dict:
    verified = verify_contract(
        manifest_path, bundle_path, public_key_path,
        candidate_path=candidate_path,
        reference_path=reference_path,
        corpus_path=corpus_path,
        eligible_case_ids_path=eligible_case_ids_path,
        retention_policy_path=retention_policy_path,
    )
    results_raw = Path(results_path).read_bytes()
    return score_verified(verified, results_raw)


def safe_score(**kwargs) -> dict:
    try:
        return score(**kwargs)
    except Exception as error:
        return {
            "schema_version": "aidefend.gnn_trigger_result.v1",
            "status": "ERROR",
            "reason": "contract, artifact, or result verification failed",
            "error_class": type(error).__name__,
        }

Replay and reconcile

Retain the verified contract inputs, complete raw row bytes, matrix-coverage result, raw candidate/reference metrics, aggregate result, and exception class. Independently replay the same manifest digest and raw-result digest and compare the canonical aggregate. Test duplicate or unknown matrix keys, invalid UTF-8, duplicate JSON keys, NaN/Infinity, blank records, null measured metrics, non-null unmeasured metrics, missing rows, changed clean values, insufficient eligible/non-target populations, exact-threshold values, and each threshold breach. A missing row or too-small measured population is INSUFFICIENT_DATA; malformed or mismatched evidence is ERROR; any measured breach remains FAIL.

This stage owns paired trigger-response measurement. It emits findings and evidence for the production envelope; it does not make a deployment release decision or reopen any source artifact.

AID-D-012.001-G003

For an applicable graph-input candidate, replay the exact test in a separately credentialed, digest-pinned isolated runner and retain signed raw and aggregate results in append-only storage.

Independent replay

For a graph-input candidate admitted to D-012.001, a separately credentialed CI job must verify the exact signed contract, run its digest-pinned image with no network and read-only inputs, replay the same perturbation records and seeds, and compare canonical raw outputs and the aggregate verdict. Non-graph assets are excluded before this implementation path; a missing runner or result for an admitted graph candidate is ERROR.

set -euo pipefail
INPUT_SNAPSHOT="$(mktemp -d)"
trap 'rm -rf -- "$INPUT_SNAPSHOT"' EXIT
chmod 0700 "$INPUT_SNAPSHOT"
for source in evidence/test-manifest.json evidence/test-manifest.sig evidence/perturbation-records.jsonl evidence/paired-results.jsonl; do
  test -f "$source"
  test ! -L "$source"
  install -m 0400 -- "$source" "$INPUT_SNAPSHOT/$(basename "$source")"
done
cosign verify-blob \
  --key trust/gnn-test-manifest.pub \
  --bundle "$INPUT_SNAPSHOT/test-manifest.sig" \
  "$INPUT_SNAPSHOT/test-manifest.json"
RUNNER_DIGEST="$(python - "$INPUT_SNAPSHOT/test-manifest.json" <<'PY'
import json
import re
import sys
from pathlib import Path

def unique_object(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate JSON key: {key}")
        value[key] = item
    return value

def reject_constant(value):
    raise ValueError(f"non-finite JSON constant: {value}")

manifest = json.loads(
    Path(sys.argv[1]).read_text(encoding="utf-8", errors="strict"),
    object_pairs_hook=unique_object,
    parse_constant=reject_constant,
)
digest = manifest.get("runner_image_digest") if isinstance(manifest, dict) else None
if not isinstance(digest, str) or re.fullmatch(r"sha256:[0-9a-f]{64}", digest) is None:
    raise SystemExit("signed runner image digest is absent or invalid")
print(digest)
PY
)"
RUNNER_IMAGE="${GNN_RUNNER_IMAGE:?set a full approved OCI repository@sha256 digest reference}"
case "$RUNNER_IMAGE" in
  *@"$RUNNER_DIGEST") ;;
  *) echo "GNN_RUNNER_IMAGE digest does not match the signed manifest" >&2; exit 1 ;;
esac
mkdir -p replay
cosign verify --key trust/gnn-runner.pub "$RUNNER_IMAGE" > replay/runner-image-verification.json
docker run --rm --network=none --read-only \
  --tmpfs /tmp:rw,noexec,nosuid,size=512m \
  -v "$PWD/artifacts:/work/artifacts:ro" \
  -v "$PWD/references:/work/references:ro" \
  -v "$INPUT_SNAPSHOT:/work/input:ro" \
  -v "$PWD/replay:/work/output:rw" \
  "$RUNNER_IMAGE" replay \
  --manifest /work/input/test-manifest.json \
  --perturbations /work/input/perturbation-records.jsonl \
  --output /work/output
python gnn_security/verify_replay_equivalence.py \
  --original "$INPUT_SNAPSHOT/paired-results.jsonl" \
  --replay replay/paired-results.jsonl \
  --manifest "$INPUT_SNAPSHOT/test-manifest.json" \
  --receipt replay/replay-receipt.json
cosign sign-blob --yes --key env://GNN_REPLAY_SIGNING_KEY \
  --bundle replay/replay-receipt.sig replay/replay-receipt.json

Bind the replay receipt to every artifact digest, seed, runner digest, raw-output digest, and verdict. Store the manifest, signatures, raw predictions, perturbation records, results, replay receipt, and exceptions in append-only or WORM storage for the duration selected by the signed organization/legal/incident retention policy bound in the test contract; record the applied policy version and disposition date with the evidence. A completed replay that validly diverges from the original canonical outputs or verdict is FAIL; a missing or failed replay is ERROR.

Run the detector and route findings

Action: independently rerun the complete contract-bound graph test with the exact pinned runner, inputs, perturbations and seeds, compare canonical raw outputs/verdicts, sign the replay receipt and retain it append-only.

Boundary: Re-verify the schema and original paired-execution receipt as authenticated inputs. This stage owns independent reproducibility and equivalence; it does not overwrite the original detection verdict.

AID-D-012.001-G004

Apply a pinned-trust production envelope that binds directed graph semantics and enforces the signed cap on unmeasured eligible rows before accepting the evaluation for promotion.

Accept one shared verified generation

The scorer is not a release gate by itself. The production wrapper pins trust keys in the verifier image, calls the contract verifier once for the complete candidate/reference/corpus/eligibility/retention set, captures the raw result bytes once, and passes those exact objects to the scorer. A signed execution policy binds graph semantics, the perturbation generator, retention policy, and a maximum unmeasured fraction for every operator. Missing or over-cap coverage is INSUFFICIENT_DATA; malformed or mismatched evidence is ERROR; a measured trigger-response finding remains FAIL.

from __future__ import annotations

# File: gnn_security/production_acceptance.py

import hashlib
import math
import re
from pathlib import Path

from gnn_security.contract import (
    ContractError,
    capture_signed_bytes,
    strict_json,
    verify_contract,
)
from gnn_security.score_trigger_response import (
    parse_result_rows,
    safe_score_verified,
)


CONTRACT_KEY = Path("/opt/aidefend/trust/gnn-test-contract.pub")
EXECUTION_POLICY_KEY = Path("/opt/aidefend/trust/gnn-execution-policy.pub")
RUNNER_RECEIPT_KEY = Path("/opt/aidefend/trust/gnn-runner-receipt.pub")
POLICY_FIELDS = {
    "schema_version", "manifest_sha256", "graph_semantics",
    "max_unmeasured_fraction_by_operator",
    "perturbation_generator_sha256", "retention_policy_sha256",
}
RECEIPT_FIELDS = {
    "schema_version", "manifest_sha256", "execution_policy_sha256",
    "graph_semantics", "perturbation_generator_sha256",
    "raw_results_sha256", "runner_image_digest",
}
SHA256 = re.compile(r"^[0-9a-f]{64}$")


def verify_json(
    path: Path,
    bundle: Path,
    key: Path,
    fields: set[str],
) -> tuple[dict, str]:
    raw = capture_signed_bytes(path, bundle, key)
    document = strict_json(raw)
    if not isinstance(document, dict) or set(document) != fields:
        raise ValueError("signed production-envelope schema mismatch")
    return document, hashlib.sha256(raw).hexdigest()


def production_score(
    *,
    manifest_path: Path,
    manifest_bundle_path: Path,
    candidate_path: Path,
    reference_path: Path,
    corpus_path: Path,
    eligible_case_ids_path: Path,
    retention_policy_path: Path,
    execution_policy_path: Path,
    execution_policy_bundle_path: Path,
    runner_receipt_path: Path,
    runner_receipt_bundle_path: Path,
    results_path: Path,
) -> dict:
    required = {
        "manifest_path": manifest_path,
        "manifest_bundle_path": manifest_bundle_path,
        "candidate_path": candidate_path,
        "reference_path": reference_path,
        "corpus_path": corpus_path,
        "eligible_case_ids_path": eligible_case_ids_path,
        "retention_policy_path": retention_policy_path,
        "execution_policy_path": execution_policy_path,
        "execution_policy_bundle_path": execution_policy_bundle_path,
        "runner_receipt_path": runner_receipt_path,
        "runner_receipt_bundle_path": runner_receipt_bundle_path,
        "results_path": results_path,
    }
    missing = sorted(name for name, value in required.items() if value is None)
    if missing:
        raise ValueError("production acceptance lacks: " + ", ".join(missing))

    verified = verify_contract(
        Path(manifest_path),
        Path(manifest_bundle_path),
        CONTRACT_KEY,
        candidate_path=Path(candidate_path),
        reference_path=Path(reference_path),
        corpus_path=Path(corpus_path),
        eligible_case_ids_path=Path(eligible_case_ids_path),
        retention_policy_path=Path(retention_policy_path),
    )
    manifest = verified.manifest
    policy, policy_sha256 = verify_json(
        Path(execution_policy_path),
        Path(execution_policy_bundle_path),
        EXECUTION_POLICY_KEY,
        POLICY_FIELDS,
    )
    receipt, receipt_sha256 = verify_json(
        Path(runner_receipt_path),
        Path(runner_receipt_bundle_path),
        RUNNER_RECEIPT_KEY,
        RECEIPT_FIELDS,
    )
    results_raw = Path(results_path).read_bytes()
    if not results_raw:
        raise ValueError("raw result population is empty")
    results_sha256 = hashlib.sha256(results_raw).hexdigest()

    if policy["schema_version"] != "aidefend.gnn-trigger-execution-policy.v1":
        raise ValueError("unsupported execution policy")
    if receipt["schema_version"] != "aidefend.gnn-runner-receipt.v1":
        raise ValueError("unsupported runner receipt")
    if policy["manifest_sha256"] != verified.manifest_sha256:
        raise ValueError("execution policy is bound to another manifest")
    if policy["graph_semantics"] not in {"DIRECTED", "UNDIRECTED"}:
        raise ValueError("graph semantics are not explicit")
    for field in ("perturbation_generator_sha256", "retention_policy_sha256"):
        if not isinstance(policy[field], str) or not SHA256.fullmatch(policy[field]):
            raise ValueError("execution-policy digest is invalid: " + field)
    if policy["retention_policy_sha256"] != manifest["retention_policy_sha256"]:
        raise ValueError("execution policy names another retention policy")

    operators = set(manifest["operators"])
    caps = policy["max_unmeasured_fraction_by_operator"]
    if not isinstance(caps, dict) or set(caps) != operators:
        raise ValueError("unmeasured-fraction policy does not cover all operators")
    for operator, value in caps.items():
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or not 0.0 <= float(value) <= 1.0
        ):
            raise ValueError("invalid unmeasured cap for " + operator)

    expected_receipt = {
        "manifest_sha256": verified.manifest_sha256,
        "execution_policy_sha256": policy_sha256,
        "graph_semantics": policy["graph_semantics"],
        "perturbation_generator_sha256": policy["perturbation_generator_sha256"],
        "raw_results_sha256": results_sha256,
        "runner_image_digest": manifest["runner_image_digest"],
    }
    if any(receipt.get(field) != value for field, value in expected_receipt.items()):
        raise ValueError("runner receipt binding differs")

    # Run the full scorer first so duplicate, unknown, or malformed rows are ERROR,
    # not misreported as a mere coverage shortfall.
    result = safe_score_verified(verified, results_raw)
    envelope_binding = {
        "execution_policy_sha256": policy_sha256,
        "runner_receipt_sha256": receipt_sha256,
        "graph_semantics": policy["graph_semantics"],
        "retention_policy_sha256": policy["retention_policy_sha256"],
    }
    if result["status"] == "ERROR":
        result.update(envelope_binding)
        return result

    eligible = set(verified.eligible_case_ids)
    if not eligible:
        raise ContractError("verified eligible population is empty")
    rows = parse_result_rows(results_raw)
    expected_per_operator = {
        operator: len(eligible)
        * len(manifest["budgets_by_operator"][operator])
        * len(manifest["seeds"])
        for operator in operators
    }
    if any(value <= 0 for value in expected_per_operator.values()):
        raise ContractError("eligible operator denominator is empty")
    unmeasured = {operator: 0 for operator in operators}
    observed_eligible = {operator: 0 for operator in operators}
    for row in rows:
        operator = row["operator"]
        if operator in operators and row["case_id"] in eligible:
            observed_eligible[operator] += 1
            if row["status"] == "UNMEASURED":
                unmeasured[operator] += 1

    coverage = {}
    coverage_failures = []
    for operator in sorted(operators):
        expected = expected_per_operator[operator]
        observed = observed_eligible[operator]
        fraction = unmeasured[operator] / expected
        coverage[operator] = {
            "expected_rows": expected,
            "observed_rows": observed,
            "unmeasured_rows": unmeasured[operator],
            "unmeasured_fraction": fraction,
            "signed_cap": float(caps[operator]),
        }
        if observed != expected:
            coverage_failures.append({
                "operator": operator,
                "reason": "eligible matrix coverage differs",
            })
        elif fraction > float(caps[operator]):
            coverage_failures.append({
                "operator": operator,
                "reason": "unmeasured eligible fraction exceeds signed cap",
            })

    verified.assert_fresh()
    result.update(envelope_binding)
    result["coverage_by_operator"] = coverage
    result["coverage_failures"] = coverage_failures
    if coverage_failures and result["status"] != "FAIL":
        result["status"] = "INSUFFICIENT_DATA"
        result["reason"] = "production coverage gate is incomplete"
    return result


def safe_production_score(**kwargs) -> dict:
    try:
        return production_score(**kwargs)
    except Exception as error:
        return {
            "schema_version": "aidefend.gnn_trigger_result.v1",
            "status": "ERROR",
            "reason": "production envelope verification failed",
            "error_class": type(error).__name__,
        }

Replay and retain the acceptance evidence

Test directed and undirected policies, a manifest/policy mismatch, changed candidate/reference/corpus/eligibility/retention bytes, substituted results, duplicate or unknown rows, an empty eligible denominator, missing rows, each per-operator cap boundary, expired contracts, and receipt mismatches. Retain the captured signed inputs, raw rows, scorer result, coverage table, and replay receipt in the authenticated evidence path. An independent runner must reproduce the raw-result digest and aggregate result before a PASS is accepted. The wrapper never converts a detector FAIL to a clean or insufficient result.

This stage owns production-envelope acceptance for graph trigger-response scanning. Asset applicability is resolved before invocation; it does not replace the detector or the independent replay.

Open-source tools

PyTorch Geometric
Deep Graph Library (DGL)
DeepRobust (graph perturbation and attack generation for bounded test cases; case definitions and scoring remain external)
safetensors
NumPy
SciPy
pytest
Cosign

Threat mappings

MITRE ATLAS

AML.T0018 Manipulate AI Model
AML.T0018.000 Manipulate AI Model: Poison AI Model
AML.T0020 Training Data Poisoning
AML.T0043.004 Craft Adversarial Data: Insert Backdoor Trigger

MAESTRO

Backdoor Attacks (L1)
Data Poisoning (Training Phase) (L1)
Data Poisoning (L2)

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

ML02:2023 Data Poisoning Attack
ML10:2023 Model Poisoning

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

NISTAML.021 Clean-label Backdoor
NISTAML.023 Backdoor Poisoning
NISTAML.024 Targeted Poisoning
NISTAML.026 Model Poisoning (Integrity)

Cisco Integrated AI Security and Safety Framework

AISubtech-9.2.2 Backdoors and Trojans
AITech-6.1 Training Data Poisoning (backdoor-related behavior under preregistered bounded graph perturbations)

Google Secure AI Framework 2.0 - Risks

DP: Data Poisoning
MST: Model Source Tampering

Databricks AI Security Framework 3.0

Datasets 3.1: Data poisoning
Model 7.1: Backdoor machine learning / Trojaned model