From 964fee1106d153cb9b7c5d3e9be97ea016b01316 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sun, 5 Jul 2026 09:28:07 +0800 Subject: [PATCH] fix(security): protect cookiecloud update uploads (#6053) --- app/api/servcookie.py | 23 +++++- app/core/config.py | 2 + tests/test_cookiecloud_routes.py | 117 +++++++++++++++++++++++++++++++ 3 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 tests/test_cookiecloud_routes.py diff --git a/app/api/servcookie.py b/app/api/servcookie.py index a2108f31..76e09af9 100644 --- a/app/api/servcookie.py +++ b/app/api/servcookie.py @@ -1,10 +1,11 @@ import gzip +import hmac import json from typing import Annotated, Callable, Any, Dict, Optional import aiofiles from anyio import Path as AsyncPath -from fastapi import APIRouter, Body, Depends, HTTPException, Path, Request, Response +from fastapi import APIRouter, Body, Depends, Header, HTTPException, Path, Request, Response from fastapi.responses import PlainTextResponse from fastapi.routing import APIRoute @@ -44,6 +45,24 @@ async def verify_server_enabled(): return True +async def verify_update_auth( + x_cookiecloud_auth: Annotated[ + Optional[str], Header(alias="X-CookieCloud-Auth") + ] = None, +): + """ + 校验CookieCloud上传接口的可选共享认证头。 + """ + expected_header = (settings.COOKIECLOUD_AUTH_HEADER or "").strip() + if not expected_header: + return True + + provided_header = (x_cookiecloud_auth or "").strip() + if not hmac.compare_digest(provided_header, expected_header): + raise HTTPException(status_code=403, detail="CookieCloud认证失败") + return True + + cookie_router = APIRouter( route_class=GzipRoute, tags=["servcookie"], @@ -61,7 +80,7 @@ async def post_root(): return "Hello MoviePilot! COOKIECLOUD API ROOT = /cookiecloud" -@cookie_router.post("/update") +@cookie_router.post("/update", dependencies=[Depends(verify_update_auth)]) async def update_cookie(req: schemas.CookieData): """ 上传Cookie数据 diff --git a/app/core/config.py b/app/core/config.py index 047c7018..3d7a5f6d 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -377,6 +377,8 @@ class ConfigModel(BaseModel): COOKIECLOUD_KEY: Optional[str] = None # CookieCloud端对端加密密码 COOKIECLOUD_PASSWORD: Optional[str] = None + # CookieCloud本地上传接口的X-CookieCloud-Auth期望值，留空表示不校验 + COOKIECLOUD_AUTH_HEADER: Optional[str] = None # CookieCloud同步间隔（分钟） COOKIECLOUD_INTERVAL: Optional[int] = 60 * 24 # CookieCloud同步黑名单，多个域名,分割 diff --git a/tests/test_cookiecloud_routes.py b/tests/test_cookiecloud_routes.py new file mode 100644 index 00000000..6a994cf4 --- /dev/null +++ b/tests/test_cookiecloud_routes.py @@ -0,0 +1,117 @@ +import json +from types import SimpleNamespace + +import httpx +import pytest +from fastapi import FastAPI + +from app.api import servcookie + +pytestmark = pytest.mark.anyio + + +@pytest.fixture() +def anyio_backend(): + return "asyncio" + + +@pytest.fixture() +def cookiecloud_app(tmp_path, monkeypatch): + settings = SimpleNamespace( + COOKIE_PATH=tmp_path, + COOKIECLOUD_ENABLE_LOCAL=True, + COOKIECLOUD_AUTH_HEADER=None, + ) + monkeypatch.setattr(servcookie, "settings", settings) + + app = FastAPI() + app.include_router(servcookie.cookie_router, prefix="/cookiecloud") + return app + + +def make_client(app): + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://testserver", + ) + + +async def test_update_rejects_when_local_cookiecloud_disabled(cookiecloud_app): + servcookie.settings.COOKIECLOUD_ENABLE_LOCAL = False + servcookie.settings.COOKIECLOUD_AUTH_HEADER = "secret" + + async with make_client(cookiecloud_app) as client: + response = await client.post( + "/cookiecloud/update", + json={"uuid": "abcde", "encrypted": "payload"}, + ) + + assert response.status_code == 400 + assert response.json()["detail"] == "本地CookieCloud服务器未启用" + + +@pytest.mark.parametrize("auth_header", [None, "", " "]) +async def test_update_allows_legacy_clients_when_auth_header_unconfigured( + cookiecloud_app, auth_header +): + servcookie.settings.COOKIECLOUD_AUTH_HEADER = auth_header + + async with make_client(cookiecloud_app) as client: + response = await client.post( + "/cookiecloud/update", + json={"uuid": "abcde", "encrypted": "payload"}, + ) + + assert response.status_code == 200 + assert response.json() == {"action": "done"} + assert json.loads((servcookie.settings.COOKIE_PATH / "abcde.json").read_text()) == { + "encrypted": "payload" + } + + +async def test_update_allows_matching_auth_header(cookiecloud_app): + servcookie.settings.COOKIECLOUD_AUTH_HEADER = " secret-token " + + async with make_client(cookiecloud_app) as client: + response = await client.post( + "/cookiecloud/update", + json={"uuid": "abcde", "encrypted": "payload"}, + headers={"X-CookieCloud-Auth": "secret-token"}, + ) + + assert response.status_code == 200 + assert response.json() == {"action": "done"} + + +@pytest.mark.parametrize("headers", [{}, {"X-CookieCloud-Auth": "wrong"}]) +async def test_update_rejects_missing_or_wrong_auth_header(cookiecloud_app, headers): + servcookie.settings.COOKIECLOUD_AUTH_HEADER = "secret-token" + + async with make_client(cookiecloud_app) as client: + response = await client.post( + "/cookiecloud/update", + json={"uuid": "abcde", "encrypted": "payload"}, + headers=headers, + ) + + assert response.status_code == 403 + assert response.json()["detail"] == "CookieCloud认证失败" + + +async def test_get_routes_do_not_require_auth_header(cookiecloud_app, monkeypatch): + servcookie.settings.COOKIECLOUD_AUTH_HEADER = "secret-token" + + async def load_encrypt_data(uuid): + assert uuid == "abcde" + return {"encrypted": "payload"} + + monkeypatch.setattr(servcookie, "load_encrypt_data", load_encrypt_data) + + async with make_client(cookiecloud_app) as client: + get_response = await client.get("/cookiecloud/get/abcde") + post_response = await client.post("/cookiecloud/get/abcde") + + assert get_response.status_code == 200 + assert get_response.json() == {"encrypted": "payload"} + assert post_response.status_code == 200 + assert post_response.json() == {"encrypted": "payload"}