AID-M-003.006: Graph-Structured Model Behavioral Baselining
Creates signed, version-bound baselines for graph-structured AI models and datasets, including graph snapshot digests, node and edge schema, deterministic node ordering, structural metric profiles, and GNN probe-node output references. This Model-side technique owns reference artifact creation, periodic recalibration, and empirical validation that selected graph metrics or probe outputs are meaningful for the graph family.
Pillars: model
Lifecycle: validation, operation
Scope boundary
Owns signed graph snapshots, node and edge schemas, deterministic node ordering, structural metric profiles, GNN probe-node output references, recalibration, and validation that selected graph signals are meaningful.
AID-M-003.002: AID-M-003.006 owns graph-structured and GNN-specific reference artifacts; AID-M-003.002 owns general predictive and operational references. Graph topology, schema, node order, and probe outputs cannot be reduced to generic model KPIs without losing the governed graph population.
AID-D-012: AID-M-003.006 creates and validates graph and GNN reference artifacts; AID-D-012 applies live trigger-response, topology, and graph-anomaly detection against those references. A reference artifact is not an alert, and live detection must not redefine its own baseline.
AID-D-016.001: AID-M-003.006 baselines graph-structured model and dataset behavior; AID-D-016.001 discovers and monitors the runtime communication graph of cooperating agents. A GNN data graph and an agent interaction graph may use similar metrics but represent different nodes, edges, owners, and security claims.
Implementation guidance
- AID-M-003.006-G001: Create and verify signed, expiring, version-bound GNN probe-node output baseline bundles using safe tensor serialization.
- AID-M-003.006-G002: Create signed graph-structure metric baselines that bind graph snapshots, node order, schema, and selected structural fingerprints.
- AID-M-003.006-G003: Validate graph structural metrics against paired, bounded perturbation experiments with actual frozen-model inference before using those metrics in Detect-side monitoring; probe-output baselines are defined separately.
AID-M-003.006-G001
Create and verify signed, expiring, version-bound GNN probe-node output baseline bundles using safe tensor serialization.
Concept:
A graph-structured model probe bundle is a signed validation reference, not proof that a GNN or self-supervised model is clean. Keep only numeric arrays in SafeTensors, compute model and graph digests from the actual files, enforce the exact tensor schema, and bind each current output to the same model, graph, and node order with a short-lived receipt signed by a protected inference-attestation identity.
Step 1: Build the baseline from reviewed validation arrays
from __future__ import annotations
import math as _aidefend_math
import os as _aidefend_os
# Admission injects these required values from one signature-verified, versioned runtime profile.
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
def _aidefend_nonnegative_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 < 0:
raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
return value
def _aidefend_fraction(name):
value = _aidefend_positive_float(name)
if value > 1:
raise RuntimeError(f"runtime-profile field must be in (0, 1]: {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")
BASELINE_LIFETIME_SECONDS = _aidefend_positive_int("M003006_BASELINE_LIFETIME_SECONDS")
MIN_LABEL_MATCH_RATE = _aidefend_fraction("M003006_MIN_LABEL_MATCH_RATE")
MAX_MEAN_COSINE_DISTANCE = _aidefend_positive_float("M003006_MAX_MEAN_COSINE_DISTANCE")
# File: modeling/build_gnn_probe_baseline.py
import hashlib
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
import numpy as np
from safetensors.numpy import save_file
BASE = Path("baselines/gnn")
MODEL_PATH = Path("artifacts/gnn-model.safetensors")
GRAPH_PATH = Path("validation/graph-snapshot.safetensors")
NODE_IDS_PATH = Path("validation/probe_node_ids.json")
TENSOR_PATH = BASE / "probe_baseline.safetensors"
MANIFEST_PATH = BASE / "probe_manifest.json"
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 canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
for required in (MODEL_PATH, GRAPH_PATH, NODE_IDS_PATH):
if not required.is_file():
raise SystemExit(f"required validation input missing: {required}")
node_ids = json.loads(NODE_IDS_PATH.read_text(encoding="utf-8"))
if (
not isinstance(node_ids, list)
or not node_ids
or any(not isinstance(node_id, str) or not node_id for node_id in node_ids)
or len(node_ids) != len(set(node_ids))
):
raise SystemExit("probe node IDs must be a non-empty unique string list")
embeddings = np.load("validation/probe_embeddings.npy", allow_pickle=False)
probe_mask = np.load("validation/probe_mask.npy", allow_pickle=False)
probe_expected = np.load("validation/probe_expected.npy", allow_pickle=False)
node_count = len(node_ids)
if embeddings.dtype != np.dtype("float32") or embeddings.ndim != 2 or embeddings.shape[0] != node_count or embeddings.shape[1] == 0:
raise SystemExit("probe_embeddings must be finite float32 with shape [node_count, dimensions]")
if probe_mask.dtype != np.dtype("int8") or probe_mask.shape != (node_count,):
raise SystemExit("probe_mask must be int8 with shape [node_count]")
if probe_expected.dtype != np.dtype("int64") or probe_expected.shape != (node_count,):
raise SystemExit("probe_expected must be int64 with shape [node_count]")
if not np.isfinite(embeddings).all():
raise SystemExit("probe embeddings contain NaN or infinity")
if not np.isin(probe_mask, np.array([0, 1], dtype=np.int8)).all() or int(probe_mask.sum()) == 0:
raise SystemExit("probe_mask must contain only 0/1 and select at least one node")
if np.any(probe_expected < 0):
raise SystemExit("probe_expected contains a negative class ID")
BASE.mkdir(parents=True, exist_ok=True)
save_file(
{
"probe_embeddings": np.ascontiguousarray(embeddings),
"probe_expected": np.ascontiguousarray(probe_expected),
"probe_mask": np.ascontiguousarray(probe_mask),
},
str(TENSOR_PATH),
)
node_order = canonical(node_ids)
now = datetime.now(timezone.utc).replace(microsecond=0)
if MAX_MEAN_COSINE_DISTANCE > 2:
raise RuntimeError("runtime-profile cosine-distance threshold exceeds its mathematical domain")
manifest = {
"schema_version": "aidefend.gnn_probe_baseline.v2",
"model_sha256": sha256_file(MODEL_PATH),
"graph_snapshot_sha256": sha256_file(GRAPH_PATH),
"node_order_digest": hashlib.sha256(node_order).hexdigest(),
"node_ids_sha256": sha256_file(NODE_IDS_PATH),
"tensor_sha256": sha256_file(TENSOR_PATH),
"created_at": now.isoformat().replace("+00:00", "Z"),
"expires_at": (now + timedelta(seconds=BASELINE_LIFETIME_SECONDS)).isoformat().replace("+00:00", "Z"),
"min_label_match_rate": MIN_LABEL_MATCH_RATE,
"max_mean_cosine_distance": MAX_MEAN_COSINE_DISTANCE,
}
MANIFEST_PATH.write_bytes(canonical(manifest))set -euo pipefail
python modeling/build_gnn_probe_baseline.py
cosign sign-blob --yes --key keys/baseline-signing.key --bundle baselines/gnn/probe_manifest.sigstore.json baselines/gnn/probe_manifest.jsonStep 2: Issue a signed receipt for the current inference output
from __future__ import annotations
import math as _aidefend_math
import os as _aidefend_os
# Admission injects these required values from one signature-verified, versioned runtime profile.
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
def _aidefend_nonnegative_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 < 0:
raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
return value
def _aidefend_fraction(name):
value = _aidefend_positive_float(name)
if value > 1:
raise RuntimeError(f"runtime-profile field must be in (0, 1]: {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")
RECEIPT_LIFETIME_SECONDS = _aidefend_positive_int("M003006_RECEIPT_LIFETIME_SECONDS")
# File: validation/create_gnn_output_receipt.py
import hashlib
import json
import os
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path
import numpy as np
from safetensors.numpy import load_file
MODEL_PATH = Path("artifacts/gnn-model.safetensors")
GRAPH_PATH = Path("validation/graph-snapshot.safetensors")
NODE_IDS_PATH = Path("validation/probe_node_ids.json")
OUTPUT_PATH = Path("validation/current_probe_outputs.safetensors")
RECEIPT_PATH = Path("validation/current_probe_outputs.receipt.json")
RUN_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
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 canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
run_id = os.environ.get("INFERENCE_RUN_ID", "")
if not RUN_ID.fullmatch(run_id):
raise SystemExit("INFERENCE_RUN_ID is missing or invalid")
node_ids = json.loads(NODE_IDS_PATH.read_text(encoding="utf-8"))
if not isinstance(node_ids, list) or not node_ids or len(node_ids) != len(set(node_ids)):
raise SystemExit("node-order input is invalid")
current = load_file(str(OUTPUT_PATH))
if set(current) != {"probe_embeddings", "probe_predictions"}:
raise SystemExit("current output SafeTensors keys do not match the exact schema")
embeddings = current["probe_embeddings"]
predictions = current["probe_predictions"]
if embeddings.dtype != np.dtype("float32") or embeddings.ndim != 2 or embeddings.shape[0] != len(node_ids) or embeddings.shape[1] == 0:
raise SystemExit("current probe_embeddings shape or dtype mismatch")
if predictions.dtype != np.dtype("int64") or predictions.shape != (len(node_ids),):
raise SystemExit("current probe_predictions shape or dtype mismatch")
if not np.isfinite(embeddings).all() or np.any(predictions < 0):
raise SystemExit("current probe outputs contain non-finite values or negative labels")
now = datetime.now(timezone.utc).replace(microsecond=0)
receipt = {
"schema_version": "aidefend.gnn_probe_output_receipt.v1",
"run_id": run_id,
"output_sha256": sha256_file(OUTPUT_PATH),
"model_sha256": sha256_file(MODEL_PATH),
"graph_snapshot_sha256": sha256_file(GRAPH_PATH),
"node_order_digest": hashlib.sha256(canonical(node_ids)).hexdigest(),
"created_at": now.isoformat().replace("+00:00", "Z"),
"expires_at": (now + timedelta(seconds=RECEIPT_LIFETIME_SECONDS)).isoformat().replace("+00:00", "Z"),
}
RECEIPT_PATH.write_bytes(canonical(receipt))set -euo pipefail
export INFERENCE_RUN_ID="$CI_PIPELINE_ID-$CI_JOB_ID"
python validation/create_gnn_output_receipt.py
# Run signing in an isolated attestation job; the model-serving process must not access this key.
cosign sign-blob --yes --key keys/inference-output-receipt.key --bundle validation/current_probe_outputs.receipt.sigstore.json validation/current_probe_outputs.receipt.jsonStep 3: Verify both signatures and every binding before comparing probes
from __future__ import annotations
import math as _aidefend_math
import os as _aidefend_os
# Admission injects these required values from one signature-verified, versioned runtime profile.
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
def _aidefend_nonnegative_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 < 0:
raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
return value
def _aidefend_fraction(name):
value = _aidefend_positive_float(name)
if value > 1:
raise RuntimeError(f"runtime-profile field must be in (0, 1]: {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")
MAX_FUTURE_SKEW_SECONDS = _aidefend_nonnegative_int("M003006_MAX_FUTURE_SKEW_SECONDS")
MAX_BASELINE_LIFETIME_SECONDS = _aidefend_positive_int("M003006_MAX_BASELINE_LIFETIME_SECONDS")
MAX_RECEIPT_LIFETIME_SECONDS = _aidefend_positive_int("M003006_MAX_RECEIPT_LIFETIME_SECONDS")
MAX_RECEIPT_PREDATE_SECONDS = _aidefend_nonnegative_int("M003006_MAX_RECEIPT_PREDATE_SECONDS")
PROBE_VERIFY_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("M003006_PROBE_VERIFY_COMMAND_TIMEOUT_SECONDS")
# File: validation/verify_gnn_probe_baseline.py
import hashlib
import json
import math
import subprocess
import os
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
import numpy as np
from safetensors.numpy import load_file
BASE = Path("baselines/gnn")
MODEL_PATH = Path("artifacts/gnn-model.safetensors")
GRAPH_PATH = Path("validation/graph-snapshot.safetensors")
NODE_IDS_PATH = Path("validation/probe_node_ids.json")
CURRENT_OUTPUT = Path("validation/current_probe_outputs.safetensors")
RECEIPT_PATH = Path("validation/current_probe_outputs.receipt.json")
MANIFEST_FIELDS = {
"schema_version", "model_sha256", "graph_snapshot_sha256",
"node_order_digest", "node_ids_sha256", "tensor_sha256",
"created_at", "expires_at", "min_label_match_rate",
"max_mean_cosine_distance",
}
RECEIPT_FIELDS = {
"schema_version", "run_id", "output_sha256", "model_sha256",
"graph_snapshot_sha256", "node_order_digest", "created_at", "expires_at",
}
def fail(message: str) -> None:
raise SystemExit(message)
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 canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def read_canonical(raw: bytes, label: str) -> dict:
value = json.loads(raw)
if not isinstance(value, dict) or canonical(value) != raw:
fail(f"{label} must be a canonical JSON object")
return value
def capture(source: Path, target: Path, label: str) -> bytes:
raw = source.read_bytes()
if not raw:
fail(f"{label} is empty")
target.write_bytes(raw)
os.chmod(target, 0o400)
return raw
def parse_time(value: object, label: str) -> datetime:
if not isinstance(value, str):
fail(f"{label} must be an RFC 3339 timestamp")
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
fail(f"{label} is not a valid RFC 3339 timestamp")
if parsed.utcoffset() is None:
fail(f"{label} must include a timezone")
return parsed.astimezone(timezone.utc)
manifest_path = BASE / "probe_manifest.json"
manifest_bundle_path = BASE / "probe_manifest.sigstore.json"
receipt_bundle_path = Path("validation/current_probe_outputs.receipt.sigstore.json")
baseline_key_path = Path("trust/baseline-signing.pub")
receipt_key_path = Path("trust/inference-output-receipt.pub")
# Capture every externally supplied payload, bundle, and trust key once. Cosign and
# the canonical parser consume only these private read-only snapshots.
with tempfile.TemporaryDirectory(prefix="gnn-probe-verification-") as directory:
snapshot_root = Path(directory)
os.chmod(snapshot_root, 0o700)
manifest_snapshot = snapshot_root / "probe_manifest.json"
manifest_bundle_snapshot = snapshot_root / "probe_manifest.sigstore.json"
receipt_snapshot = snapshot_root / "current_probe_outputs.receipt.json"
receipt_bundle_snapshot = snapshot_root / "current_probe_outputs.receipt.sigstore.json"
baseline_key_snapshot = snapshot_root / "baseline-signing.pub"
receipt_key_snapshot = snapshot_root / "inference-output-receipt.pub"
manifest_raw = capture(manifest_path, manifest_snapshot, "GNN baseline manifest")
manifest_bundle_raw = capture(manifest_bundle_path, manifest_bundle_snapshot, "GNN baseline bundle")
receipt_raw = capture(RECEIPT_PATH, receipt_snapshot, "GNN output receipt")
receipt_bundle_raw = capture(receipt_bundle_path, receipt_bundle_snapshot, "GNN output-receipt bundle")
baseline_key_raw = capture(baseline_key_path, baseline_key_snapshot, "baseline trust key")
receipt_key_raw = capture(receipt_key_path, receipt_key_snapshot, "receipt trust key")
subprocess.run([
"cosign", "verify-blob", "--key", str(baseline_key_snapshot),
"--bundle", str(manifest_bundle_snapshot), str(manifest_snapshot),
], check=True, capture_output=True, text=True, timeout=PROBE_VERIFY_COMMAND_TIMEOUT_SECONDS)
subprocess.run([
"cosign", "verify-blob", "--key", str(receipt_key_snapshot),
"--bundle", str(receipt_bundle_snapshot), str(receipt_snapshot),
], check=True, capture_output=True, text=True, timeout=PROBE_VERIFY_COMMAND_TIMEOUT_SECONDS)
captured = (
(manifest_snapshot, manifest_raw),
(manifest_bundle_snapshot, manifest_bundle_raw),
(receipt_snapshot, receipt_raw),
(receipt_bundle_snapshot, receipt_bundle_raw),
(baseline_key_snapshot, baseline_key_raw),
(receipt_key_snapshot, receipt_key_raw),
)
if any(path.read_bytes() != expected for path, expected in captured):
fail("GNN verification snapshot changed during signature checks")
manifest = read_canonical(manifest_raw, "GNN baseline manifest")
receipt = read_canonical(receipt_raw, "GNN output receipt")
if set(manifest) != MANIFEST_FIELDS or manifest["schema_version"] != "aidefend.gnn_probe_baseline.v2":
fail("GNN manifest fields or schema do not match")
if set(receipt) != RECEIPT_FIELDS or receipt["schema_version"] != "aidefend.gnn_probe_output_receipt.v1":
fail("GNN output receipt fields or schema do not match")
if not isinstance(receipt["run_id"], str) or not receipt["run_id"]:
fail("GNN output receipt run_id is missing")
now = datetime.now(timezone.utc)
manifest_created = parse_time(manifest["created_at"], "manifest created_at")
manifest_expires = parse_time(manifest["expires_at"], "manifest expires_at")
receipt_created = parse_time(receipt["created_at"], "receipt created_at")
receipt_expires = parse_time(receipt["expires_at"], "receipt expires_at")
future_limit = now + timedelta(seconds=MAX_FUTURE_SKEW_SECONDS)
if manifest_created > future_limit or receipt_created > future_limit:
fail("GNN evidence is future-dated")
if manifest_expires <= now or manifest_expires <= manifest_created or manifest_expires - manifest_created > timedelta(seconds=MAX_BASELINE_LIFETIME_SECONDS):
fail("GNN baseline is expired or exceeds its maximum lifetime")
if receipt_expires <= now or receipt_expires <= receipt_created or receipt_expires - receipt_created > timedelta(seconds=MAX_RECEIPT_LIFETIME_SECONDS):
fail("GNN output receipt is expired or exceeds its maximum lifetime")
if receipt_created + timedelta(seconds=MAX_RECEIPT_PREDATE_SECONDS) < manifest_created:
fail("GNN output receipt predates the baseline")
node_ids = json.loads(NODE_IDS_PATH.read_text(encoding="utf-8"))
if (
not isinstance(node_ids, list)
or not node_ids
or any(not isinstance(node_id, str) or not node_id for node_id in node_ids)
or len(node_ids) != len(set(node_ids))
):
fail("probe node IDs must be a non-empty unique string list")
node_order_digest = hashlib.sha256(canonical(node_ids)).hexdigest()
actual_model_sha = sha256_file(MODEL_PATH)
actual_graph_sha = sha256_file(GRAPH_PATH)
if manifest["model_sha256"] != actual_model_sha or receipt["model_sha256"] != actual_model_sha:
fail("model digest is not bound consistently")
if manifest["graph_snapshot_sha256"] != actual_graph_sha or receipt["graph_snapshot_sha256"] != actual_graph_sha:
fail("graph snapshot digest is not bound consistently")
if manifest["node_order_digest"] != node_order_digest or receipt["node_order_digest"] != node_order_digest:
fail("probe node ordering changed")
if manifest["node_ids_sha256"] != sha256_file(NODE_IDS_PATH):
fail("probe node ID file digest mismatch")
tensor_path = BASE / "probe_baseline.safetensors"
if manifest["tensor_sha256"] != sha256_file(tensor_path):
fail("baseline SafeTensors digest mismatch")
if receipt["output_sha256"] != sha256_file(CURRENT_OUTPUT):
fail("current output does not match the signed receipt")
baseline = load_file(str(tensor_path))
current = load_file(str(CURRENT_OUTPUT))
if set(baseline) != {"probe_embeddings", "probe_expected", "probe_mask"}:
fail("baseline SafeTensors keys do not match the exact schema")
if set(current) != {"probe_embeddings", "probe_predictions"}:
fail("current SafeTensors keys do not match the exact schema")
expected_embeddings = baseline["probe_embeddings"]
probe_expected = baseline["probe_expected"]
probe_mask_raw = baseline["probe_mask"]
current_embeddings = current["probe_embeddings"]
current_predictions = current["probe_predictions"]
node_count = len(node_ids)
if expected_embeddings.dtype != np.dtype("float32") or expected_embeddings.ndim != 2 or expected_embeddings.shape[0] != node_count or expected_embeddings.shape[1] == 0:
fail("baseline probe_embeddings shape or dtype mismatch")
if probe_expected.dtype != np.dtype("int64") or probe_expected.shape != (node_count,):
fail("baseline probe_expected shape or dtype mismatch")
if probe_mask_raw.dtype != np.dtype("int8") or probe_mask_raw.shape != (node_count,):
fail("baseline probe_mask shape or dtype mismatch")
if current_embeddings.dtype != np.dtype("float32") or current_embeddings.shape != expected_embeddings.shape:
fail("current probe_embeddings shape or dtype mismatch")
if current_predictions.dtype != np.dtype("int64") or current_predictions.shape != (node_count,):
fail("current probe_predictions shape or dtype mismatch")
if not np.isfinite(expected_embeddings).all() or not np.isfinite(current_embeddings).all():
fail("probe embeddings contain NaN or infinity")
if np.any(probe_expected < 0) or np.any(current_predictions < 0):
fail("probe labels contain a negative class ID")
if not np.isin(probe_mask_raw, np.array([0, 1], dtype=np.int8)).all():
fail("probe_mask contains values other than 0 and 1")
probe_mask = probe_mask_raw == 1
if int(probe_mask.sum()) == 0:
fail("probe_mask selects zero nodes")
min_label_match = manifest["min_label_match_rate"]
max_cosine_distance = manifest["max_mean_cosine_distance"]
if isinstance(min_label_match, bool) or not isinstance(min_label_match, (int, float)) or not math.isfinite(min_label_match) or not 0.0 <= min_label_match <= 1.0:
fail("min_label_match_rate is outside [0, 1]")
if isinstance(max_cosine_distance, bool) or not isinstance(max_cosine_distance, (int, float)) or not math.isfinite(max_cosine_distance) or not 0.0 <= max_cosine_distance <= 2.0:
fail("max_mean_cosine_distance is outside [0, 2]")
left = expected_embeddings[probe_mask].astype(np.float64)
right = current_embeddings[probe_mask].astype(np.float64)
left_norm = np.linalg.norm(left, axis=1)
right_norm = np.linalg.norm(right, axis=1)
if np.any(left_norm <= 1e-12) or np.any(right_norm <= 1e-12):
fail("selected probe contains a zero-norm embedding")
cosine_distances = 1.0 - np.sum(left * right, axis=1) / (left_norm * right_norm)
if not np.isfinite(cosine_distances).all() or np.any(cosine_distances < -1e-6) or np.any(cosine_distances > 2.000001):
fail("cosine distance is non-finite or outside its numerical range")
cosine_distances = np.clip(cosine_distances, 0.0, 2.0)
label_match_rate = float(np.mean(current_predictions[probe_mask] == probe_expected[probe_mask]))
mean_cosine_distance = float(np.mean(cosine_distances))
passed = label_match_rate >= float(min_label_match) and mean_cosine_distance <= float(max_cosine_distance)
evidence = {
"schema_version": "aidefend.gnn_probe_verification.v2",
"verified_at": now.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
"baseline_manifest_sha256": hashlib.sha256(manifest_raw).hexdigest(),
"output_receipt_sha256": hashlib.sha256(receipt_raw).hexdigest(),
"run_id": receipt["run_id"],
"model_sha256": actual_model_sha,
"graph_snapshot_sha256": actual_graph_sha,
"node_order_digest": node_order_digest,
"selected_probe_count": int(probe_mask.sum()),
"label_match_rate": label_match_rate,
"mean_cosine_distance": mean_cosine_distance,
"passed": passed,
}
result_path = Path("artifacts/gnn-probe-baseline-verification.json")
result_path.write_bytes(canonical(evidence))
if not passed:
fail("GNN probe baseline comparison failed")set -euo pipefail
python validation/verify_gnn_probe_baseline.py
cosign sign-blob --yes --key keys/baseline-review.key --bundle artifacts/gnn-probe-baseline-verification.sigstore.json artifacts/gnn-probe-baseline-verification.json
cosign verify-blob --key trust/baseline-review.pub --bundle artifacts/gnn-probe-baseline-verification.sigstore.json artifacts/gnn-probe-baseline-verification.jsonAction: The baseline signer, output-receipt signer, and review signer should be separate protected identities. Promotion may consume the signed comparison result; production detection and alert tuning remain with Detect-side controls.
AID-M-003.006-G002
Create signed graph-structure metric baselines that bind graph snapshots, node order, schema, and selected structural fingerprints.
Concept:
A graph baseline must describe the actual graph artifact the model uses, not a hand-written metric note. Bind each metric profile to the graph file digest, node-order digest, graph schema, model version, and measurement code version. Treat spectral and energy metrics as optional evidence: compute them only when the graph size makes exact eigendecomposition feasible, and record INSUFFICIENT_DATA instead of pretending every graph can be handled the same way.
Step 1: Compute a canonical graph-structure profile
from __future__ import annotations
# File: modeling/build_graph_structure_baseline.py
import hashlib
import json
import math
import os
from datetime import datetime, timezone
from pathlib import Path
from statistics import mean, pstdev
import networkx as nx
import numpy as np
from numpy.linalg import eigvalsh
GRAPH_PATH = Path(os.environ.get("GRAPH_PATH", "validation/graph.graphml"))
MODEL_ID = os.environ["MODEL_ID"]
OUTPUT_PATH = Path("baselines/graph/structure-baseline.json")
# Evaluator-owned implementation ceiling; callers cannot raise it through env.
MAX_EXACT_SPECTRAL_NODES = 1_000
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 canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def percentile(values: list[float], pct: float) -> float:
if not values:
return 0.0
return float(np.percentile(np.asarray(values, dtype=float), pct))
if not GRAPH_PATH.is_file():
raise SystemExit(f"graph snapshot missing: {GRAPH_PATH}")
if not MODEL_ID.strip():
raise SystemExit("MODEL_ID must identify the graph model version")
G = nx.read_graphml(GRAPH_PATH)
if G.number_of_nodes() == 0:
raise SystemExit("graph snapshot contains zero nodes")
node_order = [str(node) for node in sorted(G.nodes(), key=str)]
edge_type_values = sorted({str(data.get("type", "default")) for _, _, data in G.edges(data=True)})
node_attribute_schema = sorted({
(str(key), type(value).__name__)
for _, attributes in G.nodes(data=True)
for key, value in attributes.items()
})
edge_attribute_schema = sorted({
(str(key), type(value).__name__)
for _, _, attributes in G.edges(data=True)
for key, value in attributes.items()
})
degrees = [float(deg) for _, deg in G.degree()]
betweenness = list(nx.betweenness_centrality(G, k=min(500, G.number_of_nodes()), seed=7).values())
undirected = G.to_undirected()
components = list(nx.connected_components(undirected))
metrics = {
"node_count": int(G.number_of_nodes()),
"edge_count": int(G.number_of_edges()),
"is_directed": bool(G.is_directed()),
"density": float(nx.density(G)),
"degree_mean": float(mean(degrees)),
"degree_stddev": float(pstdev(degrees)) if len(degrees) > 1 else 0.0,
"degree_max": float(max(degrees)),
"betweenness_p95": percentile([float(v) for v in betweenness], 95),
"clustering_mean": float(mean(nx.clustering(undirected).values())),
"connected_component_count": int(len(components)),
"largest_component_ratio": float(max(len(c) for c in components) / G.number_of_nodes()),
}
spectral = {"status": "INSUFFICIENT_DATA", "reason": "graph exceeds exact spectral baseline limit"}
if G.number_of_nodes() <= MAX_EXACT_SPECTRAL_NODES:
adjacency = nx.to_numpy_array(G, nodelist=node_order, dtype=float)
vals = (
np.linalg.eigvals(adjacency) if G.is_directed()
else eigvalsh(adjacency)
)
energy = float(np.sum(np.abs(vals)))
if not math.isfinite(energy):
raise SystemExit("spectral energy is not finite")
spectral = {"status": "PASS", "graph_energy": energy, "spectral_radius": float(np.max(np.abs(vals)))}
artifact = {
"schema_version": "aidefend.graph_structure_baseline.v1",
"created_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
"model_id": MODEL_ID,
"graph_path": str(GRAPH_PATH),
"graph_sha256": sha256_file(GRAPH_PATH),
"node_order_digest": hashlib.sha256(canonical(node_order)).hexdigest(),
"node_count": G.number_of_nodes(),
"graph_schema": {
"directed": G.is_directed(),
"multigraph": G.is_multigraph(),
"node_attributes": node_attribute_schema,
"edge_attributes": edge_attribute_schema,
"edge_type_values": edge_type_values,
},
"measurement_code_sha256": sha256_file(Path(__file__)),
"evaluator_runtime": {
"networkx": nx.__version__,
"numpy": np.__version__,
"exact_spectral_node_ceiling": MAX_EXACT_SPECTRAL_NODES,
},
"metrics": metrics,
"spectral_metrics": spectral,
}
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_PATH.write_bytes(canonical(artifact))
print(json.dumps(artifact, indent=2))Step 2: Publish through the governed baseline registry
set -euo pipefail
export MODEL_ID="fraud-gnn:v17"
export GRAPH_PATH="validation/graph.graphml"
python modeling/build_graph_structure_baseline.py
sha256sum baselines/graph/structure-baseline.jsonAction: The generated JSON and digest are candidate inputs only; they cannot produce control PASS. Submit the exact bytes through the organization's authorized baseline-publisher path. The registry must authenticate the publisher, bind graph/model/schema/node-order/parser/runtime/measurement-code generations, enforce policy currentness, publish one immutable version, and return an independently readable reference whose digest matches the submitted bytes. Missing, stale, revoked, release-mismatched, unpublished, or unreadback evidence remains INSUFFICIENT_DATA or ERROR. Detect-side controls may consume the accepted reference later, but this technique only owns its creation and validation.
Verify safely
A read-only identity that cannot alter the controlled asset must obtain immutable graph and model manifests, safe tensor artifacts, raw probe outputs, structural metrics, and perturbation receipts; validate the effective asset, policy, input, and scope versions or digests; then reconstruct canonical node and edge order, recompute structure metrics and probe outputs, rerun all perturbations, and compare complete output and bundle digests.
AID-M-003.006-G003
Validate graph structural metrics against paired, bounded perturbation experiments with actual frozen-model inference before using those metrics in Detect-side monitoring; probe-output baselines are defined separately.
Concept and evaluation contract
A correlation across perturbation sizes is not evidence that a metric separates harmful from benign changes of the same size. Use a frozen release-candidate node classifier exported to ONNX, the exact reviewed validation graph, and paired challenge and benign graph exports for every policy-required kind, bounded level and seed. Use several distinct seeds at each level. The signed v2 profile binds every graph export, the unchanged test labels and population, selection limits, the evaluator code digest and the pinned runner. Never retrain the model or change the policy to obtain a result.
This example validates structural metrics on simple undirected graphs. It accepts node additions/removals, edge changes, numeric feature changes and degree-preserving topology rewiring; required kinds and levels come from the approved threat and validation policy. Other graph semantics or probe-output metrics need their own reviewed adapter and paired evaluation; this example does not validate them. A structural metric cannot detect a feature-only change that leaves the structure unchanged and must remain unselected when that required experiment fails.
Step 1: Freeze paired experiments and sign a v2 profile
Export each case as an NPZ containing node_ids, float32 features, integer labels, boolean test_mask and a two-row integer edge_index. Preserve original node identities, test labels, test population and relative node order. Node cases may add or remove non-test nodes; edge and topology cases do not change features or nodes; feature cases do not change topology. Topology cases in this adapter preserve each node's degree. Generate paired exports with the declared seed, retain the producer's generation record, and independently review their intended meaning before signing. The evaluator verifies actual change budgets and rejects unequal pairs or identical inputs relabeled as new trials.
The reviewed policy supplies policy_version, class_count, max_node_count, onnx_inputs (features and adjacency), onnx_output, runner_image_digest, perturbation_plan (required kind to positive bounded levels), perturbation_seeds (at least two distinct seeds), cases (one kind/level/seed pair with challenge_file and benign_file basenames for every combination), min_pairs_per_group, min_abs_pearson_correlation, min_pair_separation_rate, min_challenge_accuracy_drop, max_benign_accuracy_drop and positive metric_effect_floors for graph_energy, degree_mean and clustering_mean. Levels count changed nodes, feature cells or the symmetric difference of edges; both members must have equal full node/edge/feature budgets. Case digests also bind the exact numeric values. Set the limits before measurement. Rebuild and re-sign v1 profiles; extra or missing fields are rejected.
# File: modeling/build_graph_experiment_profile.py
import hashlib
import json
import os
from pathlib import Path
import numpy as np
from validate_graph_baseline_metrics import canonical, validate_profile
def digest(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
# Approve the policy before measurement; do not tune it after seeing the results.
profile = json.loads(Path("evals/reviewed-experiment-policy.json").read_text(encoding="utf-8"))
model, graph = Path("release/model.onnx"), Path("release/validation-graph.npz")
with np.load(graph, allow_pickle=False) as data:
profile["node_count"] = len(data["node_ids"])
profile["feature_dimension"] = data["features"].shape[1]
profile["test_node_ids_sha256"] = hashlib.sha256(
canonical(sorted(map(int, data["node_ids"][data["test_mask"]])))).hexdigest()
profile.update(schema_version="aidefend.graph-perturbation-profile.v2",
model_sha256=digest(model), graph_npz_sha256=digest(graph),
evaluator_sha256=digest(Path("modeling/validate_graph_baseline_metrics.py")),
runtime_profile_version=os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"],
runtime_profile_sha256=os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"])
for case in profile["cases"]:
for role in ("challenge", "benign"):
name = case[role + "_file"]
if Path(name).name != name or not name.endswith(".npz"):
raise ValueError("paired filename must be a basename")
path = Path("evals/paired-graphs") / name
if path.is_symlink() or not path.is_file():
raise ValueError("paired graph missing or unsafe")
case[role + "_sha256"] = digest(path)
validate_profile(profile)
Path("evals/graph-experiment-profile.json").write_bytes(canonical(profile))
python modeling/build_graph_experiment_profile.py
cosign sign-blob --yes --key env://GRAPH_EXPERIMENT_PROFILE_SIGNING_KEY --bundle evals/graph-experiment-profile.sig evals/graph-experiment-profile.jsonStep 2: Verify all inputs, measure both members and gate every group
from __future__ import annotations
# File: modeling/validate_graph_baseline_metrics.py
import argparse
import hashlib
import io
import json
import math
import os
import re
import subprocess
from pathlib import Path
import networkx as nx
import numpy as np
import onnxruntime as ort
SHA256 = re.compile(r"^[a-f0-9]{64}$")
IMAGE = re.compile(r"^[a-z0-9][a-z0-9._/:~-]*@sha256:[a-f0-9]{64}$")
FILE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*\.npz$")
KINDS = {"node", "edge", "feature", "topology"}
METRICS = ("graph_energy", "degree_mean", "clustering_mean")
PROFILE_FIELDS = {
"schema_version", "policy_version", "model_sha256", "graph_npz_sha256", "evaluator_sha256",
"node_count", "feature_dimension", "class_count", "test_node_ids_sha256",
"perturbation_plan", "perturbation_seeds", "cases", "max_node_count",
"min_abs_pearson_correlation", "min_pair_separation_rate", "min_pairs_per_group",
"min_challenge_accuracy_drop", "max_benign_accuracy_drop", "metric_effect_floors",
"onnx_inputs", "onnx_output", "runner_image_digest", "runtime_profile_version",
"runtime_profile_sha256",
}
CASE_FIELDS = {"kind", "level", "seed", "challenge_file", "challenge_sha256",
"benign_file", "benign_sha256"}
def positive_int(value):
return type(value) is int and value > 0
def finite_number(value):
return type(value) in (int, float) and math.isfinite(value)
def canonical(value):
return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
def sha256_file(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def reject_duplicates(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("duplicate profile field")
result[key] = value
return result
def validate_profile(profile):
if type(profile) is not dict or set(profile) != PROFILE_FIELDS:
raise ValueError("graph experiment profile schema differs; v1 is not accepted")
if profile["schema_version"] != "aidefend.graph-perturbation-profile.v2":
raise ValueError("graph experiment profile version differs")
for field in ("node_count", "feature_dimension", "class_count", "max_node_count", "min_pairs_per_group"):
if not positive_int(profile[field]):
raise ValueError("experiment dimensions must be positive integers")
if (profile["node_count"] > profile["max_node_count"] or profile["min_pairs_per_group"] < 2
or type(profile["policy_version"]) is not str or not profile["policy_version"].strip()):
raise ValueError("experiment coverage or policy is invalid")
for field in ("model_sha256", "graph_npz_sha256", "evaluator_sha256", "test_node_ids_sha256", "runtime_profile_sha256"):
if type(profile[field]) is not str or not SHA256.fullmatch(profile[field]):
raise ValueError("experiment input digest is invalid")
if (type(profile["runner_image_digest"]) is not str or not IMAGE.fullmatch(profile["runner_image_digest"])
or profile["runtime_profile_version"] != os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"]
or profile["runtime_profile_sha256"] != os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"]):
raise ValueError("runner or runtime profile differs")
for field in ("min_abs_pearson_correlation", "min_pair_separation_rate", "min_challenge_accuracy_drop"):
if not finite_number(profile[field]) or not 0 < profile[field] <= 1:
raise ValueError("selection threshold must be in (0, 1]")
if (not finite_number(profile["max_benign_accuracy_drop"])
or not 0 <= profile["max_benign_accuracy_drop"] < profile["min_challenge_accuracy_drop"]):
raise ValueError("benign and challenge quality limits overlap")
floors = profile["metric_effect_floors"]
if (type(floors) is not dict or set(floors) != set(METRICS)
or any(not finite_number(value) or value <= 0 for value in floors.values())):
raise ValueError("positive per-metric separation floors are required")
inputs = profile["onnx_inputs"]
if (type(inputs) is not dict or set(inputs) != {"features", "adjacency"}
or any(type(value) is not str or not value for value in inputs.values())
or len(set(inputs.values())) != 2 or type(profile["onnx_output"]) is not str
or not profile["onnx_output"]):
raise ValueError("ONNX tensor names are invalid")
seeds = profile["perturbation_seeds"]
if (type(seeds) is not list or len(seeds) < profile["min_pairs_per_group"]
or any(type(seed) is not int or seed < 0 for seed in seeds)
or seeds != sorted(set(seeds))):
raise ValueError("several distinct seeds are required for each group")
plan = profile["perturbation_plan"]
if type(plan) is not dict or not plan or set(plan) - KINDS:
raise ValueError("policy must name supported perturbation kinds")
for levels in plan.values():
if (type(levels) is not list or not levels or any(not positive_int(level) for level in levels)
or levels != sorted(set(levels))):
raise ValueError("policy must name positive bounded levels")
expected = {(kind, level, seed) for kind, levels in plan.items() for level in levels for seed in seeds}
cases = profile["cases"]
if type(cases) is not list or len(cases) != len(expected):
raise ValueError("a challenge and benign pair is required for every kind, level and seed")
seen, unique_trials = set(), {}
for case in cases:
if type(case) is not dict or set(case) != CASE_FIELDS:
raise ValueError("paired case schema differs")
if type(case["kind"]) is not str or type(case["level"]) is not int or type(case["seed"]) is not int:
raise ValueError("paired case identity is invalid")
key = (case["kind"], case["level"], case["seed"])
if key not in expected or key in seen:
raise ValueError("unexpected or duplicate paired case")
seen.add(key)
for role in ("challenge", "benign"):
if (type(case[role + "_file"]) is not str or not FILE.fullmatch(case[role + "_file"])
or type(case[role + "_sha256"]) is not str or not SHA256.fullmatch(case[role + "_sha256"])):
raise ValueError("case filename or digest is invalid")
digests = (case["challenge_sha256"], case["benign_sha256"])
trials = unique_trials.setdefault(key[:2], set())
if digests[0] == digests[1] or digests in trials:
raise ValueError("different seeds cannot relabel the same paired inputs")
trials.add(digests)
return profile
def load_verified_profile(path, signature):
timeout = float(os.environ["M003006_COMMAND_TIMEOUT_SECONDS"])
if not math.isfinite(timeout) or timeout <= 0:
raise ValueError("signed timeout is invalid")
subprocess.run(["cosign", "verify-blob", "--key", os.environ["GRAPH_EXPERIMENT_PROFILE_VERIFY_KEY"],
"--bundle", str(signature), str(path)], check=True, capture_output=True,
text=True, timeout=timeout)
return validate_profile(json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicates))
def load_graph(path, profile, *, baseline=False, expected_sha256=None):
graph_bytes = path.read_bytes()
if expected_sha256 is not None and hashlib.sha256(graph_bytes).hexdigest() != expected_sha256:
raise ValueError("graph bytes changed or differ from signed profile")
with np.load(io.BytesIO(graph_bytes), allow_pickle=False) as archive:
if set(archive.files) != {"node_ids", "features", "labels", "test_mask", "edge_index"}:
raise ValueError("graph NPZ schema differs")
ids, features, labels, mask, edges = (np.asarray(archive[name]) for name in
("node_ids", "features", "labels", "test_mask", "edge_index"))
count = len(ids)
if (ids.shape != (count,) or not 0 < count <= profile["max_node_count"]
or (baseline and count != profile["node_count"])
or not np.issubdtype(ids.dtype, np.integer) or len(set(map(int, ids))) != count
or features.shape != (count, profile["feature_dimension"]) or features.dtype != np.float32
or not np.isfinite(features).all() or labels.shape != (count,)
or not np.issubdtype(labels.dtype, np.integer) or np.any(labels < 0)
or np.any(labels >= profile["class_count"]) or mask.shape != (count,)
or mask.dtype != np.bool_ or not bool(mask.any()) or edges.ndim != 2 or edges.shape[0] != 2
or not np.issubdtype(edges.dtype, np.integer)):
raise ValueError("graph arrays, dimensions, labels or test population are invalid")
if hashlib.sha256(canonical(sorted(map(int, ids[mask])))).hexdigest() != profile["test_node_ids_sha256"]:
raise ValueError("test-node population differs from signed profile")
graph = nx.Graph()
graph.add_nodes_from(range(count))
normalized = []
for source, target in edges.T:
source, target = int(source), int(target)
if source == target or not 0 <= source < count or not 0 <= target < count:
raise ValueError("invalid edge endpoint or self-loop")
normalized.append(tuple(sorted((source, target))))
if len(set(normalized)) != len(normalized):
raise ValueError("duplicate edge")
graph.add_edges_from(normalized)
return graph, features, labels.astype(np.int64), mask, ids
def change_budget(original, changed, kind, level):
graph, features, labels, mask, ids = original
other, other_features, other_labels, other_mask, other_ids = changed
old = {int(node): i for i, node in enumerate(ids)}
new = {int(node): i for i, node in enumerate(other_ids)}
# Labels, the test population and existing node order cannot be changed to manufacture accuracy.
common = set(old) & set(new)
if ([int(node) for node in ids if int(node) in common] != [int(node) for node in other_ids if int(node) in common]
or any(labels[old[node]] != other_labels[new[node]] or mask[old[node]] != other_mask[new[node]] for node in common)):
raise ValueError("labels, test mask or existing node order changed")
edge_set = lambda g, names: {tuple(sorted((int(names[a]), int(names[b])))) for a, b in g.edges()}
budget = {
"nodes": len(set(old) ^ set(new)),
"edges": len(edge_set(graph, ids) ^ edge_set(other, other_ids)),
"features": sum(int(np.count_nonzero(features[old[node]] != other_features[new[node]])) for node in common),
}
primary = "edges" if kind == "topology" else kind + "s" if kind != "feature" else "features"
if budget[primary] != level:
raise ValueError("actual perturbation exceeds or differs from the signed level")
if ((kind in {"edge", "topology"} and (budget["nodes"] or budget["features"]))
or (kind == "feature" and (budget["nodes"] or budget["edges"]))
or (kind == "node" and budget["features"])):
raise ValueError("perturbation changes undeclared surfaces")
if kind == "topology" and any(graph.degree(old[node]) != other.degree(new[node]) for node in common):
raise ValueError("this topology experiment requires degree-preserving rewiring")
return budget
def normalized_adjacency(graph):
adjacency = nx.to_numpy_array(graph, nodelist=range(len(graph)), dtype=np.float32, weight=None)
adjacency += np.eye(len(graph), dtype=np.float32)
inverse = np.diag(np.power(adjacency.sum(axis=1), np.float32(-0.5)))
return np.asarray(inverse @ adjacency @ inverse, dtype=np.float32)
def structural_metrics(graph):
adjacency = nx.to_numpy_array(graph, nodelist=range(len(graph)), dtype=np.float64, weight=None)
return {"graph_energy": float(np.abs(np.linalg.eigvalsh(adjacency)).sum()),
"degree_mean": float(np.mean([degree for _, degree in graph.degree()])),
"clustering_mean": float(np.mean(list(nx.clustering(graph).values())))}
def evaluate(session, data, profile):
graph, features, labels, mask, _ = data
names = profile["onnx_inputs"]
logits = np.asarray(session.run([profile["onnx_output"]],
{names["features"]: features, names["adjacency"]: normalized_adjacency(graph)})[0])
if logits.shape != (len(graph), profile["class_count"]) or not np.isfinite(logits).all():
raise ValueError("GNN output shape or values are invalid")
return {"accuracy": float(np.mean(logits.argmax(axis=1)[mask] == labels[mask])),
"logits_sha256": hashlib.sha256(np.ascontiguousarray(logits).tobytes()).hexdigest(),
"metrics": structural_metrics(graph)}
def pearson(values, target):
left, right = np.asarray(values, dtype=np.float64), np.asarray(target, dtype=np.float64)
if len(left) < 3 or np.isclose(left.std(), 0.0) or np.isclose(right.std(), 0.0):
return None
value = float(np.corrcoef(left, right)[0, 1])
return value if math.isfinite(value) else None
def select_metrics(baseline, pairs, profile):
decisions = {}
for metric in METRICS:
groups = []
for kind, levels in sorted(profile["perturbation_plan"].items()):
for level in levels:
group = [pair for pair in pairs if pair["kind"] == kind and pair["level"] == level]
valid = [pair for pair in group if
baseline["accuracy"] - pair["challenge"]["accuracy"] >= profile["min_challenge_accuracy_drop"]
and baseline["accuracy"] - pair["benign"]["accuracy"] <= profile["max_benign_accuracy_drop"]]
separated = sum(abs(pair["challenge"]["metrics"][metric] - baseline["metrics"][metric])
- abs(pair["benign"]["metrics"][metric] - baseline["metrics"][metric])
>= profile["metric_effect_floors"][metric] for pair in valid)
rate = separated / len(valid) if valid else None
groups.append({"kind": kind, "level": level, "valid_pairs": len(valid),
"separation_rate": rate,
"sufficient": len(valid) >= profile["min_pairs_per_group"]
and len(valid) == len(group) == len(profile["perturbation_seeds"]),
"separates": rate is not None and rate >= profile["min_pair_separation_rate"]})
correlation = pearson([pair["challenge"]["metrics"][metric] for pair in pairs],
[pair["challenge"]["accuracy"] for pair in pairs])
selected = (bool(groups) and all(group["sufficient"] and group["separates"] for group in groups)
and correlation is not None and abs(correlation) >= profile["min_abs_pearson_correlation"])
decisions[metric] = {"groups": groups, "correlation": correlation,
"status": "VALIDATED_CANDIDATE" if selected else "UNSELECTED",
"recommended_for_detect_monitoring": selected}
return decisions
def run_experiment(profile, model_path, graph_path, cases_dir, profile_sha256):
validate_profile(profile)
model_bytes = model_path.read_bytes()
if hashlib.sha256(model_bytes).hexdigest() != profile["model_sha256"]:
raise ValueError("model bytes differ from signed profile")
if sha256_file(Path(__file__)) != profile["evaluator_sha256"]:
raise ValueError("evaluation code differs from signed profile")
original = load_graph(graph_path, profile, baseline=True, expected_sha256=profile["graph_npz_sha256"])
# Verify and stage every paired input before evaluating any model or emitting a recommendation.
staged, distinct_inputs = [], {}
for case in sorted(profile["cases"], key=lambda case: (case["kind"], case["level"], case["seed"])):
inputs, budgets = {}, {}
for role in ("challenge", "benign"):
candidate = cases_dir / case[role + "_file"]
if candidate.is_symlink() or not candidate.is_file():
raise ValueError("paired graph missing, unsafe or changed")
inputs[role] = load_graph(candidate, profile, expected_sha256=case[role + "_sha256"])
budgets[role] = change_budget(original, inputs[role], case["kind"], case["level"])
if budgets["challenge"] != budgets["benign"]:
raise ValueError("challenge and benign changes have different actual budgets")
semantic_pair = []
for role in ("challenge", "benign"):
graph, features, labels, mask, ids = inputs[role]
semantic_pair.append(hashlib.sha256(canonical({"ids": ids.tolist(), "features": features.tolist(),
"labels": labels.tolist(), "mask": mask.tolist(), "edges": sorted(map(list, graph.edges()))})).hexdigest())
seen = distinct_inputs.setdefault((case["kind"], case["level"]), set())
if tuple(semantic_pair) in seen:
raise ValueError("different seeds relabel the same graph content")
seen.add(tuple(semantic_pair))
staged.append((case, inputs, budgets["challenge"]))
options = ort.SessionOptions()
options.intra_op_num_threads = 1
options.inter_op_num_threads = 1
session = ort.InferenceSession(model_bytes, sess_options=options, providers=["CPUExecutionProvider"])
if (set(profile["onnx_inputs"].values()) - {item.name for item in session.get_inputs()}
or profile["onnx_output"] not in {item.name for item in session.get_outputs()}):
raise ValueError("signed ONNX tensor names are unavailable")
baseline = evaluate(session, original, profile)
pairs = [{**case, "actual_budget": budget,
"challenge": evaluate(session, inputs["challenge"], profile),
"benign": evaluate(session, inputs["benign"], profile)} for case, inputs, budget in staged]
return {"schema_version": "aidefend.graph-metric-validation.v2", "policy_version": profile["policy_version"],
"profile_sha256": profile_sha256, "model_sha256": profile["model_sha256"],
"evaluator_sha256": profile["evaluator_sha256"],
"graph_npz_sha256": profile["graph_npz_sha256"], "test_node_ids_sha256": profile["test_node_ids_sha256"],
"runner_image_digest": profile["runner_image_digest"],
"runtime_profile_version": profile["runtime_profile_version"],
"runtime_profile_sha256": profile["runtime_profile_sha256"],
"baseline": baseline, "pairs": pairs, "metric_decisions": select_metrics(baseline, pairs, profile),
"claim_boundary": "Monitoring candidates for the signed experiment scope only; no finding, proof of a clean model or guarantee of detecting attacks."}
def main():
parser = argparse.ArgumentParser()
for name in ("profile", "profile-signature", "model", "graph", "cases", "output"):
parser.add_argument("--" + name, type=Path, required=True)
args = parser.parse_args()
profile = load_verified_profile(args.profile, args.profile_signature)
report = run_experiment(profile, args.model, args.graph, args.cases, sha256_file(args.profile))
args.output.parent.mkdir(parents=True, exist_ok=True)
temporary = args.output.with_suffix(args.output.suffix + ".tmp")
temporary.write_bytes(canonical(report) + b"\n")
temporary.replace(args.output)
if __name__ == "__main__":
main()
Step 3: Independently replay and sign the complete report
The digest-pinned runner contains locked Python, NumPy, NetworkX, ONNX Runtime and Cosign versions. Its inference contract takes node features and a dense normalized adjacency and returns class logits per node; other contracts need a separately reviewed adapter. Run the exact verified inputs twice with read-only inputs and no network, compare canonical report bytes, then sign and independently read back under the expected reviewer identity. The report preserves each pair's digests, actual budget, output digest, measured accuracy and metrics, and each group's coverage and separation.
set -euo pipefail
umask 077
: "${M003006_COMMAND_TIMEOUT_SECONDS:?required from signed runtime profile}"
: "${AIDEFEND_RUNTIME_PROFILE_VERSION:?required from signed runtime profile}"
: "${AIDEFEND_RUNTIME_PROFILE_SHA256:?required from signed runtime profile}"
INPUT_STAGE="$(mktemp -d)"
cleanup() { if [[ -n "${INPUT_STAGE:-}" && -d "$INPUT_STAGE" ]]; then rm -rf -- "$INPUT_STAGE"; fi; }
trap cleanup EXIT HUP INT TERM
chmod 0700 "$INPUT_STAGE"
mkdir -m 0700 "$INPUT_STAGE/cases"
for INPUT in evals/graph-experiment-profile.json evals/graph-experiment-profile.sig release/model.onnx release/validation-graph.npz keys/graph-experiment-profile.pub modeling/validate_graph_baseline_metrics.py; do
[[ -f "$INPUT" && ! -L "$INPUT" ]] || { echo "missing or unsafe input: $INPUT" >&2; exit 1; }
done
install -m 0400 -- evals/graph-experiment-profile.json "$INPUT_STAGE/profile.json"
install -m 0400 -- evals/graph-experiment-profile.sig "$INPUT_STAGE/profile.sig"
install -m 0400 -- release/model.onnx "$INPUT_STAGE/model.onnx"
install -m 0400 -- release/validation-graph.npz "$INPUT_STAGE/validation-graph.npz"
install -m 0400 -- keys/graph-experiment-profile.pub "$INPUT_STAGE/profile.pub"
install -m 0400 -- modeling/validate_graph_baseline_metrics.py "$INPUT_STAGE/validate.py"
# The public key is pinned by deployment policy, never supplied by the profile.
cosign verify-blob --key "$INPUT_STAGE/profile.pub" --bundle "$INPUT_STAGE/profile.sig" "$INPUT_STAGE/profile.json"
jq -er '[.cases[] | .challenge_file, .benign_file] | unique | .[]' "$INPUT_STAGE/profile.json" > "$INPUT_STAGE/case-files"
while IFS= read -r CASE_FILE; do
[[ "$CASE_FILE" =~ ^[A-Za-z0-9][A-Za-z0-9_.-]*\.npz$ ]] || exit 1
[[ -f "evals/paired-graphs/$CASE_FILE" && ! -L "evals/paired-graphs/$CASE_FILE" ]] || exit 1
install -m 0400 -- "evals/paired-graphs/$CASE_FILE" "$INPUT_STAGE/cases/$CASE_FILE"
done < "$INPUT_STAGE/case-files"
RUNNER_IMAGE="$(jq -er '.runner_image_digest' "$INPUT_STAGE/profile.json")"
[[ "$RUNNER_IMAGE" =~ ^[a-z0-9][a-z0-9._/:~-]*@sha256:[a-f0-9]{64}$ ]] || exit 1
mkdir -p evidence/graph-replay-a evidence/graph-replay-b
for OUTPUT in graph-replay-a graph-replay-b; do
[[ -d "evidence/$OUTPUT" && ! -L "evidence/$OUTPUT" ]] || exit 1
docker run --rm --network=none --read-only --cap-drop=ALL --security-opt=no-new-privileges --tmpfs /tmp:rw,noexec,nosuid,size=64m \
-e M003006_COMMAND_TIMEOUT_SECONDS="$M003006_COMMAND_TIMEOUT_SECONDS" \
-e AIDEFEND_RUNTIME_PROFILE_VERSION="$AIDEFEND_RUNTIME_PROFILE_VERSION" \
-e AIDEFEND_RUNTIME_PROFILE_SHA256="$AIDEFEND_RUNTIME_PROFILE_SHA256" \
-e GRAPH_EXPERIMENT_PROFILE_VERIFY_KEY=/inputs/profile.pub \
-v "$INPUT_STAGE:/inputs:ro" -v "$PWD/evidence/$OUTPUT:/out" "$RUNNER_IMAGE" \
python /inputs/validate.py --profile /inputs/profile.json --profile-signature /inputs/profile.sig --model /inputs/model.onnx --graph /inputs/validation-graph.npz --cases /inputs/cases --output /out/metric-validation.json
done
cmp evidence/graph-replay-a/metric-validation.json evidence/graph-replay-b/metric-validation.json
cosign sign-blob --yes --key env://GRAPH_METRIC_REVIEW_KEY --bundle evidence/graph-metric-validation.sig evidence/graph-replay-a/metric-validation.json
cosign verify-blob --key keys/graph-metric-review.pub --bundle evidence/graph-metric-validation.sig evidence/graph-replay-a/metric-validation.json
Step 4: Verify positive and negative fixtures
CI must run a compatible frozen ONNX/NPZ fixture and reject a wrong signature, model or graph digest, changed labels or test population, duplicate or missing case, a single seed, identical trials relabeled with new seeds, wrong-kind changes, unequal or excessive budgets, unavailable tensor names and invalid output shape before any recommendation. Valid replay reports must be byte-identical. Include mean degree correlating with accuracy across edge counts but equaling the benign control at each count: it must remain unselected.
Action: A metric is a monitoring candidate only when every required kind/level group has enough measured independent pairs, challenges actually exceed the signed quality-loss limit, benign controls stay within their limit, matched-pair separation meets the signed threshold and the correlation check passes. Missing coverage, constant observations, weak separation or an uninformative challenge leaves it unselected; preserve the gap, never invent scores or relax the policy. A candidate applies only to this exact model, graph population, adapter and measured perturbation scope. It is not a finding, proof of a clean graph/model or guarantee of detecting attacks. Detect-side adoption and operational calibration remain separate.
Open-source tools
- PyTorch Geometric
- Deep Graph Library (DGL)
- NetworkX
- NumPy
- SciPy
- safetensors
- Cosign
- MLflow
- DVC
Commercial / hosted tools
- Neo4j Enterprise Edition
- TigerGraph Savanna
- Amazon SageMaker AI
- Google Vertex AI
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
- N/A
Google Secure AI Framework 2.0 - Risks
- N/A
Databricks AI Security Framework 3.0
- Evaluation 6.2: Insufficient evaluation data
- Algorithms 5.1: Lack of tracking and reproducibility of experiments