Module 8: Capstone Project — Secured AI System

6. Documenting Security Decisions

Overview

Code without documentation is code with an expiration date. You can build the most sophisticated security system in the world, but if nobody understands why certain decisions were made, the first new developer who touches the code will break a defense without knowing it. Security documentation isn't bureaucracy — it's the institutional memory that protects the system when you're not around.

Security decisions are especially hard to document because they involve invisible trade-offs. Why did you choose token-based rate limiting instead of IP-based? Why does the guardrail reject at a score of 0.7 and not 0.8? Why do you use bcrypt and not argon2? Without the documented context, these decisions look arbitrary and get changed without understanding the consequences.

In this capsule you'll implement a complete documentation system: Architecture Decision Records (ADR) for security decisions, a final OWASP mapping that shows the state of each LLM01-LLM10 risk with evidence from each module, honest documentation of residual risks, and automatic documentation generation from the code.


Architecture Decision Records (ADR) for security

What an ADR is

An Architecture Decision Record is a short document that captures a significant technical decision along with its context and consequences. Unlike code comments, an ADR documents the "why" at the architectural level — not the implementation, but the decision that motivated it.

ADR template for AI security

# ADR-{number}: {title}

**Status:** {proposed | accepted | deprecated | replaced}
**Date:** {YYYY-MM-DD}
**Authors:** {names}

## Context
{What problem we're solving and what constraints exist}

## Decision
{What we decided to do}

## Consequences
{What this decision implies — positive and negative}

## Alternatives Considered
{What other options we evaluated and why we discarded them}

Example: 3 ADRs for a secure AI system

ADR-001: Input Sanitization Multi-Layer

# ADR-001: Input Sanitization with a Multi-Layer Pipeline

**Status:** Accepted
**Date:** 2026-02-15
**Authors:** Security Team

## Context
User prompts can contain direct injections,
encoding tricks (Base64, ROT13), or multi-turn attacks.
A single sanitization layer isn't enough — attackers
evolve their techniques faster than we update rules.

## Decision
Implement a 3-layer pipeline:
1. Decode layer: normalizes encodings (Base64, URL, Unicode)
2. Pattern layer: detects known injection patterns
3. Semantic layer: classifies intent with embedding similarity

## Consequences
Positive:
- Defense in depth against multiple vectors
- Each layer is independent and testable
Negative:
- Additional latency of ~50ms per layer
- Operational complexity of maintaining 3 systems

## Alternatives Considered
- Regex only: discarded due to false negatives in semantic attacks
- Embeddings only: discarded due to high computational cost
- External WAF: discarded because it doesn't understand AI context

ADR-002: Rate Limiting by Token

# ADR-002: Rate Limiting Based on Token Count

**Status:** Accepted
**Date:** 2026-02-20
**Authors:** Backend Team

## Context
Traditional rate limiting by requests/minute doesn't protect
against cost attacks in LLMs. A single request with a 100K-token
prompt costs more than 100 short requests.

## Decision
Implement dual rate limiting:
- By requests: max 60/min per user
- By tokens: max 50,000 tokens/hour per user

## Consequences
Positive:
- Direct protection against cost attacks
- Legitimate users rarely hit the token limit
Negative:
- Requires estimating tokens before sending to the LLM
- Token counting adds ~5ms of latency

## Alternatives Considered
- Request-based only: doesn't protect against long prompts
- Monthly budget cap: too slow to detect spikes
- Pre-auth token estimation: rejected due to complexity

SecurityDecisionRecord class

from pydantic import BaseModel, Field, computed_field
from enum import Enum
from datetime import datetime, timezone
from typing import Optional


class ADRStatus(str, Enum):
    PROPOSED = "proposed"
    ACCEPTED = "accepted"
    DEPRECATED = "deprecated"
    REPLACED = "replaced"


class SecurityCategory(str, Enum):
    INPUT_VALIDATION = "input_validation"
    OUTPUT_FILTERING = "output_filtering"
    AUTHENTICATION = "authentication"
    RATE_LIMITING = "rate_limiting"
    DATA_PROTECTION = "data_protection"
    MONITORING = "monitoring"
    INCIDENT_RESPONSE = "incident_response"
    COST_CONTROL = "cost_control"


class Alternative(BaseModel):
    """An alternative considered and discarded."""
    name: str
    description: str
    reason_rejected: str


class Consequence(BaseModel):
    """Positive or negative consequence of the decision."""
    description: str
    is_positive: bool
    impact_area: str  # "performance", "security", "cost", "complexity"


class SecurityDecisionRecord(BaseModel):
    """Specialized ADR for AI security decisions."""
    adr_id: str
    title: str
    status: ADRStatus
    date: str
    authors: list[str]
    category: SecurityCategory
    context: str
    decision: str
    consequences: list[Consequence] = Field(default_factory=list)
    alternatives_considered: list[Alternative] = Field(default_factory=list)
    related_adrs: list[str] = Field(default_factory=list)
    owasp_risks_addressed: list[str] = Field(default_factory=list)
    superseded_by: Optional[str] = None

    @computed_field
    @property
    def trade_off_summary(self) -> str:
        """Summarizes the trade-off balance of the decision."""
        positives = [c for c in self.consequences if c.is_positive]
        negatives = [c for c in self.consequences if not c.is_positive]
        return (
            f"{len(positives)} benefits, "
            f"{len(negatives)} costs"
        )

    def to_markdown(self) -> str:
        """Generates the ADR in markdown format."""
        lines = [
            f"# {self.adr_id}: {self.title}",
            "",
            f"**Status:** {self.status.value}",
            f"**Date:** {self.date}",
            f"**Authors:** {', '.join(self.authors)}",
            f"**Category:** {self.category.value}",
            f"**OWASP:** {', '.join(self.owasp_risks_addressed) or 'N/A'}",
            "",
            "## Context",
            self.context,
            "",
            "## Decision",
            self.decision,
            "",
            "## Consequences",
        ]
        for c in self.consequences:
            icon = "✅" if c.is_positive else "⚠️"
            lines.append(f"- {icon} [{c.impact_area}] {c.description}")

        if self.alternatives_considered:
            lines.extend(["", "## Alternatives Considered"])
            for alt in self.alternatives_considered:
                lines.append(f"### {alt.name}")
                lines.append(f"{alt.description}")
                lines.append(f"**Reason for discarding:** {alt.reason_rejected}")
                lines.append("")

        if self.related_adrs:
            lines.extend(["", "## Related ADRs"])
            for related in self.related_adrs:
                lines.append(f"- {related}")

        return "\n".join(lines)


class ADRRegistry(BaseModel):
    """Centralized registry of all security decisions."""
    records: list[SecurityDecisionRecord] = Field(default_factory=list)

    def add(self, record: SecurityDecisionRecord):
        self.records.append(record)

    def by_category(self) -> dict[str, list[SecurityDecisionRecord]]:
        result: dict[str, list[SecurityDecisionRecord]] = {}
        for r in self.records:
            result.setdefault(r.category.value, []).append(r)
        return result

    def active_decisions(self) -> list[SecurityDecisionRecord]:
        return [r for r in self.records if r.status == ADRStatus.ACCEPTED]

    def find_by_owasp(self, risk_id: str) -> list[SecurityDecisionRecord]:
        """Finds ADRs that address a specific OWASP risk."""
        return [
            r for r in self.records
            if risk_id in r.owasp_risks_addressed
        ]

    def generate_index(self) -> str:
        """Generates a markdown index of all ADRs."""
        lines = ["# Security Decisions Index", ""]
        lines.append("| ID | Title | Status | Category | OWASP |")
        lines.append("|-----|--------|--------|-----------|-------|")
        for r in self.records:
            owasp = ", ".join(r.owasp_risks_addressed) or "—"
            lines.append(
                f"| {r.adr_id} | {r.title} | {r.status.value} | "
                f"{r.category.value} | {owasp} |"
            )
        return "\n".join(lines)


# --- Usage example ---
registry = ADRRegistry()

adr1 = SecurityDecisionRecord(
    adr_id="ADR-001",
    title="Input Sanitization Multi-Layer",
    status=ADRStatus.ACCEPTED,
    date="2026-02-15",
    authors=["security-lead"],
    category=SecurityCategory.INPUT_VALIDATION,
    context="Prompts can contain direct injections and encoding tricks.",
    decision="3-layer pipeline: decode, pattern, semantic.",
    consequences=[
        Consequence(description="Defense in depth", is_positive=True, impact_area="security"),
        Consequence(description="~150ms additional latency", is_positive=False, impact_area="performance"),
    ],
    alternatives_considered=[
        Alternative(name="Regex only", description="Pattern matching with regular expressions",
                    reason_rejected="False negatives in semantic attacks"),
    ],
    owasp_risks_addressed=["LLM01", "LLM02"],
)

adr2 = SecurityDecisionRecord(
    adr_id="ADR-002",
    title="Rate Limiting by Token Count",
    status=ADRStatus.ACCEPTED,
    date="2026-02-20",
    authors=["backend-team"],
    category=SecurityCategory.RATE_LIMITING,
    context="Request-based rate limiting doesn't protect against cost attacks.",
    decision="Dual rate limiting: requests/min + tokens/hour.",
    consequences=[
        Consequence(description="Protection against cost attacks", is_positive=True, impact_area="cost"),
        Consequence(description="~5ms latency for token counting", is_positive=False, impact_area="performance"),
    ],
    owasp_risks_addressed=["LLM04"],
)

registry.add(adr1)
registry.add(adr2)

print(registry.generate_index())
print()
print(adr1.to_markdown())

Explanation: SecurityDecisionRecord extends the ADR concept with fields specific to AI security: category, OWASP mapping, and consequences typed by impact area. The ADRRegistry lets you search decisions by OWASP risk, which connects directly to the final mapping.


Documenting trade-offs

Every security decision has a cost. Documenting these trade-offs with real data prevents someone from "optimizing" the system by removing a defense without understanding its purpose.

import time
from pydantic import BaseModel, Field


class TradeOffMeasurement(BaseModel):
    """Concrete measurement of a security trade-off."""
    defense_name: str
    metric_name: str
    value_without_defense: float
    value_with_defense: float
    unit: str
    acceptable_threshold: float

    @property
    def overhead(self) -> float:
        return self.value_with_defense - self.value_without_defense

    @property
    def overhead_percentage(self) -> float:
        if self.value_without_defense == 0:
            return 0.0
        return (self.overhead / self.value_without_defense) * 100

    @property
    def is_acceptable(self) -> bool:
        return self.value_with_defense <= self.acceptable_threshold

    def summary(self) -> str:
        status = "✅ ACCEPTABLE" if self.is_acceptable else "⚠️ EXCEEDS THRESHOLD"
        return (
            f"{self.defense_name}{self.metric_name}:\n"
            f"  Without defense: {self.value_without_defense:.1f} {self.unit}\n"
            f"  With defense: {self.value_with_defense:.1f} {self.unit}\n"
            f"  Overhead: +{self.overhead:.1f} {self.unit} ({self.overhead_percentage:.1f}%)\n"
            f"  Threshold: {self.acceptable_threshold:.1f} {self.unit}\n"
            f"  Status: {status}"
        )


def measure_defense_overhead(
    defense_name: str,
    operation_without: callable,
    operation_with: callable,
    iterations: int = 100,
    threshold_ms: float = 200.0,
) -> TradeOffMeasurement:
    """Measures the real overhead of a defense in milliseconds."""
    # Measure without defense
    start = time.perf_counter()
    for _ in range(iterations):
        operation_without()
    time_without = ((time.perf_counter() - start) / iterations) * 1000

    # Measure with defense
    start = time.perf_counter()
    for _ in range(iterations):
        operation_with()
    time_with = ((time.perf_counter() - start) / iterations) * 1000

    return TradeOffMeasurement(
        defense_name=defense_name,
        metric_name="latency",
        value_without_defense=time_without,
        value_with_defense=time_with,
        unit="ms",
        acceptable_threshold=threshold_ms,
    )


# --- Example: measure sanitization overhead ---
import re

INJECTION_PATTERNS = [
    re.compile(r"ignor[ae]\s+instrucciones", re.IGNORECASE),
    re.compile(r"system\s*prompt", re.IGNORECASE),
    re.compile(r"DAN\s*mode", re.IGNORECASE),
    re.compile(r"(?i)base64|rot13|hex\s*encode"),
]


def process_without_defense():
    text = "¿Cuál es el precio del producto X en la tienda de Madrid?"
    return text.lower().strip()


def process_with_defense():
    text = "¿Cuál es el precio del producto X en la tienda de Madrid?"
    text = text.lower().strip()
    for pattern in INJECTION_PATTERNS:
        pattern.search(text)
    return text


measurement = measure_defense_overhead(
    defense_name="Input Sanitization (regex)",
    operation_without=process_without_defense,
    operation_with=process_with_defense,
    iterations=1000,
    threshold_ms=5.0,
)

print(measurement.summary())


# --- Complete trade-off documentation ---
class TradeOffRegistry(BaseModel):
    """Registry of all measured security trade-offs."""
    measurements: list[TradeOffMeasurement] = Field(default_factory=list)

    def add(self, m: TradeOffMeasurement):
        self.measurements.append(m)

    def generate_report(self) -> str:
        lines = [
            "# Security Trade-Off Report",
            "",
            "| Defense | Metric | Without | With | Overhead | Status |",
            "|---------|---------|-----|-----|----------|--------|",
        ]
        for m in self.measurements:
            status = "✅" if m.is_acceptable else "⚠️"
            lines.append(
                f"| {m.defense_name} | {m.metric_name} | "
                f"{m.value_without_defense:.1f}{m.unit} | "
                f"{m.value_with_defense:.1f}{m.unit} | "
                f"+{m.overhead_percentage:.1f}% | {status} |"
            )
        unacceptable = [m for m in self.measurements if not m.is_acceptable]
        if unacceptable:
            lines.extend(["", "## Defenses that exceed the threshold"])
            for m in unacceptable:
                lines.append(f"- **{m.defense_name}**: review implementation or adjust threshold")
        return "\n".join(lines)


registry = TradeOffRegistry()
registry.add(measurement)
registry.add(TradeOffMeasurement(
    defense_name="Output PII Redaction",
    metric_name="latency",
    value_without_defense=12.0,
    value_with_defense=45.0,
    unit="ms",
    acceptable_threshold=100.0,
))
registry.add(TradeOffMeasurement(
    defense_name="Semantic Guardrail (embeddings)",
    metric_name="latency",
    value_without_defense=15.0,
    value_with_defense=180.0,
    unit="ms",
    acceptable_threshold=150.0,
))

print()
print(registry.generate_report())

Explanation: measure_defense_overhead produces real data, not estimates. The TradeOffRegistry generates a report that justifies each defense with concrete numbers. When the overhead exceeds the threshold, the report flags it explicitly so the team can evaluate whether to optimize or adjust the threshold.


Final OWASP Mapping

The OWASP mapping is the central audit artifact: it shows which OWASP LLM Top 10 risks are mitigated, with which defenses, and what evidence backs that mitigation.

from pydantic import BaseModel, Field, computed_field
from enum import Enum


class MitigationStatus(str, Enum):
    NOT_MITIGATED = "not_mitigated"
    PARTIALLY_MITIGATED = "partially_mitigated"
    MITIGATED = "mitigated"
    NOT_APPLICABLE = "not_applicable"


class DefenseEvidence(BaseModel):
    """Evidence that a defense exists and works."""
    defense_name: str
    module_source: str  # "Module 2", "Module 5", etc.
    implementation_file: str
    test_coverage: str
    last_verified: str


class ResidualRisk(BaseModel):
    """Risk that remains after applying mitigations."""
    description: str
    likelihood: str  # "low", "medium", "high"
    impact: str
    mitigation_plan: str


class OWASPRiskEntry(BaseModel):
    """State of a specific OWASP LLM risk."""
    risk_id: str
    risk_name: str
    description: str
    status: MitigationStatus
    defense_layers: list[DefenseEvidence] = Field(default_factory=list)
    residual_risks: list[ResidualRisk] = Field(default_factory=list)
    notes: str = ""

    @computed_field
    @property
    def defense_count(self) -> int:
        return len(self.defense_layers)

    @computed_field
    @property
    def has_residual_risk(self) -> bool:
        return len(self.residual_risks) > 0


class OWASPFinalMapping(BaseModel):
    """Complete OWASP LLM Top 10 mapping with evidence."""
    project_name: str
    assessment_date: str
    assessor: str
    risks: list[OWASPRiskEntry] = Field(default_factory=list)

    @computed_field
    @property
    def overall_coverage(self) -> float:
        """Percentage of mitigated or partially mitigated risks."""
        applicable = [r for r in self.risks if r.status != MitigationStatus.NOT_APPLICABLE]
        if not applicable:
            return 0.0
        mitigated = [
            r for r in applicable
            if r.status in (MitigationStatus.MITIGATED, MitigationStatus.PARTIALLY_MITIGATED)
        ]
        return (len(mitigated) / len(applicable)) * 100

    @computed_field
    @property
    def fully_mitigated_count(self) -> int:
        return sum(1 for r in self.risks if r.status == MitigationStatus.MITIGATED)

    def add_risk(self, entry: OWASPRiskEntry):
        self.risks.append(entry)

    def get_risk(self, risk_id: str) -> OWASPRiskEntry | None:
        for r in self.risks:
            if r.risk_id == risk_id:
                return r
        return None

    def generate_status_report(self) -> str:
        """Generates an OWASP status report in markdown format."""
        status_icons = {
            MitigationStatus.MITIGATED: "🟢",
            MitigationStatus.PARTIALLY_MITIGATED: "🟡",
            MitigationStatus.NOT_MITIGATED: "🔴",
            MitigationStatus.NOT_APPLICABLE: "⚪",
        }

        lines = [
            f"# OWASP LLM Top 10 — Assessment Report",
            f"**Project:** {self.project_name}",
            f"**Date:** {self.assessment_date}",
            f"**Assessor:** {self.assessor}",
            f"**Overall coverage:** {self.overall_coverage:.0f}%",
            f"**Fully mitigated:** {self.fully_mitigated_count}/{len(self.risks)}",
            "",
            "## Status by Risk",
            "",
            "| Risk ID | Name | Status | Defenses | Residual Risk |",
            "|---------|--------|--------|----------|-----------------|",
        ]

        for r in self.risks:
            icon = status_icons[r.status]
            residual = "Yes" if r.has_residual_risk else "No"
            lines.append(
                f"| {r.risk_id} | {r.risk_name} | "
                f"{icon} {r.status.value} | {r.defense_count} | {residual} |"
            )

        # Detail by risk
        lines.extend(["", "## Detail by Risk", ""])
        for r in self.risks:
            lines.append(f"### {r.risk_id}: {r.risk_name}")
            lines.append(f"**Status:** {status_icons[r.status]} {r.status.value}")
            lines.append(f"\n{r.description}\n")

            if r.defense_layers:
                lines.append("**Defenses implemented:**")
                for d in r.defense_layers:
                    lines.append(
                        f"- [{d.module_source}] {d.defense_name} "
                        f"(`{d.implementation_file}`) — Tests: {d.test_coverage}"
                    )

            if r.residual_risks:
                lines.append("\n**Residual risks:**")
                for rr in r.residual_risks:
                    lines.append(
                        f"- {rr.description} "
                        f"(likelihood: {rr.likelihood}, impact: {rr.impact})"
                    )
                    lines.append(f"  Plan: {rr.mitigation_plan}")

            if r.notes:
                lines.append(f"\n**Notes:** {r.notes}")
            lines.append("")

        return "\n".join(lines)


# --- Build the final mapping ---
mapping = OWASPFinalMapping(
    project_name="SecureAI Chat System",
    assessment_date="2026-03-14",
    assessor="Security Team",
)

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM01",
    risk_name="Prompt Injection",
    description="Model manipulation via instructions injected into the input.",
    status=MitigationStatus.MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Input Sanitization Pipeline",
            module_source="Module 2",
            implementation_file="security/input_validator.py",
            test_coverage="95% — 50 adversarial prompts",
            last_verified="2026-03-10"
        ),
        DefenseEvidence(
            defense_name="Semantic Guardrail",
            module_source="Module 4",
            implementation_file="security/guardrail.py",
            test_coverage="90% — embedding similarity checks",
            last_verified="2026-03-12"
        ),
    ],
    residual_risks=[
        ResidualRisk(
            description="Novel encoding techniques not covered by the decoder",
            likelihood="low",
            impact="medium",
            mitigation_plan="Update the adversarial dataset monthly with new techniques"
        )
    ]
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM02",
    risk_name="Insecure Output Handling",
    description="Model output used unsanitized in downstream contexts.",
    status=MitigationStatus.MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Output Filter Pipeline",
            module_source="Module 3",
            implementation_file="security/output_filter.py",
            test_coverage="92% — PII redaction + code injection",
            last_verified="2026-03-11"
        ),
    ],
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM03",
    risk_name="Training Data Poisoning",
    description="Training data manipulated to alter the model's behavior.",
    status=MitigationStatus.NOT_APPLICABLE,
    notes="We use pre-trained models from providers (OpenAI/Anthropic). We don't do fine-tuning."
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM04",
    risk_name="Model Denial of Service",
    description="Attacks that exhaust the model's resources or generate excessive costs.",
    status=MitigationStatus.MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Dual Rate Limiter (requests + tokens)",
            module_source="Module 5",
            implementation_file="security/rate_limiter.py",
            test_coverage="88% — load tests with 1000 RPS",
            last_verified="2026-03-09"
        ),
        DefenseEvidence(
            defense_name="Cost Controller",
            module_source="Module 6",
            implementation_file="security/cost_controller.py",
            test_coverage="85% — budget cap tests",
            last_verified="2026-03-10"
        ),
    ],
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM05",
    risk_name="Supply Chain Vulnerabilities",
    description="Compromised third-party dependencies, plugins, or models.",
    status=MitigationStatus.PARTIALLY_MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Dependency scanning (Dependabot)",
            module_source="Module 7",
            implementation_file=".github/dependabot.yml",
            test_coverage="CI/CD automated",
            last_verified="2026-03-14"
        ),
    ],
    residual_risks=[
        ResidualRisk(
            description="No integrity validation of downloaded models",
            likelihood="low",
            impact="high",
            mitigation_plan="Implement checksum verification for model downloads"
        )
    ]
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM06",
    risk_name="Sensitive Information Disclosure",
    description="The model reveals confidential information in its responses.",
    status=MitigationStatus.MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="PII Redactor",
            module_source="Module 3",
            implementation_file="security/pii_redactor.py",
            test_coverage="94% — regex + NER patterns",
            last_verified="2026-03-12"
        ),
        DefenseEvidence(
            defense_name="System Prompt Protection",
            module_source="Module 2",
            implementation_file="security/prompt_shield.py",
            test_coverage="90% — extraction attack suite",
            last_verified="2026-03-11"
        ),
    ],
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM07",
    risk_name="Insecure Plugin Design",
    description="LLM plugins/tools with excessive permissions or no validation.",
    status=MitigationStatus.PARTIALLY_MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Tool Permission System",
            module_source="Module 5",
            implementation_file="security/tool_permissions.py",
            test_coverage="80% — permission boundary tests",
            last_verified="2026-03-10"
        ),
    ],
    residual_risks=[
        ResidualRisk(
            description="Third-party tools don't have complete sandboxing",
            likelihood="medium",
            impact="high",
            mitigation_plan="Implement container-based sandboxing for tool execution"
        )
    ]
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM08",
    risk_name="Excessive Agency",
    description="The model takes autonomous actions without user confirmation.",
    status=MitigationStatus.MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Human-in-the-loop for destructive actions",
            module_source="Module 5",
            implementation_file="security/action_guard.py",
            test_coverage="100% — all destructive actions require confirmation",
            last_verified="2026-03-13"
        ),
    ],
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM09",
    risk_name="Overreliance",
    description="Users blindly trust the model's outputs without verifying.",
    status=MitigationStatus.PARTIALLY_MITIGATED,
    defense_layers=[
        DefenseEvidence(
            defense_name="Confidence disclaimers on outputs",
            module_source="Module 3",
            implementation_file="security/output_disclaimer.py",
            test_coverage="70% — UI integration tests",
            last_verified="2026-03-08"
        ),
    ],
    notes="Limited mitigation — depends on user behavior."
))

mapping.add_risk(OWASPRiskEntry(
    risk_id="LLM10",
    risk_name="Model Theft",
    description="Extraction or theft of the model through the API.",
    status=MitigationStatus.NOT_APPLICABLE,
    notes="We use third-party models via API. The model is not our asset."
))

print(mapping.generate_status_report())

Explanation: The mapping is exhaustive: each OWASP risk has a status, defenses with evidence traceable to specific modules, and residual risks documented honestly. NOT_APPLICABLE is a valid state — not everything applies to your architecture.


Residual risks

Honest documentation of what is NOT covered is as important as documenting what is. A false sense of security is more dangerous than knowing where the gaps are.

from pydantic import BaseModel, Field
from enum import Enum


class RiskLikelihood(str, Enum):
    RARE = "rare"
    UNLIKELY = "unlikely"
    POSSIBLE = "possible"
    LIKELY = "likely"


class RiskImpact(str, Enum):
    NEGLIGIBLE = "negligible"
    MINOR = "minor"
    MODERATE = "moderate"
    MAJOR = "major"
    CATASTROPHIC = "catastrophic"


class ResidualRiskEntry(BaseModel):
    """A risk that isn't completely mitigated."""
    risk_id: str
    title: str
    description: str
    category: str
    likelihood: RiskLikelihood
    impact: RiskImpact
    why_not_mitigated: str
    acceptance_rationale: str
    monitoring_strategy: str
    review_date: str


class ResidualRiskRegister(BaseModel):
    """Registry of consciously accepted residual risks."""
    risks: list[ResidualRiskEntry] = Field(default_factory=list)

    def add(self, risk: ResidualRiskEntry):
        self.risks.append(risk)

    def risk_matrix(self) -> str:
        """Generates a text risk matrix."""
        lines = [
            "# Residual Risk Matrix",
            "",
            "| Risk ID | Title | Likelihood | Impact | Action |",
            "|---------|--------|------------|--------|--------|",
        ]
        for r in self.risks:
            # Likely + major/catastrophic risks require action
            needs_action = (
                r.likelihood in (RiskLikelihood.LIKELY, RiskLikelihood.POSSIBLE)
                and r.impact in (RiskImpact.MAJOR, RiskImpact.CATASTROPHIC)
            )
            action = "⚠️ PRIORITIZE" if needs_action else "📋 MONITOR"
            lines.append(
                f"| {r.risk_id} | {r.title} | {r.likelihood.value} | "
                f"{r.impact.value} | {action} |"
            )
        return "\n".join(lines)

    def generate_acceptance_document(self) -> str:
        """Generates a formal risk acceptance document."""
        lines = [
            "# Residual Risk Acceptance Document",
            "",
            "The following risks have been identified, evaluated,",
            "and consciously accepted by the security team.",
            "",
        ]
        for r in self.risks:
            lines.extend([
                f"## {r.risk_id}: {r.title}",
                f"**Category:** {r.category}",
                f"**Likelihood:** {r.likelihood.value} | **Impact:** {r.impact.value}",
                "",
                f"**Description:** {r.description}",
                "",
                f"**Why isn't it mitigated?** {r.why_not_mitigated}",
                "",
                f"**Acceptance rationale:** {r.acceptance_rationale}",
                "",
                f"**Monitoring:** {r.monitoring_strategy}",
                "",
                f"**Next review:** {r.review_date}",
                "",
                "---",
                "",
            ])
        return "\n".join(lines)


# --- Project residual risks ---
register = ResidualRiskRegister()

register.add(ResidualRiskEntry(
    risk_id="RR-001",
    title="Model Poisoning via Fine-tuning Data",
    description="If fine-tuning is done in the future, the training data could be poisoned.",
    category="Training Data",
    likelihood=RiskLikelihood.UNLIKELY,
    impact=RiskImpact.MAJOR,
    why_not_mitigated="We currently don't do fine-tuning. Mitigation requires a data validation pipeline.",
    acceptance_rationale="Doesn't apply in the current architecture. It will be reviewed if fine-tuning is adopted.",
    monitoring_strategy="Check whether the team plans fine-tuning at each quarterly review.",
    review_date="2026-06-01"
))

register.add(ResidualRiskEntry(
    risk_id="RR-002",
    title="Novel Zero-day Prompt Injection",
    description="A completely new injection technique that isn't in the adversarial dataset.",
    category="Prompt Injection",
    likelihood=RiskLikelihood.POSSIBLE,
    impact=RiskImpact.MODERATE,
    why_not_mitigated="Impossible to prevent unknown attacks. Defense based on post-facto detection.",
    acceptance_rationale="The 3 defense layers cover known patterns. The anomaly detector covers the gap.",
    monitoring_strategy="Update the adversarial dataset monthly. Subscription to AI security feeds.",
    review_date="2026-04-15"
))

register.add(ResidualRiskEntry(
    risk_id="RR-003",
    title="Supply Chain Attack in LLM Provider Dependency",
    description="The LLM provider (OpenAI/Anthropic) suffers a compromise that affects our responses.",
    category="Supply Chain",
    likelihood=RiskLikelihood.RARE,
    impact=RiskImpact.CATASTROPHIC,
    why_not_mitigated="We have no control over the provider's infrastructure.",
    acceptance_rationale="Inherent risk of using third-party services. Partially mitigated by output filtering.",
    monitoring_strategy="Monitor providers' status pages. Alerts on changes in model behavior.",
    review_date="2026-06-01"
))

print(register.risk_matrix())
print()
print(register.generate_acceptance_document())

Explanation: Residual risks are documented with an explicit reason for why they aren't mitigated and a conscious acceptance decision. This protects the team: if the risk materializes, the documentation shows it was an informed decision, not an oversight.


Generating automatic documentation

Manual documentation goes stale quickly. This scanner analyzes the source code looking for security patterns and generates up-to-date documentation automatically.

import re
from pydantic import BaseModel, Field
from pathlib import Path
from datetime import datetime, timezone


class SecurityPattern(BaseModel):
    """A security pattern detected in the code."""
    pattern_type: str
    file_path: str
    line_number: int
    code_snippet: str
    description: str


class SecurityDocGenerator(BaseModel):
    """Generates security documentation by scanning source code."""
    patterns_found: list[SecurityPattern] = Field(default_factory=list)

    # Patterns we look for in the code
    SECURITY_PATTERNS: dict[str, str] = {
        r"rate_limit|RateLimit": "Rate Limiting",
        r"sanitiz|Sanitiz|validate_input": "Input Sanitization",
        r"pii_redact|PIIRedact|redact_pii": "PII Redaction",
        r"guardrail|Guardrail|guard_rail": "Guardrail",
        r"encrypt|decrypt|hash_password|bcrypt|argon2": "Cryptography",
        r"jwt|JWT|bearer|Bearer|oauth|OAuth": "Authentication",
        r"rbac|RBAC|role_required|permission": "Authorization",
        r"audit_log|AuditLog|log_security": "Audit Logging",
    }

    def scan_content(self, file_path: str, content: str):
        """Scans a file's content looking for security patterns."""
        lines = content.split("\n")
        for line_num, line in enumerate(lines, 1):
            for pattern, description in self.SECURITY_PATTERNS.items():
                if re.search(pattern, line):
                    self.patterns_found.append(SecurityPattern(
                        pattern_type=description,
                        file_path=file_path,
                        line_number=line_num,
                        code_snippet=line.strip()[:120],
                        description=description
                    ))

    def scan_directory(self, directory: str, extensions: list[str] | None = None):
        """Scans a directory recursively."""
        if extensions is None:
            extensions = [".py"]

        dir_path = Path(directory)
        if not dir_path.exists():
            print(f"Directory not found: {directory}")
            return

        for file_path in dir_path.rglob("*"):
            if file_path.suffix in extensions and file_path.is_file():
                try:
                    content = file_path.read_text(encoding="utf-8")
                    self.scan_content(str(file_path), content)
                except (UnicodeDecodeError, PermissionError):
                    continue

    def generate_security_inventory(self) -> str:
        """Generates an inventory of all security mechanisms found."""
        # Group by pattern type
        by_type: dict[str, list[SecurityPattern]] = {}
        for p in self.patterns_found:
            by_type.setdefault(p.pattern_type, []).append(p)

        lines = [
            "# Security Mechanism Inventory",
            f"*Auto-generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}*",
            "",
            f"**Total patterns found:** {len(self.patterns_found)}",
            f"**Categories:** {len(by_type)}",
            "",
        ]

        for pattern_type, patterns in sorted(by_type.items()):
            lines.append(f"## {pattern_type} ({len(patterns)} instances)")
            lines.append("")

            # Group by file
            by_file: dict[str, list[SecurityPattern]] = {}
            for p in patterns:
                by_file.setdefault(p.file_path, []).append(p)

            for file_path, file_patterns in by_file.items():
                lines.append(f"### `{file_path}`")
                for p in file_patterns:
                    lines.append(f"- Line {p.line_number}: `{p.code_snippet}`")
                lines.append("")

        return "\n".join(lines)

    def coverage_summary(self) -> dict[str, int]:
        """Coverage summary by security category."""
        coverage: dict[str, int] = {}
        for p in self.patterns_found:
            coverage[p.pattern_type] = coverage.get(p.pattern_type, 0) + 1
        return dict(sorted(coverage.items(), key=lambda x: x[1], reverse=True))


# --- Example with inline code ---
generator = SecurityDocGenerator()

sample_code = '''
from security.rate_limiter import RateLimiter

class ChatEndpoint:
    def __init__(self):
        self.rate_limiter = RateLimiter(max_requests=60)
        self.guardrail = SemanticGuardrail(threshold=0.7)
        self.pii_redactor = PIIRedactor()

    async def handle_chat(self, request):
        # Rate limit check
        self.rate_limiter.check(request.user_id)
        sanitized = validate_input(request.message)
        response = await self.llm.generate(sanitized)
        clean_response = self.pii_redactor.redact(response)
        audit_log.log_security("chat_processed", request.user_id)
        return clean_response

    def authenticate(self, token: str):
        payload = jwt.decode(token, SECRET_KEY)
        if not rbac.has_permission(payload["role"], "chat"):
            raise Forbidden()
'''

generator.scan_content("app/endpoints/chat.py", sample_code)

print(generator.generate_security_inventory())
print("\nCoverage Summary:")
for category, count in generator.coverage_summary().items():
    print(f"  {category}: {count} instances")

Explanation: The scanner uses regex to detect known security patterns in the code. It doesn't replace a human auditor, but it generates an up-to-date inventory that serves as a starting point for audits and as living documentation that regenerates with each CI/CD run.


Troubleshooting

ProblemCauseSolution
The ADR registry doesn't find related decisionsowasp_risks_addressed empty in many ADRsMake the OWASP field required in SecurityDecisionRecord validation with min_length=1
The OWASP mapping shows everything as "mitigated" without real evidenceDefenseEvidence accepted without verifying test_coverageAdd a validator that requires test_coverage with a parseable numeric percentage
The documentation scanner generates false positivesRegex too broad (e.g., "permission" in comments)Limit the scan to active code lines, excluding comments and docstrings
Trade-off measurements vary a lot between runsOperating system noise in micro-benchmarksRun with iterations >= 1000 and discard the extreme percentiles (p5/p95)
The residual risk document doesn't get updatedNo periodic review processAdd the review_date to a shared calendar and configure automatic alerts

Exercises

Exercise 1: Create an ADR Deprecation System

Implement a function that lets you deprecate an ADR and create a new one that replaces it. It should update the original ADR's status to deprecated, create the new ADR referencing the previous one, and log both changes in the ADRRegistry.

See solution
from pydantic import BaseModel, Field
from enum import Enum
from datetime import datetime, timezone


class ADRStatus(str, Enum):
    PROPOSED = "proposed"
    ACCEPTED = "accepted"
    DEPRECATED = "deprecated"
    REPLACED = "replaced"


class SimpleADR(BaseModel):
    adr_id: str
    title: str
    status: ADRStatus
    date: str
    decision: str
    superseded_by: str | None = None
    supersedes: str | None = None


class SimpleRegistry(BaseModel):
    records: list[SimpleADR] = Field(default_factory=list)
    changelog: list[dict] = Field(default_factory=list)

    def add(self, adr: SimpleADR):
        self.records.append(adr)

    def find(self, adr_id: str) -> SimpleADR | None:
        for r in self.records:
            if r.adr_id == adr_id:
                return r
        return None

    def deprecate_and_replace(
        self,
        old_id: str,
        new_id: str,
        new_title: str,
        new_decision: str,
        reason: str,
    ) -> tuple[SimpleADR, SimpleADR]:
        """Deprecates an ADR and creates its replacement."""
        old = self.find(old_id)
        if not old:
            raise ValueError(f"ADR {old_id} not found")
        if old.status == ADRStatus.DEPRECATED:
            raise ValueError(f"ADR {old_id} is already deprecated")

        old.status = ADRStatus.REPLACED
        old.superseded_by = new_id

        new = SimpleADR(
            adr_id=new_id,
            title=new_title,
            status=ADRStatus.ACCEPTED,
            date=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
            decision=new_decision,
            supersedes=old_id,
        )
        self.add(new)

        self.changelog.append({
            "action": "deprecate_and_replace",
            "old_id": old_id,
            "new_id": new_id,
            "reason": reason,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        })

        return old, new

    def show_lineage(self, adr_id: str) -> list[str]:
        """Shows an ADR's replacement chain."""
        chain = []
        current = self.find(adr_id)
        # Search backward
        while current and current.supersedes:
            current = self.find(current.supersedes)
            if current:
                chain.insert(0, f"{current.adr_id} ({current.status.value})")
        # Add the current one
        current = self.find(adr_id)
        if current:
            chain.append(f"{current.adr_id} ({current.status.value})")
        # Search forward
        while current and current.superseded_by:
            current = self.find(current.superseded_by)
            if current:
                chain.append(f"{current.adr_id} ({current.status.value})")
        return chain


# --- Example ---
registry = SimpleRegistry()
registry.add(SimpleADR(
    adr_id="ADR-001",
    title="Rate Limiting by IP",
    status=ADRStatus.ACCEPTED,
    date="2026-01-15",
    decision="Rate limiting based on IP address."
))

old, new = registry.deprecate_and_replace(
    old_id="ADR-001",
    new_id="ADR-005",
    new_title="Rate Limiting by Token Count",
    new_decision="Dual rate limiting: requests/min + tokens/hour per user.",
    reason="IP-based rate limiting doesn't protect against cost attacks in LLMs."
)

print(f"Deprecated: {old.adr_id}{old.status.value}")
print(f"New: {new.adr_id}{new.status.value}")
print(f"Lineage: {' → '.join(registry.show_lineage('ADR-005'))}")

# Expected output:
# Deprecated: ADR-001 → replaced
# New: ADR-005 → accepted
# Lineage: ADR-001 (replaced) → ADR-005 (accepted)

Explanation: The system maintains bidirectional traceability between ADRs — supersedes and superseded_by let you navigate the full chain of a decision's evolution. The changelog records the reason for each deprecation for auditing.

Exercise 2: Implement an OWASPProgressTracker

Create a class that tracks the progress of the OWASP mapping over time: from not_mitigated to partially_mitigated to mitigated. It should log each status change with a date and evidence.

See solution
from pydantic import BaseModel, Field
from enum import Enum
from datetime import datetime, timezone


class MitigationStatus(str, Enum):
    NOT_MITIGATED = "not_mitigated"
    PARTIALLY_MITIGATED = "partially_mitigated"
    MITIGATED = "mitigated"


STATUS_ORDER = [
    MitigationStatus.NOT_MITIGATED,
    MitigationStatus.PARTIALLY_MITIGATED,
    MitigationStatus.MITIGATED,
]


class StatusChange(BaseModel):
    risk_id: str
    from_status: MitigationStatus
    to_status: MitigationStatus
    evidence: str
    module_source: str
    changed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class OWASPProgressTracker(BaseModel):
    """Tracks the evolution of the OWASP status throughout development."""
    current_status: dict[str, MitigationStatus] = Field(default_factory=dict)
    history: list[StatusChange] = Field(default_factory=list)

    def initialize_risks(self, risk_ids: list[str]):
        """Initializes all risks as not mitigated."""
        for risk_id in risk_ids:
            self.current_status[risk_id] = MitigationStatus.NOT_MITIGATED

    def update_status(
        self,
        risk_id: str,
        new_status: MitigationStatus,
        evidence: str,
        module_source: str,
    ):
        """Updates a risk's status with evidence."""
        old_status = self.current_status.get(risk_id, MitigationStatus.NOT_MITIGATED)
        if old_status == new_status:
            return

        change = StatusChange(
            risk_id=risk_id,
            from_status=old_status,
            to_status=new_status,
            evidence=evidence,
            module_source=module_source,
        )
        self.history.append(change)
        self.current_status[risk_id] = new_status

    def progress_report(self) -> str:
        """Generates a progress report with a timeline."""
        total = len(self.current_status)
        mitigated = sum(
            1 for s in self.current_status.values()
            if s == MitigationStatus.MITIGATED
        )
        partial = sum(
            1 for s in self.current_status.values()
            if s == MitigationStatus.PARTIALLY_MITIGATED
        )

        lines = [
            "# OWASP Progress Report",
            f"Mitigated: {mitigated}/{total}",
            f"Partially: {partial}/{total}",
            f"Not mitigated: {total - mitigated - partial}/{total}",
            f"Coverage: {((mitigated + partial) / total * 100) if total else 0:.0f}%",
            "",
            "## Timeline",
        ]
        for change in self.history:
            lines.append(
                f"- [{change.changed_at.strftime('%Y-%m-%d')}] "
                f"{change.risk_id}: {change.from_status.value}{change.to_status.value} "
                f"({change.module_source})"
            )
        return "\n".join(lines)


# --- Example: simulate progress during development ---
tracker = OWASPProgressTracker()
tracker.initialize_risks([f"LLM{str(i).zfill(2)}" for i in range(1, 11)])

# Simulation: each module mitigates certain risks
tracker.update_status("LLM01", MitigationStatus.PARTIALLY_MITIGATED,
    "Input validation pipeline implemented", "Module 2")
tracker.update_status("LLM02", MitigationStatus.PARTIALLY_MITIGATED,
    "Basic output filter implemented", "Module 3")
tracker.update_status("LLM01", MitigationStatus.MITIGATED,
    "Semantic guardrail added as a second layer", "Module 4")
tracker.update_status("LLM04", MitigationStatus.MITIGATED,
    "Dual rate limiter (requests + tokens)", "Module 5")
tracker.update_status("LLM06", MitigationStatus.MITIGATED,
    "PII redactor with NER patterns", "Module 3")
tracker.update_status("LLM02", MitigationStatus.MITIGATED,
    "Output filter with code injection detection", "Module 3")

print(tracker.progress_report())

# Expected output:
# OWASP Progress Report
# Mitigated: 4/10
# Partially: 0/10
# Not mitigated: 6/10
# Coverage: 40%
# ...

Explanation: The tracker keeps a complete history of how each risk evolved during development. This is invaluable for audits: it shows the mitigation was incremental and each step is backed by evidence traceable to a specific module.

Exercise 3: Build a SecurityDocValidator

Implement a validator that takes an OWASPFinalMapping and an ADRRegistry and verifies that: each mitigated risk has at least one defense with evidence, each referenced defense has a corresponding ADR, and there are no orphan ADRs without a reference from the mapping.

See solution
from pydantic import BaseModel, Field
from enum import Enum


class ValidationSeverity(str, Enum):
    ERROR = "error"
    WARNING = "warning"
    INFO = "info"


class ValidationIssue(BaseModel):
    severity: ValidationSeverity
    category: str
    message: str


class SecurityDocValidator(BaseModel):
    """Validates consistency between the OWASP mapping and the ADR registry."""
    issues: list[ValidationIssue] = Field(default_factory=list)

    def validate(
        self,
        owasp_risks: list[dict],
        adr_records: list[dict],
    ) -> list[ValidationIssue]:
        """
        Validates consistency between the mapping and the ADRs.
        owasp_risks: list with risk_id, status, defense_names
        adr_records: list with adr_id, title, owasp_risks_addressed
        """
        self.issues = []

        # 1. Each mitigated risk must have at least one defense
        for risk in owasp_risks:
            if risk["status"] == "mitigated" and not risk.get("defense_names"):
                self.issues.append(ValidationIssue(
                    severity=ValidationSeverity.ERROR,
                    category="missing_evidence",
                    message=f"{risk['risk_id']}: marked as mitigated but has no defense evidence"
                ))

        # 2. Each defense should have a corresponding ADR
        all_defense_names = set()
        for risk in owasp_risks:
            for name in risk.get("defense_names", []):
                all_defense_names.add(name)

        adr_titles = {adr["title"] for adr in adr_records}
        for defense in all_defense_names:
            # Look for an ADR whose title contains the defense or vice versa
            has_adr = any(
                defense.lower() in title.lower() or title.lower() in defense.lower()
                for title in adr_titles
            )
            if not has_adr:
                self.issues.append(ValidationIssue(
                    severity=ValidationSeverity.WARNING,
                    category="missing_adr",
                    message=f"Defense '{defense}' has no corresponding ADR"
                ))

        # 3. Orphan ADRs: ADRs that don't map to any OWASP risk
        all_addressed_risks = set()
        for adr in adr_records:
            for risk_id in adr.get("owasp_risks_addressed", []):
                all_addressed_risks.add(risk_id)

        mapping_risk_ids = {r["risk_id"] for r in owasp_risks}
        orphan_risks = all_addressed_risks - mapping_risk_ids
        for orphan in orphan_risks:
            self.issues.append(ValidationIssue(
                severity=ValidationSeverity.WARNING,
                category="orphan_adr_reference",
                message=f"ADR references {orphan} but it's not in the OWASP mapping"
            ))

        # 4. Info: risks with no ADR at all
        risks_in_adrs = set()
        for adr in adr_records:
            risks_in_adrs.update(adr.get("owasp_risks_addressed", []))

        for risk in owasp_risks:
            if risk["risk_id"] not in risks_in_adrs and risk["status"] != "not_applicable":
                self.issues.append(ValidationIssue(
                    severity=ValidationSeverity.INFO,
                    category="no_adr_coverage",
                    message=f"{risk['risk_id']}: no ADR addresses this risk"
                ))

        return self.issues

    def summary(self) -> str:
        errors = sum(1 for i in self.issues if i.severity == ValidationSeverity.ERROR)
        warnings = sum(1 for i in self.issues if i.severity == ValidationSeverity.WARNING)
        infos = sum(1 for i in self.issues if i.severity == ValidationSeverity.INFO)
        lines = [
            "Validation Summary",
            f"  Errors:   {errors}",
            f"  Warnings: {warnings}",
            f"  Info:     {infos}",
            "",
        ]
        for issue in self.issues:
            icon = {"error": "❌", "warning": "⚠️", "info": "ℹ️"}
            lines.append(f"  {icon[issue.severity.value]} [{issue.category}] {issue.message}")
        return "\n".join(lines)


# --- Example ---
validator = SecurityDocValidator()

owasp_risks = [
    {"risk_id": "LLM01", "status": "mitigated", "defense_names": ["Input Sanitization Pipeline"]},
    {"risk_id": "LLM02", "status": "mitigated", "defense_names": []},  # Error: no evidence
    {"risk_id": "LLM03", "status": "not_applicable", "defense_names": []},
    {"risk_id": "LLM04", "status": "mitigated", "defense_names": ["Rate Limiter"]},
]

adr_records = [
    {"adr_id": "ADR-001", "title": "Input Sanitization Multi-Layer",
     "owasp_risks_addressed": ["LLM01"]},
    {"adr_id": "ADR-002", "title": "Rate Limiting by Token",
     "owasp_risks_addressed": ["LLM04", "LLM99"]},  # LLM99 is orphan
]

validator.validate(owasp_risks, adr_records)
print(validator.summary())

# Expected output:
# Validation Summary
#   Errors:   1
#   Warnings: 3
#   Info:     1
#
#   ❌ [missing_evidence] LLM02: marked as mitigated but has no defense evidence
#   ⚠️ [missing_adr] Defense 'Rate Limiter' has no corresponding ADR
#   ⚠️ [missing_adr] Defense 'Input Sanitization Pipeline' has no corresponding ADR
#   ⚠️ [orphan_adr_reference] ADR references LLM99 but it's not in the OWASP mapping
#   ℹ️ [no_adr_coverage] LLM02: no ADR addresses this risk

Explanation: The validator works like a linter for your security documentation — it finds inconsistencies before an auditor does. Errors are serious problems (claims without evidence), warnings are potential gaps, and infos are improvement opportunities.

Exercise 4: Generate an automatic Security README

Create a function that takes an OWASPFinalMapping, an ADRRegistry, and a ResidualRiskRegister and generates a complete project README.md that includes: executive summary, OWASP coverage table, list of key decisions, and accepted risks.

See solution
from datetime import datetime, timezone


def generate_security_readme(
    project_name: str,
    owasp_data: list[dict],
    adr_data: list[dict],
    residual_risks: list[dict],
) -> str:
    """Generates a complete project security README."""
    # Compute metrics
    total_risks = len(owasp_data)
    applicable = [r for r in owasp_data if r["status"] != "not_applicable"]
    mitigated = [r for r in applicable if r["status"] == "mitigated"]
    partial = [r for r in applicable if r["status"] == "partially_mitigated"]
    coverage = (len(mitigated) + len(partial)) / len(applicable) * 100 if applicable else 0

    active_adrs = [a for a in adr_data if a.get("status") == "accepted"]
    high_risks = [r for r in residual_risks if r.get("impact") in ("major", "catastrophic")]

    readme = f"""# {project_name} — Security Documentation

> Auto-generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}

## Executive Summary

This document describes the security measures implemented in **{project_name}**.
The system covers **{len(mitigated)}/{len(applicable)}** OWASP LLM Top 10 risks fully
mitigated, with an overall coverage of **{coverage:.0f}%**.

**{len(active_adrs)}** security architecture decisions have been made and documented as ADRs,
and **{len(residual_risks)}** accepted residual risks have been identified,
of which **{len(high_risks)}** require priority monitoring.

## OWASP LLM Top 10 Coverage

| Risk ID | Name | Status | Defenses |
|---------|--------|--------|----------|
"""
    status_icons = {
        "mitigated": "🟢",
        "partially_mitigated": "🟡",
        "not_mitigated": "🔴",
        "not_applicable": "⚪",
    }

    for r in owasp_data:
        icon = status_icons.get(r["status"], "?")
        defenses = len(r.get("defense_names", []))
        readme += f"| {r['risk_id']} | {r['name']} | {icon} {r['status']} | {defenses} |\n"

    readme += f"""
## Security Decisions (ADRs)

| ID | Decision | Status | OWASP |
|----|----------|--------|-------|
"""
    for adr in adr_data:
        owasp = ", ".join(adr.get("owasp_risks", [])) or "—"
        readme += f"| {adr['adr_id']} | {adr['title']} | {adr.get('status', 'accepted')} | {owasp} |\n"

    readme += f"""
## Accepted Residual Risks

| Risk | Title | Likelihood | Impact |
|------|--------|------------|--------|
"""
    for rr in residual_risks:
        readme += f"| {rr['risk_id']} | {rr['title']} | {rr['likelihood']} | {rr['impact']} |\n"

    readme += """
## How to Contribute to Security

1. Before adding a new defense, create an ADR documenting the decision
2. Update the OWASP mapping when you implement a new mitigation
3. Run the documentation scanner after significant changes
4. Review residual risks quarterly

## Contact

To report vulnerabilities: security@example.com
"""
    return readme


# --- Example ---
readme = generate_security_readme(
    project_name="SecureAI Chat System",
    owasp_data=[
        {"risk_id": "LLM01", "name": "Prompt Injection", "status": "mitigated",
         "defense_names": ["Input Sanitization", "Semantic Guardrail"]},
        {"risk_id": "LLM02", "name": "Insecure Output", "status": "mitigated",
         "defense_names": ["Output Filter"]},
        {"risk_id": "LLM03", "name": "Training Data Poisoning", "status": "not_applicable",
         "defense_names": []},
        {"risk_id": "LLM04", "name": "Model DoS", "status": "mitigated",
         "defense_names": ["Rate Limiter", "Cost Controller"]},
        {"risk_id": "LLM05", "name": "Supply Chain", "status": "partially_mitigated",
         "defense_names": ["Dependabot"]},
        {"risk_id": "LLM06", "name": "Info Disclosure", "status": "mitigated",
         "defense_names": ["PII Redactor"]},
        {"risk_id": "LLM07", "name": "Insecure Plugin", "status": "partially_mitigated",
         "defense_names": ["Tool Permissions"]},
        {"risk_id": "LLM08", "name": "Excessive Agency", "status": "mitigated",
         "defense_names": ["HITL Guard"]},
        {"risk_id": "LLM09", "name": "Overreliance", "status": "partially_mitigated",
         "defense_names": ["Disclaimers"]},
        {"risk_id": "LLM10", "name": "Model Theft", "status": "not_applicable",
         "defense_names": []},
    ],
    adr_data=[
        {"adr_id": "ADR-001", "title": "Input Sanitization Multi-Layer",
         "status": "accepted", "owasp_risks": ["LLM01"]},
        {"adr_id": "ADR-002", "title": "Rate Limiting by Token",
         "status": "accepted", "owasp_risks": ["LLM04"]},
    ],
    residual_risks=[
        {"risk_id": "RR-001", "title": "Model Poisoning", "likelihood": "unlikely", "impact": "major"},
        {"risk_id": "RR-002", "title": "Zero-day Injection", "likelihood": "possible", "impact": "moderate"},
    ]
)

print(readme)

# Output: complete README with all sections

Explanation: The README is generated from structured data, not free text. This means that each time the OWASP mapping or the ADRs are updated, the README can be regenerated automatically and will always reflect the current state of the system.


Summary

  • 📋 Architecture Decision Records (ADR) capture the "why" behind each security decision, protecting against accidental changes that weaken the defenses
  • 🔍 SecurityDecisionRecord extends the traditional ADR with a security category, OWASP mapping, and trade-offs typed by impact area
  • ⚖️ Documenting trade-offs with real measurements (not estimates) prevents "optimizations" that remove defenses without understanding their cost
  • 🗺️ The OWASP Final Mapping with OWASPFinalMapping is the central audit artifact: each LLM01-LLM10 risk with status, defenses, and traceable evidence
  • ⚠️ Honest documentation of residual risks with ResidualRiskRegister shows informed decisions, not oversights
  • 🤖 Automatic documentation generation with SecurityDocGenerator eliminates staleness by scanning the code directly
  • 🔗 Cross-validation between the OWASP mapping and ADRs detects inconsistencies before an auditor finds them
  • 📊 An auto-generated Security README provides executive visibility into the security state without continuous manual effort

Module completed: You've built a comprehensive AI security system — from input validation to incident response and auditable documentation.


Additional resources

  1. Architecture Decision Records — Official repository of the ADR format with templates and tools
  2. OWASP LLM Top 10 — Official list of security risks for LLMs
  3. Michael Nygard — Documenting Architecture Decisions — The original article that introduced ADRs
  4. Google DORA — Documentation Practices — Research on the impact of documentation on productivity
  5. Thoughtworks Tech Radar — ADR Tools — Analysis of ADR tools by Thoughtworks
  6. NIST AI Risk Management Framework — Federal framework for AI risk management
  7. EU AI Act — Documentation Requirements — Documentation requirements of the European AI Act

Created: March 2026 Version: 1.0