From fc8ac4e6021994535e04df8d894a15f69be4b1e6 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 16 Jul 2026 23:52:02 +0800 Subject: [PATCH] feat(telegram): accept signed callback forwards --- apps/api/src/api/v1/telegram_webhook.py | 166 +++++++++++++++++- ...test_telegram_callback_ingress_contract.py | 142 +++++++++++++++ 2 files changed, 305 insertions(+), 3 deletions(-) diff --git a/apps/api/src/api/v1/telegram_webhook.py b/apps/api/src/api/v1/telegram_webhook.py index 4f6392fae..0909f0d3b 100644 --- a/apps/api/src/api/v1/telegram_webhook.py +++ b/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) diff --git a/apps/api/tests/test_telegram_callback_ingress_contract.py b/apps/api/tests/test_telegram_callback_ingress_contract.py index 2d5fc94b0..8cc6499a4 100644 --- a/apps/api/tests/test_telegram_callback_ingress_contract.py +++ b/apps/api/tests/test_telegram_callback_ingress_contract.py @@ -1,5 +1,9 @@ from __future__ import annotations +import hashlib +import hmac +import json +import time from datetime import UTC, datetime, timedelta from types import SimpleNamespace @@ -369,6 +373,144 @@ class _CanonicalWebhookGateway: }) +class _CallbackForwardRequest: + def __init__(self, body: bytes) -> None: + self._body = body + + async def body(self) -> bytes: + return self._body + + +class _CallbackForwardRedis: + def __init__(self) -> None: + self.values: dict[str, str] = {} + + async def set(self, key: str, value: str, **_kwargs) -> bool: + if key in self.values: + return False + self.values[key] = value + return True + + async def setex(self, key: str, _ttl: int, value: str) -> None: + self.values[key] = value + + async def delete(self, key: str) -> None: + self.values.pop(key, None) + + +def _signed_callback_forward_body(token: str) -> tuple[bytes, str, str]: + body = json.dumps( + { + "callback_query": { + "data": "detail:INC-20260716-TEST", + "from": {"id": 7, "username": "operator"}, + "id": "callback-forward-77", + "message": { + "chat": {"id": -1001}, + "message_id": 9, + "text": "incident", + }, + }, + "update_id": 77, + }, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + timestamp = str(int(time.time())) + signature = "sha256=" + hmac.new( + token.encode("utf-8"), + timestamp.encode("ascii") + b"." + body, + hashlib.sha256, + ).hexdigest() + return body, timestamp, signature + + +@pytest.mark.asyncio +async def test_hmac_callback_forward_processes_only_registered_info_action( + monkeypatch, +) -> None: + token = "test-forwarder-token" + body, timestamp, signature = _signed_callback_forward_body(token) + gateway = _CanonicalWebhookGateway() + redis = _CallbackForwardRedis() + monkeypatch.setattr( + telegram_webhook_api, + "settings", + SimpleNamespace(OPENCLAW_TG_BOT_TOKEN=token), + ) + monkeypatch.setattr( + telegram_gateway_module, + "get_telegram_gateway", + lambda: gateway, + ) + monkeypatch.setattr("src.core.redis_client.get_redis", lambda: redis) + + result = await telegram_webhook_api.telegram_callback_forward( + _CallbackForwardRequest(body), + x_telegram_forward_timestamp=timestamp, + x_telegram_forward_signature=signature, + ) + + assert result == { + "ok": True, + "status": "processed", + "update_id": 77, + "action": "detail", + } + assert gateway.receipts[0]["ingress_source"] == ( + "openclaw_188_hmac_forwarder" + ) + + +@pytest.mark.asyncio +async def test_hmac_callback_forward_rejects_invalid_signature(monkeypatch) -> None: + token = "test-forwarder-token" + body, timestamp, _signature = _signed_callback_forward_body(token) + monkeypatch.setattr( + telegram_webhook_api, + "settings", + SimpleNamespace(OPENCLAW_TG_BOT_TOKEN=token), + ) + + with pytest.raises(HTTPException) as exc_info: + await telegram_webhook_api.telegram_callback_forward( + _CallbackForwardRequest(body), + x_telegram_forward_timestamp=timestamp, + x_telegram_forward_signature="sha256=invalid", + ) + + assert exc_info.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_hmac_callback_forward_rejects_write_action(monkeypatch) -> None: + token = "test-forwarder-token" + body, timestamp, _signature = _signed_callback_forward_body(token) + unsafe_body = body.replace( + b"detail:INC-20260716-TEST", + b"approve:APR-1:1:nonce", + ) + signature = "sha256=" + hmac.new( + token.encode("utf-8"), + timestamp.encode("ascii") + b"." + unsafe_body, + hashlib.sha256, + ).hexdigest() + monkeypatch.setattr( + telegram_webhook_api, + "settings", + SimpleNamespace(OPENCLAW_TG_BOT_TOKEN=token), + ) + + with pytest.raises(HTTPException) as exc_info: + await telegram_webhook_api.telegram_callback_forward( + _CallbackForwardRequest(unsafe_body), + x_telegram_forward_timestamp=timestamp, + x_telegram_forward_signature=signature, + ) + + assert exc_info.value.status_code == 403 + + @pytest.mark.asyncio async def test_canonical_webhook_forwards_callback_to_gateway_handler(monkeypatch) -> None: gateway = _CanonicalWebhookGateway()