Module 5: Secrets Management
6. Token Lifecycle and Audit Trails
Overview
In the previous capsules you learned to store secrets securely (03, 05), rotate them automatically (04), and choose between cloud providers (05). One critical piece is missing: who accesses your secrets? when? from where? And when a secret reaches the end of its useful life, how do you retire it securely?
Token lifecycle management and audit trails are the two capabilities that transform secrets management from "secure storage" to "enterprise management." The lifecycle defines a secret's phases: creation → distribution → use → rotation → revocation → destruction. Audit trails record every interaction with every secret, creating an immutable log that's essential for incident detection, post-mortem investigations, and compliance.
In this capsule you'll implement a complete lifecycle management system with defined states and transitions, an audit logger that records every operation on secrets, and the emergency revocation procedures you need when you suspect a secret was compromised.
Token Lifecycle: the 6 phases
Every secret goes through a defined lifecycle. Understanding the phases lets you manage secrets systematically:
┌──────────┐ ┌──────────────┐ ┌──────────┐
│ Creation │ → │ Distribution │ → │ Active │
└──────────┘ └──────────────┘ └────┬─────┘
│
┌─────────▼──────────┐
│ Rotation │
│ (new key created, │
│ old deprecated) │
└─────────┬──────────┘
│
┌─────────▼──────────┐
│ Revocation │
│ (key invalidated) │
└─────────┬──────────┘
│
┌─────────▼──────────┐
│ Destruction │
│ (key purged) │
└────────────────────┘
Lifecycle implementation
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:
"""Manages the lifecycle of multiple 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())}")
# Expected output:
# [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: who, what, when
An audit trail records every operation on every secret. It's the log you consult when investigating an incident, preparing a compliance audit, or needing to understand a secret's history.
Audit logger implementation
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:
"""Audit logger for operations on 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: accesses to 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}")
# Expected output:
# 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 and ISO 27001
Audit trails aren't just for debugging — they're compliance requirements:
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": [
"Who has access to which secrets?",
"When was the last rotation of each secret?",
"Is there an audit trail of every access?",
"How is access revoked when an employee leaves?",
"Are there alerts for anomalous access?",
],
"your_module_covers": [
"Lifecycle management (creation → destruction)",
"Audit trail of every operation",
"Anomaly detection (excessive access, repeated failures)",
"Emergency revocation procedures",
"Rotation schedules with evidence",
],
},
"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": [
"Credential management with a defined lifecycle",
"Encryption of secrets at rest and in transit",
"Access logging with a defined retention",
"Periodic access review",
],
},
}
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("What auditors ask:")
for q in info["what_auditors_ask"]:
print(f" ❓ {q}")Token Expiration Policies
Define expiration policies per secret type:
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="High financial value, simple rotation",
),
ExpirationPolicy(
secret_type="Database Passwords",
max_age_days=90,
warning_before_days=14,
action_on_expiry="alert_and_rotate",
auto_rotate=True,
rationale="Requires a 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="Rotation invalidates active tokens",
),
ExpirationPolicy(
secret_type="Encryption Keys",
max_age_days=365,
warning_before_days=60,
action_on_expiry="alert_only",
auto_rotate=False,
rationale="Requires re-encryption of existing data",
),
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, easy to rotate",
),
]
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
When you suspect a secret was compromised, you need an immediate revocation procedure:
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:
"""Emergency revocation procedure."""
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}")Integration: lifecycle + audit in a single system
class ManagedSecretsSystem:
"""Integrated system for 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
"The audit logs are huge and expensive to store"
Implement tiered retention: recent logs (30 days) in the main database, historical ones (1 year) in cold storage (S3/GCS), and add sampling for routine accesses. Only failed accesses and write operations need full retention.
"I don't know what actor to put in the audit log"
Use the identity of the service making the request: service name + pod/container ID. In FastAPI, extract the service identity from the authentication header or the JWT token.
"Emergency revocation caused downtime because we didn't have a replacement ready"
Keep a "break glass" procedure with pre-generated emergency secrets stored in a separate secure location. When you revoke, you activate the emergency secrets while you generate the permanent ones.
"The audit log has no checksums and could be tampered with"
Add checksums (like in AuditEntry._compute_checksum) and consider sending the logs to an immutable system (CloudTrail, write-once storage). For compliance, audit logs must be tamper-evident.
Exercises
Exercise 1: Implement audit log retention
Create a system that archives old logs and keeps only recent ones in memory:
See solution
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)Exercise 2: Create a compliance report
Generate a report that shows the compliance status of your secrets management:
See solution
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))Exercise 3: Implement anomaly detection based on time of day
Detect secret accesses outside business hours (9am-6pm):
See solution
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']}")Exercise 4: Simulate a complete revocation incident
Simulate a scenario where a key is detected in a public repo and the full procedure is executed:
See solution
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)Summary
- Token lifecycle has 6 defined phases: creation → distribution → active → rotation → revocation → destruction — with controlled valid transitions
- The LifecycleManager manages each secret's state with safe transitions, automatic expiration, and access tracking
- Audit trails record every operation (create, read, update, delete, rotate, revoke) with actor, timestamp, source IP, and an integrity checksum
- Anomaly detection identifies suspicious patterns: excessive access (>50 reads/hour), repeated failed attempts, and off-hours accesses
- Compliance (SOC 2, ISO 27001) requires evidence of lifecycle management and audit trails — this system generates that evidence automatically
- Expiration policies vary by secret type: 30 days for LLM keys, 90 for databases, 180+ for signing keys, 7 days for service accounts
- Emergency revocation is a critical procedure: a runbook per severity, immediate revocation, replacement keys, and a post-mortem
- The integrated system combines lifecycle + audit + emergency revocation into a single reusable component for the project
Next capsule: In capsule 07 you'll learn the least privilege principle applied to secrets, integration with FastAPI using dependency injection, and the fallback and resilience patterns for when the secrets service is unavailable.
Resources
- NIST SP 800-57 Key Management Recommendations — Standard for cryptographic key lifecycle management
- SOC 2 Compliance Requirements — Compliance framework with audit trail requirements
- ISO 27001 Annex A Controls — Security controls including credential management
- AWS CloudTrail — AWS's native audit trail service for Secrets Manager
- GCP Cloud Audit Logs — GCP's native audit logging for Secret Manager
- OWASP Logging Cheat Sheet — Security logging best practices
- Vault Audit Devices — HashiCorp Vault audit backends
- Python Logging Best Practices — Official Python logging reference
Created: March 2026 Version: 1.0