wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(telegram): persist controlled callback action receipts

Browse Source
This commit is contained in:
Your Name
2026-07-18 16:55:43 +08:00
parent 79489635d8
commit 7d1968835a
4 changed files with 427 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
apps/api/src/api/v1/telegram_webhook.py
View File

@@ -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),
}

55
apps/api/src/services/telegram_callback_correlation.py Normal file
View File

@@ -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",
]

166
apps/api/src/services/telegram_gateway.py
View File

@@ -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 群組）"""

177
apps/api/tests/test_telegram_callback_ingress_contract.py
View File

@@ -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"