From 5dbd3b2bfc638a9b20b24becb0f2f22d1207d482 Mon Sep 17 00:00:00 2001
From: Akkia <AkkiaS7@Outlook.com>
Date: Fri, 13 Feb 2026 19:08:13 +0800
Subject: [PATCH 01/11] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E9=A3=9E?=
 =?UTF-8?q?=E4=B9=A6=20webhook=20=E7=BB=91=E5=AE=9A=E5=8A=9F=E8=83=BD?=
 =?UTF-8?q?=EF=BC=8C=E6=94=AF=E6=8C=81=E9=80=89=E6=8B=A9=E7=BB=91=E5=AE=9A?=
 =?UTF-8?q?=E6=96=B9=E5=BC=8F?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .github/copilot-instructions.md            |   6 +-
 README.md                                  |  26 ++-
 src/stickerhub/adapters/feishu_sender.py   |  71 +++++++-
 src/stickerhub/adapters/telegram_source.py | 148 ++++++++++++++++-
 src/stickerhub/core/ports.py               |   2 +-
 src/stickerhub/main.py                     |  29 ++--
 src/stickerhub/services/binding.py         | 178 +++++++++++++++++++++
 src/stickerhub/services/relay.py           |  14 +-
 tests/test_binding_sqlite.py               |  82 ++++++++++
 tests/test_telegram_source.py              |  21 +++
 10 files changed, 544 insertions(+), 33 deletions(-)

diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
index 5562bb6..ac26bd1 100644
--- a/.github/copilot-instructions.md
+++ b/.github/copilot-instructions.md
@@ -26,9 +26,13 @@
 
 ## 不可破坏的业务行为
 
-- `/bind`（无参数）应创建或复用当前平台身份，并返回魔法字符串。
+- 配置 `FEISHU_APP_ID`/`FEISHU_APP_SECRET` 时，Telegram `/bind`（无参数）应先展示两种绑定方式按钮：`飞书机器人`、`飞书 Webhook`。
+- 选择 `飞书机器人` 后，流程与原有魔法字符串一致：生成魔法字符串并提示在飞书机器人执行 `/bind <code>`。
+- 选择 `飞书 Webhook` 后，应提示用户输入 webhook 地址并完成绑定。
 - `/bind <code>` 应将当前平台账号绑定到该魔法字符串对应身份。
 - 绑定冲突策略为可覆盖：以魔法字符串所属身份为高优先级。
+- 飞书机器人绑定与 webhook 绑定互斥，切换方式后仅新方式生效。
+- 未配置 `FEISHU_APP_ID`/`FEISHU_APP_SECRET` 时，不支持 webhook 绑定（与飞书能力关闭保持一致）。
 - Telegram 单贴纸流程必须保持两步：
   - 先立即转发表情。
   - 再提示是否发送整包。
diff --git a/README.md b/README.md
index c3b52e0..ebc3211 100644
--- a/README.md
+++ b/README.md
@@ -22,7 +22,8 @@ StickerHub 是一个 Telegram 表情素材（Sticker/Image/GIF/Video）转换与
 
 配置飞书应用后，额外支持：
 
-- 双端绑定机制（`/bind`），将 Telegram 身份与飞书身份关联
+- `/bind` 双模式绑定（飞书机器人 / 飞书自定义机器人 Webhook）
+- 绑定模式二选一：切换后仅新配置生效
 - 单个贴纸自动转发到飞书
 - 整包发送时可选「📤 发送到飞书」，按每批 10 个并发发送
 - 飞书事件接收方式：**长连接（Long Connection）**，项目不对外开放 HTTP 端口
@@ -60,7 +61,7 @@ LOG_LEVEL=INFO
 说明：
 
 - `TELEGRAM_BOT_API_TOKEN`：必填，Telegram Bot API Token
-- `FEISHU_APP_ID` / `FEISHU_APP_SECRET`：可选，填写后启用飞书转发和 `/bind` 功能
+- `FEISHU_APP_ID` / `FEISHU_APP_SECRET`：可选，填写后启用飞书转发和 `/bind` 功能（包括 webhook 绑定所需的图片上传能力）
 - 飞书需在应用后台开启机器人收发消息权限(im:message)以及获取与上传图片或文件资源权限(im:resource) ![lark_permission.png](docs/lark_permission.png)
 - 飞书事件添加接收消息(im.message.receive_v1)并启用长连接事件能力。 ![lark_event.png](docs/lark_event.png)
 
@@ -91,12 +92,21 @@ docker compose -f docker-compose.local.yml up -d --build
 docker compose -f docker-compose.local.yml logs -f stickerhub
 ```
 
-## 绑定流程（需配置飞书）
-![20260213-050307.gif](../../Downloads/20260213-050307.gif)
-1. 在 Telegram 或飞书任一端发送：`/bind`
-2. 机器人返回一串魔法字符串
-3. 在另一端发送：`/bind <魔法字符串>`
-4. 绑定成功后，Telegram 发来的素材会路由到对应飞书身份
+## 绑定流程（需配置飞书 App）
+
+1. 在 Telegram 发送：`/bind`
+2. 选择绑定方式：
+   - `飞书机器人`：走魔法字符串绑定
+   - `飞书 Webhook`：输入飞书自定义机器人的 webhook 地址
+3. `飞书机器人`流程：
+   - Telegram 返回魔法字符串
+   - 在飞书机器人里发送 `/bind <魔法字符串>` 完成绑定
+4. `飞书 Webhook`流程：
+   - Telegram 提示输入 webhook 地址
+   - 输入后立即生效，后续素材通过 webhook 发送
+5. 两种绑定方式互斥，切换后仅新方式生效
+
+> 注意：未配置 `FEISHU_APP_ID` / `FEISHU_APP_SECRET` 时，不支持飞书 webhook 绑定与飞书转发。
 
 ## 已知限制
 
diff --git a/src/stickerhub/adapters/feishu_sender.py b/src/stickerhub/adapters/feishu_sender.py
index 6b9c1a4..e2de91b 100644
--- a/src/stickerhub/adapters/feishu_sender.py
+++ b/src/stickerhub/adapters/feishu_sender.py
@@ -18,10 +18,11 @@ def __init__(
         self._app_secret = app_secret
         self._base_url = "https://open.feishu.cn/open-apis"
 
-    async def send(self, asset: StickerAsset, target_user_id: str) -> None:
+    async def send(self, asset: StickerAsset, target_mode: str, target: str) -> None:
         logger.debug(
-            "准备发送图片到飞书: receive_id=%s file=%s mime=%s size=%s",
-            target_user_id,
+            "准备发送图片到飞书: mode=%s target=%s file=%s mime=%s size=%s",
+            target_mode,
+            target,
             asset.file_name,
             asset.mime_type,
             len(asset.content),
@@ -29,7 +30,13 @@ async def send(self, asset: StickerAsset, target_user_id: str) -> None:
         async with httpx.AsyncClient(timeout=30.0) as client:
             token = await self._get_tenant_token(client)
             image_key = await self._upload_image(client, token, asset)
-            await self._send_image_message(client, token, image_key, target_user_id)
+            if target_mode == "bot":
+                await self._send_image_message(client, token, image_key, target)
+                return
+            if target_mode == "webhook":
+                await self._send_webhook_image(client, image_key=image_key, webhook_url=target)
+                return
+            raise RuntimeError(f"不支持的飞书目标类型: {target_mode}")
 
     async def send_text(
         self,
@@ -62,6 +69,18 @@ async def send_text(
                 raise RuntimeError(f"发送飞书文本消息失败: {payload}")
             logger.info("飞书文本消息已发送: receive_id=%s", receive_id)
 
+    async def send_webhook_text(self, text: str, webhook_url: str) -> None:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            await self._send_webhook_message(
+                client,
+                webhook_url=webhook_url,
+                message={
+                    "msg_type": "text",
+                    "content": {"text": text},
+                },
+            )
+        logger.info("飞书 webhook 文本消息已发送")
+
     async def _get_tenant_token(self, client: httpx.AsyncClient) -> str:
         response = await client.post(
             f"{self._base_url}/auth/v3/tenant_access_token/internal",
@@ -129,3 +148,47 @@ async def _send_image_message(
             raise RuntimeError(f"发送飞书消息失败: {payload}")
 
         logger.info("素材已发送到飞书: receive_id=%s", receive_id)
+
+    async def _send_webhook_image(
+        self,
+        client: httpx.AsyncClient,
+        image_key: str,
+        webhook_url: str,
+    ) -> None:
+        await self._send_webhook_message(
+            client,
+            webhook_url=webhook_url,
+            message={
+                "msg_type": "image",
+                "content": {"image_key": image_key},
+            },
+        )
+        logger.info("素材已发送到飞书 webhook")
+
+    async def _send_webhook_message(
+        self,
+        client: httpx.AsyncClient,
+        webhook_url: str,
+        message: dict[str, object],
+    ) -> None:
+        response = await client.post(
+            webhook_url,
+            headers={"Content-Type": "application/json"},
+            json=message,
+        )
+        try:
+            payload = response.json()
+        except json.JSONDecodeError:
+            payload = {"raw_body": response.text}
+
+        if response.status_code != 200:
+            raise RuntimeError(
+                "发送飞书 webhook 消息失败: " f"status={response.status_code} payload={payload}"
+            )
+
+        status_code = payload.get("StatusCode")
+        code = payload.get("code")
+        status_ok = status_code in (None, 0, "0")
+        code_ok = code in (None, 0, "0")
+        if not status_ok or not code_ok:
+            raise RuntimeError(f"发送飞书 webhook 消息失败: {payload}")
diff --git a/src/stickerhub/adapters/telegram_source.py b/src/stickerhub/adapters/telegram_source.py
index e822fe4..b2ef8c0 100644
--- a/src/stickerhub/adapters/telegram_source.py
+++ b/src/stickerhub/adapters/telegram_source.py
@@ -34,12 +34,15 @@
 
 AssetHandler = Callable[[StickerAsset], Awaitable[None]]
 BindHandler = Callable[[str, str | None], Awaitable[str]]
+WebhookBindHandler = Callable[[str, str], Awaitable[str]]
 PackBatchMarkerHandler = Callable[[str, int, int, int, int, str], Awaitable[None]]
 NormalizeHandler = Callable[[StickerAsset], Awaitable[StickerAsset]]
 
 PACK_CALLBACK_PREFIX = "send_pack:"
 STOP_PACK_CALLBACK_PREFIX = "stop_pack:"
+BIND_MODE_CALLBACK_PREFIX = "bind_mode:"
 PACK_REQUEST_TTL_SECONDS = 15 * 60
+WEBHOOK_BIND_REQUEST_TTL_SECONDS = 10 * 60
 PACK_BATCH_SIZE = 10
 
 
@@ -64,6 +67,12 @@ class RunningStickerPackTask:
     cancel_requested: bool = False
 
 
+@dataclass(slots=True)
+class PendingWebhookBindRequest:
+    telegram_user_id: str
+    created_at: int
+
+
 def build_telegram_usage_text(feishu_enabled: bool = True) -> str:
     lines = [
         "StickerHub 使用说明：",
@@ -81,7 +90,7 @@ def build_telegram_usage_text(feishu_enabled: bool = True) -> str:
         ]
     )
     if feishu_enabled:
-        lines.append("/bind [魔法字符串]")
+        lines.append("/bind（选择飞书机器人或 webhook 绑定）")
     return "\n".join(lines)
 
 
@@ -91,7 +100,7 @@ def get_telegram_bot_commands(feishu_enabled: bool = True) -> list[BotCommand]:
         BotCommand("start", "开始使用并查看说明"),
     ]
     if feishu_enabled:
-        commands.insert(0, BotCommand("bind", "绑定账号：/bind 或 /bind <code>"))
+        commands.insert(0, BotCommand("bind", "绑定飞书：/bind 或 /bind <code>"))
     return commands
 
 
@@ -99,6 +108,7 @@ def build_telegram_application(
     token: str,
     on_asset: AssetHandler,
     on_bind: BindHandler,
+    on_bind_webhook: WebhookBindHandler | None = None,
     on_pack_batch_marker: PackBatchMarkerHandler | None = None,
     on_normalize: NormalizeHandler | None = None,
     feishu_enabled: bool = True,
@@ -106,6 +116,7 @@ def build_telegram_application(
     application = Application.builder().token(token).build()
     pending_pack_requests: dict[str, PendingStickerPackRequest] = {}
     running_pack_tasks: dict[str, RunningStickerPackTask] = {}
+    pending_webhook_requests: dict[str, PendingWebhookBindRequest] = {}
 
     async def handle_bind(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         if not update.message or not update.effective_user:
@@ -113,12 +124,65 @@ async def handle_bind(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non
 
         try:
             arg = context.args[0] if context.args else None
-            reply = await on_bind(str(update.effective_user.id), arg)
-            await update.message.reply_text(reply)
+            telegram_user_id = str(update.effective_user.id)
+            if arg:
+                reply = await on_bind(telegram_user_id, arg)
+                await update.message.reply_text(reply)
+                return
+
+            if not feishu_enabled or on_bind_webhook is None:
+                reply = await on_bind(telegram_user_id, None)
+                await update.message.reply_text(reply)
+                return
+
+            await update.message.reply_text(
+                "请选择飞书绑定方式：",
+                reply_markup=_build_bind_mode_keyboard(telegram_user_id),
+            )
         except Exception as exc:  # noqa: BLE001
             logger.exception("处理 Telegram /bind 失败")
             await update.message.reply_text(f"绑定失败: {exc}")
 
+    async def handle_bind_mode_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        del context
+        query = update.callback_query
+        if not query:
+            return
+        await query.answer()
+
+        data = query.data or ""
+        parsed = _parse_bind_mode_callback_data(data)
+        if not parsed:
+            return
+
+        mode, owner_telegram_user_id = parsed
+        effective_user = update.effective_user
+        if not effective_user or str(effective_user.id) != owner_telegram_user_id:
+            await query.answer("仅发起绑定的用户可以点击该按钮", show_alert=True)
+            return
+
+        if mode == "bot":
+            reply = await on_bind(owner_telegram_user_id, None)
+            await query.edit_message_text(
+                f"{reply}\n\n请在飞书机器人里发送上面的 /bind 命令完成绑定。"
+            )
+            return
+
+        if mode == "webhook":
+            if on_bind_webhook is None:
+                await query.answer("当前未启用 webhook 绑定", show_alert=True)
+                return
+
+            pending_webhook_requests[owner_telegram_user_id] = PendingWebhookBindRequest(
+                telegram_user_id=owner_telegram_user_id,
+                created_at=int(time.time()),
+            )
+            await query.edit_message_text(
+                "请直接发送飞书自定义机器人的 Webhook 地址。\n"
+                "示例：\n"
+                "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx"
+            )
+
     async def handle_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         del context
         if not update.message:
@@ -310,6 +374,31 @@ async def handle_unsupported_message(
         del context
         if not update.message:
             return
+
+        _cleanup_pending_webhook_requests(pending_webhook_requests)
+
+        if update.effective_user:
+            telegram_user_id = str(update.effective_user.id)
+            pending = pending_webhook_requests.get(telegram_user_id)
+            if pending:
+                if not update.message.text:
+                    await update.message.reply_text(
+                        "正在等待你输入飞书 Webhook 地址，请发送文本链接，或重新输入 /bind。"
+                    )
+                    return
+
+                pending_webhook_requests.pop(telegram_user_id, None)
+                try:
+                    if on_bind_webhook is None:
+                        await update.message.reply_text("当前未启用 webhook 绑定")
+                        return
+                    reply = await on_bind_webhook(telegram_user_id, update.message.text.strip())
+                    await update.message.reply_text(reply)
+                except Exception as exc:  # noqa: BLE001
+                    logger.exception("处理 Telegram webhook 绑定失败")
+                    await update.message.reply_text(f"绑定失败: {exc}")
+                return
+
         await update.message.reply_text(
             "暂不支持该消息类型。\n\n" + build_telegram_usage_text(feishu_enabled)
         )
@@ -327,6 +416,12 @@ async def handle_unsupported_message(
 
     if feishu_enabled:
         application.add_handler(CommandHandler("bind", handle_bind))
+        application.add_handler(
+            CallbackQueryHandler(
+                handle_bind_mode_callback,
+                pattern=rf"^{BIND_MODE_CALLBACK_PREFIX}",
+            )
+        )
     application.add_handler(CommandHandler("help", handle_help))
     application.add_handler(CommandHandler("start", handle_help))
     application.add_handler(
@@ -538,6 +633,23 @@ def _build_stop_keyboard(task_id: str) -> InlineKeyboardMarkup:
     )
 
 
+def _build_bind_mode_keyboard(telegram_user_id: str) -> InlineKeyboardMarkup:
+    return InlineKeyboardMarkup(
+        [
+            [
+                InlineKeyboardButton(
+                    text="飞书机器人",
+                    callback_data=f"{BIND_MODE_CALLBACK_PREFIX}bot:{telegram_user_id}",
+                ),
+                InlineKeyboardButton(
+                    text="飞书 Webhook",
+                    callback_data=f"{BIND_MODE_CALLBACK_PREFIX}webhook:{telegram_user_id}",
+                ),
+            ]
+        ]
+    )
+
+
 async def _offer_send_pack_button(
     message: Message,
     context: ContextTypes.DEFAULT_TYPE,
@@ -798,6 +910,19 @@ def _cleanup_pending_requests(pending_pack_requests: dict[str, PendingStickerPac
         pending_pack_requests.pop(token, None)
 
 
+def _cleanup_pending_webhook_requests(
+    pending_webhook_requests: dict[str, PendingWebhookBindRequest],
+) -> None:
+    now = int(time.time())
+    expired_users = [
+        user_id
+        for user_id, req in pending_webhook_requests.items()
+        if now - req.created_at > WEBHOOK_BIND_REQUEST_TTL_SECONDS
+    ]
+    for user_id in expired_users:
+        pending_webhook_requests.pop(user_id, None)
+
+
 def _safe_pack_name(set_name: str) -> str:
     return re.sub(r"[^A-Za-z0-9_-]", "_", set_name)
 
@@ -816,6 +941,21 @@ def _parse_pack_callback_data(data: str) -> tuple[str, str] | None:
     return mode, token
 
 
+def _parse_bind_mode_callback_data(data: str) -> tuple[str, str] | None:
+    if not data.startswith(BIND_MODE_CALLBACK_PREFIX):
+        return None
+    rest = data[len(BIND_MODE_CALLBACK_PREFIX) :]
+    parts = rest.split(":", 1)
+    if len(parts) != 2:
+        return None
+    mode, telegram_user_id = parts
+    if mode not in {"bot", "webhook"}:
+        return None
+    if not telegram_user_id:
+        return None
+    return mode, telegram_user_id
+
+
 async def _identity_normalize(asset: StickerAsset) -> StickerAsset:
     """无操作归一化器，原样返回素材。"""
     return asset
diff --git a/src/stickerhub/core/ports.py b/src/stickerhub/core/ports.py
index f32c1ed..b5b91e6 100644
--- a/src/stickerhub/core/ports.py
+++ b/src/stickerhub/core/ports.py
@@ -9,5 +9,5 @@ async def normalize(self, asset: StickerAsset) -> StickerAsset:
 
 
 class TargetPlatformSender(Protocol):
-    async def send(self, asset: StickerAsset, target_user_id: str) -> None:
+    async def send(self, asset: StickerAsset, target_mode: str, target: str) -> None:
         """将素材发送到目标平台。"""
diff --git a/src/stickerhub/main.py b/src/stickerhub/main.py
index c95b419..ed1a0eb 100644
--- a/src/stickerhub/main.py
+++ b/src/stickerhub/main.py
@@ -56,23 +56,29 @@ async def _pack_batch_marker(
             end_index: int,
             set_name: str,
         ) -> None:
-            target_user_id = await binding_service.get_target_user_id(
+            target = await binding_service.get_feishu_target(
                 source_platform="telegram",
                 source_user_id=source_user_id,
-                target_platform="feishu",
             )
-            if not target_user_id:
+            if not target:
                 raise RuntimeError("当前 Telegram 账号未绑定飞书。请先执行 /bind 完成跨平台绑定")
 
             assert feishu_sender is not None  # narrowing for type checker
-            await feishu_sender.send_text(
-                text=(
-                    f"—— 表情包《{set_name}》批次 {batch_no}/{total_batches} "
-                    f"（第 {start_index}-{end_index} 个） ——"
-                ),
-                receive_id=target_user_id,
-                receive_id_type="open_id",
+            marker_text = (
+                f"—— 表情包《{set_name}》批次 {batch_no}/{total_batches} "
+                f"（第 {start_index}-{end_index} 个） ——"
             )
+            if target.mode == "bot":
+                await feishu_sender.send_text(
+                    text=marker_text,
+                    receive_id=target.target,
+                    receive_id_type="open_id",
+                )
+            else:
+                await feishu_sender.send_webhook_text(
+                    text=marker_text,
+                    webhook_url=target.target,
+                )
 
         on_pack_batch_marker = _pack_batch_marker
 
@@ -80,6 +86,9 @@ async def _pack_batch_marker(
         token=settings.telegram_bot_api_token,
         on_asset=relay_use_case.relay,
         on_bind=lambda user_id, arg: binding_service.handle_bind_command("telegram", user_id, arg),
+        on_bind_webhook=lambda user_id, url: binding_service.handle_bind_webhook(
+            "telegram", user_id, url
+        ),
         on_pack_batch_marker=on_pack_batch_marker,
         on_normalize=normalizer.normalize,
         feishu_enabled=feishu_enabled,
diff --git a/src/stickerhub/services/binding.py b/src/stickerhub/services/binding.py
index 5b7fc9e..d1b376f 100644
--- a/src/stickerhub/services/binding.py
+++ b/src/stickerhub/services/binding.py
@@ -4,11 +4,20 @@
 import sqlite3
 import time
 import uuid
+from dataclasses import dataclass
 from pathlib import Path
+from typing import Literal
+from urllib.parse import urlparse
 
 logger = logging.getLogger(__name__)
 
 
+@dataclass(slots=True)
+class FeishuTarget:
+    mode: Literal["bot", "webhook"]
+    target: str
+
+
 class BindingStore:
     def __init__(self, db_path: str) -> None:
         self._db_path = Path(db_path)
@@ -40,6 +49,13 @@ async def ensure_initialized(self) -> None:
                         created_at INTEGER NOT NULL,
                         used_at INTEGER
                     );
+
+                    CREATE TABLE IF NOT EXISTS feishu_webhook_bindings (
+                        hub_id TEXT PRIMARY KEY,
+                        webhook_url TEXT NOT NULL,
+                        created_at INTEGER NOT NULL,
+                        updated_at INTEGER NOT NULL
+                    );
                     """
                 )
                 conn.commit()
@@ -237,6 +253,90 @@ async def get_platform_user_id(self, platform: str, hub_id: str) -> str | None:
 
         return str(row["platform_user_id"]) if row else None
 
+    async def bind_feishu_webhook(self, hub_id: str, webhook_url: str) -> dict[str, str | None]:
+        now = int(time.time())
+        async with self._lock:
+            with self._connect() as conn:
+                webhook_row = conn.execute(
+                    """
+                    SELECT webhook_url
+                    FROM feishu_webhook_bindings
+                    WHERE hub_id = ?
+                    """,
+                    (hub_id,),
+                ).fetchone()
+                previous_webhook = str(webhook_row["webhook_url"]) if webhook_row else None
+
+                replaced_row = conn.execute(
+                    """
+                    SELECT platform_user_id
+                    FROM platform_bindings
+                    WHERE platform = 'feishu' AND hub_id = ?
+                    LIMIT 1
+                    """,
+                    (hub_id,),
+                ).fetchone()
+                replaced_user_id = str(replaced_row["platform_user_id"]) if replaced_row else None
+
+                conn.execute(
+                    """
+                    DELETE FROM platform_bindings
+                    WHERE platform = 'feishu' AND hub_id = ?
+                    """,
+                    (hub_id,),
+                )
+
+                conn.execute(
+                    """
+                    INSERT INTO feishu_webhook_bindings(
+                        hub_id, webhook_url, created_at, updated_at
+                    )
+                    VALUES (?, ?, ?, ?)
+                    ON CONFLICT(hub_id)
+                    DO UPDATE SET
+                        webhook_url = excluded.webhook_url,
+                        updated_at = excluded.updated_at
+                    """,
+                    (hub_id, webhook_url, now, now),
+                )
+                conn.commit()
+
+        logger.info("飞书 webhook 绑定成功: hub_id=%s replaced_user=%s", hub_id, replaced_user_id)
+        return {
+            "previous_webhook": previous_webhook,
+            "replaced_user_id": replaced_user_id,
+        }
+
+    async def clear_feishu_webhook(self, hub_id: str) -> bool:
+        async with self._lock:
+            with self._connect() as conn:
+                cursor = conn.execute(
+                    """
+                    DELETE FROM feishu_webhook_bindings
+                    WHERE hub_id = ?
+                    """,
+                    (hub_id,),
+                )
+                conn.commit()
+                deleted = cursor.rowcount > 0
+
+        if deleted:
+            logger.info("已清理飞书 webhook 绑定: hub_id=%s", hub_id)
+        return deleted
+
+    async def get_feishu_webhook(self, hub_id: str) -> str | None:
+        async with self._lock:
+            with self._connect() as conn:
+                row = conn.execute(
+                    """
+                    SELECT webhook_url
+                    FROM feishu_webhook_bindings
+                    WHERE hub_id = ?
+                    """,
+                    (hub_id,),
+                ).fetchone()
+        return str(row["webhook_url"]) if row else None
+
     def _connect(self) -> sqlite3.Connection:
         conn = sqlite3.connect(self._db_path)
         conn.row_factory = sqlite3.Row
@@ -290,6 +390,8 @@ async def handle_bind_command(
             return f"绑定失败: {reason}"
 
         details = await self._store.force_bind_platform(platform, platform_user_id, hub_id)
+        if platform == "feishu":
+            await self._store.clear_feishu_webhook(hub_id)
         logger.info(
             "绑定成功(覆盖模式): platform=%s user=%s previous_hub=%s replaced_user=%s",
             platform,
@@ -299,6 +401,43 @@ async def handle_bind_command(
         )
         return "绑定成功，已更新当前平台身份映射"
 
+    async def handle_bind_webhook(
+        self,
+        source_platform: str,
+        source_user_id: str,
+        webhook_url: str,
+    ) -> str:
+        normalized_url = _normalize_feishu_webhook_url(webhook_url)
+        if not normalized_url:
+            logger.warning(
+                "Webhook 绑定失败: 平台=%s user=%s 原因=URL格式不合法",
+                source_platform,
+                source_user_id,
+            )
+            return (
+                "绑定失败: Webhook 地址格式不合法。\n"
+                "请填写飞书自定义机器人 Webhook 地址，例如：\n"
+                "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx"
+            )
+
+        hub_id = await self._store.get_hub_id(source_platform, source_user_id)
+        if not hub_id:
+            hub_id = uuid.uuid4().hex
+            await self._store.bind_platform(source_platform, source_user_id, hub_id)
+
+        details = await self._store.bind_feishu_webhook(hub_id, normalized_url)
+        logger.info(
+            (
+                "Webhook 绑定成功: source_platform=%s source_user=%s "
+                "previous_webhook=%s replaced_user=%s"
+            ),
+            source_platform,
+            source_user_id,
+            details.get("previous_webhook"),
+            details.get("replaced_user_id"),
+        )
+        return "绑定成功，已切换为飞书 Webhook 转发模式"
+
     async def get_target_user_id(
         self,
         source_platform: str,
@@ -323,3 +462,42 @@ async def get_target_user_id(
             bool(target_user_id),
         )
         return target_user_id
+
+    async def get_feishu_target(
+        self,
+        source_platform: str,
+        source_user_id: str,
+    ) -> FeishuTarget | None:
+        hub_id = await self._store.get_hub_id(source_platform, source_user_id)
+        if not hub_id:
+            logger.debug(
+                "未找到源平台绑定: source_platform=%s source_user_id=%s",
+                source_platform,
+                source_user_id,
+            )
+            return None
+
+        webhook = await self._store.get_feishu_webhook(hub_id)
+        if webhook:
+            return FeishuTarget(mode="webhook", target=webhook)
+
+        open_id = await self._store.get_platform_user_id("feishu", hub_id)
+        if open_id:
+            return FeishuTarget(mode="bot", target=open_id)
+
+        return None
+
+
+def _normalize_feishu_webhook_url(url: str) -> str | None:
+    normalized = url.strip()
+    if not normalized:
+        return None
+
+    parsed = urlparse(normalized)
+    if parsed.scheme.lower() != "https":
+        return None
+    if not parsed.netloc:
+        return None
+    if "/open-apis/bot/v2/hook/" not in parsed.path:
+        return None
+    return normalized
diff --git a/src/stickerhub/services/relay.py b/src/stickerhub/services/relay.py
index c14b35a..8b9d481 100644
--- a/src/stickerhub/services/relay.py
+++ b/src/stickerhub/services/relay.py
@@ -33,12 +33,11 @@ async def relay(self, asset: StickerAsset) -> None:
             logger.debug("未配置目标平台发送器，跳过飞书转发")
             return
 
-        target_user_id = await self._binding_service.get_target_user_id(
+        target = await self._binding_service.get_feishu_target(
             source_platform=asset.source_platform,
             source_user_id=asset.source_user_id,
-            target_platform="feishu",
         )
-        if not target_user_id:
+        if not target:
             logger.info(
                 "用户未绑定飞书，跳过飞书转发: user=%s",
                 asset.source_user_id,
@@ -46,11 +45,16 @@ async def relay(self, asset: StickerAsset) -> None:
             return
 
         normalized = await self._normalizer.normalize(asset)
-        await self._target_sender.send(normalized, target_user_id=target_user_id)
+        await self._target_sender.send(
+            normalized,
+            target_mode=target.mode,
+            target=target.target,
+        )
         logger.info(
-            "转发成功: source=%s user=%s kind=%s mime=%s",
+            "转发成功: source=%s user=%s mode=%s kind=%s mime=%s",
             asset.source_platform,
             asset.source_user_id,
+            target.mode,
             normalized.media_kind,
             normalized.mime_type,
         )
diff --git a/tests/test_binding_sqlite.py b/tests/test_binding_sqlite.py
index 6efde4c..9e88bb0 100644
--- a/tests/test_binding_sqlite.py
+++ b/tests/test_binding_sqlite.py
@@ -75,6 +75,68 @@ async def _rebind_replaces_existing_account_on_same_hub(db_path: str) -> None:
     assert await service.get_target_user_id("telegram", "tg_x", "feishu") == "ou_new"
 
 
+async def _bind_webhook_flow(db_path: str) -> None:
+    store = BindingStore(db_path)
+    service = BindingService(store=store, magic_ttl_seconds=600)
+    await service.initialize()
+
+    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/test_webhook"
+    reply = await service.handle_bind_webhook("telegram", "tg_webhook", webhook_url)
+    assert "绑定成功" in reply
+
+    target = await service.get_feishu_target("telegram", "tg_webhook")
+    assert target is not None
+    assert target.mode == "webhook"
+    assert target.target == webhook_url
+    assert await service.get_target_user_id("telegram", "tg_webhook", "feishu") is None
+
+
+async def _switch_from_bot_to_webhook(db_path: str) -> None:
+    store = BindingStore(db_path)
+    service = BindingService(store=store, magic_ttl_seconds=600)
+    await service.initialize()
+
+    tg_reply = await service.handle_bind_command("telegram", "tg_switch", None)
+    code = re.search(r"/bind\s+([A-Z0-9]+)", tg_reply).group(1)  # type: ignore[union-attr]
+    assert "绑定成功" in await service.handle_bind_command("feishu", "ou_switch_old", code)
+
+    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/switch_webhook"
+    assert "绑定成功" in await service.handle_bind_webhook("telegram", "tg_switch", webhook_url)
+
+    target = await service.get_feishu_target("telegram", "tg_switch")
+    assert target is not None
+    assert target.mode == "webhook"
+    assert target.target == webhook_url
+    assert await service.get_target_user_id("telegram", "tg_switch", "feishu") is None
+
+
+async def _switch_from_webhook_to_bot(db_path: str) -> None:
+    store = BindingStore(db_path)
+    service = BindingService(store=store, magic_ttl_seconds=600)
+    await service.initialize()
+
+    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/switch_back"
+    assert "绑定成功" in await service.handle_bind_webhook("telegram", "tg_back", webhook_url)
+
+    tg_reply = await service.handle_bind_command("telegram", "tg_back", None)
+    code = re.search(r"/bind\s+([A-Z0-9]+)", tg_reply).group(1)  # type: ignore[union-attr]
+    assert "绑定成功" in await service.handle_bind_command("feishu", "ou_new", code)
+
+    target = await service.get_feishu_target("telegram", "tg_back")
+    assert target is not None
+    assert target.mode == "bot"
+    assert target.target == "ou_new"
+
+
+async def _bind_webhook_invalid_url(db_path: str) -> None:
+    store = BindingStore(db_path)
+    service = BindingService(store=store, magic_ttl_seconds=600)
+    await service.initialize()
+
+    reply = await service.handle_bind_webhook("telegram", "tg_invalid", "http://example.com/abc")
+    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)))
@@ -93,3 +155,23 @@ def test_rebind_current_account_to_new_hub(tmp_path) -> None:
 def test_rebind_replaces_existing_account_on_same_hub(tmp_path) -> None:
     db_path = tmp_path / "binding.db"
     asyncio.run(_rebind_replaces_existing_account_on_same_hub(str(db_path)))
+
+
+def test_bind_webhook_flow(tmp_path) -> None:
+    db_path = tmp_path / "binding.db"
+    asyncio.run(_bind_webhook_flow(str(db_path)))
+
+
+def test_switch_from_bot_to_webhook(tmp_path) -> None:
+    db_path = tmp_path / "binding.db"
+    asyncio.run(_switch_from_bot_to_webhook(str(db_path)))
+
+
+def test_switch_from_webhook_to_bot(tmp_path) -> None:
+    db_path = tmp_path / "binding.db"
+    asyncio.run(_switch_from_webhook_to_bot(str(db_path)))
+
+
+def test_bind_webhook_invalid_url(tmp_path) -> None:
+    db_path = tmp_path / "binding.db"
+    asyncio.run(_bind_webhook_invalid_url(str(db_path)))
diff --git a/tests/test_telegram_source.py b/tests/test_telegram_source.py
index af07aef..7168071 100644
--- a/tests/test_telegram_source.py
+++ b/tests/test_telegram_source.py
@@ -3,6 +3,7 @@
 from types import SimpleNamespace
 
 from stickerhub.adapters.telegram_source import (
+    BIND_MODE_CALLBACK_PREFIX,
     PACK_CALLBACK_PREFIX,
     PendingStickerPackRequest,
     RunningStickerPackTask,
@@ -10,6 +11,7 @@
     _deduplicate_filename,
     _detect_sticker_mime,
     _has_running_task_for_user,
+    _parse_bind_mode_callback_data,
     _parse_pack_callback_data,
     _send_single_sticker_reply,
 )
@@ -111,6 +113,25 @@ def test_token_with_colon(self) -> None:
         assert result == ("zip", "token:with:colons")
 
 
+class TestParseBindModeCallbackData:
+    def test_valid_bot_mode(self) -> None:
+        result = _parse_bind_mode_callback_data(f"{BIND_MODE_CALLBACK_PREFIX}bot:123")
+        assert result == ("bot", "123")
+
+    def test_valid_webhook_mode(self) -> None:
+        result = _parse_bind_mode_callback_data(f"{BIND_MODE_CALLBACK_PREFIX}webhook:456")
+        assert result == ("webhook", "456")
+
+    def test_invalid_prefix(self) -> None:
+        assert _parse_bind_mode_callback_data("wrong:bot:123") is None
+
+    def test_invalid_mode(self) -> None:
+        assert _parse_bind_mode_callback_data(f"{BIND_MODE_CALLBACK_PREFIX}unknown:123") is None
+
+    def test_missing_user_id(self) -> None:
+        assert _parse_bind_mode_callback_data(f"{BIND_MODE_CALLBACK_PREFIX}bot:") is None
+
+
 class TestDeduplicateFilename:
     def test_no_conflict(self) -> None:
         assert _deduplicate_filename("sticker.png", set()) == "sticker.png"

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 02/11] 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 03/11] 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 04/11] 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 05/11] 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 06/11] 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 07/11] 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 08/11] 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 09/11] 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 10/11] 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 <AkkiaS7@Outlook.com>
Date: Fri, 13 Feb 2026 20:20:38 +0800
Subject: [PATCH 11/11] 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]"