diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index a0719720d..1d7669b8c 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -4,15 +4,14 @@ import asyncio import mimetypes -from collections.abc import AsyncIterator from typing import Annotated, Any, Dict, List, Mapping, Optional, cast from urllib.parse import urlsplit -import aiofiles from anyio import Path as AsyncPath -from fastapi import Depends, Header, HTTPException, Query, Response, Security +from fastapi import Depends, Header, HTTPException, Query, Request, Response, Security from starlette import status -from starlette.responses import StreamingResponse +from starlette.responses import FileResponse +from starlette.staticfiles import StaticFiles from app.adapters.web.security.access import ( resource_token_cookie, @@ -761,7 +760,7 @@ def reset_plugin( "/file/{plugin_id}/{filepath:path}", summary="获取插件静态文件", response_model=None, - response_class=StreamingResponse, + response_class=FileResponse, responses={ 200: { "description": "插件静态资源", @@ -770,23 +769,31 @@ def reset_plugin( "application/javascript": {"schema": {"type": "string"}}, "text/css": {"schema": {"type": "string"}}, }, - } + }, + 304: {"description": "插件静态资源未修改,复用浏览器缓存"}, }, ) async def plugin_static_file( plugin_id: str, filepath: str, + request: Request, _: None = Depends(_verify_plugin_static_file_access), -) -> StreamingResponse: +) -> Response: """ - 获取插件静态文件 + 在访问校验后返回插件文件,支持私有缓存及 ETag、修改时间条件请求。 + + 当前插件版本的入口可长期复用;其余文件每次校验, + 避免无版本入口或同名文件更新后继续使用旧内容。 """ # 基础安全检查 if ".." in filepath or ".." in plugin_id: logger.warning(f"Static File API: Path traversal attempt detected: {plugin_id}/{filepath}") raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden") - source_plugin_id = get_plugin_manager().get_plugin_source_id(plugin_id) + manager = get_plugin_manager() + # 联邦入口使用小写 URL,运行时注册表保留插件 ID 原有大小写。 + runtime_id = next((pid for pid in manager.get_plugin_ids() if pid.lower() == plugin_id.lower()), plugin_id) + source_plugin_id = manager.get_plugin_source_id(runtime_id) plugin_base_dir = ( AsyncPath(get_api_runtime_config_snapshot().root_path) / "app" / "plugins" / source_plugin_id.lower() ) @@ -818,31 +825,23 @@ async def plugin_static_file( elif suffix == ".css" and not guessed_response_type: response_type = "text/css" - try: - # 异步生成器函数,用于流式读取文件 - async def file_generator() -> AsyncIterator[bytes]: - """ - 分块异步读取插件文件供下载响应使用。 - """ - async with aiofiles.open(plugin_file_path, mode="rb") as file: - # 8KB 块大小 - while chunk := await file.read(8192): - yield chunk - - return StreamingResponse( - file_generator(), - media_type=response_type, - headers={ - "Content-Disposition": f"inline; filename={plugin_file_path.name}", - "Cache-Control": "no-cache, must-revalidate", - }, - ) - except Exception as e: - logger.error( - f"Error creating/sending StreamingResponse for {plugin_file_path}: {e}", - exc_info=True, - ) - raise HTTPException(status_code=500, detail="Internal Server Error") + version = request.query_params.get("v") + current_version = manager.get_plugin_attr(runtime_id, "plugin_version") + versioned = bool(version and current_version and version == str(current_version)) + cache_control = "private, max-age=31536000, immutable" if versioned else "private, no-cache" + response = FileResponse( + str(plugin_file_path), + stat_result=await plugin_file_path.stat(), + media_type=response_type, + headers={"Cache-Control": cache_control}, + filename=plugin_file_path.name, + content_disposition_type="inline", + ) + if StaticFiles().is_not_modified(response.headers, request.headers): + return Response(status_code=304, headers={ + key: response.headers[key] for key in ("cache-control", "etag", "last-modified") + }) + return response @router.get( diff --git a/app/application/plugin/rating.py b/app/application/plugin/rating.py index 5e97032b8..36d013230 100644 --- a/app/application/plugin/rating.py +++ b/app/application/plugin/rating.py @@ -4,9 +4,13 @@ from __future__ import annotations -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Hashable, Mapping, Sequence +from copy import deepcopy +from time import monotonic from typing import Any +from cachetools import TTLCache + StatisticLoader = Callable[[], Awaitable[Mapping[str, Any]]] RatingsLoader = Callable[[list[str] | None], Awaitable[Mapping[str, Any]]] RatingLoader = Callable[[str], Awaitable[Mapping[str, Any]]] @@ -32,29 +36,55 @@ class PluginRatingService: ratings: RatingsLoader, rating: RatingLoader, submit: RatingSubmitter, + cache_scope: Callable[[], Hashable] = lambda: None, ) -> None: """ - 保存安装状态读取和远程评分窄端口。 + 保存安装状态读取和远程评分窄端口,创建当前 lifespan 的有界缓存。 + + :param cache_scope: 当前中心端、实例身份和统计开关组成的缓存隔离标识 """ self._installed_plugins = installed_plugins self._statistic = statistic self._ratings = ratings self._rating = rating self._submit = submit + self._cache_scope = cache_scope + # 仅由主事件循环访问;随 lifespan 释放,不把实例评分持久化到共享缓存。 + self._metrics: TTLCache[tuple[Hashable, str, tuple[str, ...] | None], Mapping[str, Any]] = TTLCache( + maxsize=128, ttl=300, timer=monotonic + ) + self._revision = 0 + + async def _cached_query( + self, kind: str, plugin_ids: tuple[str, ...] | None, loader: StatisticLoader + ) -> Mapping[str, Any]: + """ + 复用五分钟内的指标结果,按中心端与实例身份隔离,失败空结果不缓存。 + + 评分提交期间发出的旧查询不得覆盖提交后的缓存;向调用方返回独立副本。 + """ + key = (self._cache_scope(), kind, plugin_ids) + cached = self._metrics.get(key) + if cached is not None: + return deepcopy(cached) + revision = self._revision + result = await loader() + if result and revision == self._revision: + self._metrics[key] = deepcopy(result) + return deepcopy(result) async def statistic(self) -> Mapping[str, Any]: """ - 返回插件安装统计。 + 返回插件安装统计,复用当前实例最近五分钟的非空结果。 """ - return await self._statistic() + return await self._cached_query("statistic", None, self._statistic) - async def ratings( - self, plugin_ids: list[str] | None - ) -> Mapping[str, Any]: + async def ratings(self, plugin_ids: list[str] | None) -> Mapping[str, Any]: """ - 批量返回插件评分快照。 + 批量返回插件评分快照,同一 ID 集合五分钟内不重复外呼。 """ - return await self._ratings(plugin_ids) + ids = tuple(sorted(set(plugin_ids))) if plugin_ids is not None else None + return await self._cached_query("ratings", ids, lambda: self._ratings(plugin_ids)) async def rating(self, plugin_id: str) -> Mapping[str, Any]: """ @@ -68,7 +98,11 @@ class PluginRatingService: """ if plugin_id not in self._installed_plugins(): raise PluginNotInstalledError(f"插件 {plugin_id} 未安装,无法评分") - return await self._submit(plugin_id, rating) + result = await self._submit(plugin_id, rating) + if result is not None: + self._revision += 1 + self._metrics.clear() + return result _rating_service: PluginRatingService | None = None diff --git a/app/startup/initializers/plugins.py b/app/startup/initializers/plugins.py index b842e6505..a921f436f 100644 --- a/app/startup/initializers/plugins.py +++ b/app/startup/initializers/plugins.py @@ -625,6 +625,11 @@ def configure_plugin_services() -> None: ratings=MoviePilotServerHelper.async_get_plugin_ratings, rating=MoviePilotServerHelper.async_get_plugin_rating, submit=MoviePilotServerHelper.async_submit_plugin_rating, + cache_scope=lambda: ( + get_runtime_setting("MP_SERVER_HOST"), + MoviePilotServerHelper.get_user_uid(), + bool(get_runtime_setting("PLUGIN_STATISTIC_SHARE")), + ), ) ) diff --git a/docs/mcp-api.md b/docs/mcp-api.md index d17e4a608..d9e2a9f4d 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -36,6 +36,16 @@ 缺失或超时返回 `{"url":null}`,不要求先安装插件。本地来源只返回显式声明的链接。 探测总时限为五秒,结果缓存五分钟;接口不返回文档正文,也不新增 Agent operation。 +### 插件静态资源与指标缓存 + +`GET /api/v1/plugin/file/{plugin_id}/{filepath}` 继续校验资源访问权限,返回私有浏览器缓存、 +`ETag` 和 `Last-Modified`;条件请求命中时返回无响应体的 `304`。`v` 参数与当前插件版本 +一致时允许长期缓存,其他资源每次重新校验,更新后不复用旧文件内容。插件资源仍不进入 +前端 Service Worker 的通用脚本缓存。 + +插件批量评分和安装统计的非空结果在当前进程缓存五分钟,按中心端、实例身份和统计开关 +隔离;提交评分成功会使指标缓存失效。单插件详情仍实时查询,中心端失败的空结果不缓存。 + ### 插件生命周期忙碌与超时 `plugin.runtime.status`(`GET /api/v1/plugin/runtime`)的可选 `active_operation` 返回当前 diff --git a/tests/test_plugin_endpoint.py b/tests/test_plugin_endpoint.py index 93c12cad2..608561e82 100644 --- a/tests/test_plugin_endpoint.py +++ b/tests/test_plugin_endpoint.py @@ -11,7 +11,9 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest +from fastapi import Request from starlette.responses import Response +from starlette.testclient import TestClient from app import schemas from app.api.endpoints import plugin as plugin_endpoint @@ -1363,17 +1365,12 @@ def test_virtual_instance_static_file_reads_from_source_directory(tmp_path, monk lambda: MagicMock(root_path=tmp_path), ) - response = asyncio.run(plugin_static_file("DemoPluginwork", "dist/remoteEntry.js", None)) + request = Request({"type": "http", "headers": [], "query_string": b""}) + response = asyncio.run(plugin_static_file("DemoPluginwork", "dist/remoteEntry.js", request, None)) - async def read_body() -> bytes: - """ - 读取流式响应的全部测试内容。 - """ - return b"".join([chunk async for chunk in response.body_iterator]) - - assert asyncio.run(read_body()) == b"export default 'shared'" + assert TestClient(response).get("/").content == b"export default 'shared'" assert response.media_type == "application/javascript" - assert response.headers["cache-control"] == "no-cache, must-revalidate" + assert response.headers["cache-control"] == "private, no-cache" plugin_manager.get_plugin_source_id.assert_called_once_with("DemoPluginwork") diff --git a/tests/test_plugin_rating.py b/tests/test_plugin_rating.py index b01fe693e..a16c9b51d 100644 --- a/tests/test_plugin_rating.py +++ b/tests/test_plugin_rating.py @@ -150,3 +150,67 @@ def test_plugin_rating_service_reset_isolates_lifespans(monkeypatch) -> None: with pytest.raises(RuntimeError, match="尚未完成初始化"): rating_module.get_plugin_rating_service() + + +@pytest.mark.asyncio +async def test_metrics_cache_expires_and_isolates_identity_and_callers(monkeypatch) -> None: + """ + 五分钟内同集合查询复用副本,缓存到期或中心端身份变化后重新读取。 + """ + clock = [0.0] + scope = ["instance-one"] + monkeypatch.setattr(rating_module, "monotonic", lambda: clock[0]) + ratings = AsyncMock(return_value={"Demo": {"average_rating": 4.0}}) + statistic = AsyncMock(return_value={"Demo": 20}) + service = PluginRatingService( + installed_plugins=lambda: ["Demo"], statistic=statistic, ratings=ratings, + rating=AsyncMock(), submit=AsyncMock(), cache_scope=lambda: scope[0], + ) + first = await service.ratings(["Demo", "Other"]) + first["Demo"]["average_rating"] = 1 + assert (await service.ratings(["Other", "Demo", "Demo"]))["Demo"]["average_rating"] == 4 + ratings.assert_awaited_once() + assert await service.statistic() == await service.statistic() == {"Demo": 20} + statistic.assert_awaited_once() + clock[0] = 301 + await service.ratings(["Demo", "Other"]) + await service.statistic() + assert ratings.await_count == statistic.await_count == 2 + scope[0] = "instance-two" + await service.ratings(["Demo", "Other"]) + await service.statistic() + assert ratings.await_count == statistic.await_count == 3 + + +@pytest.mark.asyncio +async def test_failed_metrics_are_not_cached_and_submit_invalidates_inflight_reads() -> None: + """ + 空失败结果立即允许重试,成功评分后在途旧结果不能再次写入缓存。 + """ + old_result = asyncio.Future() + ratings = AsyncMock(side_effect=[{}, {"Demo": {"average_rating": 1.0}}]) + service = PluginRatingService( + installed_plugins=lambda: ["Demo"], statistic=AsyncMock(return_value={}), + ratings=ratings, rating=AsyncMock(), submit=AsyncMock(return_value={"average_rating": 5.0}), + ) + assert await service.ratings(["Demo"]) == {} + assert await service.ratings(["Demo"]) == {"Demo": {"average_rating": 1.0}} + assert ratings.await_count == 2 + await service.submit("Demo", 5.0) + + async def pending_query(_ids): + """ + 保留一个评分提交前发出的查询,模拟中心端迟到的旧响应。 + """ + return await old_result + + ratings.side_effect = pending_query + pending = asyncio.create_task(service.ratings(["Demo"])) + await asyncio.sleep(0) + await service.submit("Demo", 5.0) + old_result.set_result({"Demo": {"average_rating": 1.0}}) + await pending + ratings.side_effect = None + ratings.return_value = {"Demo": {"average_rating": 5.0}} + assert await service.ratings(["Demo"]) == ratings.return_value + assert ratings.await_count == 4 diff --git a/tests/test_plugin_static.py b/tests/test_plugin_static.py new file mode 100644 index 000000000..c2de52d90 --- /dev/null +++ b/tests/test_plugin_static.py @@ -0,0 +1,101 @@ +""" +验证插件静态资源的浏览器缓存、条件请求和访问安全边界。 +""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from fastapi import FastAPI, HTTPException +from starlette.testclient import TestClient + +from app.api.endpoints import plugin + + +@pytest.fixture +def static_files(tmp_path, monkeypatch): + """ + 在临时插件目录与独立路由中提供资源,保留真实访问校验依赖。 + """ + root = tmp_path / "app/plugins/demo/dist" + root.mkdir(parents=True) + (root / "remoteEntry.js").write_text("export const version = 1", encoding="utf-8") + (root / "chunk.js").write_text("export const value = 1", encoding="utf-8") + manager = Mock() + manager.get_plugin_source_id.return_value = "Demo" + manager.get_plugin_ids.return_value = ["Demo"] + manager.get_plugin_attr.return_value = "1.0.0" + manager.get_plugin_auth_providers.return_value = [] + verify = Mock() + monkeypatch.setattr(plugin, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(plugin, "verify_resource_token", verify) + monkeypatch.setattr(plugin, "get_api_runtime_config_snapshot", lambda: SimpleNamespace(root_path=tmp_path)) + app = FastAPI() + app.include_router(plugin.router, prefix="/plugin") + return TestClient(app), root, manager, verify + + +def test_current_version_entry_is_private_and_cacheable(static_files): + """ + 只有当前版本参数可授予长期私有缓存,且返回文件校验头。 + """ + client, _, manager, _ = static_files + response = client.get("/plugin/file/demo/dist/remoteEntry.js?v=1.0.0") + assert response.status_code == 200 + assert response.content == b"export const version = 1" + assert response.headers["content-type"].startswith("application/javascript") + assert response.headers["cache-control"] == "private, max-age=31536000, immutable" + assert response.headers["etag"] + assert response.headers["last-modified"] + manager.get_plugin_attr.assert_called_once_with("Demo", "plugin_version") + manager.get_plugin_attr.return_value = "2.0.0" + assert client.get("/plugin/file/Demo/dist/remoteEntry.js?v=1.0.0").headers["cache-control"] == "private, no-cache" + assert "immutable" in client.get("/plugin/file/Demo/dist/remoteEntry.js?v=2.0.0").headers["cache-control"] + + +@pytest.mark.parametrize("validator", ["etag", "last-modified"]) +def test_unversioned_chunks_revalidate_without_resending_body(static_files, validator): + """ + 无版本 chunk 保留缓存内容,条件命中时仍校验权限并返回无响应体的 304。 + """ + client, _, _, verify = static_files + url = "/plugin/file/Demo/dist/chunk.js" + first = client.get(url) + header = "If-None-Match" if validator == "etag" else "If-Modified-Since" + response = client.get(url, headers={header: first.headers[validator]}) + assert response.status_code == 304 + assert response.content == b"" + assert response.headers["cache-control"] == "private, no-cache" + assert response.headers[validator] == first.headers[validator] + assert verify.call_count == 2 + verify.side_effect = HTTPException(status_code=403, detail="Forbidden") + assert client.get(url, headers={header: first.headers[validator]}).status_code == 403 + + +def test_changed_file_and_nonmatching_etag_return_new_body(static_files): + """ + 文件改变或 ETag 不匹配时,不被旧修改时间错误判为未修改。 + """ + client, root, _, _ = static_files + url = "/plugin/file/Demo/dist/chunk.js" + first = client.get(url) + (root / "chunk.js").write_text("export const value = 'updated'", encoding="utf-8") + response = client.get(url, headers={ + "If-None-Match": first.headers["etag"], + "If-Modified-Since": "Wed, 01 Jan 2099 00:00:00 GMT", + }) + assert response.status_code == 200 + assert response.headers["etag"] != first.headers["etag"] + assert response.content == b"export const value = 'updated'" + + +def test_missing_and_escaped_files_are_not_cacheable(static_files, tmp_path): + """ + 缺失资源与指向插件外的符号链接保持原有拒绝语义。 + """ + client, root, _, _ = static_files + outside = tmp_path / "outside.js" + outside.write_text("private", encoding="utf-8") + (root / "escaped.js").symlink_to(outside) + assert client.get("/plugin/file/Demo/dist/escaped.js").status_code == 403 + assert client.get("/plugin/file/Demo/dist/missing.js?v=1.0.0").status_code == 404