Security

Security hardening for agent systems: authentication, authorization, data protection, and threat mitigation.

Security Layers

Security Layers
├── Network
│   ├── TLS/mTLS
│   ├── Network Policies
│   ├── WAF/DDoS Protection
│   └── Zero Trust Network
├── Application
│   ├── Authentication
│   ├── Authorization (RBAC/ABAC)
│   ├── Input Validation
│   ├── Output Encoding
│   └── Rate Limiting
├── Data
│   ├── Encryption at Rest
│   ├── Encryption in Transit
│   ├── Key Management
│   ├── Data Classification
│   └── PII Protection
├── Infrastructure
│   ├── Container Security
│   ├── Image Scanning
│   ├── Runtime Protection
│   └── Compliance
└── Operational
    ├── Secrets Management
    ├── Audit Logging
    ├── Incident Response
    └── Vulnerability Management

Authentication & Authorization

JWT Authentication

from jose import jwt, JWTError
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

class JWTAuth:
    def __init__(self, secret_key: str, algorithm: str = "HS256", 
                 access_token_expire: int = 30):  # minutes
        self.secret_key = secret_key
        self.algorithm = algorithm
        self.expire_minutes = access_token_expire
        self.security = HTTPBearer()

    def create_token(self, user_id: str, roles: list[str] = None, 
                     extra_claims: dict = None) -> str:
        payload = {
            "sub": user_id,
            "roles": roles or [],
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(minutes=self.expire_minutes)
        }
        if extra_claims:
            payload.update(extra_claims)
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

    def verify_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except JWTError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token"
            )

    async def get_current_user(self, credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
        payload = self.verify_token(credentials.credentials)
        return User(
            id=payload["sub"],
            roles=payload.get("roles", [])
        )

RBAC (Role-Based Access Control)

from enum import Enum
from functools import wraps

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    ADMIN = "admin"
    AGENT_EXECUTE = "agent:execute"
    AGENT_MANAGE = "agent:manage"
    TOOL_USE = "tool:use"
    DATA_READ = "data:read"
    DATA_WRITE = "data:write"

ROLE_PERMISSIONS = {
    "viewer": [Permission.READ, Permission.DATA_READ],
    "user": [Permission.READ, Permission.WRITE, Permission.DATA_READ, Permission.DATA_WRITE, 
             Permission.AGENT_EXECUTE, Permission.TOOL_USE],
    "developer": [Permission.READ, Permission.WRITE, Permission.DELETE, 
                  Permission.DATA_READ, Permission.DATA_WRITE,
                  Permission.AGENT_EXECUTE, Permission.AGENT_MANAGE, Permission.TOOL_USE],
    "admin": list(Permission)
}

def require_permission(*permissions: Permission):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Get current user from dependency
            current_user = kwargs.get('current_user')
            if not current_user:
                raise HTTPException(401, "Authentication required")

            user_permissions = ROLE_PERMISSIONS.get(current_user.role, [])
            if not all(p in user_permissions for p in permissions):
                raise HTTPException(403, "Insufficient permissions")
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@router.post("/agents")
@require_permission(Permission.AGENT_MANAGE)
async def create_agent(agent: AgentCreate, current_user: User = Depends(get_current_user)):
    return await agent_service.create(agent, current_user)

ABAC (Attribute-Based Access Control)

from dataclasses import dataclass
from typing import Callable

@dataclass
class AccessContext:
    user: User
    resource: Resource
    action: str
    context: dict

class PolicyEngine:
    def __init__(self):
        self.policies: list[Policy] = []

    def add_policy(self, policy: Policy):
        self.policies.append(policy)

    async def check(self, context: AccessContext) -> AuthResult:
        for policy in self.policies:
            if policy.applies(context):
                result = await policy.evaluate(context)
                if result.decision == Decision.DENY:
                    return AuthResult(allowed=False, reason=result.reason)
                if result.decision == Decision.ALLOW:
                    return AuthResult(allowed=True)
        return AuthResult(allowed=False, reason="No matching policy")

@dataclass
class Policy:
    name: str
    condition: Callable[[AccessContext], bool]
    effect: Decision
    reason: str

    def applies(self, context: AccessContext) -> bool:
        return self.condition(context)

    async def evaluate(self, context: AccessContext) -> PolicyResult:
        if self.applies(context):
            return PolicyResult(decision=self.effect, reason=self.reason)
        return PolicyResult(decision=Decision.NOT_APPLICABLE)

class Decision(Enum):
    ALLOW = "allow"
    DENY = "deny"
    NOT_APPLICABLE = "not_applicable"

# Example policies
policy_engine = PolicyEngine()

policy_engine.add_policy(Policy(
    name="owner_access",
    condition=lambda ctx: ctx.user.id == ctx.resource.owner_id,
    effect=Decision.ALLOW,
    reason="Resource owner has full access"
))

policy_engine.add_policy(Policy(
    name="admin_access",
    condition=lambda ctx: "admin" in ctx.user.roles,
    effect=Decision.ALLOW,
    reason="Admins have full access"
))

policy_engine.add_policy(Policy(
    name="business_hours_only",
    condition=lambda ctx: 9 <= datetime.utcnow().hour <= 17,
    effect=Decision.DENY,
    reason="Operation only allowed during business hours"
))

policy_engine.add_policy(Policy(
    name="data_sensitivity",
    condition=lambda ctx: ctx.resource.sensitivity == "high" and 
                           Permission.DATA_WRITE not in ctx.user.permissions,
    effect=Decision.DENY,
    reason="Insufficient permissions for high-sensitivity data"
))

Data Protection

Encryption at Rest

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os

class EncryptionService:
    def __init__(self, master_key: bytes = None):
        if master_key:
            self.key = master_key
        else:
            # Generate from environment
            password = os.environ.get("ENCRYPTION_PASSWORD", "").encode()
            salt = os.environ.get("ENCRYPTION_SALT", "fixed-salt").encode()
            kdf = PBKDF2HMAC(
                algorithm=hashes.SHA256(),
                length=32,
                salt=salt,
                iterations=100000,
            )
            self.key = base64.urlsafe_b64encode(kdf.derive(password))
        self.fernet = Fernet(self.key)

    def encrypt(self, data: str) -> str:
        return self.fernet.encrypt(data.encode()).decode()

    def decrypt(self, token: str) -> str:
        return self.fernet.decrypt(token.encode()).decode()

    def encrypt_bytes(self, data: bytes) -> bytes:
        return self.fernet.encrypt(data)

    def decrypt_bytes(self, token: bytes) -> bytes:
        return self.fernet.decrypt(token)

Field-Level Encryption

from pydantic import BaseModel, Field
from typing import Optional

class EncryptedField:
    def __init__(self, encryption_service: EncryptionService):
        self.encryption = encryption_service

    def encrypt_field(self, value: str) -> str:
        return self.encryption.encrypt(value)

    def decrypt_field(self, value: str) -> str:
        return self.encryption.decrypt(value)

# Pydantic model with encrypted fields
class UserRecord(BaseModel):
    id: str
    email: str
    encrypted_ssn: Optional[str] = Field(default=None, description="Encrypted SSN")
    encrypted_api_key: Optional[str] = Field(default=None, description="Encrypted API key")

    def set_ssn(self, ssn: str, encryption: EncryptionService):
        self.encrypted_ssn = encryption.encrypt(ssn)

    def get_ssn(self, encryption: EncryptionService) -> str:
        return encryption.decrypt(self.encrypted_ssn) if self.encrypted_ssn else None

Key Management

class KeyManager:
    def __init__(self, vault_client: VaultClient):
        self.vault = vault_client

    async def get_key(self, key_name: str) -> bytes:
        """Retrieve encryption key from Vault"""
        secret = await self.vault.read_secret(f"keys/{key_name}")
        return base64.b64decode(secret["data"]["key"])

    async def rotate_key(self, key_name: str) -> bytes:
        """Generate new key version"""
        new_key = Fernet.generate_key()
        await self.vault.write_secret(f"keys/{key_name}", {"key": base64.b64encode(new_key).decode()})
        return new_key

    async def get_key_version(self, key_name: str, version: int) -> bytes:
        """Get specific key version for decryption"""
        secret = await self.vault.read_secret_version(f"keys/{key_name}", version)
        return base64.b64decode(secret["data"]["key"])

# Key rotation scheduler
class KeyRotationScheduler:
    def __init__(self, key_manager: KeyManager, interval_days: int = 90):
        self.key_manager = key_manager
        self.interval = interval_days

    async def rotate_all_keys(self):
        keys = await self.list_keys()
        for key in keys:
            if await self.should_rotate(key):
                await self.key_manager.rotate_key(key)
                await self.reencrypt_data(key)

PII Protection

PII Detection & Masking

import re
from typing import Optional

class PIIDetector:
    PATTERNS = {
        "email": re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
        "phone": re.compile(r'(\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}'),
        "ssn": re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
        "credit_card": re.compile(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b'),
        "ip_address": re.compile(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b'),
        "api_key": re.compile(r'(sk|pk)_[a-zA-Z0-9]{32,}'),
        "jwt": re.compile(r'eyJ[A-Za-z0-9_-]+\.eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+'),
    }

    def detect(self, text: str) -> dict[str, list[str]]:
        found = {}
        for pii_type, pattern in self.PATTERNS.items():
            matches = pattern.findall(text)
            if matches:
                found[pii_type] = matches
        return found

    def mask(self, text: str, mask_char: str = "*") -> str:
        result = text
        for pii_type, pattern in self.PATTERNS.items():
            result = pattern.sub(lambda m: mask_char * len(m.group()), result)
        return result

    def redact(self, text: str, pii_types: list[str] = None) -> str:
        types = pii_types or list(self.PATTERNS.keys())
        result = text
        for pii_type in types:
            if pii_type in self.PATTERNS:
                result = self.PATTERNS[pii_type].sub("[REDACTED]", result)
        return result

# Usage in logging
class PIISafeLogger:
    def __init__(self, logger, pii_detector: PIIDetector):
        self.logger = logger
        self.detector = pii_detector

    def info(self, message: str, **kwargs):
        # Redact PII from message
        safe_message = self.detector.redact(message)
        safe_kwargs = {k: self.detector.redact(str(v)) if isinstance(v, str) else v 
                       for k, v in kwargs.items()}
        self.logger.info(safe_message, **safe_kwargs)

PII in Logs & Metrics

class PIISafeMetrics:
    def __init__(self, pii_detector: PIIDetector):
        self.detector = pii_detector

    def sanitize_labels(self, labels: dict) -> dict:
        """Remove PII from metric labels"""
        sanitized = {}
        for k, v in labels.items():
            if isinstance(v, str):
                sanitized[k] = self.detector.redact(v)
            else:
                sanitized[k] = v
        return sanitized

    def record_request(self, agent: str, user_id: str, **labels):
        # Hash user_id instead of storing directly
        safe_labels = {
            "agent": self.sanitize_agent_name(labels.get("agent")),
            "user_hash": hashlib.sha256(user_id.encode()).hexdigest()[:8],
            **{k: v for k, v in labels.items() if k not in ["user_id", "email", "name"]}
        }
        REQUEST_COUNTER.labels(**safe_labels).inc()

Secrets Management

Vault Integration

import hvac

class VaultClient:
    def __init__(self, url: str, token: str = None):
        self.client = hvac.Client(url=url, token=token or os.environ.get("VAULT_TOKEN"))

    async def read_secret(self, path: str) -> dict:
        response = self.client.secrets.kv.v2.read_secret_version(path=path)
        return response["data"]["data"]

    async def write_secret(self, path: str, data: dict):
        self.client.secrets.kv.v2.create_or_update_secret(path=path, secret=data)

    async def generate_database_credentials(self, role: str) -> DatabaseCredentials:
        response = self.client.secrets.database.generate_credentials(role)
        return DatabaseCredentials(
            username=response["data"]["username"],
            password=response["data"]["password"]
        )

    async def issue_certificate(self, role: str, common_name: str) -> Certificate:
        response = self.client.secrets.pki.generate_certificate(
            name=role,
            common_name=common_name
        )
        return Certificate(
            certificate=response["data"]["certificate"],
            private_key=response["data"]["private_key"],
            ca_chain=response["data"]["ca_chain"]
        )

# Kubernetes External Secrets Operator
# external-secret.yaml
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
  name: agent-secrets
spec:
  refreshInterval: 1h
  secretStoreRef:
    name: vault-backend
    kind: ClusterSecretStore
  target:
    name: agent-secrets
    creationPolicy: Owner
  data:
  - secretKey: openai-api-key
    remoteRef:
      key: secret/agents/openai
      property: api_key
  - secretKey: langchain-api-key
    remoteRef:
      key: secret/agents/langchain
      property: api_key

Environment-Based Secrets

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore"
    )

    # API Keys
    OPENAI_API_KEY: str
    ANTHROPIC_API_KEY: Optional[str] = None
    ANTHROPIC_API_KEY: Optional[str] = None
    LANGCHAIN_API_KEY: Optional[str] = None

    # Database
    DATABASE_URL: str
    DATABASE_POOL_SIZE: int = 20

    # Redis
    REDIS_URL: str

    # Security
    SECRET_KEY: str
    ENCRYPTION_KEY: str
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    # External Services
    VAULT_URL: Optional[str] = None
    VAULT_TOKEN: Optional[str] = None

    # Feature Flags
    ENABLE_TRACING: bool = True
    ENABLE_METRICS: bool = True

@lru_cache()
def get_settings() -> Settings:
    return Settings()

Network Security

mTLS Configuration

# Istio PeerAuthentication
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: production
spec:
  mtls:
    mode: STRICT
---
# AuthorizationPolicy
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: agent-service-authz
spec:
  selector:
    matchLabels:
      app: agent-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/production/sa/agent-service"]
    - source:
        principals: ["cluster.local/ns/production/sa/gateway"]
  rules:
  - to:
    - operation:
        methods: ["GET", "POST"]
        paths: ["/health/*", "/api/*"]
---
# DestinationRule for mTLS
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: agent-service
spec:
  host: agent-service
  trafficPolicy:
    tls:
      mode: ISTIO_MUTUAL

Network Policies (Kubernetes)

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agent-service-netpol
  namespace: production
spec:
  podSelector:
    matchLabels:
      app: agent-service
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          name: ingress-nginx
    - podSelector:
        matchLabels:
          app: gateway
    ports:
    - protocol: TCP
      port: 8080
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: kube-system
    ports:
    - protocol: TCP
      port: 53  # DNS
    - protocol: UDP
      port: 53
  - to:
    - podSelector:
        matchLabels:
          app: postgres
    ports:
    - protocol: TCP
      port: 5432
  - to:
    - podSelector:
        matchLabels:
          app: redis
    ports:
    - protocol: TCP
      port: 6379
  - to:
    - namespaceSelector: {}
    ports:
    - protocol: TCP
      port: 443  # HTTPS for external APIs

WAF Rules (ModSecurity/Cloudflare)

# ModSecurity rules for agent endpoints
SecRule REQUEST_HEADERS:Content-Type "application/json" \
  "id:1001,phase:2,deny,status:400,msg:'Invalid Content-Type'"

SecRule REQUEST_BODY "@contains ignore previous instructions" \
  "id:1002,phase:2,deny,status:403,msg:'Prompt injection attempt'"

SecRule REQUEST_BODY "@contains ignore previous instructions" \
  "id:1003,phase:2,deny,status:403,msg:'Prompt injection attempt'"

SecRule REQUEST_BODY "@rx (?i)(ignore|ignore all|forget|disregard).*(previous|prior|above|earlier).*(instruction|instruction|prompt|directive)" \
  "id:1004,phase:2,deny,status:403,msg:'Prompt injection pattern detected'"

# Rate limiting
SecRule IP:BF_COUNTER "@gt 100" \
  "id:1005,phase:1,deny,status:429,msg:'Rate limit exceeded'"

Container Security

Image Scanning

# Trivy scan in CI
- name: Scan image
  uses: aquasecurity/trivy-action@master
  with:
    image-ref: ${{ env.REGISTRY }}/agent-service:${{ github.sha }}
    format: 'sarif'
    output: 'trivy-results.sarif'
    severity: 'CRITICAL,HIGH'
    exit-code: '1'
    ignore-unfixed: true

# Syft for SBOM
- name: Generate SBOM
  uses: anchore/sbom-action@v0
  with:
    image: ${{ env.REGISTRY }}/agent-service:${{ github.sha }}
    format: spdx-json
    output-file: sbom.spdx.json

Dockerfile Security

# Dockerfile
FROM python:3.11-slim AS builder

# Non-root user
RUN groupadd -r appuser && useradd -r -g appuser appuser

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

FROM python:3.11-slim

COPY --from=builder /root/.local /home/appuser/.local
COPY --chown=appuser:appuser . /app

USER appuser
WORKDIR /app

ENV PATH=/home/appuser/.local/bin:$PATH
ENV PYTHONUNBUFFERED=1

# Read-only root filesystem
# (Set in Kubernetes: readOnlyRootFilesystem: true)

EXPOSE 8080
CMD ["python", "-m", "agent_service"]

Runtime Security (Falco)

# falco-rules.yaml
- rule: Terminal shell in container
  desc: Detect interactive shell in container
  condition: >
    spawned_process and container
    and proc.name in (bash, sh, zsh, ksh, fish)
    and proc.aname[1] in (-i, -l, --interactive, --login)
  output: "Shell spawned in container (user=%user.name container=%container.name)"
  priority: WARNING
  tags: [container, shell, mitre_execution]

- rule: Suspicious network connection
  condition: >
    container and
    fd.type in (ipv4, ipv6) and
    fd.sip not in (allowed_ips) and
    fd.sport not in (allowed_ports)
  output: "Suspicious outbound connection (container=%container.name ip=%fd.rip)"
  priority: WARNING

Supply Chain Security

SBOM Generation

# Generate SBOM with Syft
syft packages dir:. -o cyclonedx-json=sbom.json

# Or with Trivy
trivy sbom --format cyclonedx --output sbom.json .

Image Signing (Cosign)

# Sign image
cosign sign --yes ghcr.io/myorg/agent-service:v1.2.3

# Verify
cosign verify --certificate-identity-regexp=".*" \
  --certificate-oidc-issuer-regexp=".*" \
  ghcr.io/myorg/agent-service:v1.2.3

SLSA Provenance

# .github/workflows/slsa.yml
name: SLSA Build
on: [push]
jobs:
  build:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      id-token: write
      attestations: write
    uses: slsa-framework/slsa-github-generator/.github/workflows/generator_generic_slsa3.yml@v1.9.0
    with:
      base64-subjects: "${{ hash_files('agent-service') }}"

Compliance

SOC2 / PCI-DSS / HIPAA Checks

class ComplianceChecker:
    def __init__(self):
        self.checks = {
            "SOC2": SOC2Checks(),
            "PCI-DSS": PCIDSSChecks(),
            "HIPAA": HIPAAChecks(),
            "GDPR": GDPRChecks()
        }

    async def assess(self, framework: str, resources: list[Resource]) -> ComplianceReport:
        checks = self.checks.get(framework)
        if not checks:
            raise ValueError(f"Unknown framework: {framework}")

        results = []
        for resource in resources:
            result = await checks.run(resource)
            results.append(result)

        return ComplianceReport(
            framework=framework,
            passed=sum(1 for r in results if r.passed),
            failed=sum(1 for r in results if not r.passed),
            findings=[r for r in results if not r.passed]
        )

class SOC2Checks:
    async def run(self, resource: Resource) -> CheckResult:
        checks = [
            self.check_encryption_at_rest(resource),
            self.check_encryption_in_transit(resource),
            self.check_access_control(resource),
            self.check_audit_logging(resource),
            self.check_incident_response(resource),
            self.check_vulnerability_management(resource)
        ]
        results = await asyncio.gather(*checks)
        return CheckResult(
            passed=all(r.passed for r in results),
            details=results
        )

Audit Logging

Structured Audit Logs

class AuditLogger:
    def __init__(self, logger, encryption: EncryptionService):
        self.logger = logger
        self.encryption = encryption

    def log(self, event: AuditEvent):
        # Encrypt sensitive fields
        encrypted_data = self.encryption.encrypt(json.dumps(event.data))

        log_entry = {
            "event_id": event.event_id,
            "event_type": event.event_type,
            "actor": event.actor,
            "resource": event.resource,
            "action": event.action,
            "outcome": event.outcome,
            "encrypted_data": encrypted_data,
            "timestamp": datetime.utcnow().isoformat(),
            "correlation_id": event.correlation_id
        }

        self.logger.info("audit_event", **log_entry)

@dataclass
class AuditEvent:
    event_id: str
    event_type: str  # create, read, update, delete, login, logout, permission_change
    actor: Actor
    resource: Resource
    action: str
    outcome: str  # success, failure, partial
    data: dict
    correlation_id: str = None

@dataclass
class Actor:
    id: str
    type: str  # user, service, system
    roles: list[str]
    ip_address: str = None

@dataclass
class Resource:
    id: str
    type: str
    owner_id: str = None
    sensitivity: str = "low"  # low, medium, high, critical

Log Retention & Integrity

class AuditLogManager:
    def __init__(self, storage: StorageBackend, encryption: EncryptionService):
        self.storage = storage
        self.encryption = encryption

    async def write(self, event: AuditEvent):
        # Immutable append-only write
        log_entry = self.format_entry(event)
        await self.storage.append("audit_logs", log_entry)

    async def verify_integrity(self, start: datetime, end: datetime) -> IntegrityReport:
        """Verify log integrity using hash chains"""
        entries = await self.storage.query("audit_logs", start, end)

        prev_hash = None
        violations = []

        for entry in entries:
            current_hash = self.compute_hash(entry, prev_hash)
            if entry.hash != current_hash:
                violations.append(IntegrityViolation(
                    entry_id=entry.id,
                    expected=current_hash,
                    actual=entry.hash
                ))
            prev_hash = current_hash

        return IntegrityReport(
            total_entries=len(entries),
            violations=violations,
            verified=len(violations) == 0
        )

Incident Response

Incident Response Plan

class IncidentResponse:
    def __init__(self, notifier: Notifier, runbook_store: RunbookStore):
        self.notifier = notifier
        self.runbooks = runbook_store

    async def handle_incident(self, alert: Alert) -> IncidentResponse:
        incident = Incident(
            id=str(uuid4()),
            alert=alert,
            severity=alert.severity,
            status="open",
            created_at=datetime.utcnow(),
            assignee=await self.assign_owner(alert)
        )

        # Notify
        await self.notifier.notify(incident)

        # Execute runbook
        runbook = await self.runbooks.get(alert.category)
        if runbook:
            await self.execute_runbook(runbook, incident)

        return incident

    async def execute_runbook(self, runbook: Runbook, incident: Incident):
        for step in runbook.steps:
            if step.automated:
                await self.execute_automated_step(step, incident)
            else:
                await self.notify_manual_step(incident, step)

            if step.requires_approval:
                await self.wait_for_approval(incident, step)

        incident.status = "resolved"
        incident.resolved_at = datetime.utcnow()
        await self.save(incident)

Runbook Example

# runbooks/security_breach.yaml
name: security_breach
category: security
severity: critical
steps:
  - name: isolate_affected_systems
    description: "Isolate affected systems from network"
    automated: true
    script: "scripts/isolate_system.sh"
    timeout: 300

  - name: preserve_evidence
    description: "Preserve forensic evidence"
    automated: false
    instructions: |
      1. Take disk snapshots
      2. Capture memory dumps
      3. Save relevant logs
    timeout: 1800
    requires_approval: true

  - name: rotate_credentials
    description: "Rotate all compromised credentials"
    automated: true
    script: "scripts/rotate_credentials.sh"
    timeout: 600

  - name: notify_stakeholders
    description: "Notify stakeholders and regulators"
    automated: false
    instructions: "Send notification per communication plan"
    timeout: 3600
    requires_approval: true

Vulnerability Management

Automated Scanning

class VulnerabilityScanner:
    def __init__(self):
        self.scanners = {
            "container": TrivyScanner(),
            "code": SemgrepScanner(),
            "dependency": DependencyScanner(),
            "secret": SecretScanner()
        }

    async def scan_all(self, target: ScanTarget) -> ScanReport:
        results = await asyncio.gather(*[
            scanner.scan(target) for scanner in self.scanners.values()
        )

        vulnerabilities = []
        for result in results:
            vulnerabilities.extend(result.vulnerabilities)

        return ScanReport(
            target=target,
            vulnerabilities=vulnerabilities,
            summary=self.summarize(vulnerabilities)
        )

    def summarize(self, vulns: list[Vulnerability]) -> ScanSummary:
        by_severity = defaultdict(int)
        for v in vulns:
            by_severity[v.severity] += 1

        return ScanSummary(
            total=len(vulns),
            by_severity=dict(by_severity),
            fixable=sum(1 for v in vulns if v.fixable),
            highest_severity=max(v.severity for v in vulns) if vulns else "none"
        )

# CI Integration
# .github/workflows/vuln-scan.yml
name: Vulnerability Scan
on: [push, pull_request, schedule]
jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run Trivy
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          format: 'sarif'
          output: 'trivy-results.sarif'
      - name: Upload to Code Scanning
        uses: github/codeql-action/upload-sarif@v3
        with:
          sarif_file: 'trivy-results.sarif'

Related Notes

• Deployment Architecture — Secure deployment
• Observability — Security monitoring
• Reliability — Circuit breakers, reliability
• Compliance — Compliance
• Cost Optimization — Security cost optimization
• Secure Coding — Secure coding practices
• Supply Chain Security — Supply chain security
• Deployment Architecture — Secure deployment
• Reliability — Circuit breakers, reliability
• Observability — Security monitoring
• Secure Coding — Secure coding
• Supply Chain Security — Supply chain security
• CI CD Best Practices — CI/CD security
• Infrastructure as Code — IaC security
• Research Directions — Security research

Sources: OWASP Top 10, OWASP ASVS, NIST Cybersecurity Framework, MITRE ATT&CK, NIST SSDF, SLSA, OWASP LLM Top 10, OWASP LLM AI Security, Microsoft SDL, AWS Well-Architected Security Pillar, Google Cloud Security, Azure Security Benchmark, CNCF Security, Falco, Trivy, Grype, Syft, Cosign, Sigstore, Kyverno, OPA, Gatekeeper, Falco, Vault, External Secrets, Cert-Manager, Let's Encrypt, cert-manager, Istio, Linkerd, Cilium, NetworkPolicy, mTLS, SPIFFE, SPIRE, HashiCorp Vault, HashiCorp Boundary, HashiCorp Consul, HashiCorp Nomad.