Codex/harden public api auth

.env.example

@@ -496,6 +496,8 @@ WEBUI_AUTO_BUILD=true
 # 设为 true 启用密码保护；首次访问时在网页设置初始密码，可在「系统设置 > 修改密码」中修改
 # 忘记密码可在服务器执行: python -m src.auth reset_password
 ADMIN_AUTH_ENABLED=false
+# 监听 0.0.0.0/:: 且 ADMIN_AUTH_ENABLED=false 时默认拒绝启动；仅可信内网临时部署才设为 true
+FINAGENT_ALLOW_INSECURE_PUBLIC_BIND=false
 # ADMIN_SESSION_MAX_AGE_HOURS=24  # Session 有效期（小时）
 
 # ===========================================

api/middlewares/error_handler.py

@@ -20,6 +20,12 @@
 
 logger = logging.getLogger(__name__)
 
+INTERNAL_ERROR_RESPONSE = {
+    "error": "internal_error",
+    "message": "Internal server error",
+    "detail": None,
+}
+
 
 class ErrorHandlerMiddleware(BaseHTTPMiddleware):
     """
@@ -59,11 +65,7 @@ async def dispatch(
             # 返回统一格式的错误响应
             return JSONResponse(
                 status_code=500,
-                content={
-                    "error": "internal_error",
-                    "message": "服务器内部错误，请稍后重试",
-                    "detail": str(e) if logger.isEnabledFor(logging.DEBUG) else None
-                }
+                content=INTERNAL_ERROR_RESPONSE,
             )
 
 
@@ -82,6 +84,19 @@ def add_error_handlers(app) -> None:
     @app.exception_handler(HTTPException)
     async def http_exception_handler(request: Request, exc: HTTPException):
         """处理 HTTP 异常"""
+        if exc.status_code >= 500:
+            logger.error(
+                "HTTPException %s at %s %s: %s",
+                exc.status_code,
+                request.method,
+                request.url.path,
+                exc.detail,
+            )
+            return JSONResponse(
+                status_code=exc.status_code,
+                content=INTERNAL_ERROR_RESPONSE,
+            )
+
         # 如果 detail 已经是 ErrorResponse 格式的 dict，直接使用
         if isinstance(exc.detail, dict) and "error" in exc.detail and "message" in exc.detail:
             return JSONResponse(
@@ -120,9 +135,5 @@ async def general_exception_handler(request: Request, exc: Exception):
         )
         return JSONResponse(
             status_code=500,
-            content={
-                "error": "internal_error",
-                "message": "服务器内部错误",
-                "detail": None
-            }
+            content=INTERNAL_ERROR_RESPONSE,
         )

docker/docker-compose.yml

@@ -34,6 +34,8 @@ x-common: &common
 
     # Web/API service bind address (must be 0.0.0.0 inside container)
     - WEBUI_HOST=0.0.0.0
+    # The server service publishes API_PORT to the host, so auth defaults on.
+    - ADMIN_AUTH_ENABLED=${ADMIN_AUTH_ENABLED:-true}
     # API_PORT 从 .env 文件读取，无需在此硬编码
 
     # 代理设置（如果需要）

docs/CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
+- [修复] 公开监听 Web/API 时默认要求开启登录认证，并统一清洗 500 错误响应，避免公网部署误暴露接口或泄露内部异常详情
 - [修复] 补回 GitHub Actions 每日股票分析工作流，恢复 fork 后手动/定时运行能力（fixes #31）。
 - [新功能] 打板策略（screen_board_play）：通过 AKShare 获取龙虎榜、涨停池、连板天梯、涨停概念数据，自动筛选 1进2/2进3/3进4 晋级股，输出板块策略 + 连板预测 + 个股买入建议
 - [改进] 打板策略结果在 UI 中按连板层级（1进2 / 2进3 / 3进4）分组展示，不同层级使用不同颜色区分

docs/deploy-webui-cloud.md

@@ -304,6 +304,12 @@ ADMIN_AUTH_ENABLED=true
 
 重启服务后，第一次访问网页时会要求设置初始密码。设置完成后，每次打开设置页面都需要输入密码，可以防止 API Key 等敏感配置被他人看到。
 
+从本版本起，如果服务监听 `0.0.0.0` 或 `::` 且 `ADMIN_AUTH_ENABLED=false`，FinAgent 会拒绝启动，避免把分析、历史记录和配置接口误暴露到公网。仅在受信任内网或临时排障场景下，才可以显式设置：
+
+```env
+FINAGENT_ALLOW_INSECURE_PUBLIC_BIND=true
+```
+
 > 如果忘了密码，可以在服务器上执行：`python -m src.auth reset_password`
 
 ---

main.py

@@ -21,6 +21,7 @@
 - 效率优先：关注筹码集中度好的股票
 - 买点偏好：缩量回踩 MA5/MA10 支撑
 """
+import ipaddress
 import os
 from pathlib import Path
 from typing import Dict, Optional
@@ -606,6 +607,8 @@ def start_api_server(host: str, port: int, config: Config) -> None:
     import threading
     import uvicorn
 
+    enforce_public_bind_auth(host)
+
     def run_server():
         level_name = (config.log_level or "INFO").lower()
         uvicorn.run(
@@ -627,6 +630,41 @@ def _is_truthy_env(var_name: str, default: str = "true") -> bool:
     return value not in {"0", "false", "no", "off"}
 
 
+def _is_public_bind_address(host: str) -> bool:
+    normalized = (host or "").strip().lower().strip("[]")
+    if normalized in {"", "*"}:
+        return True
+
+    try:
+        return ipaddress.ip_address(normalized).is_unspecified
+    except ValueError:
+        return False
+
+
+def enforce_public_bind_auth(host: str) -> None:
+    """Refuse public Web/API binds unless auth is enabled or explicitly waived."""
+    if not _is_public_bind_address(host):
+        return
+
+    from src.auth import is_auth_enabled
+
+    if is_auth_enabled():
+        return
+
+    if _is_truthy_env("FINAGENT_ALLOW_INSECURE_PUBLIC_BIND", "false"):
+        logger.warning(
+            "Starting public Web/API listener with ADMIN_AUTH_ENABLED=false because "
+            "FINAGENT_ALLOW_INSECURE_PUBLIC_BIND=true is set."
+        )
+        return
+
+    raise RuntimeError(
+        "Refusing to start public Web/API listener with ADMIN_AUTH_ENABLED=false. "
+        "Set ADMIN_AUTH_ENABLED=true before binding to 0.0.0.0/::, or set "
+        "FINAGENT_ALLOW_INSECURE_PUBLIC_BIND=true only for a trusted private network."
+    )
+
+
 def start_bot_stream_clients(config: Config) -> None:
     """Start bot stream clients when enabled in config."""
     # 启动钉钉 Stream 客户端
@@ -783,6 +821,7 @@ def main() -> int:
             bot_clients_started = True
         except Exception as e:
             logger.error(f"启动 FastAPI 服务失败: {e}")
+            return 1
 
     if bot_clients_started:
         start_bot_stream_clients(config)

server.py

@@ -45,6 +45,9 @@
 
 if __name__ == "__main__":
     import uvicorn
+    from main import enforce_public_bind_auth
+
+    enforce_public_bind_auth("0.0.0.0")
 
     uvicorn.run(
         "server:app",

tests/test_error_handler.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+"""Regression tests for API error response sanitization."""
+
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from api.middlewares.error_handler import add_error_handlers
+
+
+def test_http_500_detail_does_not_leak_exception_text():
+    app = FastAPI()
+    add_error_handlers(app)
+
+    @app.get("/boom")
+    def boom():
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error": "internal_error",
+                "message": "secret path C:/prod/.env token=sk-test",
+            },
+        )
+
+    response = TestClient(app).get("/boom")
+
+    assert response.status_code == 500
+    payload = response.json()
+    assert payload["error"] == "internal_error"
+    assert payload.get("detail") is None
+    assert "secret path" not in response.text
+    assert "sk-test" not in response.text

tests/test_main_schedule_mode.py

@@ -389,6 +389,34 @@ def test_bootstrap_logging_failure_does_not_block_startup(self) -> None:
         self.assertEqual(exit_code, 0)
         run_mock.assert_called_once()
 
+    def test_public_bind_requires_auth_when_insecure_override_absent(self) -> None:
+        with patch.dict(os.environ, {"FINAGENT_ALLOW_INSECURE_PUBLIC_BIND": "false"}, clear=False), \
+             patch("src.auth.is_auth_enabled", return_value=False):
+            with self.assertRaisesRegex(RuntimeError, "ADMIN_AUTH_ENABLED=true"):
+                main.enforce_public_bind_auth("0.0.0.0")
+
+    def test_public_bind_allows_loopback_without_auth(self) -> None:
+        with patch("src.auth.is_auth_enabled", side_effect=AssertionError("loopback should not check auth")):
+            main.enforce_public_bind_auth("127.0.0.1")
+
+    def test_public_bind_allows_explicit_insecure_override(self) -> None:
+        with patch.dict(os.environ, {"FINAGENT_ALLOW_INSECURE_PUBLIC_BIND": "true"}, clear=False), \
+             patch("src.auth.is_auth_enabled", return_value=False):
+            main.enforce_public_bind_auth("0.0.0.0")
+
+    def test_main_returns_nonzero_when_server_startup_guard_fails(self) -> None:
+        args = self._make_args(serve_only=True, host="0.0.0.0")
+        config = self._make_config()
+
+        with patch("main.parse_arguments", return_value=args), \
+             patch("main.get_config", return_value=config), \
+             patch("main.setup_logging"), \
+             patch("main.prepare_webui_frontend_assets", return_value=True), \
+             patch("main.start_api_server", side_effect=RuntimeError("guard failed")):
+            exit_code = main.main()
+
+        self.assertEqual(exit_code, 1)
+
     def test_run_full_analysis_import_failure_propagates(self) -> None:
         """P1: import failures in run_full_analysis must propagate, not be swallowed."""
         args = self._make_args()