diff --git a/package.v2.json b/package.v2.json index 2e465485..45d88033 100644 --- a/package.v2.json +++ b/package.v2.json @@ -294,11 +294,12 @@ "name": "Webhook消息推送", "description": "接收 Webhook 消息并推送到通知客户端。", "labels": "通知,工具", - "version": "1.0", + "version": "1.1", "icon": "https://raw.githubusercontent.com/InfinityPacer/MoviePilot-Plugins/main/icons/webhooknotify.png", "author": "InfinityPacer", "level": 1, "history": { + "v1.1": "新增 APIKEY 认证配置。", "v1.0": "新增通用 Webhook 通知入口。" }, "release": true diff --git a/plugins.v2/webhooknotify/README.md b/plugins.v2/webhooknotify/README.md index 3ebed9ff..c6212be8 100644 --- a/plugins.v2/webhooknotify/README.md +++ b/plugins.v2/webhooknotify/README.md @@ -7,12 +7,14 @@ ## 版本更新日志 +- v1.1 + - 新增 APIKEY 认证配置。 - v1.0 - 新增通用 Webhook 通知入口。 ## 功能概览 -- 提供受 MoviePilot 公共 `API_TOKEN` 保护的 `GET` 和 `POST` Webhook。 +- 提供支持独立 API Key 的 `GET` 和 `POST` Webhook，避免向外部系统提供 MoviePilot 公共 `API_TOKEN`。 - 接收外部请求中的 `title` 和 `body`，任意一项非空即可转发为所配置类型的消息。 - 支持选择 MoviePilot 消息类型，默认为“插件”，用于匹配通知渠道的接收设置。 - 复用 MoviePilot 通知历史和已配置的 WebPush、Telegram、微信等通知渠道。 @@ -22,8 +24,15 @@ | 方法 | 地址 | 说明 | | --- | --- | --- | -| `POST` | `/api/v1/plugin/WebhookNotify/webhook?token=API_TOKEN` | 从 JSON 请求体接收入站通知 | -| `GET` | `/api/v1/plugin/WebhookNotify/webhook?token=API_TOKEN&title=...&body=...` | 从查询参数接收入站通知 | +| `POST` | `/api/v1/plugin/WebhookNotify/webhook` | 从 JSON 请求体接收入站通知 | +| `GET` | `/api/v1/plugin/WebhookNotify/webhook?title=...&body=...` | 从查询参数接收入站通知 | + +认证信息支持以下任一传递方式： + +- 请求头：`X-API-KEY: ` +- 查询参数：`?apikey=` + +配置“APIKEY”后，`API_KEY` 必须使用插件中配置的 Key；留空时则使用 MoviePilot 公共 `API_TOKEN`。两种模式使用相同的请求头和查询参数名称。 `POST` 请求头使用 `Content-Type: application/json`，请求体格式如下： @@ -48,7 +57,8 @@ ```bash curl -X POST \ - "https://moviepilot.example.com/api/v1/plugin/WebhookNotify/webhook?token=API_TOKEN" \ + "https://moviepilot.example.com/api/v1/plugin/WebhookNotify/webhook" \ + -H "X-API-KEY: WEBHOOK_API_KEY" \ -H "Content-Type: application/json" \ -d '{"title":"路由故障","body":"主线路连续 3 次探测失败"}' ``` @@ -58,7 +68,7 @@ curl -X POST \ ```bash curl --get \ "https://moviepilot.example.com/api/v1/plugin/WebhookNotify/webhook" \ - --data-urlencode "token=API_TOKEN" \ + --data-urlencode "apikey=WEBHOOK_API_KEY" \ --data-urlencode "body=主线路连续 3 次探测失败" ``` @@ -66,8 +76,9 @@ curl --get \ | 配置项 | 标识 | 类型 | 默认值 | 说明 | 备注 | | --- | --- | --- | --- | --- | --- | -| 启用插件 | `enabled` | bool | `false` | 是否接收入站 Webhook | 使用主程序公共 `API_TOKEN`，无需重复配置 | +| 启用插件 | `enabled` | bool | `false` | 是否接收入站 Webhook | 停用时接口返回 `503` | | [消息类型](#cfg-notify_type) | `notify_type` | enum | `Plugin`（插件） | 设置通知分类和渠道过滤类型 | 调用方不能通过 Webhook 覆盖 | +| [APIKEY](#cfg-api_key) | `api_key` | string | 空 | 设置 Webhook 专用凭据 | 留空时使用主程序公共 `API_TOKEN` | ## 深入说明 @@ -78,23 +89,30 @@ MoviePilot 的通知渠道可分别选择接收哪些消息类型。Webhook 消 可选值为：`Download`（资源下载）、`Organize`（整理入库）、`Subscribe`（订阅）、`SiteMessage`（站点）、`MediaServer`（媒体服务器）、`Manual`（手动处理）、`Plugin`（插件）、`Agent`（智能体）和 `Other`（其它）。默认使用 `Plugin`（插件）；配置缺失或不是有效枚举值时同样按“插件”处理。 + +#### APIKEY（`api_key`） + +配置后，Webhook 只接受该独立 Key，不再接受 MoviePilot 公共 `API_TOKEN`。外部监控系统仍通过 `X-API-KEY` 请求头或 `apikey` 查询参数传递凭据。留空时由主程序使用同样的入口校验公共 `API_TOKEN`。 + +建议使用足够长的随机字符串，并优先通过 `X-API-KEY` 请求头传递，避免查询参数被反向代理访问日志记录。独立 Key 泄露后只需在插件配置中轮换，不影响 MoviePilot 的其他公共 API。 + ## 使用步骤 1. 在插件市场安装并启用 Webhook消息推送。 2. 选择消息类型，并确认 MoviePilot 中至少有一个已启用通知渠道允许接收该类型。 -3. 将调用方的 Webhook 地址配置为插件 API 地址，将 `API_TOKEN` 放在 `token` 查询参数中。 +3. 建议配置 APIKEY，并在调用方中通过 `X-API-KEY` 请求头传递；未配置时使用 MoviePilot 公共 `API_TOKEN`。 4. 使用 `GET` 查询参数或 `POST` JSON 发送通知，`title` 和 `body` 至少提供一项，确认客户端收到通知。 ## 注意事项 / 已知风险 -- `API_TOKEN` 具备 MoviePilot 公共集成权限，请只在受信任的监控系统中使用，并通过 HTTPS 传输。 +- 独立 API Key 和公共 `API_TOKEN` 都属于敏感凭据，请通过 HTTPS 传输；优先使用请求头，避免凭据进入 URL 日志。 - 插件只负责提交通知，不会为重复请求做去重；健康监测的失败阈值、恢复判定和重试策略由调用方负责。 - 消息是否实际发送仍受 MoviePilot 通知渠道配置和所选消息类型开关影响。 ## 故障排查 -- 插件请求日志位于 `/config/logs/plugins/webhooknotify.log`，记录请求方式、消息类型以及标题、正文是否存在，不记录 `API_TOKEN` 或消息内容。 -- 返回 `401`：检查 URL 中的 `token` 是否为当前 MoviePilot 公共 `API_TOKEN`。 +- 插件请求日志位于 `/config/logs/plugins/webhooknotify.log`，记录请求方式、消息类型以及标题、正文是否存在，不记录 API Key 或消息内容。 +- 返回 `401`：已配置独立 API Key 时检查插件 Key；未配置时检查 MoviePilot 公共 `API_TOKEN`。凭据应通过 `X-API-KEY` 或 `apikey` 传递。 - 返回 `422`：检查参数长度、JSON 格式，并确认 `title`、`body` 至少有一项为非空字符串。 - 返回 `503`：插件在 WebUI 中处于停用状态。 - 接口返回成功但客户端无消息：检查通知渠道是否启用，以及是否允许接收插件配置的消息类型。 diff --git a/plugins.v2/webhooknotify/__init__.py b/plugins.v2/webhooknotify/__init__.py index 485951fe..2bf64e72 100644 --- a/plugins.v2/webhooknotify/__init__.py +++ b/plugins.v2/webhooknotify/__init__.py @@ -1,15 +1,28 @@ +from secrets import compare_digest from typing import Annotated, Any, Dict, List, Optional, Tuple -from fastapi import Depends, HTTPException, Query, status +from fastapi import Depends, HTTPException, Query, Security, status +from fastapi.security import APIKeyHeader, APIKeyQuery from pydantic import BaseModel, Field, model_validator from app import schemas -from app.core.security import verify_apitoken from app.log import logger from app.plugins import _PluginBase from app.schemas import NotificationType +api_key_header = APIKeyHeader( + name="X-API-KEY", + auto_error=False, + scheme_name="webhooknotify_api_key_header", +) +api_key_query = APIKeyQuery( + name="apikey", + auto_error=False, + scheme_name="webhooknotify_api_key_query", +) + + class WebhookNotifyPayload(BaseModel): """外部 Webhook 通知的消息载荷。""" @@ -35,7 +48,7 @@ class WebhookNotify(_PluginBase): plugin_name = "Webhook消息推送" plugin_desc = "接收 Webhook 消息并推送到通知客户端。" plugin_icon = "https://raw.githubusercontent.com/InfinityPacer/MoviePilot-Plugins/main/icons/webhooknotify.png" - plugin_version = "1.0" + plugin_version = "1.1" plugin_author = "InfinityPacer" author_url = "https://github.com/InfinityPacer" plugin_config_prefix = "webhooknotify_" @@ -45,9 +58,10 @@ class WebhookNotify(_PluginBase): _enabled = False # MoviePilot 通知渠道按消息类型过滤，默认使用专用的插件分类。 _notify_type = NotificationType.Plugin + _api_key = "" def init_plugin(self, config: dict = None): - """加载插件开关和消息类型；公共 API_TOKEN 不在插件配置中重复保存。""" + """加载插件开关、消息类型和可选的独立 API Key。""" config = config or {} self._enabled = bool(config.get("enabled", False)) notify_type = config.get("notify_type", NotificationType.Plugin.name) @@ -56,6 +70,8 @@ def init_plugin(self, config: dict = None): if isinstance(notify_type, str) else NotificationType.Plugin ) + api_key = config.get("api_key") + self._api_key = api_key.strip() if isinstance(api_key, str) else "" def get_state(self) -> bool: """返回 Webhook 接收能力是否启用。""" @@ -67,27 +83,29 @@ def get_command() -> List[Dict[str, Any]]: return [] def get_api(self) -> List[Dict[str, Any]]: - """注册使用公共 API_TOKEN 认证的入站 Webhook。""" + """按独立 Key 配置状态注册对应的入站认证方式。""" + use_plugin_key = bool(self._api_key) + dependencies = [Depends(self._verify_api_key)] if use_plugin_key else [] return [ { "path": "/webhook", "endpoint": self.receive_webhook, "methods": ["POST"], - # 插件注册器的默认 apikey 只读取 X-API-KEY/apikey；认证依赖由 - # receive_webhook 的参数显式声明，以保持 MoviePilot 原生 ?token= 约定。 - "allow_anonymous": True, + "allow_anonymous": use_plugin_key, + "dependencies": dependencies.copy(), "response_model": schemas.Response, "summary": "接收 Webhook JSON 通知", - "description": "使用 MoviePilot 公共 API_TOKEN 接收 JSON；title 和 body 至少提供一项。", + "description": "使用 API Key 接收 JSON；title 和 body 至少提供一项。", }, { "path": "/webhook", "endpoint": self.receive_webhook_get, "methods": ["GET"], - "allow_anonymous": True, + "allow_anonymous": use_plugin_key, + "dependencies": dependencies.copy(), "response_model": schemas.Response, "summary": "接收 Webhook 查询通知", - "description": "使用 MoviePilot 公共 API_TOKEN 接收查询参数；title 和 body 至少提供一项。", + "description": "使用 API Key 接收查询参数；title 和 body 至少提供一项。", }, ] @@ -137,6 +155,27 @@ def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: }, ], }, + { + "component": "VRow", + "content": [ + { + "component": "VCol", + "props": {"cols": 12, "md": 6}, + "content": [ + { + "component": "VTextField", + "props": { + "model": "api_key", + "label": "APIKEY", + "type": "password", + "placeholder": "留空则使用主程序 API Token", + "clearable": True, + }, + } + ], + }, + ], + }, { "component": "VRow", "content": [ @@ -149,7 +188,7 @@ def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: "props": { "type": "info", "variant": "tonal", - "text": "GET/POST /api/v1/plugin/WebhookNotify/webhook?token=API_TOKEN，title 和 body 至少提供一项。", + "text": "GET/POST /api/v1/plugin/WebhookNotify/webhook，使用 X-API-KEY 或 apikey 认证，title 和 body 至少提供一项。", }, } ], @@ -161,6 +200,7 @@ def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: ], { "enabled": False, "notify_type": NotificationType.Plugin.name, + "api_key": "", } def get_page(self) -> Optional[List[dict]]: @@ -174,14 +214,12 @@ def stop_service(self): def receive_webhook( self, payload: WebhookNotifyPayload, - _: Annotated[str, Depends(verify_apitoken)], ) -> schemas.Response: """接收 POST JSON，并把外部消息交给 MoviePilot 通知链。""" return self._post_notification(payload, request_method="POST") def receive_webhook_get( self, - _: Annotated[str, Depends(verify_apitoken)], title: Annotated[Optional[str], Query(max_length=200)] = None, body: Annotated[Optional[str], Query(max_length=10000)] = None, ) -> schemas.Response: @@ -196,6 +234,24 @@ def receive_webhook_get( request_method="GET", ) + def _verify_api_key( + self, + key_query: Annotated[Optional[str], Security(api_key_query)] = None, + key_header: Annotated[Optional[str], Security(api_key_header)] = None, + ) -> str: + """校验插件独立 Key，并与主程序保持相同的凭据读取顺序。""" + supplied_key = key_header or key_query + if ( + not supplied_key + or not self._api_key + or not compare_digest(supplied_key, self._api_key) + ): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="API Key 校验不通过", + ) + return supplied_key + def _post_notification( self, payload: WebhookNotifyPayload, diff --git a/tests/v2/webhooknotify/test_webhooknotify.py b/tests/v2/webhooknotify/test_webhooknotify.py index 363966c1..4e67de7f 100644 --- a/tests/v2/webhooknotify/test_webhooknotify.py +++ b/tests/v2/webhooknotify/test_webhooknotify.py @@ -2,11 +2,12 @@ from unittest.mock import MagicMock import pytest -from fastapi import FastAPI, HTTPException +from fastapi import Depends, FastAPI, HTTPException from fastapi.testclient import TestClient from pydantic import ValidationError from app.core.config import settings +from app.core.security import verify_apikey from app.schemas.types import NotificationType from app.utils.object import ObjectUtils @@ -20,7 +21,10 @@ def _build_test_app(plugin: WebhookNotify) -> FastAPI: for api_definition in plugin.get_api(): api = api_definition.copy() path = api.pop("path") - api.pop("allow_anonymous") + allow_anonymous = api.pop("allow_anonymous") + dependencies = api.setdefault("dependencies", []) + if not allow_anonymous: + dependencies.append(Depends(verify_apikey)) test_app.add_api_route(path, **api) return test_app @@ -50,7 +54,7 @@ def _assert_notification( class TestWebhookNotify: - """Webhook 请求校验、公共 API_TOKEN 认证和通知转发契约。""" + """Webhook 请求校验、双模式 API Key 认证和通知转发契约。""" def test_payload_accepts_either_field_and_rejects_empty_message(self): assert WebhookNotifyPayload(title="告警").body is None @@ -62,16 +66,27 @@ def test_payload_accepts_either_field_and_rejects_empty_message(self): with pytest.raises(ValidationError, match="title 和 body 至少提供一项"): WebhookNotifyPayload(title=" ", body="\n") - def test_api_registers_get_and_post_with_public_token_dependency(self): + def test_api_uses_default_auth_without_plugin_key(self): plugin = WebhookNotify() + plugin.init_plugin({"api_key": " "}) api_definitions = plugin.get_api() assert len(api_definitions) == 2 assert {tuple(api["methods"]) for api in api_definitions} == {("GET",), ("POST",)} assert all(api["path"] == "/webhook" for api in api_definitions) - assert all(api["allow_anonymous"] is True for api in api_definitions) + assert all(api["allow_anonymous"] is False for api in api_definitions) + assert all(api.get("dependencies") == [] for api in api_definitions) assert all(api["response_model"] for api in api_definitions) + def test_api_uses_plugin_auth_when_plugin_key_is_configured(self): + plugin = WebhookNotify() + plugin.init_plugin({"api_key": "plugin-key"}) + + api_definitions = plugin.get_api() + + assert all(api["allow_anonymous"] is True for api in api_definitions) + assert all(len(api["dependencies"]) == 1 for api in api_definitions) + def test_data_page_is_not_exposed(self): assert WebhookNotify().get_page() is None assert ObjectUtils.check_method(WebhookNotify.get_page) is False @@ -81,12 +96,15 @@ def test_form_defaults_to_disabled_and_exposes_all_notification_types(self): form, defaults = plugin.get_form() enabled_row = form[0]["content"][0] notify_type_row = form[0]["content"][1] + api_key_row = form[0]["content"][2] enabled_field = enabled_row["content"][0]["content"][0] notify_type_field = notify_type_row["content"][0]["content"][0] + api_key_field = api_key_row["content"][0]["content"][0] assert defaults == { "enabled": False, "notify_type": NotificationType.Plugin.name, + "api_key": "", } assert enabled_row["component"] == "VRow" assert notify_type_row["component"] == "VRow" @@ -100,22 +118,39 @@ def test_form_defaults_to_disabled_and_exposes_all_notification_types(self): {"title": item.value, "value": item.name} for item in NotificationType ] - info_row = form[0]["content"][2] + assert api_key_row["component"] == "VRow" + assert api_key_row["content"][0]["props"] == {"cols": 12, "md": 6} + assert api_key_field["component"] == "VTextField" + assert api_key_field["props"]["model"] == "api_key" + assert api_key_field["props"]["label"] == "APIKEY" + assert api_key_field["props"]["type"] == "password" + info_row = form[0]["content"][3] assert info_row["component"] == "VRow" assert info_row["content"][0]["component"] == "VCol" assert info_row["content"][0]["props"] == {"cols": 12} assert info_row["content"][0]["content"][0]["component"] == "VAlert" - def test_missing_token_is_rejected_and_valid_token_forwards_message(self, monkeypatch): + @pytest.mark.parametrize( + ("request_kwargs",), + [ + ({"headers": {"X-API-KEY": "moviepilot-key"}},), + ({"params": {"apikey": "moviepilot-key"}},), + ], + ) + def test_default_auth_accepts_public_api_token( + self, + monkeypatch, + request_kwargs, + ): plugin = WebhookNotify() plugin.init_plugin({"enabled": True}) post_message = _mock_notification_chain(plugin) - monkeypatch.setattr(settings, "API_TOKEN", "unit-test-token") + monkeypatch.setattr(settings, "API_TOKEN", "moviepilot-key") client = TestClient(_build_test_app(plugin)) payload = {"title": "路由故障", "body": "主线路不可达"} unauthorized = client.post("/webhook", json=payload) - authorized = client.post("/webhook?token=unit-test-token", json=payload) + authorized = client.post("/webhook", json=payload, **request_kwargs) assert unauthorized.status_code == 401 assert authorized.status_code == 200 @@ -127,6 +162,48 @@ def test_missing_token_is_rejected_and_valid_token_forwards_message(self, monkey text="主线路不可达", ) + @pytest.mark.parametrize( + ("request_kwargs",), + [ + ({"headers": {"X-API-KEY": "plugin-key"}},), + ({"params": {"apikey": "plugin-key"}},), + ], + ) + def test_plugin_key_replaces_public_api_token( + self, + monkeypatch, + request_kwargs, + ): + plugin = WebhookNotify() + plugin.init_plugin({"enabled": True, "api_key": "plugin-key"}) + post_message = _mock_notification_chain(plugin) + monkeypatch.setattr(settings, "API_TOKEN", "moviepilot-key") + client = TestClient(_build_test_app(plugin)) + payload = {"title": "独立认证"} + + missing = client.post("/webhook", json=payload) + public_token = client.post( + "/webhook", + json=payload, + headers={"X-API-KEY": "moviepilot-key"}, + ) + wrong_header_overrides_query = client.post( + "/webhook?apikey=plugin-key", + json=payload, + headers={"X-API-KEY": "wrong-key"}, + ) + authorized = client.post("/webhook", json=payload, **request_kwargs) + + assert missing.status_code == 401 + assert public_token.status_code == 401 + assert wrong_header_overrides_query.status_code == 401 + assert authorized.status_code == 200 + _assert_notification( + post_message, + mtype=NotificationType.Plugin, + title="独立认证", + ) + @pytest.mark.parametrize( ("payload", "expected_title", "expected_text"), [ @@ -141,7 +218,7 @@ def test_post_accepts_title_or_body(self, monkeypatch, payload, expected_title, monkeypatch.setattr(settings, "API_TOKEN", "unit-test-token") client = TestClient(_build_test_app(plugin)) - response = client.post("/webhook?token=unit-test-token", json=payload) + response = client.post("/webhook?apikey=unit-test-token", json=payload) assert response.status_code == 200 _assert_notification( @@ -164,7 +241,7 @@ def test_get_accepts_title_or_body(self, monkeypatch, params, expected_title, ex post_message = _mock_notification_chain(plugin) monkeypatch.setattr(settings, "API_TOKEN", "unit-test-token") client = TestClient(_build_test_app(plugin)) - params["token"] = "unit-test-token" + params["apikey"] = "unit-test-token" response = client.get("/webhook", params=params) @@ -178,16 +255,19 @@ def test_get_accepts_title_or_body(self, monkeypatch, params, expected_title, ex def test_submission_logs_metadata_without_message_content(self, monkeypatch): plugin = WebhookNotify() - plugin.init_plugin({"enabled": True}) + plugin.init_plugin({"enabled": True, "api_key": "unit-test-token"}) _mock_notification_chain(plugin) log_info = MagicMock() monkeypatch.setattr(webhooknotify.logger, "info", log_info) + client = TestClient(_build_test_app(plugin)) - plugin.receive_webhook( - WebhookNotifyPayload(title="敏感标题", body="敏感正文"), - "unit-test-token", + response = client.post( + "/webhook", + headers={"X-API-KEY": "unit-test-token"}, + json={"title": "敏感标题", "body": "敏感正文"}, ) + assert response.status_code == 200 assert log_info.call_count == 2 logged_values = " ".join( str(value) @@ -206,8 +286,8 @@ def test_get_and_post_reject_missing_content(self, monkeypatch): monkeypatch.setattr(settings, "API_TOKEN", "unit-test-token") client = TestClient(_build_test_app(plugin)) - post_response = client.post("/webhook?token=unit-test-token", json={}) - get_response = client.get("/webhook?token=unit-test-token") + post_response = client.post("/webhook?apikey=unit-test-token", json={}) + get_response = client.get("/webhook?apikey=unit-test-token") assert post_response.status_code == 422 assert get_response.status_code == 422 @@ -220,7 +300,7 @@ def test_plugin_defaults_to_disabled_and_returns_service_unavailable(self): assert plugin.get_state() is False with pytest.raises(HTTPException) as exc_info: - plugin.receive_webhook(WebhookNotifyPayload(title="标题"), "unit-test-token") + plugin.receive_webhook(WebhookNotifyPayload(title="标题")) assert exc_info.value.status_code == 503 post_message.assert_not_called() @@ -251,7 +331,6 @@ def test_configured_notification_type_is_forwarded_or_defaults_to_plugin( response = plugin.receive_webhook( WebhookNotifyPayload(title="标题"), - "unit-test-token", ) assert response.success is True