Módulo 5: Secrets Management
6. Token Lifecycle y Audit Trails
Descripción
En las cápsulas anteriores aprendiste a almacenar secrets de forma segura (03, 05), a rotarlos automáticamente (04), y a elegir entre proveedores cloud (05). Falta una pieza crítica: ¿quién accede a tus secrets? ¿cuándo? ¿desde dónde? Y cuando un secret llega al final de su vida útil, ¿cómo lo retiras de forma segura?
Token lifecycle management y audit trails son las dos capacidades que transforman secrets management de "almacenamiento seguro" a "gestión enterprise." El lifecycle define las fases de un secret: creación → distribución → uso → rotación → revocación → destrucción. Los audit trails registran cada interacción con cada secret, creando un registro inmutable que es esencial para detección de incidentes, investigaciones post-mortem, y compliance.
En esta cápsula vas a implementar un sistema completo de lifecycle management con estados y transiciones definidas, un audit logger que registra cada operación sobre secrets, y los procedimientos de revocación de emergencia que necesitas cuando sospechas que un secret fue comprometido.
Token Lifecycle: las 6 fases
Cada secret pasa por un ciclo de vida definido. Entender las fases te permite gestionar secrets de forma sistemática:
┌──────────┐ ┌──────────────┐ ┌──────────┐
│ Creation │ → │ Distribution │ → │ Active │
└──────────┘ └──────────────┘ └────┬─────┘
│
┌─────────▼──────────┐
│ Rotation │
│ (new key created, │
│ old deprecated) │
└─────────┬──────────┘
│
┌─────────▼──────────┐
│ Revocation │
│ (key invalidated) │
└─────────┬──────────┘
│
┌─────────▼──────────┐
│ Destruction │
│ (key purged) │
└────────────────────┘
Implementación del lifecycle
import json
import time
import uuid
from enum import Enum
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Callable
import logging
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("lifecycle")
class SecretState(Enum):
CREATED = "created"
DISTRIBUTED = "distributed"
ACTIVE = "active"
ROTATING = "rotating"
DEPRECATED = "deprecated"
REVOKED = "revoked"
DESTROYED = "destroyed"
VALID_TRANSITIONS = {
SecretState.CREATED: [SecretState.DISTRIBUTED, SecretState.REVOKED],
SecretState.DISTRIBUTED: [SecretState.ACTIVE, SecretState.REVOKED],
SecretState.ACTIVE: [SecretState.ROTATING, SecretState.DEPRECATED, SecretState.REVOKED],
SecretState.ROTATING: [SecretState.DEPRECATED, SecretState.REVOKED],
SecretState.DEPRECATED: [SecretState.REVOKED],
SecretState.REVOKED: [SecretState.DESTROYED],
SecretState.DESTROYED: [],
}
@dataclass
class SecretLifecycle:
secret_id: str
name: str
state: SecretState
created_at: datetime
created_by: str
state_history: list[dict] = field(default_factory=list)
expires_at: Optional[datetime] = None
last_accessed: Optional[datetime] = None
access_count: int = 0
metadata: dict = field(default_factory=dict)
def can_transition_to(self, new_state: SecretState) -> bool:
return new_state in VALID_TRANSITIONS.get(self.state, [])
def transition(self, new_state: SecretState, actor: str, reason: str = "") -> bool:
if not self.can_transition_to(new_state):
logger.warning(
f"Invalid transition: {self.name} {self.state.value} → {new_state.value}"
)
return False
old_state = self.state
self.state = new_state
self.state_history.append({
"from": old_state.value,
"to": new_state.value,
"actor": actor,
"reason": reason,
"timestamp": datetime.utcnow().isoformat(),
})
logger.info(f"[{self.name}] {old_state.value} → {new_state.value} by {actor}")
return True
def record_access(self):
self.last_accessed = datetime.utcnow()
self.access_count += 1
@property
def is_usable(self) -> bool:
if self.state not in (SecretState.ACTIVE, SecretState.DEPRECATED):
return False
if self.expires_at and datetime.utcnow() >= self.expires_at:
return False
return True
@property
def age_days(self) -> int:
return (datetime.utcnow() - self.created_at).days
def summary(self) -> dict:
return {
"name": self.name,
"state": self.state.value,
"age_days": self.age_days,
"access_count": self.access_count,
"is_usable": self.is_usable,
"transitions": len(self.state_history),
"last_accessed": self.last_accessed.isoformat() if self.last_accessed else None,
}
class LifecycleManager:
"""Gestiona el lifecycle de múltiples secrets."""
def __init__(self):
self._secrets: dict[str, SecretLifecycle] = {}
def create_secret(
self,
name: str,
created_by: str,
ttl_days: Optional[int] = None,
metadata: Optional[dict] = None,
) -> SecretLifecycle:
secret = SecretLifecycle(
secret_id=f"sec-{uuid.uuid4().hex[:8]}",
name=name,
state=SecretState.CREATED,
created_at=datetime.utcnow(),
created_by=created_by,
expires_at=datetime.utcnow() + timedelta(days=ttl_days) if ttl_days else None,
metadata=metadata or {},
)
self._secrets[name] = secret
return secret
def distribute(self, name: str, actor: str, target: str) -> bool:
secret = self._secrets.get(name)
if not secret:
return False
return secret.transition(
SecretState.DISTRIBUTED, actor, f"Distributed to {target}"
)
def activate(self, name: str, actor: str) -> bool:
secret = self._secrets.get(name)
if not secret:
return False
return secret.transition(SecretState.ACTIVE, actor, "Activated for use")
def deprecate(self, name: str, actor: str, reason: str = "") -> bool:
secret = self._secrets.get(name)
if not secret:
return False
return secret.transition(SecretState.DEPRECATED, actor, reason)
def revoke(self, name: str, actor: str, reason: str = "") -> bool:
secret = self._secrets.get(name)
if not secret:
return False
return secret.transition(SecretState.REVOKED, actor, reason)
def emergency_revoke_all(self, actor: str, reason: str) -> list[str]:
revoked = []
for name, secret in self._secrets.items():
if secret.state in (SecretState.ACTIVE, SecretState.DEPRECATED):
if secret.transition(SecretState.REVOKED, actor, f"EMERGENCY: {reason}"):
revoked.append(name)
return revoked
def get_expired(self) -> list[SecretLifecycle]:
return [
s for s in self._secrets.values()
if s.expires_at and datetime.utcnow() >= s.expires_at
and s.state in (SecretState.ACTIVE, SecretState.DEPRECATED)
]
def dashboard(self) -> dict:
by_state = {}
for secret in self._secrets.values():
state = secret.state.value
by_state.setdefault(state, []).append(secret.name)
return {
"total": len(self._secrets),
"by_state": by_state,
"expired": [s.name for s in self.get_expired()],
}
manager = LifecycleManager()
openai = manager.create_secret("openai-api-key", "admin", ttl_days=30)
manager.distribute("openai-api-key", "admin", "api-service")
manager.activate("openai-api-key", "api-service")
anthropic = manager.create_secret("anthropic-api-key", "admin", ttl_days=30)
manager.distribute("anthropic-api-key", "admin", "api-service")
manager.activate("anthropic-api-key", "api-service")
db = manager.create_secret("database-password", "dba", ttl_days=90)
manager.distribute("database-password", "dba", "all-services")
manager.activate("database-password", "migration-runner")
openai.record_access()
openai.record_access()
openai.record_access()
print("Lifecycle Dashboard:")
print(json.dumps(manager.dashboard(), indent=2))
print("\nSecret Summaries:")
for name in ["openai-api-key", "anthropic-api-key", "database-password"]:
secret = manager._secrets[name]
print(f" {json.dumps(secret.summary())}")
# Output esperado:
# [openai-api-key] created → distributed by admin
# [openai-api-key] distributed → active by api-service
# ...
# Lifecycle Dashboard:
# {
# "total": 3,
# "by_state": {
# "active": ["openai-api-key", "anthropic-api-key", "database-password"]
# },
# "expired": []
# }Audit Trails: quién, qué, cuándo
Un audit trail registra cada operación sobre cada secret. Es el log que consultas cuando investigas un incidente, preparas una auditoría de compliance, o necesitas entender el historial de un secret.
Implementación del audit logger
import json
import time
import uuid
import hashlib
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Optional
from enum import Enum
class AuditAction(Enum):
CREATE = "create"
READ = "read"
UPDATE = "update"
DELETE = "delete"
ROTATE = "rotate"
REVOKE = "revoke"
DISTRIBUTE = "distribute"
LIST = "list"
FAILED_ACCESS = "failed_access"
@dataclass
class AuditEntry:
entry_id: str
timestamp: str
action: str
secret_name: str
actor: str
success: bool
source_ip: str = "unknown"
service: str = "unknown"
details: dict = field(default_factory=dict)
checksum: str = ""
def __post_init__(self):
if not self.checksum:
self.checksum = self._compute_checksum()
def _compute_checksum(self) -> str:
data = f"{self.timestamp}:{self.action}:{self.secret_name}:{self.actor}"
return hashlib.sha256(data.encode()).hexdigest()[:16]
class SecretAuditLogger:
"""Logger de auditoría para operaciones sobre secrets."""
def __init__(self, log_file: Optional[str] = None):
self._entries: list[AuditEntry] = []
self._log_file = log_file
self._logger = logging.getLogger("audit")
def log(
self,
action: AuditAction,
secret_name: str,
actor: str,
success: bool = True,
source_ip: str = "unknown",
service: str = "unknown",
details: Optional[dict] = None,
) -> AuditEntry:
entry = AuditEntry(
entry_id=f"aud-{uuid.uuid4().hex[:8]}",
timestamp=datetime.utcnow().isoformat(),
action=action.value,
secret_name=secret_name,
actor=actor,
success=success,
source_ip=source_ip,
service=service,
details=details or {},
)
self._entries.append(entry)
log_line = json.dumps(asdict(entry), ensure_ascii=False)
self._logger.info(log_line)
if self._log_file:
with open(self._log_file, "a") as f:
f.write(log_line + "\n")
return entry
def query(
self,
secret_name: Optional[str] = None,
actor: Optional[str] = None,
action: Optional[AuditAction] = None,
success_only: bool = False,
limit: int = 100,
) -> list[AuditEntry]:
results = self._entries
if secret_name:
results = [e for e in results if e.secret_name == secret_name]
if actor:
results = [e for e in results if e.actor == actor]
if action:
results = [e for e in results if e.action == action.value]
if success_only:
results = [e for e in results if e.success]
return results[-limit:]
def detect_anomalies(self) -> list[dict]:
anomalies = []
actor_counts: dict[str, int] = {}
failed_counts: dict[str, int] = {}
recent_window = 3600
now = time.time()
for entry in self._entries:
entry_time = datetime.fromisoformat(entry.timestamp).timestamp()
if now - entry_time > recent_window:
continue
actor_counts[entry.actor] = actor_counts.get(entry.actor, 0) + 1
if not entry.success:
key = f"{entry.actor}:{entry.secret_name}"
failed_counts[key] = failed_counts.get(key, 0) + 1
for actor, count in actor_counts.items():
if count > 50:
anomalies.append({
"type": "excessive_access",
"actor": actor,
"count": count,
"window": "1 hour",
"severity": "WARNING",
})
for key, count in failed_counts.items():
if count > 5:
actor, secret = key.split(":", 1)
anomalies.append({
"type": "repeated_failures",
"actor": actor,
"secret": secret,
"count": count,
"severity": "CRITICAL",
})
return anomalies
def summary(self) -> dict:
total = len(self._entries)
by_action = {}
by_actor = {}
failures = 0
for entry in self._entries:
by_action[entry.action] = by_action.get(entry.action, 0) + 1
by_actor[entry.actor] = by_actor.get(entry.actor, 0) + 1
if not entry.success:
failures += 1
return {
"total_entries": total,
"by_action": by_action,
"by_actor": by_actor,
"failures": failures,
"failure_rate": f"{(failures/total*100):.1f}%" if total > 0 else "0%",
}
audit = SecretAuditLogger()
audit.log(AuditAction.CREATE, "openai-api-key", "admin@company.com",
service="secrets-cli", source_ip="10.0.1.5")
audit.log(AuditAction.READ, "openai-api-key", "api-service",
service="api-server", source_ip="10.0.2.10")
audit.log(AuditAction.READ, "openai-api-key", "api-service",
service="api-server", source_ip="10.0.2.10")
audit.log(AuditAction.READ, "database-password", "migration-job",
service="k8s-job", source_ip="10.0.3.20")
audit.log(AuditAction.FAILED_ACCESS, "admin-key", "unknown-process",
success=False, service="unknown", source_ip="192.168.1.100",
details={"reason": "policy denied"})
audit.log(AuditAction.ROTATE, "openai-api-key", "rotation-scheduler",
service="cron", details={"old_version": 1, "new_version": 2})
print("Audit Summary:")
print(json.dumps(audit.summary(), indent=2))
print("\nQuery: accesos a openai-api-key:")
for entry in audit.query(secret_name="openai-api-key"):
status = "✅" if entry.success else "❌"
print(f" {status} [{entry.action}] by {entry.actor} from {entry.source_ip}")
print("\nQuery: failed accesses:")
for entry in audit.query(success_only=False):
if not entry.success:
print(f" ❌ [{entry.action}] {entry.actor} → {entry.secret_name}: {entry.details}")
# Output esperado:
# Audit Summary:
# {
# "total_entries": 6,
# "by_action": {"create": 1, "read": 3, "failed_access": 1, "rotate": 1},
# "by_actor": {"admin@company.com": 1, "api-service": 2, ...},
# "failures": 1,
# "failure_rate": "16.7%"
# }Compliance: SOC 2 e ISO 27001
Los audit trails no son solo para debugging — son requisitos de compliance:
compliance_requirements = {
"SOC2": {
"relevant_criteria": [
"CC6.1: Logical access controls",
"CC6.2: Access provisioning and removal",
"CC6.3: Role-based access",
"CC7.2: Monitoring anomalous activity",
"CC8.1: Change management",
],
"what_auditors_ask": [
"¿Quién tiene acceso a qué secrets?",
"¿Cuándo fue la última rotación de cada secret?",
"¿Hay audit trail de cada acceso?",
"¿Cómo se revoca el acceso cuando un empleado se va?",
"¿Hay alertas de acceso anómalo?",
],
"your_module_covers": [
"Lifecycle management (creation → destruction)",
"Audit trail de cada operación",
"Anomaly detection (excessive access, repeated failures)",
"Emergency revocation procedures",
"Rotation schedules con evidencia",
],
},
"ISO_27001": {
"relevant_controls": [
"A.9.2: User access management",
"A.9.4: System and application access control",
"A.10.1: Cryptographic controls",
"A.12.4: Logging and monitoring",
"A.18.1: Compliance with legal requirements",
],
"key_requirements": [
"Gestión de credenciales con lifecycle definido",
"Encryption de secrets at rest y in transit",
"Logging de accesos con retención definida",
"Revisión periódica de accesos",
],
},
}
for framework, info in compliance_requirements.items():
print(f"\n=== {framework} ===")
if "relevant_criteria" in info:
print("Criteria:")
for c in info["relevant_criteria"]:
print(f" - {c}")
if "what_auditors_ask" in info:
print("Lo que preguntan los auditores:")
for q in info["what_auditors_ask"]:
print(f" ❓ {q}")Token Expiration Policies
Define políticas de expiración por tipo de secret:
from dataclasses import dataclass
from typing import Optional
@dataclass
class ExpirationPolicy:
secret_type: str
max_age_days: int
warning_before_days: int
action_on_expiry: str
auto_rotate: bool
rationale: str
policies = [
ExpirationPolicy(
secret_type="LLM API Keys",
max_age_days=30,
warning_before_days=7,
action_on_expiry="auto_rotate",
auto_rotate=True,
rationale="Alto valor financiero, rotación simple",
),
ExpirationPolicy(
secret_type="Database Passwords",
max_age_days=90,
warning_before_days=14,
action_on_expiry="alert_and_rotate",
auto_rotate=True,
rationale="Requiere connection pool restart",
),
ExpirationPolicy(
secret_type="JWT Signing Keys",
max_age_days=180,
warning_before_days=30,
action_on_expiry="alert_only",
auto_rotate=False,
rationale="Rotación invalida tokens activos",
),
ExpirationPolicy(
secret_type="Encryption Keys",
max_age_days=365,
warning_before_days=60,
action_on_expiry="alert_only",
auto_rotate=False,
rationale="Requiere re-encryption de datos existentes",
),
ExpirationPolicy(
secret_type="Service Account Tokens",
max_age_days=7,
warning_before_days=1,
action_on_expiry="auto_rotate",
auto_rotate=True,
rationale="Short-lived, fácil de rotar",
),
]
print("Expiration Policies:")
print(f"{'Type':<25} {'Max Age':<10} {'Warning':<10} {'Auto-Rotate':<12} {'On Expiry':<20}")
print("-" * 80)
for p in policies:
auto = "Yes" if p.auto_rotate else "No"
print(f"{p.secret_type:<25} {p.max_age_days}d{'':<6} {p.warning_before_days}d{'':<6} {auto:<12} {p.action_on_expiry:<20}")Emergency Revocation
Cuando sospechas que un secret fue comprometido, necesitas un procedimiento de revocación inmediata:
import json
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
class IncidentSeverity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class RevocationEvent:
incident_id: str
severity: IncidentSeverity
affected_secrets: list[str]
trigger: str
initiated_by: str
timestamp: str
steps_taken: list[str] = field(default_factory=list)
status: str = "in_progress"
class EmergencyRevocation:
"""Procedimiento de revocación de emergencia."""
def __init__(self, lifecycle_manager: LifecycleManager, audit_logger: SecretAuditLogger):
self.lifecycle = lifecycle_manager
self.audit = audit_logger
self._incidents: list[RevocationEvent] = []
def initiate(
self,
severity: IncidentSeverity,
affected_secrets: list[str],
trigger: str,
initiated_by: str,
) -> RevocationEvent:
incident = RevocationEvent(
incident_id=f"INC-{uuid.uuid4().hex[:6].upper()}",
severity=severity,
affected_secrets=affected_secrets,
trigger=trigger,
initiated_by=initiated_by,
timestamp=datetime.utcnow().isoformat(),
)
self.audit.log(
AuditAction.REVOKE,
",".join(affected_secrets),
initiated_by,
details={"incident_id": incident.incident_id, "trigger": trigger},
)
for secret_name in affected_secrets:
success = self.lifecycle.revoke(
secret_name, initiated_by,
f"Emergency: {trigger} (Incident: {incident.incident_id})"
)
if success:
incident.steps_taken.append(f"Revoked: {secret_name}")
else:
incident.steps_taken.append(f"Failed to revoke: {secret_name}")
incident.steps_taken.append("Notified incident response team")
incident.steps_taken.append("Checked audit logs for unauthorized access")
self._incidents.append(incident)
return incident
def get_runbook(self, severity: IncidentSeverity) -> list[str]:
runbooks = {
IncidentSeverity.CRITICAL: [
"1. REVOKE all affected secrets IMMEDIATELY",
"2. Notify incident response team (Slack #security-incidents)",
"3. Check audit trail for unauthorized access in last 24h",
"4. Check LLM provider usage dashboards for anomalies",
"5. Generate new secrets and update secrets manager",
"6. Deploy with new secrets",
"7. Monitor for 1 hour for any remaining unauthorized access",
"8. File incident report within 24 hours",
],
IncidentSeverity.HIGH: [
"1. REVOKE affected secrets within 1 hour",
"2. Notify team lead and security team",
"3. Review audit logs for the affected secrets",
"4. Generate replacement secrets",
"5. Deploy with new secrets during next window",
"6. File incident report within 48 hours",
],
IncidentSeverity.MEDIUM: [
"1. Schedule revocation within 24 hours",
"2. Review audit logs for anomalies",
"3. Prepare replacement secrets",
"4. Deploy with new secrets in next scheduled deployment",
"5. Document in weekly security review",
],
IncidentSeverity.LOW: [
"1. Add to next rotation cycle",
"2. Review access policies",
"3. Document in monthly security review",
],
}
return runbooks.get(severity, [])
manager = LifecycleManager()
audit = SecretAuditLogger()
openai = manager.create_secret("openai-api-key", "admin", ttl_days=30)
manager.distribute("openai-api-key", "admin", "api-service")
manager.activate("openai-api-key", "api-service")
db = manager.create_secret("database-password", "dba", ttl_days=90)
manager.distribute("database-password", "dba", "all-services")
manager.activate("database-password", "migration-runner")
emergency = EmergencyRevocation(manager, audit)
print("=== Emergency Revocation Runbook (CRITICAL) ===")
for step in emergency.get_runbook(IncidentSeverity.CRITICAL):
print(f" {step}")
incident = emergency.initiate(
severity=IncidentSeverity.CRITICAL,
affected_secrets=["openai-api-key"],
trigger="API key found in public GitHub repository",
initiated_by="security-team@company.com",
)
print(f"\nIncident {incident.incident_id}:")
print(f" Severity: {incident.severity.value}")
print(f" Trigger: {incident.trigger}")
print(" Steps taken:")
for step in incident.steps_taken:
print(f" - {step}")Integración: lifecycle + audit en un solo sistema
class ManagedSecretsSystem:
"""Sistema integrado de lifecycle management + audit trails."""
def __init__(self):
self.lifecycle = LifecycleManager()
self.audit = SecretAuditLogger()
self.emergency = EmergencyRevocation(self.lifecycle, self.audit)
def create_and_activate(
self, name: str, actor: str, ttl_days: Optional[int] = None
) -> SecretLifecycle:
secret = self.lifecycle.create_secret(name, actor, ttl_days)
self.audit.log(AuditAction.CREATE, name, actor,
details={"ttl_days": ttl_days})
self.lifecycle.distribute(name, actor, "target-service")
self.audit.log(AuditAction.DISTRIBUTE, name, actor)
self.lifecycle.activate(name, actor)
return secret
def read_secret(self, name: str, actor: str, source_ip: str = "unknown") -> Optional[str]:
secret = self.lifecycle._secrets.get(name)
if not secret or not secret.is_usable:
self.audit.log(AuditAction.FAILED_ACCESS, name, actor,
success=False, source_ip=source_ip,
details={"reason": "not found or not usable"})
return None
secret.record_access()
self.audit.log(AuditAction.READ, name, actor, source_ip=source_ip)
return f"secret-value-for-{name}"
def rotate_secret(self, name: str, actor: str) -> bool:
success = self.lifecycle.deprecate(name, actor, "Rotated")
if success:
new_secret = self.lifecycle.create_secret(
f"{name}", actor, ttl_days=30
)
self.audit.log(AuditAction.ROTATE, name, actor)
return success
def health_report(self) -> dict:
return {
"lifecycle": self.lifecycle.dashboard(),
"audit": self.audit.summary(),
"anomalies": self.audit.detect_anomalies(),
}
system = ManagedSecretsSystem()
system.create_and_activate("openai-api-key", "admin", ttl_days=30)
system.create_and_activate("anthropic-api-key", "admin", ttl_days=30)
system.create_and_activate("database-password", "dba", ttl_days=90)
system.read_secret("openai-api-key", "api-service", "10.0.2.10")
system.read_secret("openai-api-key", "api-service", "10.0.2.10")
system.read_secret("nonexistent-key", "unknown", "192.168.1.100")
print("System Health Report:")
print(json.dumps(system.health_report(), indent=2))Troubleshooting
"Los audit logs son enormes y costosos de almacenar"
Implementa retención por capas: logs recientes (30 días) en la base de datos principal, históricos (1 año) en cold storage (S3/GCS), y agrega sampling para accesos rutinarios. Solo los accesos fallidos y las operaciones de escritura necesitan retención completa.
"No sé qué actor poner en el audit log"
Usa la identidad del servicio que hace la request: nombre del servicio + ID del pod/container. En FastAPI, extrae el service identity del header de autenticación o del JWT token.
"Emergency revocation causó downtime porque no teníamos replacement ready"
Mantén un procedimiento de "break glass" con secrets de emergencia pre-generados y almacenados en un lugar seguro separado. Cuando revocas, activas los secrets de emergencia mientras generas los definitivos.
"El audit log no tiene checksums y podría ser manipulado"
Agrega checksums (como en AuditEntry._compute_checksum) y considera enviar los logs a un sistema inmutable (CloudTrail, write-once storage). Para compliance, los audit logs deben ser tamper-evident.
Ejercicios
Ejercicio 1: Implementa retención de audit logs
Crea un sistema que archive logs antiguos y mantenga solo los recientes en memoria:
Ver solución
class AuditLogWithRetention(SecretAuditLogger):
def __init__(self, retention_days: int = 30, archive_file: str = "audit_archive.jsonl"):
super().__init__()
self.retention_days = retention_days
self.archive_file = archive_file
def archive_old_entries(self) -> int:
cutoff = datetime.utcnow() - timedelta(days=self.retention_days)
to_archive = []
to_keep = []
for entry in self._entries:
entry_time = datetime.fromisoformat(entry.timestamp)
if entry_time < cutoff:
to_archive.append(entry)
else:
to_keep.append(entry)
if to_archive:
with open(self.archive_file, "a") as f:
for entry in to_archive:
f.write(json.dumps(asdict(entry)) + "\n")
self._entries = to_keep
return len(to_archive)Ejercicio 2: Crea un reporte de compliance
Genera un reporte que muestre el estado de compliance de tu secrets management:
Ver solución
def compliance_report(system: ManagedSecretsSystem) -> dict:
report = {
"generated_at": datetime.utcnow().isoformat(),
"checks": [],
}
dashboard = system.lifecycle.dashboard()
audit_summary = system.audit.summary()
report["checks"].append({
"control": "All secrets have defined lifecycle",
"status": "PASS" if dashboard["total"] > 0 else "FAIL",
"detail": f"{dashboard['total']} secrets tracked",
})
report["checks"].append({
"control": "Audit trail is active",
"status": "PASS" if audit_summary["total_entries"] > 0 else "FAIL",
"detail": f"{audit_summary['total_entries']} entries recorded",
})
report["checks"].append({
"control": "No expired secrets in active state",
"status": "PASS" if not dashboard["expired"] else "FAIL",
"detail": f"Expired: {dashboard['expired'] or 'none'}",
})
report["checks"].append({
"control": "Failed access attempts are logged",
"status": "PASS" if audit_summary.get("failures", 0) >= 0 else "FAIL",
"detail": f"{audit_summary.get('failures', 0)} failures logged",
})
return report
report = compliance_report(system)
print(json.dumps(report, indent=2))Ejercicio 3: Implementa anomaly detection basada en hora del día
Detecta accesos a secrets fuera del horario laboral (9am-6pm):
Ver solución
def detect_off_hours_access(audit: SecretAuditLogger, work_start: int = 9, work_end: int = 18):
anomalies = []
for entry in audit._entries:
entry_time = datetime.fromisoformat(entry.timestamp)
hour = entry_time.hour
if hour < work_start or hour >= work_end:
anomalies.append({
"type": "off_hours_access",
"entry_id": entry.entry_id,
"actor": entry.actor,
"secret": entry.secret_name,
"hour": hour,
"severity": "WARNING",
})
return anomalies
off_hours = detect_off_hours_access(system.audit)
print(f"Off-hours accesses: {len(off_hours)}")
for a in off_hours:
print(f" {a['actor']} accessed {a['secret']} at hour {a['hour']}")Ejercicio 4: Simula un incidente completo de revocación
Simula un escenario donde se detecta una key en un repo público y se ejecuta el procedimiento completo:
Ver solución
def simulate_incident(system: ManagedSecretsSystem):
print("=== INCIDENT SIMULATION ===")
print("1. Alert: API key detected in public GitHub repo")
incident = system.emergency.initiate(
severity=IncidentSeverity.CRITICAL,
affected_secrets=["openai-api-key"],
trigger="Key found in public repo by GitHub Secret Scanning",
initiated_by="github-alert@company.com",
)
print(f"2. Incident created: {incident.incident_id}")
print(f"3. Steps taken: {incident.steps_taken}")
print("4. Generating replacement key...")
system.create_and_activate("openai-api-key-v2", "security-team", ttl_days=30)
print("5. Checking audit trail for unauthorized access...")
suspicious = system.audit.query(
secret_name="openai-api-key",
success_only=False,
)
print(f" Found {len(suspicious)} access records to review")
print(f"6. Incident {incident.incident_id} resolved")
simulate_incident(system)Resumen
- Token lifecycle tiene 6 fases definidas: creation → distribution → active → rotation → revocation → destruction — con transiciones válidas controladas
- El LifecycleManager gestiona el estado de cada secret con transiciones seguras, expiración automática, y tracking de accesos
- Audit trails registran cada operación (create, read, update, delete, rotate, revoke) con actor, timestamp, source IP, y checksum de integridad
- Anomaly detection identifica patrones sospechosos: acceso excesivo (>50 reads/hora), intentos fallidos repetidos, y accesos fuera de horario
- Compliance (SOC 2, ISO 27001) requiere evidencia de lifecycle management y audit trails — este sistema genera esa evidencia automáticamente
- Expiration policies varían por tipo de secret: 30 días para LLM keys, 90 para databases, 180+ para signing keys, 7 días para service accounts
- Emergency revocation es un procedimiento crítico: runbook por severidad, revocación inmediata, replacement keys, y post-mortem
- El sistema integrado combina lifecycle + audit + emergency revocation en un solo componente reutilizable para el proyecto
Próxima cápsula: En la cápsula 07 vas a aprender el principio de least privilege aplicado a secrets, la integración con FastAPI usando dependency injection, y los patrones de fallback y resilience cuando el secrets service no está disponible.
Recursos
- NIST SP 800-57 Key Management Recommendations — Estándar para lifecycle management de keys criptográficas
- SOC 2 Compliance Requirements — Framework de compliance con requisitos de audit trails
- ISO 27001 Annex A Controls — Controles de seguridad incluyendo gestión de credenciales
- AWS CloudTrail — Servicio de audit trail nativo de AWS para Secrets Manager
- GCP Cloud Audit Logs — Audit logging nativo de GCP para Secret Manager
- OWASP Logging Cheat Sheet — Mejores prácticas de logging de seguridad
- Vault Audit Devices — Audit backends de HashiCorp Vault
- Python Logging Best Practices — Referencia oficial de logging en Python
Creado: Marzo 2026 Versión: 1.0