FastAPI Auth: Authorization Patterns — RBAC, ABAC, and Beyond

from __future__ import annotations
from uuid import UUID, uuid4

from sqlalchemy import ForeignKey, String, Table, Column
from sqlalchemy.orm import Mapped, mapped_column, relationship

from src.db.base import Base

# Association tables
user_roles = Table(
    "user_roles", Base.metadata,
    Column("user_id", ForeignKey("users.id"), primary_key=True),
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
)

role_permissions = Table(
    "role_permissions", Base.metadata,
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
    Column("permission_id", ForeignKey("permissions.id"), primary_key=True),
)


class Role(Base):
    __tablename__ = "roles"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    name: Mapped[str] = mapped_column(String(50), unique=True)  # "admin", "editor", "viewer"
    description: Mapped[str] = mapped_column(String(200))
    permissions: Mapped[list["Permission"]] = relationship(secondary=role_permissions)


class Permission(Base):
    __tablename__ = "permissions"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    resource: Mapped[str] = mapped_column(String(50))  # "documents", "users"
    action: Mapped[str] = mapped_column(String(50))     # "create", "read", "update", "delete"

    @property
    def key(self) -> str:
        return f"{self.resource}:{self.action}"

from functools import wraps
from fastapi import Depends, HTTPException, status


async def get_user_permissions(
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> set[str]:
    """Load all permissions for the current user across all their roles."""
    result = await db.execute(
        select(Permission)
        .join(role_permissions)
        .join(Role)
        .join(user_roles)
        .where(user_roles.c.user_id == user.id)
    )
    permissions = result.scalars().all()
    return {p.key for p in permissions}


def require_permission(resource: str, action: str):
    """Dependency factory: check if the user has a specific permission."""
    required = f"{resource}:{action}"

    async def checker(
        permissions: set[str] = Depends(get_user_permissions),
    ):
        if required not in permissions:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing permission: {required}",
            )

    return checker


def require_any_permission(*perms: str):
    """Dependency factory: check if the user has ANY of the listed permissions."""

    async def checker(
        permissions: set[str] = Depends(get_user_permissions),
    ):
        if not permissions.intersection(perms):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing one of: {', '.join(perms)}",
            )

    return checker

from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import Any


class Effect(Enum):
    ALLOW = "allow"
    DENY = "deny"


@dataclass
class PolicyContext:
    subject: dict[str, Any]     # User attributes
    resource: dict[str, Any]    # Resource attributes
    action: str                 # The operation
    environment: dict[str, Any] # Contextual attributes


@dataclass
class Policy:
    name: str
    effect: Effect
    condition: callable  # (PolicyContext) -> bool

    def evaluate(self, ctx: PolicyContext) -> Effect | None:
        """Returns the effect if the condition matches, None otherwise."""
        if self.condition(ctx):
            return self.effect
        return None


# Define policies as code
policies: list[Policy] = [
    # Admins can do anything
    Policy(
        name="admin_full_access",
        effect=Effect.ALLOW,
        condition=lambda ctx: "admin" in ctx.subject.get("roles", []),
    ),

    # Editors can update documents during business hours
    Policy(
        name="editor_business_hours",
        effect=Effect.ALLOW,
        condition=lambda ctx: (
            "editor" in ctx.subject.get("roles", [])
            and ctx.action in ("read", "update")
            and 9 <= ctx.environment.get("hour", 0) <= 17
        ),
    ),

    # Users can only delete their own documents
    Policy(
        name="owner_delete",
        effect=Effect.ALLOW,
        condition=lambda ctx: (
            ctx.action == "delete"
            and ctx.resource.get("owner_id") == ctx.subject.get("user_id")
        ),
    ),

    # Deny access to classified documents unless clearance matches
    Policy(
        name="classification_check",
        effect=Effect.DENY,
        condition=lambda ctx: (
            ctx.resource.get("classification") == "confidential"
            and ctx.subject.get("clearance") != "confidential"
        ),
    ),
]


def evaluate_policies(ctx: PolicyContext) -> bool:
    """Evaluate all policies. Deny takes precedence over Allow."""
    result = Effect.DENY  # Default deny

    for policy in policies:
        effect = policy.evaluate(ctx)
        if effect == Effect.DENY:
            return False  # Explicit deny — short circuit
        if effect == Effect.ALLOW:
            result = Effect.ALLOW

    return result == Effect.ALLOW

from src.core.abac import PolicyContext, evaluate_policies


def require_abac(action: str):
    """ABAC dependency factory — evaluates policies against request context."""

    async def checker(
        request: Request,
        user: User = Depends(get_current_user),
        resource: dict | None = None,
    ):
        ctx = PolicyContext(
            subject={
                "user_id": str(user.id),
                "roles": user.role_names,
                "department": user.department,
                "clearance": user.clearance_level,
            },
            resource=resource or {},
            action=action,
            environment={
                "hour": datetime.now(timezone.utc).hour,
                "ip": request.client.host,
                "method": request.method,
            },
        )

        if not evaluate_policies(ctx):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Access denied by policy for action: {action}",
            )

    return checker


# Usage — with resource attributes loaded from DB
@router.put("/documents/{doc_id}")
async def update_document(
    doc_id: UUID,
    body: UpdateDocumentRequest,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    document = await db.get(Document, doc_id)
    if not document:
        raise HTTPException(status_code=404)

    # Build resource context and evaluate
    checker = require_abac("update")
    await checker(
        request=...,
        user=user,
        resource={
            "owner_id": str(document.owner_id),
            "classification": document.classification,
        },
    )

    # Proceed with update
    ...

class ResourceRelation(Base):
    __tablename__ = "resource_relations"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    subject_type: Mapped[str] = mapped_column(String(50))   # "user", "team"
    subject_id: Mapped[UUID] = mapped_column()
    relation: Mapped[str] = mapped_column(String(50))        # "owner", "editor", "viewer"
    resource_type: Mapped[str] = mapped_column(String(50))   # "document", "folder"
    resource_id: Mapped[UUID] = mapped_column()

    __table_args__ = (
        UniqueConstraint(
            "subject_type", "subject_id", "relation",
            "resource_type", "resource_id",
            name="uq_relation",
        ),
        Index("ix_resource_lookup", "resource_type", "resource_id"),
    )

# Relation hierarchy — owner inherits editor, editor inherits viewer
RELATION_HIERARCHY: dict[str, set[str]] = {
    "owner": {"owner", "editor", "viewer"},
    "editor": {"editor", "viewer"},
    "viewer": {"viewer"},
}


async def check_relation(
    db: AsyncSession,
    user_id: UUID,
    resource_type: str,
    resource_id: UUID,
    required_relation: str,
) -> bool:
    """Check if user has the required (or higher) relation to a resource."""
    # Get all relations for this user and resource
    result = await db.execute(
        select(ResourceRelation.relation).where(
            ResourceRelation.subject_type == "user",
            ResourceRelation.subject_id == user_id,
            ResourceRelation.resource_type == resource_type,
            ResourceRelation.resource_id == resource_id,
        )
    )
    user_relations = {r for (r,) in result.all()}

    # Check via team memberships too
    team_ids = await get_user_team_ids(db, user_id)
    if team_ids:
        result = await db.execute(
            select(ResourceRelation.relation).where(
                ResourceRelation.subject_type == "team",
                ResourceRelation.subject_id.in_(team_ids),
                ResourceRelation.resource_type == resource_type,
                ResourceRelation.resource_id == resource_id,
            )
        )
        user_relations.update(r for (r,) in result.all())

    # Check if any relation grants the required access
    for rel in user_relations:
        if required_relation in RELATION_HIERARCHY.get(rel, set()):
            return True

    return False