fix(api): enforce one access boundary for HTTP and WebSocket

api/main.py

@@ -33,6 +33,7 @@
 from fastapi.responses import FileResponse
 
 from .routers import crawler_router, data_router, websocket_router
+from .security import APIAccessMiddleware
 
 # Project root directory (used for running subprocesses like uv run main.py)
 PROJECT_ROOT = Path(__file__).parent.parent
@@ -59,6 +60,7 @@
     allow_methods=["*"],
     allow_headers=["*"],
 )
+app.add_middleware(APIAccessMiddleware)
 
 # Register routers
 app.include_router(crawler_router, prefix="/api")
@@ -202,4 +204,4 @@ async def get_config_options():
 
 
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8080)
+    uvicorn.run(app, host="127.0.0.1", port=8080, proxy_headers=False)

api/security.py

@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2025 relakkes@gmail.com
+#
+# This file is part of MediaCrawler project.
+# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/api\security.py
+# GitHub: https://github.com/NanmiCoder
+# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
+#
+# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
+# 1. 不得用于任何商业用途。
+# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
+# 3. 不得进行大规模爬取或对平台造成运营干扰。
+# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
+# 5. 不得用于任何非法或不当的用途。
+#
+# 详细许可条款请参阅项目根目录下的LICENSE文件。
+# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
+
+"""Access boundary shared by HTTP endpoints and WebSocket handshakes."""
+
+import ipaddress
+import os
+import secrets
+from urllib.parse import urlsplit
+
+from starlette.datastructures import Headers
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+
+LOCAL_DEV_ORIGINS = {
+    "http://localhost:5173", "http://localhost:3000",
+    "http://127.0.0.1:5173", "http://127.0.0.1:3000",
+}
+
+
+def _is_loopback(host: str) -> bool:
+    try:
+        return ipaddress.ip_address(host).is_loopback
+    except ValueError:
+        return False
+
+
+class APIAccessMiddleware:
+    """Local-only by default; setting a token requires it for every API client."""
+
+    def __init__(self, app: ASGIApp):
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        if scope["type"] not in {"http", "websocket"} or not scope["path"].startswith("/api/"):
+            await self.app(scope, receive, send)
+            return
+
+        headers = Headers(scope=scope)
+        token = os.environ.get("MEDIACRAWLER_API_TOKEN", "")
+        status_code = 403
+        if token:
+            scheme, _, credential = headers.get("authorization", "").partition(" ")
+            allowed = scheme.lower() == "bearer" and secrets.compare_digest(
+                credential.encode(), token.encode()
+            )
+            status_code = 401
+        else:
+            peer = (scope.get("client") or ("", 0))[0]
+            try:
+                host = urlsplit("//" + headers.get("host", "")).hostname or ""
+            except ValueError:
+                host = ""
+            allowed = _is_loopback(peer) and (host == "localhost" or _is_loopback(host))
+            origin = headers.get("origin")
+            if origin:
+                scheme = "https" if scope["scheme"] in {"https", "wss"} else "http"
+                same_origin = f"{scheme}://{headers.get('host', '')}"
+                allowed = allowed and origin in LOCAL_DEV_ORIGINS | {same_origin}
+
+        if allowed:
+            await self.app(scope, receive, send)
+        elif scope["type"] == "websocket":
+            await send({"type": "websocket.close", "code": 1008})
+        else:
+            response = JSONResponse(
+                {"detail": "API access requires a trusted local client or configured Bearer token"},
+                status_code=status_code,
+                headers={"WWW-Authenticate": "Bearer"} if token else None,
+            )
+            await response(scope, receive, send)

docs/api-access.md

@@ -0,0 +1,16 @@
+# WebUI API 访问边界
+
+默认运行 `uv run python -m api.main`，仅监听 `127.0.0.1:8080`。
+浏览器通过 `http://localhost:8080` 或 `http://127.0.0.1:8080` 使用 WebUI，无需额外配置。
+API 校验回环客户端、Host 与浏览器 Origin；本机 Vite 3000/5173 端口仍可访问。
+HTTP 与 WebSocket 使用同一个访问边界，CORS 不是身份验证。
+
+需要远程 API 时，在服务进程环境中设置随机的 `MEDIACRAWLER_API_TOKEN`，
+并在每个 HTTP 请求和 WebSocket 握手中发送 `Authorization: Bearer <token>`。
+设置 token 后，本机客户端也必须携带它。请在可信 TLS 反向代理后暴露服务，
+不要把 token 放入 URL 查询参数或源码。当前 WebUI 没有 token 输入功能；
+远程使用浏览器界面可通过 SSH 本地端口转发访问默认回环服务。
+
+命令行自行使用 `uvicorn --host ...` 不会移除 API 的访问检查。反向代理必须可靠地
+覆盖转发头；不能通过伪造 Host 或 X-Forwarded-For 将远程匿名请求当成本机请求。
+没有 token 时不要把一个把全部用户代理为回环客户端的服务暴露到公网。

tests/test_api_access.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2025 relakkes@gmail.com
+#
+# This file is part of MediaCrawler project.
+# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/tests/test_api_access.py
+# GitHub: https://github.com/NanmiCoder
+# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
+#
+# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
+# 1. 不得用于任何商业用途。
+# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
+# 3. 不得进行大规模爬取或对平台造成运营干扰。
+# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
+# 5. 不得用于任何非法或不当的用途。
+#
+# 详细许可条款请参阅项目根目录下的LICENSE文件。
+# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
+
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from starlette.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+from api.main import app
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("path", ["/api/crawler/status", "/api/crawler/logs", "/api/data/files", "/api/env/check"])
+async def test_remote_api_requests_require_authentication(monkeypatch, path):
+    monkeypatch.delenv("MEDIACRAWLER_API_TOKEN", raising=False)
+    transport = httpx.ASGITransport(app=app, client=("198.51.100.8", 1234))
+    async with httpx.AsyncClient(transport=transport, base_url="http://crawler.example") as client:
+        response = await client.get(path)
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_remote_token_and_local_same_origin_access(monkeypatch):
+    monkeypatch.setenv("MEDIACRAWLER_API_TOKEN", "test-only-token")
+    transport = httpx.ASGITransport(app=app, client=("198.51.100.8", 1234))
+    async with httpx.AsyncClient(transport=transport, base_url="http://crawler.example") as client:
+        assert (await client.get("/api/crawler/status")).status_code == 401
+        response = await client.get("/api/crawler/status", headers={"Authorization": "Bearer test-only-token"})
+        assert response.status_code == 200
+    monkeypatch.delenv("MEDIACRAWLER_API_TOKEN")
+    transport = httpx.ASGITransport(app=app, client=("127.0.0.1", 1234))
+    async with httpx.AsyncClient(transport=transport, base_url="http://localhost:8080") as client:
+        assert (await client.get("/api/crawler/status", headers={"Origin": "http://localhost:8080"})).status_code == 200
+        assert (await client.get("/api/crawler/status", headers={"Origin": "https://untrusted.example"})).status_code == 403
+        assert (await client.get("/api/crawler/status", headers={"Host": "rebound.example"})).status_code == 403
+
+
+@pytest.mark.parametrize("path", ["/api/ws/logs", "/api/ws/status"])
+def test_cross_origin_websocket_is_rejected(monkeypatch, path):
+    monkeypatch.delenv("MEDIACRAWLER_API_TOKEN", raising=False)
+    async def local_client(scope, receive, send):
+        scope["client"] = ("127.0.0.1", 1234)
+        await app(scope, receive, send)
+
+    with TestClient(local_client, base_url="http://localhost") as client:
+        with client.websocket_connect("ws://localhost" + path, headers={"Origin": "http://localhost"}):
+            pass
+        with pytest.raises(WebSocketDisconnect):
+            with client.websocket_connect("ws://localhost" + path, headers={"Origin": "https://untrusted.example"}):
+                pytest.fail("Untrusted WebSocket was accepted")
+
+
+@pytest.mark.parametrize("path", ["/api/ws/logs", "/api/ws/status"])
+def test_websocket_token_policy_matches_http(monkeypatch, path):
+    monkeypatch.setenv("MEDIACRAWLER_API_TOKEN", "test-only-token")
+    with TestClient(app) as client:
+        with pytest.raises(WebSocketDisconnect):
+            with client.websocket_connect("ws://localhost" + path, headers={"Authorization": "Bearer wrong-token"}):
+                pytest.fail("Invalid token was accepted")
+        with client.websocket_connect("ws://localhost" + path, headers={"Authorization": "Bearer test-only-token"}):
+            pass
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("host", ["[", "[not-ip]"])
+async def test_malformed_host_is_denied(monkeypatch, host):
+    monkeypatch.delenv("MEDIACRAWLER_API_TOKEN", raising=False)
+    transport = httpx.ASGITransport(app=app, client=("127.0.0.1", 1234))
+    async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
+        response = await client.get("/api/crawler/status", headers={"Host": host})
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_untrusted_request_cannot_start_or_stop_a_process(monkeypatch):
+    from api.services import crawler_manager
+    monkeypatch.delenv("MEDIACRAWLER_API_TOKEN", raising=False)
+    start = AsyncMock()
+    stop = AsyncMock()
+    monkeypatch.setattr(crawler_manager, "start", start)
+    monkeypatch.setattr(crawler_manager, "stop", stop)
+    transport = httpx.ASGITransport(app=app, client=("198.51.100.8", 1234))
+    async with httpx.AsyncClient(transport=transport, base_url="http://localhost:8080") as client:
+        assert (await client.post("/api/crawler/start", json={"platform": "xhs"})).status_code == 403
+        assert (await client.post("/api/crawler/stop")).status_code == 403
+    start.assert_not_awaited()
+    stop.assert_not_awaited()

tests/test_api_limits.py

@@ -1,4 +1,21 @@
 # -*- coding: utf-8 -*-
+# Copyright (c) 2025 relakkes@gmail.com
+#
+# This file is part of MediaCrawler project.
+# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/tests\test_api_limits.py
+# GitHub: https://github.com/NanmiCoder
+# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
+#
+# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
+# 1. 不得用于任何商业用途。
+# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
+# 3. 不得进行大规模爬取或对平台造成运营干扰。
+# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
+# 5. 不得用于任何非法或不当的用途。
+#
+# 详细许可条款请参阅项目根目录下的LICENSE文件。
+# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
+
 import pytest
 import config
 from unittest.mock import AsyncMock, patch
@@ -8,6 +25,11 @@
 from api.services.crawler_manager import CrawlerManager
 from api.main import app
 
+
+@pytest.fixture(autouse=True)
+def api_authentication(monkeypatch):
+    monkeypatch.setenv("MEDIACRAWLER_API_TOKEN", "test-api-limits")
+
 @pytest.mark.asyncio
 async def test_cmd_arg_crawler_max_notes_count():
     # Store original values
@@ -63,7 +85,7 @@ def test_crawler_manager_build_command():
     assert cmd2[idx_comments + 1] == "5"
 
 def test_api_start_crawler_with_limits():
-    client = TestClient(app)
+    client = TestClient(app, headers={"Authorization": "Bearer test-api-limits"})
 
     with patch("api.routers.crawler.crawler_manager.start", new_callable=AsyncMock) as mock_start:
         mock_start.return_value = True
@@ -88,7 +110,7 @@ def test_api_start_crawler_with_limits():
         assert called_request.max_comments_count == 5
 
 def test_api_start_crawler_without_limits():
-    client = TestClient(app)
+    client = TestClient(app, headers={"Authorization": "Bearer test-api-limits"})
 
     with patch("api.routers.crawler.crawler_manager.start", new_callable=AsyncMock) as mock_start:
         mock_start.return_value = True
@@ -121,7 +143,7 @@ def test_api_start_crawler_without_limits():
     ],
 )
 def test_api_rejects_invalid_limits(field_name, value):
-    client = TestClient(app)
+    client = TestClient(app, headers={"Authorization": "Bearer test-api-limits"})
     payload = {
         "platform": "xhs",
         "login_type": "qrcode",