Scope boundary

Owns authorized, credential-free, outside-in discovery and validation of organization-owned Internet-reachable AI endpoints and agent services, followed by ownership, exposure-intent, authentication, exception, and retirement reconciliation.

AID-M-001.004: AID-M-001.005 discovers the organization's inbound public AI attack surface; AID-M-001.004 discovers enterprise consumption of externally hosted AI and SaaS capabilities. An externally observed owned endpoint is not the same asset population as an external service used by employees or workloads.

AID-M-001.006: AID-M-001.005 observes public exposure from outside the trusted network; AID-M-001.006 observes private-network and internal-control-plane exposure from an authorized internal scope. Public and private discovery may share schemas, but their eligible populations, vantage points, authorization, and exposure claims remain separate.

Implementation guidance

• AID-M-001.005-G001: Run authorized outside-in discovery and AI-route validation from a network that does not inherit enterprise trust or credentials.
• AID-M-001.005-G002: Reconcile externally observed AI endpoints to declared inventory and fail the exposure gate on unknown, ownerless, expired, or authentication-drift findings.
• AID-M-001.005-G003: Discover organization-published agents on provider-owned domains through tenant inventory and authorized channel probes.
• AID-M-001.005-G004: Use a versioned runtime fingerprint profile covering nonstandard serving ports and reconcile external scan-service observations.

AID-M-001.005-G001

Concept:

An internal cloud inventory cannot prove what an unauthenticated Internet client can reach. Seed discovery only with root domains the organization owns or is explicitly authorized to test, then probe from an external runner with no corporate VPN, cookies, client certificates, or cloud credentials. Redirects must remain on the original host so a discovered endpoint cannot make the scanner follow traffic into an unrelated domain. Third-party SaaS consumption belongs to AID-M-001.004.

Step 1: Enumerate owned names and probe common AI routes

from __future__ import annotations

# File: exposure/write_scan_metadata.py

import argparse
import hashlib
import json
import re
from datetime import datetime
from pathlib import Path

DOMAIN = re.compile(r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def nonempty_lines(path: Path) -> list[str]:
    return [line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


parser = argparse.ArgumentParser()
parser.add_argument("--scope", required=True)
parser.add_argument("--hosts", required=True)
parser.add_argument("--probes", required=True)
parser.add_argument("--started-at", required=True)
parser.add_argument("--finished-at", required=True)
parser.add_argument("--amass-version", required=True)
parser.add_argument("--httpx-version", required=True)
parser.add_argument("--output", required=True)
args = parser.parse_args()

scope_path = Path(args.scope)
hosts_path = Path(args.hosts)
probes_path = Path(args.probes)
roots = sorted(set(value.lower().rstrip(".") for value in nonempty_lines(scope_path)))
if not roots or any(not DOMAIN.fullmatch(root) for root in roots):
    raise SystemExit("authorized scope is empty or contains an invalid root domain")
started = datetime.fromisoformat(args.started_at.replace("Z", "+00:00"))
finished = datetime.fromisoformat(args.finished_at.replace("Z", "+00:00"))
if started.tzinfo is None or finished.tzinfo is None or finished < started:
    raise SystemExit("scan timestamps are invalid")
host_count = len(set(nonempty_lines(hosts_path)))
probe_count = len(nonempty_lines(probes_path))
if host_count == 0 or probe_count == 0:
    raise SystemExit("not_measurable: scanner produced no hosts or probe records")
metadata = {
    "schema_version": "aidefend.public_ai_scan.v1",
    "completed": True,
    "started_at": started.isoformat(),
    "finished_at": finished.isoformat(),
    "authorized_roots": roots,
    "authorized_scope_sha256": sha256_file(scope_path),
    "scanners": {"amass": args.amass_version, "httpx": args.httpx_version},
    "discovered_hosts_sha256": sha256_file(hosts_path),
    "probe_jsonl_sha256": sha256_file(probes_path),
    "discovered_host_count": host_count,
    "probe_record_count": probe_count,
}
Path(args.output).write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8")

set -euo pipefail
mkdir -p out artifacts
sort -u scope/owned-root-domains.txt > out/authorized-root-domains.txt
test -s out/authorized-root-domains.txt
STARTED_AT=$(date -u +%Y-%m-%dT%H:%M:%SZ)
AMASS_VERSION=$(amass -version 2>&1 | head -n 1)
HTTPX_VERSION=$(httpx -version 2>&1 | head -n 1)
amass enum -passive -df out/authorized-root-domains.txt -o out/amass-hosts.txt
{ cat out/authorized-root-domains.txt; cat out/amass-hosts.txt; } | sort -u > out/discovered-hosts.txt
httpx -l out/discovered-hosts.txt \
  -path /v1/models,/v1/chat/completions,/openapi.json,/docs,/mcp,/.well-known/agent.json,/.well-known/ai-plugin.json,/gradio_api/info \
  -json -status-code -title -content-length -hash sha256 -probe \
  -follow-host-redirects -rate-limit 50 \
  -o out/public-ai-route-probes.jsonl
FINISHED_AT=$(date -u +%Y-%m-%dT%H:%M:%SZ)
python exposure/write_scan_metadata.py \
  --scope out/authorized-root-domains.txt \
  --hosts out/discovered-hosts.txt \
  --probes out/public-ai-route-probes.jsonl \
  --started-at "$STARTED_AT" --finished-at "$FINISHED_AT" \
  --amass-version "$AMASS_VERSION" --httpx-version "$HTTPX_VERSION" \
  --output artifacts/public-ai-exposure-scan-metadata.json

Step 2: Preserve completeness evidence

The metadata binds the authorized scope, exact scanner versions, start and finish times, discovered hosts, and raw probe JSONL. A run with no probe records is not measurable. The reconciliation gate below independently re-hashes every input, rejects stale or out-of-scope data, and classifies each response before creating a finding. A generic 404 on an AI-looking path proves only that the host answered; it is an observation, not proof that an undeclared AI endpoint exists.

Step 3: Independently reproduce high-risk findings

Repeat every high-risk request from a second credential-free runner or commercial EASM export. A 401 or 403 proves an authentication boundary answered; other statuses do not prove authentication. Run at least daily and after DNS, ingress, model-serving, MCP, A2A, notebook, or demo-environment changes.

Action: Retain the authorized scope, raw JSONL, scan metadata, reconciliation result, and independent recheck for at least the exposure-remediation SLA period.

AID-M-001.005-G002

Concept:

The security outcome is fresh proof that every responding AI route is in authorized scope, has one unambiguous declaration, and matches its approved exposure state. A scan is not trusted merely because a JSONL file exists: verify its signed-input metadata, time window, scope digest, scanner versions, and raw-file digests first.

Step 1: Maintain the declared endpoint register

# File: inventory/declared_public_ai_endpoints.csv
canonical_url,owner,environment,public_intent,auth_required,retire_after,response_sha256
https://api.example.com/v1/models,ai-platform,production,true,true,,
https://demo.example.com/gradio_api/info,ai-research,test,true,true,2026-08-01T00:00:00Z,

response_sha256 is optional per row. When present, it must be the approved SHA-256 of the unauthenticated response body emitted by the external scanner and is enforced as drift evidence.

Step 2: Produce deterministic reconciliation evidence

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
MAX_SCAN_AGE_SECONDS = _aidefend_positive_int("M001005_MAX_SCAN_AGE_SECONDS")
MAX_SCAN_DURATION_SECONDS = _aidefend_positive_int("M001005_MAX_SCAN_DURATION_SECONDS")
MAX_FUTURE_SKEW_SECONDS = _aidefend_nonnegative_int("M001005_MAX_FUTURE_SKEW_SECONDS")

# File: exposure/reconcile_public_ai_endpoints.py

import csv
import hashlib
import json
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

OBSERVED = Path("out/public-ai-route-probes.jsonl")
DISCOVERED = Path("out/discovered-hosts.txt")
SCOPE = Path("out/authorized-root-domains.txt")
SCAN_METADATA = Path("artifacts/public-ai-exposure-scan-metadata.json")
DECLARED = Path("inventory/declared_public_ai_endpoints.csv")
EXCEPTIONS = Path("inventory/public_ai_exposure_exceptions.csv")
MAX_SCAN_AGE = timedelta(seconds=MAX_SCAN_AGE_SECONDS)
MAX_SCAN_DURATION = timedelta(seconds=MAX_SCAN_DURATION_SECONDS)
MAX_FUTURE_SKEW = timedelta(seconds=MAX_FUTURE_SKEW_SECONDS)
ALLOWED_ENVIRONMENTS = {"production", "staging", "test", "development", "demo"}
REQUIRED_COLUMNS = {
    "canonical_url", "owner", "environment", "public_intent",
    "auth_required", "retire_after", "response_sha256",
}
SHA256 = re.compile(r"^(?:sha256:)?([0-9a-f]{64})$")
AI_SPECIFIC_PATHS = {
    "/v1/models",
    "/v1/chat/completions",
    "/v1/completions",
    "/mcp",
    "/.well-known/agent.json",
    "/.well-known/ai-plugin.json",
    "/gradio_api/info",
}
GENERIC_AI_HINTS = {
    "agent", "ai", "anthropic", "bedrock", "chat", "completion", "gradio",
    "inference", "llm", "mcp", "model", "openai", "prompt", "vertex",
}
HARD_FAIL_SEVERITIES = {"critical", "high"}
EXCEPTION_COLUMNS = {"canonical_url", "finding_type", "expires_at", "reason"}


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def nonempty_lines(path: Path) -> list[str]:
    return [line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


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


def normalize_root(value: str) -> str:
    return value.strip().lower().rstrip(".")


def host_in_scope(host: str, roots: set[str]) -> bool:
    normalized = normalize_root(host)
    return any(normalized == root or normalized.endswith("." + root) for root in roots)


def canonical_url(value: str) -> str:
    parsed = urlsplit(value.strip())
    if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname:
        raise ValueError("endpoint URL must use http/https and include a host")
    if parsed.username or parsed.password:
        raise ValueError("endpoint URL must not contain credentials")
    port = parsed.port
    host = parsed.hostname.lower().rstrip(".")
    netloc = host if port is None or (parsed.scheme.lower(), port) in {("http", 80), ("https", 443)} else f"{host}:{port}"
    path = parsed.path.rstrip("/") or "/"
    return urlunsplit((parsed.scheme.lower(), netloc, path, "", ""))


def parse_bool(value: str, field: str) -> bool:
    normalized = value.strip().lower()
    if normalized not in {"true", "false"}:
        raise ValueError(f"{field} must be true or false")
    return normalized == "true"


def normalize_sha256(value: str) -> str | None:
    if not value.strip():
        return None
    match = SHA256.fullmatch(value.strip().lower())
    if not match:
        raise ValueError("response_sha256 must be a SHA-256 digest")
    return match.group(1)


def response_sha256(row: dict) -> str | None:
    hashes = row.get("hash")
    candidates = []
    if isinstance(hashes, dict):
        candidates.extend([hashes.get("body_sha256"), hashes.get("sha256")])
    candidates.extend([row.get("body_sha256"), row.get("sha256")])
    for value in candidates:
        if isinstance(value, str):
            match = SHA256.fullmatch(value.lower())
            if match:
                return match.group(1)
    return None


def route_path(url: str) -> str:
    path = urlsplit(url).path.rstrip("/") or "/"
    return path.lower()


def row_text(row: dict) -> str:
    values = [str(row.get(key, "")) for key in ("title", "webserver", "tech", "body_preview", "url", "input")]
    return " ".join(values).lower()


def classify_ai_candidate(row: dict, status: int, response_url: str) -> tuple[bool, list[str]]:
    path = route_path(response_url)
    if status in {404, 410}:
        return False, ["negative_http_response"]
    evidence = []
    if path in AI_SPECIFIC_PATHS:
        evidence.append(f"ai_specific_path:{path}")
    text = row_text(row)
    matched_hints = sorted(hint for hint in GENERIC_AI_HINTS if hint in text)
    if matched_hints and path in {"/", "/docs", "/openapi.json"}:
        evidence.append("ai_text_hint:" + ",".join(matched_hints[:5]))
    return bool(evidence), evidence


def severity_for_unknown_candidate(status: int) -> str:
    if 200 <= status <= 299:
        return "high"
    if status in {301, 302, 307, 308, 401, 403, 405, 429}:
        return "medium"
    if 500 <= status <= 599:
        return "medium"
    return "low"


def load_exceptions(now: datetime) -> list[dict]:
    if not EXCEPTIONS.is_file():
        return []
    with EXCEPTIONS.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        if set(reader.fieldnames or []) != EXCEPTION_COLUMNS:
            raise ValueError("exception CSV columns do not match required schema")
        exceptions = []
        for row in reader:
            expires_at = parse_time(row["expires_at"])
            if expires_at <= now:
                raise ValueError("expired exposure exception remains in inventory: " + row["canonical_url"])
            exceptions.append({
                "canonical_url": canonical_url(row["canonical_url"]),
                "finding_type": row["finding_type"].strip(),
                "expires_at": expires_at.isoformat(),
                "reason": row["reason"].strip(),
            })
        return exceptions


def exception_applies(finding: dict, exceptions: list[dict]) -> dict | None:
    for exception in exceptions:
        if exception["canonical_url"] == finding.get("url") and exception["finding_type"] == finding.get("type"):
            return exception
    return None


def load_and_verify_scan(now: datetime) -> tuple[dict, set[str], list[dict]]:
    metadata = json.loads(SCAN_METADATA.read_text(encoding="utf-8"))
    if metadata.get("schema_version") != "aidefend.public_ai_scan.v1" or metadata.get("completed") is not True:
        raise ValueError("scan metadata schema/completion invalid")
    roots = {normalize_root(value) for value in nonempty_lines(SCOPE)}
    if not roots or set(metadata.get("authorized_roots", [])) != roots:
        raise ValueError("scan metadata scope differs from authorized scope")
    if metadata.get("authorized_scope_sha256") != sha256_file(SCOPE):
        raise ValueError("authorized scope digest mismatch")
    if metadata.get("discovered_hosts_sha256") != sha256_file(DISCOVERED):
        raise ValueError("discovered-host digest mismatch")
    if metadata.get("probe_jsonl_sha256") != sha256_file(OBSERVED):
        raise ValueError("probe JSONL digest mismatch")
    discovered_hosts = set(nonempty_lines(DISCOVERED))
    if len(discovered_hosts) != metadata.get("discovered_host_count"):
        raise ValueError("discovered-host count mismatch")
    for host in discovered_hosts:
        if not host_in_scope(host, roots):
            raise ValueError("discovered host escaped authorized scope: " + host)
    raw_lines = nonempty_lines(OBSERVED)
    if len(raw_lines) != metadata.get("probe_record_count") or not raw_lines:
        raise ValueError("probe count is empty or differs from metadata")
    scanners = metadata.get("scanners", {})
    if any(not isinstance(scanners.get(name), str) or not scanners[name].strip() for name in ("amass", "httpx")):
        raise ValueError("scanner version evidence missing")
    started = parse_time(metadata["started_at"])
    finished = parse_time(metadata["finished_at"])
    if started > finished or finished - started > MAX_SCAN_DURATION:
        raise ValueError("scan duration is invalid")
    if finished > now + MAX_FUTURE_SKEW or now - finished > MAX_SCAN_AGE:
        raise ValueError("scan metadata is stale or from the future")
    return metadata, roots, [json.loads(line) for line in raw_lines]


def load_declarations() -> dict[str, dict]:
    with DECLARED.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        if set(reader.fieldnames or []) != REQUIRED_COLUMNS:
            raise ValueError("declared endpoint CSV columns do not match required schema")
        rows = list(reader)
    declared: dict[str, dict] = {}
    for row in rows:
        url = canonical_url(row["canonical_url"])
        if url in declared:
            raise ValueError("duplicate canonical endpoint declaration: " + url)
        owner = row["owner"].strip()
        environment = row["environment"].strip().lower()
        if not owner:
            raise ValueError("endpoint owner is required: " + url)
        if environment not in ALLOWED_ENVIRONMENTS:
            raise ValueError("invalid endpoint environment: " + url)
        retire_after = parse_time(row["retire_after"]) if row["retire_after"].strip() else None
        declared[url] = {
            **row,
            "owner": owner,
            "environment": environment,
            "public_intent": parse_bool(row["public_intent"], "public_intent"),
            "auth_required": parse_bool(row["auth_required"], "auth_required"),
            "retire_after_parsed": retire_after,
            "response_sha256_normalized": normalize_sha256(row["response_sha256"]),
        }
    return declared


now = datetime.now(timezone.utc)
metadata, roots, observed_rows = load_and_verify_scan(now)
declared = load_declarations()
exceptions = load_exceptions(now)
findings: list[dict] = []
observations: list[dict] = []
observed_urls: set[str] = set()
unanswered = 0

for row in observed_rows:
    try:
        status = int(row["status_code"])
    except (KeyError, TypeError, ValueError):
        unanswered += 1
        findings.append({"type": "probe_without_http_response", "input": str(row.get("input") or row.get("url") or "")})
        continue
    if not 100 <= status <= 599:
        findings.append({"type": "invalid_http_status", "status_code": status})
        continue

    response_url = canonical_url(str(row.get("url") or row.get("input") or ""))
    candidate_urls = [response_url]
    for key in ("input", "final_url", "location"):
        value = row.get(key)
        if isinstance(value, str) and "://" in value:
            candidate_urls.append(canonical_url(value))
    hosts = {urlsplit(value).hostname for value in candidate_urls}
    if any(host is None or not host_in_scope(host, roots) for host in hosts):
        findings.append({"type": "out_of_scope_probe_or_redirect", "urls": candidate_urls})
        continue
    if len(hosts) != 1:
        findings.append({"type": "cross_host_redirect", "urls": candidate_urls})
        continue
    if response_url in observed_urls:
        findings.append({"type": "duplicate_probe_record", "url": response_url, "status_code": status})
        continue
    observed_urls.add(response_url)

    record = declared.get(response_url)
    if record is None:
        probable_ai, evidence = classify_ai_candidate(row, status, response_url)
        if not probable_ai:
            observations.append({
                "type": "non_ai_route_response",
                "url": response_url,
                "status_code": status,
            })
            continue
        finding = {
            "type": "undeclared_public_ai_endpoint",
            "severity": severity_for_unknown_candidate(status),
            "url": response_url,
            "status_code": status,
            "evidence": evidence,
        }
        exception = exception_applies(finding, exceptions)
        if exception:
            observations.append({"type": "suppressed_finding", "finding": finding, "exception": exception})
        else:
            findings.append(finding)
        continue
    if record["retire_after_parsed"] and now >= record["retire_after_parsed"]:
        findings.append({"type": "expired_endpoint_still_exposed", "url": response_url, "status_code": status})
    if not record["public_intent"]:
        findings.append({"type": "exposure_not_approved", "url": response_url, "status_code": status})
    if record["auth_required"] and status not in {401, 403}:
        findings.append({"type": "authentication_boundary_not_proven", "url": response_url, "status_code": status})
    expected_fingerprint = record["response_sha256_normalized"]
    if expected_fingerprint is not None:
        observed_fingerprint = response_sha256(row)
        if observed_fingerprint is None:
            findings.append({"type": "response_fingerprint_unmeasured", "url": response_url})
        elif observed_fingerprint != expected_fingerprint:
            findings.append({
                "type": "response_fingerprint_drift",
                "url": response_url,
                "expected": expected_fingerprint,
                "observed": observed_fingerprint,
            })

for url, record in declared.items():
    expired = record["retire_after_parsed"] and now >= record["retire_after_parsed"]
    if record["public_intent"] and not expired and url not in observed_urls:
        findings.append({"type": "declared_public_endpoint_not_measured", "severity": "high", "url": url})

severity_counts = {
    severity: sum(1 for finding in findings if finding.get("severity", "high") == severity)
    for severity in ("critical", "high", "medium", "low")
}
hard_fail_count = sum(severity_counts[severity] for severity in HARD_FAIL_SEVERITIES)

artifact = {
    "schema_version": "aidefend.public_ai_exposure.v2",
    "generated_at": now.isoformat(),
    "expires_at": (now + MAX_SCAN_AGE).isoformat(),
    "scan_finished_at": metadata["finished_at"],
    "scan_metadata_sha256": sha256_file(SCAN_METADATA),
    "authorized_scope_sha256": sha256_file(SCOPE),
    "observed_sha256": sha256_file(OBSERVED),
    "declared_inventory_sha256": sha256_file(DECLARED),
    "exception_inventory_sha256": sha256_file(EXCEPTIONS) if EXCEPTIONS.is_file() else None,
    "measured_route_count": len(observed_urls),
    "unanswered_probe_count": unanswered,
    "finding_count": len(findings),
    "observation_count": len(observations),
    "severity_counts": severity_counts,
    "findings": findings,
    "observations": observations,
    "status": "fail" if hard_fail_count else "review" if findings else "pass",
}
Path("artifacts").mkdir(exist_ok=True)
output = Path("artifacts/public-ai-exposure-reconciliation.json")
output.write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n", encoding="utf-8")
if hard_fail_count:
    raise SystemExit(f"public AI exposure reconciliation failed with {hard_fail_count} high/critical finding(s)")
if findings:
    print(f"public AI exposure reconciliation requires review: {severity_counts}")

Step 3: Sign and independently verify the evidence

set -euo pipefail
python exposure/reconcile_public_ai_endpoints.py
cosign sign-blob --yes --key keys/exposure-evidence.key \
  --bundle artifacts/public-ai-exposure-reconciliation.sig \
  artifacts/public-ai-exposure-reconciliation.json
cosign verify-blob --key keys/exposure-evidence.pub \
  --bundle artifacts/public-ai-exposure-reconciliation.sig \
  artifacts/public-ai-exposure-reconciliation.json

Action: Route every finding to the endpoint owner and applicable authentication, ingress, patching, or eviction control. Closure requires a fresh in-scope rescan and a newly signed reconciliation result; a closed ticket or stale scanner file is not evidence.

AID-M-001.005-G003

Step 1: Include hosted publishing scope

The root-domain scanner does not cover agents hosted on shared SaaS domains. Build a separate signed scope of authorized tenant/environment IDs and provider resource IDs from administrative exports, then join their published channels to the approved exposure register. This is discovery of agents the organization publishes; AID-M-001.004 still owns consumption of external AI services. Do not enumerate unrelated tenants or fuzz a provider's shared namespace without permission.

Step 2: Collect one supported platform inventory

For Microsoft Copilot Studio, use the Power Platform inventory API with a delegated authorized administrator token. The example queries all agent rows and refuses incomplete pagination. It does not fabricate support for a managed identity or app-only token. Validate policy limits at startup, bind the token's tenant to the authorized scope, and retain request scope, timestamps, API/schema version, page digests, and the complete result digest as collection evidence. Do not log the token.

# hosted_inventory.py -- requests, delegated Power Platform administrator token
import json, requests

def collect_agents(token: str, policy: dict) -> list[dict]:
    endpoint = "https://api.powerplatform.com/resourcequery/resources/query?api-version=2024-10-01"
    query = {"TableName": "PowerPlatformResources", "Clauses": [
        {"$type": "where", "FieldName": "type", "Operator": "==",
         "Values": ["'microsoft.copilotstudio/agents'"]}],
        "Options": {"Top": policy["page_size"]}}
    rows, tokens, total = {}, set(), None
    with requests.Session() as session:
        session.trust_env = False
        for _ in range(policy["max_pages"]):
            with session.post(endpoint, headers={"Authorization": "Bearer " + token},
                              json=query, timeout=policy["request_timeout_seconds"],
                              allow_redirects=False, stream=True) as response:
                if response.status_code != 200:
                    raise RuntimeError("inventory_http_" + str(response.status_code))
                chunks, size = [], 0
                for chunk in response.iter_content(65536):
                    size += len(chunk)
                    if size > policy["max_page_bytes"]:
                        raise RuntimeError("inventory_page_limit")
                    chunks.append(chunk)
                page = json.loads(b"".join(chunks))
            if not isinstance(page.get("data"), list) or page.get("count") != len(page["data"]):
                raise RuntimeError("inventory_response_shape")
            if total is None: total = page["totalRecords"]
            if type(total) is not int or total < 0 or page["totalRecords"] != total:
                raise RuntimeError("inventory_population_changed")
            for row in page["data"]:
                if not isinstance(row, dict) or not isinstance(row.get("properties"), dict):
                    raise RuntimeError("resource_shape")
                tenant, environment, name = row.get("tenantId"), row["properties"].get("environmentId"), row.get("name")
                if (tenant != policy["tenant_id"] or row.get("type") != "microsoft.copilotstudio/agents"
                        or not isinstance(environment, str) or not environment
                        or not isinstance(name, str) or not name):
                    raise RuntimeError("resource_scope_or_identity")
                rid = (tenant, environment, name)
                if rid in rows:
                    raise RuntimeError("duplicate_resource_identity")
                rows[rid] = row
            token_next = page.get("skipToken")
            if token_next:
                if not isinstance(token_next, str) or token_next in tokens:
                    raise RuntimeError("invalid_or_repeated_skip_token")
                tokens.add(token_next)
                query["Options"]["SkipToken"] = token_next
                continue
            if page.get("resultTruncated") not in (False, 0, "false") or len(rows) != total:
                raise RuntimeError("incomplete_inventory")
            return list(rows.values())
    raise RuntimeError("inventory_pagination_limit")

Inventory metadata is not a complete exposure verdict. The agent inventory schema has coverage and refresh limitations, including classic bots, nullable fields, and truncated per-agent capability lists. Cross-check environment coverage and published versions through supported admin readback. A missing field remains unknown. Reconcile connector counts against returned operations; do not assume a displayed subset is the whole capability set. If the API is unavailable in the tenant, use an administrator export plus independent reconciliation, preserving the unavailable coverage as INSUFFICIENT_DATA until resolved.

Step 3: Verify actual exposure

Derive exact authorized endpoints from the published agent/channel configuration. Probe only those endpoints from a clean external client, including demo/custom website paths, and compare observed audiences with the signed public-agent exception list. Collect independent no-credential and unapproved-user readbacks, recording actual agent responses rather than accepting a generic website 200 as proof. Pass findings and the full tenant/resource/version population to AID-H-004.006. Publication restrictions and incident isolation are owned there and in Isolate, not by this inventory method.

AID-M-001.005-G004

Step 1: Expand the authorized port and route population

The baseline 80/443 web scan is insufficient for self-hosted AI runtimes. Add known ports and read-only paths for the deployment's actual products and versions. Candidate profiles include Ollama on 11434 with /api/tags and /api/version, LiteLLM on 4000 with /v1/models, and the configured management paths for Langflow, n8n, vLLM, or agent runtimes. Product defaults are discovery seeds, not a complete inventory: custom ports come from DNS, cloud/IPAM, Kubernetes Services, load balancers, and the exposure register. Never issue model generation, tool execution, /api/pull, or other state-changing probes just to identify a service.

Step 2: Run bounded read-only probes

Use the same authorized, validated host list and evidence recorder as AID-M-001.005-G001. This illustrative ProjectDiscovery httpx invocation covers a small approved profile; set rate, timeout, ports, paths, and output limits from the signed scan policy and pin the tool version. Keep the profile JSON/digest with the raw output and record unsuccessful attempts, not only matches. Verify the deployed httpx version's CLI flags before promotion.

# Values below are a small test profile, not universal production defaults.
httpx -l out/discovered-hosts.txt \
  -ports http:80,4000,5678,7860,11434,https:443 \
  -path /api/tags,/api/version,/v1/models,/health,/healthz \
  -sc -title -server -probe -hash sha256 -json -rl 5 -timeout 5 -retries 0 \
  -rstr 1048576 -rsts 1048576 -duc \
  -o out/runtime-profile-probes.jsonl

Do not enable cross-host redirects or run unknown vulnerability templates. Use signatures over response shape and headers to identify candidates; a port, path, status, or page title alone does not prove an exposed model or authentication bypass. Reproduce suspected unauthorized access using synthetic test resources and the service owner's approved test plan. An HTTP authentication error from one path does not establish protection of every path.

Step 3: Reconcile independent discovery sources

Query Shodan/Censys for the organization's authorized IP ranges and hostnames, then compare timestamped observations with current direct probes and declared inventory. Third-party scan data can be stale or incomplete; it never replaces live verification or authorizes scanning a newly discovered third-party host. Require the union of approved ports/endpoints to be attempted, and report missing attempts or unresolvable scope as insufficient coverage. Reuse the same versioned profile with AID-M-001.006 for private targets.

Open-source tools

• OWASP Amass
• ProjectDiscovery httpx
• ProjectDiscovery Nuclei
• Cosign

Commercial / hosted tools

• Shodan Monitor (outside-in exposed-service discovery)
• Censys Attack Surface Management
• Microsoft Defender External Attack Surface Management
• Palo Alto Networks Cortex Xpanse
• CrowdStrike Falcon Surface

Threat mappings

MITRE ATLAS

• AML.T0006 Active Scanning (outside-in discovery finds exposed AI services before adversaries operationalize the same scan)
• AML.T0007 Discover AI Artifacts (reconciliation identifies publicly discoverable AI artifacts and endpoints)
• AML.T0040 AI Model Inference API Access (validation finds unintended or unauthenticated inference API exposure)
• AML.T0049 Exploit Public-Facing Application (exposure validation identifies public AI applications requiring remediation)
• AML.T0116 Autonomous Reconnaissance (outside-in discovery closes the unknown exposed surface an autonomous agent would enumerate)
• AML.T0000 Search Open Technical Databases (reconciles public scan-service observations to identify unapproved AI exposure)
• AML.T0000.003 Search Open Technical Databases: Scan Databases (reconciles third-party scan records with live probes and the approved AI exposure register)
• AML.T0006.000 Active Scanning: Enumerate Hosted AI Resources (tenant inventory and authorized channel probes identify unintentionally reachable hosted agents)
• AML.T0006.002 Active Scanning: Scan for Exposed AI Infrastructure (authorized port and route fingerprints identify exposed AI runtimes)
• AML.T0132 Misconfigured or Publicly Exposed AI Services (independent access probes and inventory reconciliation surface unapproved public AI services)

MAESTRO

• Orchestration Attacks (L4) (outside-in validation identifies exposed AI orchestration and administration surfaces)
• Integration Risks (L7) (public MCP and A2A endpoints are externally reachable agent integrations)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• ML05:2023 Model Theft (finding unintended public model and inference access reduces direct theft paths)

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.031 Model Extraction (unintended public inference access enables extraction)

Cisco Integrated AI Security and Safety Framework

• AISubtech-14.1.2 Insufficient Access Controls (External probes identify routes whose required authentication is not proven)

Google Secure AI Framework 2.0 - Risks

• MXF: Model Exfiltration (outside-in discovery finds exposed model artifacts)
• MRE: Model Reverse Engineering (outside-in discovery finds exposed inference APIs that enable query-based model extraction)
• IIC: Insecure Integrated Component (public MCP and agent endpoints are externally reachable integrated components)

Databricks AI Security Framework 3.0

• Model Serving - Inference requests 9.11: Model Inference API Access
• Platform 12.7: Initial Access (unintended Internet exposure creates an initial-access path)
• Agents - Tools MCP Server 13.20: Insecure Server Configuration (external validation identifies publicly exposed or weakly protected MCP servers)