From 2e863c8b432c0e8061ae5eddfa98140f73d0779c Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Sun, 21 Jun 2026 09:06:22 +0800 Subject: [PATCH] refactor(auth): replace legacy middleware sentinel --- apps/api/app/middleware/auth.py | 160 +++++------------------------ tests/unit/test_auth_middleware.py | 94 +++++++++++++++++ 2 files changed, 121 insertions(+), 133 deletions(-) create mode 100644 tests/unit/test_auth_middleware.py diff --git a/apps/api/app/middleware/auth.py b/apps/api/app/middleware/auth.py index d4fd33fcf..b5005c9da 100644 --- a/apps/api/app/middleware/auth.py +++ b/apps/api/app/middleware/auth.py @@ -1,166 +1,60 @@ """ -Legacy auth middleware. +Authentication dependency compatibility layer. -Disabled because it depends on the removed DI container. Rebuild it around the -canonical JWT settings and SQLAlchemy-backed user repository before reuse. +Canonical bearer-token parsing lives in app.auth. This module remains only so +legacy imports have a safe target while workspace dependencies are rebuilt. """ -raise RuntimeError("apps.api.app.middleware.auth is disabled: rebuild auth dependency wiring before importing it") +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 from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer -from apps.api.app.dependencies import get_container -from packages.domain.auth import jwt_service from packages.domain.entities import User +from packages.ports.user_repository import UserRepository -security = HTTPBearer() +optional_bearer_scheme = HTTPBearer(auto_error=False) async def get_current_user( - credentials: HTTPAuthorizationCredentials = Depends(security), + authenticated_user: AuthenticatedUser = Depends(get_authenticated_user), ) -> User: - """ - 获取当前登录用户 - - 从 Authorization header 中提取 JWT token 并验证 - - Raises: - HTTPException: Token 无效或过期 - - Returns: - 当前用户对象 - """ - token = credentials.credentials - - try: - # 验证 token - payload = jwt_service.verify_token(token) - user_id = payload.get("sub") - - if not user_id: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid token: missing user_id", - headers={"WWW-Authenticate": "Bearer"}, - ) - - # 从数据库获取用户 - container = get_container() - user = container.user_repository.find_by_id(user_id) - - if not user: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="User not found", - headers={"WWW-Authenticate": "Bearer"}, - ) - - return user - - except Exception as e: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail=f"Invalid token: {str(e)}", - headers={"WWW-Authenticate": "Bearer"}, - ) + return authenticated_user.user async def get_current_user_optional( - credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)), + credentials: HTTPAuthorizationCredentials | None = Depends(optional_bearer_scheme), + user_repository: UserRepository = Depends(get_user_repository), ) -> User | None: - """ - 获取当前登录用户（可选） - - 如果没有提供 token，返回 None 而不是抛出异常 - - Returns: - 当前用户对象或 None - """ - if not credentials: + if credentials is None: return None - try: - return await get_current_user(credentials) + 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)) -> tuple[str, str]: - """ - 要求用户可以访问指定工作空间 - - Args: - workspace_id: 工作空间 ID - user: 当前用户 - - Raises: - HTTPException: 用户没有访问权限 - - Returns: - (workspace_id, user_role) - """ - container = get_container() - permission_checker = container.permission_checker - - has_access, role = permission_checker.check_workspace_access(workspace_id, user.id) - - if not has_access: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="You don't have access to this workspace", - ) - - return workspace_id, role + raise _workspace_dependency_not_ready(workspace_id, user.id) def require_workspace_admin(workspace_id: str, user: User = Depends(get_current_user)) -> str: - """ - 要求用户是工作空间的 Admin 或 Owner - - Args: - workspace_id: 工作空间 ID - user: 当前用户 - - Raises: - HTTPException: 用户没有管理权限 - - Returns: - workspace_id - """ - container = get_container() - permission_checker = container.permission_checker - - if not permission_checker.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 + raise _workspace_dependency_not_ready(workspace_id, user.id) def require_workspace_owner(workspace_id: str, user: User = Depends(get_current_user)) -> str: - """ - 要求用户是工作空间的 Owner + raise _workspace_dependency_not_ready(workspace_id, user.id) - Args: - workspace_id: 工作空间 ID - user: 当前用户 - Raises: - HTTPException: 用户不是 Owner - - Returns: - workspace_id - """ - container = get_container() - permission_checker = container.permission_checker - - if not permission_checker.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 +def _workspace_dependency_not_ready(workspace_id: str, user_id: str) -> HTTPException: + return HTTPException( + status_code=status.HTTP_501_NOT_IMPLEMENTED, + detail=( + "Workspace authorization dependencies require SQLAlchemy workspace-member " + f"repository wiring before use: workspace_id={workspace_id}, user_id={user_id}" + ), + ) diff --git a/tests/unit/test_auth_middleware.py b/tests/unit/test_auth_middleware.py new file mode 100644 index 000000000..ef93634a8 --- /dev/null +++ b/tests/unit/test_auth_middleware.py @@ -0,0 +1,94 @@ +import asyncio +import sys +from pathlib import Path + +import jwt +import pytest +from fastapi import HTTPException +from fastapi.security import HTTPAuthorizationCredentials + +ROOT = Path(__file__).resolve().parents[2] +API_ROOT = ROOT / "apps" / "api" +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + +from app.config import settings +from app.middleware.auth import get_current_user, get_current_user_optional, require_workspace_access + +from packages.domain.auth import password_hasher +from packages.domain.entities import User + + +class InMemoryUserRepository: + def __init__(self): + self.users = {} + + def save(self, user): + self.users[user.id] = user + + def find_by_id(self, user_id): + return self.users.get(user_id) + + def find_by_email(self, email): + return next((user for user in self.users.values() if user.email == email), None) + + def find_by_username(self, username): + return next((user for user in self.users.values() if user.username == username), None) + + def find_by_verification_token(self, token): + return None + + def find_by_password_reset_token(self, token): + return None + + def delete(self, user_id): + return self.users.pop(user_id, None) is not None + + +def _repo_with_user(): + repo = InMemoryUserRepository() + user = User( + id="user-1", + email="user@example.com", + username="user", + display_name="User", + password_hash=password_hasher.hash_password("Password1"), + ) + repo.save(user) + return repo + + +def _credentials(token_type="user_auth"): + token = jwt.encode({"sub": "user-1", "sid": "session-1", "type": token_type}, settings.JWT_SECRET_KEY, "HS256") + return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) + + +def test_legacy_middleware_get_current_user_delegates_to_canonical_auth(): + user = asyncio.run(get_current_user(asyncio.run(_authenticated_user()))) + + assert user.id == "user-1" + + +def test_legacy_middleware_optional_user_returns_none_without_credentials(): + assert asyncio.run(get_current_user_optional(None, _repo_with_user())) is None + + +def test_legacy_middleware_optional_user_returns_user_with_valid_credentials(): + user = asyncio.run(get_current_user_optional(_credentials(), _repo_with_user())) + + assert user is not None + assert user.id == "user-1" + + +def test_workspace_dependency_fails_closed_until_repository_is_wired(): + with pytest.raises(HTTPException) as error: + require_workspace_access("workspace-1", _repo_with_user().find_by_id("user-1")) + + assert error.value.status_code == 501 + assert "SQLAlchemy workspace-member repository" in error.value.detail + + +async def _authenticated_user(): + from app.auth import get_current_user as get_authenticated_user + + return await get_authenticated_user(_credentials(), _repo_with_user())