"""Single source of truth for AI / LLM surface detection signals.
This module is the canonical catalogue consumed by every distributed AI-recon
hook (domain_recon, port_scan/masscan/nmap, http_probe, resource_enum, js_recon,
vuln_scan, subdomain_takeover, vhost_sni_enum, add_mitre, the OSINT enrichments,
and the future central ai_surface_recon module). Every hook imports the constants
it needs from this file and never duplicates the data inline.
Naming convention (see internal/ADVERSARIAL_AI/AI_SURFACE_RECON.md §7.0):
- Properties prefixed ``ai_`` or ``is_ai_`` are AI surface annotations.
- Values prefixed ``ai-``, ``llm-``, or ``AML.T`` are AI classifications on
fields whose own name is generic (Technology.category, MitreData.id, etc.).
This file is forward-only: later integration laps fill the empty stubs at the
bottom. The import path stays stable from day one so distributed hooks land
cleanly across laps.
"""
from __future__ import annotations
import re
from typing import Pattern
# ---------------------------------------------------------------------------
# domain_recon (AI_SURFACE_RECON.md §2.1 #1)
# ---------------------------------------------------------------------------
# AI provider substrings to look for inside TXT records (SPF, DKIM, DMARC,
# verification tokens). Order matters: the first match wins so that the
# strongest signal (a clear vendor domain) outranks generic CDN hints.
AI_TXT_PATTERNS: list[tuple[Pattern[str], str]] = [
(re.compile(r"\banthropic\.com\b", re.IGNORECASE), "anthropic"),
(re.compile(r"\bopenai\.com\b", re.IGNORECASE), "openai"),
(re.compile(r"\bhuggingface\.co\b", re.IGNORECASE), "huggingface"),
(re.compile(r"\bcohere\.com\b", re.IGNORECASE), "cohere"),
(re.compile(r"\breplicate\.com\b", re.IGNORECASE), "replicate"),
(re.compile(r"\blangchain\.com\b", re.IGNORECASE), "langchain"),
(re.compile(r"\blangfuse\.com\b", re.IGNORECASE), "langfuse"),
(re.compile(r"\blangsmith\.com\b", re.IGNORECASE), "langsmith"),
(re.compile(r"\btogether\.ai\b", re.IGNORECASE), "together"),
(re.compile(r"\bgroq\.com\b", re.IGNORECASE), "groq"),
(re.compile(r"\bmistral\.ai\b", re.IGNORECASE), "mistral"),
]
# NS-record substrings that hint at AI-friendly hosting providers. Always a
# weak signal — these providers host plenty of non-AI sites — so the consumer
# must only set ``Subdomain.ai_service_hint = "ai-hosting-candidate"`` when no
# stronger TXT hint already exists.
AI_NS_HINT_PATTERNS: list[tuple[Pattern[str], str]] = [
(re.compile(r"\bvercel-dns\b", re.IGNORECASE), "vercel"),
(re.compile(r"\bnsone\.net\b", re.IGNORECASE), "netlify"),
(re.compile(r"\breplit\b", re.IGNORECASE), "replit"),
(re.compile(r"\bmodal-dns\b", re.IGNORECASE), "modal"),
(re.compile(r"\bhuggingface\.co\b", re.IGNORECASE), "huggingface-spaces"),
]
# ---------------------------------------------------------------------------
# port_scan / masscan_scan (AI_SURFACE_RECON.md §2.1 #2)
# ---------------------------------------------------------------------------
# Port → AI service descriptor.
#
# Each entry carries ``name`` (the Technology.name we MERGE in the graph),
# ``category`` (Technology.category value, always prefixed ``ai-``), and an
# optional ``disambiguate`` flag. When ``disambiguate`` is True the port is
# shared between AI and non-AI services (e.g. 8000 is also a generic dev
# server) and the lookup must be gated on a corroborating signal — a matching
# header or title from http_probe — before any AI annotation is written. The
# port_scan hook never sets an AI tag on a ``disambiguate=True`` port on its
# own; that promotion happens later in the central ai_surface_recon module
# once chat-shape probes confirm the surface.
AI_PORTS: dict[int, dict[str, str | bool]] = {
# ─── Local model runtimes ─────────────────────────────────────────────
11434: {"name": "ollama", "category": "ai-runtime"},
1234: {"name": "lm-studio", "category": "ai-runtime", "disambiguate": True},
3000: {"name": "bentoml-or-langflow-or-openllm", "category": "ai-runtime", "disambiguate": True},
5000: {"name": "mlflow-or-flask", "category": "ai-mlops", "disambiguate": True},
5001: {"name": "koboldcpp", "category": "ai-runtime", "disambiguate": True},
8000: {"name": "vllm-or-chroma-or-langserve-or-nim-or-mlc-or-faster-whisper", "category": "ai-runtime", "disambiguate": True},
8001: {"name": "triton-or-vllm-or-redis-insight", "category": "ai-runtime", "disambiguate": True},
8002: {"name": "triton-metrics", "category": "ai-runtime", "disambiguate": True},
8080: {"name": "open-webui-or-weaviate-or-localai-or-whisper-cpp", "category": "ai-frontend", "disambiguate": True},
8880: {"name": "kokoro-tts", "category": "ai-runtime"},
30000: {"name": "sglang", "category": "ai-runtime"},
# ─── Vector databases ─────────────────────────────────────────────────
6333: {"name": "qdrant", "category": "ai-vector-db"},
6334: {"name": "qdrant-grpc", "category": "ai-vector-db"},
19530: {"name": "milvus", "category": "ai-vector-db"},
9091: {"name": "milvus-metrics-or-invokeai-or-prometheus", "category": "ai-vector-db", "disambiguate": True},
50051: {"name": "weaviate-grpc", "category": "ai-vector-db", "disambiguate": True},
# ─── Proxies / gateways ───────────────────────────────────────────────
4000: {"name": "litellm", "category": "ai-proxy", "disambiguate": True},
# ─── Frontends / web UIs ──────────────────────────────────────────────
7860: {"name": "gradio-or-automatic1111-or-langflow", "category": "ai-frontend"},
7865: {"name": "fooocus", "category": "ai-frontend", "disambiguate": True},
8188: {"name": "comfyui", "category": "ai-frontend"},
8501: {"name": "streamlit", "category": "ai-frontend"},
3001: {"name": "anythingllm", "category": "ai-frontend", "disambiguate": True},
9090: {"name": "invokeai", "category": "ai-frontend", "disambiguate": True},
# ─── MLOps / observability stacks ─────────────────────────────────────
6006: {"name": "phoenix-arize-or-tensorboard", "category": "ai-mlops", "disambiguate": True},
6900: {"name": "argilla", "category": "ai-mlops"},
8081: {"name": "autogen-studio", "category": "ai-mlops", "disambiguate": True},
8123: {"name": "langgraph-or-clickhouse", "category": "ai-framework", "disambiguate": True},
8265: {"name": "ray-dashboard", "category": "ai-mlops", "disambiguate": True},
2024: {"name": "langgraph-dev", "category": "ai-framework"},
}
# ---------------------------------------------------------------------------
# nmap_scan (AI_SURFACE_RECON.md §2.1 #4)
# ---------------------------------------------------------------------------
# Regex applied to nmap's ``product`` / ``version`` fields. On match, set
# ``Service.ai_runtime_version`` to the matched substring so downstream CVE
# lookups can join against AI library CVE clusters in later laps.
AI_NMAP_VERSION_PATTERNS: list[tuple[Pattern[str], str]] = [
(re.compile(r"\bOllama/", re.IGNORECASE), "ollama"),
(re.compile(r"\bvllm/", re.IGNORECASE), "vllm"),
(re.compile(r"\bLiteLLM/", re.IGNORECASE), "litellm"),
(re.compile(r"\bTGI/|text-generation-inference/", re.IGNORECASE), "tgi"),
(re.compile(r"\btriton-server/", re.IGNORECASE), "triton"),
(re.compile(r"\bllama\.cpp/", re.IGNORECASE), "llama.cpp"),
]
# ---------------------------------------------------------------------------
# http_probe (AI_SURFACE_RECON.md §2.1 #5)
# ---------------------------------------------------------------------------
# AI-stack header signature regex. Matched against captured response header
# *names* (case-insensitive). First match wins; ordering matters because some
# headers (``x-litellm-*``) hint at a proxy in front of a real runtime, so the
# runtime headers come first.
#
# Tuple shape: (header_name_pattern, framework_name, technology_category).
AI_HEADER_PATTERNS: list[tuple[Pattern[str], str, str]] = [
# Runtimes — strongest signal
(re.compile(r"^x-vllm-", re.IGNORECASE), "vllm", "ai-runtime"),
(re.compile(r"^x-tgi-", re.IGNORECASE), "tgi", "ai-runtime"),
(re.compile(r"^x-tei-", re.IGNORECASE), "text-embeddings-inference", "ai-runtime"),
(re.compile(r"^x-bentoml-", re.IGNORECASE), "bentoml", "ai-runtime"),
(re.compile(r"^x-baseten-", re.IGNORECASE), "baseten", "ai-runtime"),
(re.compile(r"^x-modal-", re.IGNORECASE), "modal", "ai-runtime"),
(re.compile(r"^x-replicate-", re.IGNORECASE), "replicate", "ai-runtime"),
(re.compile(r"^x-runpod-", re.IGNORECASE), "runpod", "ai-runtime"),
# Frameworks / orchestrators
(re.compile(r"^x-langchain-", re.IGNORECASE), "langchain", "ai-framework"),
(re.compile(r"^x-llamaindex-", re.IGNORECASE), "llamaindex", "ai-framework"),
(re.compile(r"^langfuse-", re.IGNORECASE), "langfuse", "ai-framework"),
# Proxies / gateways
(re.compile(r"^x-litellm-", re.IGNORECASE), "litellm", "ai-proxy"),
(re.compile(r"^x-helicone-", re.IGNORECASE), "helicone", "ai-proxy"),
(re.compile(r"^x-portkey-", re.IGNORECASE), "portkey", "ai-proxy"),
(re.compile(r"^x-omniroute-", re.IGNORECASE), "omniroute", "ai-proxy"),
(re.compile(r"^cf-aig-", re.IGNORECASE), "cloudflare-ai-gateway", "ai-proxy"),
(re.compile(r"^together-", re.IGNORECASE), "together", "ai-proxy"),
# SDK clients (proxied vendor calls)
(re.compile(r"^openai-(organization|version|processing-ms)", re.IGNORECASE), "openai", "ai-sdk-client"),
(re.compile(r"^anthropic-(version|beta|ratelimit-)", re.IGNORECASE), "anthropic", "ai-sdk-client"),
# Azure OpenAI: x-ms-region + azureml-model-session are unique to AOAI.
# The combination is the strongest signal, but each alone is high-confidence
# for an AOAI-fronted endpoint.
(re.compile(r"^x-ms-region$|^azureml-model-session$", re.IGNORECASE), "azure-openai", "ai-sdk-client"),
# Fireworks AI's unique ratelimit suffix and account header.
(re.compile(r"^x-ratelimit-limit-tokens-cache-adjusted-prompt$|^x-fireworks-account-id$", re.IGNORECASE), "fireworks", "ai-sdk-client"),
# MCP
(re.compile(r"^x-mcp-", re.IGNORECASE), "mcp", "ai-framework"),
]
# Page-title regex catalogue for AI frontend products. Matched against the
# captured ``title`` from httpx. Each entry maps to a Technology(name) under
# Technology.category = "ai-frontend".
AI_TITLE_PATTERNS: list[tuple[Pattern[str], str]] = [
# ─── Chat / generic LLM frontends ─────────────────────────────────────
(re.compile(r"\bOpen WebUI\b", re.IGNORECASE), "open-webui"),
(re.compile(r"\bLibreChat\b", re.IGNORECASE), "librechat"),
(re.compile(r"\bAnythingLLM\b", re.IGNORECASE), "anythingllm"),
(re.compile(r"\bFlowise\b", re.IGNORECASE), "flowise"),
(re.compile(r"\bLangflow\b", re.IGNORECASE), "langflow"),
(re.compile(r"\bDify\b", re.IGNORECASE), "dify"),
(re.compile(r"\bComfyUI\b", re.IGNORECASE), "comfyui"),
(re.compile(r"\bGradio\b", re.IGNORECASE), "gradio"),
(re.compile(r"\bStreamlit\b", re.IGNORECASE), "streamlit"),
# Specific clones first, then the generic ChatGPT-clone fallback.
(re.compile(r"\bBetterChatGPT\b", re.IGNORECASE), "betterchatgpt"),
(re.compile(r"\bOnyx\b|\bDanswer\b", re.IGNORECASE), "onyx"),
(re.compile(r"\bChatGPT\b", re.IGNORECASE), "chatgpt-clone"),
(re.compile(r"\bHuggingFace Chat UI\b", re.IGNORECASE), "hf-chat-ui"),
(re.compile(r"\bLobeChat\b|\bLobeHub\b", re.IGNORECASE), "lobechat"),
(re.compile(r"\bNextChat\b", re.IGNORECASE), "nextchat"),
(re.compile(r"\bSillyTavern\b", re.IGNORECASE), "sillytavern"),
(re.compile(r"\bJan\b\s*-\s*Open\s*Source", re.IGNORECASE), "jan"),
(re.compile(r"\bh2oGPT\b", re.IGNORECASE), "h2ogpt"),
(re.compile(r"\bPrivateGPT\b", re.IGNORECASE), "privategpt"),
(re.compile(r"\bQuivr\b", re.IGNORECASE), "quivr"),
# ─── Image-gen UIs (run on Gradio shell) ──────────────────────────────
# InvokeAI: exact title "Invoke - Community Edition" set via index.html
(re.compile(r"\bInvoke\s*-\s*Community Edition\b", re.IGNORECASE), "invokeai"),
# A1111 / Forge: title is just "Stable Diffusion". This is broad — body
# fingerprint (txt2img_textarea) is the high-confidence channel; title
# here is a corroborating hint.
(re.compile(r"^Stable Diffusion$", re.IGNORECASE), "automatic1111"),
# ─── MLOps / observability frontends ──────────────────────────────────
# Word-boundary patterns: title field from httpx is the bare text inside
# ..., not the markup itself. Anchored "^...$" patterns
# require the title to be exactly the product name (no surrounding text).
(re.compile(r"^MLflow$", re.IGNORECASE), "mlflow"),
(re.compile(r"^Labelstudio$", re.IGNORECASE), "label-studio"),
(re.compile(r"\bRay Dashboard\b", re.IGNORECASE), "ray-dashboard"),
(re.compile(r"\bRedisInsight\b", re.IGNORECASE), "redis-insight"),
(re.compile(r"\bAutoGen Studio\b", re.IGNORECASE), "autogen-studio"),
(re.compile(r"\bLangfuse\b", re.IGNORECASE), "langfuse-ui"),
(re.compile(r"\bArize Phoenix\b|^Phoenix$", re.IGNORECASE), "phoenix-arize"),
(re.compile(r"\bArgilla\b", re.IGNORECASE), "argilla"),
(re.compile(r"\bGPT Researcher\b", re.IGNORECASE), "gpt-researcher"),
]
# AI Wappalyzer-style fingerprints. Each entry matches against the HTTP
# response *body* (HTML / JS bundle text). Patterns are deliberately specific
# enough that a casual mention in documentation will NOT trip them — we want
# fingerprints of the deployed product, not blog posts about it.
#
# Tuple shape: (body_regex, framework_name, technology_category).
# First match wins. Iteration order: runtimes > frameworks > frontends > sdk.
AI_BODY_FINGERPRINTS: list[tuple[Pattern[str], str, str]] = [
# --- Runtimes ---------------------------------------------------------
# TGI: the streaming form action / API client path
(re.compile(r"""(?:action|href|fetch\()\s*=?\s*["']/generate_stream["']""", re.IGNORECASE), "tgi", "ai-runtime"),
# vLLM: session cookie literal often echoed in error pages or JS
(re.compile(r"\bvllm_session\b", re.IGNORECASE), "vllm", "ai-runtime"),
# --- Frameworks -------------------------------------------------------
# LangChain JS globals injected into the page by langchain/langgraph apps
(re.compile(r"window\.__LANGCHAIN__|window\.__LANGCHAIN_TRACING_V2__", re.IGNORECASE), "langchain", "ai-framework"),
# LangChain JS package import string visible in unminified bundles
(re.compile(r"""@langchain/(core|community|langgraph|openai|anthropic)["']""", re.IGNORECASE), "langchain", "ai-framework"),
# LlamaIndex JS package + global
(re.compile(r"""@llamaindex/(core|cloud|community)["']|window\.LlamaIndex\b""", re.IGNORECASE), "llamaindex", "ai-framework"),
# --- Image-gen UIs (run on Gradio shell — these patterns disambiguate) ---
# A1111 Stable Diffusion WebUI — high-confidence textarea IDs + JS hooks
(re.compile(r"\btxt2img_textarea\b|\bimg2img_textarea\b|\bonAfterUiUpdate\b|\bgradioApp\(\)", re.IGNORECASE), "automatic1111", "ai-frontend"),
# Fooocus — literal version tag in their script.js shipped to browser
(re.compile(r"\bfooocus_v2\b", re.IGNORECASE), "fooocus", "ai-frontend"),
# InvokeAI — branded favicon asset path is shipped in the HTML
(re.compile(r"invoke-favicon\.svg|/src/main\.tsx['\"]", re.IGNORECASE), "invokeai", "ai-frontend"),
# ComfyUI frontend (Vue rewrite) — distinctive splash + manifest pattern
(re.compile(r'aria-label=["\']Loading ComfyUI["\']|\bcomfy-splash-bg\b', re.IGNORECASE), "comfyui", "ai-frontend"),
# --- General frontends ------------------------------------------------
# Gradio: custom element tag + the runtime config global it bootstraps
(re.compile(r" str | None:
"""Return the provider name if the TXT record value matches a known AI vendor.
Used by domain_recon's AI TXT hint hook. Returns the first match (patterns
are ordered by strength) or None if no AI signal is present.
"""
if not record_value:
return None
for pattern, hint in AI_TXT_PATTERNS:
if pattern.search(record_value):
return hint
return None
def match_ai_ns_hint(record_value: str) -> str | None:
"""Return the provider hint if the NS record matches a known AI-hosting platform.
Always the weakest signal. The caller must only apply this hint when no
stronger TXT-derived hint has already been recorded for the host.
"""
if not record_value:
return None
for pattern, hint in AI_NS_HINT_PATTERNS:
if pattern.search(record_value):
return hint
return None
def lookup_ai_port(port: int) -> dict[str, str | bool] | None:
"""Return the AI service descriptor for a port, or None.
The descriptor carries ``name``, ``category``, and optionally
``disambiguate``. When ``disambiguate`` is True, the port is shared between
AI and non-AI services and the caller must require a corroborating signal
(matching header / title from http_probe) before promoting the annotation
to a Technology graph node.
"""
return AI_PORTS.get(port)
def match_ai_header(header_name: str) -> tuple[str, str] | None:
"""Return (framework_name, technology_category) on first matching header pattern.
Matched against the header *name* only. Header values are not inspected
here — that responsibility belongs to the central ai_surface_recon module
in phase 15.
"""
if not header_name:
return None
for pattern, framework, category in AI_HEADER_PATTERNS:
if pattern.search(header_name):
return framework, category
return None
def match_ai_title(title: str) -> str | None:
"""Return the AI frontend product name if the page title matches."""
if not title:
return None
for pattern, product in AI_TITLE_PATTERNS:
if pattern.search(title):
return product
return None
def match_ai_nmap_version(product_or_version: str) -> str | None:
"""Return the AI runtime name if the nmap product/version string matches."""
if not product_or_version:
return None
for pattern, runtime in AI_NMAP_VERSION_PATTERNS:
if pattern.search(product_or_version):
return runtime
return None
def match_ai_body_fingerprint(body: str) -> tuple[str, str] | None:
"""Return (framework_name, technology_category) on the first body match.
Scans against `AI_BODY_FINGERPRINTS` — the Wappalyzer-style catalogue
of regexes that match AI-product signatures embedded in the response
body (HTML markup, shipped JS, form actions). Returns None if no entry
matches. Bodies larger than a few hundred KB should be capped by the
caller before invoking this helper.
"""
if not body:
return None
for pattern, framework, category in AI_BODY_FINGERPRINTS:
if pattern.search(body):
return framework, category
return None
# ---------------------------------------------------------------------------
# resource_enum helpers
# ---------------------------------------------------------------------------
def match_ai_path(path: str) -> str | None:
"""Return the ``ai_interface_type`` enum value if the URL path matches
a known LLM-shape route, else ``None``.
Iterates ``AI_PATH_PATTERNS`` in catalogue order (vendor-specific first,
generic fallbacks last). Matches on the path component only — query string
must be stripped by the caller. Case-insensitive.
Uses ``re.search`` (not ``re.match``) so patterns anchored with
``(?:^|/)`` correctly fire on suffix positions (e.g. LangServe's
``/agents/foo/stream``). Most patterns are explicitly anchored with
``^``, so search-vs-match makes no behavioural difference there.
The returned value is one of: ``llm-chat``, ``llm-completion``,
``llm-embedding``, ``llm-tool-call``, ``sse-stream``, ``mcp``,
``llm-graphql``. Callers that want the explicit ``non-llm`` sentinel
must substitute it themselves when this helper returns ``None``.
"""
if not path:
return None
for pattern, interface_type in AI_PATH_PATTERNS:
if pattern.search(path):
return interface_type
return None
def is_ai_rag_path(path: str, parent_is_ai: bool = False) -> bool:
"""True if the URL path looks like a RAG ingestion or retrieval endpoint.
Some patterns (``/upload``, ``/search``, ``/query``) are too generic to
flag standalone — they collide with file-upload and search bars on every
e-commerce site. Those patterns carry an ``ambiguous=True`` flag in the
catalogue and only fire when ``parent_is_ai`` is True (i.e. the parent
BaseURL or Service is already AI-tagged via another signal channel).
Unambiguous patterns (``/v1/vector_stores``, ``/vectors/upsert``,
``/collections//points``) fire regardless of ``parent_is_ai``.
"""
if not path:
return False
for pattern, requires_parent_ai in AI_RAG_PATH_PATTERNS:
if pattern.search(path):
if requires_parent_ai and not parent_is_ai:
continue
return True
return False
def is_ai_prompt_param(name: str) -> bool:
"""True if a parameter name is a known prompt-injection vector.
Matches case-insensitively against ``AI_PARAM_NAMES``. The caller is
expected to gate this check on the parent Endpoint already being
AI-classified (``ai_interface_type IS NOT NULL`` and != ``non-llm``);
a parameter named ``text`` on a contact form is not prompt-injectable
in any meaningful sense.
"""
if not name:
return False
return name.strip().lower() in AI_PARAM_NAMES
# ---------------------------------------------------------------------------
# js_recon helpers
# ---------------------------------------------------------------------------
def _redact_secret(value: str) -> str:
"""Return a safe-to-store rendering of an API key for graph storage.
Shows the first 6 and last 4 characters, masking the middle. Avoids
persisting the full credential while still letting an operator confirm
which key matched. Short strings (<14 chars) are masked entirely.
"""
if not value:
return ""
if len(value) < 14:
return "*" * len(value)
return f"{value[:6]}...{value[-4:]}"
def _disambiguate_google_key(
content: str, span: tuple[int, int]
) -> tuple[str, str, str]:
"""Return ``(sdk_name, severity, confidence)`` for an ``AIzaSy*`` match.
Scans ±2KB around the match span for Gemini-specific tokens. When any
are present, the key is a Gemini credential (critical). Otherwise it's
almost certainly a Maps / Firebase / YouTube key — keep the finding but
downgrade severity so operators don't chase a false positive.
"""
start, end = span
window_start = max(0, start - 2048)
window_end = min(len(content), end + 2048)
context = content[window_start:window_end]
for token in _GEMINI_CONTEXT_TOKENS:
if token in context:
return ("Google Gemini API Key", "critical", "high")
return ("Google API Key (likely Maps/Firebase)", "medium", "low")
def match_ai_sdk(content: str, max_bytes: int = 524288) -> list[dict]:
"""Scan a JS blob for AI/LLM signals across four channels.
Returns a list of finding dicts with the shape::
{
"category": "ai-sdk-client" | "ai-sdk-key-literal"
| "ai-sdk-browser-allowed" | "ai-frontend-detected"
| "ai-provider-url",
"sdk_name": "OpenAI" / "Anthropic" / "Open WebUI" / ...,
"severity": "info" | "low" | "medium" | "high" | "critical",
"confidence": "low" | "medium" | "high",
"matched_text": the full matched substring (kept for evidence),
"sample": redacted rendering of captured key, "" otherwise,
"byte_offset": start index of the match in the content,
"detection_method": "ai_sdk_catalogue",
}
Constructor-context key matches suppress overlapping prefix-anchored
matches on the same byte range, so a ``new OpenAI({apiKey:'sk-...'})``
yields a single high-confidence finding instead of two duplicates.
The ``AIzaSy`` Google-key disambiguation rule runs post-match: the helper
scans ±2KB around each Google key for Gemini SDK / endpoint tokens and
only flags as Gemini when at least one is present.
Bodies larger than ``max_bytes`` (default 512KB) are truncated before
scanning. JS bundles past this size are typically minified vendor blobs
that have already been catalogued upstream.
"""
if not content:
return []
if len(content) > max_bytes:
content = content[:max_bytes]
findings: list[dict] = []
claimed_ranges: list[tuple[int, int]] = []
def _record(category: str, sdk_name: str, severity: str, confidence: str,
match: re.Match, *, captured_value: str = "") -> None:
span = match.span()
# Suppress prefix-anchored hits that overlap a constructor-context
# match already recorded for the same span.
for c_start, c_end in claimed_ranges:
if span[0] >= c_start and span[1] <= c_end:
return
findings.append({
"category": category,
"sdk_name": sdk_name,
"severity": severity,
"confidence": confidence,
"matched_text": match.group(0)[:500],
# Full captured secret kept here so callers can dedup against the
# existing Secret taxonomy. NOT persisted as-is — the mixin uses
# this as a Cypher needle but stores only the redacted ``sample``.
"captured_value": captured_value,
"sample": _redact_secret(captured_value) if captured_value else "",
"byte_offset": span[0],
"detection_method": "ai_sdk_catalogue",
})
# 1) Constructor-context key literals run FIRST so they win priority over
# the prefix-anchored fallback patterns below.
for pattern, sdk_name, severity, confidence in AI_KEY_CONSTRUCTOR_PATTERNS:
for match in pattern.finditer(content):
captured = match.group(1) if match.groups() else match.group(0)
_record("ai-sdk-key-literal", sdk_name, severity, confidence,
match, captured_value=captured)
claimed_ranges.append(match.span())
# 2) Prefix-anchored key literals (Google key gets disambiguated).
for pattern, sdk_name, severity, confidence in AI_KEY_PREFIX_PATTERNS:
for match in pattern.finditer(content):
if "AIzaSy" in pattern.pattern:
resolved_sdk, resolved_sev, resolved_conf = (
_disambiguate_google_key(content, match.span())
)
_record("ai-sdk-key-literal", resolved_sdk, resolved_sev,
resolved_conf, match, captured_value=match.group(0))
else:
_record("ai-sdk-key-literal", sdk_name, severity, confidence,
match, captured_value=match.group(0))
# 3) SDK imports.
for pattern, sdk_name, severity, confidence in AI_SDK_IMPORT_PATTERNS:
for match in pattern.finditer(content):
_record("ai-sdk-client", sdk_name, severity, confidence, match)
# 4) Browser-mode opt-in flags.
for pattern, flag_name, severity, confidence in AI_BROWSER_FLAG_PATTERNS:
for match in pattern.finditer(content):
_record("ai-sdk-browser-allowed", flag_name, severity, confidence,
match)
# 5) Frontend product markers in shipped JS.
seen_products: set[str] = set()
for pattern, product, severity, confidence in AI_FRONTEND_JS_PATTERNS:
for match in pattern.finditer(content):
# One frontend finding per product per file — markers cluster.
if product in seen_products:
continue
seen_products.add(product)
_record("ai-frontend-detected", product, severity, confidence,
match)
break
# 6) Provider base URLs / proxy URLs (contextual signals).
seen_urls: set[str] = set()
for pattern, label, severity, confidence in AI_PROVIDER_URL_PATTERNS:
for match in pattern.finditer(content):
if label in seen_urls:
continue
seen_urls.add(label)
_record("ai-provider-url", label, severity, confidence, match)
break
return findings
def resolve_ai_tool_arg_path(spec: dict, dialect: str, param_name: str) -> str | None:
"""Return a JSON Pointer to ``param_name`` inside the tool-schema ``spec``.
Walks the spec using the JSON Pointer prefix registered in
``AI_TOOL_ARG_PATH_DIALECTS`` for the named dialect. Returns the
full pointer (e.g. ``/parameters/properties/query``) on success, or
``None`` if the dialect is unknown, the spec doesn't contain the
expected shape, or the parameter is not in the schema.
This is a placeholder for the Phase-15 ai_surface_recon module — the
spec is only populated once that module discovers an OpenAPI /
ai-plugin.json / MCP tools/list document. Until then the resolver is
effectively a no-op (no spec, no resolution).
"""
if not spec or not dialect or not param_name:
return None
prefix = next((p for d, p in AI_TOOL_ARG_PATH_DIALECTS if d == dialect), None)
if prefix is None:
return None
# Walk the spec by the prefix components and verify the parameter is
# in the resulting properties dict.
cursor: dict | None = spec
for segment in prefix.strip("/").split("/"):
if not isinstance(cursor, dict):
return None
cursor = cursor.get(segment)
if not isinstance(cursor, dict) or param_name not in cursor:
return None
return f"{prefix}/{param_name}"
# ===========================================================================
# Central ai_surface_recon module constants (active probing).
# Consumed only by recon/main_recon_modules/ai_surface_recon.py and its
# partial-recon twin. These drive the active, protocol-aware workloads:
# chat-shape probes, MCP handshake paths, OpenAPI discovery, vector-DB
# confirmation reads, and model-family guessing. All benign; no payloads.
# ===========================================================================
# Canonical chat/completion/SSE paths to POST a 1-token "ping" against when a
# host shows an AI signal but the crawl found no classified chat path.
AI_CHAT_PROBE_PATHS: list[str] = [
"/v1/chat/completions",
"/openai/v1/chat/completions", # Groq prefixes OpenAI paths with /openai
"/v1/completions",
"/v1/fim/completions", # Mistral fill-in-middle (completion-shaped)
"/v1/responses",
"/v1/messages",
"/v2/chat", # Cohere v2 chat
"/v1/sonar", # Perplexity Sonar
"/api/chat",
"/api/generate",
"/chat/completions",
"/completion",
"/generate",
"/generate_stream", # TGI streaming variant
"/invocations", # TGI / SageMaker invoke
"/invoke",
"/stream",
]
# Response-shape classifiers: (family, list-of-required-json-keys (dot paths),
# ai_interface_type). Matched against the parsed JSON body of a chat probe.
# A 401/422 with an OpenAI-style {"error": ...} body still counts as a positive
# OpenAI-compatible detection (handled in code, not here).
AI_CHAT_RESPONSE_SHAPES: list[tuple[str, list[str], str]] = [
("openai", ["choices"], "llm-chat"),
("anthropic", ["content", "stop_reason"], "llm-chat"),
("ollama", ["response"], "llm-completion"),
("ollama-chat", ["message", "done"], "llm-chat"),
("gemini", ["candidates"], "llm-chat"),
("langserve", ["output"], "llm-chat"),
]
# MCP transport probe paths (Streamable HTTP + legacy SSE conventions).
AI_MCP_PROBE_PATHS: list[str] = [
"/mcp",
"/sse",
"/messages",
"/message",
"/api/mcp",
"/mcp/sse",
"/",
]
# OpenAPI / manifest / model-listing discovery GETs.
AI_OPENAPI_DISCOVERY_PATHS: list[str] = [
"/.well-known/ai-plugin.json",
"/openapi.json",
"/swagger.json",
"/v3/api-docs",
"/v1/models",
"/models",
"/api/tags",
"/api/version",
]
# Vector-DB confirmation reads: tech_name -> ordered list of (path, expected
# substring) attempts. A 200 whose body contains the substring (empty == any
# 200) confirms the service; the first endpoint that answers wins. Multiple
# endpoints per DB absorb version/path drift (e.g. Chroma v1 -> v2 heartbeat).
#
# Candidates are unioned from TWO sources in _confirm_vector_dbs:
# - the port catalogue (AI_PORTS entries whose category == "ai-vector-db"), and
# - http_probe body/title fingerprints (info["ai_framework_name"]).
# The second source is what lets a DB on a SHARED port still confirm: Chroma
# (8000, catalogued ai-runtime) and Weaviate (8080, catalogued ai-frontend)
# are only reachable via their http_probe fingerprint, not the port category.
AI_VECTOR_DB_READS: dict[str, list[tuple[str, str]]] = {
# Chroma: heartbeat key is `"nanosecond heartbeat"` on both v1 and v2.
"chroma": [("/api/v2/heartbeat", "heartbeat"),
("/api/v1/heartbeat", "heartbeat"),
("/api/v1/collections", "")],
# Qdrant: root returns {"title":"qdrant - vector search engine",...};
# /collections returns {"result":{...},"status":"ok",...}.
"qdrant": [("/", "qdrant"), ("/collections", "result")],
# Weaviate: /v1/meta always carries a "modules" object; ready is 200-empty.
"weaviate": [("/v1/meta", "modules"), ("/v1/.well-known/ready", "")],
# Milvus: REST surface varies by version/port; /healthz on the metrics
# port answers "OK". (gRPC port 19530 won't speak HTTP — best-effort.)
"milvus": [("/v1/vector/collections", ""), ("/healthz", "")],
}
# Lowercase tokens used to map a model id / family string to a family bucket.
AI_MODEL_FAMILY_TOKENS: list[str] = [
"gpt", "o1", "o3", "claude", "llama", "mistral", "mixtral", "qwen",
"gemini", "gemma", "command-r", "command", "phi", "deepseek", "yi",
"falcon", "vicuna", "codellama", "starcoder", "nous", "openchat",
]
def classify_ai_chat_response(parsed_json: dict) -> str | None:
"""Return an ai_interface_type from a parsed chat-probe JSON body, or None.
Matches the strongest shape in AI_CHAT_RESPONSE_SHAPES (all required
top-level keys present). Ordered by specificity (first match wins).
"""
if not isinstance(parsed_json, dict):
return None
keys = set(parsed_json.keys())
for _family, required, iface in AI_CHAT_RESPONSE_SHAPES:
if all(k in keys for k in required):
return iface
return None
def guess_model_family(model_ids: list[str]) -> str | None:
"""Map a list of model id/family strings to a single family token.
Picks the first AI_MODEL_FAMILY_TOKENS hit across the ids (longest token
first so 'codellama' beats 'llama'). Returns None when nothing matches.
"""
if not model_ids:
return None
blob = " ".join(str(m).lower() for m in model_ids if m)
for token in sorted(AI_MODEL_FAMILY_TOKENS, key=len, reverse=True):
if token in blob:
return token
return None
def pick_tool_dialect(spec: dict) -> str | None:
"""Guess the tool-schema dialect of a single tool spec for resolve_ai_tool_arg_path.
openai-functions -> has a 'function' wrapper or top-level 'parameters'
anthropic-tools -> has 'input_schema'
mcp-tools-list -> has 'inputSchema'
"""
if not isinstance(spec, dict):
return None
if "inputSchema" in spec:
return "mcp-tools-list"
if "input_schema" in spec:
return "anthropic-tools"
if "function" in spec and isinstance(spec.get("function"), dict):
return "openai-functions"
if "parameters" in spec:
return "openai-functions"
return None