nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-11-01 13:15:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(security): env-gated auth bypass, WebAuthn hardening, passkey/JWT fixes

Browse Source 
- Gate dev auth bypass on AB_DEV_NO_AUTH=1 (was: any non-CI build), warn loudly when on
- WebAuthn rp_id/origin from settings when set; LRU-cap the service cache; match challenge by clientDataJSON
- Passkey: enforce login-IP whitelist, remove user enumeration, stop echoing str(e) in 500s
- Auth: migrate session/JWT on username change; guard decode_token(None); logout is POST + samesite=strict cookie
- Setup: require default admin account before pre-auth /setup/complete
- JWT secret written 0600; move IP-allowlist helper into module/security (new ip_allowlist.py)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014w1Z6Nxy6XTRgkFXqPr9Zh
This commit is contained in:
Estrella Pan
2026-07-02 17:05:34 +02:00
parent 4d7104ca7f
commit 0f44efa2af
12 changed files with 263 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
backend/src/module/api/auth.py
View File

@@ -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"}

87
backend/src/module/api/passkey.py
View File

@@ -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)

40
backend/src/module/api/setup.py
View File

@@ -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

34
backend/src/module/mcp/security.py
View File

@@ -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.

22
backend/src/module/security/api.py
View File

@@ -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 <token>`` header matching ``login_tokens``.
3. HttpOnly ``token`` cookie containing a valid JWT with an active session.
"""

39
backend/src/module/security/ip_allowlist.py Normal file
View File

@@ -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()

18
backend/src/module/security/jwt.py
View File

@@ -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)
# 密码加密&验证

45
backend/src/module/security/webauthn.py
View File

@@ -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

18
backend/src/test/test_api_auth.py
View File

@@ -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

8
backend/src/test/test_api_passkey.py
View File

@@ -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
# ---------------------------------------------------------------------------

2
backend/src/test/test_auth.py
View File

@@ -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:

1
backend/src/test/test_mcp_security.py
View File

@@ -10,7 +10,6 @@ from starlette.testclient import TestClient
from module.mcp.security import McpAccessMiddleware, _is_allowed, clear_network_cache
# ---------------------------------------------------------------------------
# _is_allowed() unit tests
# ---------------------------------------------------------------------------