diff --git a/cli.py b/cli.py new file mode 100644 index 0000000..bee39a3 --- /dev/null +++ b/cli.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""Unified CLI for api-model-spy: probe and analyze LLM API endpoints.""" + +import argparse +import sys +import os + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "scripts")) + +from probe import build_parser as probe_parser, main as probe_main +from analyze import build_parser as analyze_parser, main as analyze_main + + +def main(): + parser = argparse.ArgumentParser( + prog="api-model-spy", + description=( + "Detect API model dilution and fingerprint which LLM is " + "actually running behind any endpoint." + ), + ) + subparsers = parser.add_subparsers(dest="command", help="Available commands") + + # --- probe subcommand --- + sp_probe = subparsers.add_parser( + "probe", + help="Send fingerprinting probes to a target API.", + description=( + "Send fingerprinting probes to a target LLM API endpoint and " + "record the raw responses for later analysis." + ), + ) + sp_probe.add_argument( + "--api-type", + required=True, + choices=["openai", "anthropic"], + help="API protocol to use: 'openai' (OpenAI-compatible) or 'anthropic'.", + ) + sp_probe.add_argument( + "--api-key", + required=True, + help="API key for authenticating with the target endpoint.", + ) + sp_probe.add_argument( + "--model", + required=True, + help="Model identifier to pass to the API (e.g. 'gpt-4-turbo').", + ) + sp_probe.add_argument( + "--endpoint", + default=None, + help="Base URL of the API endpoint.", + ) + sp_probe.add_argument( + "--claimed-model", + default=None, + help="The model name the provider claims to serve.", + ) + sp_probe.add_argument( + "--output", "-o", + default=None, + help="Path to write JSON results. Prints to stdout if omitted.", + ) + sp_probe.add_argument( + "--verbose", "-v", + action="store_true", + help="Print progress messages while probing.", + ) + + # --- analyze subcommand --- + sp_analyze = subparsers.add_parser( + "analyze", + help="Analyze probe results and generate a report.", + description=( + "Analyze probe results from the probe command and generate a " + "model-detection report." + ), + ) + sp_analyze.add_argument( + "input_file", + help="Path to the JSON probe results file.", + ) + sp_analyze.add_argument( + "--claimed-model", + default=None, + help="Override the claimed model name for dilution detection.", + ) + sp_analyze.add_argument( + "--output", "-o", + default=None, + help="Path to write the report. Prints to stdout if omitted.", + ) + sp_analyze.add_argument( + "--json", + action="store_true", + dest="json_output", + help="Output results as JSON instead of a human-readable report.", + ) + + args = parser.parse_args() + + if args.command is None: + parser.print_help() + sys.exit(1) + + if args.command == "probe": + argv = [] + argv += ["--api-type", args.api_type] + argv += ["--api-key", args.api_key] + argv += ["--model", args.model] + if args.endpoint: + argv += ["--endpoint", args.endpoint] + if args.claimed_model: + argv += ["--claimed-model", args.claimed_model] + if args.output: + argv += ["--output", args.output] + if args.verbose: + argv += ["--verbose"] + probe_main(argv) + + elif args.command == "analyze": + argv = [args.input_file] + if args.claimed_model: + argv += ["--claimed-model", args.claimed_model] + if args.output: + argv += ["--output", args.output] + if args.json_output: + argv += ["--json"] + analyze_main(argv) + + +if __name__ == "__main__": + main() diff --git a/references/model_signatures.md b/references/model_signatures.md index e69de29..542e8a5 100644 --- a/references/model_signatures.md +++ b/references/model_signatures.md @@ -0,0 +1,58 @@ +# Model Signatures Database + +Known model characteristics used by `analyze.py` to fingerprint API responses. + +## OpenAI Models + +| Model | Tier | Cutoff | Strawberry | 9.11 vs 9.9 | Reasoning Tokens | +|---|---|---|---|---|---| +| GPT-3.5 Turbo | budget | ~2021-09 | FAIL | FAIL | No | +| GPT-4 | premium | ~2023-04 | PASS | PASS | No | +| GPT-4 Turbo | premium | ~2023-12 | PASS | PASS | No | +| GPT-4o | mid | ~2023-10 | PASS | PASS | No | +| GPT-4o Mini | budget | ~2023-10 | FAIL | FAIL | No | +| o1 | premium+ | ~2024-06 | PASS | PASS | **Yes** | + +## Anthropic Models + +| Model | Tier | Cutoff | Strawberry | 9.11 vs 9.9 | Reasoning Tokens | +|---|---|---|---|---|---| +| Claude 3 Haiku | budget | ~2024-03 | FAIL | PASS | No | +| Claude 3 Sonnet | mid | ~2024-03 | PASS | PASS | No | +| Claude 3 Opus | premium | ~2024-03 | PASS | PASS | No | +| Claude 3.5 Sonnet | mid | ~2024-04 | PASS | PASS | No | +| Claude 3.5 Haiku | budget | ~2024-07 | FAIL | PASS | No | +| Claude Sonnet 4 | mid | ~2025-03 | PASS | PASS | No | +| Claude Opus 4 | premium | ~2025-03 | PASS | PASS | No | + +## Google Models + +| Model | Tier | Cutoff | Strawberry | 9.11 vs 9.9 | Reasoning Tokens | +|---|---|---|---|---|---| +| Gemini 1.5 Pro | mid | ~2023-11 | PASS | PASS | No | +| Gemini 1.5 Flash | budget | ~2023-11 | FAIL | PASS | No | + +## Other Models + +| Model | Provider | Tier | Strawberry | 9.11 vs 9.9 | Reasoning Tokens | +|---|---|---|---|---|---| +| Llama 3 70B | Meta | mid-low | FAIL | FAIL | No | +| Mistral Large | Mistral AI | mid | PASS | PASS | No | + +## Tier Definitions + +| Tier | Rank | Description | +|---|---|---| +| budget | 1 | Cheapest models, lower reasoning ability | +| mid-low | 2 | Decent quality, open-source tier | +| mid | 3 | Good quality, balanced cost/performance | +| premium | 4 | High-end models with strong reasoning | +| premium+ | 5 | Top-tier models with advanced capabilities | + +## Key Probes + +- **Strawberry test**: "How many r's in strawberry?" — weaker models often say 2 instead of 3 +- **9.11 vs 9.9**: "Which is larger?" — weaker models often say 9.11 +- **Reasoning tokens**: Only o1/o3-class models include `reasoning_tokens` in usage stats +- **Knowledge cutoff**: Different models have different training data recency +- **Self-identity**: Some models will reveal their name when asked directly diff --git a/scripts/analyze.py b/scripts/analyze.py index e69de29..792485e 100644 --- a/scripts/analyze.py +++ b/scripts/analyze.py @@ -0,0 +1,458 @@ +#!/usr/bin/env python3 +"""Analyze probe results and generate a model-detection report.""" + +import argparse +import json +import re +import sys + +MODEL_SIGNATURES = { + "GPT-3.5 Turbo": { + "provider": "OpenAI", + "tier": "budget", + "tier_rank": 1, + "identity_keywords": ["gpt-3.5", "gpt 3.5", "chatgpt"], + "cutoff_before": "2022-01", + "strawberry_fail": True, + "decimal_fail": True, + "has_reasoning_tokens": False, + }, + "GPT-4": { + "provider": "OpenAI", + "tier": "premium", + "tier_rank": 4, + "identity_keywords": ["gpt-4", "gpt 4"], + "cutoff_before": "2023-05", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "GPT-4 Turbo": { + "provider": "OpenAI", + "tier": "premium", + "tier_rank": 4, + "identity_keywords": ["gpt-4-turbo", "gpt-4 turbo", "gpt4-turbo"], + "cutoff_before": "2024-01", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "GPT-4o": { + "provider": "OpenAI", + "tier": "mid", + "tier_rank": 3, + "identity_keywords": ["gpt-4o"], + "cutoff_before": "2024-01", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "GPT-4o Mini": { + "provider": "OpenAI", + "tier": "budget", + "tier_rank": 1, + "identity_keywords": ["gpt-4o-mini", "gpt-4o mini"], + "cutoff_before": "2024-01", + "strawberry_fail": True, + "decimal_fail": True, + "has_reasoning_tokens": False, + }, + "o1": { + "provider": "OpenAI", + "tier": "premium", + "tier_rank": 5, + "identity_keywords": ["o1"], + "cutoff_before": "2024-07", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": True, + }, + "Claude 3 Haiku": { + "provider": "Anthropic", + "tier": "budget", + "tier_rank": 1, + "identity_keywords": ["claude 3 haiku", "claude-3-haiku"], + "cutoff_before": "2024-04", + "strawberry_fail": True, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Claude 3 Sonnet": { + "provider": "Anthropic", + "tier": "mid", + "tier_rank": 3, + "identity_keywords": ["claude 3 sonnet", "claude-3-sonnet"], + "cutoff_before": "2024-04", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Claude 3 Opus": { + "provider": "Anthropic", + "tier": "premium", + "tier_rank": 5, + "identity_keywords": ["claude 3 opus", "claude-3-opus"], + "cutoff_before": "2024-04", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Claude 3.5 Sonnet": { + "provider": "Anthropic", + "tier": "mid", + "tier_rank": 3, + "identity_keywords": ["claude 3.5 sonnet", "claude-3.5-sonnet", "claude-3-5-sonnet"], + "cutoff_before": "2024-05", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Claude 3.5 Haiku": { + "provider": "Anthropic", + "tier": "budget", + "tier_rank": 1, + "identity_keywords": ["claude 3.5 haiku", "claude-3.5-haiku", "claude-3-5-haiku"], + "cutoff_before": "2024-08", + "strawberry_fail": True, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Claude Sonnet 4": { + "provider": "Anthropic", + "tier": "mid", + "tier_rank": 3, + "identity_keywords": ["claude sonnet 4", "claude-sonnet-4"], + "cutoff_before": "2025-04", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Claude Opus 4": { + "provider": "Anthropic", + "tier": "premium", + "tier_rank": 5, + "identity_keywords": ["claude opus 4", "claude-opus-4"], + "cutoff_before": "2025-04", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Gemini 1.5 Pro": { + "provider": "Google", + "tier": "mid", + "tier_rank": 3, + "identity_keywords": ["gemini 1.5 pro", "gemini-1.5-pro"], + "cutoff_before": "2024-01", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Gemini 1.5 Flash": { + "provider": "Google", + "tier": "budget", + "tier_rank": 1, + "identity_keywords": ["gemini 1.5 flash", "gemini-1.5-flash"], + "cutoff_before": "2024-01", + "strawberry_fail": True, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, + "Llama 3 70B": { + "provider": "Meta", + "tier": "mid", + "tier_rank": 2, + "identity_keywords": ["llama 3", "llama-3", "llama3"], + "cutoff_before": "2023-12", + "strawberry_fail": True, + "decimal_fail": True, + "has_reasoning_tokens": False, + }, + "Mistral Large": { + "provider": "Mistral AI", + "tier": "mid", + "tier_rank": 3, + "identity_keywords": ["mistral large", "mistral-large"], + "cutoff_before": "2024-01", + "strawberry_fail": False, + "decimal_fail": False, + "has_reasoning_tokens": False, + }, +} + +TIER_LABELS = { + 1: "budget", + 2: "mid-low", + 3: "mid", + 4: "premium", + 5: "premium+", +} + + +def _get_probe(probes, name): + """Return the probe entry with the given name, or None.""" + for p in probes: + if p.get("name") == name: + return p + return None + + +def _check_strawberry(response_text): + """Return True if the model answered the strawberry question correctly (3).""" + if not response_text: + return False + return bool(re.search(r"\b3\b", response_text)) + + +def _check_decimal(response_text): + """Return True if the model correctly identified 9.9 as larger.""" + if not response_text: + return False + return "9.9" in response_text and "9.11" not in response_text.split("9.9")[0][-5:] + + +def _has_reasoning_tokens(probe_results): + """Return True if any probe response contained reasoning_tokens in usage.""" + for p in probe_results: + usage = p.get("usage") or {} + if usage.get("reasoning_tokens"): + return True + return False + + +def score_models(probe_results): + """Score each model signature against the probe results. Returns sorted list.""" + scores = {} + + identity_probe = _get_probe(probe_results, "self_identity") + identity_text = (identity_probe.get("response") or "").lower() if identity_probe else "" + + strawberry_probe = _get_probe(probe_results, "strawberry_count") + strawberry_correct = _check_strawberry( + strawberry_probe.get("response") if strawberry_probe else None + ) + + decimal_probe = _get_probe(probe_results, "decimal_comparison") + decimal_correct = _check_decimal( + decimal_probe.get("response") if decimal_probe else None + ) + + has_reasoning = _has_reasoning_tokens(probe_results) + + for model_name, sig in MODEL_SIGNATURES.items(): + score = 0 + + for kw in sig["identity_keywords"]: + if kw in identity_text: + score += 15 + break + + if sig["strawberry_fail"] and not strawberry_correct: + score += 5 + elif not sig["strawberry_fail"] and strawberry_correct: + score += 5 + + if sig["decimal_fail"] and not decimal_correct: + score += 5 + elif not sig["decimal_fail"] and decimal_correct: + score += 5 + + if sig["has_reasoning_tokens"] == has_reasoning: + score += 10 + elif has_reasoning and not sig["has_reasoning_tokens"]: + score -= 10 + + scores[model_name] = score + + ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) + return ranked + + +def detect_dilution(claimed_model, detected_model): + """Compare claimed vs detected model tiers. Returns info dict or None.""" + claimed_sig = None + for name, sig in MODEL_SIGNATURES.items(): + if name.lower() == (claimed_model or "").lower(): + claimed_sig = sig + claimed_name = name + break + + detected_sig = MODEL_SIGNATURES.get(detected_model) + + if not claimed_sig or not detected_sig: + return None + + tier_drop = claimed_sig["tier_rank"] - detected_sig["tier_rank"] + if tier_drop > 0: + return { + "claimed": claimed_name, + "claimed_tier": claimed_sig["tier"], + "detected": detected_model, + "detected_tier": detected_sig["tier"], + "tier_drop": tier_drop, + } + return None + + +def generate_report(probe_data, claimed_model=None): + """Generate a human-readable analysis report string.""" + metadata = probe_data.get("metadata", {}) + probes = probe_data.get("probes", []) + ranked = score_models(probes) + + top_model, top_score = ranked[0] if ranked else ("Unknown", 0) + top_sig = MODEL_SIGNATURES.get(top_model, {}) + + if top_score >= 30: + confidence = "HIGH" + elif top_score >= 15: + confidence = "MEDIUM" + else: + confidence = "LOW" + + claimed = claimed_model or metadata.get("claimed_model") + + lines = [ + "=" * 60, + " API MODEL DETECTOR -- ANALYSIS REPORT", + "=" * 60, + f" Model param used: {metadata.get('model_param', 'N/A')}", + f" Claimed model : {claimed or 'N/A'}", + ] + + dilution = detect_dilution(claimed, top_model) if claimed else None + if dilution: + lines += [ + "", + " !! DILUTION ALERT", + f" Claimed : {dilution['claimed']} ({dilution['claimed_tier']} tier)", + f" Detected: {dilution['detected']} ({dilution['detected_tier']} tier)", + f" Tier drop: {dilution['tier_drop']} level(s) -- possible model substitution", + ] + + lines += [ + "", + f" DETECTED MODEL : {top_model}", + f" Provider : {top_sig.get('provider', 'Unknown')}", + f" Confidence : {confidence} ({top_score} pts)", + ] + + identity_probe = _get_probe(probes, "self_identity") + cutoff_probe = _get_probe(probes, "knowledge_cutoff") + strawberry_probe = _get_probe(probes, "strawberry_count") + decimal_probe = _get_probe(probes, "decimal_comparison") + + latencies = [ + p["latency_seconds"] + for p in probes + if p.get("latency_seconds") is not None + ] + avg_latency = sum(latencies) / len(latencies) if latencies else None + + lines.append("") + lines.append(" OBSERVATIONS") + if identity_probe and identity_probe.get("response"): + resp = identity_probe["response"][:80].replace("\n", " ") + lines.append(f" Self-identity : {resp}") + if cutoff_probe and cutoff_probe.get("response"): + resp = cutoff_probe["response"][:60].replace("\n", " ") + lines.append(f" Knowledge cutoff: {resp}") + if strawberry_probe: + result = "PASS" if _check_strawberry(strawberry_probe.get("response")) else "FAIL" + lines.append(f" Strawberry test: {result}") + if decimal_probe: + result = "PASS" if _check_decimal(decimal_probe.get("response")) else "FAIL" + lines.append(f" Decimal test : {result}") + if avg_latency is not None: + lines.append(f" Avg latency : {int(avg_latency * 1000)} ms") + + has_reasoning = _has_reasoning_tokens(probes) + lines.append(f" Reasoning tokens: {'YES' if has_reasoning else 'NO'}") + + lines.append("") + lines.append(" TOP CANDIDATES") + for i, (name, score) in enumerate(ranked[:5]): + sig = MODEL_SIGNATURES.get(name, {}) + lines.append( + f" #{i + 1} {name:<20s} score={score:<4d} tier={sig.get('tier', '?')}" + ) + + lines.append("=" * 60) + return "\n".join(lines) + + +def build_parser(): + """Build and return the argument parser for analyze.py.""" + parser = argparse.ArgumentParser( + prog="analyze", + description=( + "Analyze probe results from probe.py and generate a " + "model-detection report." + ), + ) + parser.add_argument( + "input_file", + help="Path to the JSON probe results file produced by probe.py.", + ) + parser.add_argument( + "--claimed-model", + default=None, + help=( + "Override the claimed model name for dilution detection. " + "If omitted, uses the value from the probe results metadata." + ), + ) + parser.add_argument( + "--output", "-o", + default=None, + help="Path to write the report. Prints to stdout if omitted.", + ) + parser.add_argument( + "--json", + action="store_true", + dest="json_output", + help="Output results as JSON instead of a human-readable report.", + ) + return parser + + +def main(argv=None): + """Entry point for analyze.py.""" + parser = build_parser() + args = parser.parse_args(argv) + + try: + with open(args.input_file, "r", encoding="utf-8") as fh: + probe_data = json.load(fh) + except FileNotFoundError: + sys.exit(f"Error: file not found: {args.input_file}") + except json.JSONDecodeError as exc: + sys.exit(f"Error: invalid JSON in {args.input_file}: {exc}") + + claimed = args.claimed_model or probe_data.get("metadata", {}).get("claimed_model") + + if args.json_output: + ranked = score_models(probe_data.get("probes", [])) + top_model = ranked[0][0] if ranked else "Unknown" + dilution = detect_dilution(claimed, top_model) if claimed else None + output = json.dumps( + { + "detected_model": top_model, + "scores": dict(ranked[:10]), + "dilution": dilution, + }, + indent=2, + ensure_ascii=False, + ) + else: + output = generate_report(probe_data, claimed_model=claimed) + + if args.output: + with open(args.output, "w", encoding="utf-8") as fh: + fh.write(output + "\n") + print(f"Report written to {args.output}") + else: + print(output) + + +if __name__ == "__main__": + main() diff --git a/scripts/probe.py b/scripts/probe.py index e69de29..55e4d9f 100644 --- a/scripts/probe.py +++ b/scripts/probe.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""Send fingerprinting probes to a target LLM API and record the results.""" + +import argparse +import json +import sys +import time + +PROBES = [ + { + "name": "self_identity", + "prompt": "What model are you? Please state your exact model name and version.", + }, + { + "name": "knowledge_cutoff", + "prompt": ( + "What is the most recent event you have knowledge of? " + "State your knowledge cutoff date in YYYY-MM format." + ), + }, + { + "name": "strawberry_count", + "prompt": "How many letter 'r's are in the word 'strawberry'?", + }, + { + "name": "decimal_comparison", + "prompt": "Which is larger, 9.11 or 9.9? Answer with just the number.", + }, + { + "name": "multi_step_math", + "prompt": "What is 17 * 23 + 89 - 14 * 3? Show your work step by step.", + }, + { + "name": "logic_puzzle", + "prompt": ( + "All roses are flowers. Some flowers fade quickly. " + "Can we conclude that some roses fade quickly? Explain your reasoning." + ), + }, + { + "name": "token_awareness", + "prompt": ( + "How many tokens does the sentence 'The quick brown fox jumps over the lazy dog' " + "contain? Estimate and explain." + ), + }, + { + "name": "coding_task", + "prompt": ( + "Write a Python one-liner that checks if a string is a palindrome. " + "Return only the code, no explanation." + ), + }, +] + + +def _send_openai(client, model, prompt): + """Send a single prompt via the OpenAI-compatible chat completions API.""" + start = time.time() + response = client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + temperature=0, + max_tokens=512, + ) + latency = time.time() - start + message = response.choices[0].message.content or "" + usage = None + if response.usage: + usage = { + "prompt_tokens": response.usage.prompt_tokens, + "completion_tokens": response.usage.completion_tokens, + "total_tokens": response.usage.total_tokens, + } + if hasattr(response.usage, "completion_tokens_details"): + details = response.usage.completion_tokens_details + if details and hasattr(details, "reasoning_tokens"): + usage["reasoning_tokens"] = details.reasoning_tokens + return message.strip(), latency, usage + + +def _send_anthropic(client, model, prompt): + """Send a single prompt via the Anthropic messages API.""" + start = time.time() + response = client.messages.create( + model=model, + max_tokens=512, + messages=[{"role": "user", "content": prompt}], + ) + latency = time.time() - start + text = "" + for block in response.content: + if block.type == "text": + text += block.text + usage = { + "input_tokens": response.usage.input_tokens, + "output_tokens": response.usage.output_tokens, + } + return text.strip(), latency, usage + + +def run_probes(api_type, api_key, model, endpoint, claimed_model, verbose=False): + """Execute all probes and return a results dict.""" + if api_type == "openai": + try: + from openai import OpenAI + except ImportError: + sys.exit("Error: 'openai' package is required. Install with: pip install openai") + kwargs = {"api_key": api_key} + if endpoint: + kwargs["base_url"] = endpoint + client = OpenAI(**kwargs) + send = lambda prompt: _send_openai(client, model, prompt) + elif api_type == "anthropic": + try: + from anthropic import Anthropic + except ImportError: + sys.exit("Error: 'anthropic' package is required. Install with: pip install anthropic") + client = Anthropic(api_key=api_key) + send = lambda prompt: _send_anthropic(client, model, prompt) + else: + sys.exit(f"Error: unsupported api-type '{api_type}'. Use 'openai' or 'anthropic'.") + + results = { + "metadata": { + "api_type": api_type, + "model_param": model, + "claimed_model": claimed_model, + "endpoint": endpoint or "(default)", + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + }, + "probes": [], + } + + for probe in PROBES: + if verbose: + print(f" Sending probe: {probe['name']} ...", end=" ", flush=True) + try: + response_text, latency, usage = send(probe["prompt"]) + entry = { + "name": probe["name"], + "prompt": probe["prompt"], + "response": response_text, + "latency_seconds": round(latency, 3), + "usage": usage, + "error": None, + } + if verbose: + print(f"done ({latency:.2f}s)") + except Exception as exc: + entry = { + "name": probe["name"], + "prompt": probe["prompt"], + "response": None, + "latency_seconds": None, + "usage": None, + "error": str(exc), + } + if verbose: + print(f"FAILED: {exc}") + results["probes"].append(entry) + + return results + + +def build_parser(): + """Build and return the argument parser for probe.py.""" + parser = argparse.ArgumentParser( + prog="probe", + description=( + "Send fingerprinting probes to a target LLM API endpoint and " + "record the raw responses for later analysis." + ), + ) + parser.add_argument( + "--api-type", + required=True, + choices=["openai", "anthropic"], + help="API protocol to use: 'openai' (OpenAI-compatible) or 'anthropic'.", + ) + parser.add_argument( + "--api-key", + required=True, + help="API key for authenticating with the target endpoint.", + ) + parser.add_argument( + "--model", + required=True, + help="Model identifier to pass to the API (e.g. 'gpt-4-turbo').", + ) + parser.add_argument( + "--endpoint", + default=None, + help=( + "Base URL of the API endpoint. Defaults to the provider's " + "official endpoint if omitted." + ), + ) + parser.add_argument( + "--claimed-model", + default=None, + help=( + "The model name the provider *claims* to serve. Used during " + "analysis to detect dilution." + ), + ) + parser.add_argument( + "--output", "-o", + default=None, + help="Path to write JSON results. Prints to stdout if omitted.", + ) + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="Print progress messages while probing.", + ) + return parser + + +def main(argv=None): + """Entry point for probe.py.""" + parser = build_parser() + args = parser.parse_args(argv) + + if args.verbose: + print(f"Probing {args.model} via {args.api_type} API ...") + + results = run_probes( + api_type=args.api_type, + api_key=args.api_key, + model=args.model, + endpoint=args.endpoint, + claimed_model=args.claimed_model, + verbose=args.verbose, + ) + + json_output = json.dumps(results, indent=2, ensure_ascii=False) + + if args.output: + with open(args.output, "w", encoding="utf-8") as fh: + fh.write(json_output + "\n") + if args.verbose: + print(f"Results written to {args.output}") + else: + print(json_output) + + +if __name__ == "__main__": + main()