From 84a3fa17a9ec8ff7beca976cfcfd62aa58e5469d Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 16 Jul 2026 23:49:23 +0800 Subject: [PATCH] fix(telegram): require durable callback receipts --- apps/api/src/api/v1/telegram.py | 1 + .../src/services/telegram_button_registry.py | 43 ++++--- apps/api/src/services/telegram_gateway.py | 107 ++++++++++++++++-- ...test_telegram_callback_ingress_contract.py | 69 +++++++++-- 4 files changed, 186 insertions(+), 34 deletions(-) diff --git a/apps/api/src/api/v1/telegram.py b/apps/api/src/api/v1/telegram.py index 019f98ce4..e12d3e7fa 100644 --- a/apps/api/src/api/v1/telegram.py +++ b/apps/api/src/api/v1/telegram.py @@ -564,6 +564,7 @@ async def test_push( async def telegram_health() -> dict: """Telegram Gateway health with callback transport truth.""" gateway = get_telegram_gateway() + await gateway.refresh_callback_ingress_receipt() ingress = gateway.callback_ingress_status() configured = bool(settings.OPENCLAW_TG_BOT_TOKEN and settings.SRE_GROUP_CHAT_ID) interactive_ready = bool(ingress.get("interactive_buttons_ready") is True) diff --git a/apps/api/src/services/telegram_button_registry.py b/apps/api/src/services/telegram_button_registry.py index 7b2171f28..00c83c978 100644 --- a/apps/api/src/services/telegram_button_registry.py +++ b/apps/api/src/services/telegram_button_registry.py @@ -43,18 +43,22 @@ _STATIC_ACTIONS: dict[str, TelegramButtonActionContract] = { "approve": TelegramButtonActionContract( "approve", "nonce", "TelegramGateway.handle_callback", "high", "approval_control_plane", "approval_and_execution_receipt", + state="blocked_missing_controlled_apply_contract", ), "reject": TelegramButtonActionContract( "reject", "nonce", "TelegramGateway.handle_callback", "medium", "approval_control_plane", "approval_state_receipt", + state="blocked_missing_controlled_apply_contract", ), "silence": TelegramButtonActionContract( "silence", "nonce", "TelegramGateway._handle_silence", "medium", "telegram_silence_store", "telegram_callback_reply_receipt", + state="blocked_missing_controlled_apply_contract", ), "snooze": TelegramButtonActionContract( "snooze", "nonce", "TelegramGateway._handle_snooze", "low", "telegram_snooze_store", "telegram_callback_reply_receipt", + state="blocked_missing_controlled_apply_contract", ), "tune": TelegramButtonActionContract( "tune", "nonce", "TelegramGateway._handle_auto_tuning", "low", @@ -64,22 +68,27 @@ _STATIC_ACTIONS: dict[str, TelegramButtonActionContract] = { "log_manual_fix": TelegramButtonActionContract( "log_manual_fix", "nonce", "TelegramGateway.handle_callback:log_manual_fix", "low", "incident_timeline", "timeline_write_receipt", + state="blocked_missing_controlled_apply_contract", ), "drift_view": TelegramButtonActionContract( "drift_view", "nonce", "TelegramGateway._handle_drift_action", "low", "drift_readback", "telegram_callback_reply_receipt", + state="blocked_missing_controlled_apply_contract", ), "drift_adopt": TelegramButtonActionContract( "drift_adopt", "nonce", "TelegramGateway._handle_drift_action", "high", "drift_control_plane", "drift_post_verifier", + state="blocked_missing_controlled_apply_contract", ), "drift_revert": TelegramButtonActionContract( "drift_revert", "nonce", "TelegramGateway._handle_drift_action", "high", "drift_control_plane", "drift_post_verifier", + state="blocked_missing_controlled_apply_contract", ), "la": TelegramButtonActionContract( "la", "llm_short_id", "TelegramGateway._handle_llm_action_callback", "dynamic", "callback_dispatcher", "callback_dispatch_receipt", + state="blocked_missing_controlled_apply_contract", ), } @@ -99,23 +108,16 @@ def _callback_action(callback_data: str) -> str: def _registry_action_contract(action: str) -> TelegramButtonActionContract | None: contract = _STATIC_ACTIONS.get(action) if contract is not None: - return contract + return contract if contract.state == "active" else None if action in _AI_ADVISORY_ACTIONS: - return TelegramButtonActionContract( - action, - "advisory", - "TelegramGateway._handle_ai_advisory_action", - "low", - "ai_advisory_control_plane", - "telegram_callback_reply_receipt", - ) + return None # The YAML registry is the existing source of truth for category actions. # Import lazily so registry validation never creates a module cycle. from src.services.callback_dispatcher import get_action_spec spec = get_action_spec(action) - if spec is None: + if spec is None or spec.callback_format != "info" or spec.risk != "low": return None return TelegramButtonActionContract( action=spec.name, @@ -166,12 +168,12 @@ def registered_info_callback_actions() -> frozenset[str]: names = { action for action, contract in _STATIC_ACTIONS.items() - if contract.callback_format == "info" + if contract.callback_format == "info" and contract.state == "active" } names.update( spec.name for spec in load_action_registry().values() - if spec.callback_format == "info" + if spec.callback_format == "info" and spec.risk == "low" ) return frozenset(names) @@ -189,6 +191,7 @@ def telegram_button_registry_snapshot() -> dict[str, object]: risk="low", executor="ai_advisory_control_plane", verifier="telegram_callback_reply_receipt", + state="blocked_missing_controlled_apply_contract", ) for action in sorted(_AI_ADVISORY_ACTIONS) ) @@ -200,6 +203,13 @@ def telegram_button_registry_snapshot() -> dict[str, object]: risk=spec.risk, executor=f"mcp:{spec.mcp_provider}/{spec.mcp_tool}", verifier="callback_dispatch_receipt", + state=( + "active" + if spec.callback_format == "info" and spec.risk == "low" + else "break_glass_only" + if spec.risk == "critical" + else "blocked_missing_controlled_apply_contract" + ), ) for spec in load_action_registry().values() if spec.name not in _STATIC_ACTIONS @@ -207,8 +217,15 @@ def telegram_button_registry_snapshot() -> dict[str, object]: rows = sorted((asdict(item) for item in contracts), key=lambda item: item["action"]) return { "schema_version": "telegram_button_action_registry_v1", - "registered_action_count": len(rows), + "declared_action_count": len(rows), + "registered_action_count": sum( + 1 for item in rows if item["state"] == "active" + ), + "blocked_action_count": sum( + 1 for item in rows if item["state"] != "active" + ), "unknown_action_policy": "do_not_render", + "write_action_policy": "fail_closed_until_controlled_apply_contract", "actions": rows, } diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index faad17160..89fcfec30 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -135,7 +135,9 @@ POLLING_LEADER_KEY = "telegram:polling:leader" POLLING_LEADER_TTL = 45 # seconds - Pod 宕掉後 45s 自動轉移 POLLING_LEADER_RENEW = 20 # seconds - 每 20s 續約 POLLING_LEADER_WATCH = 30 # seconds - 非 Leader Pod 每 30s 嘗試接管 -CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS = 24 * 60 * 60 +CALLBACK_INGRESS_RECEIPT_KEY = "telegram:callback_ingress:last_receipt:v1" +CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS = 5 * 60 +CALLBACK_INGRESS_RECEIPT_TTL_SECONDS = 10 * 60 logger = structlog.get_logger(__name__) _TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s]+") @@ -5557,6 +5559,8 @@ class TelegramGateway: self._last_update_id = 0 self._last_callback_ingress_receipt_at: datetime | None = None self._last_callback_ingress_source = "" + self._last_callback_ingress_receipt_durable = False + self._callback_ingress_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 @@ -5609,23 +5613,94 @@ class TelegramGateway: """Runtime binding; new monitoring sends are still policy-gated.""" return settings.SRE_GROUP_CHAT_ID - def record_callback_ingress_receipt( + async def record_callback_ingress_receipt( self, *, source: str, update_id: int | None, ) -> None: - """Record one process-local callback receipt without callback payload.""" - self._last_callback_ingress_receipt_at = datetime.now(UTC) - self._last_callback_ingress_source = str(source or "unknown")[:80] + """Persist one redaction-safe shared receipt without callback payload.""" + received_at = datetime.now(UTC) + normalized_source = str(source or "unknown")[:80] + self._last_callback_ingress_receipt_at = received_at + self._last_callback_ingress_source = normalized_source + self._last_callback_ingress_receipt_durable = False if isinstance(update_id, int): self._last_update_id = max(int(self._last_update_id or 0), update_id) + receipt = { + "schema_version": "telegram_callback_ingress_receipt_v1", + "received_at": received_at.isoformat(), + "source": normalized_source, + "update_id": update_id if isinstance(update_id, int) else None, + } + try: + redis = get_redis() + await redis.setex( + CALLBACK_INGRESS_RECEIPT_KEY, + CALLBACK_INGRESS_RECEIPT_TTL_SECONDS, + json.dumps(receipt, separators=(",", ":"), sort_keys=True), + ) + self._last_callback_ingress_receipt_durable = True + self._callback_ingress_shared_read_error = "" + except Exception as exc: + self._callback_ingress_shared_read_error = type(exc).__name__ + logger.error( + "telegram_callback_ingress_receipt_persist_failed", + error_type=type(exc).__name__, + callback_payload_persisted=False, + ) + + async def refresh_callback_ingress_receipt(self) -> bool: + """Refresh the process mirror from Redis before health or send gates.""" + try: + redis = get_redis() + raw = await redis.get(CALLBACK_INGRESS_RECEIPT_KEY) + except Exception as exc: + self._last_callback_ingress_receipt_durable = False + self._callback_ingress_shared_read_error = type(exc).__name__ + return False + + if not raw: + self._last_callback_ingress_receipt_durable = False + self._callback_ingress_shared_read_error = "" + return False + try: + parsed = json.loads(str(raw)) + received_at = datetime.fromisoformat(str(parsed["received_at"])) + if received_at.tzinfo is None: + received_at = received_at.replace(tzinfo=UTC) + else: + received_at = received_at.astimezone(UTC) + if received_at > datetime.now(UTC): + raise ValueError("future_callback_receipt") + source = str(parsed.get("source") or "unknown")[:80] + update_id = parsed.get("update_id") + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + self._last_callback_ingress_receipt_durable = False + self._callback_ingress_shared_read_error = type(exc).__name__ + logger.warning( + "telegram_callback_ingress_receipt_invalid", + error_type=type(exc).__name__, + ) + return False + + self._last_callback_ingress_receipt_at = received_at + self._last_callback_ingress_source = source + self._last_callback_ingress_receipt_durable = True + self._callback_ingress_shared_read_error = "" + if isinstance(update_id, int): + self._last_update_id = max(int(self._last_update_id or 0), update_id) + return True + def callback_ingress_status(self) -> dict[str, object]: """Return callback transport truth used by health and render gates.""" now = datetime.now(UTC) last_receipt = getattr(self, "_last_callback_ingress_receipt_at", None) source = str(getattr(self, "_last_callback_ingress_source", "") or "") + receipt_durable = bool( + getattr(self, "_last_callback_ingress_receipt_durable", False) + ) age_seconds: int | None = None if isinstance(last_receipt, datetime): normalized = ( @@ -5635,7 +5710,8 @@ class TelegramGateway: ) age_seconds = max(0, int((now - normalized).total_seconds())) receipt_fresh = bool( - age_seconds is not None + receipt_durable + and age_seconds is not None and age_seconds <= CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS ) polling_declared_active = bool(getattr(self, "_polling_active", False)) @@ -5671,8 +5747,12 @@ class TelegramGateway: "polling_active": polling_active, "polling_declared_active": polling_declared_active, "polling_task_active": polling_task_active, - "receipt_available": isinstance(last_receipt, datetime), + "receipt_available": bool( + receipt_durable and isinstance(last_receipt, datetime) + ), "receipt_fresh": receipt_fresh, + "receipt_durable": receipt_durable, + "local_receipt_observed": isinstance(last_receipt, datetime), "receipt_source": source or None, "last_receipt_at": ( last_receipt.astimezone(UTC).isoformat() @@ -5686,7 +5766,15 @@ class TelegramGateway: "stale": bool(isinstance(last_receipt, datetime) and not receipt_fresh), "interactive_buttons_ready": interactive_ready, "fail_closed_reason": ( - None if interactive_ready else "callback_ingress_not_verified" + None + if interactive_ready + else "callback_ingress_shared_receipt_unavailable" + if isinstance(last_receipt, datetime) and not receipt_durable + else "callback_ingress_not_verified" + ), + "shared_read_error": ( + str(getattr(self, "_callback_ingress_shared_read_error", "") or "") + or None ), } @@ -6779,6 +6867,7 @@ class TelegramGateway: raise TelegramGatewayError("HTTP client not initialized") payload = normalize_telegram_send_message_payload(method, payload) + await self.refresh_callback_ingress_receipt() payload = self._apply_callback_ingress_render_gate(payload) source_envelope_extra = payload.pop(_AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY, None) if canonical_route_receipt is not None: @@ -13090,7 +13179,7 @@ class TelegramGateway: logger.warning("telegram_callback_invalid", update_id=update_id) return - self.record_callback_ingress_receipt( + await self.record_callback_ingress_receipt( source=ingress_source, update_id=update_id, ) diff --git a/apps/api/tests/test_telegram_callback_ingress_contract.py b/apps/api/tests/test_telegram_callback_ingress_contract.py index fc4b7ece3..2d5fc94b0 100644 --- a/apps/api/tests/test_telegram_callback_ingress_contract.py +++ b/apps/api/tests/test_telegram_callback_ingress_contract.py @@ -59,22 +59,56 @@ def test_callback_ingress_defaults_to_unverified_fail_closed() -> None: assert status["fail_closed_reason"] == "callback_ingress_not_verified" -def test_callback_ingress_receipt_has_24_hour_freshness_contract() -> None: +def test_callback_ingress_receipt_has_five_minute_freshness_contract() -> None: gateway = _ConfiguredGateway() - assert CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS == 24 * 60 * 60 + assert CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS == 5 * 60 gateway._last_callback_ingress_source = "canonical_webhook" - gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(hours=23) + gateway._last_callback_ingress_receipt_durable = True + gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(minutes=4) fresh = gateway.callback_ingress_status() assert fresh["receipt_fresh"] is True assert fresh["interactive_buttons_ready"] is True - gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(hours=25) + gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(minutes=6) stale = gateway.callback_ingress_status() assert stale["stale"] is True assert stale["interactive_buttons_ready"] is False +class _SharedReceiptRedis: + def __init__(self) -> None: + self.value: str | None = None + + async def setex(self, _key: str, _ttl: int, value: str) -> None: + self.value = value + + async def get(self, _key: str) -> str | None: + return self.value + + +@pytest.mark.asyncio +async def test_callback_ingress_receipt_is_shared_across_gateway_processes( + monkeypatch, +) -> None: + redis = _SharedReceiptRedis() + monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: redis) + writer = _ConfiguredGateway() + reader = _ConfiguredGateway() + + await writer.record_callback_ingress_receipt( + source="canonical_webhook", + update_id=88, + ) + assert await reader.refresh_callback_ingress_receipt() is True + + status = reader.callback_ingress_status() + assert status["receipt_durable"] is True + assert status["receipt_fresh"] is True + assert status["interactive_buttons_ready"] is True + assert reader._last_update_id == 88 + + def test_polling_flag_without_live_task_is_stale_not_ready() -> None: gateway = _ConfiguredGateway() gateway._polling_active = True @@ -154,23 +188,31 @@ def test_ready_markup_still_removes_unknown_ghost_callbacks() -> None: assert all("scale:" not in str(button) for button in rendered) -def test_machine_registry_excludes_ghosts_and_describes_shadow_tune() -> None: +def test_machine_registry_excludes_ghosts_and_fails_closed_write_actions() -> None: assert callback_action_is_registered("detail:INC-20260716-TEST") is True assert callback_action_is_registered("view:ERR:1:nonce") is False assert callback_action_is_registered("scale:RES:1:nonce") is False - tune = resolve_callback_action_contract("tune:APR-1:1:nonce") - assert tune is not None - assert tune.state == "shadow_candidate_only" + assert resolve_callback_action_contract("tune:APR-1:1:nonce") is None + assert resolve_callback_action_contract("approve:APR-1:1:nonce") is None snapshot = telegram_button_registry_snapshot() action_names = {row["action"] for row in snapshot["actions"]} assert snapshot["schema_version"] == "telegram_button_action_registry_v1" assert snapshot["unknown_action_policy"] == "do_not_render" + assert snapshot["write_action_policy"] == ( + "fail_closed_until_controlled_apply_contract" + ) assert "view" not in action_names assert "scale" not in action_names + actions = {row["action"]: row for row in snapshot["actions"]} + assert actions["tune"]["state"] == "shadow_candidate_only" + assert actions["approve"]["state"] == ( + "blocked_missing_controlled_apply_contract" + ) + assert actions["secops_block_ip"]["state"] == "break_glass_only" -def test_every_yaml_action_has_a_renderable_registry_contract() -> None: +def test_only_low_risk_read_only_yaml_actions_are_renderable() -> None: registry = load_action_registry() assert registry @@ -181,9 +223,12 @@ def test_every_yaml_action_has_a_renderable_registry_contract() -> None: else f"{action}:INC-20260716-TEST" ) contract = resolve_callback_action_contract(callback_data) - assert contract is not None, action - assert contract.handler == "callback_dispatcher.dispatch_action" - assert contract.executor.startswith("mcp:") + if spec.callback_format == "info" and spec.risk == "low": + assert contract is not None, action + assert contract.handler == "callback_dispatcher.dispatch_action" + assert contract.executor.startswith("mcp:") + else: + assert contract is None, action def test_security_parser_accepts_every_registered_info_action() -> None: