diff --git a/backend/src/module/api/auth.py b/backend/src/module/api/auth.py index 22f899dd..f445b4dc 100644 --- a/backend/src/module/api/auth.py +++ b/backend/src/module/api/auth.py @@ -22,13 +22,24 @@ router = APIRouter(prefix="/auth", tags=["auth"]) _TOKEN_EXPIRY_DAYS = 1 _TOKEN_MAX_AGE = 86400 +# Pseudo-username returned by get_current_user() for callers authenticated via +# a bearer API token (settings.security.login_tokens) rather than a cookie +# session. Such callers have no real session to refresh/logout/update. +_API_TOKEN_USER = "api_token_user" + def _issue_token(username: str, response: Response) -> dict: """Create a JWT, set it as an HttpOnly cookie, and return the bearer payload.""" token = create_access_token( data={"sub": username}, expires_delta=timedelta(days=_TOKEN_EXPIRY_DAYS) ) - response.set_cookie(key="token", value=token, httponly=True, max_age=_TOKEN_MAX_AGE) + response.set_cookie( + key="token", + value=token, + httponly=True, + max_age=_TOKEN_MAX_AGE, + samesite="strict", + ) return {"access_token": token, "token_type": "bearer"} @@ -42,11 +53,18 @@ async def login(response: Response, form_data=Depends(OAuth2PasswordRequestForm) return u_response(resp) -@router.get( - "/refresh_token", response_model=dict, dependencies=[Depends(get_current_user)] -) -async def refresh(response: Response, token: str = Cookie(None)): +@router.get("/refresh_token", response_model=dict) +async def refresh( + response: Response, + token: str = Cookie(None), + current_user: str = Depends(get_current_user), +): """Refresh the current session token and update the active-user timestamp.""" + if current_user == _API_TOKEN_USER: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="API token authentication has no cookie session to refresh.", + ) payload = decode_token(token) username = payload.get("sub") if payload else None if not username: @@ -57,15 +75,18 @@ async def refresh(response: Response, token: str = Cookie(None)): return _issue_token(username, response) -@router.get( - "/logout", response_model=APIResponse, dependencies=[Depends(get_current_user)] -) -async def logout(response: Response, token: str = Cookie(None)): +@router.post("/logout", response_model=APIResponse) +async def logout( + response: Response, + token: str = Cookie(None), + current_user: str = Depends(get_current_user), +): """Invalidate the session and clear the token cookie.""" - payload = decode_token(token) - username = payload.get("sub") if payload else None - if username: - active_user.remove(username) + if current_user != _API_TOKEN_USER: + payload = decode_token(token) + username = payload.get("sub") if payload else None + if username: + active_user.remove(username) response.delete_cookie(key="token") return JSONResponse( status_code=200, @@ -73,11 +94,19 @@ async def logout(response: Response, token: str = Cookie(None)): ) -@router.post("/update", response_model=dict, dependencies=[Depends(get_current_user)]) +@router.post("/update", response_model=dict) async def update_user( - user_data: UserUpdate, response: Response, token: str = Cookie(None) + user_data: UserUpdate, + response: Response, + token: str = Cookie(None), + current_user: str = Depends(get_current_user), ): """Update credentials for the current user and re-issue a fresh token.""" + if current_user == _API_TOKEN_USER: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="API token authentication cannot update user credentials.", + ) payload = decode_token(token) old_user = payload.get("sub") if payload else None if not old_user: @@ -85,4 +114,12 @@ async def update_user( status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized" ) if await update_user_info(user_data, old_user): - return {**_issue_token(old_user, response), "message": "update success"} + # Re-issue the token under the *new* username (if it changed) and + # migrate the active-user session entry, so the old username isn't + # left stranded (unable to log out) and the new one isn't rejected + # as "not in active_user" on the very next request. + new_username = user_data.username or old_user + if new_username != old_user: + active_user.remove(old_user) + active_user.add(new_username) + return {**_issue_token(new_username, response), "message": "update success"} diff --git a/backend/src/module/api/passkey.py b/backend/src/module/api/passkey.py index b91fedf2..4980039f 100644 --- a/backend/src/module/api/passkey.py +++ b/backend/src/module/api/passkey.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import JSONResponse, Response from sqlmodel import select +from module.conf import settings from module.database.engine import async_session_factory from module.database.passkey import PasskeyDatabase from module.models import APIResponse @@ -21,7 +22,7 @@ from module.models.passkey import ( PasskeyList, ) from module.models.user import User -from module.security.api import active_user, get_current_user +from module.security.api import active_user, check_login_ip, get_current_user from module.security.auth_strategy import PasskeyAuthStrategy from module.security.jwt import create_access_token from module.security.webauthn import get_webauthn_service @@ -29,29 +30,37 @@ from module.security.webauthn import get_webauthn_service logger = logging.getLogger(__name__) router = APIRouter(prefix="/passkey", tags=["passkey"]) +_GENERIC_ERROR = "Internal server error." + def _get_webauthn_from_request(request: Request): """ 从请求中构造 WebAuthnService - 优先使用浏览器的 Origin header(与 clientDataJSON 中的 origin 一致) + + 若配置了 settings.security.webauthn_rp_id / webauthn_origin(非空), + 直接使用它们;否则回退到从请求头推断(Origin -> Referer -> Host), + 这些请求头在反向代理场景下可能被客户端伪造,因此建议生产环境显式配置。 """ from urllib.parse import urlparse - origin = request.headers.get("origin") + origin = settings.security.webauthn_origin if not origin: - # Fallback: 从 Referer 或 Host 推断 - referer = request.headers.get("referer", "") - if referer: - parsed = urlparse(referer) - origin = f"{parsed.scheme}://{parsed.netloc}" - else: - host = request.headers.get("host", "localhost:7892") - forwarded_proto = request.headers.get("x-forwarded-proto") - scheme = forwarded_proto if forwarded_proto else request.url.scheme - origin = f"{scheme}://{host}" + origin = request.headers.get("origin") + if not origin: + # Fallback: 从 Referer 或 Host 推断 + referer = request.headers.get("referer", "") + if referer: + parsed = urlparse(referer) + origin = f"{parsed.scheme}://{parsed.netloc}" + else: + host = request.headers.get("host", "localhost:7892") + forwarded_proto = request.headers.get("x-forwarded-proto") + scheme = forwarded_proto if forwarded_proto else request.url.scheme + origin = f"{scheme}://{host}" - parsed_origin = urlparse(origin) - rp_id = parsed_origin.hostname or "localhost" + rp_id = settings.security.webauthn_rp_id + if not rp_id: + rp_id = urlparse(origin).hostname or "localhost" return get_webauthn_service(rp_id, "AutoBangumi", origin) @@ -96,7 +105,7 @@ async def get_registration_options( raise except Exception as e: logger.error(f"Failed to generate registration options: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=_GENERIC_ERROR) @router.post("/register/verify", response_model=APIResponse) @@ -147,13 +156,17 @@ async def verify_registration( raise except Exception as e: logger.error(f"Failed to register passkey: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=_GENERIC_ERROR) # ============ 认证流程 ============ -@router.post("/auth/options", response_model=dict) +@router.post( + "/auth/options", + response_model=dict, + dependencies=[Depends(check_login_ip)], +) async def get_passkey_login_options( auth_data: PasskeyAuthStart, request: Request, @@ -174,7 +187,7 @@ async def get_passkey_login_options( return options except Exception as e: logger.error(f"Failed to generate discoverable login options: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=_GENERIC_ERROR) # Username-based mode async with async_session_factory() as session: @@ -184,15 +197,19 @@ async def get_passkey_login_options( select(User).where(User.username == auth_data.username) ) user = result.scalar_one_or_none() - if not user: - raise HTTPException(status_code=404, detail="User not found") - passkey_db = PasskeyDatabase(session) - passkeys = await passkey_db.get_passkeys_by_user_id(user.id) + passkeys = [] + if user: + passkey_db = PasskeyDatabase(session) + passkeys = await passkey_db.get_passkeys_by_user_id(user.id) - if not passkeys: + if not user or not passkeys: + # Same response whether the username doesn't exist or simply + # has no passkeys registered, so this endpoint can't be used + # to enumerate valid usernames. raise HTTPException( - status_code=400, detail="No passkeys registered for this user" + status_code=400, + detail="No passkeys available for this username.", ) options = webauthn.generate_authentication_options( @@ -204,10 +221,14 @@ async def get_passkey_login_options( raise except Exception as e: logger.error(f"Failed to generate login options: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=_GENERIC_ERROR) -@router.post("/auth/verify", response_model=dict) +@router.post( + "/auth/verify", + response_model=dict, + dependencies=[Depends(check_login_ip)], +) async def login_with_passkey( auth_data: PasskeyAuthFinish, response: Response, @@ -233,7 +254,13 @@ async def login_with_passkey( token = create_access_token( data={"sub": username}, expires_delta=timedelta(days=1) ) - response.set_cookie(key="token", value=token, httponly=True, max_age=86400) + response.set_cookie( + key="token", + value=token, + httponly=True, + max_age=86400, + samesite="strict", + ) active_user.add(username) return {"access_token": token, "token_type": "bearer"} @@ -265,7 +292,7 @@ async def list_passkeys(username: str = Depends(get_current_user)): raise except Exception as e: logger.error(f"Failed to list passkeys: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=_GENERIC_ERROR) @router.post("/delete", response_model=APIResponse) @@ -299,4 +326,4 @@ async def delete_passkey( raise except Exception as e: logger.error(f"Failed to delete passkey: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=_GENERIC_ERROR) diff --git a/backend/src/module/api/setup.py b/backend/src/module/api/setup.py index f6a66805..9818b775 100644 --- a/backend/src/module/api/setup.py +++ b/backend/src/module/api/setup.py @@ -5,7 +5,7 @@ from pathlib import Path from urllib.parse import urlparse import httpx -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel, Field from module.conf import VERSION, settings @@ -13,7 +13,7 @@ from module.models import Config, ResponseModel from module.models.config import NotificationProvider as ProviderConfig from module.network import RequestContent from module.notification import PROVIDER_REGISTRY -from module.security.jwt import get_password_hash +from module.security.jwt import get_password_hash, verify_password logger = logging.getLogger(__name__) @@ -31,6 +31,39 @@ def _require_setup_needed(): raise HTTPException(status_code=403, detail="Setup already completed.") +async def _require_default_admin_or_authenticated(request: Request) -> None: + """Extra guard for /setup/complete specifically. + + ``_require_setup_needed()`` only compares the on-disk config to its + defaults, which stays true on an upgraded install that never ran the + wizard even after the real admin already changed their password some + other way (e.g. the old settings UI). Without this check, an + unauthenticated caller could hit /setup/complete on such an install and + overwrite the admin's real credentials. Allow completion only if the + caller already has a valid session, or the "admin" account still has the + untouched factory-default password. + """ + from module.security.api import get_current_user + + try: + await get_current_user(request, token=request.cookies.get("token")) + return + except HTTPException: + pass + + from module.database import Database + + async with Database() as db: + try: + user = await db.user.get_user("admin") + except HTTPException: + # No admin account yet (fresh install) — setup may proceed. + return + + if not verify_password("adminadmin", user.password): + raise HTTPException(status_code=403, detail="Setup already completed.") + + def _validate_scheme(url: str) -> None: """Reject non-HTTP schemes and URLs without a hostname.""" parsed = urlparse(url) @@ -288,9 +321,10 @@ async def test_notification(req: TestNotificationRequest): @router.post("/complete", response_model=ResponseModel) -async def complete_setup(req: SetupCompleteRequest): +async def complete_setup(req: SetupCompleteRequest, request: Request): """Save all wizard configuration and mark setup as complete.""" _require_setup_needed() + await _require_default_admin_or_authenticated(request) try: # 1. Update user credentials diff --git a/backend/src/module/mcp/security.py b/backend/src/module/mcp/security.py index 24b30227..5eed9f3a 100644 --- a/backend/src/module/mcp/security.py +++ b/backend/src/module/mcp/security.py @@ -1,45 +1,21 @@ """MCP access control: configurable IP whitelist and bearer token authentication.""" -import ipaddress import logging -from functools import lru_cache from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import JSONResponse from module.conf import settings +from module.security.ip_allowlist import ( # noqa: F401 (re-exported for existing importers) + _is_allowed, + _parse_network, + clear_network_cache, +) logger = logging.getLogger(__name__) -@lru_cache(maxsize=128) -def _parse_network(cidr: str) -> ipaddress.IPv4Network | ipaddress.IPv6Network | None: - try: - return ipaddress.ip_network(cidr, strict=False) - except ValueError: - logger.warning("[MCP] Invalid CIDR in whitelist: %s", cidr) - return None - - -def _is_allowed(host: str, whitelist: list[str]) -> bool: - """Return True if *host* falls within any CIDR range in *whitelist*.""" - try: - addr = ipaddress.ip_address(host) - except ValueError: - return False - for cidr in whitelist: - net = _parse_network(cidr) - if net and addr in net: - return True - return False - - -def clear_network_cache(): - """Clear the parsed network cache (call after config reload).""" - _parse_network.cache_clear() - - class McpAccessMiddleware(BaseHTTPMiddleware): """Configurable access control for MCP endpoint. diff --git a/backend/src/module/security/api.py b/backend/src/module/security/api.py index 3a3ec55a..9183f718 100644 --- a/backend/src/module/security/api.py +++ b/backend/src/module/security/api.py @@ -1,3 +1,5 @@ +import logging +import os import secrets from datetime import datetime, timedelta @@ -6,11 +8,13 @@ from fastapi.security import OAuth2PasswordBearer from module.conf import settings from module.database import Database -from module.mcp.security import _is_allowed from module.models.user import User, UserUpdate +from .ip_allowlist import _is_allowed from .jwt import verify_token +logger = logging.getLogger(__name__) + oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") # Session lifetime mirrors the JWT expiry used at issuance (see auth.py / @@ -56,12 +60,16 @@ class SessionStore: active_user = SessionStore() -try: - from module.__version__ import VERSION -except ImportError: - VERSION = "DEV_VERSION" +# Opt-in only: unlike the old `VERSION == "DEV_VERSION"` check (which was +# true for every unofficial/self-built image), this requires an explicit +# environment variable so auth can never be silently bypassed in production. +DEV_AUTH_BYPASS = os.environ.get("AB_DEV_NO_AUTH") == "1" -DEV_AUTH_BYPASS = VERSION == "DEV_VERSION" +if DEV_AUTH_BYPASS: + logger.warning( + "!!! AB_DEV_NO_AUTH=1 is set — authentication is BYPASSED for every " + "request. This must never be set in production. !!!" + ) def check_login_ip(request: Request): @@ -84,7 +92,7 @@ async def get_current_user(request: Request, token: str = Cookie(None)): """FastAPI dependency that validates the current session. Accepts authentication via (in order of precedence): - 1. DEV_AUTH_BYPASS when running as DEV_VERSION. + 1. DEV_AUTH_BYPASS when ``AB_DEV_NO_AUTH=1`` is set in the environment. 2. ``Authorization: Bearer `` header matching ``login_tokens``. 3. HttpOnly ``token`` cookie containing a valid JWT with an active session. """ diff --git a/backend/src/module/security/ip_allowlist.py b/backend/src/module/security/ip_allowlist.py new file mode 100644 index 00000000..4c3ce23f --- /dev/null +++ b/backend/src/module/security/ip_allowlist.py @@ -0,0 +1,39 @@ +"""CIDR-based IP allowlist helpers, shared by the login and MCP guards. + +Lives in ``module.security`` (not ``module.mcp``) so the security layer does +not depend on the MCP layer; ``module.mcp.security`` re-exports these names +for backward compatibility. +""" + +import ipaddress +import logging +from functools import lru_cache + +logger = logging.getLogger(__name__) + + +@lru_cache(maxsize=128) +def _parse_network(cidr: str) -> ipaddress.IPv4Network | ipaddress.IPv6Network | None: + try: + return ipaddress.ip_network(cidr, strict=False) + except ValueError: + logger.warning("Invalid CIDR in whitelist: %s", cidr) + return None + + +def _is_allowed(host: str, whitelist: list[str]) -> bool: + """Return True if *host* falls within any CIDR range in *whitelist*.""" + try: + addr = ipaddress.ip_address(host) + except ValueError: + return False + for cidr in whitelist: + net = _parse_network(cidr) + if net and addr in net: + return True + return False + + +def clear_network_cache(): + """Clear the parsed network cache (call after config reload).""" + _parse_network.cache_clear() diff --git a/backend/src/module/security/jwt.py b/backend/src/module/security/jwt.py index 22388406..22668946 100644 --- a/backend/src/module/security/jwt.py +++ b/backend/src/module/security/jwt.py @@ -14,6 +14,7 @@ def _load_or_create_secret() -> str: secret = secrets.token_hex(32) _SECRET_PATH.parent.mkdir(parents=True, exist_ok=True) _SECRET_PATH.write_text(secret) + _SECRET_PATH.chmod(0o600) return secret @@ -37,7 +38,9 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None): # 解码 Token -def decode_token(token: str): +def decode_token(token: str | None): + if not token: + return None try: payload = jwt.decode(token, app_pwd_key, algorithms=[app_pwd_algorithm]) username = payload.get("sub") @@ -48,14 +51,11 @@ def decode_token(token: str): return None -def verify_token(token: str): - token_data = decode_token(token) - if token_data is None: - return None - expires = token_data.get("exp") - if datetime.now(timezone.utc) >= datetime.fromtimestamp(expires, tz=timezone.utc): - raise JWTError("Token expired") - return token_data +def verify_token(token: str | None): + # jose's jwt.decode() already validates "exp" and raises JWTError (caught + # above) for expired tokens, so a token returned by decode_token() is + # never expired — no separate expiry recheck is needed here. + return decode_token(token) # 密码加密&验证 diff --git a/backend/src/module/security/webauthn.py b/backend/src/module/security/webauthn.py index c0deb492..67fe3a15 100644 --- a/backend/src/module/security/webauthn.py +++ b/backend/src/module/security/webauthn.py @@ -7,6 +7,7 @@ import base64 import json import logging import time +from collections import OrderedDict from typing import List, Optional from webauthn import ( @@ -16,6 +17,10 @@ from webauthn import ( verify_authentication_response, verify_registration_response, ) +from webauthn.helpers import ( + parse_authentication_credential_json, + parse_client_data_json, +) from webauthn.helpers.cose import COSEAlgorithmIdentifier from webauthn.helpers.structs import ( AuthenticatorSelectionCriteria, @@ -298,13 +303,20 @@ class WebAuthnService: Raises: ValueError: 验证失败 """ - # Try all discoverable challenges to find the matching one - expected_challenge = None - for b64key, (challenge, _, lk) in list(self._challenges.items()): - if lk.startswith("auth_discoverable_"): - expected_challenge = challenge - del self._challenges[b64key] - break + # Pop the challenge the authenticator actually signed over (from + # clientDataJSON), not just the first stored discoverable challenge — + # with multiple concurrent discoverable logins in flight, popping by + # position let one caller consume another caller's challenge. + try: + parsed_credential = parse_authentication_credential_json(credential) + client_data = parse_client_data_json( + parsed_credential.response.client_data_json + ) + except Exception as e: + logger.warning(f"Failed to parse discoverable authentication response: {e}") + raise ValueError("Invalid authentication response") + + expected_challenge = self._pop_challenge_by_value(client_data.challenge) if not expected_challenge: raise ValueError("Challenge not found or expired") @@ -355,15 +367,24 @@ class WebAuthnService: # 全局 WebAuthn 服务实例存储 -_webauthn_services: dict[str, WebAuthnService] = {} +# Keyed by "rp_id:origin"; when rp_id/origin are derived from request headers +# (see api/passkey.py) a hostile client could otherwise grow this dict +# without bound, so it is capped as an LRU cache. +_WEBAUTHN_SERVICES_MAX = 8 +_webauthn_services: "OrderedDict[str, WebAuthnService]" = OrderedDict() def get_webauthn_service(rp_id: str, rp_name: str, origin: str) -> WebAuthnService: """ 获取或创建 WebAuthnService 实例 - 使用缓存以保持 challenge 状态 + 使用缓存以保持 challenge 状态;缓存容量有限(LRU 淘汰) """ key = f"{rp_id}:{origin}" - if key not in _webauthn_services: - _webauthn_services[key] = WebAuthnService(rp_id, rp_name, origin) - return _webauthn_services[key] + if key in _webauthn_services: + _webauthn_services.move_to_end(key) + return _webauthn_services[key] + if len(_webauthn_services) >= _WEBAUTHN_SERVICES_MAX: + _webauthn_services.popitem(last=False) + service = WebAuthnService(rp_id, rp_name, origin) + _webauthn_services[key] = service + return service diff --git a/backend/src/test/test_api_auth.py b/backend/src/test/test_api_auth.py index 9956707d..b6d52733 100644 --- a/backend/src/test/test_api_auth.py +++ b/backend/src/test/test_api_auth.py @@ -65,8 +65,8 @@ class TestAuthRequired: @patch("module.security.api.DEV_AUTH_BYPASS", False) def test_logout_unauthorized(self, unauthed_client): - """GET /auth/logout without auth returns 401.""" - response = unauthed_client.get("/api/v1/auth/logout") + """POST /auth/logout without auth returns 401.""" + response = unauthed_client.post("/api/v1/auth/logout") assert response.status_code == 401 @patch("module.security.api.DEV_AUTH_BYPASS", False) @@ -135,17 +135,17 @@ class TestRefreshToken: # --------------------------------------------------------------------------- -# GET /auth/logout +# POST /auth/logout # --------------------------------------------------------------------------- class TestLogout: def test_logout_success(self, authed_client): - """GET /auth/logout clears session and returns success.""" + """POST /auth/logout clears session and returns success.""" token = create_access_token(data={"sub": "testuser"}) authed_client.cookies.set("token", token) with patch("module.api.auth.active_user", _store("testuser")): - response = authed_client.get("/api/v1/auth/logout") + response = authed_client.post("/api/v1/auth/logout") assert response.status_code == 200 data = response.json() @@ -237,21 +237,21 @@ class TestRefreshTokenCookieBehavior: class TestLogoutCookieBehavior: def test_logout_removes_only_current_user(self, authed_client): - """GET /logout removes the current user from active_user, not others.""" + """POST /logout removes the current user from active_user, not others.""" token = create_access_token(data={"sub": "testuser"}) authed_client.cookies.set("token", token) active_users = _store("testuser", "otheruser") with patch("module.api.auth.active_user", active_users): - response = authed_client.get("/api/v1/auth/logout") + response = authed_client.post("/api/v1/auth/logout") assert response.status_code == 200 assert "testuser" not in active_users assert "otheruser" in active_users def test_logout_with_no_cookie_still_succeeds(self, authed_client): - """GET /logout with no cookie clears nothing but returns success.""" + """POST /logout with no cookie clears nothing but returns success.""" with patch("module.api.auth.decode_token", return_value=None): with patch("module.api.auth.active_user", SessionStore()): - response = authed_client.get("/api/v1/auth/logout") + response = authed_client.post("/api/v1/auth/logout") assert response.status_code == 200 diff --git a/backend/src/test/test_api_passkey.py b/backend/src/test/test_api_passkey.py index ad034ddf..62e199a5 100644 --- a/backend/src/test/test_api_passkey.py +++ b/backend/src/test/test_api_passkey.py @@ -294,7 +294,11 @@ class TestAuthOptions: assert "challenge" in data def test_get_auth_options_user_not_found(self, unauthed_client, mock_webauthn): - """POST /passkey/auth/options with non-existent user returns 404.""" + """POST /passkey/auth/options with non-existent user returns 400. + + Same status/message as "user exists but has no passkeys" so this + endpoint can't be used to enumerate valid usernames. + """ with patch( "module.api.passkey._get_webauthn_from_request", return_value=mock_webauthn ): @@ -313,7 +317,7 @@ class TestAuthOptions: "/api/v1/passkey/auth/options", json={"username": "nonexistent"} ) - assert response.status_code == 404 + assert response.status_code == 400 # --------------------------------------------------------------------------- diff --git a/backend/src/test/test_auth.py b/backend/src/test/test_auth.py index b944435d..7882f93d 100644 --- a/backend/src/test/test_auth.py +++ b/backend/src/test/test_auth.py @@ -154,7 +154,6 @@ class TestGetCurrentUser: @staticmethod def _mock_request(authorization=""): """Create a mock Request with the given Authorization header.""" - from unittest.mock import MagicMock request = MagicMock() request.headers = {"authorization": authorization} @@ -261,7 +260,6 @@ class TestGetCurrentUser: class TestCheckLoginIp: @staticmethod def _make_request(host: str | None): - from unittest.mock import MagicMock request = MagicMock() if host is None: diff --git a/backend/src/test/test_mcp_security.py b/backend/src/test/test_mcp_security.py index 6d96397d..48bed948 100644 --- a/backend/src/test/test_mcp_security.py +++ b/backend/src/test/test_mcp_security.py @@ -10,7 +10,6 @@ from starlette.testclient import TestClient from module.mcp.security import McpAccessMiddleware, _is_allowed, clear_network_cache - # --------------------------------------------------------------------------- # _is_allowed() unit tests # ---------------------------------------------------------------------------