From 65034ec16eb7f9cb1ffbaabcd734efa71e746954 Mon Sep 17 00:00:00 2001 From: LouisHong <30621586+Activer007@users.noreply.github.com> Date: Fri, 13 Mar 2026 23:15:40 +0800 Subject: [PATCH] =?UTF-8?q?#602=20[PR=203]=20=E8=AE=A4=E8=AF=81=E5=AE=89?= =?UTF-8?q?=E5=85=A8=E4=B8=8E=E5=88=9D=E5=A7=8B=E5=8C=96=E6=B5=81=E7=A8=8B?= =?UTF-8?q?=20(Auth=20&=20Security)=20(#639)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(auth): add runtime auth settings endpoint * feat(auth): robust runtime auth settings and security hardening - Added '/api/v1/auth/settings' to enable/disable Web authentication at runtime - Implemented session secret rotation (HMAC invalidation) on any auth toggle - Fixed TOCTOU race condition in settings update via mandatory session validation - Added comprehensive integration tests for auth re-enablement and rate limiting - Improved credential persistence: added 'ENV_FILE' support and atomic 'replace' operations - Refactored auth initialization flow to prevent password overwrites and ensure rollback on failure * fix(auth): harden auth settings race checks * fix(auth): share config writer and propagate rotation failures * fix(auth): order toggle persistence and document worker scope * fix(auth): enhance error handling and logging for auth toggle application --------- Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com> --- README.md | 5 +- api/v1/endpoints/auth.py | 262 ++++++++++++- apps/dsa-web/src/api/auth.ts | 25 ++ docs/CHANGELOG.md | 12 + src/auth.py | 63 ++- src/services/system_config_service.py | 12 + tests/test_auth.py | 74 ++++ tests/test_auth_api.py | 526 ++++++++++++++++++++++---- 8 files changed, 883 insertions(+), 96 deletions(-) diff --git a/README.md b/README.md index b3a1d496c..78e222f5b 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,9 @@ > 历史报告详情会优先展示 AI 返回的原始「狙击点位」文本，避免区间价、条件说明等复杂内容在历史回看时被压缩成单个数字。 +> Web 管理认证支持运行时开关；如果系统中已保留管理员密码，重新开启认证时必须提供当前密码，避免在认证关闭窗口内直接获取新的管理员会话。 +> 多进程/多 worker 部署时，认证开关仅在当前进程即时生效；需重启或滚动重启全部 worker 以统一状态。 + ### 技术栈与数据来源 | 类型 | 支持 | @@ -294,7 +297,7 @@ LITELLM_MODEL=openai/deepseek-chat 包含完整的配置管理、任务监控和手动分析功能。 -**可选密码保护**：在 `.env` 中设置 `ADMIN_AUTH_ENABLED=true` 可启用 Web 登录，首次访问在网页设置初始密码，保护 Settings 中的 API 密钥等敏感配置。详见 [完整指南](docs/full-guide.md)。 +**可选密码保护**：在 `.env` 中设置 `ADMIN_AUTH_ENABLED=true` 可启用 Web 登录，首次访问在网页设置初始密码，保护 Settings 中的 API 密钥等敏感配置。系统设置现支持运行时开启或关闭认证；关闭认证不会删除已保存密码，后续可直接重新启用。详见 [完整指南](docs/full-guide.md)。 ### 智能导入 diff --git a/api/v1/endpoints/auth.py b/api/v1/endpoints/auth.py index 02669cf46..06a504979 100644 --- a/api/v1/endpoints/auth.py +++ b/api/v1/endpoints/auth.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, Request from fastapi.responses import JSONResponse, Response from pydantic import BaseModel, Field +from api.deps import get_system_config_service from src.auth import ( COOKIE_NAME, SESSION_MAX_AGE_HOURS_DEFAULT, @@ -18,14 +19,20 @@ from src.auth import ( clear_rate_limit, create_session, get_client_ip, + has_stored_password, is_auth_enabled, is_password_changeable, is_password_set, record_login_failure, + refresh_auth_state, + rotate_session_secret, set_initial_password, verify_password, + verify_stored_password, verify_session, ) +from src.config import Config, setup_env +from src.core.config_manager import ConfigManager logger = logging.getLogger(__name__) @@ -51,6 +58,17 @@ class ChangePasswordRequest(BaseModel): new_password_confirm: str = Field(default="", alias="newPasswordConfirm") +class AuthSettingsRequest(BaseModel): + """Update auth enablement and initial password settings.""" + + model_config = {"populate_by_name": True} + + auth_enabled: bool = Field(alias="authEnabled") + password: str = Field(default="") + password_confirm: str | None = Field(default=None, alias="passwordConfirm") + current_password: str = Field(default="", alias="currentPassword") + + def _cookie_params(request: Request) -> dict: """Build cookie params including Secure based on request.""" secure = False @@ -76,6 +94,66 @@ def _cookie_params(request: Request) -> dict: } +def _apply_auth_enabled(enabled: bool, request: Request | None = None) -> bool: + """Persist auth toggle to .env and reload runtime config.""" + manager_applied = False + if request is not None: + try: + service = get_system_config_service(request) + service.apply_simple_updates( + updates=[("ADMIN_AUTH_ENABLED", "true" if enabled else "false")], + mask_token="******", + ) + manager_applied = True + except Exception as exc: + logger.warning( + "Failed to apply auth toggle via shared SystemConfigService, falling back: %s", + exc, + exc_info=True, + ) + manager_applied = False + + if not manager_applied: + try: + manager = ConfigManager() + manager.apply_updates( + updates=[("ADMIN_AUTH_ENABLED", "true" if enabled else "false")], + sensitive_keys=set(), + mask_token="******", + ) + manager_applied = True + except Exception as exc: + logger.error("Failed to apply auth toggle via ConfigManager: %s", exc, exc_info=True) + manager_applied = False + + if not manager_applied: + return False + + Config.reset_instance() + setup_env(override=True) + refresh_auth_state() + return True + + +def _password_set_for_response(auth_enabled: bool) -> bool: + """Avoid exposing stored-password state when auth is disabled.""" + return is_password_set() if auth_enabled else False + + +def _set_session_cookie(response: Response, session_value: str, request: Request) -> None: + """Attach the admin session cookie to a response.""" + params = _cookie_params(request) + response.set_cookie( + key=COOKIE_NAME, + value=session_value, + httponly=params["httponly"], + samesite=params["samesite"], + secure=params["secure"], + path=params["path"], + max_age=params["max_age"], + ) + + @router.get( "/status", summary="Get auth status", @@ -91,11 +169,182 @@ async def auth_status(request: Request): return { "authEnabled": auth_enabled, "loggedIn": logged_in, - "passwordSet": is_password_set() if auth_enabled else False, + "passwordSet": _password_set_for_response(auth_enabled), "passwordChangeable": is_password_changeable() if auth_enabled else False, } +@router.post( + "/settings", + summary="Update auth settings", + description=( + "Enable or disable password login. When enabling without an existing password, " + "password + passwordConfirm are required. When re-enabling with a stored password, " + "currentPassword is required." + ), +) +async def auth_update_settings(request: Request, body: AuthSettingsRequest): + """Manage auth enablement from the settings page.""" + target_enabled = body.auth_enabled + current_enabled = is_auth_enabled() + stored_password_exists = has_stored_password() + + password = (body.password or "").strip() + confirm = (body.password_confirm or "").strip() + current_password = (body.current_password or "").strip() + + if target_enabled: + if password or confirm: + if stored_password_exists: + return JSONResponse( + status_code=400, + content={ + "error": "password_already_set", + "message": "已存在管理员密码，请启用认证后通过修改密码功能更新", + }, + ) + if not password: + return JSONResponse( + status_code=400, + content={"error": "password_required", "message": "请输入要设置的管理员密码"}, + ) + if password != confirm: + return JSONResponse( + status_code=400, + content={"error": "password_mismatch", "message": "两次输入的密码不一致"}, + ) + if has_stored_password(): + return JSONResponse( + status_code=400, + content={ + "error": "password_already_set", + "message": "已存在管理员密码，请启用认证后通过修改密码功能更新", + }, + ) + err = set_initial_password(password) + if err: + return JSONResponse( + status_code=400, + content={"error": "invalid_password", "message": err}, + ) + elif not stored_password_exists: + return JSONResponse( + status_code=400, + content={"error": "password_required", "message": "开启密码登录前请先设置密码"}, + ) + else: + # P1 Vulnerability Fix: Enforce current-password check independent of global cached flag + # We must verify they actually possess a valid admin session, otherwise an attacker + # could hit a race condition when auth becomes enabled mid-flight. + # This triggers whenever trying to enable/keep enabled an existing auth setup. + cookie_val = request.cookies.get(COOKIE_NAME) + # if target_enabled is True here, they are requesting to enable or keep auth enabled + is_valid_session = cookie_val and verify_session(cookie_val) + + if not is_valid_session: + if not current_password: + return JSONResponse( + status_code=400, + content={"error": "current_required", "message": "重新开启认证前请输入当前密码"}, + ) + ip = get_client_ip(request) + if not check_rate_limit(ip): + return JSONResponse( + status_code=429, + content={ + "error": "rate_limited", + "message": "Too many failed attempts. Please try again later.", + }, + ) + if not verify_stored_password(current_password): + record_login_failure(ip) + return JSONResponse( + status_code=401, + content={"error": "invalid_password", "message": "当前密码错误"}, + ) + clear_rate_limit(ip) + else: + if current_enabled: + cookie_val = request.cookies.get(COOKIE_NAME) + is_valid_session = cookie_val and verify_session(cookie_val) + + if not is_valid_session: + if not current_password: + return JSONResponse( + status_code=400, + content={"error": "current_required", "message": "关闭认证前请输入当前密码"}, + ) + ip = get_client_ip(request) + if not check_rate_limit(ip): + return JSONResponse( + status_code=429, + content={ + "error": "rate_limited", + "message": "Too many failed attempts. Please try again later.", + }, + ) + if not verify_stored_password(current_password): + record_login_failure(ip) + return JSONResponse( + status_code=401, + content={"error": "invalid_password", "message": "当前密码错误"}, + ) + clear_rate_limit(ip) + + if target_enabled != current_enabled: + if not _apply_auth_enabled(target_enabled, request=request): + return JSONResponse( + status_code=500, + content={"error": "internal_error", "message": "Failed to update auth settings"}, + ) + if not rotate_session_secret(): + rollback_ok = _apply_auth_enabled(current_enabled, request=request) + if not rollback_ok: + logger.error("Failed to roll back auth state after session secret rotation failure") + return JSONResponse( + status_code=500, + content={"error": "internal_error", "message": "Failed to rotate session secret"}, + ) + else: + if not _apply_auth_enabled(target_enabled, request=request): + return JSONResponse( + status_code=500, + content={"error": "internal_error", "message": "Failed to update auth settings"}, + ) + + if target_enabled: + session_val = create_session() + if not session_val: + rollback_ok = _apply_auth_enabled(current_enabled, request=request) + if not rollback_ok: + logger.error("Failed to roll back auth state after session creation failure") + return JSONResponse( + status_code=500, + content={"error": "internal_error", "message": "Failed to create session"}, + ) + resp = JSONResponse( + content={ + "authEnabled": True, + "loggedIn": True, + "passwordSet": _password_set_for_response(True), + "passwordChangeable": True, + } + ) + _set_session_cookie(resp, session_val, request) + return resp + + resp = JSONResponse( + content={ + "authEnabled": False, + "loggedIn": False, + "passwordSet": _password_set_for_response(False), + "passwordChangeable": False, + } + ) + resp.delete_cookie(key=COOKIE_NAME, path="/") + return resp + + @router.post( "/login", summary="Login or set initial password", @@ -161,16 +410,7 @@ async def auth_login(request: Request, body: LoginRequest): ) resp = JSONResponse(content={"ok": True}) - params = _cookie_params(request) - resp.set_cookie( - key=COOKIE_NAME, - value=session_val, - httponly=params["httponly"], - samesite=params["samesite"], - secure=params["secure"], - path=params["path"], - max_age=params["max_age"], - ) + _set_session_cookie(resp, session_val, request) return resp diff --git a/apps/dsa-web/src/api/auth.ts b/apps/dsa-web/src/api/auth.ts index a4a3d6704..d90bbd92d 100644 --- a/apps/dsa-web/src/api/auth.ts +++ b/apps/dsa-web/src/api/auth.ts @@ -13,6 +13,31 @@ export const authApi = { return data; }, + async updateSettings( + authEnabled: boolean, + password?: string, + passwordConfirm?: string, + currentPassword?: string + ): Promise { + const body: { + authEnabled: boolean; + password?: string; + passwordConfirm?: string; + currentPassword?: string; + } = { authEnabled }; + if (password !== undefined) { + body.password = password; + } + if (passwordConfirm !== undefined) { + body.passwordConfirm = passwordConfirm; + } + if (currentPassword !== undefined) { + body.currentPassword = currentPassword; + } + const { data } = await apiClient.post('/api/v1/auth/settings', body); + return data; + }, + async login(password: string, passwordConfirm?: string): Promise { const body: { password: string; passwordConfirm?: string } = { password }; if (passwordConfirm !== undefined) { diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 2d75f8710..843f96ba0 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -10,6 +10,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] ### Added +- 🔐 **Auth settings API** — new `POST /api/v1/auth/settings` endpoint to enable or disable Web authentication at runtime and set the initial admin password when needed + +### Changed +- 🔐 **Auth password state semantics** — stored password existence is now tracked independently from auth enablement; when auth is disabled, `/api/v1/auth/status` returns `passwordSet=false` while preserving the saved password for future re-enable +- 🔐 **Auth settings re-enable hardening** — re-enabling auth with a stored password now requires `currentPassword`, and failed session creation rolls back the auth toggle to avoid lockout + +### Fixed +- 🐛 **Session secret rotation on Windows** — use atomic replace so auth toggles invalidate existing sessions even when `.session_secret` already exists +- 🐛 **Auth toggle atomicity** — persist `ADMIN_AUTH_ENABLED` before rotating session secret; on rotation failure, roll back to the previous auth state - openclaw Skill 集成指南 — 新增 [docs/openclaw-skill-integration.md](openclaw-skill-integration.md)，说明如何通过 openclaw Skill 调用 DSA API - ⚙️ **LLM channel protocol/test UX** — `.env` and Web settings now share the same channel shape (`LLM_CHANNELS` + `LLM__PROTOCOL/BASE_URL/API_KEY/MODELS/ENABLED`); settings page adds per-channel connection testing, primary/fallback/vision model selection, and protocol-aware model prefixing @@ -18,6 +27,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - 🐛 **P0 基本面聚合稳定性修复** (#614) — 修复 `get_stock_info` 板块语义回归（新增 `belong_boards` 并保留 `boards` 兼容别名）、引入基本面上下文精简返回以控制 token、为基本面缓存增加最大条目淘汰，并补齐 ETF 总体状态聚合与 NaN 板块字段过滤，保证 fail-open 与最小入侵。 - 🔧 **GitHub Actions 搜索引擎环境变量补充** — 工作流新增 `MINIMAX_API_KEYS`、`BRAVE_API_KEYS`、`SEARXNG_BASE_URLS` 环境变量映射，使 GitHub Actions 用户可配置 MiniMax、Brave、SearXNG 搜索服务（此前 v3.5.0 已添加 provider 实现但缺少工作流配置） +### Notes +- ⚠️ **Multi-worker auth toggles** — runtime auth updates are process-local; multi-worker deployments must restart/roll workers to keep auth state consistent + ## [3.5.0] - 2026-03-12 ### Added diff --git a/src/auth.py b/src/auth.py index 81893ed17..5d952c539 100644 --- a/src/auth.py +++ b/src/auth.py @@ -69,7 +69,8 @@ def _get_credential_path() -> Path: def _is_auth_enabled_from_env() -> bool: """Read ADMIN_AUTH_ENABLED from .env file.""" _ensure_env_loaded() - env_path = Path(__file__).resolve().parent.parent / ".env" + env_file = os.getenv("ENV_FILE") + env_path = Path(env_file) if env_file else Path(__file__).resolve().parent.parent / ".env" if not env_path.exists(): return False values = dotenv_values(env_path) @@ -77,6 +78,26 @@ def _is_auth_enabled_from_env() -> bool: return val in ("true", "1", "yes") +def rotate_session_secret() -> bool: + """Rotate the session signing secret to invalidate all active sessions.""" + global _session_secret + data_dir = _get_data_dir() + secret_path = data_dir / ".session_secret" + data_dir.mkdir(parents=True, exist_ok=True) + new_secret = secrets.token_bytes(32) + try: + tmp_path = secret_path.with_suffix(".tmp") + tmp_path.write_bytes(new_secret) + tmp_path.chmod(0o600) + tmp_path.replace(secret_path) + _session_secret = new_secret + logger.info("Session secret rotated successfully") + return True + except OSError as e: + logger.error("Failed to rotate .session_secret: %s", e) + return False + + def _load_session_secret() -> Optional[bytes]: """Load or create session secret.""" global _session_secret @@ -92,8 +113,10 @@ def _load_session_secret() -> Optional[bytes]: if len(_session_secret) != 32: logger.warning("Invalid .session_secret length, regenerating") _session_secret = None - else: - return _session_secret + if rotate_session_secret(): + return _session_secret + return None + return _session_secret data_dir.mkdir(parents=True, exist_ok=True) new_secret = secrets.token_bytes(32) @@ -163,6 +186,14 @@ def _load_credential_from_file() -> bool: return False +def refresh_auth_state() -> None: + """Reload auth-related state from disk and env.""" + global _auth_enabled, _session_secret + _auth_enabled = None + _session_secret = None + _load_credential_from_file() + + def is_auth_enabled() -> bool: """Return whether admin authentication is enabled (ADMIN_AUTH_ENABLED=true).""" global _auth_enabled @@ -172,12 +203,23 @@ def is_auth_enabled() -> bool: return _auth_enabled +def has_stored_password() -> bool: + """Return whether a valid stored password hash exists on disk.""" + return _load_credential_from_file() + + +def verify_stored_password(password: str) -> bool: + """Verify password against stored credential even when auth is disabled.""" + if not has_stored_password(): + return False + return _verify_password_hash(password, _password_hash_salt, _password_hash_stored) + + def is_password_set() -> bool: """Return whether initial password has been set (credential file exists and valid).""" if not is_auth_enabled(): return False - _load_credential_from_file() - return _password_hash_stored is not None + return has_stored_password() def is_password_changeable() -> bool: @@ -229,7 +271,8 @@ def set_initial_password(password: str) -> Optional[str]: tmp_path = cred_path.with_suffix(".tmp") tmp_path.write_text(content) tmp_path.chmod(0o600) - tmp_path.rename(cred_path) + tmp_path.replace(cred_path) + _load_credential_from_file() return None except OSError as e: logger.error("Failed to write credential file: %s", e) @@ -240,9 +283,7 @@ def verify_password(password: str) -> bool: """Verify password against stored credential. Constant-time where applicable.""" if not is_auth_enabled(): return True - if not is_password_set(): - return False - return _verify_password_hash(password, _password_hash_salt, _password_hash_stored) + return verify_stored_password(password) def change_password(current: str, new: str) -> Optional[str]: @@ -279,7 +320,7 @@ def change_password(current: str, new: str) -> Optional[str]: tmp_path = cred_path.with_suffix(".tmp") tmp_path.write_text(content) tmp_path.chmod(0o600) - tmp_path.rename(cred_path) + tmp_path.replace(cred_path) # Reload into memory so subsequent verify_password uses new hash _load_credential_from_file() return None @@ -404,7 +445,7 @@ def overwrite_password(new_password: str) -> Optional[str]: tmp_path = cred_path.with_suffix(".tmp") tmp_path.write_text(content) tmp_path.chmod(0o600) - tmp_path.rename(cred_path) + tmp_path.replace(cred_path) _load_credential_from_file() return None except OSError as e: diff --git a/src/services/system_config_service.py b/src/services/system_config_service.py index 536a47e24..c74f4be03 100644 --- a/src/services/system_config_service.py +++ b/src/services/system_config_service.py @@ -264,6 +264,18 @@ class SystemConfigService: "warnings": warnings, } + def apply_simple_updates( + self, + updates: Sequence[Tuple[str, str]], + mask_token: str = "******", + ) -> None: + """Apply raw key updates without validation (internal service use only).""" + self._manager.apply_updates( + updates=updates, + sensitive_keys=set(), + mask_token=mask_token, + ) + def _collect_issues(self, items: Sequence[Dict[str, str]], mask_token: str) -> List[Dict[str, Any]]: """Collect field-level and cross-field validation issues.""" current_map = self._manager.read_config_map() diff --git a/tests/test_auth.py b/tests/test_auth.py index 296eb3935..21ebfb61b 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -128,6 +128,36 @@ class AuthSessionTestCase(unittest.TestCase): self._patch_env_and_run(test_fn=run) + def test_rotate_session_secret_overwrites_existing(self) -> None: + def run(): + secret_path = self.data_dir / ".session_secret" + secret_path.write_bytes(b"a" * 32) + secret_path.chmod(0o600) + old_secret = secret_path.read_bytes() + + auth.rotate_session_secret() + + new_secret = secret_path.read_bytes() + self.assertNotEqual(old_secret, new_secret) + self.assertEqual(auth._session_secret, new_secret) + + self._patch_env_and_run(test_fn=run) + + def test_load_session_secret_regenerates_invalid_length(self) -> None: + def run(): + secret_path = self.data_dir / ".session_secret" + secret_path.write_bytes(b"x") + secret_path.chmod(0o600) + + tok = auth.create_session() + self.assertTrue(tok) + + new_secret = secret_path.read_bytes() + self.assertEqual(len(new_secret), 32) + self.assertNotEqual(new_secret, b"x") + + self._patch_env_and_run(test_fn=run) + class AuthRateLimitTestCase(unittest.TestCase): """Test rate limiting.""" @@ -172,11 +202,55 @@ class AuthSetPasswordTestCase(unittest.TestCase): def run(): err = auth.set_initial_password("password123") self.assertIsNone(err) + self.assertIsNotNone(auth._password_hash_stored) self.assertTrue(auth.is_password_set()) self.assertTrue(auth.verify_password("password123")) self._run_with_patch(run) + def test_has_stored_password_remains_true_after_auth_disabled(self) -> None: + def run(): + err = auth.set_initial_password("password123") + self.assertIsNone(err) + self.assertTrue(auth.has_stored_password()) + + auth._auth_enabled = False + self.assertTrue(auth.has_stored_password()) + self.assertFalse(auth.is_password_set()) + + self._run_with_patch(run) + + def test_verify_stored_password_when_auth_disabled(self) -> None: + def run(): + err = auth.set_initial_password("password123") + self.assertIsNone(err) + + auth._auth_enabled = False + self.assertTrue(auth.verify_stored_password("password123")) + self.assertFalse(auth.verify_stored_password("wrongpass")) + + self._run_with_patch(run) + + def test_is_auth_enabled_from_env_respects_env_file(self) -> None: + custom_env = self.data_dir / "custom.env" + custom_env.write_text("ADMIN_AUTH_ENABLED=true\n", encoding="utf-8") + + with patch.dict(os.environ, {"ENV_FILE": str(custom_env)}): + auth._auth_enabled = None + self.assertTrue(auth._is_auth_enabled_from_env()) + + def test_refresh_auth_state_clears_session_secret_cache(self) -> None: + def run(): + first_secret = auth.create_session() + self.assertTrue(first_secret) + self.assertIsNotNone(auth._session_secret) + + auth._session_secret = b"x" * 32 + auth.refresh_auth_state() + self.assertNotEqual(auth._session_secret, b"x" * 32) + + self._run_with_patch(run) + def test_set_initial_password_invalid(self) -> None: def run(): self.assertIsNotNone(auth.set_initial_password("")) diff --git a/tests/test_auth_api.py b/tests/test_auth_api.py index 3b5829a9e..834b2f449 100644 --- a/tests/test_auth_api.py +++ b/tests/test_auth_api.py @@ -1,14 +1,18 @@ # -*- coding: utf-8 -*- """Integration tests for auth API endpoints (login, logout, change-password, API protection).""" +import asyncio import os import sys import tempfile import unittest from pathlib import Path -from unittest.mock import patch, MagicMock +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch -from fastapi.testclient import TestClient +from dotenv import dotenv_values +from fastapi.responses import Response +from starlette.requests import Request # Keep this test runnable when optional LLM runtime deps are not installed. try: @@ -17,7 +21,8 @@ except ModuleNotFoundError: sys.modules["litellm"] = MagicMock() import src.auth as auth -from api.app import create_app +from api.middlewares.auth import AuthMiddleware +from api.v1.endpoints import auth as auth_endpoint from src.config import Config @@ -50,9 +55,6 @@ class AuthApiTestCase(unittest.TestCase): self.auth_patcher.start() self.data_dir_patcher.start() - app = create_app(static_dir=self.data_dir / "empty-static") - self.client = TestClient(app) - def tearDown(self) -> None: self.auth_patcher.stop() self.data_dir_patcher.stop() @@ -61,113 +63,491 @@ class AuthApiTestCase(unittest.TestCase): os.environ.pop("DATABASE_PATH", None) self.temp_dir.cleanup() + def _read_auth_enabled_from_env(self) -> bool: + values = dotenv_values(self.env_path) + return (values.get("ADMIN_AUTH_ENABLED") or "").strip().lower() in ("true", "1", "yes") + + @staticmethod + def _build_request(cookies=None): + return SimpleNamespace( + headers={}, + url=SimpleNamespace(scheme="http"), + cookies=cookies or {}, + client=SimpleNamespace(host="127.0.0.1"), + ) + def test_auth_status_when_password_not_set(self) -> None: - response = self.client.get("/api/v1/auth/status") - self.assertEqual(response.status_code, 200) - data = response.json() + data = asyncio.run(auth_endpoint.auth_status(self._build_request())) self.assertTrue(data["authEnabled"]) self.assertFalse(data["passwordSet"]) self.assertFalse(data["loggedIn"]) def test_login_first_time_set_initial_password(self) -> None: - response = self.client.post( - "/api/v1/auth/login", - json={"password": "newpass123", "passwordConfirm": "newpass123"}, + response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="newpass123", passwordConfirm="newpass123"), + ) ) self.assertEqual(response.status_code, 200) - self.assertIn("dsa_session", response.cookies) - self.assertTrue(response.json().get("ok")) + self.assertIn("dsa_session=", response.headers["set-cookie"]) + self.assertIn(b'"ok":true', response.body) def test_login_first_time_mismatch_rejected(self) -> None: - response = self.client.post( - "/api/v1/auth/login", - json={"password": "pass1", "passwordConfirm": "pass2"}, + response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="pass1", passwordConfirm="pass2"), + ) ) self.assertEqual(response.status_code, 400) - self.assertIn("password_mismatch", response.json().get("error", "")) + self.assertIn(b'"error":"password_mismatch"', response.body) def test_login_after_set_normal_login(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "mypass456", "passwordConfirm": "mypass456"}, + first_response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="mypass456", passwordConfirm="mypass456"), + ) ) - response = self.client.post( - "/api/v1/auth/login", - json={"password": "mypass456"}, + self.assertEqual(first_response.status_code, 200) + + response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="mypass456"), + ) ) self.assertEqual(response.status_code, 200) - self.assertTrue(response.json().get("ok")) + self.assertIn(b'"ok":true', response.body) def test_login_wrong_password_returns_401(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "correct", "passwordConfirm": "correct"}, + first_response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="correct", passwordConfirm="correct"), + ) ) - response = self.client.post( - "/api/v1/auth/login", - json={"password": "wrong"}, + self.assertEqual(first_response.status_code, 200) + + response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="wrong"), + ) ) self.assertEqual(response.status_code, 401) def test_logout_clears_cookie(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "passwd6", "passwordConfirm": "passwd6"}, - ) - self.assertIn("dsa_session", self.client.cookies) - self.client.post("/api/v1/auth/logout") - response = self.client.get("/api/v1/system/config") - self.assertEqual(response.status_code, 401, "After logout, protected API should return 401") + response = asyncio.run(auth_endpoint.auth_logout(self._build_request())) + self.assertEqual(response.status_code, 204) + self.assertIn("dsa_session=", response.headers["set-cookie"]) def test_change_password_requires_session(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "oldpass6", "passwordConfirm": "oldpass6"}, + first_response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="oldpass6", passwordConfirm="oldpass6"), + ) ) - response = self.client.post( - "/api/v1/auth/change-password", - json={ - "currentPassword": "oldpass6", - "newPassword": "newpass6", - "newPasswordConfirm": "newpass6", - }, + self.assertEqual(first_response.status_code, 200) + + response = asyncio.run( + auth_endpoint.auth_change_password( + auth_endpoint.ChangePasswordRequest( + currentPassword="oldpass6", + newPassword="newpass6", + newPasswordConfirm="newpass6", + ) + ) ) self.assertIn(response.status_code, (200, 204)) def test_change_password_wrong_current_rejected(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "actual6", "passwordConfirm": "actual6"}, + first_response = asyncio.run( + auth_endpoint.auth_login( + self._build_request(), + auth_endpoint.LoginRequest(password="actual6", passwordConfirm="actual6"), + ) ) - response = self.client.post( - "/api/v1/auth/change-password", - json={ - "currentPassword": "wrong", - "newPassword": "new123", - "newPasswordConfirm": "new123", - }, + self.assertEqual(first_response.status_code, 200) + + response = asyncio.run( + auth_endpoint.auth_change_password( + auth_endpoint.ChangePasswordRequest( + currentPassword="wrong", + newPassword="new123", + newPasswordConfirm="new123", + ) + ) ) self.assertEqual(response.status_code, 400) def test_protected_api_returns_401_without_session(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "passwd6", "passwordConfirm": "passwd6"}, - ) - client_no_cookie = TestClient( - create_app(static_dir=self.data_dir / "empty-static"), - raise_server_exceptions=False, - ) - response = client_no_cookie.get("/api/v1/system/config") + scope = { + "type": "http", + "method": "GET", + "path": "/api/v1/system/config", + "headers": [], + "query_string": b"", + "scheme": "http", + "client": ("127.0.0.1", 1234), + "server": ("testserver", 80), + "root_path": "", + } + request = Request(scope) + middleware = AuthMiddleware(app=MagicMock()) + + with patch("api.middlewares.auth.is_auth_enabled", return_value=True): + response = asyncio.run(middleware.dispatch(request, AsyncMock(return_value=Response(status_code=200)))) + self.assertEqual(response.status_code, 401) def test_protected_api_accessible_with_session(self) -> None: - self.client.post( - "/api/v1/auth/login", - json={"password": "passwd6", "passwordConfirm": "passwd6"}, - ) - response = self.client.get("/api/v1/system/config") + scope = { + "type": "http", + "method": "GET", + "path": "/api/v1/system/config", + "headers": [(b"cookie", b"dsa_session=test-session")], + "query_string": b"", + "scheme": "http", + "client": ("127.0.0.1", 1234), + "server": ("testserver", 80), + "root_path": "", + } + request = Request(scope) + middleware = AuthMiddleware(app=MagicMock()) + next_response = Response(status_code=200) + call_next = AsyncMock(return_value=next_response) + + with patch("api.middlewares.auth.is_auth_enabled", return_value=True): + with patch("api.middlewares.auth.verify_session", return_value=True): + response = asyncio.run(middleware.dispatch(request, call_next)) + self.assertEqual(response.status_code, 200) + call_next.assert_awaited_once() + + def test_auth_settings_requires_session_when_auth_enabled(self) -> None: + scope = { + "type": "http", + "method": "POST", + "path": "/api/v1/auth/settings", + "headers": [], + "query_string": b"", + "scheme": "http", + "client": ("127.0.0.1", 1234), + "server": ("testserver", 80), + "root_path": "", + } + request = Request(scope) + middleware = AuthMiddleware(app=MagicMock()) + + with patch("api.middlewares.auth.is_auth_enabled", return_value=True): + response = asyncio.run(middleware.dispatch(request, AsyncMock(return_value=Response(status_code=200)))) + + self.assertEqual(response.status_code, 401) + + def test_auth_settings_is_reachable_when_auth_disabled(self) -> None: + scope = { + "type": "http", + "method": "POST", + "path": "/api/v1/auth/settings", + "headers": [], + "query_string": b"", + "scheme": "http", + "client": ("127.0.0.1", 1234), + "server": ("testserver", 80), + "root_path": "", + } + request = Request(scope) + middleware = AuthMiddleware(app=MagicMock()) + next_response = Response(status_code=200) + call_next = AsyncMock(return_value=next_response) + + with patch("api.middlewares.auth.is_auth_enabled", return_value=False): + response = asyncio.run(middleware.dispatch(request, call_next)) + + self.assertEqual(response.status_code, 200) + call_next.assert_awaited_once() + + def test_auth_settings_enable_sets_initial_password_and_logs_in(self) -> None: + self.env_path.write_text( + "STOCK_LIST=600519\nGEMINI_API_KEY=test\nADMIN_AUTH_ENABLED=false\n", + encoding="utf-8", + ) + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.refresh_auth_state() + + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest( + authEnabled=True, + password="initpass123", + passwordConfirm="initpass123", + ), + ) + ) + + self.assertEqual(response.status_code, 200) + self.assertIn(b'"authEnabled":true', response.body) + self.assertIn(b'"loggedIn":true', response.body) + self.assertIn(b'"passwordSet":true', response.body) + self.assertIn("dsa_session=", response.headers["set-cookie"]) + self.assertIn("ADMIN_AUTH_ENABLED=true", self.env_path.read_text(encoding="utf-8")) + + def test_auth_settings_enable_requires_password_when_missing(self) -> None: + self.env_path.write_text( + "STOCK_LIST=600519\nGEMINI_API_KEY=test\nADMIN_AUTH_ENABLED=false\n", + encoding="utf-8", + ) + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.refresh_auth_state() + + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=True), + ) + ) + + self.assertEqual(response.status_code, 400) + self.assertIn(b'"error":"password_required"', response.body) + + def test_auth_settings_rechecks_password_before_initial_write(self) -> None: + self.env_path.write_text( + "STOCK_LIST=600519\nGEMINI_API_KEY=test\nADMIN_AUTH_ENABLED=false\n", + encoding="utf-8", + ) + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.refresh_auth_state() + + with patch.object( + auth_endpoint, + "has_stored_password", + side_effect=[False, True], + ) as has_password_mock: + with patch.object(auth_endpoint, "set_initial_password") as set_password_mock: + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest( + authEnabled=True, + password="initpass123", + passwordConfirm="initpass123", + ), + ) + ) + + self.assertEqual(has_password_mock.call_count, 2) + set_password_mock.assert_not_called() + self.assertEqual(response.status_code, 400) + self.assertIn(b'"error":"password_already_set"', response.body) + + def test_auth_settings_disable_clears_cookie_and_hides_password_state(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False, currentPassword="passwd6"), + ) + ) + + self.assertEqual(response.status_code, 200) + self.assertIn(b'"authEnabled":false', response.body) + self.assertIn(b'"loggedIn":false', response.body) + self.assertIn(b'"passwordSet":false', response.body) + self.assertIn("ADMIN_AUTH_ENABLED=false", self.env_path.read_text(encoding="utf-8")) + self.assertIn("dsa_session=", response.headers["set-cookie"]) + + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + status_response = asyncio.run(auth_endpoint.auth_status(self._build_request())) + self.assertFalse(status_response["authEnabled"]) + self.assertFalse(status_response["passwordSet"]) + + def test_auth_settings_disable_requires_current_password_when_auth_enabled(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False), + ) + ) + + self.assertEqual(response.status_code, 400) + self.assertIn(b'"error":"current_required"', response.body) + self.assertIn("ADMIN_AUTH_ENABLED=true", self.env_path.read_text(encoding="utf-8")) + + def test_auth_settings_toggle_fails_when_secret_rotation_fails(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + with patch.object(auth_endpoint, "rotate_session_secret", return_value=False): + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False, currentPassword="passwd6"), + ) + ) + + self.assertEqual(response.status_code, 500) + self.assertIn(b'"error":"internal_error"', response.body) + self.assertIn("ADMIN_AUTH_ENABLED=true", self.env_path.read_text(encoding="utf-8")) + + def test_auth_settings_enable_with_existing_password_reuses_stored_password(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + disable_response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False, currentPassword="passwd6"), + ) + ) + self.assertEqual(disable_response.status_code, 200) + + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + enable_response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=True, currentPassword="passwd6"), + ) + ) + + self.assertEqual(enable_response.status_code, 200) + self.assertIn(b'"authEnabled":true', enable_response.body) + self.assertIn(b'"passwordSet":true', enable_response.body) + self.assertIn(b'"loggedIn":true', enable_response.body) + self.assertIn("dsa_session=", enable_response.headers["set-cookie"]) + + def test_auth_settings_enable_with_existing_password_requires_current_password(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + disable_response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False, currentPassword="passwd6"), + ) + ) + self.assertEqual(disable_response.status_code, 200) + + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=True), + ) + ) + + self.assertEqual(response.status_code, 400) + self.assertIn(b'"error":"current_required"', response.body) + self.assertIn("ADMIN_AUTH_ENABLED=false", self.env_path.read_text(encoding="utf-8")) + + def test_auth_settings_enable_with_existing_password_rejects_wrong_current_password(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + disable_response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False, currentPassword="passwd6"), + ) + ) + self.assertEqual(disable_response.status_code, 200) + + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=True, currentPassword="wrongpass"), + ) + ) + + self.assertEqual(response.status_code, 401) + self.assertIn(b'"error":"invalid_password"', response.body) + self.assertIn("ADMIN_AUTH_ENABLED=false", self.env_path.read_text(encoding="utf-8")) + + def test_auth_settings_enable_rolls_back_when_session_creation_fails(self) -> None: + self.env_path.write_text( + "STOCK_LIST=600519\nGEMINI_API_KEY=test\nADMIN_AUTH_ENABLED=false\n", + encoding="utf-8", + ) + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.refresh_auth_state() + with patch.object(auth_endpoint, "create_session", return_value=""): + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest( + authEnabled=True, + password="initpass123", + passwordConfirm="initpass123", + ), + ) + ) + + self.assertEqual(response.status_code, 500) + self.assertIn(b'"error":"internal_error"', response.body) + self.assertIn("ADMIN_AUTH_ENABLED=false", self.env_path.read_text(encoding="utf-8")) + + def test_auth_settings_rejects_overwriting_existing_password(self) -> None: + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + auth.set_initial_password("passwd6") + disable_response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest(authEnabled=False, currentPassword="passwd6"), + ) + ) + self.assertEqual(disable_response.status_code, 200) + + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(), + auth_endpoint.AuthSettingsRequest( + authEnabled=True, + password="newpass123", + passwordConfirm="newpass123", + ), + ) + ) + + self.assertEqual(response.status_code, 400) + self.assertIn(b'"error":"password_already_set"', response.body) + + def test_auth_settings_enable_requires_valid_session_cookie_against_toctou(self) -> None: + """Verify fix for P1 vulnerability: passing authEnabled=True without currentPassword + must be rejected if the caller lacks a cryptographically valid session, even if + is_auth_enabled() evaluates to True during handler execution (TOCTOU race condition). + """ + self.env_path.write_text( + "STOCK_LIST=600519\nGEMINI_API_KEY=test\nADMIN_AUTH_ENABLED=false\n", + encoding="utf-8", + ) + with patch.object(auth, "_is_auth_enabled_from_env", side_effect=self._read_auth_enabled_from_env): + # 1. Setup an existing password, auth is currently disabled + auth.set_initial_password("passwd6") + + # 2. Simulate the race condition: + # The middleware let the request through because auth was supposedly False. + # But just before the handler runs, another thread enables auth. + self.env_path.write_text( + "STOCK_LIST=600519\nGEMINI_API_KEY=test\nADMIN_AUTH_ENABLED=true\n", + encoding="utf-8", + ) + auth.refresh_auth_state() # simulate the flip to True + + # 3. The attacker tries to re-enable auth without a password or valid cookie + response = asyncio.run( + auth_endpoint.auth_update_settings( + self._build_request(cookies={"dsa_session": "invalid"}), + auth_endpoint.AuthSettingsRequest(authEnabled=True), + ) + ) + + # 4. Must be rejected because they lack a valid session + NO current_password + self.assertEqual(response.status_code, 400) + self.assertIn(b'"error":"current_required"', response.body) if __name__ == "__main__":