From 507f703cb5bce56ab0afb3806d24e6ee25d34ed4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:32:46 +0000 Subject: [PATCH 01/10] Initial plan From 0754b6827ef86bde654b8283349d76f15863b0b5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:36:27 +0000 Subject: [PATCH 02/10] fix: address security review comments - webhook URL logging, SSRF protection, memory leak, and type safety Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- .env.example | 4 +++ src/stickerhub/adapters/feishu_sender.py | 24 ++++++++++++++-- src/stickerhub/adapters/telegram_source.py | 5 ++++ src/stickerhub/config.py | 4 +++ src/stickerhub/core/ports.py | 6 ++-- src/stickerhub/main.py | 1 + src/stickerhub/services/binding.py | 32 ++++++++++++++++++---- 7 files changed, 67 insertions(+), 9 deletions(-) diff --git a/.env.example b/.env.example index b61f26f..2dee6db 100644 --- a/.env.example +++ b/.env.example @@ -9,5 +9,9 @@ FEISHU_APP_SECRET= BINDING_DB_PATH=data/stickerhub.db BIND_MAGIC_TTL_SECONDS=600 +# 飞书 Webhook 域名白名单(逗号分隔,留空则使用默认值) +# 默认:open.feishu.cn,open.larksuite.com +FEISHU_WEBHOOK_ALLOWED_HOSTS= + # 日志级别(DEBUG / INFO / WARNING / ERROR) LOG_LEVEL=INFO diff --git a/src/stickerhub/adapters/feishu_sender.py b/src/stickerhub/adapters/feishu_sender.py index e2de91b..c80055e 100644 --- a/src/stickerhub/adapters/feishu_sender.py +++ b/src/stickerhub/adapters/feishu_sender.py @@ -1,5 +1,6 @@ import json import logging +from typing import Literal import httpx @@ -18,11 +19,15 @@ def __init__( self._app_secret = app_secret self._base_url = "https://open.feishu.cn/open-apis" - async def send(self, asset: StickerAsset, target_mode: str, target: str) -> None: + async def send( + self, asset: StickerAsset, target_mode: Literal["bot", "webhook"], target: str + ) -> None: + # 避免在日志中暴露 webhook URL 中的敏感 token + safe_target = target if target_mode == "bot" else _mask_webhook_url(target) logger.debug( "准备发送图片到飞书: mode=%s target=%s file=%s mime=%s size=%s", target_mode, - target, + safe_target, asset.file_name, asset.mime_type, len(asset.content), @@ -192,3 +197,18 @@ async def _send_webhook_message( code_ok = code in (None, 0, "0") if not status_ok or not code_ok: raise RuntimeError(f"发送飞书 webhook 消息失败: {payload}") + + +def _mask_webhook_url(webhook_url: str) -> str: + """脱敏 webhook URL,仅保留 host 和末尾部分,避免泄露敏感 token""" + try: + from urllib.parse import urlparse + + parsed = urlparse(webhook_url) + if parsed.path and len(parsed.path) > 20: + masked_path = f"{parsed.path[:20]}...{parsed.path[-8:]}" + else: + masked_path = parsed.path + return f"{parsed.scheme}://{parsed.netloc}{masked_path}" + except Exception: # noqa: BLE001 + return "[webhook_url_masked]" diff --git a/src/stickerhub/adapters/telegram_source.py b/src/stickerhub/adapters/telegram_source.py index b2ef8c0..5b8a107 100644 --- a/src/stickerhub/adapters/telegram_source.py +++ b/src/stickerhub/adapters/telegram_source.py @@ -122,6 +122,8 @@ async def handle_bind(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non if not update.message or not update.effective_user: return + _cleanup_pending_webhook_requests(pending_webhook_requests) + try: arg = context.args[0] if context.args else None telegram_user_id = str(update.effective_user.id) @@ -150,6 +152,8 @@ async def handle_bind_mode_callback(update: Update, context: ContextTypes.DEFAUL return await query.answer() + _cleanup_pending_webhook_requests(pending_webhook_requests) + data = query.data or "" parsed = _parse_bind_mode_callback_data(data) if not parsed: @@ -194,6 +198,7 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> return _cleanup_pending_requests(pending_pack_requests) + _cleanup_pending_webhook_requests(pending_webhook_requests) try: asset = await _extract_asset(update.message, context) diff --git a/src/stickerhub/config.py b/src/stickerhub/config.py index 890cc49..3bdee9c 100644 --- a/src/stickerhub/config.py +++ b/src/stickerhub/config.py @@ -21,6 +21,10 @@ class Settings(BaseSettings): validation_alias=AliasChoices("BINDING_DB_PATH", "BINDING_STORE_PATH"), ) bind_magic_ttl_seconds: int = Field(default=600, alias="BIND_MAGIC_TTL_SECONDS") + feishu_webhook_allowed_hosts: list[str] = Field( + default=["open.feishu.cn", "open.larksuite.com"], + alias="FEISHU_WEBHOOK_ALLOWED_HOSTS", + ) log_level: str = Field(default="INFO", alias="LOG_LEVEL") model_config = SettingsConfigDict( diff --git a/src/stickerhub/core/ports.py b/src/stickerhub/core/ports.py index b5b91e6..bad3031 100644 --- a/src/stickerhub/core/ports.py +++ b/src/stickerhub/core/ports.py @@ -1,4 +1,4 @@ -from typing import Protocol +from typing import Literal, Protocol from stickerhub.core.models import StickerAsset @@ -9,5 +9,7 @@ async def normalize(self, asset: StickerAsset) -> StickerAsset: class TargetPlatformSender(Protocol): - async def send(self, asset: StickerAsset, target_mode: str, target: str) -> None: + async def send( + self, asset: StickerAsset, target_mode: Literal["bot", "webhook"], target: str + ) -> None: """将素材发送到目标平台。""" diff --git a/src/stickerhub/main.py b/src/stickerhub/main.py index ed1a0eb..414cee9 100644 --- a/src/stickerhub/main.py +++ b/src/stickerhub/main.py @@ -27,6 +27,7 @@ async def async_main() -> None: binding_service = BindingService( store=BindingStore(settings.binding_db_path), magic_ttl_seconds=settings.bind_magic_ttl_seconds, + webhook_allowed_hosts=settings.feishu_webhook_allowed_hosts, ) await binding_service.initialize() diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index d1b376f..9f4c4a6 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -344,9 +344,18 @@ def _connect(self) -> sqlite3.Connection: class BindingService: - def __init__(self, store: BindingStore, magic_ttl_seconds: int = 600) -> None: + def __init__( + self, + store: BindingStore, + magic_ttl_seconds: int = 600, + webhook_allowed_hosts: list[str] | None = None, + ) -> None: self._store = store self._magic_ttl_seconds = magic_ttl_seconds + self._webhook_allowed_hosts = webhook_allowed_hosts or [ + "open.feishu.cn", + "open.larksuite.com", + ] async def initialize(self) -> None: await self._store.ensure_initialized() @@ -407,15 +416,17 @@ async def handle_bind_webhook( source_user_id: str, webhook_url: str, ) -> str: - normalized_url = _normalize_feishu_webhook_url(webhook_url) + normalized_url = _normalize_feishu_webhook_url(webhook_url, self._webhook_allowed_hosts) if not normalized_url: logger.warning( - "Webhook 绑定失败: 平台=%s user=%s 原因=URL格式不合法", + "Webhook 绑定失败: 平台=%s user=%s 原因=URL格式不合法或域名不在白名单内", source_platform, source_user_id, ) + allowed_hosts_str = ", ".join(self._webhook_allowed_hosts) return ( - "绑定失败: Webhook 地址格式不合法。\n" + "绑定失败: Webhook 地址格式不合法或域名不在白名单内。\n" + f"允许的域名:{allowed_hosts_str}\n" "请填写飞书自定义机器人 Webhook 地址,例如:\n" "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx" ) @@ -488,7 +499,13 @@ async def get_feishu_target( return None -def _normalize_feishu_webhook_url(url: str) -> str | None: +def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str]) -> str | None: + """ + 验证并归一化飞书 Webhook URL。 + - 必须是 https 协议 + - 域名必须在白名单内(防止 SSRF) + - 路径必须包含 /open-apis/bot/v2/hook/ + """ normalized = url.strip() if not normalized: return None @@ -498,6 +515,11 @@ def _normalize_feishu_webhook_url(url: str) -> str | None: return None if not parsed.netloc: return None + + # 域名白名单校验(SSRF 防护) + if parsed.netloc.lower() not in [host.lower() for host in allowed_hosts]: + return None + if "/open-apis/bot/v2/hook/" not in parsed.path: return None return normalized From 98fc36e29bc27429eb1a1f872e321210f1f90ea4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:37:29 +0000 Subject: [PATCH 03/10] test: add coverage for webhook cleanup and domain whitelist validation Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- README.md | 6 ++++++ tests/test_binding_sqlite.py | 30 ++++++++++++++++++++++++++++++ tests/test_telegram_source.py | 19 +++++++++++++++++++ 3 files changed, 55 insertions(+) diff --git a/README.md b/README.md index ebc3211..2fbf83f 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,11 @@ FEISHU_APP_SECRET= BINDING_DB_PATH=data/stickerhub.db BIND_MAGIC_TTL_SECONDS=600 + +# 飞书 Webhook 域名白名单(逗号分隔,留空则使用默认值) +# 默认:open.feishu.cn,open.larksuite.com +FEISHU_WEBHOOK_ALLOWED_HOSTS= + LOG_LEVEL=INFO ``` @@ -62,6 +67,7 @@ LOG_LEVEL=INFO - `TELEGRAM_BOT_API_TOKEN`:必填,Telegram Bot API Token - `FEISHU_APP_ID` / `FEISHU_APP_SECRET`:可选,填写后启用飞书转发和 `/bind` 功能(包括 webhook 绑定所需的图片上传能力) +- `FEISHU_WEBHOOK_ALLOWED_HOSTS`:飞书 Webhook 域名白名单(防止 SSRF 攻击),默认仅允许 `open.feishu.cn` 和 `open.larksuite.com` - 飞书需在应用后台开启机器人收发消息权限(im:message)以及获取与上传图片或文件资源权限(im:resource) ![lark_permission.png](docs/lark_permission.png) - 飞书事件添加接收消息(im.message.receive_v1)并启用长连接事件能力。 ![lark_event.png](docs/lark_event.png) diff --git a/tests/test_binding_sqlite.py b/tests/test_binding_sqlite.py index 9e88bb0..18a56cf 100644 --- a/tests/test_binding_sqlite.py +++ b/tests/test_binding_sqlite.py @@ -137,6 +137,31 @@ async def _bind_webhook_invalid_url(db_path: str) -> None: assert "格式不合法" in reply +async def _bind_webhook_domain_whitelist_enforcement(db_path: str) -> None: + """测试域名白名单校验(SSRF 防护)""" + store = BindingStore(db_path) + # 自定义白名单,仅允许 open.feishu.cn + service = BindingService( + store=store, magic_ttl_seconds=600, webhook_allowed_hosts=["open.feishu.cn"] + ) + await service.initialize() + + # 合法域名应通过 + valid_url = "https://open.feishu.cn/open-apis/bot/v2/hook/valid_token" + reply = await service.handle_bind_webhook("telegram", "tg_whitelist_ok", valid_url) + assert "绑定成功" in reply + + # 不在白名单的域名应被拒绝(防止 SSRF) + blocked_url = "https://evil.com/open-apis/bot/v2/hook/malicious" + reply = await service.handle_bind_webhook("telegram", "tg_whitelist_block", blocked_url) + assert "白名单" in reply or "格式不合法" in reply + + # open.larksuite.com 不在自定义白名单中,应被拒绝 + larksuite_url = "https://open.larksuite.com/open-apis/bot/v2/hook/token" + reply = await service.handle_bind_webhook("telegram", "tg_whitelist_lark", larksuite_url) + assert "白名单" in reply or "格式不合法" in reply + + def test_bind_flow_with_sqlite(tmp_path) -> None: db_path = tmp_path / "binding.db" asyncio.run(_bind_flow(str(db_path))) @@ -175,3 +200,8 @@ def test_switch_from_webhook_to_bot(tmp_path) -> None: def test_bind_webhook_invalid_url(tmp_path) -> None: db_path = tmp_path / "binding.db" asyncio.run(_bind_webhook_invalid_url(str(db_path))) + + +def test_bind_webhook_domain_whitelist(tmp_path) -> None: + db_path = tmp_path / "binding.db" + asyncio.run(_bind_webhook_domain_whitelist_enforcement(str(db_path))) diff --git a/tests/test_telegram_source.py b/tests/test_telegram_source.py index 7168071..5bd1af4 100644 --- a/tests/test_telegram_source.py +++ b/tests/test_telegram_source.py @@ -6,8 +6,10 @@ BIND_MODE_CALLBACK_PREFIX, PACK_CALLBACK_PREFIX, PendingStickerPackRequest, + PendingWebhookBindRequest, RunningStickerPackTask, _cleanup_pending_requests, + _cleanup_pending_webhook_requests, _deduplicate_filename, _detect_sticker_mime, _has_running_task_for_user, @@ -176,3 +178,20 @@ async def reply_document(self, *args: object, **kwargs: object) -> None: assert message.document_called is True assert message.animation_called is False + + +def test_cleanup_pending_webhook_requests_removes_expired_only() -> None: + now = int(time.time()) + pending = { + "expired_user": PendingWebhookBindRequest( + telegram_user_id="expired_user", + created_at=now - 3600, + ), + "fresh_user": PendingWebhookBindRequest( + telegram_user_id="fresh_user", + created_at=now, + ), + } + _cleanup_pending_webhook_requests(pending) + assert "expired_user" not in pending + assert "fresh_user" in pending From e4c058ec9567a1de990b76e786628ec142b9c8a3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:39:05 +0000 Subject: [PATCH 04/10] refactor: address code review feedback - extract constants, improve logging, fix test naming Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- src/stickerhub/adapters/feishu_sender.py | 8 ++++++-- src/stickerhub/services/binding.py | 18 +++++++++++++++++- tests/test_binding_sqlite.py | 4 ++-- 3 files changed, 25 insertions(+), 5 deletions(-) diff --git a/src/stickerhub/adapters/feishu_sender.py b/src/stickerhub/adapters/feishu_sender.py index c80055e..6c5b4a1 100644 --- a/src/stickerhub/adapters/feishu_sender.py +++ b/src/stickerhub/adapters/feishu_sender.py @@ -199,14 +199,18 @@ async def _send_webhook_message( raise RuntimeError(f"发送飞书 webhook 消息失败: {payload}") +PATH_PREFIX_LENGTH = 20 +PATH_SUFFIX_LENGTH = 8 + + def _mask_webhook_url(webhook_url: str) -> str: """脱敏 webhook URL,仅保留 host 和末尾部分,避免泄露敏感 token""" try: from urllib.parse import urlparse parsed = urlparse(webhook_url) - if parsed.path and len(parsed.path) > 20: - masked_path = f"{parsed.path[:20]}...{parsed.path[-8:]}" + if parsed.path and len(parsed.path) > PATH_PREFIX_LENGTH: + masked_path = f"{parsed.path[:PATH_PREFIX_LENGTH]}...{parsed.path[-PATH_SUFFIX_LENGTH:]}" else: masked_path = parsed.path return f"{parsed.scheme}://{parsed.netloc}{masked_path}" diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index 9f4c4a6..c999b68 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -418,10 +418,13 @@ async def handle_bind_webhook( ) -> str: normalized_url = _normalize_feishu_webhook_url(webhook_url, self._webhook_allowed_hosts) if not normalized_url: + # 脱敏 URL 用于日志 + masked_url = _mask_url_for_log(webhook_url) logger.warning( - "Webhook 绑定失败: 平台=%s user=%s 原因=URL格式不合法或域名不在白名单内", + "Webhook 绑定失败: 平台=%s user=%s 原因=URL格式不合法或域名不在白名单内 url=%s", source_platform, source_user_id, + masked_url, ) allowed_hosts_str = ", ".join(self._webhook_allowed_hosts) return ( @@ -523,3 +526,16 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str]) -> str | N if "/open-apis/bot/v2/hook/" not in parsed.path: return None return normalized + + +def _mask_url_for_log(url: str) -> str: + """脱敏 URL 用于日志输出,避免泄露敏感 token""" + try: + parsed = urlparse(url) + if parsed.path and len(parsed.path) > 30: + masked_path = f"{parsed.path[:20]}...{parsed.path[-8:]}" + else: + masked_path = parsed.path + return f"{parsed.scheme}://{parsed.netloc}{masked_path}" + except Exception: # noqa: BLE001 + return "[url_masked]" diff --git a/tests/test_binding_sqlite.py b/tests/test_binding_sqlite.py index 18a56cf..3b736c3 100644 --- a/tests/test_binding_sqlite.py +++ b/tests/test_binding_sqlite.py @@ -137,7 +137,7 @@ async def _bind_webhook_invalid_url(db_path: str) -> None: assert "格式不合法" in reply -async def _bind_webhook_domain_whitelist_enforcement(db_path: str) -> None: +async def _test_bind_webhook_domain_whitelist(db_path: str) -> None: """测试域名白名单校验(SSRF 防护)""" store = BindingStore(db_path) # 自定义白名单,仅允许 open.feishu.cn @@ -204,4 +204,4 @@ def test_bind_webhook_invalid_url(tmp_path) -> None: def test_bind_webhook_domain_whitelist(tmp_path) -> None: db_path = tmp_path / "binding.db" - asyncio.run(_bind_webhook_domain_whitelist_enforcement(str(db_path))) + asyncio.run(_test_bind_webhook_domain_whitelist(str(db_path))) From f02f5b8dd59c2b2c3a34b65e271bbe45249a1697 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:40:23 +0000 Subject: [PATCH 05/10] refactor: use consistent constants for URL masking across modules Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- src/stickerhub/adapters/feishu_sender.py | 3 ++- src/stickerhub/services/binding.py | 12 ++++++++++-- tests/test_binding_sqlite.py | 4 ++-- 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/src/stickerhub/adapters/feishu_sender.py b/src/stickerhub/adapters/feishu_sender.py index 6c5b4a1..1ece4a6 100644 --- a/src/stickerhub/adapters/feishu_sender.py +++ b/src/stickerhub/adapters/feishu_sender.py @@ -201,6 +201,7 @@ async def _send_webhook_message( PATH_PREFIX_LENGTH = 20 PATH_SUFFIX_LENGTH = 8 +PATH_MASK_THRESHOLD = PATH_PREFIX_LENGTH + PATH_SUFFIX_LENGTH def _mask_webhook_url(webhook_url: str) -> str: @@ -209,7 +210,7 @@ def _mask_webhook_url(webhook_url: str) -> str: from urllib.parse import urlparse parsed = urlparse(webhook_url) - if parsed.path and len(parsed.path) > PATH_PREFIX_LENGTH: + if parsed.path and len(parsed.path) > PATH_MASK_THRESHOLD: masked_path = f"{parsed.path[:PATH_PREFIX_LENGTH]}...{parsed.path[-PATH_SUFFIX_LENGTH:]}" else: masked_path = parsed.path diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index c999b68..52f2c9a 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -528,12 +528,20 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str]) -> str | N return normalized +# Constants for URL masking (shared with feishu_sender.py) +_URL_PATH_PREFIX_LENGTH = 20 +_URL_PATH_SUFFIX_LENGTH = 8 +_URL_PATH_MASK_THRESHOLD = _URL_PATH_PREFIX_LENGTH + _URL_PATH_SUFFIX_LENGTH + + def _mask_url_for_log(url: str) -> str: """脱敏 URL 用于日志输出,避免泄露敏感 token""" try: parsed = urlparse(url) - if parsed.path and len(parsed.path) > 30: - masked_path = f"{parsed.path[:20]}...{parsed.path[-8:]}" + if parsed.path and len(parsed.path) > _URL_PATH_MASK_THRESHOLD: + masked_path = ( + f"{parsed.path[:_URL_PATH_PREFIX_LENGTH]}...{parsed.path[-_URL_PATH_SUFFIX_LENGTH:]}" + ) else: masked_path = parsed.path return f"{parsed.scheme}://{parsed.netloc}{masked_path}" diff --git a/tests/test_binding_sqlite.py b/tests/test_binding_sqlite.py index 3b736c3..6754779 100644 --- a/tests/test_binding_sqlite.py +++ b/tests/test_binding_sqlite.py @@ -137,7 +137,7 @@ async def _bind_webhook_invalid_url(db_path: str) -> None: assert "格式不合法" in reply -async def _test_bind_webhook_domain_whitelist(db_path: str) -> None: +async def _bind_webhook_domain_whitelist(db_path: str) -> None: """测试域名白名单校验(SSRF 防护)""" store = BindingStore(db_path) # 自定义白名单,仅允许 open.feishu.cn @@ -204,4 +204,4 @@ def test_bind_webhook_invalid_url(tmp_path) -> None: def test_bind_webhook_domain_whitelist(tmp_path) -> None: db_path = tmp_path / "binding.db" - asyncio.run(_test_bind_webhook_domain_whitelist(str(db_path))) + asyncio.run(_bind_webhook_domain_whitelist(str(db_path))) From 060d5dfd5b5bc399150a741de56d645668f2e90b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:42:46 +0000 Subject: [PATCH 06/10] refactor: extract URL masking to shared utility module to eliminate duplication Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- src/stickerhub/adapters/feishu_sender.py | 23 ++------------------ src/stickerhub/services/binding.py | 25 +++------------------- src/stickerhub/utils/url_masking.py | 27 ++++++++++++++++++++++++ 3 files changed, 32 insertions(+), 43 deletions(-) create mode 100644 src/stickerhub/utils/url_masking.py diff --git a/src/stickerhub/adapters/feishu_sender.py b/src/stickerhub/adapters/feishu_sender.py index 1ece4a6..e672a5c 100644 --- a/src/stickerhub/adapters/feishu_sender.py +++ b/src/stickerhub/adapters/feishu_sender.py @@ -5,6 +5,7 @@ import httpx from stickerhub.core.models import StickerAsset +from stickerhub.utils.url_masking import mask_url logger = logging.getLogger(__name__) @@ -23,7 +24,7 @@ async def send( self, asset: StickerAsset, target_mode: Literal["bot", "webhook"], target: str ) -> None: # 避免在日志中暴露 webhook URL 中的敏感 token - safe_target = target if target_mode == "bot" else _mask_webhook_url(target) + safe_target = target if target_mode == "bot" else mask_url(target) logger.debug( "准备发送图片到飞书: mode=%s target=%s file=%s mime=%s size=%s", target_mode, @@ -197,23 +198,3 @@ async def _send_webhook_message( code_ok = code in (None, 0, "0") if not status_ok or not code_ok: raise RuntimeError(f"发送飞书 webhook 消息失败: {payload}") - - -PATH_PREFIX_LENGTH = 20 -PATH_SUFFIX_LENGTH = 8 -PATH_MASK_THRESHOLD = PATH_PREFIX_LENGTH + PATH_SUFFIX_LENGTH - - -def _mask_webhook_url(webhook_url: str) -> str: - """脱敏 webhook URL,仅保留 host 和末尾部分,避免泄露敏感 token""" - try: - from urllib.parse import urlparse - - parsed = urlparse(webhook_url) - if parsed.path and len(parsed.path) > PATH_MASK_THRESHOLD: - masked_path = f"{parsed.path[:PATH_PREFIX_LENGTH]}...{parsed.path[-PATH_SUFFIX_LENGTH:]}" - else: - masked_path = parsed.path - return f"{parsed.scheme}://{parsed.netloc}{masked_path}" - except Exception: # noqa: BLE001 - return "[webhook_url_masked]" diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index 52f2c9a..72cab81 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -9,6 +9,8 @@ from typing import Literal from urllib.parse import urlparse +from stickerhub.utils.url_masking import mask_url + logger = logging.getLogger(__name__) @@ -419,7 +421,7 @@ async def handle_bind_webhook( normalized_url = _normalize_feishu_webhook_url(webhook_url, self._webhook_allowed_hosts) if not normalized_url: # 脱敏 URL 用于日志 - masked_url = _mask_url_for_log(webhook_url) + masked_url = mask_url(webhook_url) logger.warning( "Webhook 绑定失败: 平台=%s user=%s 原因=URL格式不合法或域名不在白名单内 url=%s", source_platform, @@ -526,24 +528,3 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str]) -> str | N if "/open-apis/bot/v2/hook/" not in parsed.path: return None return normalized - - -# Constants for URL masking (shared with feishu_sender.py) -_URL_PATH_PREFIX_LENGTH = 20 -_URL_PATH_SUFFIX_LENGTH = 8 -_URL_PATH_MASK_THRESHOLD = _URL_PATH_PREFIX_LENGTH + _URL_PATH_SUFFIX_LENGTH - - -def _mask_url_for_log(url: str) -> str: - """脱敏 URL 用于日志输出,避免泄露敏感 token""" - try: - parsed = urlparse(url) - if parsed.path and len(parsed.path) > _URL_PATH_MASK_THRESHOLD: - masked_path = ( - f"{parsed.path[:_URL_PATH_PREFIX_LENGTH]}...{parsed.path[-_URL_PATH_SUFFIX_LENGTH:]}" - ) - else: - masked_path = parsed.path - return f"{parsed.scheme}://{parsed.netloc}{masked_path}" - except Exception: # noqa: BLE001 - return "[url_masked]" diff --git a/src/stickerhub/utils/url_masking.py b/src/stickerhub/utils/url_masking.py new file mode 100644 index 0000000..1b9886a --- /dev/null +++ b/src/stickerhub/utils/url_masking.py @@ -0,0 +1,27 @@ +"""Shared constants and utilities for URL masking to prevent credential leakage in logs.""" + +from urllib.parse import urlparse + +# URL masking constants +PATH_PREFIX_LENGTH = 20 +PATH_SUFFIX_LENGTH = 8 +PATH_MASK_THRESHOLD = PATH_PREFIX_LENGTH + PATH_SUFFIX_LENGTH + + +def mask_url(url: str) -> str: + """ + 脱敏 URL 用于日志输出,避免泄露敏感 token。 + + 仅保留协议、域名、路径前 20 个字符和后 8 个字符,中间用 ... 替代。 + """ + try: + parsed = urlparse(url) + if parsed.path and len(parsed.path) > PATH_MASK_THRESHOLD: + masked_path = ( + f"{parsed.path[:PATH_PREFIX_LENGTH]}...{parsed.path[-PATH_SUFFIX_LENGTH:]}" + ) + else: + masked_path = parsed.path + return f"{parsed.scheme}://{parsed.netloc}{masked_path}" + except Exception: # noqa: BLE001 + return "[url_masked]" From 48c6045659ee740ddf00f5b6dac796ffdc025134 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 11:44:18 +0000 Subject: [PATCH 07/10] fix: improve exception handling and avoid exposing whitelist in error messages Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- src/stickerhub/services/binding.py | 3 +-- src/stickerhub/utils/url_masking.py | 3 ++- tests/test_binding_sqlite.py | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index 72cab81..b0d2fbd 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -428,10 +428,9 @@ async def handle_bind_webhook( source_user_id, masked_url, ) - allowed_hosts_str = ", ".join(self._webhook_allowed_hosts) + # 不在用户消息中暴露完整白名单,避免为攻击者提供信息 return ( "绑定失败: Webhook 地址格式不合法或域名不在白名单内。\n" - f"允许的域名:{allowed_hosts_str}\n" "请填写飞书自定义机器人 Webhook 地址,例如:\n" "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx" ) diff --git a/src/stickerhub/utils/url_masking.py b/src/stickerhub/utils/url_masking.py index 1b9886a..dc3fb1b 100644 --- a/src/stickerhub/utils/url_masking.py +++ b/src/stickerhub/utils/url_masking.py @@ -23,5 +23,6 @@ def mask_url(url: str) -> str: else: masked_path = parsed.path return f"{parsed.scheme}://{parsed.netloc}{masked_path}" - except Exception: # noqa: BLE001 + except (ValueError, TypeError, AttributeError): + # urlparse 可能抛出 ValueError,或传入 None 导致 TypeError/AttributeError return "[url_masked]" diff --git a/tests/test_binding_sqlite.py b/tests/test_binding_sqlite.py index 6754779..cf20cb4 100644 --- a/tests/test_binding_sqlite.py +++ b/tests/test_binding_sqlite.py @@ -154,12 +154,12 @@ async def _bind_webhook_domain_whitelist(db_path: str) -> None: # 不在白名单的域名应被拒绝(防止 SSRF) blocked_url = "https://evil.com/open-apis/bot/v2/hook/malicious" reply = await service.handle_bind_webhook("telegram", "tg_whitelist_block", blocked_url) - assert "白名单" in reply or "格式不合法" in reply + assert "白名单" in reply # open.larksuite.com 不在自定义白名单中,应被拒绝 larksuite_url = "https://open.larksuite.com/open-apis/bot/v2/hook/token" reply = await service.handle_bind_webhook("telegram", "tg_whitelist_lark", larksuite_url) - assert "白名单" in reply or "格式不合法" in reply + assert "白名单" in reply def test_bind_flow_with_sqlite(tmp_path) -> None: From 7851551a492daf676c63849dddb7331410b5e79d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 12:14:32 +0000 Subject: [PATCH 08/10] fix: improve URL masking, use hostname for whitelist validation, add option to disable whitelist, use JSON format for env config Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- .env.example | 7 ++++-- README.md | 11 ++++++--- src/stickerhub/config.py | 21 +++++++++++++++-- src/stickerhub/main.py | 2 +- src/stickerhub/services/binding.py | 35 ++++++++++++++++++++--------- src/stickerhub/utils/url_masking.py | 12 ++++++++-- tests/test_binding_sqlite.py | 30 ++++++++++++++++++++++++- 7 files changed, 96 insertions(+), 22 deletions(-) diff --git a/.env.example b/.env.example index 2dee6db..92e9648 100644 --- a/.env.example +++ b/.env.example @@ -9,8 +9,11 @@ FEISHU_APP_SECRET= BINDING_DB_PATH=data/stickerhub.db BIND_MAGIC_TTL_SECONDS=600 -# 飞书 Webhook 域名白名单(逗号分隔,留空则使用默认值) -# 默认:open.feishu.cn,open.larksuite.com +# 飞书 Webhook 域名白名单(JSON 格式的字符串数组) +# 不设置:使用默认白名单 ["open.feishu.cn","open.larksuite.com"] +# 设为 []:禁用白名单校验(允许任意域名,请谨慎使用) +# 设为自定义列表:如 ["open.feishu.cn","custom.domain.com"] +# FEISHU_WEBHOOK_ALLOWED_HOSTS=["open.feishu.cn","open.larksuite.com"] FEISHU_WEBHOOK_ALLOWED_HOSTS= # 日志级别(DEBUG / INFO / WARNING / ERROR) diff --git a/README.md b/README.md index 2fbf83f..7b3ea44 100644 --- a/README.md +++ b/README.md @@ -56,8 +56,10 @@ FEISHU_APP_SECRET= BINDING_DB_PATH=data/stickerhub.db BIND_MAGIC_TTL_SECONDS=600 -# 飞书 Webhook 域名白名单(逗号分隔,留空则使用默认值) -# 默认:open.feishu.cn,open.larksuite.com +# 飞书 Webhook 域名白名单(JSON 格式的字符串数组) +# 不设置:使用默认白名单 ["open.feishu.cn","open.larksuite.com"] +# 设为 []:禁用白名单校验(允许任意域名,请谨慎使用) +# 设为自定义列表:如 ["open.feishu.cn","custom.domain.com"] FEISHU_WEBHOOK_ALLOWED_HOSTS= LOG_LEVEL=INFO @@ -67,7 +69,10 @@ LOG_LEVEL=INFO - `TELEGRAM_BOT_API_TOKEN`:必填,Telegram Bot API Token - `FEISHU_APP_ID` / `FEISHU_APP_SECRET`:可选,填写后启用飞书转发和 `/bind` 功能(包括 webhook 绑定所需的图片上传能力) -- `FEISHU_WEBHOOK_ALLOWED_HOSTS`:飞书 Webhook 域名白名单(防止 SSRF 攻击),默认仅允许 `open.feishu.cn` 和 `open.larksuite.com` +- `FEISHU_WEBHOOK_ALLOWED_HOSTS`:飞书 Webhook 域名白名单(JSON 格式,防止 SSRF 攻击) + - 不设置:使用默认白名单 `["open.feishu.cn", "open.larksuite.com"]` + - 设为 `[]`:禁用白名单校验(允许任意域名,**请谨慎使用**) + - 设为自定义列表:如 `["open.feishu.cn", "custom.domain.com"]` - 飞书需在应用后台开启机器人收发消息权限(im:message)以及获取与上传图片或文件资源权限(im:resource) ![lark_permission.png](docs/lark_permission.png) - 飞书事件添加接收消息(im.message.receive_v1)并启用长连接事件能力。 ![lark_event.png](docs/lark_event.png) diff --git a/src/stickerhub/config.py b/src/stickerhub/config.py index 3bdee9c..30d2a54 100644 --- a/src/stickerhub/config.py +++ b/src/stickerhub/config.py @@ -21,9 +21,15 @@ class Settings(BaseSettings): validation_alias=AliasChoices("BINDING_DB_PATH", "BINDING_STORE_PATH"), ) bind_magic_ttl_seconds: int = Field(default=600, alias="BIND_MAGIC_TTL_SECONDS") - feishu_webhook_allowed_hosts: list[str] = Field( - default=["open.feishu.cn", "open.larksuite.com"], + feishu_webhook_allowed_hosts: list[str] | None = Field( + default=None, alias="FEISHU_WEBHOOK_ALLOWED_HOSTS", + description=( + "飞书 Webhook 域名白名单(JSON 格式,如 " + '["open.feishu.cn","open.larksuite.com"])。' + "设为 null 或空列表 [] 禁用白名单校验。" + "不设置时使用默认白名单。" + ), ) log_level: str = Field(default="INFO", alias="LOG_LEVEL") @@ -33,3 +39,14 @@ class Settings(BaseSettings): case_sensitive=False, extra="ignore", ) + + def get_webhook_allowed_hosts(self) -> list[str] | None: + """ + 获取 webhook 域名白名单。 + - None: 使用默认白名单 ["open.feishu.cn", "open.larksuite.com"] + - []: 禁用白名单校验 + - [...]: 使用自定义白名单 + """ + if self.feishu_webhook_allowed_hosts is None: + return ["open.feishu.cn", "open.larksuite.com"] + return self.feishu_webhook_allowed_hosts diff --git a/src/stickerhub/main.py b/src/stickerhub/main.py index 414cee9..a77f5a4 100644 --- a/src/stickerhub/main.py +++ b/src/stickerhub/main.py @@ -27,7 +27,7 @@ async def async_main() -> None: binding_service = BindingService( store=BindingStore(settings.binding_db_path), magic_ttl_seconds=settings.bind_magic_ttl_seconds, - webhook_allowed_hosts=settings.feishu_webhook_allowed_hosts, + webhook_allowed_hosts=settings.get_webhook_allowed_hosts(), ) await binding_service.initialize() diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index b0d2fbd..07a8184 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -354,10 +354,8 @@ def __init__( ) -> None: self._store = store self._magic_ttl_seconds = magic_ttl_seconds - self._webhook_allowed_hosts = webhook_allowed_hosts or [ - "open.feishu.cn", - "open.larksuite.com", - ] + # None 表示使用默认白名单,[] 表示禁用白名单,其他表示自定义白名单 + self._webhook_allowed_hosts = webhook_allowed_hosts async def initialize(self) -> None: await self._store.ensure_initialized() @@ -441,6 +439,12 @@ async def handle_bind_webhook( await self._store.bind_platform(source_platform, source_user_id, hub_id) details = await self._store.bind_feishu_webhook(hub_id, normalized_url) + # 脱敏 previous_webhook 避免泄露旧凭据 + previous_webhook_masked = ( + mask_url(details["previous_webhook"]) + if details.get("previous_webhook") + else None + ) logger.info( ( "Webhook 绑定成功: source_platform=%s source_user=%s " @@ -448,7 +452,7 @@ async def handle_bind_webhook( ), source_platform, source_user_id, - details.get("previous_webhook"), + previous_webhook_masked, details.get("replaced_user_id"), ) return "绑定成功,已切换为飞书 Webhook 转发模式" @@ -503,12 +507,16 @@ async def get_feishu_target( return None -def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str]) -> str | None: +def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str] | None) -> str | None: """ 验证并归一化飞书 Webhook URL。 - 必须是 https 协议 - - 域名必须在白名单内(防止 SSRF) + - 域名必须在白名单内(防止 SSRF),除非 allowed_hosts 为空列表(禁用白名单) - 路径必须包含 /open-apis/bot/v2/hook/ + + Args: + url: 待验证的 webhook URL + allowed_hosts: 域名白名单。None 表示禁用白名单,[] 也表示禁用白名单,其他表示使用指定白名单 """ normalized = url.strip() if not normalized: @@ -517,12 +525,17 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str]) -> str | N parsed = urlparse(normalized) if parsed.scheme.lower() != "https": return None - if not parsed.netloc: + + # 必须有合法主机名 + if not parsed.hostname: return None - # 域名白名单校验(SSRF 防护) - if parsed.netloc.lower() not in [host.lower() for host in allowed_hosts]: - return None + # 域名白名单校验(SSRF 防护)——仅基于 hostname,不限制端口 + # allowed_hosts 为空列表时禁用白名单校验 + if allowed_hosts is not None and len(allowed_hosts) > 0: + hostname = parsed.hostname.lower() + if hostname not in [host.lower() for host in allowed_hosts]: + return None if "/open-apis/bot/v2/hook/" not in parsed.path: return None diff --git a/src/stickerhub/utils/url_masking.py b/src/stickerhub/utils/url_masking.py index dc3fb1b..6cc0106 100644 --- a/src/stickerhub/utils/url_masking.py +++ b/src/stickerhub/utils/url_masking.py @@ -12,17 +12,25 @@ def mask_url(url: str) -> str: """ 脱敏 URL 用于日志输出,避免泄露敏感 token。 - 仅保留协议、域名、路径前 20 个字符和后 8 个字符,中间用 ... 替代。 + 仅保留协议、域名(hostname)、路径前 20 个字符和后 8 个字符,中间用 ... 替代。 + 不包含 userinfo、端口、query、fragment 等敏感信息。 """ try: parsed = urlparse(url) + + # 仅在解析结果有明确的协议和主机名时才返回拼接后的 URL + if not parsed.scheme or not parsed.hostname: + return "[url_masked]" + if parsed.path and len(parsed.path) > PATH_MASK_THRESHOLD: masked_path = ( f"{parsed.path[:PATH_PREFIX_LENGTH]}...{parsed.path[-PATH_SUFFIX_LENGTH:]}" ) else: masked_path = parsed.path - return f"{parsed.scheme}://{parsed.netloc}{masked_path}" + + # 仅使用 hostname,避免将 userinfo、端口等敏感信息写入日志 + return f"{parsed.scheme}://{parsed.hostname}{masked_path}" except (ValueError, TypeError, AttributeError): # urlparse 可能抛出 ValueError,或传入 None 导致 TypeError/AttributeError return "[url_masked]" diff --git a/tests/test_binding_sqlite.py b/tests/test_binding_sqlite.py index cf20cb4..619d43c 100644 --- a/tests/test_binding_sqlite.py +++ b/tests/test_binding_sqlite.py @@ -146,11 +146,16 @@ async def _bind_webhook_domain_whitelist(db_path: str) -> None: ) await service.initialize() - # 合法域名应通过 + # 合法域名应通过(不限制端口) valid_url = "https://open.feishu.cn/open-apis/bot/v2/hook/valid_token" reply = await service.handle_bind_webhook("telegram", "tg_whitelist_ok", valid_url) assert "绑定成功" in reply + # 合法域名 + 自定义端口也应通过 + valid_url_with_port = "https://open.feishu.cn:8443/open-apis/bot/v2/hook/token_with_port" + reply = await service.handle_bind_webhook("telegram", "tg_whitelist_port", valid_url_with_port) + assert "绑定成功" in reply + # 不在白名单的域名应被拒绝(防止 SSRF) blocked_url = "https://evil.com/open-apis/bot/v2/hook/malicious" reply = await service.handle_bind_webhook("telegram", "tg_whitelist_block", blocked_url) @@ -162,6 +167,24 @@ async def _bind_webhook_domain_whitelist(db_path: str) -> None: assert "白名单" in reply +async def _bind_webhook_whitelist_disabled(db_path: str) -> None: + """测试禁用白名单校验""" + store = BindingStore(db_path) + # 空列表表示禁用白名单 + service = BindingService(store=store, magic_ttl_seconds=600, webhook_allowed_hosts=[]) + await service.initialize() + + # 任意域名都应通过(白名单已禁用) + custom_url = "https://custom.domain.com/open-apis/bot/v2/hook/custom_token" + reply = await service.handle_bind_webhook("telegram", "tg_no_whitelist", custom_url) + assert "绑定成功" in reply + + # 即使是非常规域名也应通过 + another_url = "https://example.org/open-apis/bot/v2/hook/another_token" + reply = await service.handle_bind_webhook("telegram", "tg_no_whitelist2", another_url) + assert "绑定成功" in reply + + def test_bind_flow_with_sqlite(tmp_path) -> None: db_path = tmp_path / "binding.db" asyncio.run(_bind_flow(str(db_path))) @@ -205,3 +228,8 @@ def test_bind_webhook_invalid_url(tmp_path) -> None: def test_bind_webhook_domain_whitelist(tmp_path) -> None: db_path = tmp_path / "binding.db" asyncio.run(_bind_webhook_domain_whitelist(str(db_path))) + + +def test_bind_webhook_whitelist_disabled(tmp_path) -> None: + db_path = tmp_path / "binding.db" + asyncio.run(_bind_webhook_whitelist_disabled(str(db_path))) From fb626623a0fa787012e5a06f5c02435f4d449548 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 12:15:25 +0000 Subject: [PATCH 09/10] refactor: simplify whitelist check logic and clarify docstring Co-authored-by: AkkiaS7 <68485070+AkkiaS7@users.noreply.github.com> --- src/stickerhub/services/binding.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index 07a8184..0636634 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -511,12 +511,15 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str] | None) -> """ 验证并归一化飞书 Webhook URL。 - 必须是 https 协议 - - 域名必须在白名单内(防止 SSRF),除非 allowed_hosts 为空列表(禁用白名单) + - 域名必须在白名单内(防止 SSRF),除非白名单为空列表(禁用白名单) - 路径必须包含 /open-apis/bot/v2/hook/ Args: url: 待验证的 webhook URL - allowed_hosts: 域名白名单。None 表示禁用白名单,[] 也表示禁用白名单,其他表示使用指定白名单 + allowed_hosts: 域名白名单。 + - None: 使用默认白名单 ["open.feishu.cn", "open.larksuite.com"] + - []: 禁用白名单校验(允许任意域名) + - [...]: 使用指定的自定义白名单 """ normalized = url.strip() if not normalized: @@ -531,8 +534,8 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str] | None) -> return None # 域名白名单校验(SSRF 防护)——仅基于 hostname,不限制端口 - # allowed_hosts 为空列表时禁用白名单校验 - if allowed_hosts is not None and len(allowed_hosts) > 0: + # allowed_hosts 为空列表或 None 时的处理在外层逻辑中已完成 + if allowed_hosts: hostname = parsed.hostname.lower() if hostname not in [host.lower() for host in allowed_hosts]: return None From 529f00021c794c78bf7cf5312388f3333981b6df Mon Sep 17 00:00:00 2001 From: Akkia Date: Fri, 13 Feb 2026 20:20:38 +0800 Subject: [PATCH 10/10] fix pre-commit error --- src/stickerhub/services/binding.py | 8 +++----- src/stickerhub/utils/url_masking.py | 2 +- 2 files changed, 4 insertions(+), 6 deletions(-) diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py index 0636634..90946e3 100644 --- a/src/stickerhub/services/binding.py +++ b/src/stickerhub/services/binding.py @@ -441,9 +441,7 @@ async def handle_bind_webhook( details = await self._store.bind_feishu_webhook(hub_id, normalized_url) # 脱敏 previous_webhook 避免泄露旧凭据 previous_webhook_masked = ( - mask_url(details["previous_webhook"]) - if details.get("previous_webhook") - else None + mask_url(details["previous_webhook"]) if details.get("previous_webhook") else None ) logger.info( ( @@ -513,7 +511,7 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str] | None) -> - 必须是 https 协议 - 域名必须在白名单内(防止 SSRF),除非白名单为空列表(禁用白名单) - 路径必须包含 /open-apis/bot/v2/hook/ - + Args: url: 待验证的 webhook URL allowed_hosts: 域名白名单。 @@ -528,7 +526,7 @@ def _normalize_feishu_webhook_url(url: str, allowed_hosts: list[str] | None) -> parsed = urlparse(normalized) if parsed.scheme.lower() != "https": return None - + # 必须有合法主机名 if not parsed.hostname: return None diff --git a/src/stickerhub/utils/url_masking.py b/src/stickerhub/utils/url_masking.py index 6cc0106..4f14cb5 100644 --- a/src/stickerhub/utils/url_masking.py +++ b/src/stickerhub/utils/url_masking.py @@ -17,7 +17,7 @@ def mask_url(url: str) -> str: """ try: parsed = urlparse(url) - + # 仅在解析结果有明确的协议和主机名时才返回拼接后的 URL if not parsed.scheme or not parsed.hostname: return "[url_masked]"