feat(telegram): accept signed callback forwards

apps/api/src/api/v1/telegram_webhook.py

@@ -9,7 +9,10 @@ WS4 Hermes NL 接入：@mention 觸發 12-Agent 自然語言問答。
 """
 from __future__ import annotations
 
-from typing import Optional
+import hashlib
+import hmac
+import json
+import time
 
 import structlog
 from fastapi import APIRouter, Depends, Header, HTTPException, Request
@@ -20,9 +23,14 @@ logger = structlog.get_logger(__name__)
 
 router = APIRouter(prefix="/telegram", tags=["Telegram Webhook"])
 
+_CALLBACK_FORWARD_MAX_BODY_BYTES = 64 * 1024
+_CALLBACK_FORWARD_MAX_CLOCK_SKEW_SECONDS = 60
+_CALLBACK_FORWARD_DEDUP_TTL_SECONDS = 10 * 60
+_CALLBACK_FORWARD_DEDUP_PREFIX = "telegram:callback_forward:dedup:v1:"
+
 
 def _verify_secret_token(
-    x_telegram_bot_api_secret_token: Optional[str] = Header(None),
+    x_telegram_bot_api_secret_token: str | None = Header(None),
 ) -> None:
     """
     驗證 Telegram Webhook Secret Token（ADR-094 P0-2 修）
@@ -48,6 +56,154 @@ def _verify_secret_token(
         raise HTTPException(status_code=403, detail="Invalid webhook secret")
 
 
+def _verify_callback_forward_signature(
+    body: bytes,
+    *,
+    timestamp: str | None,
+    signature: str | None,
+) -> None:
+    """Authenticate the host188 forwarder without logging secret material."""
+    token = str(getattr(settings, "OPENCLAW_TG_BOT_TOKEN", "") or "")
+    if not token:
+        raise HTTPException(
+            status_code=503,
+            detail="Telegram callback forwarder authentication is not configured",
+        )
+    if not body or len(body) > _CALLBACK_FORWARD_MAX_BODY_BYTES:
+        raise HTTPException(status_code=413, detail="Invalid callback forward body size")
+    try:
+        forwarded_at = int(str(timestamp or ""))
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail="Invalid callback forward signature") from exc
+    if abs(int(time.time()) - forwarded_at) > _CALLBACK_FORWARD_MAX_CLOCK_SKEW_SECONDS:
+        raise HTTPException(status_code=403, detail="Expired callback forward signature")
+
+    provided = str(signature or "").strip().lower()
+    signed = str(forwarded_at).encode("ascii") + b"." + body
+    expected = "sha256=" + hmac.new(
+        token.encode("utf-8"),
+        signed,
+        hashlib.sha256,
+    ).hexdigest()
+    if not hmac.compare_digest(provided, expected):
+        raise HTTPException(status_code=403, detail="Invalid callback forward signature")
+
+
+@router.post("/callback-forward")
+async def telegram_callback_forward(
+    request: Request,
+    x_telegram_forward_timestamp: str | None = Header(
+        None,
+        alias="X-Telegram-Forward-Timestamp",
+    ),
+    x_telegram_forward_signature: str | None = Header(
+        None,
+        alias="X-Telegram-Forward-Signature",
+    ),
+) -> dict:
+    """Receive one HMAC-authenticated callback from the sole polling owner."""
+    body_bytes = await request.body()
+    _verify_callback_forward_signature(
+        body_bytes,
+        timestamp=x_telegram_forward_timestamp,
+        signature=x_telegram_forward_signature,
+    )
+    try:
+        body = json.loads(body_bytes)
+    except (TypeError, ValueError, json.JSONDecodeError) as exc:
+        raise HTTPException(status_code=400, detail="Invalid callback forward JSON") from exc
+
+    if not isinstance(body, dict) or set(body) - {"update_id", "callback_query"}:
+        raise HTTPException(status_code=400, detail="Invalid callback forward envelope")
+    update_id = body.get("update_id")
+    callback = body.get("callback_query")
+    if not isinstance(update_id, int) or not isinstance(callback, dict):
+        raise HTTPException(status_code=400, detail="Invalid callback forward envelope")
+
+    callback_query_id = str(callback.get("id") or "")
+    callback_data = str(callback.get("data") or "")
+    user_id = (callback.get("from") or {}).get("id")
+    message = callback.get("message") or {}
+    chat_id = (message.get("chat") or {}).get("id")
+    message_id = message.get("message_id")
+    if not all(
+        [callback_query_id, callback_data, user_id, chat_id, message_id]
+    ):
+        raise HTTPException(status_code=400, detail="Incomplete callback forward envelope")
+
+    from src.services.telegram_button_registry import (
+        resolve_callback_action_contract,
+    )
+
+    contract = resolve_callback_action_contract(callback_data)
+    if (
+        contract is None
+        or contract.callback_format != "info"
+        or contract.risk != "low"
+        or contract.state != "active"
+    ):
+        raise HTTPException(status_code=403, detail="Callback action is not forwardable")
+
+    dedup_id = hashlib.sha256(callback_query_id.encode("utf-8")).hexdigest()[:24]
+    dedup_key = f"{_CALLBACK_FORWARD_DEDUP_PREFIX}{update_id}:{dedup_id}"
+    try:
+        from src.core.redis_client import get_redis
+
+        redis = get_redis()
+        claimed = await redis.set(
+            dedup_key,
+            "processing",
+            nx=True,
+            ex=_CALLBACK_FORWARD_DEDUP_TTL_SECONDS,
+        )
+    except Exception as exc:
+        logger.error(
+            "telegram_callback_forward_dedup_unavailable",
+            error_type=type(exc).__name__,
+        )
+        raise HTTPException(
+            status_code=503,
+            detail="Callback forward deduplication unavailable",
+        ) from exc
+    if not claimed:
+        return {
+            "ok": True,
+            "status": "duplicate_ignored",
+            "update_id": update_id,
+        }
+
+    from src.services.telegram_gateway import get_telegram_gateway
+
+    try:
+        gateway = get_telegram_gateway()
+        await gateway._handle_callback_query(
+            update_id,
+            callback,
+            ingress_source="openclaw_188_hmac_forwarder",
+        )
+        await redis.setex(
+            dedup_key,
+            _CALLBACK_FORWARD_DEDUP_TTL_SECONDS,
+            "processed",
+        )
+    except Exception:
+        await redis.delete(dedup_key)
+        raise
+
+    logger.info(
+        "telegram_callback_forward_processed",
+        update_id=update_id,
+        action=contract.action,
+        callback_payload_logged=False,
+    )
+    return {
+        "ok": True,
+        "status": "processed",
+        "update_id": update_id,
+        "action": contract.action,
+    }
+
+
 @router.post("/webhook", dependencies=[Depends(_verify_secret_token)])
 async def telegram_webhook(request: Request) -> dict:
     """
@@ -107,7 +263,11 @@ async def telegram_webhook(request: Request) -> dict:
             if group_id and member_user_id:
                 try:
                     from src.core.redis_client import get_redis  # type: ignore[import]
-                    from src.hermes.approvers import sync_member_joined, sync_member_left  # type: ignore[import]
+                    from src.hermes.approvers import (  # type: ignore[import]
+                        sync_member_joined,
+                        sync_member_left,
+                    )
+
                     redis = get_redis()
                     if status in ("member", "administrator", "creator"):
                         await sync_member_joined(redis, group_id, member_user_id)