Scope boundary

Owns enterprise-wide discovery and inventory reconciliation for externally hosted AI services, embedded SaaS AI features, browser AI applications, provider APIs, and hosted agent endpoints observed through identity, endpoint, browser, SaaS, proxy, DNS, gateway, and provider telemetry.

AID-M-001.005: AID-M-001.004 discovers enterprise users and workloads consuming third-party or externally hosted AI services; AID-M-001.005 discovers the organization's own Internet-reachable AI services from an outside vantage point. Outbound shadow-AI use and inbound public attack-surface exposure are different populations and must not be combined.

AID-M-001.006: AID-M-001.004 discovers externally hosted AI consumption across enterprise telemetry; AID-M-001.006 discovers organization-owned AI endpoints reachable only through private networks or control planes. The first follows external service use, while the second enumerates internal service exposure under an explicitly authorized private scope.

Implementation guidance

• AID-M-001.004-G001: Ingest AI-service discovery telemetry from identity, network, endpoint, browser, SaaS, and API-gateway logs into a normalized AI service observation schema.
• AID-M-001.004-G002: Classify observed AI services against a governed AI service catalog and assign sanctioned, restricted, unknown, or blocked status with technical evidence.
• AID-M-001.004-G003: Reconcile discovered AI services with declared AI asset and dependency inventories, then produce drift evidence for downstream controls.

AID-M-001.004-G001

Purpose and scope

Normalize the exact discovery-source population selected for this tenant and release. Source eligibility and AI-service candidate signals come from a reviewed, current discovery profile—not a hard-coded vendor-domain list. Every source member produces a coverage row even when it yields zero candidates; failed, stale, missing, or partial members are INSUFFICIENT_DATA.

Governed discovery profile

{
  "schema_version": "aidefend.ai-service-discovery-profile.v1",
  "profile_version": "discovery-42",
  "tenant_id": "acme",
  "release_scope": "enterprise-egress-2026-08-28",
  "issued_at": "2026-08-28T00:00:00Z",
  "expires_at": "2026-08-29T00:00:00Z",
  "members": [
    {
      "member_id": "swg-prod",
      "source_type": "secure_web_gateway",
      "raw_path": "in/swg-prod.csv",
      "raw_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
      "host_suffixes": ["api.openai.com", "claude.ai"],
      "accepted_signals": ["oauth_ai_app", "model_api_route"],
      "attempt_status": "COMPLETED"
    },
    {
      "member_id": "model-gateway-prod",
      "source_type": "model_gateway",
      "raw_path": "in/model-gateway-prod.csv",
      "raw_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
      "host_suffixes": ["models.acme.example"],
      "accepted_signals": ["governed_model_gateway"],
      "attempt_status": "COMPLETED"
    }
  ]
}

Normalize stable source bytes and prove coverage

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit

SHA = set("0123456789abcdef")
PROFILE_FIELDS = {
    "schema_version", "profile_version", "tenant_id", "release_scope",
    "issued_at", "expires_at", "members",
}
MEMBER_FIELDS = {
    "member_id", "source_type", "raw_path", "raw_sha256",
    "host_suffixes", "accepted_signals", "attempt_status",
}


def strict_json(path: Path) -> dict:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key: {key}")
            value[key] = item
        return value
    value = json.loads(
        path.read_text(encoding="utf-8"),
        object_pairs_hook=unique,
        parse_constant=lambda token: (_ for _ in ()).throw(ValueError(token)),
    )
    if not isinstance(value, dict):
        raise ValueError("profile must be an object")
    return value


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 host_matches(host: str, suffix: str) -> bool:
    host = host.lower().rstrip(".")
    suffix = suffix.lower().rstrip(".")
    return host == suffix or host.endswith("." + suffix)


parser = argparse.ArgumentParser()
parser.add_argument("--profile", type=Path, required=True)
parser.add_argument("--expected-tenant", required=True)
parser.add_argument("--expected-release-scope", required=True)
parser.add_argument("--events", type=Path, required=True)
parser.add_argument("--coverage", type=Path, required=True)
args = parser.parse_args()

profile = strict_json(args.profile)
if set(profile) != PROFILE_FIELDS or (
    profile["schema_version"] != "aidefend.ai-service-discovery-profile.v1"
    or profile["tenant_id"] != args.expected_tenant
    or profile["release_scope"] != args.expected_release_scope
    or not isinstance(profile["profile_version"], str)
    or not profile["profile_version"]
    or not when(profile["issued_at"]) <= datetime.now(timezone.utc)
       < when(profile["expires_at"])
    or not isinstance(profile["members"], list)
    or not profile["members"]
):
    raise SystemExit("discovery profile identity, scope, or currentness differs")

member_ids: set[str] = set()
events: list[dict] = []
coverage: list[dict] = []
for member in profile["members"]:
    if not isinstance(member, dict) or set(member) != MEMBER_FIELDS:
        raise SystemExit("discovery member schema differs")
    member_id = member["member_id"]
    if (
        not isinstance(member_id, str) or not member_id
        or member_id in member_ids
        or member["attempt_status"] != "COMPLETED"
        or not isinstance(member["host_suffixes"], list)
        or not isinstance(member["accepted_signals"], list)
        or len(member["host_suffixes"]) != len(set(member["host_suffixes"]))
        or len(member["accepted_signals"]) != len(set(member["accepted_signals"]))
    ):
        raise SystemExit("discovery member identity or completed attempt differs")
    member_ids.add(member_id)
    raw_path = Path(member["raw_path"])
    raw = raw_path.read_bytes()
    raw_sha256 = hashlib.sha256(raw).hexdigest()
    if (
        raw_path.is_symlink() or not raw
        or len(member["raw_sha256"]) != 64
        or set(member["raw_sha256"]) - SHA
        or raw_sha256 != member["raw_sha256"]
    ):
        raise SystemExit("discovery source bytes differ")
    reader = csv.DictReader(io.StringIO(raw.decode("utf-8", errors="strict")))
    headers = reader.fieldnames or []
    required = {
        "timestamp", "url", "host", "principal_id", "trace_id", "ai_signal",
        "service_tenant", "data_signals",
    }
    if len(headers) != len(set(headers)) or not required.issubset(headers):
        raise SystemExit("discovery source schema differs")

    candidate_count = 0
    for ordinal, row in enumerate(reader, start=2):
        raw_url = row["url"].strip()
        host = (urlsplit(raw_url).hostname or row["host"]).lower().rstrip(".")
        signal = row["ai_signal"].strip()
        is_candidate = (
            any(host_matches(host, suffix) for suffix in member["host_suffixes"])
            or signal in member["accepted_signals"]
        )
        if not is_candidate:
            continue
        if not host or not row["trace_id"].strip():
            raise SystemExit("candidate observation lacks stable identity")
        candidate_count += 1
        events.append({
            "schema_version": "aidefend.ai-service-observation.v2",
            "tenant_id": profile["tenant_id"],
            "release_scope": profile["release_scope"],
            "profile_version": profile["profile_version"],
            "source_type": member["source_type"],
            "source_member_id": member_id,
            "source_record_ordinal": ordinal,
            "raw_source_sha256": raw_sha256,
            "observed_at": row["timestamp"],
            "app_hostname": host,
            "app_url": raw_url or None,
            "principal_id": row["principal_id"] or None,
            "trace_id": row["trace_id"],
            "service_tenant": row["service_tenant"] or None,
            "data_signals": sorted(filter(None, row["data_signals"].split(";"))),
            "candidate_signal": signal or "governed_host_fact",
        })
    coverage.append({
        "member_id": member_id,
        "source_type": member["source_type"],
        "raw_sha256": raw_sha256,
        "candidate_count": candidate_count,
        "attempt_status": "COMPLETED",
    })

args.events.parent.mkdir(parents=True, exist_ok=True)
args.events.write_text(
    "".join(json.dumps(item, sort_keys=True) + "\n" for item in events),
    encoding="utf-8",
)
observation_bytes = args.events.read_bytes()
artifact = {
    "schema_version": "aidefend.ai-service-discovery-coverage.v2",
    "tenant_id": profile["tenant_id"],
    "release_scope": profile["release_scope"],
    "profile_version": profile["profile_version"],
    "profile_sha256": hashlib.sha256(args.profile.read_bytes()).hexdigest(),
    "expected_member_count": len(profile["members"]),
    "measured_member_count": len(coverage),
    "candidate_event_count": len(events),
    "observations_sha256": hashlib.sha256(observation_bytes).hexdigest(),
    "members": coverage,
    "outcome": "PASS",
}
args.coverage.write_text(
    json.dumps(artifact, sort_keys=True, separators=(",", ":")) + "\n",
    encoding="utf-8",
)

Action: Obtain the profile through the organization's governed configuration path and verify its trust/currentness before this normalizer runs. The resulting PASS means the declared discovery members were attempted from their retained exact bytes; it does not mean every candidate is sanctioned or that zero candidates proves no AI use. Classification and owner reconciliation remain separate steps.

AID-M-001.004-G002

Purpose and authority

Classify every candidate observation against one current catalog for the exact tenant and release scope. The catalog is a governed authority input, not a mutable local hint file. Capture its bytes once, verify them through the organization's catalog trust path, and pass the expected digest and authority generation to the classifier. Unknown, ambiguous, blocked, or disallowed observations are findings; merely producing a non-empty output cannot yield PASS.

Governed catalog

{
  "schema_version": "aidefend.ai-service-catalog.v2",
  "catalog_version": "catalog-103",
  "authority_generation": "catalog-authority-2026-08-28T00:00:00Z",
  "tenant_id": "acme",
  "issued_at": "2026-08-28T00:00:00Z",
  "expires_at": "2026-08-29T00:00:00Z",
  "services": [
    {
      "service_id": "openai-enterprise",
      "status": "sanctioned",
      "owner_team": "ai-platform",
      "approved_tenants": ["acme-enterprise"],
      "allowed_data_classes": ["public", "internal"],
      "patterns": [
        {"kind": "exact", "value": "api.openai.com", "priority": 100},
        {"kind": "suffix", "value": "chatgpt.com", "priority": 90}
      ]
    }
  ]
}

Classify deterministically and reconcile the full candidate population

from __future__ import annotations

import argparse
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path

CATALOG_FIELDS = {
    "schema_version", "catalog_version", "authority_generation",
    "tenant_id", "issued_at", "expires_at", "services",
}
SERVICE_FIELDS = {
    "service_id", "status", "owner_team", "approved_tenants",
    "allowed_data_classes", "patterns",
}
PATTERN_FIELDS = {"kind", "value", "priority"}
DIGEST_CHARS = set("0123456789abcdef")


def strict_json_bytes(raw: bytes, label: str) -> object:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key in {label}: {key}")
            value[key] = item
        return value
    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique,
        parse_constant=lambda token: (_ for _ in ()).throw(ValueError(token)),
    )


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


def matches(host: str, pattern: dict) -> bool:
    value = pattern["value"]
    return host == value if pattern["kind"] == "exact" else (
        host == value or host.endswith("." + value)
    )


parser = argparse.ArgumentParser()
parser.add_argument("--catalog", type=Path, required=True)
parser.add_argument("--catalog-sha256", required=True)
parser.add_argument("--authority-generation", required=True)
parser.add_argument("--tenant", required=True)
parser.add_argument("--release-scope", required=True)
parser.add_argument("--coverage", type=Path, required=True)
parser.add_argument("--observations", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()

catalog_raw = args.catalog.read_bytes()
if (
    len(args.catalog_sha256) != 64
    or set(args.catalog_sha256) - DIGEST_CHARS
    or hashlib.sha256(catalog_raw).hexdigest() != args.catalog_sha256
):
    raise SystemExit("catalog stable-byte binding differs")
catalog = strict_json_bytes(catalog_raw, "catalog")
if not isinstance(catalog, dict) or set(catalog) != CATALOG_FIELDS or (
    catalog["schema_version"] != "aidefend.ai-service-catalog.v2"
    or catalog["authority_generation"] != args.authority_generation
    or catalog["tenant_id"] != args.tenant
    or not when(catalog["issued_at"]) <= datetime.now(timezone.utc)
       < when(catalog["expires_at"])
    or not isinstance(catalog["services"], list)
    or not catalog["services"]
):
    raise SystemExit("catalog schema, scope, authority, or currentness differs")

services: list[dict] = []
service_ids: set[str] = set()
for service in catalog["services"]:
    if not isinstance(service, dict) or set(service) != SERVICE_FIELDS:
        raise SystemExit("catalog service schema differs")
    if (
        not isinstance(service["service_id"], str) or not service["service_id"]
        or service["service_id"] in service_ids
        or service["status"] not in {"sanctioned", "restricted", "blocked"}
        or not isinstance(service["patterns"], list)
        or not service["patterns"]
        or any(not isinstance(value, list) for value in (
            service["approved_tenants"], service["allowed_data_classes"]
        ))
    ):
        raise SystemExit("catalog service identity or policy differs")
    service_ids.add(service["service_id"])
    seen_patterns: set[tuple[str, str]] = set()
    for pattern in service["patterns"]:
        if not isinstance(pattern, dict) or set(pattern) != PATTERN_FIELDS:
            raise SystemExit("catalog host pattern schema differs")
        identity = (pattern["kind"], str(pattern["value"]).lower().rstrip("."))
        if (
            pattern["kind"] not in {"exact", "suffix"}
            or not identity[1] or identity in seen_patterns
            or isinstance(pattern["priority"], bool)
            or not isinstance(pattern["priority"], int)
        ):
            raise SystemExit("catalog host pattern is invalid")
        pattern["value"] = identity[1]
        seen_patterns.add(identity)
    services.append(service)

coverage_raw = args.coverage.read_bytes()
coverage = strict_json_bytes(coverage_raw, "coverage")
observation_raw = args.observations.read_bytes()
if not isinstance(coverage, dict) or (
    coverage.get("outcome") != "PASS"
    or coverage.get("tenant_id") != args.tenant
    or coverage.get("release_scope") != args.release_scope
    or coverage.get("observations_sha256")
       != hashlib.sha256(observation_raw).hexdigest()
):
    raise SystemExit("candidate population coverage is incomplete or unbound")

observations = [
    strict_json_bytes(line, f"observation {index}")
    for index, line in enumerate(observation_raw.splitlines(), start=1)
    if line.strip()
]
identities: set[tuple[str, str, int]] = set()
classified: list[dict] = []
finding_count = 0
for observation in observations:
    if not isinstance(observation, dict):
        raise SystemExit("observation is not an object")
    identity = (
        observation.get("source_member_id"),
        observation.get("trace_id"),
        observation.get("source_record_ordinal"),
    )
    if (
        observation.get("tenant_id") != args.tenant
        or observation.get("release_scope") != args.release_scope
        or identity in identities
        or not all(identity)
    ):
        raise SystemExit("observation scope or identity differs")
    identities.add(identity)
    host = str(observation.get("app_hostname", "")).lower().rstrip(".")
    candidates = [
        (pattern["priority"], service, pattern)
        for service in services
        for pattern in service["patterns"]
        if matches(host, pattern)
    ]
    if not candidates:
        classification, service_id, matched_pattern = "unknown", None, None
        allowed_data: set[str] = set()
    else:
        top_priority = max(item[0] for item in candidates)
        winners = [item for item in candidates if item[0] == top_priority]
        winner_ids = {item[1]["service_id"] for item in winners}
        if len(winner_ids) != 1:
            classification, service_id, matched_pattern = "ambiguous", None, None
            allowed_data = set()
        else:
            _, service, pattern = winners[0]
            service_id = service["service_id"]
            matched_pattern = {"kind": pattern["kind"], "value": pattern["value"]}
            approved_tenant = observation.get("service_tenant")
            classification = service["status"]
            if service["approved_tenants"] and approved_tenant not in service["approved_tenants"]:
                classification = "blocked"
            allowed_data = set(service["allowed_data_classes"])
    data_signals = set(observation.get("data_signals") or [])
    disallowed = sorted(data_signals - allowed_data)
    finding = classification in {"unknown", "ambiguous", "blocked"} or bool(disallowed)
    finding_count += int(finding)
    classified.append({
        **observation,
        "catalog_version": catalog["catalog_version"],
        "catalog_sha256": args.catalog_sha256,
        "authority_generation": catalog["authority_generation"],
        "service_id": service_id,
        "matched_pattern": matched_pattern,
        "classification_status": classification,
        "disallowed_data_signals": disallowed,
        "finding": finding,
    })

artifact = {
    "schema_version": "aidefend.ai-service-classification.v2",
    "tenant_id": args.tenant,
    "release_scope": args.release_scope,
    "catalog_version": catalog["catalog_version"],
    "catalog_sha256": args.catalog_sha256,
    "coverage_sha256": hashlib.sha256(coverage_raw).hexdigest(),
    "expected_observation_count": coverage.get("candidate_event_count"),
    "classified_observation_count": len(classified),
    "finding_count": finding_count,
    "outcome": "FAIL" if finding_count else "PASS",
    "observations": classified,
}
if artifact["expected_observation_count"] != len(classified):
    raise SystemExit("classified population differs from coverage receipt")
args.output.write_text(
    json.dumps(artifact, sort_keys=True, separators=(",", ":")) + "\n",
    encoding="utf-8",
)
if artifact["outcome"] != "PASS":
    raise SystemExit("AI service classification contains findings")

Action: The catalog adapter must verify the configured authority and current generation before invoking this code. Preserve unknown and ambiguous results as findings rather than choosing the first matching pattern. A catalog or population trust failure is ERROR or INSUFFICIENT_DATA; a fully classified population with security findings is FAIL.

AID-M-001.004-G003

Consume the actual G002 classification artifact

Export the governed declared-service inventory below and verify its signature, tenant/release scope, current revision and digest before use. Consume the complete JSON aidefend.ai-service-classification.v2 output from G002, including FAIL results containing findings; it is not JSONL. Bind the exact captured artifact to its verified upstream receipt. A CI job must preserve and reconcile a finding artifact even when G002 exits nonzero; an execution error without an authenticated artifact remains ERROR.

# File: discovery/declared_ai_services.json
{
  "declared_services": [
    {
      "service_id": "openai-enterprise",
      "owner_team": "ai-platform",
      "approved_principals": ["group:approved-ai-users"],
      "inventory_source": "AID-M-001.002",
      "dependency_record_id": "dep-openai-enterprise-001"
    }
  ]
}

# File: discovery/reconcile_ai_services.py
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path

def reconcile(artifact, declared, *, tenant, release_scope):
    if (artifact.get("schema_version") != "aidefend.ai-service-classification.v2"
            or artifact.get("tenant_id") != tenant
            or artifact.get("release_scope") != release_scope):
        raise ValueError("classification artifact scope/schema differs")
    events = artifact.get("observations")
    expected = artifact.get("expected_observation_count")
    if not isinstance(events, list) or type(expected) is not int or expected < 0:
        raise ValueError("classification population schema differs")
    if (expected != len(events) or artifact.get("classified_observation_count") != len(events)
            or artifact.get("outcome") == "INSUFFICIENT_DATA"):
        return {"status": "INSUFFICIENT_DATA", "findings": [], "reason": "incomplete upstream population"}
    if artifact.get("outcome") not in {"PASS", "FAIL"}:
        raise ValueError("upstream producer error/unknown state")
    records = declared["declared_services"]
    inventory = {row["service_id"]: row for row in records}
    if len(inventory) != len(records):
        raise ValueError("duplicate declared service")
    findings, identities, upstream_findings = [], set(), 0
    for event in events:
        identity = (event["source_member_id"], event["trace_id"], event["source_record_ordinal"])
        if identity in identities or event.get("tenant_id") != tenant or event.get("release_scope") != release_scope:
            raise ValueError("duplicate or misbound observation")
        identities.add(identity)
        service = event.get("service_id")
        classification = event["classification_status"]
        if classification not in {"sanctioned", "restricted", "blocked", "unknown", "ambiguous"}:
            raise ValueError("unknown classification state")
        disallowed = event.get("disallowed_data_signals")
        if not isinstance(disallowed, list) or any(not isinstance(x, str) for x in disallowed):
            raise ValueError("data classification signals malformed")
        observed_finding = classification in {"blocked", "unknown", "ambiguous"} or bool(disallowed)
        if event.get("finding") is not observed_finding:
            raise ValueError("classification finding state differs")
        upstream_findings += int(observed_finding)
        reasons = []
        if service is None or service not in inventory:
            reasons.append("undeclared_ai_service_observed")
        if classification in {"blocked", "unknown", "ambiguous"}:
            reasons.append("classification_" + classification)
        if disallowed:
            reasons.append("data_scope_drift")
        owner = inventory.get(service, {}).get("owner_team")
        if not owner:
            reasons.append("inventory_owner_missing")
        if reasons:
            findings.append({"source_member_id": identity[0], "trace_id": identity[1],
                             "source_record_ordinal": identity[2], "service_id": service,
                             "app_hostname": event["app_hostname"], "owner_team": owner,
                             "reasons": reasons, "disallowed_data_signals": disallowed})
    if (artifact.get("finding_count") != upstream_findings
            or artifact["outcome"] != ("FAIL" if upstream_findings else "PASS")):
        raise ValueError("upstream count or finding outcome is inconsistent")
    return {"status": "FAIL" if findings else "PASS", "findings": findings,
            "observed_count": len(events), "catalog_sha256": artifact["catalog_sha256"],
            "coverage_sha256": artifact["coverage_sha256"]}

def main():
    parser = argparse.ArgumentParser()
    for name in ("classification", "declared", "output"):
        parser.add_argument("--" + name, type=Path, required=True)
    for name in ("classification-sha256", "declared-sha256", "tenant", "release-scope"):
        parser.add_argument("--" + name, required=True)
    args = parser.parse_args()
    # Expected digests come from verified current upstream/inventory receipts.
    captured = []
    for path, expected in [(args.classification, args.classification_sha256),
                           (args.declared, args.declared_sha256)]:
        raw = path.read_bytes()
        if hashlib.sha256(raw).hexdigest() != expected:
            raise ValueError("captured input does not match verified receipt")
        captured.append(json.loads(raw))
    result = reconcile(*captured, tenant=args.tenant, release_scope=args.release_scope)
    result.update(schema_version="aidefend.ai-service-reconciliation.v2",
                  tenant_id=args.tenant, release_scope=args.release_scope,
                  classification_sha256=args.classification_sha256,
                  declared_sha256=args.declared_sha256)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8")
    if result["status"] != "PASS":
        raise SystemExit(1)

if __name__ == "__main__":
    main()

Route findings and verify the result

Invoke the script with --classification <G002-output.json>, --declared <verified-inventory.json>, their receipt-derived --classification-sha256 and --declared-sha256, --tenant, --release-scope and --output. The trusted launcher supplies those digests, never a discovery request. Unknown/ambiguous services retain their source/trace identity even when service_id is null; blocked known services and disallowed data remain findings. A responsible owner does not turn a finding into PASS. Principal/device entitlements require a separate authoritative IAM join; this reference path makes no claim to have measured those permissions. Missing IAM coverage remains unknown when that join is required by deployment policy. Sign the resulting artifact, read it back independently, and replay mixed known/unknown, blocked known, ambiguous, data drift, missing owner, duplicate trace, empty-but-completely-measured and partial-population cases. Inventory reconciliation routes findings to Harden/Isolate; it does not enforce access.

Open-source tools

• OpenTelemetry
• Zeek
• OpenSearch
• Elasticsearch 8.16+ AGPLv3 source build
• Sigma CLI
• pySigma

Commercial / hosted tools

• Microsoft Purview
• Microsoft Defender for Cloud Apps
• Microsoft Entra Global Secure Access
• Netskope One AI Security
• Zscaler AI Security
• Palo Alto Networks Prisma Access
• Palo Alto Networks Enterprise Data Loss Prevention
• Google Security Operations (YARA-L 2.0 detection rules)

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• N/A

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AITech-8.2 Data Exfiltration / Exposure (Detects observed data-class sharing outside approved AI-service or tenant scope)
• AITech-8.3 Information Disclosure (Detects observed data-class sharing outside approved AI-service or tenant scope)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Governance 4.1: Lack of traceability and transparency of model assets