Scope boundary

Owns secret challenge execution, null calibration, complete population reconciliation, statistical scoring, and replayable black-box ownership review evidence for a compatible suspect endpoint. A trigger is not an ownership verdict.

AID-H-013.002: AID-H-013.003 produces a bounded statistical review trigger from endpoint behavior; AID-H-013.002 proves exact artifact or canonical tensor-state match from authorized suspect bytes. Behavioral similarity cannot be reported as byte or tensor identity.

AID-DV-004: AID-H-013.003 uses a preregistered statistical challenge population to test an ownership hypothesis; AID-DV-004 uses deception-specific markers and beacons to expose misuse or leakage. Behavioral fingerprinting and trap activation have different null models and investigation claims.

Warning

Statistical Review Signal, Not Standalone Ownership Proof

Behavior can converge across unrelated models and can change through fine-tuning, distillation, system prompts, routing, sampling, or adaptive countermeasures. Use a preregistered secret population, independent null calibration, policy-bound thresholds, and independent replay. Treat a trigger only as a reason for bounded forensic review.

Implementation guidance

• AID-H-013.003-G001: Use a secret, signed and independently calibrated structured-response challenge population to produce a statistically bounded black-box ownership review trigger.

AID-H-013.003-G001

Use black-box fingerprint verification only when the model owner pre-registered a secret challenge population for the exact released model, the suspect endpoint exposes a deterministic structured response class, and the investigation has authorization to query it. A signature-verified policy binds the endpoint, mTLS trust material, registry and scorer schema, exact challenge population, retry/error budget, independent null-model calibration receipt, and statistical review thresholds. Free-form marker matching and fixed universal sample counts are not accepted.

Runnable calibrated black-box verifier

# File: model_ip/blackbox_fingerprint_verifier.py
from __future__ import annotations

from contextlib import ExitStack
import hashlib
import json
import math
import os
import ssl
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path

import httpx
from scipy.stats import binomtest

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

OWNERSHIP_CHALLENGE_BUNDLE_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_OWNERSHIP_CHALLENGE_BUNDLE_VERIFY_TIMEOUT_SECONDS")



class InsufficientData(RuntimeError):
    pass


class ContractViolation(RuntimeError):
    pass


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


def verified_json(
    path: Path, signature: Path, public_key: Path
) -> tuple[dict, str]:
    raw = path.read_bytes()
    sig = signature.read_bytes()
    if not raw or not sig:
        raise ValueError("signed JSON or signature is empty")
    with tempfile.TemporaryDirectory(prefix="ownership-verify-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        snapshot = root / "artifact.json"
        signature_snapshot = root / "artifact.sig"
        snapshot.write_bytes(raw)
        signature_snapshot.write_bytes(sig)
        os.chmod(snapshot, 0o400)
        os.chmod(signature_snapshot, 0o400)
        subprocess.run([
            "cosign", "verify-blob", "--key", str(public_key),
            "--bundle", str(signature_snapshot), str(snapshot),
        ], check=True, capture_output=True, text=True, timeout=OWNERSHIP_CHALLENGE_BUNDLE_VERIFY_TIMEOUT_SECONDS)
        if snapshot.read_bytes() != raw:
            raise ValueError("verified artifact changed")
    return json.loads(raw.decode("utf-8", errors="strict")), sha256(raw)


def load_contract(
    policy_path: Path,
    policy_signature: Path,
    policy_key: Path,
    registry_path: Path,
    registry_signature: Path,
    registry_key: Path,
    calibration_path: Path,
    calibration_signature: Path,
    calibration_key: Path,
) -> tuple[dict, dict, dict, str, str]:
    policy, policy_sha256 = verified_json(
        policy_path, policy_signature, policy_key
    )
    policy_fields = {
        "schema_version", "policy_version", "suspect_endpoint",
        "expected_model_route", "registry_sha256",
        "calibration_receipt_sha256", "ca_bundle_sha256",
        "client_certificate_sha256", "not_before", "not_after",
        "request_timeout_seconds", "response_max_bytes", "retry_attempts",
        "minimum_challenges", "maximum_p_value",
        "minimum_match_rate",
    }
    if not isinstance(policy, dict) or set(policy) != policy_fields:
        raise ValueError("ownership policy schema differs")
    if policy["schema_version"] != "aidefend.blackbox-ownership-policy.v1":
        raise ValueError("unsupported ownership policy")
    now = datetime.now(timezone.utc)
    not_before = datetime.fromisoformat(
        policy["not_before"].replace("Z", "+00:00")
    )
    not_after = datetime.fromisoformat(
        policy["not_after"].replace("Z", "+00:00")
    )
    if not not_before <= now <= not_after:
        raise InsufficientData("ownership policy is outside its time window")

    registry, registry_sha256 = verified_json(
        registry_path, registry_signature, registry_key
    )
    if registry_sha256 != policy["registry_sha256"]:
        raise ContractViolation("challenge registry digest differs")
    if (
        not isinstance(registry, dict)
        or set(registry) != {
            "schema_version", "reference_model_sha256", "challenges"
        }
        or registry["schema_version"]
        != "aidefend.blackbox-challenge-registry.v1"
    ):
        raise InsufficientData("challenge registry schema differs")
    challenges = registry["challenges"]
    if not isinstance(challenges, list) or len(challenges) < int(
        policy["minimum_challenges"]
    ):
        raise InsufficientData("challenge population is too small")
    ids = [row.get("id") for row in challenges if isinstance(row, dict)]
    if len(ids) != len(challenges) or len(set(ids)) != len(ids):
        raise InsufficientData("challenge IDs are missing or duplicated")
    for row in challenges:
        if set(row) != {"id", "request_body", "expected_response_class"}:
            raise InsufficientData("challenge schema differs")
        if not isinstance(row["request_body"], dict):
            raise InsufficientData("challenge request body must be an object")

    calibration, calibration_sha256 = verified_json(
        calibration_path, calibration_signature, calibration_key
    )
    if calibration_sha256 != policy["calibration_receipt_sha256"]:
        raise ContractViolation("calibration receipt digest differs")
    required_calibration = {
        "schema_version", "status", "registry_sha256",
        "reference_population_sha256", "null_match_probability",
        "challenge_count",
    }
    if (
        not isinstance(calibration, dict)
        or set(calibration) != required_calibration
        or calibration["schema_version"]
        != "aidefend.blackbox-null-calibration.v1"
        or calibration["status"] != "PASS"
        or calibration["registry_sha256"] != registry_sha256
        or int(calibration["challenge_count"]) != len(challenges)
    ):
        raise InsufficientData("calibration receipt is incomplete or unbound")
    null_probability = float(calibration["null_match_probability"])
    if not 0.0 < null_probability < 1.0:
        raise ValueError("null match probability is invalid")
    return policy, registry, calibration, policy_sha256, calibration_sha256


def run_verification(
    ca_bundle: Path,
    client_certificate: Path,
    client_private_key: Path,
    **contract_paths,
) -> dict:
    policy, registry, calibration, policy_sha256, calibration_sha256 = (
        load_contract(**contract_paths)
    )
    if sha256(ca_bundle.read_bytes()) != policy["ca_bundle_sha256"]:
        raise ContractViolation("mTLS CA bundle digest differs")
    if (
        sha256(client_certificate.read_bytes())
        != policy["client_certificate_sha256"]
    ):
        raise ContractViolation("mTLS client certificate digest differs")
    endpoint = httpx.URL(policy["suspect_endpoint"])
    if (
        endpoint.scheme != "https"
        or endpoint.username
        or endpoint.password
        or endpoint.fragment
    ):
        raise ContractViolation("suspect endpoint must be an absolute HTTPS URL")

    timeout = float(policy["request_timeout_seconds"])
    retries = int(policy["retry_attempts"])
    response_max_bytes = policy["response_max_bytes"]
    if (
        not timeout > 0.0 or retries < 0
        or isinstance(response_max_bytes, bool)
        or not isinstance(response_max_bytes, int)
        or response_max_bytes <= 0
    ):
        raise ValueError("timeout or retry policy is invalid")
    tls_context = ssl.create_default_context(cafile=str(ca_bundle))
    tls_context.minimum_version = ssl.TLSVersion.TLSv1_2
    tls_context.load_cert_chain(
        certfile=str(client_certificate), keyfile=str(client_private_key)
    )
    rows = []
    failures = []
    with httpx.Client(
        verify=tls_context,
        timeout=timeout,
        follow_redirects=False,
        trust_env=False,
    ) as client:
        for challenge in registry["challenges"]:
            response_raw = None
            payload = None
            last_error = None
            for _ in range(retries + 1):
                try:
                    with client.stream(
                        "POST", endpoint,
                        json=challenge["request_body"],
                        headers={
                            "X-AIDEFEND-Challenge-ID": challenge["id"],
                            "Cache-Control": "no-store",
                            "Accept": "application/json",
                        },
                    ) as candidate:
                        candidate.raise_for_status()
                        raw = bytearray()
                        for chunk in candidate.iter_bytes():
                            raw.extend(chunk)
                            if len(raw) > response_max_bytes:
                                raise ContractViolation("endpoint response exceeds signed byte cap")
                    response_raw = bytes(raw)
                    payload = json.loads(response_raw)
                    break
                except httpx.HTTPError as exc:
                    last_error = type(exc).__name__
            if response_raw is None or payload is None:
                failures.append({
                    "challenge_id": challenge["id"],
                    "error": last_error or "request_failed",
                })
                continue
            if (
                not isinstance(payload, dict)
                or set(payload) != {
                    "request_id", "response_class", "model_route"
                }
                or not isinstance(payload["request_id"], str)
                or not payload["request_id"]
                or not isinstance(payload["response_class"], str)
                or payload["model_route"] != policy["expected_model_route"]
            ):
                raise ContractViolation("structured response contract differs")
            matched = (
                payload["response_class"]
                == challenge["expected_response_class"]
            )
            rows.append({
                "challenge_id": challenge["id"],
                "matched": matched,
                "request_id_sha256": sha256(
                    payload["request_id"].encode()
                ),
                "response_sha256": sha256(response_raw),
            })

    if failures or len(rows) != len(registry["challenges"]):
        raise InsufficientData("challenge population is incomplete")
    hits = sum(row["matched"] for row in rows)
    count = len(rows)
    match_rate = hits / count
    p_value = float(binomtest(
        hits, count,
        p=float(calibration["null_match_probability"]),
        alternative="greater",
    ).pvalue)
    maximum_p_value = float(policy["maximum_p_value"])
    minimum_match_rate = float(policy["minimum_match_rate"])
    if (
        not math.isfinite(p_value)
        or not math.isfinite(match_rate)
        or not math.isfinite(maximum_p_value)
        or not math.isfinite(minimum_match_rate)
        or not 0.0 <= maximum_p_value <= 1.0
        or not 0.0 <= minimum_match_rate <= 1.0
    ):
        raise ValueError("statistic or policy threshold is non-finite or invalid")
    review_trigger = (
        p_value <= maximum_p_value
        and match_rate >= minimum_match_rate
    )
    return {
        "schema_version": "aidefend.blackbox-ownership-evidence.v1",
        "status": "PASS",
        "finding": "REVIEW_TRIGGER" if review_trigger else "NO_TRIGGER",
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "registry_sha256": policy["registry_sha256"],
        "calibration_receipt_sha256": calibration_sha256,
        "reference_model_sha256": registry["reference_model_sha256"],
        "suspect_endpoint_sha256": sha256(str(endpoint).encode()),
        "challenge_count": count,
        "match_count": hits,
        "match_rate": match_rate,
        "p_value": p_value,
        "null_match_probability": calibration["null_match_probability"],
        "results": rows,
    }


def run_with_status(**kwargs) -> dict:
    try:
        return run_verification(**kwargs)
    except ContractViolation as exc:
        return {"status": "FAIL", "reason": str(exc)}
    except InsufficientData as exc:
        return {"status": "INSUFFICIENT_DATA", "reason": str(exc)}
    except Exception as exc:
        return {"status": "ERROR", "reason": type(exc).__name__}

Dependency, evidence, independent replay, and outcomes

The null calibration receipt is a required upstream artifact produced by an evaluator that queried the complete policy-selected population of unrelated reference endpoints. It binds the registry digest, reference-population digest, challenge count, null match probability, evaluator identity, and raw result digest. The local boundary re-verifies the receipt and never inherits its PASS. Retain the signed policy, registry and calibration receipts, verification records, exact CA/client-certificate digests, endpoint and model-route identity, retry policy, complete challenge-ID population, hashed request IDs and responses, match vector, p-value, finding, immutable verifier image, and independent replay receipt in append-only storage. Keep challenge bodies and expected classes in a separate restricted secret store.

Action: Execute the complete secret challenge population over authenticated transport, compute the policy-bound calibrated binomial result, preserve only bounded evidence, and open forensic review only when the independently replayed finding is REVIEW_TRIGGER.

Open-source tools

• Cosign
• HTTPX
• SciPy

Threat mappings

MITRE ATLAS

• AML.T0024.002 Exfiltration via AI Inference API: Extract AI Model
• AML.T0048.004 External Harms: AI Intellectual Property Theft

MAESTRO

• Model Stealing (L1)

OWASP LLM Top 10 2026

• LLM06:2026 Unbounded Consumption (behavioral fingerprinting supports investigation of extracted or cloned models but does not prevent abusive queries at inference)

OWASP ML Top 10 2023

• ML05:2023 Model Theft

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.031 Model Extraction

Cisco Integrated AI Security and Safety Framework

• AITech-10.1 Model Extraction (Tests suspected replicas through a retained behavioral fingerprint)

Google Secure AI Framework 2.0 - Risks

• MRE: Model Reverse Engineering

Databricks AI Security Framework 3.0

• Model Management 8.2: Model theft