Permissions and Authorization¶

from typing import Annotated, Any

from fastapi import APIRouter, Depends

from ...infrastructure.auth.session.dependencies import get_current_superuser

router = APIRouter()


@router.delete("/admin/users/{username}")
async def gdpr_anonymize(
    username: str,
    _: Annotated[dict[str, Any], Depends(get_current_superuser)],
) -> dict[str, str]:
    # Only superusers reach this code
    ...

from ..common.exceptions import PermissionDeniedError


async def verify_user_permission(
    self,
    current_user: dict[str, Any],
    target_username: str,
    action: str,
) -> None:
    """Raise PermissionDeniedError if current_user can't act on target_username."""
    if current_user["username"] != target_username and not current_user["is_superuser"]:
        raise PermissionDeniedError(f"Cannot {action} for another user")

# modules/user/routes.py
@router.patch("/{username}")
async def update_user_profile(
    username: str,
    values: UserUpdate,
    current_user: Annotated[dict[str, Any], Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(async_session)],
    user_service: Annotated[UserService, Depends(get_user_service)],
) -> dict[str, str]:
    try:
        await user_service.verify_user_permission(current_user, username, "update profile")
        # ...proceed with update...

# modules/widgets/service.py
from ..common.exceptions import PermissionDeniedError, ResourceNotFoundError


class WidgetService:
    async def delete(
        self, widget_id: int, current_user: dict[str, Any], db: AsyncSession,
    ) -> None:
        widget = await crud_widgets.get(db=db, id=widget_id)
        if widget is None:
            raise ResourceNotFoundError("Widget not found")

        if widget["owner_id"] != current_user["id"] and not current_user["is_superuser"]:
            raise PermissionDeniedError("Cannot delete another user's widget")

        await crud_widgets.delete(db=db, id=widget_id)

@router.get("/me", response_model=UserRead)
async def me(
    current_user: Annotated[dict[str, Any], Depends(get_current_user)],
) -> dict[str, Any]:
    # current_user["tier"] is the joined Tier dict (or None)
    return current_user

async def export_data(self, current_user: dict[str, Any], db: AsyncSession) -> bytes:
    tier = current_user.get("tier") or {}
    if tier.get("name") not in {"pro", "enterprise"}:
        raise PermissionDeniedError("Data export requires the Pro or Enterprise tier")
    # ...generate export...

# modules/api_keys/models.py
class KeyPermission(Base, TimestampMixin):
    __tablename__ = "key_permissions"

    api_key_id: Mapped[int] = mapped_column(ForeignKey("api_keys.id", ondelete="CASCADE"))
    resource: Mapped[KeyPermissionResource] = mapped_column(index=True)
    action: Mapped[KeyPermissionAction] = mapped_column(index=True)
    conditions: Mapped[dict[str, Any] | None] = mapped_column(JSON, default=None)
    is_allowed: Mapped[bool] = mapped_column(Boolean, default=True)

class KeyPermissionResource(StrEnum):
    USER_PROFILE = "user_profile"
    ANALYTICS = "analytics"
    ADMIN = "admin"
    BILLING = "billing"
    API_KEYS = "api_keys"
    WILDCARD = "*"
    # ... plus a few legacy values inherited from the upstream template


class KeyPermissionAction(StrEnum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    CREATE = "create"
    UPDATE = "update"
    LIST = "list"
    ADMIN = "admin"
    WILDCARD = "*"

async def require_key_permission(
    resource: KeyPermissionResource,
    action: KeyPermissionAction,
    db: AsyncSession,
    api_key: dict[str, Any],
) -> None:
    has_permission = await crud_key_permissions.exists(
        db=db,
        api_key_id=api_key["id"],
        resource=resource,
        action=action,
        is_allowed=True,
    )
    # also check wildcards
    if not has_permission:
        has_wildcard = await crud_key_permissions.exists(
            db=db,
            api_key_id=api_key["id"],
            resource=KeyPermissionResource.WILDCARD,
            action=KeyPermissionAction.WILDCARD,
            is_allowed=True,
        )
        if not has_wildcard:
            raise PermissionDeniedError(f"API key lacks {resource}:{action}")

@router.delete("/widgets/{widget_id}", status_code=204)
async def delete_widget(
    widget_id: int,
    current_user: Annotated[dict[str, Any], Depends(get_current_user)],   # 1. authn
    db: Annotated[AsyncSession, Depends(async_session)],
    widget_service: Annotated[WidgetService, Depends(get_widget_service)],
) -> None:
    try:
        # Service handles:
        #   2. Existence check
        #   3. Ownership check (superuser bypass)
        #   4. Tier feature gate (e.g. "delete requires Pro tier")
        await widget_service.delete(widget_id, current_user, db)
    except Exception as e:
        http_exc = handle_exception(e)
        if http_exc:
            raise http_exc
        raise HTTPException(status_code=500, detail="An unexpected error occurred")

import pytest
from src.modules.user.service import UserService
from src.modules.common.exceptions import PermissionDeniedError


@pytest.mark.asyncio
async def test_normal_user_cannot_update_other_users():
    service = UserService()
    current_user = {"username": "alice", "is_superuser": False}

    with pytest.raises(PermissionDeniedError):
        await service.verify_user_permission(current_user, "bob", "update profile")


@pytest.mark.asyncio
async def test_superuser_can_update_other_users():
    service = UserService()
    current_user = {"username": "alice", "is_superuser": True}

    # Should not raise
    await service.verify_user_permission(current_user, "bob", "update profile")

# 1. Authenticated? — done by the dependency
# 2. Resource exists?
if widget is None:
    raise ResourceNotFoundError(...)
# 3. User owns it?
if widget["owner_id"] != current_user["id"] and not current_user["is_superuser"]:
    raise PermissionDeniedError(...)
# 4. Quota / tier OK?
if not within_tier_limits(...):
    raise PermissionDeniedError(...)