#602 [PR 3] 认证安全与初始化流程 (Auth & Security) (#639)

* 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

@@ -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)。
 
 ### 智能导入
 

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
 
 

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<AuthStatusResponse> {
+    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<AuthStatusResponse>('/api/v1/auth/settings', body);
+    return data;
+  },
+
   async login(password: string, passwordConfirm?: string): Promise<void> {
     const body: { password: string; passwordConfirm?: string } = { password };
     if (passwordConfirm !== undefined) {

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_<NAME>_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

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:

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

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

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__":