Xiaoxia AI
77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
"""
|
|
Authentication dependency compatibility layer.
|
|
|
|
Canonical bearer-token parsing lives in app.auth. This module remains only so
|
|
legacy imports have a safe target while workspace dependencies are rebuilt.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.auth import AuthenticatedUser
|
|
from app.auth import get_current_user as get_authenticated_user
|
|
from app.dependencies import get_user_repository, get_workspace_member_repository
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from packages.domain.entities import User
|
|
from packages.domain.permissions import PermissionChecker
|
|
from packages.ports.user_repository import UserRepository
|
|
from packages.ports.workspace_member_repository import WorkspaceMemberRepository
|
|
|
|
optional_bearer_scheme = HTTPBearer(auto_error=False)
|
|
|
|
|
|
async def get_current_user(
|
|
authenticated_user: AuthenticatedUser = Depends(get_authenticated_user),
|
|
) -> User:
|
|
return authenticated_user.user
|
|
|
|
|
|
async def get_current_user_optional(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(optional_bearer_scheme),
|
|
user_repository: UserRepository = Depends(get_user_repository),
|
|
) -> User | None:
|
|
if credentials is None:
|
|
return None
|
|
try:
|
|
authenticated_user = await get_authenticated_user(credentials, user_repository)
|
|
except HTTPException:
|
|
return None
|
|
return authenticated_user.user
|
|
|
|
|
|
def require_workspace_access(
|
|
workspace_id: str,
|
|
user: User = Depends(get_current_user),
|
|
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
|
|
) -> tuple[str, str]:
|
|
has_access, role = PermissionChecker(workspace_member_repository).check_workspace_access(workspace_id, user.id)
|
|
if not has_access or role is None:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You don't have access to this workspace")
|
|
return workspace_id, role
|
|
|
|
|
|
def require_workspace_admin(
|
|
workspace_id: str,
|
|
user: User = Depends(get_current_user),
|
|
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
|
|
) -> str:
|
|
if not PermissionChecker(workspace_member_repository).check_is_admin_or_owner(workspace_id, user.id):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only workspace owner or admin can perform this action",
|
|
)
|
|
return workspace_id
|
|
|
|
|
|
def require_workspace_owner(
|
|
workspace_id: str,
|
|
user: User = Depends(get_current_user),
|
|
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
|
|
) -> str:
|
|
if not PermissionChecker(workspace_member_repository).check_is_owner(workspace_id, user.id):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Only workspace owner can perform this action"
|
|
)
|
|
return workspace_id
|