Scope boundary

Owns authorized discovery and reconciliation of organization-owned AI endpoints and agent services reachable only through private networks or internal control planes. It reports exact internal exposure facts but does not authenticate, segment, patch, contain, or evict those services.

AID-M-001.004: AID-M-001.006 discovers internally reachable organization-owned AI services; AID-M-001.004 discovers use of externally hosted AI services and SaaS AI features. Internal service exposure and external service consumption require different telemetry, ownership, and reconciliation populations.

AID-M-001.005: AID-M-001.006 uses authorized private-network and control-plane scope; AID-M-001.005 uses a credential-free outside-in vantage point against public exposure. A service visible in both paths remains two independently measured exposure claims rather than one merged scan result.

AID-D-016.001: AID-M-001.006 discovers endpoints and services from network and control-plane facts; AID-D-016.001 detects new or rogue agents from runtime agent-graph behavior. A reachable endpoint is not proof of an active agent node, and an anomalous agent graph can reveal actors that static endpoint discovery misses.

Implementation guidance

• AID-M-001.006-G001: Reconcile every authorized private-network and control-plane target through signed scope, scanner, and independent replay evidence.

AID-M-001.006-G001

Scope and upstream dependencies

This method verifies exact private AI endpoint discovery and inventory reconciliation. The scanner does not define its own scope. An asset-authority compiler outside the scanner derives the target population from signed IPAM, cloud-organization, Kubernetes-fleet, and CMDB exports. Missing source coverage, adapters, targets, or replay is INSUFFICIENT_DATA; an empty scan is never a pass by itself.

Step 1: Produce a signed, target-level scope outside the scanner

# File: policy/internal-discovery-policy.json
{
  "schema_version": "aidefend.internal_discovery_policy.v1",
  "policy_version": "internal-discovery-policy-7",
  "max_scope_age_seconds": 86400,
  "adapters": {
    "nmap-httpx": "registry.example.com/aidefend/nmap-httpx@sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
    "kubernetes": "registry.example.com/aidefend/kubernetes-discovery@sha256:1123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
    "aws-config": "registry.example.com/aidefend/aws-config-discovery@sha256:2123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
  }
}

# File: inventory/internal-ai-target-manifest.json
{
  "schema_version": "aidefend.internal_ai_target_manifest.v1",
  "scope_id": "internal-ai-estate-2026-07-16",
  "issued_at": "2026-07-16T00:00:00Z",
  "expires_at": "2026-07-17T00:00:00Z",
  "policy_version": "internal-discovery-policy-7",
  "policy_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "applicability": "APPLICABLE",
  "not_applicable_reason": null,
  "authoritative_sources": [
    {
      "source_id": "ipam",
      "path": "inventory/ipam-export.json",
      "signature_path": "inventory/ipam-export.sig",
      "sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
      "eligible_target_count": 1,
      "eligible_member_ids_sha256": "1111111111111111111111111111111111111111111111111111111111111111"
    },
    {
      "source_id": "kubernetes-fleet",
      "path": "inventory/kubernetes-fleet.json",
      "signature_path": "inventory/kubernetes-fleet.sig",
      "sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
      "eligible_target_count": 1,
      "eligible_member_ids_sha256": "2222222222222222222222222222222222222222222222222222222222222222"
    },
    {
      "source_id": "cloud-org",
      "path": "inventory/cloud-org.json",
      "signature_path": "inventory/cloud-org.sig",
      "sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
      "eligible_target_count": 1,
      "eligible_member_ids_sha256": "3333333333333333333333333333333333333333333333333333333333333333"
    }
  ],
  "targets": [
    {
      "target_id": "ip:10.40.1.8:443",
      "segment_id": "vpc-prod-us-east-1",
      "adapter": "nmap-httpx",
      "locator": {"host": "10.40.1.8", "port": 443, "probe_paths": ["/v1/models", "/mcp", "/.well-known/agent.json"]},
      "source_members": [{"source_id":"ipam","member_id":"ipam-10.40.1.8-443","raw_record_sha256":"4444444444444444444444444444444444444444444444444444444444444444","authorization_id":"scan-auth-prod"}]
    },
    {
      "target_id": "k8s:eks-prod:ai-platform",
      "segment_id": "eks-prod",
      "adapter": "kubernetes",
      "locator": {"context": "eks-prod", "namespace": "ai-platform"},
      "source_members": [{"source_id":"kubernetes-fleet","member_id":"eks-prod-ai-platform","raw_record_sha256":"5555555555555555555555555555555555555555555555555555555555555555","authorization_id":"scan-auth-prod"}]
    },
    {
      "target_id": "aws:111122223333:us-east-1",
      "segment_id": "aws-org",
      "adapter": "aws-config",
      "locator": {"account": "111122223333", "region": "us-east-1"},
      "source_members": [{"source_id":"cloud-org","member_id":"111122223333-us-east-1","raw_record_sha256":"6666666666666666666666666666666666666666666666666666666666666666","authorization_id":"scan-auth-prod"}]
    }
  ]
}

# File: inventory/declared-internal-ai-endpoints.json
{
  "schema_version": "aidefend.declared_internal_ai_inventory.v1",
  "inventory_version": "cmdb-2026-07-16T00:00:00Z",
  "items": [
    {
      "canonical_url": "https://model-gateway.ai-platform.svc/v1/models",
      "resource_id": "Service/ai-platform/model-gateway",
      "owner": "ai-platform",
      "environment": "production",
      "approved": true
    }
  ]
}

cosign sign-blob --yes --key env://INTERNAL_DISCOVERY_POLICY_SIGNING_KEY --bundle policy/internal-discovery-policy.sig policy/internal-discovery-policy.json
cosign sign-blob --yes --key env://ASSET_AUTHORITY_SIGNING_KEY --bundle inventory/internal-ai-target-manifest.sig inventory/internal-ai-target-manifest.json
cosign sign-blob --yes --key env://CMDB_SIGNING_KEY --bundle inventory/declared-internal-ai-endpoints.sig inventory/declared-internal-ai-endpoints.json

Each signed source export uses {"schema_version":"aidefend.internal-discovery-source.v1","source_id":"...","eligible_members":[{"member_id":"...","target_id":"...","raw_record_path":"...","raw_record_sha256":"...","authorization_id":"..."}]}. The compiler forms a canonical union by target_id, retaining every contributing source member and authorization; overlapping sources therefore do not inflate target counts or disappear. NOT_APPLICABLE is valid only when every policy-declared authoritative source has a verified empty member population, the target list and declared private-endpoint inventory are empty, and the signed reason is present.

Step 2: Execute every target and sign scanner receipts

Pin each adapter image to the digest in policy. The runner accepts only verified target-manifest locators, invokes subprocesses without a shell, applies per-target time/output limits, and records the exact argument vector. Network scans use explicit host/port targets; Kubernetes and cloud adapters use read-only identities. Raw output must be non-empty even when no AI endpoint is found.

nmap -sT -sV --version-light -Pn --open -p 443 10.40.1.8 -oX out/ip-10.40.1.8-443-nmap.xml
httpx -u https://10.40.1.8:443 -path /v1/models,/mcp,/.well-known/agent.json -json -status-code -hash sha256 -no-fallback -o out/ip-10.40.1.8-443-httpx.jsonl
kubectl --context eks-prod get services,ingresses,deployments,statefulsets,pods -n ai-platform -o json > out/k8s-eks-prod-ai-platform.json
aws configservice select-aggregate-resource-config --configuration-aggregator-name organization --expression "SELECT accountId, awsRegion, resourceType, resourceId, configuration WHERE accountId = '111122223333' AND awsRegion = 'us-east-1'" --output json > out/aws-111122223333-us-east-1.json

# File: out/k8s-eks-prod-ai-platform-observations.jsonl
{"target_id":"k8s:eks-prod:ai-platform","segment_id":"eks-prod","resource_id":"Service/ai-platform/model-gateway","canonical_url":"https://model-gateway.ai-platform.svc/v1/models","visibility":"private","status_code":401,"ai_evidence":["kubernetes_service","ai_specific_path:/v1/models"],"body_sha256":"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee","raw_record_sha256":"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff","observed_at":"2026-07-16T12:00:00Z"}

# File: out/internal-ai-scan-receipts.json
{
  "schema_version": "aidefend.internal_ai_scan_receipts.v1",
  "scope_id": "internal-ai-estate-2026-07-16",
  "target_manifest_sha256": "9999999999999999999999999999999999999999999999999999999999999999",
  "scanner_identity": "spiffe://company.example/scanner/internal-ai-primary",
  "receipts": [
    {
      "target_id": "ip:10.40.1.8:443",
      "adapter": "nmap-httpx",
      "adapter_image_digest": "registry.example.com/aidefend/nmap-httpx@sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
      "started_at": "2026-07-16T11:50:00Z",
      "finished_at": "2026-07-16T12:02:00Z",
      "status": "COMPLETE",
      "incomplete_reason": null,
      "raw_path": "out/ip-10.40.1.8-443-adapter.json",
      "raw_sha256": "3333333333333333333333333333333333333333333333333333333333333333",
      "observations_path": "out/ip-10.40.1.8-443-observations.jsonl",
      "observations_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
      "observation_count": 0
    },
    {
      "target_id": "k8s:eks-prod:ai-platform",
      "adapter": "kubernetes",
      "adapter_image_digest": "registry.example.com/aidefend/kubernetes-discovery@sha256:1123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
      "started_at": "2026-07-16T11:55:00Z",
      "finished_at": "2026-07-16T12:05:00Z",
      "status": "COMPLETE",
      "incomplete_reason": null,
      "raw_path": "out/k8s-eks-prod-ai-platform.json",
      "raw_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
      "observations_path": "out/k8s-eks-prod-ai-platform-observations.jsonl",
      "observations_sha256": "2222222222222222222222222222222222222222222222222222222222222222",
      "observation_count": 1
    },
    {
      "target_id": "aws:111122223333:us-east-1",
      "adapter": "aws-config",
      "adapter_image_digest": "registry.example.com/aidefend/aws-config-discovery@sha256:2123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
      "started_at": "2026-07-16T11:52:00Z",
      "finished_at": "2026-07-16T12:03:00Z",
      "status": "COMPLETE",
      "incomplete_reason": null,
      "raw_path": "out/aws-111122223333-us-east-1.json",
      "raw_sha256": "4444444444444444444444444444444444444444444444444444444444444444",
      "observations_path": "out/aws-111122223333-us-east-1-observations.jsonl",
      "observations_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
      "observation_count": 0
    }
  ]
}

cosign sign-blob --yes --key env://INTERNAL_SCANNER_SIGNING_KEY --bundle out/internal-ai-scan-receipts.sig out/internal-ai-scan-receipts.json
cosign verify-blob --key keys/internal-scanner.pub --bundle out/internal-ai-scan-receipts.sig out/internal-ai-scan-receipts.json

Write one receipt for every target, including zero-observation targets. A receipt is COMPLETE only after the adapter attempted the exact locator and retained raw evidence. The signed receipt binds the target manifest, scanner identity, adapter image, timestamps, raw digest, normalized digest, and observation count; it cannot be reconstructed later from normalized findings alone.

Step 3: Verify exact coverage and reconcile the signed inventory

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")
RECONCILIATION_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("M001006_RECONCILIATION_COMMAND_TIMEOUT_SECONDS")

# File: discovery/reconcile_internal_ai.py

import hashlib
import json
import os
import re
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

POLICY = Path("policy/internal-discovery-policy.json")
POLICY_SIG = Path("policy/internal-discovery-policy.sig")
TARGETS = Path("inventory/internal-ai-target-manifest.json")
TARGETS_SIG = Path("inventory/internal-ai-target-manifest.sig")
DECLARED = Path("inventory/declared-internal-ai-endpoints.json")
DECLARED_SIG = Path("inventory/declared-internal-ai-endpoints.sig")
RECEIPTS = Path("out/internal-ai-scan-receipts.json")
RECEIPTS_SIG = Path("out/internal-ai-scan-receipts.sig")
OUTPUT = Path("evidence/internal-ai-discovery-reconciliation.json")
SHA = re.compile(r"^[0-9a-f]{64}$")
IMAGE = re.compile(r"^.+@sha256:[0-9a-f]{64}$")


_CAPTURED_BYTES: dict[Path, bytes] = {}


def regular_bytes(path: Path, *, allow_empty: bool = False) -> bytes:
    descriptor = os.open(
        path,
        os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0),
    )
    with os.fdopen(descriptor, "rb") as handle:
        status = os.fstat(handle.fileno())
        if not stat.S_ISREG(status.st_mode):
            raise ValueError(f"input is not a regular file: {path}")
        value = handle.read()
    if not allow_empty and not value:
        raise ValueError(f"input is empty: {path}")
    return value


def captured_bytes(path: Path, *, allow_empty: bool = False) -> bytes:
    if path not in _CAPTURED_BYTES:
        _CAPTURED_BYTES[path] = regular_bytes(path, allow_empty=allow_empty)
    value = _CAPTURED_BYTES[path]
    if not allow_empty and not value:
        raise ValueError(f"captured input is empty: {path}")
    return value


def verify(path: Path, signature: Path, key: Path) -> None:
    payload = regular_bytes(path)
    signature_bytes = regular_bytes(signature)
    with tempfile.TemporaryDirectory(prefix="aidefend-internal-discovery-") as stage_name:
        stage = Path(stage_name)
        os.chmod(stage, 0o700)
        payload_snapshot = stage / "payload"
        signature_snapshot = stage / "payload.sig"
        payload_snapshot.write_bytes(payload)
        signature_snapshot.write_bytes(signature_bytes)
        os.chmod(payload_snapshot, 0o400)
        os.chmod(signature_snapshot, 0o400)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(key),
                "--bundle", str(signature_snapshot), str(payload_snapshot),
            ],
            check=True, capture_output=True, text=True,
            timeout=RECONCILIATION_COMMAND_TIMEOUT_SECONDS,
        )
        if (
            regular_bytes(payload_snapshot) != payload
            or regular_bytes(signature_snapshot) != signature_bytes
        ):
            raise ValueError("private signed-input snapshot changed during verification")
    _CAPTURED_BYTES[path] = payload


def digest(path: Path, *, allow_empty: bool = False) -> str:
    return hashlib.sha256(
        captured_bytes(path, allow_empty=allow_empty)
    ).hexdigest()


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


def reject_nonfinite(value):
    raise ValueError(f"non-finite JSON number: {value}")


def load_json(path: Path) -> object:
    return json.loads(
        captured_bytes(path).decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=reject_nonfinite,
    )


def load_jsonl(path: Path) -> list[object]:
    rows = []
    text = captured_bytes(path, allow_empty=True).decode("utf-8", errors="strict")
    for line_number, line in enumerate(text.splitlines(), start=1):
        if not line.strip():
            continue
        try:
            rows.append(json.loads(
                line,
                object_pairs_hook=unique_object,
                parse_constant=reject_nonfinite,
            ))
        except ValueError as error:
            raise ValueError(f"invalid JSONL row {line_number}: {path}") from error
    return rows


def when(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def canonical_url(value: str) -> str:
    parsed = urlsplit(value.strip())
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        raise ValueError("invalid endpoint URL")
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError("endpoint URL contains disallowed fields")
    host = parsed.hostname.lower().rstrip(".")
    port = parsed.port
    netloc = host if port is None or (parsed.scheme, port) in {("http", 80), ("https", 443)} else f"{host}:{port}"
    return urlunsplit((parsed.scheme, netloc, parsed.path.rstrip("/") or "/", "", ""))


now = datetime.now(timezone.utc)
outcome = "ERROR"
reasons: list[str] = []
findings: list[dict] = []
observations: list[dict] = []
coverage: list[dict] = []
scope: dict = {}
verifier_identity = os.environ.get("VERIFIER_WORKLOAD_ID")
try:
    if not verifier_identity:
        raise ValueError("VERIFIER_WORKLOAD_ID is required")
    verify(POLICY, POLICY_SIG, Path("keys/internal-discovery-policy.pub"))
    verify(TARGETS, TARGETS_SIG, Path("keys/asset-authority.pub"))
    verify(DECLARED, DECLARED_SIG, Path("keys/cmdb.pub"))

    policy = load_json(POLICY)
    scope = load_json(TARGETS)
    declared_doc = load_json(DECLARED)
    if policy.get("schema_version") != "aidefend.internal_discovery_policy.v1":
        raise ValueError("policy schema differs")
    if scope.get("schema_version") != "aidefend.internal_ai_target_manifest.v1":
        raise ValueError("target manifest schema differs")
    if declared_doc.get("schema_version") != "aidefend.declared_internal_ai_inventory.v1":
        raise ValueError("declared inventory schema differs")

    issued, expires = when(scope["issued_at"]), when(scope["expires_at"])
    max_age = int(policy.get("max_scope_age_seconds", 0))
    if max_age <= 0 or not issued <= now < expires:
        raise ValueError("target manifest freshness is invalid")
    if (expires - issued).total_seconds() > max_age:
        raise ValueError("target manifest exceeds policy freshness")
    if scope.get("policy_version") != policy.get("policy_version"):
        raise ValueError("policy version differs")
    if scope.get("policy_sha256") != digest(POLICY):
        raise ValueError("target manifest policy digest differs")

    adapters = policy.get("adapters")
    if not isinstance(adapters, dict) or not adapters:
        raise ValueError("adapter policy is empty")
    if any(not name or not IMAGE.fullmatch(str(image)) for name, image in adapters.items()):
        raise ValueError("adapter policy contains an unpinned image")

    sources = scope.get("authoritative_sources")
    if not isinstance(sources, list) or not sources:
        raise ValueError("authoritative source population is empty")
    source_ids = set()
    eligible_members: dict[tuple[str, str], dict] = {}
    for source in sources:
        required = {
            "source_id", "path", "signature_path", "sha256",
            "eligible_target_count", "eligible_member_ids_sha256",
        }
        if set(source) != required or source["source_id"] in source_ids:
            raise ValueError("authoritative source schema or identity differs")
        source_id = source["source_id"]
        source_ids.add(source_id)
        path = Path(source["path"])
        verify(path, Path(source["signature_path"]), Path("keys/asset-authority.pub"))
        if not SHA.fullmatch(source["sha256"]) or digest(path) != source["sha256"]:
            raise ValueError("authoritative source digest differs")
        source_doc = load_json(path)
        if (
            not isinstance(source_doc, dict)
            or set(source_doc) != {"schema_version", "source_id", "eligible_members"}
            or source_doc["schema_version"] != "aidefend.internal-discovery-source.v1"
            or source_doc["source_id"] != source_id
            or not isinstance(source_doc["eligible_members"], list)
        ):
            raise ValueError("authoritative source member schema differs")
        member_ids = []
        for member in source_doc["eligible_members"]:
            fields = {
                "member_id", "target_id", "raw_record_path",
                "raw_record_sha256",
                "authorization_id",
            }
            key = (source_id, member.get("member_id") if isinstance(member, dict) else None)
            if (
                not isinstance(member, dict) or set(member) != fields
                or any(not isinstance(member[field], str) or not member[field]
                       for field in fields)
                or not SHA.fullmatch(member["raw_record_sha256"])
                or digest(Path(member["raw_record_path"])) != member["raw_record_sha256"]
                or key in eligible_members
            ):
                raise ValueError("authoritative source member is invalid or duplicated")
            eligible_members[key] = member
            member_ids.append(member["member_id"])
        member_ids_sha256 = hashlib.sha256(json.dumps(
            sorted(member_ids), separators=(",", ":")
        ).encode("utf-8")).hexdigest()
        if (
            isinstance(source["eligible_target_count"], bool)
            or not isinstance(source["eligible_target_count"], int)
            or source["eligible_target_count"] != len(member_ids)
            or not SHA.fullmatch(source["eligible_member_ids_sha256"])
            or source["eligible_member_ids_sha256"] != member_ids_sha256
        ):
            raise ValueError("authoritative source member population differs")

    declared_items = declared_doc.get("items")
    if not isinstance(declared_items, list):
        raise ValueError("declared inventory items are missing")
    declared: dict[str, dict] = {}
    for item in declared_items:
        required = {
            "canonical_url", "resource_id", "owner",
            "environment", "approved",
        }
        if set(item) != required or not isinstance(item["approved"], bool):
            raise ValueError("declared inventory item schema differs")
        key = canonical_url(item["canonical_url"])
        if key in declared or not item["resource_id"] or not item["owner"]:
            raise ValueError("declared inventory contains an invalid duplicate")
        declared[key] = item

    targets = scope.get("targets")
    if not isinstance(targets, list):
        raise ValueError("target population is missing")
    target_by_id: dict[str, dict] = {}
    claimed_members: dict[tuple[str, str], str] = {}
    for target in targets:
        required = {
            "target_id", "segment_id", "adapter", "locator", "source_members",
        }
        if set(target) != required or not target["target_id"] or not target["segment_id"]:
            raise ValueError("target schema differs")
        if target["target_id"] in target_by_id or target["adapter"] not in adapters:
            raise ValueError("target identity or adapter differs")
        if (
            not isinstance(target["locator"], dict) or not target["locator"]
            or not isinstance(target["source_members"], list)
            or not target["source_members"]
        ):
            raise ValueError("target locator or provenance is empty")
        for provenance in target["source_members"]:
            fields = {
                "source_id", "member_id", "raw_record_sha256",
                "authorization_id",
            }
            if not isinstance(provenance, dict) or set(provenance) != fields:
                raise ValueError("target source-member provenance differs")
            key = (provenance["source_id"], provenance["member_id"])
            expected = eligible_members.get(key)
            if (
                expected is None or key in claimed_members
                or expected["target_id"] != target["target_id"]
                or expected["raw_record_sha256"] != provenance["raw_record_sha256"]
                or expected["authorization_id"] != provenance["authorization_id"]
            ):
                raise ValueError("target source-member binding differs")
            claimed_members[key] = target["target_id"]
        target_by_id[target["target_id"]] = target

    applicability = scope.get("applicability")
    if applicability == "NOT_APPLICABLE":
        if targets or declared or not str(scope.get("not_applicable_reason") or "").strip():
            raise ValueError("NOT_APPLICABLE target manifest is inconsistent")
        if any(int(source["eligible_target_count"]) != 0 for source in sources):
            raise ValueError("NOT_APPLICABLE source has eligible targets")
        outcome = "NOT_APPLICABLE"
    elif applicability != "APPLICABLE" or not targets:
        raise ValueError("APPLICABLE target population is empty")
    elif set(claimed_members) != set(eligible_members):
        outcome = "INSUFFICIENT_DATA"
        reasons.append("authoritative source-member union differs from target provenance")

    if outcome != "NOT_APPLICABLE":
        verify(RECEIPTS, RECEIPTS_SIG, Path("keys/internal-scanner.pub"))
        receipt_doc = load_json(RECEIPTS)
        if receipt_doc.get("schema_version") != "aidefend.internal_ai_scan_receipts.v1":
            raise ValueError("scanner receipt schema differs")
        if set(receipt_doc) != {
            "schema_version", "scope_id", "target_manifest_sha256",
            "scanner_identity", "receipts",
        }:
            raise ValueError("scanner receipt envelope schema differs")
        if receipt_doc["scope_id"] != scope["scope_id"]:
            raise ValueError("scanner receipt scope differs")
        if receipt_doc["target_manifest_sha256"] != digest(TARGETS):
            raise ValueError("scanner receipt target digest differs")
        if not str(receipt_doc["scanner_identity"]).strip():
            raise ValueError("scanner identity is empty")
        receipts = receipt_doc["receipts"]
        if not isinstance(receipts, list):
            raise ValueError("scanner receipt population is missing")
        receipt_ids = [item.get("target_id") for item in receipts]
        if len(receipt_ids) != len(set(receipt_ids)) or set(receipt_ids) != set(target_by_id):
            outcome = "INSUFFICIENT_DATA"
            reasons.append("scanner receipt target population differs")
        seen_records: set[tuple[str, str, str]] = set()
        observed_by_url: dict[str, dict] = {}
        for receipt in receipts:
            required = {
                "target_id", "adapter", "adapter_image_digest",
                "started_at", "finished_at", "status", "incomplete_reason",
                "raw_path", "raw_sha256", "observations_path",
                "observations_sha256", "observation_count",
            }
            if set(receipt) != required or receipt["target_id"] not in target_by_id:
                raise ValueError("target receipt schema differs")
            target = target_by_id[receipt["target_id"]]
            if receipt["adapter"] != target["adapter"]:
                raise ValueError("receipt adapter differs from target")
            if receipt["adapter_image_digest"] != adapters[receipt["adapter"]]:
                raise ValueError("receipt adapter image differs from policy")
            started, finished = when(receipt["started_at"]), when(receipt["finished_at"])
            if not issued <= started <= finished <= now:
                raise ValueError("receipt timestamp is outside scope")
            if receipt["status"] == "INCOMPLETE":
                if not str(receipt["incomplete_reason"] or "").strip():
                    raise ValueError("incomplete receipt lacks reason")
                outcome = "INSUFFICIENT_DATA"
                reasons.append("incomplete target:" + receipt["target_id"])
                continue
            if receipt["status"] != "COMPLETE" or receipt["incomplete_reason"] is not None:
                raise ValueError("receipt status is invalid")

            raw = Path(receipt["raw_path"])
            normalized = Path(receipt["observations_path"])
            if not SHA.fullmatch(receipt["raw_sha256"]) or digest(raw) != receipt["raw_sha256"]:
                raise ValueError("raw adapter evidence digest differs")
            if not SHA.fullmatch(receipt["observations_sha256"]) or digest(normalized, allow_empty=True) != receipt["observations_sha256"]:
                raise ValueError("normalized observation digest differs")
            rows = load_jsonl(normalized)
            if int(receipt["observation_count"]) != len(rows):
                raise ValueError("observation count differs")
            coverage.append({
                "target_id": receipt["target_id"],
                "adapter": receipt["adapter"],
                "adapter_image_digest": receipt["adapter_image_digest"],
                "raw_sha256": receipt["raw_sha256"],
                "observations_sha256": receipt["observations_sha256"],
                "observation_count": len(rows),
            })
            for row in rows:
                required_row = {
                    "target_id", "segment_id", "resource_id", "canonical_url",
                    "visibility", "status_code", "ai_evidence", "body_sha256",
                    "raw_record_sha256", "observed_at",
                }
                if set(row) != required_row:
                    raise ValueError("normalized observation schema differs")
                if row["target_id"] != receipt["target_id"]:
                    raise ValueError("observation target differs")
                if row["segment_id"] != target["segment_id"] or row["visibility"] != "private":
                    raise ValueError("observation scope differs")
                if not started <= when(row["observed_at"]) <= finished:
                    raise ValueError("observation timestamp differs")
                if not row["resource_id"] or not row["ai_evidence"]:
                    raise ValueError("observation lacks AI evidence")
                if not SHA.fullmatch(row["raw_record_sha256"]):
                    raise ValueError("raw record digest is invalid")
                key = canonical_url(row["canonical_url"])
                record_key = (row["target_id"], row["resource_id"], key)
                if record_key in seen_records:
                    raise ValueError("duplicate normalized observation")
                seen_records.add(record_key)
                observation = {
                    "target_id": row["target_id"],
                    "resource_id": row["resource_id"],
                    "canonical_url": key,
                    "ai_evidence": row["ai_evidence"],
                    "body_sha256": row["body_sha256"],
                    "raw_record_sha256": row["raw_record_sha256"],
                }
                observations.append(observation)
                prior = observed_by_url.get(key)
                if prior is not None and prior["resource_id"] != row["resource_id"]:
                    findings.append({
                        "type": "conflicting_resource_identity",
                        "canonical_url": key,
                    })
                else:
                    observed_by_url[key] = observation

        for key, row in observed_by_url.items():
            expected = declared.get(key)
            if expected is None:
                findings.append({
                    "type": "undeclared_internal_ai_endpoint",
                    "canonical_url": key,
                })
            elif (
                expected["resource_id"] != row["resource_id"]
                or not expected["approved"]
            ):
                findings.append({
                    "type": "inventory_identity_or_approval_drift",
                    "canonical_url": key,
                })
        for key in sorted(set(declared) - set(observed_by_url)):
            findings.append({
                "type": "declared_endpoint_not_observed",
                "canonical_url": key,
            })
        if outcome != "INSUFFICIENT_DATA":
            outcome = "FAIL" if findings else "PASS"
except (OSError, subprocess.SubprocessError, KeyError, TypeError, ValueError, json.JSONDecodeError) as error:
    outcome = "ERROR"
    reasons.append(str(error))

result = {
    "schema_version": "aidefend.internal_ai_discovery_reconciliation.v2",
    "control": "AID-M-001.006",
    "scope_id": scope.get("scope_id"),
    "target_manifest_sha256": digest(TARGETS) if TARGETS.is_file() else None,
    "declared_inventory_sha256": digest(DECLARED) if DECLARED.is_file() else None,
    "scanner_receipts_sha256": digest(RECEIPTS) if RECEIPTS.is_file() else None,
    "verifier_identity": verifier_identity,
    "eligible_target_count": len(scope.get("targets", [])),
    "measured_target_count": len(coverage),
    "coverage": coverage,
    "observations": observations,
    "findings": findings,
    "outcome": outcome,
    "reasons": reasons,
    "generated_at": now.isoformat(),
}
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
OUTPUT.write_text(
    json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n",
    encoding="utf-8",
)
if outcome not in {"PASS", "NOT_APPLICABLE"}:
    raise SystemExit(f"internal AI discovery did not pass: {outcome}")

python discovery/reconcile_internal_ai.py
cosign sign-blob --yes --key env://INTERNAL_DISCOVERY_VERIFIER_SIGNING_KEY --bundle evidence/internal-ai-discovery-reconciliation.sig evidence/internal-ai-discovery-reconciliation.json
cosign verify-blob --key keys/internal-discovery-verifier.pub --bundle evidence/internal-ai-discovery-reconciliation.sig evidence/internal-ai-discovery-reconciliation.json

Independent replay and evidence

A second scanner identity that cannot alter policy, authoritative exports, target manifest, declared inventory, workloads, or first-run receipts repeats every target with the same pinned adapter images. It signs a separate receipt envelope, reruns this verifier from the retained raw files, and compares the exact target set plus canonical observations. A difference requires a signed environment-change record; unexplained partial replay is INSUFFICIENT_DATA. The promotion gate accepts only fresh signed primary and replay receipts with matching scope and a reproduced PASS.

AI runtime fingerprint coverage

Compile custom and common runtime ports from authoritative cloud, IPAM, Kubernetes Services and workload exports into the signed target manifest. Use the versioned read-only profile in AID-M-001.005-G004 for Ollama, LiteLLM, and other installed runtimes, extending it for their actual versions and configured routes. Do not infer a complete internal population from 80/443 alone or use an empty fingerprint result as evidence that no runtime exists. Bind each target's port/route profile to its receipt and independent replay.

Open-source tools

• ProjectDiscovery httpx
• kubectl
• Steampipe
• Cartography (cloud, Kubernetes, and identity observed-state graph; declared-inventory reconciliation remains external)
• Cosign

Source-available / open-weight tools

• Nmap (Nmap Public Source License; source-available)

Commercial / hosted tools

• Microsoft Defender for Cloud
• Wiz Cloud Security Platform
• AWS Config
• Google Cloud Asset Inventory

Threat mappings

MITRE ATLAS

• AML.T0006 Active Scanning (authorized internal discovery identifies privately reachable AI services before an adversary with network access performs the same scan)
• AML.T0007 Discover AI Artifacts (private control-plane and endpoint reconciliation finds undeclared internal AI artifacts and services)

MAESTRO

• Orchestration Attacks (L4) (Kubernetes and cloud-control-plane reconciliation identifies undeclared AI workloads and administration surfaces)
• Integration Risks (L7) (internal MCP and A2A endpoints are agent integrations that require inventory and ownership)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• ML05:2023 Model Theft (finding undeclared internal model stores and inference endpoints exposes access paths that require restriction)

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.031 Model Extraction (undeclared internally reachable inference APIs create ungoverned extraction paths)

Cisco Integrated AI Security and Safety Framework

• N/A

Google Secure AI Framework 2.0 - Risks

• MXF: Model Exfiltration (private discovery finds ungoverned model stores)
• MRE: Model Reverse Engineering (private discovery finds ungoverned inference APIs that enable query-based model extraction)
• IIC: Insecure Integrated Component (internal MCP, A2A, model, and retrieval endpoints are integrated components requiring inventory)

Databricks AI Security Framework 3.0

• Model Serving - Inference requests 9.11: Model Inference API Access
• Agents - Tools MCP Server 13.20: Insecure Server Configuration (internal discovery identifies undeclared or weakly governed MCP servers)