Identity & Access Management, RBAC/ABAC, and Least-Privilege Governance for Distributed Systems¶

# k8s/admission-webhook.py
def validate_deployment(admission_request):
    """Prevent deploying to prod without approval"""
    namespace = admission_request.namespace
    labels = admission_request.object.metadata.labels

    if namespace == "production":
        # Check for approval annotation
        if "access.control/approved" not in admission_request.object.metadata.annotations:
            return {
                "allowed": False,
                "status": {
                    "reason": "Production deployments require approval annotation"
                }
            }

        # Prevent PII data in shared namespaces
        if labels.get("data_class") == "pii" and namespace != "pii-isolated":
            return {
                "allowed": False,
                "status": {
                    "reason": "PII data cannot be deployed to shared namespaces"
                }
            }

    return {"allowed": True}

# cicd/token-rotation.py
import secrets
import boto3
from datetime import datetime, timedelta

class TokenRotator:
    def __init__(self, k8s_client, vault_client):
        self.k8s = k8s_client
        self.vault = vault_client

    def rotate_ci_token(self, service_account: str, namespace: str):
        """Rotate CI/CD service account token"""
        # Generate new token
        new_token = secrets.token_urlsafe(32)

        # Update in Vault
        self.vault.write(
            f"secret/cicd/{service_account}",
            token=new_token,
            expires_at=(datetime.now() + timedelta(days=90)).isoformat()
        )

        # Update Kubernetes secret
        self.k8s.patch_secret(
            namespace=namespace,
            name=f"{service_account}-token",
            data={"token": new_token}
        )

        # Revoke old token after grace period
        # (implementation depends on system)

# api/oidc_auth.py
from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from jwt import PyJWKClient

class OIDCAuth:
    def __init__(self, issuer_url: str):
        self.issuer_url = issuer_url
        self.jwks_client = PyJWKClient(f"{issuer_url}/.well-known/jwks.json")
        self.security = HTTPBearer()

    async def verify_token(
        self,
        credentials: HTTPAuthorizationCredentials = Security(HTTPBearer())
    ) -> dict:
        """Verify JWT token and return claims"""
        try:
            # Get signing key
            signing_key = self.jwks_client.get_signing_key_from_jwt(
                credentials.credentials
            )

            # Decode token
            payload = jwt.decode(
                credentials.credentials,
                signing_key.key,
                algorithms=["RS256"],
                audience="api",
                issuer=self.issuer_url
            )

            return payload
        except jwt.InvalidTokenError as e:
            raise HTTPException(
                status_code=401,
                detail=f"Invalid token: {str(e)}"
            )

    def require_role(self, *allowed_roles: str):
        """Dependency to require specific roles"""
        async def role_checker(
            claims: dict = Depends(self.verify_token)
        ) -> dict:
            user_roles = claims.get("groups", [])

            if not any(role in user_roles for role in allowed_roles):
                raise HTTPException(
                    status_code=403,
                    detail=f"Requires one of: {allowed_roles}"
                )

            return claims

        return role_checker

# Usage
auth = OIDCAuth("https://keycloak.example.com/realms/myrealm")

@app.get("/api/admin/users")
async def list_users(
    claims: dict = Depends(auth.require_role("platform-admin"))
):
    """List users - requires platform-admin role"""
    return {"users": [...]}

@app.get("/api/data/query")
async def query_data(
    claims: dict = Depends(auth.verify_token)
):
    """Query data - any authenticated user"""
    # Check tenant access
    tenant_id = claims.get("tenant_id")
    return query_data_for_tenant(tenant_id)

# api/permissions.py
from enum import Enum
from typing import List

class Permission(Enum):
    READ_DATA = "read:data"
    WRITE_DATA = "write:data"
    MANAGE_MODELS = "manage:models"
    TRIGGER_TRAINING = "trigger:training"
    VIEW_ANALYTICS = "view:analytics"

class PermissionChecker:
    def __init__(self, claims: dict):
        self.claims = claims
        self.roles = claims.get("groups", [])
        self.permissions = self._map_roles_to_permissions()

    def _map_roles_to_permissions(self) -> List[Permission]:
        """Map roles to permissions"""
        permissions = []

        if "ml-engineer" in self.roles:
            permissions.extend([
                Permission.READ_DATA,
                Permission.WRITE_DATA,
                Permission.MANAGE_MODELS,
                Permission.TRIGGER_TRAINING
            ])

        if "analyst" in self.roles:
            permissions.extend([
                Permission.READ_DATA,
                Permission.VIEW_ANALYTICS
            ])

        return permissions

    def has_permission(self, permission: Permission) -> bool:
        """Check if user has specific permission"""
        return permission in self.permissions

# Usage in FastAPI
def require_permission(permission: Permission):
    """Dependency factory for permission checks"""
    async def permission_checker(
        claims: dict = Depends(auth.verify_token)
    ) -> dict:
        checker = PermissionChecker(claims)

        if not checker.has_permission(permission):
            raise HTTPException(
                status_code=403,
                detail=f"Requires permission: {permission.value}"
            )

        return claims

    return permission_checker

@app.post("/api/ml/train")
async def train_model(
    claims: dict = Depends(require_permission(Permission.TRIGGER_TRAINING))
):
    """Train model - requires TRIGGER_TRAINING permission"""
    return {"status": "training started"}

# api/api_keys.py
from datetime import datetime, timedelta
import hashlib
import secrets

class APIKeyManager:
    def __init__(self, db):
        self.db = db

    def create_api_key(
        self,
        service_name: str,
        scopes: List[str],
        expires_days: int = 90
    ) -> str:
        """Create API key with scopes and expiration"""
        # Generate key
        key = f"sk_{secrets.token_urlsafe(32)}"
        key_hash = hashlib.sha256(key.encode()).hexdigest()

        # Store in database
        self.db.execute("""
            INSERT INTO api_keys (service_name, key_hash, scopes, expires_at)
            VALUES (%s, %s, %s, %s)
        """, (
            service_name,
            key_hash,
            scopes,
            datetime.now() + timedelta(days=expires_days)
        ))

        return key

    async def verify_api_key(
        self,
        key: str,
        required_scope: str
    ) -> dict:
        """Verify API key and check scope"""
        key_hash = hashlib.sha256(key.encode()).hexdigest()

        result = self.db.execute("""
            SELECT service_name, scopes, expires_at
            FROM api_keys
            WHERE key_hash = %s AND expires_at > NOW()
        """, (key_hash,))

        if not result:
            raise HTTPException(status_code=401, detail="Invalid API key")

        scopes = result["scopes"]
        if required_scope not in scopes:
            raise HTTPException(
                status_code=403,
                detail=f"API key missing required scope: {required_scope}"
            )

        return {
            "service_name": result["service_name"],
            "scopes": scopes
        }

# Usage
api_key_auth = APIKeyManager(db)

@app.get("/api/data/query")
async def query_data_api(
    api_key: str = Header(..., alias="X-API-Key")
):
    """Query data via API key"""
    key_info = await api_key_auth.verify_api_key(api_key, "read:data")
    return query_data(service=key_info["service_name"])

# abac/policy_engine.py
from typing import Dict, List, Any
from dataclasses import dataclass

@dataclass
class Subject:
    """Subject attributes"""
    user_id: str
    roles: List[str]
    groups: List[str]
    environment: str
    tenant_id: str = None

@dataclass
class Resource:
    """Resource attributes"""
    resource_id: str
    resource_type: str
    environment: str
    data_class: str
    tenant_id: str = None

@dataclass
class Action:
    """Action to perform"""
    action: str
    resource_type: str

class ABACPolicyEngine:
    def __init__(self):
        self.policies = []

    def add_policy(self, policy: dict):
        """Add ABAC policy"""
        self.policies.append(policy)

    def evaluate(
        self,
        subject: Subject,
        resource: Resource,
        action: Action
    ) -> bool:
        """Evaluate access based on ABAC policies"""
        for policy in self.policies:
            if self._matches_policy(subject, resource, action, policy):
                return policy["effect"] == "allow"

        # Default deny
        return False

    def _matches_policy(
        self,
        subject: Subject,
        resource: Resource,
        action: Action,
        policy: dict
    ) -> bool:
        """Check if policy matches request"""
        # Check subject conditions
        if "subject" in policy:
            if not self._match_conditions(subject, policy["subject"]):
                return False

        # Check resource conditions
        if "resource" in policy:
            if not self._match_conditions(resource, policy["resource"]):
                return False

        # Check action
        if "action" in policy:
            if action.action != policy["action"]:
                return False

        return True

    def _match_conditions(self, obj: Any, conditions: dict) -> bool:
        """Match object attributes against conditions"""
        for key, value in conditions.items():
            obj_value = getattr(obj, key, None)

            if isinstance(value, list):
                if obj_value not in value:
                    return False
            elif obj_value != value:
                return False

        return True

# Example policies
policy_engine = ABACPolicyEngine()

# Policy: Analysts can read non-PII data in dev/stage
policy_engine.add_policy({
    "effect": "allow",
    "subject": {
        "roles": ["analyst"],
        "environment": ["dev", "stage"]
    },
    "resource": {
        "data_class": ["public", "internal"]
    },
    "action": "read"
})

# Policy: ML engineers can write models in their environment
policy_engine.add_policy({
    "effect": "allow",
    "subject": {
        "roles": ["ml-engineer"],
        "environment": ["dev", "stage", "prod"]
    },
    "resource": {
        "resource_type": "model",
        "environment": "${subject.environment}"  # Same environment
    },
    "action": "write"
})

# Policy: Prevent PII data in shared namespaces
policy_engine.add_policy({
    "effect": "deny",
    "resource": {
        "data_class": "pii",
        "tenant_id": None  # Shared namespace
    },
    "action": "write"
})

# k8s/abac-webhook.py
from kubernetes import client, config
from kubernetes.client.rest import ApiException

class ABACAdmissionWebhook:
    def validate(self, admission_request):
        """Validate request based on ABAC policies"""
        # Extract attributes
        user = admission_request.user_info.username
        groups = admission_request.user_info.groups
        namespace = admission_request.namespace
        resource = admission_request.object

        # Get resource attributes
        labels = resource.metadata.labels or {}
        annotations = resource.metadata.annotations or {}

        data_class = labels.get("data_class", "internal")
        environment = labels.get("environment", "dev")
        tenant_id = labels.get("tenant_id")

        # ABAC checks
        # 1. Prevent PII in shared namespaces
        if data_class == "pii" and not tenant_id:
            return {
                "allowed": False,
                "status": {
                    "reason": "PII data requires tenant isolation"
                }
            }

        # 2. Restrict prod deployments
        if environment == "prod" and "platform-admin" not in groups:
            if "access.control/approved" not in annotations:
                return {
                    "allowed": False,
                    "status": {
                        "reason": "Production deployments require approval"
                    }
                }

        # 3. Environment matching
        user_env = self._get_user_environment(groups)
        if environment != user_env and user_env != "all":
            return {
                "allowed": False,
                "status": {
                    "reason": f"User environment {user_env} does not match resource environment {environment}"
                }
            }

        return {"allowed": True}

# jit/jit_access.py
from datetime import datetime, timedelta
import asyncio

class JITAccessManager:
    def __init__(self, db, k8s_client, postgres_client):
        self.db = db
        self.k8s = k8s_client
        self.postgres = postgres_client

    async def request_elevation(
        self,
        user_id: str,
        target_role: str,
        duration_minutes: int = 60,
        reason: str = ""
    ) -> bool:
        """Request JIT elevation with approval workflow"""
        # Check if approval required
        if target_role in ["platform-admin", "cluster-admin"]:
            # Require approval (ticket, chatops, etc.)
            approval = await self._request_approval(user_id, target_role, reason)
            if not approval:
                return False

        # Grant temporary access
        expires_at = datetime.now() + timedelta(minutes=duration_minutes)

        # Store in database
        self.db.execute("""
            INSERT INTO jit_access (user_id, role, expires_at, reason)
            VALUES (%s, %s, %s, %s)
        """, (user_id, target_role, expires_at, reason))

        # Grant in systems
        if target_role == "cluster-admin":
            await self._grant_k8s_cluster_admin(user_id, expires_at)
        elif target_role == "pgadmin":
            await self._grant_postgres_superuser(user_id, expires_at)

        return True

    async def _grant_k8s_cluster_admin(
        self,
        user_id: str,
        expires_at: datetime
    ):
        """Grant temporary cluster-admin in K8s"""
        # Create temporary ClusterRoleBinding
        binding = {
            "apiVersion": "rbac.authorization.k8s.io/v1",
            "kind": "ClusterRoleBinding",
            "metadata": {
                "name": f"jit-{user_id}-{int(expires_at.timestamp())}",
                "annotations": {
                    "expires-at": expires_at.isoformat()
                }
            },
            "roleRef": {
                "apiGroup": "rbac.authorization.k8s.io",
                "kind": "ClusterRole",
                "name": "cluster-admin"
            },
            "subjects": [{
                "kind": "User",
                "name": user_id
            }]
        }

        self.k8s.create_cluster_role_binding(binding)

        # Schedule revocation
        delay = (expires_at - datetime.now()).total_seconds()
        asyncio.create_task(self._revoke_after_delay(user_id, delay))

    async def _grant_postgres_superuser(
        self,
        user_id: str,
        expires_at: datetime
    ):
        """Grant temporary superuser in Postgres"""
        # Grant role
        self.postgres.execute(f"""
            GRANT pgadmin TO {user_id};
            ALTER ROLE {user_id} VALID UNTIL '{expires_at.isoformat()}';
        """)

        # Schedule revocation
        delay = (expires_at - datetime.now()).total_seconds()
        asyncio.create_task(self._revoke_after_delay(user_id, delay, "postgres"))