Módulo 8: Proyecto Integrador — Secured AI System
6. Documentación de Decisiones de Seguridad
Descripción
El código sin documentación es código con fecha de expiración. Puedes construir el sistema de seguridad más sofisticado del mundo, pero si nadie entiende por qué se tomaron ciertas decisiones, el primer desarrollador nuevo que toque el código va a romper una defensa sin saberlo. La documentación de seguridad no es burocracia — es la memoria institucional que protege al sistema cuando tú no estás.
Las decisiones de seguridad son especialmente difíciles de documentar porque involucran trade-offs invisibles. ¿Por qué elegiste rate limiting por token en vez de por IP? ¿Por qué el guardrail rechaza con un score de 0.7 y no 0.8? ¿Por qué usas bcrypt y no argon2? Sin el contexto documentado, estas decisiones parecen arbitrarias y se cambian sin entender las consecuencias.
En esta cápsula vas a implementar un sistema completo de documentación: Architecture Decision Records (ADR) para decisiones de seguridad, un mapping final OWASP que muestra el estado de cada riesgo LLM01-LLM10 con evidencia de cada módulo, documentación honesta de riesgos residuales, y generación automática de documentación a partir del código.
Architecture Decision Records (ADR) para seguridad
Qué es un ADR
Un Architecture Decision Record es un documento corto que captura una decisión técnica significativa junto con su contexto y consecuencias. A diferencia de los comentarios en código, un ADR documenta el "por qué" a nivel arquitectónico — no la implementación, sino la decisión que la motivó.
Template ADR para seguridad AI
# ADR-{número}: {título}
**Estado:** {propuesto | aceptado | deprecado | reemplazado}
**Fecha:** {YYYY-MM-DD}
**Autores:** {nombres}
## Contexto
{Qué problema estamos resolviendo y qué restricciones existen}
## Decisión
{Qué decidimos hacer}
## Consecuencias
{Qué implica esta decisión — positivo y negativo}
## Alternativas Consideradas
{Qué otras opciones evaluamos y por qué las descartamos}
Ejemplo: 3 ADRs para un sistema AI seguro
ADR-001: Input Sanitization Multi-Layer
# ADR-001: Input Sanitization con Pipeline Multi-Capa
**Estado:** Aceptado
**Fecha:** 2026-02-15
**Autores:** Security Team
## Contexto
Los prompts de usuario pueden contener inyecciones directas,
encoding tricks (Base64, ROT13), o ataques multi-turno.
Una sola capa de sanitización no es suficiente — los atacantes
evolucionan las técnicas más rápido de lo que actualizamos reglas.
## Decisión
Implementar un pipeline de 3 capas:
1. Decode layer: normaliza encodings (Base64, URL, Unicode)
2. Pattern layer: detecta patrones de injection conocidos
3. Semantic layer: clasifica intención con embedding similarity
## Consecuencias
Positivas:
- Defensa en profundidad contra múltiples vectores
- Cada capa es independiente y testeable
Negativas:
- Latencia adicional de ~50ms por capa
- Complejidad operacional de mantener 3 sistemas
## Alternativas Consideradas
- Solo regex: descartado por falsos negativos en ataques semánticos
- Solo embeddings: descartado por alto costo computacional
- WAF externo: descartado por no entender contexto AI
ADR-002: Rate Limiting por Token
# ADR-002: Rate Limiting Basado en Token Count
**Estado:** Aceptado
**Fecha:** 2026-02-20
**Autores:** Backend Team
## Contexto
El rate limiting tradicional por requests/minuto no protege
contra ataques de costo en LLMs. Un solo request con un prompt
de 100K tokens cuesta más que 100 requests cortas.
## Decisión
Implementar rate limiting dual:
- Por requests: max 60/min por usuario
- Por tokens: max 50,000 tokens/hora por usuario
## Consecuencias
Positivas:
- Protección directa contra cost attacks
- Usuarios legítimos raramente alcanzan el límite de tokens
Negativas:
- Requiere estimar tokens antes de enviar al LLM
- Token counting añade ~5ms de latencia
## Alternativas Consideradas
- Solo request-based: no protege contra prompts largos
- Budget cap mensual: demasiado lento para detectar spikes
- Pre-auth token estimation: rechazado por complejidad
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):
"""Una alternativa considerada y descartada."""
name: str
description: str
reason_rejected: str
class Consequence(BaseModel):
"""Consecuencia positiva o negativa de la decisión."""
description: str
is_positive: bool
impact_area: str # "performance", "security", "cost", "complexity"
class SecurityDecisionRecord(BaseModel):
"""ADR especializado para decisiones de seguridad AI."""
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:
"""Resume el balance de trade-offs de la decisión."""
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)} beneficios, "
f"{len(negatives)} costos"
)
def to_markdown(self) -> str:
"""Genera el ADR en formato markdown."""
lines = [
f"# {self.adr_id}: {self.title}",
"",
f"**Estado:** {self.status.value}",
f"**Fecha:** {self.date}",
f"**Autores:** {', '.join(self.authors)}",
f"**Categoría:** {self.category.value}",
f"**OWASP:** {', '.join(self.owasp_risks_addressed) or 'N/A'}",
"",
"## Contexto",
self.context,
"",
"## Decisión",
self.decision,
"",
"## Consecuencias",
]
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(["", "## Alternativas Consideradas"])
for alt in self.alternatives_considered:
lines.append(f"### {alt.name}")
lines.append(f"{alt.description}")
lines.append(f"**Razón de descarte:** {alt.reason_rejected}")
lines.append("")
if self.related_adrs:
lines.extend(["", "## ADRs Relacionados"])
for related in self.related_adrs:
lines.append(f"- {related}")
return "\n".join(lines)
class ADRRegistry(BaseModel):
"""Registro centralizado de todas las decisiones de seguridad."""
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]:
"""Encuentra ADRs que abordan un riesgo OWASP específico."""
return [
r for r in self.records
if risk_id in r.owasp_risks_addressed
]
def generate_index(self) -> str:
"""Genera un índice markdown de todos los ADRs."""
lines = ["# Índice de Decisiones de Seguridad", ""]
lines.append("| ID | Título | Estado | Categoría | 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)
# --- Ejemplo de uso ---
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 pueden contener injections directas y encoding tricks.",
decision="Pipeline de 3 capas: decode, pattern, semantic.",
consequences=[
Consequence(description="Defensa en profundidad", is_positive=True, impact_area="security"),
Consequence(description="~150ms latencia adicional", is_positive=False, impact_area="performance"),
],
alternatives_considered=[
Alternative(name="Solo regex", description="Pattern matching con expresiones regulares",
reason_rejected="Falsos negativos en ataques semánticos"),
],
owasp_risks_addressed=["LLM01", "LLM02"],
)
adr2 = SecurityDecisionRecord(
adr_id="ADR-002",
title="Rate Limiting por Token Count",
status=ADRStatus.ACCEPTED,
date="2026-02-20",
authors=["backend-team"],
category=SecurityCategory.RATE_LIMITING,
context="Rate limiting por requests no protege contra cost attacks.",
decision="Dual rate limiting: requests/min + tokens/hora.",
consequences=[
Consequence(description="Protección contra cost attacks", is_positive=True, impact_area="cost"),
Consequence(description="~5ms latencia por 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())
Explicación: SecurityDecisionRecord extiende el concepto de ADR con campos específicos para seguridad AI: categoría, mapping OWASP, y consecuencias tipadas por área de impacto. El ADRRegistry permite buscar decisiones por riesgo OWASP, lo que conecta directamente con el mapping final.
Documentando trade-offs
Cada decisión de seguridad tiene un costo. Documentar estos trade-offs con datos reales previene que alguien "optimice" el sistema eliminando una defensa sin entender su propósito.
import time
from pydantic import BaseModel, Field
class TradeOffMeasurement(BaseModel):
"""Medición concreta de un trade-off de seguridad."""
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" Sin defensa: {self.value_without_defense:.1f} {self.unit}\n"
f" Con defensa: {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:
"""Mide el overhead real de una defensa en milisegundos."""
# Medir sin defensa
start = time.perf_counter()
for _ in range(iterations):
operation_without()
time_without = ((time.perf_counter() - start) / iterations) * 1000
# Medir con defensa
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,
)
# --- Ejemplo: medir overhead de sanitización ---
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())
# --- Trade-off documentation completa ---
class TradeOffRegistry(BaseModel):
"""Registro de todos los trade-offs de seguridad medidos."""
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",
"",
"| Defensa | Métrica | Sin | Con | 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(["", "## Defensas que exceden el threshold"])
for m in unacceptable:
lines.append(f"- **{m.defense_name}**: revisar implementación o ajustar 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())
Explicación: measure_defense_overhead produce datos reales, no estimaciones. El TradeOffRegistry genera un reporte que justifica cada defensa con números concretos. Cuando el overhead excede el threshold, el reporte lo señala explícitamente para que el equipo evalúe si optimizar o ajustar el threshold.
OWASP Mapping Final
El mapping OWASP es el artefacto central de auditoría: muestra qué riesgos del OWASP LLM Top 10 están mitigados, con qué defensas, y qué evidencia respalda esa mitigación.
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):
"""Evidencia de que una defensa existe y funciona."""
defense_name: str
module_source: str # "Module 2", "Module 5", etc.
implementation_file: str
test_coverage: str
last_verified: str
class ResidualRisk(BaseModel):
"""Riesgo que permanece después de aplicar las mitigaciones."""
description: str
likelihood: str # "low", "medium", "high"
impact: str
mitigation_plan: str
class OWASPRiskEntry(BaseModel):
"""Estado de un riesgo OWASP LLM específico."""
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):
"""Mapping completo OWASP LLM Top 10 con evidencia."""
project_name: str
assessment_date: str
assessor: str
risks: list[OWASPRiskEntry] = Field(default_factory=list)
@computed_field
@property
def overall_coverage(self) -> float:
"""Porcentaje de riesgos mitigados o parcialmente mitigados."""
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:
"""Genera reporte de estado OWASP en formato markdown."""
status_icons = {
MitigationStatus.MITIGATED: "🟢",
MitigationStatus.PARTIALLY_MITIGATED: "🟡",
MitigationStatus.NOT_MITIGATED: "🔴",
MitigationStatus.NOT_APPLICABLE: "⚪",
}
lines = [
f"# OWASP LLM Top 10 — Assessment Report",
f"**Proyecto:** {self.project_name}",
f"**Fecha:** {self.assessment_date}",
f"**Evaluador:** {self.assessor}",
f"**Cobertura global:** {self.overall_coverage:.0f}%",
f"**Fully mitigated:** {self.fully_mitigated_count}/{len(self.risks)}",
"",
"## Estado por Riesgo",
"",
"| Risk ID | Nombre | Estado | Defensas | Riesgo Residual |",
"|---------|--------|--------|----------|-----------------|",
]
for r in self.risks:
icon = status_icons[r.status]
residual = "Sí" 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} |"
)
# Detalle por riesgo
lines.extend(["", "## Detalle por Riesgo", ""])
for r in self.risks:
lines.append(f"### {r.risk_id}: {r.risk_name}")
lines.append(f"**Estado:** {status_icons[r.status]} {r.status.value}")
lines.append(f"\n{r.description}\n")
if r.defense_layers:
lines.append("**Defensas implementadas:**")
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**Riesgos residuales:**")
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**Notas:** {r.notes}")
lines.append("")
return "\n".join(lines)
# --- Construir el mapping final ---
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="Manipulación del modelo mediante instrucciones inyectadas en el 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 no cubiertas por el decoder",
likelihood="low",
impact="medium",
mitigation_plan="Actualizar dataset adversarial mensualmente con nuevas técnicas"
)
]
))
mapping.add_risk(OWASPRiskEntry(
risk_id="LLM02",
risk_name="Insecure Output Handling",
description="Output del modelo usado sin sanitizar en contextos downstream.",
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="Datos de entrenamiento manipulados para alterar el comportamiento del modelo.",
status=MitigationStatus.NOT_APPLICABLE,
notes="Usamos modelos pre-entrenados de providers (OpenAI/Anthropic). No hacemos fine-tuning."
))
mapping.add_risk(OWASPRiskEntry(
risk_id="LLM04",
risk_name="Model Denial of Service",
description="Ataques que agotan recursos del modelo o generan costos excesivos.",
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 con 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="Dependencias, plugins o modelos de terceros comprometidos.",
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 validación de integridad de modelos descargados",
likelihood="low",
impact="high",
mitigation_plan="Implementar checksum verification para model downloads"
)
]
))
mapping.add_risk(OWASPRiskEntry(
risk_id="LLM06",
risk_name="Sensitive Information Disclosure",
description="El modelo revela información confidencial en sus respuestas.",
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="Plugins/tools del LLM con permisos excesivos o sin validación.",
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="Tools de terceros no tienen sandboxing completo",
likelihood="medium",
impact="high",
mitigation_plan="Implementar container-based sandboxing para tool execution"
)
]
))
mapping.add_risk(OWASPRiskEntry(
risk_id="LLM08",
risk_name="Excessive Agency",
description="El modelo toma acciones autónomas sin confirmación del usuario.",
status=MitigationStatus.MITIGATED,
defense_layers=[
DefenseEvidence(
defense_name="Human-in-the-loop para acciones destructivas",
module_source="Module 5",
implementation_file="security/action_guard.py",
test_coverage="100% — todas las acciones destructivas requieren confirmación",
last_verified="2026-03-13"
),
],
))
mapping.add_risk(OWASPRiskEntry(
risk_id="LLM09",
risk_name="Overreliance",
description="Usuarios confían ciegamente en outputs del modelo sin verificar.",
status=MitigationStatus.PARTIALLY_MITIGATED,
defense_layers=[
DefenseEvidence(
defense_name="Confidence disclaimers en outputs",
module_source="Module 3",
implementation_file="security/output_disclaimer.py",
test_coverage="70% — UI integration tests",
last_verified="2026-03-08"
),
],
notes="Mitigación limitada — depende del comportamiento del usuario."
))
mapping.add_risk(OWASPRiskEntry(
risk_id="LLM10",
risk_name="Model Theft",
description="Extracción o robo del modelo a través de la API.",
status=MitigationStatus.NOT_APPLICABLE,
notes="Usamos modelos de terceros via API. El modelo no es nuestro activo."
))
print(mapping.generate_status_report())
Explicación: El mapping es exhaustivo: cada riesgo OWASP tiene un estado, defensas con evidencia rastreable a módulos específicos, y riesgos residuales documentados honestamente. NOT_APPLICABLE es un estado válido — no todo aplica a tu arquitectura.
Riesgos residuales
La documentación honesta de lo que NO está cubierto es tan importante como documentar lo que sí. Un falso sentido de seguridad es más peligroso que saber dónde están los gaps.
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):
"""Un riesgo que no está completamente mitigado."""
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):
"""Registro de riesgos residuales aceptados conscientemente."""
risks: list[ResidualRiskEntry] = Field(default_factory=list)
def add(self, risk: ResidualRiskEntry):
self.risks.append(risk)
def risk_matrix(self) -> str:
"""Genera una matriz de riesgo en texto."""
lines = [
"# Residual Risk Matrix",
"",
"| Risk ID | Título | Likelihood | Impact | Acción |",
"|---------|--------|------------|--------|--------|",
]
for r in self.risks:
# Riesgos likely + major/catastrophic requieren acción
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:
"""Genera documento de aceptación formal de riesgos."""
lines = [
"# Documento de Aceptación de Riesgos Residuales",
"",
"Los siguientes riesgos han sido identificados, evaluados,",
"y aceptados conscientemente por el equipo de seguridad.",
"",
]
for r in self.risks:
lines.extend([
f"## {r.risk_id}: {r.title}",
f"**Categoría:** {r.category}",
f"**Likelihood:** {r.likelihood.value} | **Impact:** {r.impact.value}",
"",
f"**Descripción:** {r.description}",
"",
f"**¿Por qué no está mitigado?** {r.why_not_mitigated}",
"",
f"**Razón de aceptación:** {r.acceptance_rationale}",
"",
f"**Monitoreo:** {r.monitoring_strategy}",
"",
f"**Próxima revisión:** {r.review_date}",
"",
"---",
"",
])
return "\n".join(lines)
# --- Riesgos residuales del proyecto ---
register = ResidualRiskRegister()
register.add(ResidualRiskEntry(
risk_id="RR-001",
title="Model Poisoning via Fine-tuning Data",
description="Si en el futuro se hace fine-tuning, los datos de entrenamiento podrían ser envenenados.",
category="Training Data",
likelihood=RiskLikelihood.UNLIKELY,
impact=RiskImpact.MAJOR,
why_not_mitigated="Actualmente no hacemos fine-tuning. La mitigación requiere data validation pipeline.",
acceptance_rationale="No aplica en la arquitectura actual. Se revisará si se adopta fine-tuning.",
monitoring_strategy="Revisar si el equipo planea fine-tuning en cada quarterly review.",
review_date="2026-06-01"
))
register.add(ResidualRiskEntry(
risk_id="RR-002",
title="Novel Zero-day Prompt Injection",
description="Técnica de injection completamente nueva que no está en el dataset adversarial.",
category="Prompt Injection",
likelihood=RiskLikelihood.POSSIBLE,
impact=RiskImpact.MODERATE,
why_not_mitigated="Imposible prevenir ataques desconocidos. Defensa basada en detección post-facto.",
acceptance_rationale="Las 3 capas de defensa cubren patrones conocidos. El anomaly detector cubre la brecha.",
monitoring_strategy="Actualizar dataset adversarial mensualmente. Suscripción a AI security feeds.",
review_date="2026-04-15"
))
register.add(ResidualRiskEntry(
risk_id="RR-003",
title="Supply Chain Attack en Dependencia de LLM Provider",
description="El provider de LLM (OpenAI/Anthropic) sufre un compromiso que afecta nuestras respuestas.",
category="Supply Chain",
likelihood=RiskLikelihood.RARE,
impact=RiskImpact.CATASTROPHIC,
why_not_mitigated="No tenemos control sobre la infraestructura del provider.",
acceptance_rationale="Riesgo inherente a usar servicios de terceros. Mitigado parcialmente por output filtering.",
monitoring_strategy="Monitorear status pages de providers. Alertas en cambios de comportamiento del modelo.",
review_date="2026-06-01"
))
print(register.risk_matrix())
print()
print(register.generate_acceptance_document())
Explicación: Los riesgos residuales se documentan con una razón explícita de por qué no están mitigados y una decisión consciente de aceptación. Esto protege al equipo: si el riesgo se materializa, la documentación demuestra que fue una decisión informada, no un olvido.
Generando documentación automática
La documentación manual se desactualiza rápidamente. Este scanner analiza el código fuente buscando patrones de seguridad y genera documentación actualizada automáticamente.
import re
from pydantic import BaseModel, Field
from pathlib import Path
from datetime import datetime, timezone
class SecurityPattern(BaseModel):
"""Un patrón de seguridad detectado en el código."""
pattern_type: str
file_path: str
line_number: int
code_snippet: str
description: str
class SecurityDocGenerator(BaseModel):
"""Genera documentación de seguridad escaneando código fuente."""
patterns_found: list[SecurityPattern] = Field(default_factory=list)
# Patrones que buscamos en el código
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):
"""Escanea el contenido de un archivo buscando patrones de seguridad."""
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):
"""Escanea un directorio recursivamente."""
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:
"""Genera un inventario de todos los mecanismos de seguridad encontrados."""
# Agrupar por tipo de patrón
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("")
# Agrupar por archivo
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]:
"""Resumen de cobertura por categoría de seguridad."""
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))
# --- Ejemplo con código inline ---
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")
Explicación: El scanner usa regex para detectar patrones conocidos de seguridad en el código. No reemplaza a un auditor humano, pero genera un inventario actualizado que sirve como punto de partida para auditorías y como documentación viva que se regenera con cada CI/CD run.
Troubleshooting
| Problema | Causa | Solución |
|---|---|---|
| El ADR registry no encuentra decisiones relacionadas | owasp_risks_addressed vacío en muchos ADRs | Hace obligatorio el campo OWASP en la validación de SecurityDecisionRecord con min_length=1 |
| El OWASP mapping muestra todo como "mitigated" sin evidencia real | DefenseEvidence aceptada sin verificar test_coverage | Agrega un validador que requiera test_coverage con un porcentaje numérico parseable |
| El scanner de documentación genera falsos positivos | Regex demasiado amplios (e.g., "permission" en comentarios) | Limita el scan a líneas de código activo excluyendo comentarios y docstrings |
| Trade-off measurements varían mucho entre ejecuciones | Ruido del sistema operativo en micro-benchmarks | Ejecuta con iterations >= 1000 y descarta los percentiles extremos (p5/p95) |
| El documento de riesgos residuales no se actualiza | No hay proceso de revisión periódica | Agrega el review_date a un calendario compartido y configura alertas automáticas |
Ejercicios
Ejercicio 1: Crear un ADR Deprecation System
Implementa una función que permita deprecar un ADR y crear uno nuevo que lo reemplace. Debe actualizar el estado del ADR original a deprecated, crear el nuevo ADR con referencia al anterior, y registrar ambos cambios en el ADRRegistry.
Ver solución
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]:
"""Depreca un ADR y crea su reemplazo."""
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]:
"""Muestra la cadena de reemplazos de un ADR."""
chain = []
current = self.find(adr_id)
# Buscar hacia atrás
while current and current.supersedes:
current = self.find(current.supersedes)
if current:
chain.insert(0, f"{current.adr_id} ({current.status.value})")
# Agregar el actual
current = self.find(adr_id)
if current:
chain.append(f"{current.adr_id} ({current.status.value})")
# Buscar hacia adelante
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
# --- Ejemplo ---
registry = SimpleRegistry()
registry.add(SimpleADR(
adr_id="ADR-001",
title="Rate Limiting por IP",
status=ADRStatus.ACCEPTED,
date="2026-01-15",
decision="Rate limiting basado en IP address."
))
old, new = registry.deprecate_and_replace(
old_id="ADR-001",
new_id="ADR-005",
new_title="Rate Limiting por Token Count",
new_decision="Dual rate limiting: requests/min + tokens/hora por usuario.",
reason="IP-based rate limiting no protege contra cost attacks en 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'))}")
# Output esperado:
# Deprecated: ADR-001 → replaced
# New: ADR-005 → accepted
# Lineage: ADR-001 (replaced) → ADR-005 (accepted)
Explicación: El sistema mantiene trazabilidad bidireccional entre ADRs — supersedes y superseded_by permiten navegar la cadena completa de evolución de una decisión. El changelog registra la razón de cada deprecación para auditoría.
Ejercicio 2: Implementar un OWASPProgressTracker
Crea una clase que rastree el progreso del mapping OWASP a lo largo del tiempo: de not_mitigated a partially_mitigated a mitigated. Debe registrar cada cambio de estado con fecha y evidencia.
Ver solución
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):
"""Rastrea la evolución del estado OWASP a lo largo del desarrollo."""
current_status: dict[str, MitigationStatus] = Field(default_factory=dict)
history: list[StatusChange] = Field(default_factory=list)
def initialize_risks(self, risk_ids: list[str]):
"""Inicializa todos los riesgos como no mitigados."""
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,
):
"""Actualiza el estado de un riesgo con evidencia."""
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:
"""Genera reporte de progreso con 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)
# --- Ejemplo: simular progreso durante el desarrollo ---
tracker = OWASPProgressTracker()
tracker.initialize_risks([f"LLM{str(i).zfill(2)}" for i in range(1, 11)])
# Simulación: cada módulo mitiga ciertos riesgos
tracker.update_status("LLM01", MitigationStatus.PARTIALLY_MITIGATED,
"Input validation pipeline implementado", "Module 2")
tracker.update_status("LLM02", MitigationStatus.PARTIALLY_MITIGATED,
"Output filter básico implementado", "Module 3")
tracker.update_status("LLM01", MitigationStatus.MITIGATED,
"Semantic guardrail añadido como segunda capa", "Module 4")
tracker.update_status("LLM04", MitigationStatus.MITIGATED,
"Dual rate limiter (requests + tokens)", "Module 5")
tracker.update_status("LLM06", MitigationStatus.MITIGATED,
"PII redactor con NER patterns", "Module 3")
tracker.update_status("LLM02", MitigationStatus.MITIGATED,
"Output filter con code injection detection", "Module 3")
print(tracker.progress_report())
# Output esperado:
# OWASP Progress Report
# Mitigated: 4/10
# Partially: 0/10
# Not mitigated: 6/10
# Coverage: 40%
# ...
Explicación: El tracker mantiene un historial completo de cómo cada riesgo evolucionó durante el desarrollo. Esto es invaluable para auditorías: demuestra que la mitigación fue incremental y cada paso está respaldado por evidencia referenciable a un módulo específico.
Ejercicio 3: Construir un SecurityDocValidator
Implementa un validador que reciba un OWASPFinalMapping y un ADRRegistry y verifique que: cada riesgo mitigado tiene al menos una defensa con evidencia, cada defensa referenciada tiene un ADR correspondiente, y no hay ADRs huérfanos sin referencia desde el mapping.
Ver solución
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):
"""Valida consistencia entre OWASP mapping y ADR registry."""
issues: list[ValidationIssue] = Field(default_factory=list)
def validate(
self,
owasp_risks: list[dict],
adr_records: list[dict],
) -> list[ValidationIssue]:
"""
Valida consistencia entre el mapping y los ADRs.
owasp_risks: lista con risk_id, status, defense_names
adr_records: lista con adr_id, title, owasp_risks_addressed
"""
self.issues = []
# 1. Cada riesgo mitigado debe tener al menos una defensa
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. Cada defensa debería tener un ADR correspondiente
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:
# Busca ADR cuyo título contenga la defensa o viceversa
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. ADRs huérfanos: ADRs que no mapean a ningún riesgo OWASP
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: riesgos sin ADR alguno
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)
# --- Ejemplo ---
validator = SecurityDocValidator()
owasp_risks = [
{"risk_id": "LLM01", "status": "mitigated", "defense_names": ["Input Sanitization Pipeline"]},
{"risk_id": "LLM02", "status": "mitigated", "defense_names": []}, # Error: sin evidencia
{"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 por Token",
"owasp_risks_addressed": ["LLM04", "LLM99"]}, # LLM99 es huérfano
]
validator.validate(owasp_risks, adr_records)
print(validator.summary())
# Output esperado:
# Validation Summary
# Errors: 1
# Warnings: 1
# Info: 0
# ❌ [missing_evidence] LLM02: marked as mitigated but has no defense evidence
# ⚠️ [orphan_adr_reference] ADR references LLM99 but it's not in the OWASP mapping
Explicación: El validador funciona como un linter para tu documentación de seguridad — encuentra inconsistencias antes de que un auditor las encuentre. Los errores son problemas graves (afirmaciones sin evidencia), los warnings son gaps potenciales, y los infos son oportunidades de mejora.
Ejercicio 4: Generar un Security README automático
Crea una función que tome un OWASPFinalMapping, un ADRRegistry, y un ResidualRiskRegister y genere un README.md completo del proyecto que incluya: resumen ejecutivo, tabla de cobertura OWASP, lista de decisiones clave, y riesgos aceptados.
Ver solución
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:
"""Genera un README de seguridad completo del proyecto."""
# Calcular métricas
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')}
## Resumen Ejecutivo
Este documento describe las medidas de seguridad implementadas en **{project_name}**.
El sistema cubre **{len(mitigated)}/{len(applicable)}** riesgos OWASP LLM Top 10 completamente
mitigados, con una cobertura global del **{coverage:.0f}%**.
Se han tomado **{len(active_adrs)}** decisiones arquitectónicas de seguridad documentadas como ADRs,
y se han identificado **{len(residual_risks)}** riesgos residuales aceptados,
de los cuales **{len(high_risks)}** requieren monitoreo prioritario.
## OWASP LLM Top 10 Coverage
| Risk ID | Nombre | Estado | Defensas |
|---------|--------|--------|----------|
"""
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"""
## Decisiones de Seguridad (ADRs)
| ID | Decisión | Estado | 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"""
## Riesgos Residuales Aceptados
| Risk | Título | Likelihood | Impact |
|------|--------|------------|--------|
"""
for rr in residual_risks:
readme += f"| {rr['risk_id']} | {rr['title']} | {rr['likelihood']} | {rr['impact']} |\n"
readme += """
## Cómo Contribuir a la Seguridad
1. Antes de agregar una nueva defensa, crea un ADR documentando la decisión
2. Actualiza el OWASP mapping cuando implementes una nueva mitigación
3. Ejecuta el scanner de documentación después de cambios significativos
4. Revisa los riesgos residuales trimestralmente
## Contacto
Para reportar vulnerabilidades: security@example.com
"""
return readme
# --- Ejemplo ---
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 por 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: README completo con todas las secciones
Explicación: El README se genera a partir de datos estructurados, no de texto libre. Esto significa que cada vez que el mapping OWASP o los ADRs se actualizan, el README se puede regenerar automáticamente y siempre reflejará el estado actual del sistema.
Resumen
- 📋 Los Architecture Decision Records (ADR) capturan el "por qué" detrás de cada decisión de seguridad, protegiendo contra cambios accidentales que debiliten las defensas
- 🔍
SecurityDecisionRecordextiende el ADR tradicional con categoría de seguridad, mapping OWASP, y trade-offs tipados por área de impacto - ⚖️ Documentar trade-offs con mediciones reales (no estimaciones) previene "optimizaciones" que eliminen defensas sin entender su costo
- 🗺️ El OWASP Final Mapping con
OWASPFinalMappinges el artefacto central de auditoría: cada riesgo LLM01-LLM10 con estado, defensas, y evidencia trazable - ⚠️ La documentación honesta de riesgos residuales con
ResidualRiskRegisterdemuestra decisiones informadas, no olvidos - 🤖 La generación automática de documentación con
SecurityDocGeneratorelimina la desactualización al escanear el código directamente - 🔗 La validación cruzada entre mapping OWASP y ADRs detecta inconsistencias antes de que un auditor las encuentre
- 📊 Un Security README auto-generado proporciona visibilidad ejecutiva del estado de seguridad sin esfuerzo manual continuo
Módulo completado: Has construido un sistema de seguridad AI integral — desde input validation hasta incident response y documentación auditable.
Recursos adicionales
- Architecture Decision Records — Repositorio oficial del formato ADR con templates y herramientas
- OWASP LLM Top 10 — Lista oficial de riesgos de seguridad para LLMs
- Michael Nygard — Documenting Architecture Decisions — El artículo original que introdujo los ADRs
- Google DORA — Documentation Practices — Investigación sobre el impacto de la documentación en la productividad
- Thoughtworks Tech Radar — ADR Tools — Análisis de herramientas ADR por Thoughtworks
- NIST AI Risk Management Framework — Framework federal de gestión de riesgos AI
- EU AI Act — Documentation Requirements — Requisitos de documentación del AI Act europeo
Creado: Marzo 2026 Versión: 1.0