AID-D-012.003: Topology-Only Degree & Centrality Anomaly Detection
Load the graph under review explicitly and identify topology-only node outliers using calibrated degree and betweenness-centrality scores. Preserve the graph identity, policy, population, sampling parameters, raw metrics, and coverage.
Technical applicability: an AI asset whose signed technical inventory has no supported graph topology is outside this detector's scope. Once supported topology is declared, missing graph bytes, an incompatible or unreadable calibration reference, and inadequate node or sampling coverage must remain explicit rather than being treated as a clean topology.
Technical applicability: an AI asset whose signed technical inventory has no supported graph topology is outside this detector's scope. Once supported topology is declared, missing graph bytes, an incompatible or unreadable calibration reference, and inadequate node or sampling coverage must remain explicit rather than being treated as a clean topology.
Pillars: data
Lifecycle: validation, operation
Scope boundary
Owns topology-only degree and betweenness-centrality outlier findings from an explicitly loaded graph, signed calibration reference, deterministic sampling, raw metrics, and complete node-coverage accounting. It does not inspect node features, infer a backdoor, claim clique or graph-energy analysis, or mutate the graph.
AID-D-012.001: AID-D-012.003 detects node outliers from topology-only degree and centrality; AID-D-012.001 tests an exact candidate GNN under preregistered trigger and perturbation cases. A topological outlier is an investigation signal, not evidence that a candidate model responded to a backdoor trigger.
AID-D-012.002: AID-D-012.003 uses topology only and remains applicable when no compatible node-feature channel exists; AID-D-012.002 requires node features and evaluates their relationships across edges. Degree or centrality anomalies must not be reported as structure-feature inconsistency, and missing feature channels do not invalidate topology-only measurement.
Implementation guidance
- AID-D-012.003-G001: Compute calibrated topology-only degree and betweenness outlier evidence from an explicitly loaded graph with deterministic sampling and robust zero-variance handling.
- AID-D-012.003-G002: Verify a signed AI-asset topology inventory under a pinned trust root, derive technical N/A only from absent or unsupported topology, and bind applicable scoring to exact graph and reference bytes.
AID-D-012.003-G001
Compute calibrated topology-only degree and betweenness outlier evidence from an explicitly loaded graph with deterministic sampling and robust zero-variance handling.
Before you begin
Use this topology-only detector only when a signed graph-family reference declares directedness, duplicate-edge and self-loop handling, directed degree mode, deterministic betweenness sampling size and seed, robust clean-reference centers/scales, minimum node coverage, and anomaly thresholds.
Policy-bound topology scorer
from __future__ import annotations
# File: gnn_security/topology_score.py
import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
import networkx as nx
import numpy as np
from safetensors.numpy import load
REQUIRED = {
"schema_version", "policy_version", "graph_family", "graph_sha256",
"calibration_corpus_sha256", "calibration_reference_sha256",
"calibration_method", "calibration_method_version",
"directed", "duplicate_edge_policy", "self_loop_policy", "degree_mode",
"betweenness_sample_size", "sampling_seed", "min_observed_nodes",
"degree_center", "degree_scale", "degree_z_threshold",
"betweenness_center", "betweenness_scale",
"betweenness_z_threshold", "max_anomalous_node_fraction",
"issued_at", "expires_at",
}
SHA256 = re.compile(r"^[0-9a-f]{64}$")
SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(os.environ["D012_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0:
raise RuntimeError("signed runtime-profile verification timeout is invalid")
_AIDEFEND_CAPTURED: dict[Path, bytes] = {}
def _aidefend_bytes(path: Path) -> bytes:
normalized = Path(path)
if normalized not in _AIDEFEND_CAPTURED:
_AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
return _AIDEFEND_CAPTURED[normalized]
def _aidefend_write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def _aidefend_strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + token)
),
)
def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
lines = raw.decode("utf-8", errors="strict").splitlines()
if not lines or any(not line.strip() for line in lines):
raise ValueError("JSONL input is empty or contains a blank record")
return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]
def _aidefend_capture_signed_bytes(
payload_path: Path,
proof_path: Path,
key_reference: str | Path,
timeout_seconds: float,
) -> bytes:
payload_path = Path(payload_path)
proof_path = Path(proof_path)
if payload_path == proof_path:
raise RuntimeError("signed payload and signature paths collide")
payload_raw = payload_path.read_bytes()
proof_raw = proof_path.read_bytes()
key_text = str(key_reference)
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
)) or key_text.startswith("pkcs11:")
key_raw = None if remote_key else (
key_text.encode("utf-8")
if key_text.startswith("-----BEGIN ")
else Path(key_text).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
proof_snapshot = root / "payload.sig"
_aidefend_write_private(payload_snapshot, payload_raw)
_aidefend_write_private(proof_snapshot, proof_raw)
captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
key_argument = key_text
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
_aidefend_write_private(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
["cosign", "verify-blob", "--key", key_argument,
"--bundle", str(proof_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True,
timeout=timeout_seconds,
)
if any(
not hmac.compare_digest(snapshot.read_bytes(), expected)
for snapshot, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for snapshot, _ in captured:
os.chmod(snapshot, 0o600)
_AIDEFEND_CAPTURED[payload_path] = payload_raw
return payload_raw
def sha256_file(path):
return hashlib.sha256(_aidefend_bytes(Path(path))).hexdigest()
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 positive_int(value, field):
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{field} must be a positive integer")
return value
def parse_time(value):
if not isinstance(value, str):
raise ValueError("reference timestamp is not text")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("reference timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def load_reference(
manifest_path, signature_path, public_key_path, graph_path,
calibration_corpus_path, calibration_reference_path,
):
reference_raw = _aidefend_capture_signed_bytes(
Path(manifest_path), Path(signature_path), public_key_path,
SIGNATURE_VERIFY_TIMEOUT_SECONDS,
)
reference = _aidefend_strict_json(reference_raw)
if not isinstance(reference, dict) or set(reference) != REQUIRED:
raise ValueError("signed topology reference schema mismatch")
if reference["schema_version"] != "aidefend.topology_reference.v1":
raise ValueError("unsupported topology reference")
if not isinstance(reference["policy_version"], str) or not reference["policy_version"]:
raise ValueError("missing policy version")
if not isinstance(reference["graph_family"], str) or not reference["graph_family"]:
raise ValueError("missing graph family")
if not isinstance(reference["graph_sha256"], str) or not SHA256.fullmatch(reference["graph_sha256"]):
raise ValueError("invalid graph digest")
if sha256_file(graph_path) != reference["graph_sha256"]:
raise ValueError("graph digest mismatch")
for field, path in (
("calibration_corpus_sha256", calibration_corpus_path),
("calibration_reference_sha256", calibration_reference_path),
):
if not isinstance(reference[field], str) or not SHA256.fullmatch(reference[field]):
raise ValueError(f"{field} is invalid")
if sha256_file(path) != reference[field]:
raise ValueError(f"{field} artifact mismatch")
for field in ("calibration_method", "calibration_method_version"):
if not isinstance(reference[field], str) or not reference[field]:
raise ValueError(f"{field} is missing")
if not isinstance(reference["directed"], bool):
raise ValueError("directed must be boolean")
if reference["duplicate_edge_policy"] not in {"reject", "collapse"}:
raise ValueError("unsupported duplicate-edge policy")
if reference["self_loop_policy"] not in {"reject", "include"}:
raise ValueError("unsupported self-loop policy")
valid_modes = {"in", "out", "total"} if reference["directed"] else {"total"}
if reference["degree_mode"] not in valid_modes:
raise ValueError("degree mode conflicts with directedness")
positive_int(reference["betweenness_sample_size"], "betweenness_sample_size")
positive_int(reference["min_observed_nodes"], "min_observed_nodes")
seed = reference["sampling_seed"]
if isinstance(seed, bool) or not isinstance(seed, int) or seed < 0:
raise ValueError("sampling_seed must be a non-negative integer")
for field in (
"degree_center", "degree_scale", "degree_z_threshold",
"betweenness_center", "betweenness_scale",
"betweenness_z_threshold", "max_anomalous_node_fraction",
):
reference[field] = finite(reference[field], field)
if reference["degree_scale"] < 0 or reference["betweenness_scale"] < 0:
raise ValueError("reference scale cannot be negative")
if reference["degree_z_threshold"] <= 0 or reference["betweenness_z_threshold"] <= 0:
raise ValueError("z thresholds must be positive")
if not 0 <= reference["max_anomalous_node_fraction"] <= 1:
raise ValueError("anomalous-node fraction outside [0,1]")
issued = parse_time(reference["issued_at"])
expires = parse_time(reference["expires_at"])
now = datetime.now(timezone.utc)
if issued > now or expires <= issued or now >= expires:
raise ValueError("signed topology reference is not currently valid")
return reference
def build_graph(graph_path, reference):
tensors = load(_aidefend_bytes(Path(graph_path)))
if set(tensors) != {"edge_index", "node_count"}:
raise ValueError("SafeTensors topology schema mismatch")
edges = np.asarray(tensors["edge_index"])
count_tensor = np.asarray(tensors["node_count"])
if (
edges.ndim != 2 or edges.shape[0] != 2
or not np.issubdtype(edges.dtype, np.integer)
or count_tensor.size != 1
or not np.issubdtype(count_tensor.dtype, np.integer)
):
raise ValueError("topology tensor shape/type mismatch")
node_count = int(count_tensor.reshape(-1)[0])
if node_count < 1:
raise ValueError("node_count must be positive")
if edges.size and (edges.min() < 0 or edges.max() >= node_count):
raise ValueError("edge endpoint outside node range")
seen, normalized = set(), []
for source, target in edges.T.tolist():
source, target = int(source), int(target)
if source == target and reference["self_loop_policy"] == "reject":
raise ValueError("self-loop violates signed policy")
key = (
(source, target)
if reference["directed"]
else tuple(sorted((source, target)))
)
if key in seen:
if reference["duplicate_edge_policy"] == "reject":
raise ValueError("duplicate edge violates signed policy")
continue
seen.add(key)
normalized.append((source, target))
graph = nx.DiGraph() if reference["directed"] else nx.Graph()
graph.add_nodes_from(range(node_count))
graph.add_edges_from(normalized)
return graph, int(edges.shape[1]), len(normalized)
def metric_flags(values, center, scale, threshold):
flags = {}
scores = {}
for node, value in values.items():
value = finite(value, f"metric[{node}]")
if scale == 0:
flags[node] = not math.isclose(value, center, rel_tol=0, abs_tol=1e-12)
scores[node] = None if flags[node] else 0.0
else:
score = abs(value - center) / scale
flags[node] = score > threshold
scores[node] = score
return flags, scores
def score(
graph_path, manifest_path, signature_path, public_key_path,
calibration_corpus_path, calibration_reference_path,
):
_AIDEFEND_CAPTURED.clear()
reference = load_reference(
manifest_path, signature_path, public_key_path, graph_path,
calibration_corpus_path, calibration_reference_path,
)
graph, raw_edge_count, effective_edge_count = build_graph(
graph_path, reference
)
node_count = graph.number_of_nodes()
if (
node_count < reference["min_observed_nodes"]
or reference["betweenness_sample_size"] > node_count
):
return {
"status": "INSUFFICIENT_DATA",
"reason": "signed node/sampling coverage requirement not met",
"node_count": node_count,
}
if reference["directed"] and reference["degree_mode"] == "in":
degree = dict(graph.in_degree())
elif reference["directed"] and reference["degree_mode"] == "out":
degree = dict(graph.out_degree())
else:
degree = dict(graph.degree())
sample_size = reference["betweenness_sample_size"]
betweenness = nx.betweenness_centrality(
graph,
k=None if sample_size == node_count else sample_size,
normalized=True,
seed=reference["sampling_seed"],
)
degree_flags, degree_scores = metric_flags(
degree,
reference["degree_center"],
reference["degree_scale"],
reference["degree_z_threshold"],
)
centrality_flags, centrality_scores = metric_flags(
betweenness,
reference["betweenness_center"],
reference["betweenness_scale"],
reference["betweenness_z_threshold"],
)
anomalous_nodes = sorted(
node for node in graph
if degree_flags[node] or centrality_flags[node]
)
anomaly_fraction = len(anomalous_nodes) / node_count
finding = anomaly_fraction > reference["max_anomalous_node_fraction"]
all_node_scores = [
{
"node_id": node,
"degree": degree[node],
"degree_score": degree_scores[node],
"degree_anomaly": degree_flags[node],
"betweenness": betweenness[node],
"betweenness_score": centrality_scores[node],
"betweenness_anomaly": centrality_flags[node],
}
for node in sorted(graph)
]
score_artifact_sha256 = hashlib.sha256(
json.dumps(
all_node_scores, sort_keys=True, separators=(",", ":"),
allow_nan=False,
).encode()
).hexdigest()
detail = [
record for record in all_node_scores
if record["degree_anomaly"] or record["betweenness_anomaly"]
]
return {
"schema_version": "aidefend.topology_result.v1",
"status": "ANOMALY_FOUND" if finding else "NO_THRESHOLD_BREACH",
"policy_version": reference["policy_version"],
"graph_family": reference["graph_family"],
"graph_sha256": reference["graph_sha256"],
"reference_sha256": sha256_file(manifest_path),
"calibration_corpus_sha256": reference["calibration_corpus_sha256"],
"calibration_reference_sha256": reference["calibration_reference_sha256"],
"calibration_method": reference["calibration_method"],
"calibration_method_version": reference["calibration_method_version"],
"node_score_artifact_sha256": score_artifact_sha256,
"node_scores": all_node_scores,
"directed": reference["directed"],
"degree_mode": reference["degree_mode"],
"duplicate_edge_policy": reference["duplicate_edge_policy"],
"self_loop_policy": reference["self_loop_policy"],
"sampling_seed": reference["sampling_seed"],
"betweenness_sample_size": sample_size,
"node_count": node_count,
"raw_edge_count": raw_edge_count,
"effective_edge_count": effective_edge_count,
"anomalous_node_count": len(anomalous_nodes),
"anomalous_node_fraction": anomaly_fraction,
"max_anomalous_node_fraction": reference["max_anomalous_node_fraction"],
"anomalous_nodes": detail,
}
def safe_score(*args, **kwargs):
try:
return score(*args, **kwargs)
except Exception as exc:
return {
"status": "ERROR",
"reason": "reference, topology, or scoring verification failed",
"error_class": type(exc).__name__,
}
Evidence and scope: retain the signed reference and signature, graph digest, bound clean calibration corpus/reference digests and method/version, directedness and edge policies, raw/effective edge counts, signed sampling seed and size, and the canonical digest of the complete per-node metric artifact, result, and exception class. Independently replay the exact topology. NO_THRESHOLD_BREACH means only that this signed degree/betweenness policy did not breach; it is not proof that the graph or model is clean.
Run the detector and route findings
Action: compute deterministic degree and betweenness evidence for the complete eligible node population under the exact signed topology reference and emit the calibrated topology-only verdict.
Boundary: Re-verify the signed technical topology inventory before scoring. This stage owns degree and centrality scoring only, not structure-feature detection or graph admission.
Share graph identity without confusing the two measurements
D-012.002 measures structure-feature consistency; this method measures topology only. Their SafeTensors payload contracts intentionally differ. When both run on one feature-bearing snapshot, generate the topology projection below from the same verified source bytes. Preserve isolated nodes, edge order, orientation and the stable node-ID mapping; inferring node_count from the largest edge endpoint silently drops isolated nodes. Publish and sign the derived payload and bind its digest plus source snapshot digest in the topology authority manifest. Do not reuse the feature-payload signature as if it signed these different bytes, or count one score as both control outcomes.
# File: graph_io/topology_projection.py
import hashlib
import numpy as np
from safetensors.numpy import load, save
def topology_projection(verified_feature_payload: bytes):
tensors = load(verified_feature_payload)
if set(tensors) != {"edge_index", "node_features"}:
raise ValueError("feature graph payload schema differs")
edges, features = tensors["edge_index"], tensors["node_features"]
if (edges.ndim != 2 or edges.shape[0] != 2
or not np.issubdtype(edges.dtype, np.integer)
or features.ndim != 2 or features.shape[0] < 1):
raise ValueError("graph shape/type differs")
count = features.shape[0]
if edges.size and (edges.min() < 0 or edges.max() >= count):
raise ValueError("edge outside stable node-ID range")
payload = save({"edge_index": np.ascontiguousarray(edges),
"node_count": np.asarray([count], dtype=np.int64)})
return payload, {"source_graph_sha256": hashlib.sha256(verified_feature_payload).hexdigest(),
"topology_payload_sha256": hashlib.sha256(payload).hexdigest(),
"node_count": count}AID-D-012.003-G002
Verify a signed AI-asset topology inventory under a pinned trust root, derive technical N/A only from absent or unsupported topology, and bind applicable scoring to exact graph and reference bytes.
Verify the technical topology inventory
The signed inventory records whether the exact AI asset exposes topology supported by this detector. It is not a separate applicability decision. NO_GRAPH_TOPOLOGY and UNSUPPORTED_TOPOLOGY are NOT_APPLICABLE; once SUPPORTED_GRAPH is declared, graph and calibration inputs are mandatory and failures are ERROR. Inadequate measured-node or betweenness-sample coverage remains INSUFFICIENT_DATA.
# File: gnn_security/verified_topology.py
import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path
from gnn_security.topology_score import safe_score
TRUST_KEY = Path("/opt/aidefend/trust/gnn-topology-authority.pub")
FIELDS = {
"schema_version", "asset_id", "asset_sha256",
"topology_scope_state", "graph_sha256",
}
STATES = {"SUPPORTED_GRAPH", "NO_GRAPH_TOPOLOGY", "UNSUPPORTED_TOPOLOGY"}
_AIDEFEND_CAPTURED: dict[Path, bytes] = {}
def _aidefend_bytes(path: Path) -> bytes:
normalized = Path(path)
if normalized not in _AIDEFEND_CAPTURED:
_AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
return _AIDEFEND_CAPTURED[normalized]
def _aidefend_write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def _aidefend_strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + token)
),
)
def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
lines = raw.decode("utf-8", errors="strict").splitlines()
if not lines or any(not line.strip() for line in lines):
raise ValueError("JSONL input is empty or contains a blank record")
return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]
def _aidefend_capture_signed_bytes(
payload_path: Path,
proof_path: Path,
key_reference: str | Path,
timeout_seconds: float,
) -> bytes:
payload_path = Path(payload_path)
proof_path = Path(proof_path)
if payload_path == proof_path:
raise RuntimeError("signed payload and signature paths collide")
payload_raw = payload_path.read_bytes()
proof_raw = proof_path.read_bytes()
key_text = str(key_reference)
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
)) or key_text.startswith("pkcs11:")
key_raw = None if remote_key else (
key_text.encode("utf-8")
if key_text.startswith("-----BEGIN ")
else Path(key_text).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
proof_snapshot = root / "payload.sig"
_aidefend_write_private(payload_snapshot, payload_raw)
_aidefend_write_private(proof_snapshot, proof_raw)
captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
key_argument = key_text
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
_aidefend_write_private(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
["cosign", "verify-blob", "--key", key_argument,
"--bundle", str(proof_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True,
timeout=timeout_seconds,
)
if any(
not hmac.compare_digest(snapshot.read_bytes(), expected)
for snapshot, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for snapshot, _ in captured:
os.chmod(snapshot, 0o600)
_AIDEFEND_CAPTURED[payload_path] = payload_raw
return payload_raw
def sha256_file(path: Path) -> str:
return hashlib.sha256(_aidefend_bytes(Path(path))).hexdigest()
def verify_timeout() -> float:
value = float(os.environ["D012_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(value) or value <= 0:
raise ValueError("signature verification timeout must be finite and positive")
return value
def verify_inventory(path: Path, signature: Path, asset_path: Path,
graph_path: Path | None) -> tuple[dict, str]:
raw = _aidefend_capture_signed_bytes(
path, signature, TRUST_KEY, verify_timeout(),
)
inventory = _aidefend_strict_json(raw)
if (not isinstance(inventory, dict) or set(inventory) != FIELDS
or inventory["schema_version"] != "aidefend.ai-asset-topology-inventory.v1"
or not isinstance(inventory["asset_id"], str) or not inventory["asset_id"]
or inventory["topology_scope_state"] not in STATES
or inventory["asset_sha256"] != sha256_file(asset_path)):
raise ValueError("signed topology inventory differs")
state = inventory["topology_scope_state"]
if state == "SUPPORTED_GRAPH":
if graph_path is None or inventory["graph_sha256"] != sha256_file(graph_path):
raise ValueError("supported topology graph binding differs")
elif graph_path is not None or inventory["graph_sha256"] is not None:
raise ValueError("out-of-scope asset must not supply detector graph bytes")
return inventory, hashlib.sha256(raw).hexdigest()
def evaluate(inventory_path, inventory_signature, *, asset_path, graph_path=None,
manifest_path=None, signature_path=None,
calibration_corpus_path=None, calibration_reference_path=None):
_AIDEFEND_CAPTURED.clear()
inventory, inventory_sha256 = verify_inventory(
inventory_path, inventory_signature, asset_path, graph_path,
)
if inventory["topology_scope_state"] != "SUPPORTED_GRAPH":
return {
"status": "NOT_APPLICABLE",
"reason": inventory["topology_scope_state"],
"asset_sha256": inventory["asset_sha256"],
"inventory_sha256": inventory_sha256,
}
required = {
"manifest_path": manifest_path,
"signature_path": signature_path,
"calibration_corpus_path": calibration_corpus_path,
"calibration_reference_path": calibration_reference_path,
}
missing = sorted(name for name, path in required.items() if path is None)
if missing:
raise FileNotFoundError("supported topology lacks required inputs: " + ", ".join(missing))
graph_raw = _aidefend_bytes(Path(graph_path))
with tempfile.TemporaryDirectory(prefix="aidefend-d012003-graph-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
graph_snapshot = root / "graph.safetensors"
_aidefend_write_private(graph_snapshot, graph_raw)
try:
result = safe_score(
graph_snapshot, manifest_path, signature_path, TRUST_KEY,
calibration_corpus_path, calibration_reference_path,
)
if not hmac.compare_digest(graph_snapshot.read_bytes(), graph_raw):
raise RuntimeError("private graph snapshot changed during scoring")
finally:
os.chmod(graph_snapshot, 0o600)
result.update({
"asset_sha256": inventory["asset_sha256"],
"topology_inventory_sha256": inventory_sha256,
})
return result
def safe_evaluate(**kwargs):
try:
return evaluate(**kwargs)
except Exception as error:
return {
"status": "ERROR",
"reason": "technical inventory or required topology input verification failed",
"error_class": type(error).__name__,
}
Verification: preserve the signed topology inventory, pinned signer identity, exact graph and clean-calibration digests, complete per-node score population, directedness and edge policies, result, and independent replay. A deleted supported node or missing calibration object must not produce PASS.
Action: verify the exact asset and technical topology state under the pinned key, then either return deterministic technical N/A or run the calibrated scorer over the complete node population.
Boundary: This wrapper resolves only topology capability and trust bindings. It does not create a separate evidence or governance decision lane, and no inventory check becomes a scoring PASS.
Open-source tools
- NetworkX
- python-igraph
- safetensors
- NumPy
- SciPy
- Cosign
Threat mappings
MITRE ATLAS
- AML.T0020 Training Data Poisoning
- AML.T0043.004 Craft Adversarial Data: Insert Backdoor Trigger
MAESTRO
- Data Poisoning (L2)
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- ML02:2023 Data Poisoning Attack
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.013 Data Poisoning
- NISTAML.023 Backdoor Poisoning
- NISTAML.024 Targeted Poisoning
Cisco Integrated AI Security and Safety Framework
- N/A
Google Secure AI Framework 2.0 - Risks
- DP: Data Poisoning
Databricks AI Security Framework 3.0
- Datasets 3.1: Data poisoning