Security and Compliance

📖 Introduction

Security in platform engineering is fundamentally different from traditional security approaches. Instead of being a gate at the end of the development process, security becomes an enabler embedded throughout the platform. The goal is to make the secure path the easy path—developers shouldn't need to think about security because the platform handles it by default.

In my experience building platforms, the teams that succeed with security are those who treat it as a feature of the platform, not a constraint imposed from outside.

---

🎯 Security Philosophy

Shift-Left Security

Security as a Platform Feature

Traditional	Platform Engineering
Security team reviews code	Templates include security by default
Manual compliance checks	Automated policy enforcement
Separate security tools	Integrated into developer workflow
Approval gates	Guardrails with escape hatches
"Security says no"	"Platform enables secure practices"

---

🔒 Policy as Code

Why Policy as Code?

• Consistent - Same rules everywhere

• Auditable - Changes tracked in Git

• Testable - Validate before deployment

• Self-Service - No waiting for approvals

• Scalable - Enforce across thousands of resources

Open Policy Agent (OPA)

# policies/kubernetes/deployment.rego
package kubernetes.admission

import future.keywords.in

# Deny deployments without resource limits
deny[msg] {
    input.request.kind.kind == "Deployment"
    container := input.request.object.spec.template.spec.containers[_]
    not container.resources.limits
    msg := sprintf("Container '%s' must have resource limits", [container.name])
}

# Deny containers running as root
deny[msg] {
    input.request.kind.kind == "Deployment"
    container := input.request.object.spec.template.spec.containers[_]
    container.securityContext.runAsUser == 0
    msg := sprintf("Container '%s' cannot run as root", [container.name])
}

# Require approved image registries
deny[msg] {
    input.request.kind.kind == "Deployment"
    container := input.request.object.spec.template.spec.containers[_]
    not approved_registry(container.image)
    msg := sprintf("Image '%s' is not from an approved registry", [container.image])
}

approved_registry(image) {
    approved := ["gcr.io/company-prod/", "registry.company.com/"]
    startswith(image, approved[_])
}

# Require specific labels
deny[msg] {
    input.request.kind.kind == "Deployment"
    required := {"app", "team", "environment"}
    provided := {k | input.request.object.metadata.labels[k]}
    missing := required - provided
    count(missing) > 0
    msg := sprintf("Missing required labels: %v", [missing])
}

# Enforce pod disruption budgets for production
deny[msg] {
    input.request.kind.kind == "Deployment"
    input.request.object.metadata.labels.environment == "production"
    input.request.object.spec.replicas >= 3
    not has_pdb(input.request.object.metadata.name, input.request.object.metadata.namespace)
    msg := "Production deployments with 3+ replicas require a PodDisruptionBudget"
}

has_pdb(name, namespace) {
    # In practice, this would query existing PDBs
    true
}

Kyverno Policies

# policies/require-labels.yaml
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: require-labels
  annotations:
    policies.kyverno.io/title: Require Labels
    policies.kyverno.io/category: Best Practices
    policies.kyverno.io/severity: medium
spec:
  validationFailureAction: Enforce
  background: true
  rules:
    - name: check-required-labels
      match:
        any:
          - resources:
              kinds:
                - Deployment
                - Service
                - ConfigMap
      validate:
        message: "Resources must have 'team' and 'app' labels"
        pattern:
          metadata:
            labels:
              team: "?*"
              app: "?*"

---
# policies/restrict-image-registries.yaml
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: restrict-image-registries
spec:
  validationFailureAction: Enforce
  rules:
    - name: validate-registries
      match:
        any:
          - resources:
              kinds:
                - Pod
      validate:
        message: "Images must be from approved registries"
        pattern:
          spec:
            containers:
              - image: "gcr.io/company-*/* | registry.company.com/*"
            initContainers:
              - image: "gcr.io/company-*/* | registry.company.com/*"

---
# policies/add-default-security-context.yaml
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: add-default-security-context
spec:
  rules:
    - name: add-security-context
      match:
        any:
          - resources:
              kinds:
                - Pod
      mutate:
        patchStrategicMerge:
          spec:
            securityContext:
              runAsNonRoot: true
              seccompProfile:
                type: RuntimeDefault
            containers:
              - (name): "*"
                securityContext:
                  allowPrivilegeEscalation: false
                  capabilities:
                    drop:
                      - ALL

Python Policy Validator

"""
Policy validation service for platform resources.
"""

from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional
from enum import Enum
import re
import json


class PolicySeverity(Enum):
    ERROR = "error"  # Blocks deployment
    WARNING = "warning"  # Logged, may require approval
    INFO = "info"  # Informational only


@dataclass
class PolicyViolation:
    """Represents a policy violation."""
    policy_name: str
    severity: PolicySeverity
    message: str
    resource_path: str
    remediation: Optional[str] = None


@dataclass
class ValidationResult:
    """Result of policy validation."""
    valid: bool
    violations: List[PolicyViolation] = field(default_factory=list)
    
    @property
    def errors(self) -> List[PolicyViolation]:
        return [v for v in self.violations if v.severity == PolicySeverity.ERROR]
    
    @property
    def warnings(self) -> List[PolicyViolation]:
        return [v for v in self.violations if v.severity == PolicySeverity.WARNING]


class SecurityPolicyEngine:
    """Evaluate security policies against resources."""
    
    APPROVED_REGISTRIES = [
        "gcr.io/company-prod/",
        "gcr.io/company-staging/",
        "registry.company.com/",
    ]
    
    REQUIRED_LABELS = ["app", "team", "environment"]
    
    BANNED_CAPABILITIES = [
        "SYS_ADMIN", "NET_ADMIN", "SYS_PTRACE", "SYS_RAWIO"
    ]
    
    def validate(self, resource: Dict[str, Any]) -> ValidationResult:
        """Validate a resource against all policies."""
        violations = []
        
        kind = resource.get("kind", "")
        
        if kind == "Deployment":
            violations.extend(self._validate_deployment(resource))
        elif kind == "Pod":
            violations.extend(self._validate_pod(resource))
        elif kind == "Service":
            violations.extend(self._validate_service(resource))
        elif kind == "NetworkPolicy":
            violations.extend(self._validate_network_policy(resource))
        
        # Common validations
        violations.extend(self._validate_labels(resource))
        violations.extend(self._validate_annotations(resource))
        
        has_errors = any(v.severity == PolicySeverity.ERROR for v in violations)
        
        return ValidationResult(valid=not has_errors, violations=violations)
    
    def _validate_deployment(self, resource: Dict[str, Any]) -> List[PolicyViolation]:
        """Validate Deployment resources."""
        violations = []
        spec = resource.get("spec", {})
        template = spec.get("template", {})
        pod_spec = template.get("spec", {})
        
        for i, container in enumerate(pod_spec.get("containers", [])):
            path = f"spec.template.spec.containers[{i}]"
            
            # Check image registry
            image = container.get("image", "")
            if not any(image.startswith(reg) for reg in self.APPROVED_REGISTRIES):
                violations.append(PolicyViolation(
                    policy_name="approved-image-registry",
                    severity=PolicySeverity.ERROR,
                    message=f"Image '{image}' is not from an approved registry",
                    resource_path=f"{path}.image",
                    remediation=f"Use images from: {', '.join(self.APPROVED_REGISTRIES)}",
                ))
            
            # Check for latest tag
            if image.endswith(":latest") or ":" not in image:
                violations.append(PolicyViolation(
                    policy_name="no-latest-tag",
                    severity=PolicySeverity.ERROR,
                    message=f"Image '{image}' uses 'latest' tag or no tag",
                    resource_path=f"{path}.image",
                    remediation="Use specific image tags for reproducibility",
                ))
            
            # Check resource limits
            resources = container.get("resources", {})
            if not resources.get("limits"):
                violations.append(PolicyViolation(
                    policy_name="resource-limits-required",
                    severity=PolicySeverity.ERROR,
                    message=f"Container '{container.get('name')}' missing resource limits",
                    resource_path=f"{path}.resources.limits",
                    remediation="Add resources.limits with cpu and memory",
                ))
            
            # Check security context
            sec_ctx = container.get("securityContext", {})
            if sec_ctx.get("privileged", False):
                violations.append(PolicyViolation(
                    policy_name="no-privileged-containers",
                    severity=PolicySeverity.ERROR,
                    message=f"Container '{container.get('name')}' runs as privileged",
                    resource_path=f"{path}.securityContext.privileged",
                    remediation="Remove privileged: true from security context",
                ))
            
            if sec_ctx.get("runAsUser") == 0:
                violations.append(PolicyViolation(
                    policy_name="no-root-user",
                    severity=PolicySeverity.ERROR,
                    message=f"Container '{container.get('name')}' runs as root",
                    resource_path=f"{path}.securityContext.runAsUser",
                    remediation="Set runAsUser to a non-zero value",
                ))
            
            # Check capabilities
            capabilities = sec_ctx.get("capabilities", {})
            for cap in capabilities.get("add", []):
                if cap in self.BANNED_CAPABILITIES:
                    violations.append(PolicyViolation(
                        policy_name="restricted-capabilities",
                        severity=PolicySeverity.ERROR,
                        message=f"Capability '{cap}' is not allowed",
                        resource_path=f"{path}.securityContext.capabilities.add",
                        remediation=f"Remove '{cap}' from capabilities.add",
                    ))
            
            # Check probes
            if not container.get("livenessProbe"):
                violations.append(PolicyViolation(
                    policy_name="health-probes-required",
                    severity=PolicySeverity.WARNING,
                    message=f"Container '{container.get('name')}' missing liveness probe",
                    resource_path=f"{path}.livenessProbe",
                    remediation="Add livenessProbe for better reliability",
                ))
            
            if not container.get("readinessProbe"):
                violations.append(PolicyViolation(
                    policy_name="health-probes-required",
                    severity=PolicySeverity.WARNING,
                    message=f"Container '{container.get('name')}' missing readiness probe",
                    resource_path=f"{path}.readinessProbe",
                    remediation="Add readinessProbe for proper traffic management",
                ))
        
        # Check replica count for production
        metadata = resource.get("metadata", {})
        labels = metadata.get("labels", {})
        if labels.get("environment") == "production":
            replicas = spec.get("replicas", 1)
            if replicas < 3:
                violations.append(PolicyViolation(
                    policy_name="production-minimum-replicas",
                    severity=PolicySeverity.ERROR,
                    message=f"Production deployments require at least 3 replicas (found {replicas})",
                    resource_path="spec.replicas",
                    remediation="Set replicas to at least 3 for production",
                ))
        
        return violations
    
    def _validate_labels(self, resource: Dict[str, Any]) -> List[PolicyViolation]:
        """Validate required labels are present."""
        violations = []
        metadata = resource.get("metadata", {})
        labels = metadata.get("labels", {})
        
        missing = [l for l in self.REQUIRED_LABELS if l not in labels]
        if missing:
            violations.append(PolicyViolation(
                policy_name="required-labels",
                severity=PolicySeverity.ERROR,
                message=f"Missing required labels: {', '.join(missing)}",
                resource_path="metadata.labels",
                remediation=f"Add labels: {', '.join(missing)}",
            ))
        
        return violations
    
    def _validate_annotations(self, resource: Dict[str, Any]) -> List[PolicyViolation]:
        """Validate annotations."""
        violations = []
        metadata = resource.get("metadata", {})
        annotations = metadata.get("annotations", {})
        
        # Check for owner annotation
        if "platform.example.com/owner" not in annotations:
            violations.append(PolicyViolation(
                policy_name="owner-annotation",
                severity=PolicySeverity.WARNING,
                message="Missing owner annotation",
                resource_path="metadata.annotations",
                remediation="Add platform.example.com/owner annotation",
            ))
        
        return violations
    
    def _validate_pod(self, resource: Dict[str, Any]) -> List[PolicyViolation]:
        """Validate Pod resources."""
        # Pods should generally come through Deployments
        violations = []
        
        violations.append(PolicyViolation(
            policy_name="no-naked-pods",
            severity=PolicySeverity.WARNING,
            message="Direct Pod creation is discouraged; use Deployment instead",
            resource_path="kind",
            remediation="Create a Deployment to manage Pods",
        ))
        
        return violations
    
    def _validate_service(self, resource: Dict[str, Any]) -> List[PolicyViolation]:
        """Validate Service resources."""
        violations = []
        spec = resource.get("spec", {})
        
        # Check for LoadBalancer type without annotations
        if spec.get("type") == "LoadBalancer":
            annotations = resource.get("metadata", {}).get("annotations", {})
            if "service.beta.kubernetes.io/aws-load-balancer-internal" not in annotations:
                violations.append(PolicyViolation(
                    policy_name="internal-load-balancer",
                    severity=PolicySeverity.WARNING,
                    message="LoadBalancer services should be internal by default",
                    resource_path="metadata.annotations",
                    remediation="Add internal load balancer annotation or use Ingress",
                ))
        
        return violations
    
    def _validate_network_policy(self, resource: Dict[str, Any]) -> List[PolicyViolation]:
        """Validate NetworkPolicy resources."""
        violations = []
        spec = resource.get("spec", {})
        
        # Check for deny-all by default
        ingress = spec.get("ingress", [])
        egress = spec.get("egress", [])
        
        if not ingress and not egress:
            violations.append(PolicyViolation(
                policy_name="network-policy-rules",
                severity=PolicySeverity.INFO,
                message="NetworkPolicy has no rules defined",
                resource_path="spec",
                remediation="Add ingress/egress rules or this will deny all traffic",
            ))
        
        return violations


class PolicyReporter:
    """Generate policy violation reports."""
    
    def __init__(self, engine: SecurityPolicyEngine):
        self.engine = engine
    
    def validate_and_report(self, resources: List[Dict[str, Any]]) -> str:
        """Validate resources and generate report."""
        all_violations = []
        
        for resource in resources:
            result = self.engine.validate(resource)
            name = resource.get("metadata", {}).get("name", "unknown")
            kind = resource.get("kind", "unknown")
            
            for violation in result.violations:
                violation.resource_path = f"{kind}/{name}: {violation.resource_path}"
                all_violations.append(violation)
        
        return self._format_report(all_violations)
    
    def _format_report(self, violations: List[PolicyViolation]) -> str:
        """Format violations as a report."""
        if not violations:
            return "✅ All policies passed!"
        
        lines = ["Policy Validation Report", "=" * 50, ""]
        
        errors = [v for v in violations if v.severity == PolicySeverity.ERROR]
        warnings = [v for v in violations if v.severity == PolicySeverity.WARNING]
        infos = [v for v in violations if v.severity == PolicySeverity.INFO]
        
        if errors:
            lines.append(f"❌ ERRORS ({len(errors)})")
            for v in errors:
                lines.append(f"  [{v.policy_name}] {v.message}")
                lines.append(f"    Resource: {v.resource_path}")
                if v.remediation:
                    lines.append(f"    Fix: {v.remediation}")
            lines.append("")
        
        if warnings:
            lines.append(f"⚠️ WARNINGS ({len(warnings)})")
            for v in warnings:
                lines.append(f"  [{v.policy_name}] {v.message}")
            lines.append("")
        
        if infos:
            lines.append(f"ℹ️ INFO ({len(infos)})")
            for v in infos:
                lines.append(f"  [{v.policy_name}] {v.message}")
        
        return "\n".join(lines)


# Usage example
if __name__ == "__main__":
    engine = SecurityPolicyEngine()
    reporter = PolicyReporter(engine)
    
    deployment = {
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": {
            "name": "my-app",
            "labels": {"app": "my-app"},  # Missing team, environment
        },
        "spec": {
            "replicas": 1,
            "template": {
                "spec": {
                    "containers": [{
                        "name": "app",
                        "image": "nginx:latest",  # Unapproved registry, latest tag
                        "securityContext": {
                            "privileged": True,  # Not allowed
                        },
                        # Missing resources, probes
                    }]
                }
            }
        }
    }
    
    report = reporter.validate_and_report([deployment])
    print(report)

---

🔐 RBAC and Access Control

Kubernetes RBAC

# Platform RBAC structure
---
# Team namespace admin role
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: team-admin
  namespace: checkout-team
rules:
  # Full access to applications
  - apiGroups: ["apps"]
    resources: ["deployments", "replicasets", "statefulsets"]
    verbs: ["*"]
  
  # Full access to core resources
  - apiGroups: [""]
    resources: ["pods", "services", "configmaps", "secrets"]
    verbs: ["*"]
  
  # Read-only access to events and logs
  - apiGroups: [""]
    resources: ["events", "pods/log"]
    verbs: ["get", "list", "watch"]
  
  # Platform custom resources
  - apiGroups: ["platform.example.com"]
    resources: ["applications", "databases"]
    verbs: ["*"]

---
# Team developer role (limited)
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: team-developer
  namespace: checkout-team
rules:
  - apiGroups: ["apps"]
    resources: ["deployments"]
    verbs: ["get", "list", "watch", "create", "update", "patch"]
    # Note: no delete
  
  - apiGroups: [""]
    resources: ["pods", "services", "configmaps"]
    verbs: ["get", "list", "watch"]
  
  - apiGroups: [""]
    resources: ["pods/log", "pods/exec"]
    verbs: ["get", "create"]
  
  # Can't access secrets directly
  # - apiGroups: [""]
  #   resources: ["secrets"]
  #   verbs: ["get", "list"]

---
# Cluster-wide read-only for platform visibility
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: platform-viewer
rules:
  - apiGroups: [""]
    resources: ["namespaces", "nodes"]
    verbs: ["get", "list", "watch"]
  
  - apiGroups: ["apps"]
    resources: ["deployments", "replicasets"]
    verbs: ["get", "list", "watch"]

---
# Bind to team group from identity provider
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: checkout-team-admins
  namespace: checkout-team
subjects:
  - kind: Group
    name: checkout-team-admins
    apiGroup: rbac.authorization.k8s.io
roleRef:
  kind: Role
  name: team-admin
  apiGroup: rbac.authorization.k8s.io

Dynamic RBAC Generator

"""
Generate RBAC resources for platform teams.
"""

from dataclasses import dataclass
from typing import List, Dict, Any
import yaml


@dataclass
class TeamConfig:
    """Configuration for a platform team."""
    name: str
    namespace: str
    admin_group: str
    developer_group: str
    readonly_group: str
    allowed_resources: List[str]


class RBACGenerator:
    """Generate Kubernetes RBAC resources."""
    
    def __init__(self, team: TeamConfig):
        self.team = team
    
    def generate_all(self) -> List[Dict[str, Any]]:
        """Generate all RBAC resources for a team."""
        resources = []
        
        # Namespace
        resources.append(self._generate_namespace())
        
        # Roles
        resources.append(self._generate_admin_role())
        resources.append(self._generate_developer_role())
        resources.append(self._generate_readonly_role())
        
        # RoleBindings
        resources.append(self._generate_admin_binding())
        resources.append(self._generate_developer_binding())
        resources.append(self._generate_readonly_binding())
        
        # NetworkPolicy for namespace isolation
        resources.append(self._generate_network_policy())
        
        # ResourceQuota
        resources.append(self._generate_resource_quota())
        
        return resources
    
    def _generate_namespace(self) -> Dict[str, Any]:
        return {
            "apiVersion": "v1",
            "kind": "Namespace",
            "metadata": {
                "name": self.team.namespace,
                "labels": {
                    "team": self.team.name,
                    "managed-by": "platform",
                },
            },
        }
    
    def _generate_admin_role(self) -> Dict[str, Any]:
        return {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "Role",
            "metadata": {
                "name": "team-admin",
                "namespace": self.team.namespace,
            },
            "rules": [
                {
                    "apiGroups": ["", "apps", "batch"],
                    "resources": ["*"],
                    "verbs": ["*"],
                },
                {
                    "apiGroups": ["platform.example.com"],
                    "resources": self.team.allowed_resources,
                    "verbs": ["*"],
                },
            ],
        }
    
    def _generate_developer_role(self) -> Dict[str, Any]:
        return {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "Role",
            "metadata": {
                "name": "team-developer",
                "namespace": self.team.namespace,
            },
            "rules": [
                {
                    "apiGroups": ["apps"],
                    "resources": ["deployments", "replicasets"],
                    "verbs": ["get", "list", "watch", "create", "update", "patch"],
                },
                {
                    "apiGroups": [""],
                    "resources": ["pods", "services", "configmaps"],
                    "verbs": ["get", "list", "watch", "create", "update", "patch"],
                },
                {
                    "apiGroups": [""],
                    "resources": ["pods/log", "pods/exec"],
                    "verbs": ["get", "create"],
                },
                {
                    "apiGroups": ["platform.example.com"],
                    "resources": self.team.allowed_resources,
                    "verbs": ["get", "list", "watch", "create", "update"],
                },
            ],
        }
    
    def _generate_readonly_role(self) -> Dict[str, Any]:
        return {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "Role",
            "metadata": {
                "name": "team-readonly",
                "namespace": self.team.namespace,
            },
            "rules": [
                {
                    "apiGroups": ["", "apps", "batch"],
                    "resources": ["*"],
                    "verbs": ["get", "list", "watch"],
                },
            ],
        }
    
    def _generate_admin_binding(self) -> Dict[str, Any]:
        return {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "RoleBinding",
            "metadata": {
                "name": "team-admin-binding",
                "namespace": self.team.namespace,
            },
            "subjects": [{
                "kind": "Group",
                "name": self.team.admin_group,
                "apiGroup": "rbac.authorization.k8s.io",
            }],
            "roleRef": {
                "kind": "Role",
                "name": "team-admin",
                "apiGroup": "rbac.authorization.k8s.io",
            },
        }
    
    def _generate_developer_binding(self) -> Dict[str, Any]:
        return {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "RoleBinding",
            "metadata": {
                "name": "team-developer-binding",
                "namespace": self.team.namespace,
            },
            "subjects": [{
                "kind": "Group",
                "name": self.team.developer_group,
                "apiGroup": "rbac.authorization.k8s.io",
            }],
            "roleRef": {
                "kind": "Role",
                "name": "team-developer",
                "apiGroup": "rbac.authorization.k8s.io",
            },
        }
    
    def _generate_readonly_binding(self) -> Dict[str, Any]:
        return {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "RoleBinding",
            "metadata": {
                "name": "team-readonly-binding",
                "namespace": self.team.namespace,
            },
            "subjects": [{
                "kind": "Group",
                "name": self.team.readonly_group,
                "apiGroup": "rbac.authorization.k8s.io",
            }],
            "roleRef": {
                "kind": "Role",
                "name": "team-readonly",
                "apiGroup": "rbac.authorization.k8s.io",
            },
        }
    
    def _generate_network_policy(self) -> Dict[str, Any]:
        """Default deny with allow same-namespace and system."""
        return {
            "apiVersion": "networking.k8s.io/v1",
            "kind": "NetworkPolicy",
            "metadata": {
                "name": "default-deny-with-exceptions",
                "namespace": self.team.namespace,
            },
            "spec": {
                "podSelector": {},
                "policyTypes": ["Ingress", "Egress"],
                "ingress": [
                    # Allow from same namespace
                    {"from": [{"podSelector": {}}]},
                    # Allow from ingress controller
                    {"from": [{"namespaceSelector": {"matchLabels": {"name": "ingress-nginx"}}}]},
                    # Allow from monitoring
                    {"from": [{"namespaceSelector": {"matchLabels": {"name": "monitoring"}}}]},
                ],
                "egress": [
                    # Allow to same namespace
                    {"to": [{"podSelector": {}}]},
                    # Allow DNS
                    {"to": [{"namespaceSelector": {}}], "ports": [{"port": 53, "protocol": "UDP"}]},
                    # Allow to platform services
                    {"to": [{"namespaceSelector": {"matchLabels": {"name": "platform-system"}}}]},
                ],
            },
        }
    
    def _generate_resource_quota(self) -> Dict[str, Any]:
        """Default resource quota for namespace."""
        return {
            "apiVersion": "v1",
            "kind": "ResourceQuota",
            "metadata": {
                "name": "team-quota",
                "namespace": self.team.namespace,
            },
            "spec": {
                "hard": {
                    "requests.cpu": "10",
                    "requests.memory": "20Gi",
                    "limits.cpu": "20",
                    "limits.memory": "40Gi",
                    "persistentvolumeclaims": "10",
                    "services.loadbalancers": "2",
                },
            },
        }


# Usage
team = TeamConfig(
    name="checkout",
    namespace="checkout-team",
    admin_group="checkout-admins",
    developer_group="checkout-developers",
    readonly_group="checkout-viewers",
    allowed_resources=["applications", "databases", "caches"],
)

generator = RBACGenerator(team)
resources = generator.generate_all()

for resource in resources:
    print(yaml.dump(resource, default_flow_style=False))
    print("---")

---

🛡️ Supply Chain Security

Container Image Scanning

# Trivy container scanning in CI/CD
name: Container Security Scan

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Build image
        run: docker build -t ${{ github.repository }}:${{ github.sha }} .
      
      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: '${{ github.repository }}:${{ github.sha }}'
          format: 'sarif'
          output: 'trivy-results.sarif'
          severity: 'CRITICAL,HIGH'
          exit-code: '1'
      
      - name: Upload Trivy scan results
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: 'trivy-results.sarif'

Image Signing with Cosign

"""
Container image signing and verification.
"""

import subprocess
import json
from dataclasses import dataclass
from typing import Optional, List


@dataclass
class SignatureVerification:
    """Result of signature verification."""
    valid: bool
    signatures: List[dict]
    message: Optional[str] = None


class ImageSigner:
    """Sign and verify container images."""
    
    def __init__(self, key_path: str):
        self.key_path = key_path
    
    def sign(self, image: str) -> bool:
        """Sign a container image with cosign."""
        try:
            result = subprocess.run(
                [
                    "cosign", "sign",
                    "--key", self.key_path,
                    "--yes",  # Skip confirmation
                    image,
                ],
                capture_output=True,
                text=True,
            )
            return result.returncode == 0
        except Exception as e:
            print(f"Signing failed: {e}")
            return False
    
    def verify(self, image: str, public_key: str) -> SignatureVerification:
        """Verify a container image signature."""
        try:
            result = subprocess.run(
                [
                    "cosign", "verify",
                    "--key", public_key,
                    "--output", "json",
                    image,
                ],
                capture_output=True,
                text=True,
            )
            
            if result.returncode == 0:
                signatures = json.loads(result.stdout)
                return SignatureVerification(
                    valid=True,
                    signatures=signatures,
                )
            else:
                return SignatureVerification(
                    valid=False,
                    signatures=[],
                    message=result.stderr,
                )
        
        except Exception as e:
            return SignatureVerification(
                valid=False,
                signatures=[],
                message=str(e),
            )
    
    def generate_sbom(self, image: str, output_path: str) -> bool:
        """Generate Software Bill of Materials."""
        try:
            result = subprocess.run(
                [
                    "syft", image,
                    "-o", f"spdx-json={output_path}",
                ],
                capture_output=True,
            )
            return result.returncode == 0
        except Exception as e:
            print(f"SBOM generation failed: {e}")
            return False
    
    def attach_sbom(self, image: str, sbom_path: str) -> bool:
        """Attach SBOM as attestation to image."""
        try:
            result = subprocess.run(
                [
                    "cosign", "attach", "sbom",
                    "--sbom", sbom_path,
                    "--type", "spdx",
                    image,
                ],
                capture_output=True,
            )
            return result.returncode == 0
        except Exception as e:
            print(f"SBOM attachment failed: {e}")
            return False


class AdmissionController:
    """Verify image signatures before admission."""
    
    def __init__(self, trusted_keys: List[str]):
        self.trusted_keys = trusted_keys
        self.signer = ImageSigner("")  # No private key needed for verification
    
    def verify_image(self, image: str) -> bool:
        """Verify image is signed by a trusted key."""
        for key in self.trusted_keys:
            result = self.signer.verify(image, key)
            if result.valid:
                return True
        return False
    
    def admission_review(self, request: dict) -> dict:
        """Handle Kubernetes admission review."""
        pod_spec = request.get("request", {}).get("object", {}).get("spec", {})
        containers = pod_spec.get("containers", []) + pod_spec.get("initContainers", [])
        
        for container in containers:
            image = container.get("image", "")
            if not self.verify_image(image):
                return {
                    "apiVersion": "admission.k8s.io/v1",
                    "kind": "AdmissionReview",
                    "response": {
                        "uid": request.get("request", {}).get("uid"),
                        "allowed": False,
                        "status": {
                            "code": 403,
                            "message": f"Image '{image}' is not signed by a trusted key",
                        },
                    },
                }
        
        return {
            "apiVersion": "admission.k8s.io/v1",
            "kind": "AdmissionReview",
            "response": {
                "uid": request.get("request", {}).get("uid"),
                "allowed": True,
            },
        }

---

🔑 Secrets Management

External Secrets Operator

# ExternalSecret fetches from external provider
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
  name: database-credentials
  namespace: checkout-team
spec:
  refreshInterval: 1h
  secretStoreRef:
    name: aws-secrets-manager
    kind: ClusterSecretStore
  target:
    name: database-credentials
    creationPolicy: Owner
  data:
    - secretKey: username
      remoteRef:
        key: checkout/database
        property: username
    - secretKey: password
      remoteRef:
        key: checkout/database
        property: password

---
# ClusterSecretStore connects to AWS Secrets Manager
apiVersion: external-secrets.io/v1beta1
kind: ClusterSecretStore
metadata:
  name: aws-secrets-manager
spec:
  provider:
    aws:
      service: SecretsManager
      region: us-east-1
      auth:
        jwt:
          serviceAccountRef:
            name: external-secrets
            namespace: external-secrets-system

---

📊 Compliance Dashboard

"""
Compliance monitoring and reporting.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Any
from enum import Enum


class ComplianceStatus(Enum):
    COMPLIANT = "compliant"
    NON_COMPLIANT = "non_compliant"
    UNKNOWN = "unknown"


@dataclass
class ComplianceCheck:
    """Individual compliance check result."""
    name: str
    category: str
    status: ComplianceStatus
    details: str
    resource_count: int
    failing_resources: List[str] = field(default_factory=list)


@dataclass
class ComplianceReport:
    """Overall compliance report."""
    timestamp: datetime
    total_resources: int
    compliant_count: int
    non_compliant_count: int
    checks: List[ComplianceCheck]
    
    @property
    def compliance_percentage(self) -> float:
        if self.total_resources == 0:
            return 100.0
        return (self.compliant_count / self.total_resources) * 100


class ComplianceMonitor:
    """Monitor platform compliance."""
    
    def __init__(self, k8s_client):
        self.k8s = k8s_client
    
    def run_all_checks(self) -> ComplianceReport:
        """Run all compliance checks."""
        checks = []
        
        checks.append(self._check_resource_limits())
        checks.append(self._check_image_registries())
        checks.append(self._check_security_context())
        checks.append(self._check_network_policies())
        checks.append(self._check_pod_disruption_budgets())
        checks.append(self._check_rbac_bindings())
        
        total = sum(c.resource_count for c in checks)
        compliant = sum(c.resource_count - len(c.failing_resources) for c in checks)
        non_compliant = sum(len(c.failing_resources) for c in checks)
        
        return ComplianceReport(
            timestamp=datetime.utcnow(),
            total_resources=total,
            compliant_count=compliant,
            non_compliant_count=non_compliant,
            checks=checks,
        )
    
    def _check_resource_limits(self) -> ComplianceCheck:
        """Check all pods have resource limits."""
        # In production, query Kubernetes API
        return ComplianceCheck(
            name="resource-limits",
            category="resource-management",
            status=ComplianceStatus.COMPLIANT,
            details="All pods have resource limits defined",
            resource_count=150,
            failing_resources=[],
        )
    
    def _check_image_registries(self) -> ComplianceCheck:
        """Check images from approved registries."""
        failing = ["default/old-app:nginx:latest"]
        status = ComplianceStatus.NON_COMPLIANT if failing else ComplianceStatus.COMPLIANT
        
        return ComplianceCheck(
            name="approved-registries",
            category="supply-chain",
            status=status,
            details="Images must be from approved registries",
            resource_count=150,
            failing_resources=failing,
        )
    
    def _check_security_context(self) -> ComplianceCheck:
        """Check security context settings."""
        return ComplianceCheck(
            name="security-context",
            category="security",
            status=ComplianceStatus.COMPLIANT,
            details="All containers have proper security context",
            resource_count=150,
            failing_resources=[],
        )
    
    def _check_network_policies(self) -> ComplianceCheck:
        """Check namespaces have network policies."""
        failing = ["test-namespace"]
        return ComplianceCheck(
            name="network-policies",
            category="network-security",
            status=ComplianceStatus.NON_COMPLIANT if failing else ComplianceStatus.COMPLIANT,
            details="All namespaces should have network policies",
            resource_count=20,
            failing_resources=failing,
        )
    
    def _check_pod_disruption_budgets(self) -> ComplianceCheck:
        """Check production workloads have PDBs."""
        return ComplianceCheck(
            name="pod-disruption-budgets",
            category="availability",
            status=ComplianceStatus.COMPLIANT,
            details="Production workloads have PDBs",
            resource_count=50,
            failing_resources=[],
        )
    
    def _check_rbac_bindings(self) -> ComplianceCheck:
        """Check RBAC bindings follow least privilege."""
        return ComplianceCheck(
            name="rbac-least-privilege",
            category="access-control",
            status=ComplianceStatus.COMPLIANT,
            details="RBAC bindings follow least privilege principle",
            resource_count=30,
            failing_resources=[],
        )
    
    def generate_report(self, report: ComplianceReport) -> str:
        """Generate human-readable report."""
        lines = [
            "Platform Compliance Report",
            f"Generated: {report.timestamp.isoformat()}",
            "=" * 50,
            "",
            f"Overall Compliance: {report.compliance_percentage:.1f}%",
            f"Total Resources: {report.total_resources}",
            f"Compliant: {report.compliant_count}",
            f"Non-Compliant: {report.non_compliant_count}",
            "",
            "Checks:",
        ]
        
        for check in report.checks:
            status_icon = "✅" if check.status == ComplianceStatus.COMPLIANT else "❌"
            lines.append(f"  {status_icon} {check.name} ({check.category})")
            lines.append(f"     {check.details}")
            if check.failing_resources:
                lines.append(f"     Failing: {', '.join(check.failing_resources[:5])}")
        
        return "\n".join(lines)


# Usage
monitor = ComplianceMonitor(None)  # Pass k8s client
report = monitor.run_all_checks()
print(monitor.generate_report(report))

---

✅ Best Practices

Security

1. Default deny - Start restrictive, open as needed

2. Least privilege - Minimum permissions required

3. Defense in depth - Multiple security layers

4. Automate everything - Manual checks don't scale

5. Fail secure - When in doubt, deny

Compliance

1. Continuous monitoring - Not point-in-time audits

2. Self-service remediation - Help teams fix issues

3. Clear documentation - Why policies exist

4. Graduated enforcement - Warn before blocking

5. Exception process - Documented escape hatches

---

🔗 What's Next?

In Article 11: Platform Observability and Metrics, we'll explore how to measure platform success using DORA metrics, developer productivity measurements, and feedback loops.

---

📚 References

• Open Policy Agent

• Kyverno

• Falco

• Cosign and Sigstore

• External Secrets Operator

• NIST Container Security Guide