From 7d1968835a70e32bac31eb155d04a10ad1257d01 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sat, 18 Jul 2026 16:55:43 +0800 Subject: [PATCH] fix(telegram): persist controlled callback action receipts --- apps/api/src/api/v1/telegram_webhook.py | 44 ++++- .../services/telegram_callback_correlation.py | 55 ++++++ apps/api/src/services/telegram_gateway.py | 166 +++++++++++++++- ...test_telegram_callback_ingress_contract.py | 177 ++++++++++++++++-- 4 files changed, 427 insertions(+), 15 deletions(-) create mode 100644 apps/api/src/services/telegram_callback_correlation.py diff --git a/apps/api/src/api/v1/telegram_webhook.py b/apps/api/src/api/v1/telegram_webhook.py index 7500e92af..a7b2ca1bd 100644 --- a/apps/api/src/api/v1/telegram_webhook.py +++ b/apps/api/src/api/v1/telegram_webhook.py @@ -18,6 +18,10 @@ import structlog from fastapi import APIRouter, Depends, Header, HTTPException, Request from src.core.config import settings +from src.services.telegram_callback_correlation import ( + controlled_callback_correlation_fingerprint, + normalize_controlled_callback_correlation, +) logger = structlog.get_logger(__name__) @@ -113,12 +117,31 @@ async def telegram_callback_forward( 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"}: + if not isinstance(body, dict) or set(body) - { + "update_id", + "callback_query", + "controlled_correlation", + }: 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") + controlled_correlation = None + correlation_fingerprint = None + if "controlled_correlation" in body: + try: + controlled_correlation = normalize_controlled_callback_correlation( + body["controlled_correlation"] + ) + correlation_fingerprint = controlled_callback_correlation_fingerprint( + controlled_correlation + ) + except ValueError as exc: + raise HTTPException( + status_code=400, + detail="Invalid controlled callback correlation", + ) from exc callback_query_id = str(callback.get("id") or "") callback_data = str(callback.get("data") or "") @@ -176,11 +199,25 @@ async def telegram_callback_forward( try: gateway = get_telegram_gateway() - await gateway._handle_callback_query( + result = await gateway._handle_callback_query( update_id, callback, ingress_source="openclaw_188_hmac_forwarder", + ingress_correlation=controlled_correlation, ) + if not isinstance(result, dict) or result.get("success") is not True: + raise HTTPException( + status_code=502, + detail="Callback action did not complete", + ) + if ( + controlled_correlation is not None + and result.get("controlled_action_receipt_durable") is not True + ): + raise HTTPException( + status_code=503, + detail="Controlled callback receipt is not durable", + ) await redis.setex( dedup_key, _CALLBACK_FORWARD_DEDUP_TTL_SECONDS, @@ -194,6 +231,8 @@ async def telegram_callback_forward( "telegram_callback_forward_processed", update_id=update_id, action=contract.action, + correlation_present=controlled_correlation is not None, + correlation_fingerprint=correlation_fingerprint, callback_payload_logged=False, ) return { @@ -201,6 +240,7 @@ async def telegram_callback_forward( "status": "processed", "update_id": update_id, "action": contract.action, + "controlled_receipt": bool(controlled_correlation), } diff --git a/apps/api/src/services/telegram_callback_correlation.py b/apps/api/src/services/telegram_callback_correlation.py new file mode 100644 index 000000000..c830b252b --- /dev/null +++ b/apps/api/src/services/telegram_callback_correlation.py @@ -0,0 +1,55 @@ +"""Strict, redaction-safe correlation for controlled Telegram callbacks.""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping +from typing import Any + +CONTROLLED_CALLBACK_CORRELATION_SCHEMA = ( + "awoooi_controlled_callback_correlation_v1" +) +_CORRELATION_KEYS = frozenset( + {"schema_version", "trace_id", "run_id", "work_item_id"} +) +_SAFE_ID_RE = re.compile( + r"^[A-Za-z0-9](?:[A-Za-z0-9_.:-]{0,158}[A-Za-z0-9])?$" +) + + +def normalize_controlled_callback_correlation(value: Any) -> dict[str, str]: + """Return one exact correlation envelope or raise ``ValueError``.""" + if not isinstance(value, Mapping) or set(value) != _CORRELATION_KEYS: + raise ValueError("invalid_controlled_callback_correlation_shape") + if value.get("schema_version") != CONTROLLED_CALLBACK_CORRELATION_SCHEMA: + raise ValueError("invalid_controlled_callback_correlation_schema") + + normalized = {"schema_version": CONTROLLED_CALLBACK_CORRELATION_SCHEMA} + for key in ("trace_id", "run_id", "work_item_id"): + candidate = value.get(key) + if not isinstance(candidate, str) or not _SAFE_ID_RE.fullmatch(candidate): + raise ValueError(f"invalid_controlled_callback_correlation_{key}") + if candidate in {".", ".."}: + raise ValueError(f"invalid_controlled_callback_correlation_{key}") + normalized[key] = candidate + return normalized + + +def controlled_callback_correlation_fingerprint(value: Any) -> str: + """Hash normalized IDs so health/readback never needs to expose them.""" + normalized = normalize_controlled_callback_correlation(value) + canonical = json.dumps( + normalized, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return hashlib.sha256(canonical).hexdigest() + + +__all__ = [ + "CONTROLLED_CALLBACK_CORRELATION_SCHEMA", + "controlled_callback_correlation_fingerprint", + "normalize_controlled_callback_correlation", +] diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 7c7885079..83caff89c 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -77,6 +77,10 @@ from src.services.security_interceptor import ( get_security_interceptor, ) from src.services.telegram_button_registry import callback_action_is_registered +from src.services.telegram_callback_correlation import ( + controlled_callback_correlation_fingerprint, + normalize_controlled_callback_correlation, +) # ============================================================================= # Snooze/Silence Redis Keys (2026-03-27 P1 優化) @@ -136,6 +140,9 @@ POLLING_LEADER_TTL = 45 # seconds - Pod 宕掉後 45s 自動轉移 POLLING_LEADER_RENEW = 20 # seconds - 每 20s 續約 POLLING_LEADER_WATCH = 30 # seconds - 非 Leader Pod 每 30s 嘗試接管 CALLBACK_INGRESS_RECEIPT_KEY = "telegram:callback_ingress:last_receipt:v1" +CALLBACK_CONTROLLED_ACTION_RECEIPT_KEY = ( + "telegram:callback_action:last_controlled_receipt:v1" +) CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS = 5 * 60 CALLBACK_INGRESS_RECEIPT_TTL_SECONDS = 10 * 60 @@ -5635,6 +5642,12 @@ class TelegramGateway: self._last_callback_ingress_source = "" self._last_callback_ingress_receipt_durable = False self._callback_ingress_shared_read_error = "" + self._last_controlled_callback_action_at: datetime | None = None + self._last_controlled_callback_action_source = "" + self._last_controlled_callback_action = "" + self._last_controlled_callback_action_fingerprint = "" + self._last_controlled_callback_action_durable = False + self._controlled_callback_action_shared_read_error = "" # 2026-04-01 Claude Code: 分散式 Leader Election (防 2-Pod 409 互搶) self._pod_id = os.environ.get("POD_NAME", os.urandom(8).hex()) self._leader_task: asyncio.Task | None = None @@ -5730,11 +5743,18 @@ class TelegramGateway: try: redis = get_redis() raw = await redis.get(CALLBACK_INGRESS_RECEIPT_KEY) + controlled_action_raw = await redis.get( + CALLBACK_CONTROLLED_ACTION_RECEIPT_KEY + ) except Exception as exc: self._last_callback_ingress_receipt_durable = False self._callback_ingress_shared_read_error = type(exc).__name__ return False + await self._refresh_controlled_callback_action_receipt( + controlled_action_raw + ) + if not raw: self._last_callback_ingress_receipt_durable = False self._callback_ingress_shared_read_error = "" @@ -5767,6 +5787,96 @@ class TelegramGateway: self._last_update_id = max(int(self._last_update_id or 0), update_id) return True + async def record_controlled_callback_action_receipt( + self, + *, + source: str, + update_id: int, + action: str, + correlation: Mapping[str, object], + ) -> bool: + """Persist proof only after the correlated callback action succeeded.""" + try: + normalized = normalize_controlled_callback_correlation(correlation) + fingerprint = controlled_callback_correlation_fingerprint(normalized) + except ValueError: + return False + + completed_at = datetime.now(UTC) + normalized_source = str(source or "unknown")[:80] + normalized_action = str(action or "unknown")[:64] + receipt = { + "schema_version": "telegram_controlled_callback_action_receipt_v1", + "completed_at": completed_at.isoformat(), + "source": normalized_source, + "update_id": int(update_id), + "action": normalized_action, + "correlation_fingerprint": fingerprint, + } + self._last_controlled_callback_action_at = completed_at + self._last_controlled_callback_action_source = normalized_source + self._last_controlled_callback_action = normalized_action + self._last_controlled_callback_action_fingerprint = fingerprint + self._last_controlled_callback_action_durable = False + try: + redis = get_redis() + await redis.setex( + CALLBACK_CONTROLLED_ACTION_RECEIPT_KEY, + CALLBACK_INGRESS_RECEIPT_TTL_SECONDS, + json.dumps(receipt, separators=(",", ":"), sort_keys=True), + ) + except Exception as exc: + self._controlled_callback_action_shared_read_error = type(exc).__name__ + logger.error( + "telegram_controlled_callback_action_receipt_persist_failed", + error_type=type(exc).__name__, + correlation_fingerprint=fingerprint, + callback_payload_persisted=False, + ) + return False + + self._last_controlled_callback_action_durable = True + self._controlled_callback_action_shared_read_error = "" + return True + + async def _refresh_controlled_callback_action_receipt(self, raw: object) -> bool: + if not raw: + self._last_controlled_callback_action_durable = False + self._controlled_callback_action_shared_read_error = "" + return False + try: + parsed = json.loads(str(raw)) + if ( + parsed.get("schema_version") + != "telegram_controlled_callback_action_receipt_v1" + ): + raise ValueError("invalid_controlled_action_receipt_schema") + completed_at = datetime.fromisoformat(str(parsed["completed_at"])) + completed_at = ( + completed_at.replace(tzinfo=UTC) + if completed_at.tzinfo is None + else completed_at.astimezone(UTC) + ) + if completed_at > datetime.now(UTC): + raise ValueError("future_controlled_action_receipt") + fingerprint = str(parsed["correlation_fingerprint"]) + if not re.fullmatch(r"[0-9a-f]{64}", fingerprint): + raise ValueError("invalid_controlled_action_fingerprint") + source = str(parsed.get("source") or "unknown")[:80] + action = str(parsed.get("action") or "unknown")[:64] + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + self._last_controlled_callback_action_durable = False + self._controlled_callback_action_shared_read_error = type(exc).__name__ + return False + + self._last_controlled_callback_action_at = completed_at + self._last_controlled_callback_action_source = source + self._last_controlled_callback_action = action + self._last_controlled_callback_action_fingerprint = fingerprint + self._last_controlled_callback_action_durable = True + self._controlled_callback_action_shared_read_error = "" + return True + def callback_ingress_status(self) -> dict[str, object]: """Return callback transport truth used by health and render gates.""" now = datetime.now(UTC) @@ -5813,6 +5923,18 @@ class TelegramGateway: owner = "external_forwarder_unverified" configured = bool(self.bot_token and self.chat_id) interactive_ready = bool(configured and (polling_active or receipt_fresh)) + controlled_action_at = self._last_controlled_callback_action_at + controlled_action_age_seconds = ( + max(0, int((now - controlled_action_at).total_seconds())) + if isinstance(controlled_action_at, datetime) + else None + ) + controlled_action_fresh = bool( + self._last_controlled_callback_action_durable + and controlled_action_age_seconds is not None + and controlled_action_age_seconds + <= CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS + ) return { "schema_version": "telegram_callback_ingress_status_v1", "mode": mode, @@ -5850,6 +5972,32 @@ class TelegramGateway: str(getattr(self, "_callback_ingress_shared_read_error", "") or "") or None ), + "controlled_action_receipt": { + "schema_version": "telegram_controlled_callback_action_status_v1", + "available": bool( + self._last_controlled_callback_action_durable + and isinstance(controlled_action_at, datetime) + ), + "fresh": controlled_action_fresh, + "durable": self._last_controlled_callback_action_durable, + "source": self._last_controlled_callback_action_source or None, + "action": self._last_controlled_callback_action or None, + "correlation_fingerprint": ( + self._last_controlled_callback_action_fingerprint or None + ), + "completed_at": ( + controlled_action_at.astimezone(UTC).isoformat() + if isinstance(controlled_action_at, datetime) + and controlled_action_at.tzinfo + else controlled_action_at.replace(tzinfo=UTC).isoformat() + if isinstance(controlled_action_at, datetime) + else None + ), + "age_seconds": controlled_action_age_seconds, + "shared_read_error": ( + self._controlled_callback_action_shared_read_error or None + ), + }, } def _apply_callback_ingress_render_gate(self, payload: dict) -> dict: @@ -13862,7 +14010,8 @@ class TelegramGateway: callback_query: dict, *, ingress_source: str = "api_long_polling", - ) -> None: + ingress_correlation: Mapping[str, object] | None = None, + ) -> dict: """處理按鈕點擊更新""" callback_query_id = callback_query.get("id") callback_data = callback_query.get("data") @@ -13871,7 +14020,7 @@ class TelegramGateway: if not all([callback_query_id, callback_data, user_id]): logger.warning("telegram_callback_invalid", update_id=update_id) - return + return {"success": False, "reason": "invalid_callback_query"} await self.record_callback_ingress_receipt( source=ingress_source, @@ -13912,6 +14061,18 @@ class TelegramGateway: chat_id=chat_id, ) + if result.get("success") is True and ingress_correlation is not None: + receipt_durable = await self.record_controlled_callback_action_receipt( + source=ingress_source, + update_id=update_id, + action=str(result.get("action") or "unknown"), + correlation=ingress_correlation, + ) + result = { + **result, + "controlled_action_receipt_durable": receipt_durable, + } + if result.get("success") and not result.get("info_action"): # 執行資料庫更新 (簽核/拒絕) await self._execute_approval_action( @@ -13922,6 +14083,7 @@ class TelegramGateway: message_id=message_id, inbound_chat_id=chat_id, ) + return result async def _handle_chat_message(self, update_id: int, message: dict) -> None: """處理统帥的文字訊息（個人 chat 或 SRE 群組）""" diff --git a/apps/api/tests/test_telegram_callback_ingress_contract.py b/apps/api/tests/test_telegram_callback_ingress_contract.py index 8cc6499a4..6140b91fc 100644 --- a/apps/api/tests/test_telegram_callback_ingress_contract.py +++ b/apps/api/tests/test_telegram_callback_ingress_contract.py @@ -20,7 +20,13 @@ from src.services.telegram_button_registry import ( resolve_callback_action_contract, telegram_button_registry_snapshot, ) +from src.services.telegram_callback_correlation import ( + CONTROLLED_CALLBACK_CORRELATION_SCHEMA, + controlled_callback_correlation_fingerprint, + normalize_controlled_callback_correlation, +) from src.services.telegram_gateway import ( + CALLBACK_CONTROLLED_ACTION_RECEIPT_KEY, CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS, TelegramGateway, filter_telegram_reply_markup, @@ -82,13 +88,13 @@ def test_callback_ingress_receipt_has_five_minute_freshness_contract() -> None: class _SharedReceiptRedis: def __init__(self) -> None: - self.value: str | None = None + self.values: dict[str, str] = {} - async def setex(self, _key: str, _ttl: int, value: str) -> None: - self.value = value + async def setex(self, key: str, _ttl: int, value: str) -> None: + self.values[key] = value - async def get(self, _key: str) -> str | None: - return self.value + async def get(self, key: str) -> str | None: + return self.values.get(key) @pytest.mark.asyncio @@ -356,8 +362,9 @@ class _CallbackRequest: class _CanonicalWebhookGateway: - def __init__(self) -> None: + def __init__(self, *, success: bool = True) -> None: self.receipts: list[dict] = [] + self.success = success async def _handle_callback_query( self, @@ -365,12 +372,19 @@ class _CanonicalWebhookGateway: callback_query: dict, *, ingress_source: str, - ) -> None: + ingress_correlation: dict | None = None, + ) -> dict: self.receipts.append({ "update_id": update_id, "callback_query": callback_query, "ingress_source": ingress_source, + "ingress_correlation": ingress_correlation, }) + return { + "success": self.success, + "action": "detail", + "controlled_action_receipt_durable": bool(ingress_correlation), + } class _CallbackForwardRequest: @@ -398,9 +412,12 @@ class _CallbackForwardRedis: self.values.pop(key, None) -def _signed_callback_forward_body(token: str) -> tuple[bytes, str, str]: - body = json.dumps( - { +def _signed_callback_forward_body( + token: str, + *, + controlled_correlation: dict | None = None, +) -> tuple[bytes, str, str]: + payload = { "callback_query": { "data": "detail:INC-20260716-TEST", "from": {"id": 7, "username": "operator"}, @@ -412,7 +429,11 @@ def _signed_callback_forward_body(token: str) -> tuple[bytes, str, str]: }, }, "update_id": 77, - }, + } + if controlled_correlation is not None: + payload["controlled_correlation"] = controlled_correlation + body = json.dumps( + payload, separators=(",", ":"), sort_keys=True, ).encode("utf-8") @@ -456,12 +477,146 @@ async def test_hmac_callback_forward_processes_only_registered_info_action( "status": "processed", "update_id": 77, "action": "detail", + "controlled_receipt": False, } assert gateway.receipts[0]["ingress_source"] == ( "openclaw_188_hmac_forwarder" ) +def _controlled_correlation() -> dict[str, str]: + return { + "schema_version": CONTROLLED_CALLBACK_CORRELATION_SCHEMA, + "trace_id": "trace:host188:001", + "run_id": "run:host188:001", + "work_item_id": "AIA-SRE-017", + } + + +def test_controlled_callback_correlation_is_exact_and_stably_fingerprinted() -> None: + correlation = _controlled_correlation() + + assert normalize_controlled_callback_correlation(correlation) == correlation + assert len(controlled_callback_correlation_fingerprint(correlation)) == 64 + with pytest.raises(ValueError, match="shape"): + normalize_controlled_callback_correlation({**correlation, "extra": "no"}) + with pytest.raises(ValueError, match="run_id"): + normalize_controlled_callback_correlation({**correlation, "run_id": "unsafe id"}) + + +@pytest.mark.asyncio +async def test_hmac_callback_forward_carries_controlled_receipt_without_raw_ids( + monkeypatch, +) -> None: + token = "test-forwarder-token" + correlation = _controlled_correlation() + body, timestamp, signature = _signed_callback_forward_body( + token, + controlled_correlation=correlation, + ) + 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", + "controlled_receipt": True, + } + assert gateway.receipts[0]["ingress_correlation"] == correlation + assert correlation["run_id"] not in json.dumps(result) + + +@pytest.mark.asyncio +async def test_hmac_callback_forward_does_not_finalize_failed_action(monkeypatch) -> None: + token = "test-forwarder-token" + body, timestamp, signature = _signed_callback_forward_body(token) + gateway = _CanonicalWebhookGateway(success=False) + 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) + + 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=signature, + ) + + assert exc_info.value.status_code == 502 + assert redis.values == {} + + +@pytest.mark.asyncio +async def test_successful_controlled_action_receipt_is_durable_and_redacted( + monkeypatch, +) -> None: + redis = _SharedReceiptRedis() + monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: redis) + writer = _InfoCallbackGateway() + reader = _ConfiguredGateway() + correlation = _controlled_correlation() + + result = await writer._handle_callback_query( + 42, + { + "id": "callback-controlled-1", + "data": "detail:INC-20260716-TEST", + "from": {"id": 7, "username": "operator"}, + "message": { + "message_id": 99, + "text": "incident", + "chat": {"id": -1001}, + }, + }, + ingress_source="openclaw_188_hmac_forwarder", + ingress_correlation=correlation, + ) + + assert result["controlled_action_receipt_durable"] is True + assert CALLBACK_CONTROLLED_ACTION_RECEIPT_KEY in redis.values + assert await reader.refresh_callback_ingress_receipt() is True + receipt = reader.callback_ingress_status()["controlled_action_receipt"] + assert receipt["durable"] is True + assert receipt["fresh"] is True + assert receipt["source"] == "openclaw_188_hmac_forwarder" + assert receipt["action"] == "detail" + assert receipt["correlation_fingerprint"] == ( + controlled_callback_correlation_fingerprint(correlation) + ) + serialized = json.dumps(receipt) + assert correlation["trace_id"] not in serialized + assert correlation["run_id"] not in serialized + assert correlation["work_item_id"] not in serialized + + @pytest.mark.asyncio async def test_hmac_callback_forward_rejects_invalid_signature(monkeypatch) -> None: token = "test-forwarder-token"