Secure Polyglot System with Identity Federation and Encryption Lifecycle: A Complete Integration Tutorial¶

"""OIDC identity federation."""
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
import jwt
from jose import JWTError
from fastapi import HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext

from app.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class OIDCProvider:
    """OIDC provider for identity federation."""

    def __init__(self, issuer: str, client_id: str, client_secret: str):
        self.issuer = issuer
        self.client_id = client_id
        self.client_secret = client_secret
        self.jwks_uri = f"{issuer}/.well-known/jwks.json"

    def verify_token(self, token: str) -> Dict[str, Any]:
        """Verify OIDC token."""
        try:
            # In production, fetch JWKS and verify
            payload = jwt.decode(
                token,
                options={"verify_signature": False}  # Simplified for demo
            )
            return payload
        except JWTError as e:
            raise HTTPException(
                status_code=401,
                detail=f"Invalid token: {str(e)}"
            )

    def get_user_info(self, token: str) -> Dict[str, Any]:
        """Get user info from OIDC provider."""
        payload = self.verify_token(token)
        return {
            "sub": payload.get("sub"),
            "email": payload.get("email"),
            "name": payload.get("name"),
            "roles": payload.get("roles", [])
        }


class JWTManager:
    """JWT token management."""

    def __init__(self, secret_key: str, algorithm: str = "HS256"):
        self.secret_key = secret_key
        self.algorithm = algorithm

    def create_token(
        self,
        data: Dict[str, Any],
        expires_delta: Optional[timedelta] = None
    ) -> str:
        """Create JWT token."""
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(hours=24)

        to_encode.update({
            "exp": expire,
            "iat": datetime.utcnow(),
            "iss": settings.JWT_ISSUER
        })

        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)

    def verify_token(self, token: str) -> Dict[str, Any]:
        """Verify JWT token."""
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],
                options={"verify_exp": True}
            )
            return payload
        except JWTError as e:
            raise HTTPException(
                status_code=401,
                detail=f"Invalid token: {str(e)}"
            )


async def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
    """Get current user from token."""
    oidc = OIDCProvider(
        issuer=settings.OIDC_ISSUER,
        client_id=settings.OIDC_CLIENT_ID,
        client_secret=settings.OIDC_CLIENT_SECRET
    )
    return oidc.get_user_info(token)

"""RBAC and ABAC authorization."""
from typing import List, Dict, Any, Optional
from enum import Enum
from dataclasses import dataclass

from app.oidc import get_current_user
from fastapi import Depends, HTTPException


class Permission(Enum):
    """Permissions."""
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    ADMIN = "admin"


@dataclass
class Role:
    """Role definition."""
    name: str
    permissions: List[Permission]
    resources: List[str]


class RBACEngine:
    """RBAC authorization engine."""

    def __init__(self):
        self.roles = {
            "admin": Role(
                name="admin",
                permissions=[Permission.READ, Permission.WRITE, Permission.DELETE, Permission.ADMIN],
                resources=["*"]
            ),
            "user": Role(
                name="user",
                permissions=[Permission.READ, Permission.WRITE],
                resources=["users", "data"]
            ),
            "viewer": Role(
                name="viewer",
                permissions=[Permission.READ],
                resources=["users", "data"]
            )
        }

    def has_permission(
        self,
        user_roles: List[str],
        permission: Permission,
        resource: str
    ) -> bool:
        """Check if user has permission."""
        for role_name in user_roles:
            if role_name in self.roles:
                role = self.roles[role_name]
                if permission in role.permissions:
                    if "*" in role.resources or resource in role.resources:
                        return True
        return False


class ABACEngine:
    """ABAC authorization engine."""

    def evaluate_policy(
        self,
        subject: Dict[str, Any],
        resource: Dict[str, Any],
        action: str,
        environment: Dict[str, Any]
    ) -> bool:
        """Evaluate ABAC policy."""
        # Example policy: user can access their own data or admin can access all
        if subject.get("roles", []).__contains__("admin"):
            return True

        if action == "read" and subject.get("sub") == resource.get("owner"):
            return True

        if action == "write" and subject.get("sub") == resource.get("owner"):
            return True

        return False


def require_permission(permission: Permission, resource: str):
    """Decorator for permission checking."""
    def decorator(func):
        async def wrapper(
            *args,
            current_user: Dict[str, Any] = Depends(get_current_user),
            **kwargs
        ):
            rbac = RBACEngine()
            user_roles = current_user.get("roles", [])

            if not rbac.has_permission(user_roles, permission, resource):
                raise HTTPException(
                    status_code=403,
                    detail=f"Permission denied: {permission.value} on {resource}"
                )

            return await func(*args, current_user=current_user, **kwargs)
        return wrapper
    return decorator

"""Secrets management with Vault."""
import hvac
from typing import Optional, Dict, Any
import os

from observability.metrics import secrets_metrics


class VaultClient:
    """Hashicorp Vault client."""

    def __init__(self, vault_url: str, vault_token: str):
        self.client = hvac.Client(url=vault_url, token=vault_token)
        self.client.is_authenticated()

    def get_secret(self, path: str, key: Optional[str] = None) -> Any:
        """Get secret from Vault."""
        try:
            response = self.client.secrets.kv.v2.read_secret_version(path=path)
            data = response["data"]["data"]

            if key:
                return data.get(key)
            return data
        except Exception as e:
            secrets_metrics.secret_access_failures.labels(
                path=path,
                error_type=type(e).__name__
            ).inc()
            raise

    def rotate_secret(self, path: str, new_secret: str) -> bool:
        """Rotate secret in Vault."""
        try:
            self.client.secrets.kv.v2.create_or_update_secret(
                path=path,
                secret={"value": new_secret}
            )
            secrets_metrics.secret_rotations.labels(path=path).inc()
            return True
        except Exception as e:
            secrets_metrics.secret_rotation_failures.labels(
                path=path,
                error_type=type(e).__name__
            ).inc()
            raise


# Global Vault client
vault_client = VaultClient(
    vault_url=os.getenv("VAULT_ADDR", "http://vault:8200"),
    vault_token=os.getenv("VAULT_TOKEN", "")
)