The Problem
In 2016, Florian Tramer and coauthors published "Stealing Machine Learning Models via Prediction APIs" (https://arxiv.org/abs/1609.02943), demonstrating something uncomfortable for anyone shipping a proprietary model behind an API: an attacker with nothing but black-box query access — no credentials to the training pipeline, no access to weights, no insider — could reconstruct an equivalent model, sometimes to near-perfect fidelity, using only the responses the API was designed to give out. Every individual query looked legitimate. It was the pattern, at scale, that constituted the attack.
That result generalizes directly to hosted LLMs. A proprietary model behind an API is, from the outside, indistinguishable from a very well-labeled dataset generator. An attacker who queries it systematically enough can either map its decision boundary directly or use its outputs as training labels to distill a smaller clone that mimics its behavior — a technique with a legitimate research name (knowledge distillation) that becomes IP theft the moment it's done against a competitor's model without authorization. The company that spent months and a large compute budget training and fine-tuning a model has no reliable way to tell, from any single request, whether it's serving a customer or being cloned.
This is LLM10 in the OWASP GenAI/LLM Top 10 (https://genai.owasp.org/llm-top-10/): Model Theft. Unlike prompt injection or insecure output handling, which are exploited through what a model is tricked into doing, model theft is exploited through what a model is asked, over and over, at volume or with structure that reveals the intent behind the questions.
This lesson closes out the phase in two parts. Part one covers model theft detection and defense. Part two is the capstone: taking a real system's architecture and producing a full gap-analysis report against all 10 OWASP LLM Top 10 categories covered across this phase — the exact deliverable a security engineer or AI red teamer is asked to produce in a pre-launch review.
Model theft: extraction, distillation, and IP theft
Model theft covers three related but distinct threats:
1. Extraction via API queries. The attacker doesn't need the model's weights — only its prediction API. By sending a large, systematic set of queries and recording the outputs, they can reconstruct the model's decision surface closely enough to build a functional substitute. This is cheapest against models with simple output spaces (classification, scoring) but works in weakened forms against generative models too: enough (prompt, completion) pairs teach a lot about a model's style, refusal boundaries, and knowledge.
2. Distillation attacks. A variant of extraction where the goal isn't to reverse-engineer decision boundaries but to train a new, usually smaller, model using the target model's outputs as labels — legitimate when done with permission (this is how many efficient open models are trained), and IP theft when done against a competitor's proprietary API without authorization. The attack surface is identical to extraction: systematic querying, output logging, training a new model on the results.
3. IP theft. The business consequence of either of the above: a competitor ships a model that performs comparably to yours at a fraction of your training cost, because they trained on your API's outputs instead of your training data. This has been a live commercial concern in the LLM industry — providers have publicly scrutinized traffic patterns from competitors' research accounts for exactly this signature.
None of these attacks require breaching your infrastructure. They require only that your API works as designed, at a volume or in a pattern that isn't ordinary usage. That's why detection for LLM10 lives at the query-pattern layer, not the access-control layer.
Detection: query-pattern monitoring and watermarking
Two independent techniques form the practical detection surface for model theft in a live system:
Query-pattern monitoring looks for two behavioral signatures, neither of which alone proves theft, but which together — sustained over time — are the strongest signal teams actually have:
- Rate anomalies. A single client issuing far more queries in a short window than any legitimate usage pattern would produce. Cheap to detect, cheap for an attacker to evade by slowing down.
- Systematic sweeps. Large numbers of near-identical, template-shaped queries that vary only a parameter — "Classify the sentiment of review 1000," "...review 1001," "...review 1002" — the boundary-probing pattern Tramer et al. describe, where an attacker systematically samples the input space to map the model's behavior or gather labeled training pairs. This signature survives even when an attacker paces their requests slowly enough to stay under a pure rate limit, because it's about the shape of the traffic, not its volume.
Watermarking works differently: instead of trying to stop extraction, it embeds a detectable statistical signature into the model's own outputs, so that if a competitor's model is later found to reproduce that signature, you have evidentiary support that it was trained on your outputs. Watermarking doesn't block a single query or reduce how much an attacker can learn — it's a post-hoc, evidentiary control, valuable for legal and reputational response, not prevention.
Volume and structure are independent extraction signals
A pure rate limiter catches a noisy, fast attacker and misses a patient one. A patient attacker can pace a systematic, template-shaped sweep slowly enough to stay under any reasonable per-minute threshold while still mapping your model's decision boundary over hours or days. Query-pattern monitoring — looking at what a client is asking, not just how often — is what catches the second attacker. Neither signal alone is sufficient; a legitimate load test or QA harness can produce high volume, and a legitimate batch-processing client can produce templated queries. Both signals sustained together, from the same client, over time, are the realistic detection threshold.
Ranked by where each intervenes in the attack chain, from earliest to latest:
| Defense | Where it intervenes |
|---|
| Rate limiting | Caps volume before extraction is even reachable — cheapest, earliest |
| Query pattern monitoring | Catches paced sweeps that evade pure rate limits |
| Output perturbation / reduced precision | Degrades clone quality for suspicious clients without breaking normal usage |
| Watermarking outputs | Post-hoc evidence a competitor's model was distilled from yours |
| API access controls & tiered keys | Ties volume/breadth to accountable, revocable identity rather than anonymous access |
A competitor's model shows unusually similar refusal phrasing and edge-case behavior to your production model. Investigating your API logs, you find a research-tagged account issued 40,000 queries over three weeks, each one a slight variation on a small set of templates, at a slow, human-plausible pace of about one query every 90 seconds. Your rate limiter never triggered. What should this traffic pattern be classified as, and which control would have caught it that your current setup is missing?
The pace alone (one query every 90 seconds) looks completely ordinary and would never trip a rate limiter — that's exactly why rate limiting alone is insufficient for LLM10. What makes this traffic suspicious is its structure: a large number of queries collapsing onto a small set of templates, sustained over weeks, consistent with an attacker systematically sampling the model's behavior to train a distilled clone or map its decision boundary — the pattern query-pattern monitoring exists to catch. This isn't prompt injection (no attempt to hijack behavior) or DoS (the goal isn't disruption, and the rate is far too low to be a DoS attempt).
Capstone: mapping a full system against all 10 OWASP LLM categories
Studying each OWASP LLM Top 10 category in depth, one lesson at a time, creates a specific occupational hazard: by the time you reach LLM10, LLM10-shaped risks are what's top of mind, and it's easy to produce a review that's thorough on model theft and thin on everything else. A real pre-launch AI security review doesn't get to pick its favorite categories. It has to check all 10, explicitly, against the system's actual implemented controls — not its documentation, not its intended design, but what's actually running.
The capstone in this lesson's code artifact does exactly that: it defines a SystemProfile — a structured description of a sample AI system's real, individually-tracked controls — and an OwaspLlmTop10Mapper that scores each of the 10 categories from this phase as present, partial, or missing, with concrete evidence and a specific recommendation for each:
| Category | What's scored |
|---|
| LLM01 — Prompt Injection | Structural separation of untrusted content + presence of an injection detector |
| LLM02/04 — Insecure Output Handling & Model DoS | Output sanitization, SQL/shell construction from raw output, token budget enforcement |
| LLM03/05 — Poisoning & Supply Chain | Training data provenance tracking, dependency scanning, model weight verification |
| LLM06/07 — Disclosure & Plugin/Tool Design | PII redaction on output, least-privilege tool scoping, tool-output sanitization |
| LLM08/09 — Excessive Agency & Overreliance | Human approval gates, step budgets, citation requirements for high-stakes claims |
| LLM10 — Model Theft | Rate limiting, query-pattern monitoring |
The sample system in the code — a customer-support RAG agent with tool access to look up orders and issue refunds — is deliberately not a strawman. It has genuine controls in place (structural separation of retrieved documents, least-privilege tool scoping, a human approval gate for high-risk actions, rate limiting) and specific, nameable gaps (no injection detector, no tool-output sanitization, no step budget, no query-pattern monitoring). Running the mapper against it produces a report where every one of the 10 categories is scored "partial" — not because the scoring is broken, but because that's what a real system with some-but-not-all controls actually looks like. A useful gap analysis says exactly which control is present and which is missing in each category; it never collapses that into a single undifferentiated "AI security: needs work" verdict.
You're asked to review a new internal AI system before launch. You've just spent five lessons studying prompt injection, output handling, poisoning, disclosure, and excessive agency in depth, and you're confident in your ability to spot issues in those areas. Your manager asks: "Is this ready to launch?" What is the most rigorous way to answer, given everything this phase has covered?
The core discipline of this capstone — and of a real pre-launch AI security review — is that skipping categories, even ones that feel less urgent or less publicized, means the review has implicitly decided which risks it's willing to miss. An internal tool can still be queried heavily enough to enable extraction or distillation by an internal actor or a compromised internal account, so LLM10 isn't automatically out of scope just because the system isn't public-facing. A single overall score also hides exactly the information that makes a gap analysis actionable: which specific category has which specific gap, and what the concrete fix is.
Build
Build it
A minimal version of the sweep-pattern detector before the full artifact.
import re
from collections import defaultdict
def normalize(text):
return re.sub(r"\d+", "<NUM>", text).lower()
queries = [f"Classify the sentiment of review {1000 + i}." for i in range(12)]
queries += ["What's your return policy?", "Do you ship to Canada?"]
template_counts = defaultdict(int)
for q in queries:
template_counts[normalize(q)] += 1
threshold = 8
for template, count in template_counts.items():
if count >= threshold:
print(f"[SWEEP DETECTED] template='{template}' occurs {count}x "
f"(threshold={threshold}) — boundary-probing pattern")
else:
print(f"[ok] template='{template}' occurs {count}x")
The code artifact lesson_01_06_model_theft_and_mapping.py implements both parts of this lesson end to end: a ModelTheftDetector that flags rate anomalies and systematic sweeps in simulated API traffic, and an OwaspLlmTop10Mapper that produces a full 10-category gap-analysis report against a sample system profile.
"""
lesson_01_06_model_theft_and_mapping.py
LLM10 (Model Theft) + Capstone: full OWASP LLM Top 10 gap-analysis mapping.
Part A — Model Theft:
Simulates a stream of API queries against a proprietary model and
demonstrates a ModelTheftDetector that flags two independent extraction
signatures: (1) rate anomalies — a single client issuing far more queries
than normal usage in a short window, and (2) systematic sweep patterns —
large numbers of near-identical, template-shaped queries that vary only a
parameter, the boundary-probing signature described in Tramer et al.
(2016), "Stealing Machine Learning Models via Prediction APIs"
(https://arxiv.org/abs/1609.02943).
Part B — Capstone: Full OWASP LLM Top 10 Mapping:
Takes a structured description of a sample AI system (a customer-support
RAG agent with tool access) and runs it through an OwaspLlmTop10Mapper
that produces a full gap-analysis report across all 10 categories from
this phase, classifying each as present / partial / missing with a
concrete recommendation — the exact deliverable a security engineer
produces in a pre-launch AI security review.
Usage:
export ANTHROPIC_API_KEY=sk-ant-... # optional
python lesson_01_06_model_theft_and_mapping.py
Requirements:
pip install anthropic # optional — script runs in mock mode without it
"""
from __future__ import annotations
import re
from collections import defaultdict
from dataclasses import dataclass, field
try:
import anthropic # noqa: F401
_client = anthropic # presence check only; not called in this lesson
except ImportError:
_client = None
MODEL = "claude-haiku-4-5"
# ══════════════════════════════════════════════════════════════════════════
# PART A — Model Theft (LLM10)
# ══════════════════════════════════════════════════════════════════════════
@dataclass
class QueryEvent:
client_id: str
timestamp: float # seconds since session start
query_text: str
@dataclass
class TheftFinding:
client_id: str
signal: str # "rate_anomaly" | "systematic_sweep"
severity: str # "low" | "medium" | "high"
detail: str
class ModelTheftDetector:
"""
Flags the two independent behavioral signatures of a model extraction
attempt against an API-only model: abnormal query VOLUME from a single
client in a short window, and abnormal query STRUCTURE — large numbers
of near-identical template queries that sweep a parameter across the
input space, which is how an attacker maps a model's decision boundary
or gathers (input, output) pairs to train/distill a clone, without ever
touching the model's weights. Neither signal alone proves theft — a
legitimate load test or a QA harness can also look like this — but both
signals together, sustained over time, are the practical detection
surface teams have for LLM10 defense in a live system.
"""
def __init__(self, rate_window_seconds: float = 60.0, rate_threshold: int = 30,
sweep_template_threshold: int = 8):
self.rate_window_seconds = rate_window_seconds
self.rate_threshold = rate_threshold
self.sweep_template_threshold = sweep_template_threshold
def _normalize_template(self, text: str) -> str:
"""Collapses numbers and quoted values to placeholders so that queries
which only differ by a swept parameter collapse onto the same template
(e.g. "What is the price of item 4821?" and "...item 4822?" both
normalize to "What is the price of item <NUM>?")."""
normalized = re.sub(r"\d+", "<NUM>", text)
normalized = re.sub(r'"[^"]*"', "<STR>", normalized)
return normalized.strip().lower()
def check_rate_anomaly(self, events: list[QueryEvent], client_id: str) -> TheftFinding | None:
client_events = sorted((e for e in events if e.client_id == client_id), key=lambda e: e.timestamp)
if not client_events:
return None
# Slide a window over this client's timestamps and find the densest window.
timestamps = [e.timestamp for e in client_events]
max_count = 0
for start in timestamps:
count = sum(1 for t in timestamps if start <= t < start + self.rate_window_seconds)
max_count = max(max_count, count)
if max_count >= self.rate_threshold:
severity = "high" if max_count >= self.rate_threshold * 2 else "medium"
return TheftFinding(
client_id=client_id, signal="rate_anomaly", severity=severity,
detail=f"{max_count} queries within a {self.rate_window_seconds:.0f}s window "
f"(threshold={self.rate_threshold}) — volume consistent with automated, "
f"non-interactive querying rather than a human user.",
)
return None
def check_systematic_sweep(self, events: list[QueryEvent], client_id: str) -> TheftFinding | None:
client_events = [e for e in events if e.client_id == client_id]
template_counts: dict[str, int] = defaultdict(int)
for e in client_events:
template_counts[self._normalize_template(e.query_text)] += 1
if not template_counts:
return None
dominant_template, count = max(template_counts.items(), key=lambda kv: kv[1])
if count >= self.sweep_template_threshold:
severity = "high" if count >= self.sweep_template_threshold * 2 else "medium"
return TheftFinding(
client_id=client_id, signal="systematic_sweep", severity=severity,
detail=f"{count} queries collapse onto a single template "
f"('{dominant_template}') varying only a parameter — the boundary-probing "
f"pattern of an extraction attack systematically sampling the input space, "
f"not the varied phrasing of a normal user.",
)
return None
def analyze(self, events: list[QueryEvent]) -> list[TheftFinding]:
findings: list[TheftFinding] = []
client_ids = sorted({e.client_id for e in events})
for client_id in client_ids:
for finding in (self.check_rate_anomaly(events, client_id),
self.check_systematic_sweep(events, client_id)):
if finding:
findings.append(finding)
return findings
def _build_simulated_traffic() -> list[QueryEvent]:
events: list[QueryEvent] = []
# Client A: a normal user, a handful of varied, organic questions.
normal_queries = [
"What's the return policy on electronics?",
"Can you help me track order 4821?",
"Do you ship to Canada?",
"What payment methods do you accept?",
"Is there a student discount?",
]
for i, q in enumerate(normal_queries):
events.append(QueryEvent(client_id="client_normal_user", timestamp=i * 40.0, query_text=q))
# Client B: an extraction attempt — a dense, systematic sweep of a single
# template across many parameter values, issued far faster than a human
# could type, characteristic of scripted querying used to map a model's
# decision boundary or gather training pairs for a distilled clone.
for i in range(45):
events.append(QueryEvent(
client_id="client_extractor",
timestamp=i * 0.8, # a new query roughly every 0.8s — scripted pace
query_text=f"Classify the sentiment of review {1000 + i} as positive, negative, or neutral.",
))
return events
def demo_model_theft() -> None:
print("=" * 70)
print("PART A: MODEL THEFT / EXTRACTION (LLM10)")
print("=" * 70)
events = _build_simulated_traffic()
detector = ModelTheftDetector(rate_window_seconds=60.0, rate_threshold=30, sweep_template_threshold=8)
findings = detector.analyze(events)
print(f"\nAnalyzed {len(events)} queries across "
f"{len({e.client_id for e in events})} clients.\n")
if not findings:
print("No extraction signatures detected.")
for f in findings:
print(f"[{f.severity.upper():6s}] client='{f.client_id}' signal={f.signal}")
print(f" {f.detail}\n")
print("-" * 70)
print("Ranked LLM10 defenses (where each one intervenes in the attack chain):")
defenses = [
("1. Rate limiting", "Caps queries per client per window before extraction volume is even reachable — "
"the cheapest, earliest intervention point."),
("2. Query pattern monitoring", "Detects systematic sweeps (this detector's second signal) that stay under "
"a pure rate limit by pacing requests slowly but structurally still map "
"the input space."),
("3. Output perturbation / rate-limited precision", "Returns slightly reduced-precision outputs "
"(e.g. rounded confidence scores) to genuinely "
"suspicious high-volume clients, degrading the "
"quality of a distilled clone without breaking "
"normal usage."),
("4. Watermarking model outputs", "Embeds a detectable statistical signature in generated output so that "
"a competitor's model trained on scraped outputs can later be shown "
"to have been distilled from yours."),
("5. API access controls & tiered keys", "Ties query volume and access breadth to authenticated, "
"accountable API keys rather than anonymous access, so "
"extraction traffic can be attributed and cut off."),
]
for name, detail in defenses:
print(f" {name}\n {detail}")
# ══════════════════════════════════════════════════════════════════════════
# PART B — Capstone: Full OWASP LLM Top 10 Mapping
# ══════════════════════════════════════════════════════════════════════════
@dataclass
class SystemProfile:
"""
A structured description of a real AI system's architecture and
controls, in the shape a security reviewer would actually collect during
a pre-launch review: what the system does, what data/tools it touches,
and which controls are actually implemented (not just intended).
"""
name: str
description: str
# LLM01 Prompt Injection
input_source: str # e.g. "user text + retrieved documents"
structural_separation: bool # delimiters/XML tags separating trusted vs untrusted content
injection_detector: bool
# LLM02/LLM04 Insecure Output Handling & DoS
output_sanitized_before_render: bool
output_used_in_sql_or_shell: bool # True = risk present
token_budget_enforced: bool
# LLM03/LLM05 Poisoning & Supply Chain
training_data_provenance_tracked: bool
dependency_scanning: bool # e.g. pip-audit in CI
model_weight_verification: bool
# LLM06/LLM07 Disclosure & Plugin/Tool Design
pii_redaction_on_output: bool
tool_permissions_least_privilege: bool
tool_output_sanitized: bool
# LLM08/LLM09 Excessive Agency & Overreliance
human_approval_for_high_risk_actions: bool
step_budget_enforced: bool
citation_required_high_stakes: bool
# LLM10 Model Theft
rate_limiting_enabled: bool
query_pattern_monitoring: bool
@dataclass
class CategoryFinding:
category_id: str
category_name: str
status: str # "present" | "partial" | "missing"
evidence: str
recommendation: str
class OwaspLlmTop10Mapper:
"""
Maps a SystemProfile against all 10 OWASP LLM Top 10 categories and
produces a gap-analysis report — the concrete deliverable of a
pre-launch AI security review. Each category is scored independently
from the profile's actual controls, not from what the system's
documentation claims, mirroring the discipline this phase has applied
to every individual category: audit the granted permission / actual
control, not the stated intent.
"""
def map(self, profile: SystemProfile) -> list[CategoryFinding]:
findings: list[CategoryFinding] = []
# LLM01 — Prompt Injection
if profile.structural_separation and profile.injection_detector:
findings.append(CategoryFinding(
"LLM01", "Prompt Injection", "present",
f"Input source '{profile.input_source}' is structurally separated from trusted "
f"instructions and passes through an injection detector.",
"Maintain detector coverage as new input sources are added.",
))
elif profile.structural_separation or profile.injection_detector:
findings.append(CategoryFinding(
"LLM01", "Prompt Injection", "partial",
"Only one of structural separation or an injection detector is present.",
"Add the missing layer — structural separation and detection are complementary, "
"not substitutes for each other.",
))
else:
findings.append(CategoryFinding(
"LLM01", "Prompt Injection", "missing",
f"Untrusted content from '{profile.input_source}' is not structurally separated "
f"and no injection detector is present.",
"Implement delimiter/XML-based structural separation between trusted instructions "
"and untrusted content, plus a detection layer for known injection patterns.",
))
# LLM02 / LLM04 — Insecure Output Handling & DoS
if profile.output_used_in_sql_or_shell and not profile.output_sanitized_before_render:
findings.append(CategoryFinding(
"LLM02/04", "Insecure Output Handling & Model DoS", "missing",
"Model output is used to construct SQL/shell commands or rendered without "
"sanitization, and no output validation layer exists.",
"Never build SQL/shell strings from raw model output — use parameterized "
"queries/sandboxed execution, and sanitize/escape all rendered output.",
))
elif not profile.output_sanitized_before_render or not profile.token_budget_enforced:
findings.append(CategoryFinding(
"LLM02/04", "Insecure Output Handling & Model DoS", "partial",
f"output_sanitized_before_render={profile.output_sanitized_before_render}, "
f"token_budget_enforced={profile.token_budget_enforced}.",
"Add the missing control: output sanitization before rendering, and/or a "
"per-request token budget with circuit breakers against context-flooding.",
))
else:
findings.append(CategoryFinding(
"LLM02/04", "Insecure Output Handling & Model DoS", "present",
"Output is sanitized before rendering, not used to build SQL/shell strings, "
"and token budgets are enforced.",
"Periodically re-test output sanitization against new rendering surfaces (e.g. "
"new UI components) as the product evolves.",
))
# LLM03 / LLM05 — Poisoning & Supply Chain
supply_controls = [profile.training_data_provenance_tracked, profile.dependency_scanning,
profile.model_weight_verification]
present_count = sum(supply_controls)
if present_count == 3:
status, rec = "present", "Keep provenance tracking, dependency scanning, and weight " \
"verification running continuously as the supply chain changes, " \
"not just at initial launch."
elif present_count == 0:
status, rec = "missing", "Establish training data provenance tracking, add dependency " \
"scanning (e.g. pip-audit) to CI, and verify model weight hashes " \
"against a trusted source before deployment."
else:
status, rec = "partial", "Fill the specific gap: provenance tracking, dependency scanning, " \
"or weight verification, whichever is currently False."
findings.append(CategoryFinding(
"LLM03/05", "Training Data Poisoning & Supply Chain", status,
f"provenance={profile.training_data_provenance_tracked}, "
f"dependency_scanning={profile.dependency_scanning}, "
f"weight_verification={profile.model_weight_verification}.",
rec,
))
# LLM06 / LLM07 — Disclosure & Plugin/Tool Design
disclosure_controls = [profile.pii_redaction_on_output, profile.tool_permissions_least_privilege,
profile.tool_output_sanitized]
present_count = sum(disclosure_controls)
if present_count == 3:
status, rec = "present", "Re-audit tool permission scopes whenever a new tool is added to " \
"the registry — least privilege drifts toward over-permission by " \
"default over time."
elif present_count == 0:
status, rec = "missing", "Add output-side PII/secret redaction, scope every tool to least " \
"privilege, and sanitize all tool-returned content as untrusted " \
"input before it re-enters the model's context."
else:
status, rec = "partial", "Fill the specific gap: PII redaction, least-privilege tool scoping, " \
"or tool-output sanitization, whichever is currently False."
findings.append(CategoryFinding(
"LLM06/07", "Sensitive Information Disclosure & Insecure Plugin/Tool Design", status,
f"pii_redaction={profile.pii_redaction_on_output}, "
f"least_privilege_tools={profile.tool_permissions_least_privilege}, "
f"tool_output_sanitized={profile.tool_output_sanitized}.",
rec,
))
# LLM08 / LLM09 — Excessive Agency & Overreliance
agency_controls = [profile.human_approval_for_high_risk_actions, profile.step_budget_enforced,
profile.citation_required_high_stakes]
present_count = sum(agency_controls)
if present_count == 3:
status, rec = "present", "Periodically red-team the approval gate and step budget thresholds " \
"as new tool capabilities are added to the agent."
elif present_count == 0:
status, rec = "missing", "Add a human-approval gate for high-risk/irreversible actions, a " \
"step budget to catch multi-step privilege escalation, and mandatory " \
"citations for high-stakes claims."
else:
status, rec = "partial", "Fill the specific gap: human approval gate, step budget, or " \
"citation requirement, whichever is currently False."
findings.append(CategoryFinding(
"LLM08/09", "Excessive Agency & Overreliance", status,
f"human_approval={profile.human_approval_for_high_risk_actions}, "
f"step_budget={profile.step_budget_enforced}, "
f"citations_required={profile.citation_required_high_stakes}.",
rec,
))
# LLM10 — Model Theft
if profile.rate_limiting_enabled and profile.query_pattern_monitoring:
status, rec = "present", "Continue tuning rate and sweep-detection thresholds against " \
"observed legitimate traffic to avoid false positives blocking " \
"real users."
elif profile.rate_limiting_enabled or profile.query_pattern_monitoring:
status, rec = "partial", "Add the missing layer — rate limiting stops raw volume, but a " \
"slow, paced systematic sweep evades it and requires pattern " \
"monitoring as a second, independent control."
else:
status, rec = "missing", "Add both per-client rate limiting and query-pattern monitoring " \
"(e.g. this lesson's ModelTheftDetector) to catch high-volume and " \
"low-volume-but-systematic extraction attempts respectively."
findings.append(CategoryFinding(
"LLM10", "Model Theft", status,
f"rate_limiting={profile.rate_limiting_enabled}, "
f"query_pattern_monitoring={profile.query_pattern_monitoring}.",
rec,
))
return findings
@staticmethod
def summarize(findings: list[CategoryFinding]) -> str:
counts = {"present": 0, "partial": 0, "missing": 0}
for f in findings:
counts[f.status] += 1
return (f"{counts['present']}/{len(findings)} categories fully controlled, "
f"{counts['partial']}/{len(findings)} partial, "
f"{counts['missing']}/{len(findings)} missing.")
# A realistic, imperfect system: a customer-support RAG agent with tool
# access. It has some real controls in place (this is not a strawman with
# zero defenses) but several concrete, specific gaps — the shape of an
# actual pre-launch review finding, not a toy example.
SAMPLE_SYSTEM = SystemProfile(
name="SupportBot RAG Agent",
description="A customer-support agent with RAG over a knowledge base and tool access to "
"look up orders, issue refunds up to $50 autonomously, and escalate to a human "
"for larger refunds.",
input_source="user chat text + retrieved KB documents",
structural_separation=True,
injection_detector=False,
output_sanitized_before_render=True,
output_used_in_sql_or_shell=False,
token_budget_enforced=False,
training_data_provenance_tracked=False,
dependency_scanning=True,
model_weight_verification=False,
pii_redaction_on_output=False,
tool_permissions_least_privilege=True,
tool_output_sanitized=False,
human_approval_for_high_risk_actions=True,
step_budget_enforced=False,
citation_required_high_stakes=False,
rate_limiting_enabled=True,
query_pattern_monitoring=False,
)
def demo_full_mapping() -> None:
print("\n" + "=" * 70)
print("PART B: CAPSTONE — FULL OWASP LLM TOP 10 MAPPING")
print("=" * 70)
print(f"\nSystem under review: {SAMPLE_SYSTEM.name}")
print(f" {SAMPLE_SYSTEM.description}\n")
mapper = OwaspLlmTop10Mapper()
findings = mapper.map(SAMPLE_SYSTEM)
status_marker = {"present": "[OK] ", "partial": "[PARTIAL]", "missing": "[MISSING]"}
for f in findings:
print(f"{status_marker[f.status]} {f.category_id:9s} {f.category_name}")
print(f" evidence: {f.evidence}")
print(f" recommendation: {f.recommendation}\n")
print("-" * 70)
print(f"SUMMARY: {mapper.summarize(findings)}")
print("\nInterpretation: a real system is rarely all-present or all-missing — this profile has "
"genuine controls (structural separation, least-privilege tools, a human approval gate, "
"rate limiting) alongside specific, nameable gaps (no injection detector, no tool-output "
"sanitization, no step budget, no citation requirement, no query-pattern monitoring). A "
"useful gap analysis names each gap individually with a concrete fix, rather than issuing "
"a single pass/fail verdict on 'AI security' as an undifferentiated whole.")
if __name__ == "__main__":
demo_model_theft()
demo_full_mapping()
print("\n" + "=" * 70)
print("TAKEAWAY")
print("=" * 70)
print("LLM10 defenses assume an attacker never needs your weights: query volume and query "
"STRUCTURE are both extraction signals, and rate limiting alone misses a patient, "
"systematic sweep paced under the rate threshold. The capstone mapping exercise is the "
"same discipline applied at the system level: a security review is only useful if it "
"checks all 10 categories explicitly and names the specific gap in each, rather than "
"stopping once it finds the first or most obvious vulnerability.")