Scope boundary

Owns anomaly findings for node-feature relationships across observed edges using a signed threshold policy calibrated on representative clean graphs from the same graph family and feature schema. It does not self-calibrate from the graph under test, mutate edges, or claim an outlier is malicious.

AID-D-012.001: AID-D-012.002 evaluates structure-feature consistency in graph data; AID-D-012.001 evaluates a candidate GNN's behavioral response to preregistered perturbations. Graph data anomalies and model trigger response have different measured assets and outcomes.

AID-D-012.003: AID-D-012.002 requires compatible node features and scores feature relationships across edges; AID-D-012.003 requires only supported topology and scores degree and centrality. Featureless graphs can be applicable to topology detection while being technically not applicable to structure-feature analysis.

Implementation guidance

• AID-D-012.002-G001: Score edge endpoint feature similarity against a signed clean-graph reference and emit an evidence-bound structure-feature result with explicit unmeasured coverage.
• AID-D-012.002-G002: Verify a signed graph-feature inventory under a pinned trust root, derive technical N/A only from an absent or unsupported feature channel, and bind applicable scoring to exact graph, schema, and reference bytes.

AID-D-012.002-G001

Before you begin

Use this only when a signed clean graph-family reference defines a calibrated endpoint-similarity threshold, the maximum acceptable low-similarity fraction, a minimum measured-edge count, and a maximum unmeasured-edge fraction.

Signed-reference scorer

from __future__ import annotations

# File: gnn_security/structure_feature_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 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",
    "low_similarity_threshold", "max_low_similarity_fraction",
    "min_measured_edges", "max_unmeasured_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 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 reference schema mismatch")
    if reference["schema_version"] != "aidefend.structure_feature_reference.v1":
        raise ValueError("unsupported reference schema")
    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 artifact 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")

    threshold = finite(reference["low_similarity_threshold"], "low_similarity_threshold")
    low_cap = finite(
        reference["max_low_similarity_fraction"],
        "max_low_similarity_fraction",
    )
    unmeasured_cap = finite(
        reference["max_unmeasured_fraction"],
        "max_unmeasured_fraction",
    )
    if not -1 <= threshold <= 1:
        raise ValueError("cosine threshold is outside [-1,1]")
    if not 0 <= low_cap <= 1 or not 0 <= unmeasured_cap <= 1:
        raise ValueError("coverage/finding fractions are outside [0,1]")
    minimum = reference["min_measured_edges"]
    if isinstance(minimum, bool) or not isinstance(minimum, int) or minimum < 1:
        raise ValueError("min_measured_edges must be a positive integer")
    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 reference is not currently valid")
    return reference


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,
    )
    tensors = load(_aidefend_bytes(Path(graph_path)))
    if set(tensors) != {"edge_index", "node_features"}:
        raise ValueError("SafeTensors graph schema mismatch")
    edges = np.asarray(tensors["edge_index"])
    features = np.asarray(tensors["node_features"], dtype=np.float64)
    if edges.ndim != 2 or edges.shape[0] != 2 or features.ndim != 2:
        raise ValueError("graph tensor shape mismatch")
    if edges.shape[1] < 1 or features.shape[0] < 1 or features.shape[1] < 1:
        raise ValueError("graph tensors are empty")
    if not np.issubdtype(edges.dtype, np.integer):
        raise ValueError("edge_index must be integral")
    if not np.isfinite(features).all():
        raise ValueError("node features contain NaN or infinity")
    if edges.min() < 0 or edges.max() >= features.shape[0]:
        raise ValueError("edge endpoint is outside node range")

    left = features[edges[0]]
    right = features[edges[1]]
    left_norm = np.linalg.norm(left, axis=1)
    right_norm = np.linalg.norm(right, axis=1)
    measurable = (left_norm > 0) & (right_norm > 0)
    similarities = np.empty(edges.shape[1], dtype=np.float64)
    similarities.fill(np.nan)
    similarities[measurable] = (
        (left[measurable] * right[measurable]).sum(axis=1)
        / (left_norm[measurable] * right_norm[measurable])
    )

    total = int(edges.shape[1])
    measured = int(measurable.sum())
    unmeasured = total - measured
    unmeasured_fraction = unmeasured / total
    if (
        measured < reference["min_measured_edges"]
        or unmeasured_fraction > reference["max_unmeasured_fraction"]
    ):
        return {
            "status": "INSUFFICIENT_DATA",
            "reason": "signed measurement coverage requirement not met",
            "total_edge_count": total,
            "measured_edge_count": measured,
            "unmeasured_edge_count": unmeasured,
            "unmeasured_fraction": unmeasured_fraction,
        }

    score_records = [
        {
            "edge_index": index,
            "similarity": (
                float(similarities[index]) if measurable[index] else None
            ),
        }
        for index in range(total)
    ]
    score_artifact_sha256 = hashlib.sha256(
        json.dumps(
            score_records, sort_keys=True, separators=(",", ":"),
            allow_nan=False,
        ).encode()
    ).hexdigest()
    measured_values = similarities[measurable]
    low_count = int(
        (measured_values < reference["low_similarity_threshold"]).sum()
    )
    low_fraction = low_count / measured
    finding = low_fraction > reference["max_low_similarity_fraction"]
    return {
        "schema_version": "aidefend.structure_feature_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"],
        "edge_score_artifact_sha256": score_artifact_sha256,
        "edge_scores": score_records,
        "total_edge_count": total,
        "measured_edge_count": measured,
        "unmeasured_edge_count": unmeasured,
        "unmeasured_fraction": unmeasured_fraction,
        "low_similarity_edge_count": low_count,
        "low_similarity_fraction": low_fraction,
        "threshold": reference["low_similarity_threshold"],
        "max_low_similarity_fraction": reference["max_low_similarity_fraction"],
        "max_unmeasured_fraction": reference["max_unmeasured_fraction"],
    }


def safe_score(*args, **kwargs):
    try:
        return score(*args, **kwargs)
    except Exception as exc:
        return {
            "status": "ERROR",
            "reason": "reference, artifact, 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, explicit measured/unmeasured counts, threshold, both signed coverage caps, and the canonical digest of the complete per-edge score artifact (including null unmeasured edges), result, and exception class. Independently replay the exact graph and reference. NO_THRESHOLD_BREACH applies only to this calibrated structure-feature relationship; it is not proof that the graph or model is clean.

Run the detector and route findings

Action: score every eligible edge against the exact signed structure-feature reference, retain the complete per-edge artifact and emit a calibrated anomaly verdict with explicit unmeasured coverage.

Boundary: Re-verify the signed technical graph-feature inventory before scoring. This stage owns structure-feature scoring only, not topology-only detection or graph admission.

AID-D-012.002-G002

Verify the technical graph-feature inventory

The signed inventory describes an observable property of the exact graph, not a separate applicability decision. NOT_APPLICABLE is limited to NO_NODE_FEATURES or UNSUPPORTED_FEATURE_SCHEMA. Once the inventory declares SUPPORTED, every graph, schema, calibration, and reference input is mandatory; absence or mismatch is ERROR, while the scorer reserves INSUFFICIENT_DATA for inadequate measurable-edge coverage.

# File: gnn_security/verified_structure_feature.py
import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path

from gnn_security.structure_feature_score import safe_score

TRUST_KEY = Path("/opt/aidefend/trust/gnn-structure-feature-authority.pub")
FIELDS = {
    "schema_version", "asset_id", "graph_sha256",
    "feature_scope_state", "feature_schema_sha256",
}
STATES = {"SUPPORTED", "NO_NODE_FEATURES", "UNSUPPORTED_FEATURE_SCHEMA"}


_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, graph_path: Path) -> 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.graph-feature-inventory.v1"
            or not isinstance(inventory["asset_id"], str) or not inventory["asset_id"]
            or inventory["feature_scope_state"] not in STATES
            or inventory["graph_sha256"] != sha256_file(graph_path)):
        raise ValueError("signed graph-feature inventory differs")
    schema_digest = inventory["feature_schema_sha256"]
    if inventory["feature_scope_state"] == "SUPPORTED":
        if not isinstance(schema_digest, str) or len(schema_digest) != 64:
            raise ValueError("supported feature channel lacks a schema digest")
    elif schema_digest is not None:
        raise ValueError("out-of-scope feature channel must not claim a schema digest")
    return inventory, hashlib.sha256(raw).hexdigest()


def evaluate(inventory_path, inventory_signature, *, graph_path,
             feature_schema_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, graph_path,
    )
    if inventory["feature_scope_state"] != "SUPPORTED":
        return {
            "status": "NOT_APPLICABLE",
            "reason": inventory["feature_scope_state"],
            "graph_sha256": inventory["graph_sha256"],
            "inventory_sha256": inventory_sha256,
        }
    required = {
        "feature_schema_path": feature_schema_path,
        "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 graph lacks required inputs: " + ", ".join(missing))
    if sha256_file(feature_schema_path) != inventory["feature_schema_sha256"]:
        raise ValueError("feature schema differs from signed graph inventory")
    graph_raw = _aidefend_bytes(Path(graph_path))
    with tempfile.TemporaryDirectory(prefix="aidefend-d012002-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({
        "graph_inventory_sha256": inventory_sha256,
        "feature_schema_sha256": inventory["feature_schema_sha256"],
    })
    return result


def safe_evaluate(**kwargs):
    try:
        return evaluate(**kwargs)
    except Exception as error:
        return {
            "status": "ERROR",
            "reason": "technical inventory or required graph input verification failed",
            "error_class": type(error).__name__,
        }

Verification: preserve the signed graph-feature inventory, pinned signer identity, exact feature-schema and calibration-reference digests, complete per-edge score population, result, and independent replay. Re-signing an inventory with a deleted supported edge or missing prerequisite must not produce PASS.

Action: verify the exact graph and technical feature state under the pinned key, then either return the deterministic technical N/A result or run the calibrated scorer over the complete edge population.

Boundary: This wrapper resolves only the graph-feature 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

• PyTorch Geometric
• Deep Graph Library (DGL)
• PyGOD (graph outlier scoring; calibration policy and complete-coverage evidence remain external)
• 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)
• Data Tampering (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