diff --git a/agent99-control-plane.ps1 b/agent99-control-plane.ps1 index bfd210e6e..e1fc23f66 100644 --- a/agent99-control-plane.ps1 +++ b/agent99-control-plane.ps1 @@ -534,185 +534,6 @@ function Invoke-AgentStaleSshProcessGuard { $result } -function Send-AgentTelegramRelay { - param( - [string]$Message, - [object]$Relay, - [string]$ChatIdOverride = "", - [string]$PhotoPath = "", - [string]$ReplyToMessageId = "" - ) - - if (-not ($Relay -and $Relay.enabled)) { - return [pscustomobject]@{ ok = $false; error = "relay_disabled" } - } - - # Canonical Telegram routing is owned by the AWOOOI API gateway. Agent99 - # must never borrow a container token/chat binding or call Bot API directly. - # Until a correlated gateway transport is configured this path is an - # explicit no-egress terminal, not a best-effort fallback. - return [pscustomobject]@{ - ok = $false - sent = $false - providerSendPerformed = $false - routeStatus = "blocked_no_egress" - error = "canonical_telegram_gateway_transport_required" - } - - $relayHost = if ($Relay.host) { [string]$Relay.host } else { "192.168.0.110" } - $container = if ($Relay.container) { [string]$Relay.container } else { "stockplatform-v2-api-1" } - if ($container -notmatch "^[A-Za-z0-9_.-]+$") { - return [pscustomobject]@{ ok = $false; error = "invalid_relay_container" } - } - - $messageBytes = [Text.Encoding]::UTF8.GetBytes($Message) - $messageB64 = [Convert]::ToBase64String($messageBytes) - $chatOverrideB64 = if ($ChatIdOverride) { [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($ChatIdOverride)) } else { "" } - $photoSent = $false - $photoError = $null - $containerPhotoPath = "" - if ($PhotoPath -and (Test-Path -LiteralPath $PhotoPath)) { - $safePhotoName = "agent99-card-" + ([guid]::NewGuid().ToString("N")) + ".png" - $remotePhotoPath = "/tmp/" + $safePhotoName - $containerPhotoPath = "/tmp/" + $safePhotoName - $relayUser = Get-HostSshUser $relayHost - $scpArgs = @(Get-AgentSshIdentityArguments) + @( - "-o", "BatchMode=yes", - "-o", "StrictHostKeyChecking=accept-new", - "-o", "NumberOfPasswordPrompts=0", - "-o", "ConnectTimeout=10", - $PhotoPath, - "${relayUser}@${relayHost}:$remotePhotoPath" - ) - try { - $scp = Start-Process -FilePath "scp.exe" -ArgumentList $scpArgs -NoNewWindow -Wait -PassThru - $scp.Refresh() - if ($null -ne $scp.ExitCode -and $scp.ExitCode -eq 0) { - $copyRun = Invoke-SshText $relayHost "docker cp $remotePhotoPath $container`:$containerPhotoPath && rm -f $remotePhotoPath" 45 1 - if (-not $copyRun.ok) { - $photoError = "relay_docker_cp_failed: $($copyRun.output)" - $containerPhotoPath = "" - } - } else { - $photoError = "relay_scp_failed exitCode=$($scp.ExitCode)" - $containerPhotoPath = "" - } - } catch { - $photoError = "relay_scp_exception: $($_.Exception.Message)" - $containerPhotoPath = "" - } - } - - $containerPhotoB64 = if ($containerPhotoPath) { [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($containerPhotoPath)) } else { "" } - $python = 'import base64, json, mimetypes, os, sys, urllib.error, urllib.parse, urllib.request -text = base64.b64decode(sys.argv[2]).decode() -chat_override = base64.b64decode(sys.argv[3]).decode() if len(sys.argv) > 3 and sys.argv[3] and sys.argv[3] != "-" else "" -photo_path = base64.b64decode(sys.argv[4]).decode() if len(sys.argv) > 4 and sys.argv[4] and sys.argv[4] != "-" else "" -reply_to = int(sys.argv[5]) if len(sys.argv) > 5 and sys.argv[5].isdigit() else None -token = os.environ.get("TELEGRAM_BOT_TOKEN") -chat = chat_override or os.environ.get("TELEGRAM_CHAT_ID") -assert token and chat -reply_parameters = json.dumps({"message_id": reply_to}) if reply_to else None -def emit_receipt(payload): - result = payload.get("result") or {} - safe = {"message_id": result.get("message_id"), "chat_type": (result.get("chat") or {}).get("type")} - print("telegram_receipt_b64=" + base64.b64encode(json.dumps(safe).encode()).decode()) -if photo_path and os.path.exists(photo_path): - caption = text - if len(caption) > 900: - caption = caption[:900] + "\n...(truncated; see evidence)" - filename = os.path.basename(photo_path) - ctype = mimetypes.guess_type(filename)[0] or "image/png" - try: - import requests - request_data = {"chat_id": chat, "caption": caption} - if reply_parameters: - request_data["reply_parameters"] = reply_parameters - with open(photo_path, "rb") as fh: - response = requests.post( - "blocked://canonical-telegram-gateway-required/sendPhoto", - data=request_data, - files={"photo": (filename, fh, ctype)}, - timeout=30, - ) - if not response.ok: - print("photo_send_failed status=%s body=%s" % (response.status_code, response.text[:500])) - sys.exit(2) - emit_receipt(response.json()) - print("photo_sent_ok") - except ImportError: - boundary = "----agent99boundary" - fields = [("chat_id", chat), ("caption", caption)] - if reply_parameters: - fields.append(("reply_parameters", reply_parameters)) - body = bytearray() - for name, value in fields: - body.extend(("--" + boundary + "\r\n").encode()) - body.extend(("Content-Disposition: form-data; name=\"" + name + "\"\r\n\r\n").encode()) - body.extend(str(value).encode()) - body.extend(b"\r\n") - body.extend(("--" + boundary + "\r\n").encode()) - body.extend(("Content-Disposition: form-data; name=\"photo\"; filename=\"" + filename + "\"\r\n").encode()) - body.extend(("Content-Type: " + ctype + "\r\n\r\n").encode()) - with open(photo_path, "rb") as fh: - body.extend(fh.read()) - body.extend(b"\r\n") - body.extend(("--" + boundary + "--\r\n").encode()) - req = urllib.request.Request("blocked://canonical-telegram-gateway-required/sendPhoto", data=bytes(body)) - req.add_header("Content-Type", "multipart/form-data; boundary=" + boundary) - try: - response_payload = json.loads(urllib.request.urlopen(req, timeout=30).read().decode()) - emit_receipt(response_payload) - print("photo_sent_ok") - except urllib.error.HTTPError as exc: - print("photo_send_failed status=%s body=%s" % (exc.code, exc.read().decode(errors="replace")[:500])) - sys.exit(2) -else: - fields = {"chat_id": chat, "text": text, "disable_web_page_preview": "true"} - if reply_parameters: - fields["reply_parameters"] = reply_parameters - data = urllib.parse.urlencode(fields).encode() - response_payload = json.loads(urllib.request.urlopen("blocked://canonical-telegram-gateway-required/sendMessage", data=data, timeout=15).read().decode()) - emit_receipt(response_payload) - print("sent_ok") -' - $pythonB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($python)) - $bootstrap = 'import base64,sys;exec(base64.b64decode(sys.argv[1]))' - $chatOverrideArg = if ($chatOverrideB64) { $chatOverrideB64 } else { "-" } - $containerPhotoArg = if ($containerPhotoB64) { $containerPhotoB64 } else { "-" } - $replyArg = if ($ReplyToMessageId -match "^\d+$") { $ReplyToMessageId } else { "-" } - $remoteCommand = "docker exec $container python3 -c '$bootstrap' $pythonB64 $messageB64 $chatOverrideArg $containerPhotoArg $replyArg" - $run = Invoke-SshText $relayHost $remoteCommand 45 1 - $photoSent = [bool]($run.ok -and $run.output -match "photo_sent_ok") - $messageId = $null - $receiptMatch = [regex]::Match([string]$run.output, "telegram_receipt_b64=([A-Za-z0-9+/=]+)") - if ($receiptMatch.Success) { - try { - $receiptJson = [Text.Encoding]::UTF8.GetString([Convert]::FromBase64String($receiptMatch.Groups[1].Value)) | ConvertFrom-Json - $messageId = $receiptJson.message_id - } catch { - $messageId = $null - } - } - if ($containerPhotoPath) { - Invoke-SshText $relayHost "docker exec $container rm -f $containerPhotoPath" 20 1 | Out-Null - } - - [pscustomobject]@{ - ok = [bool]($run.ok -and ($run.output -match "sent_ok|photo_sent_ok")) - host = $relayHost - container = $container - exitCode = $run.exitCode - chatOverride = if ($ChatIdOverride) { "telegram-origin" } else { $null } - replyToMessageId = if ($ReplyToMessageId) { $ReplyToMessageId } else { $null } - messageId = $messageId - photoSent = $photoSent - photoError = $photoError - output = $run.output - error = if ($run.ok) { $photoError } else { $run.output } - } -} - function Get-AgentObjectValue { param( [object]$Data, @@ -1644,60 +1465,6 @@ function New-AgentTelegramCardImage { } } -function Send-AgentTelegramPhotoDirect { - param( - [string]$Token, - [string]$ChatId, - [string]$Caption, - [string]$PhotoPath, - [string]$ReplyToMessageId = "" - ) - - return [pscustomobject]@{ - ok = $false - sent = $false - providerSendPerformed = $false - routeStatus = "blocked_no_egress" - error = "canonical_telegram_gateway_transport_required" - } - - Add-Type -AssemblyName System.Net.Http -ErrorAction Stop - $client = New-Object System.Net.Http.HttpClient - $content = New-Object System.Net.Http.MultipartFormDataContent - $fileContent = $null - try { - $content.Add([System.Net.Http.StringContent]::new($ChatId), "chat_id") - $captionToSend = $Caption - if ($captionToSend.Length -gt 900) { - $captionToSend = $captionToSend.Substring(0, 900) + "`n...(truncated; see evidence)" - } - $content.Add([System.Net.Http.StringContent]::new($captionToSend), "caption") - if ($ReplyToMessageId -match "^\d+$") { - $replyJson = @{ message_id = [long]$ReplyToMessageId } | ConvertTo-Json -Compress - $content.Add([System.Net.Http.StringContent]::new($replyJson), "reply_parameters") - } - $fileBytes = [IO.File]::ReadAllBytes($PhotoPath) - $fileContent = [System.Net.Http.ByteArrayContent]::new($fileBytes) - $fileContent.Headers.ContentType = [System.Net.Http.Headers.MediaTypeHeaderValue]::Parse("image/png") - $content.Add($fileContent, "photo", [IO.Path]::GetFileName($PhotoPath)) - $response = $client.PostAsync("blocked://canonical-telegram-gateway-required/sendPhoto", $content).GetAwaiter().GetResult() - $body = $response.Content.ReadAsStringAsync().GetAwaiter().GetResult() - if (-not $response.IsSuccessStatusCode) { - throw "telegram_sendPhoto_failed status=$([int]$response.StatusCode) body=$body" - } - $payload = $body | ConvertFrom-Json - return [pscustomobject]@{ - ok = [bool]$payload.ok - messageId = if ($payload.result) { $payload.result.message_id } else { $null } - replyToMessageId = if ($ReplyToMessageId) { $ReplyToMessageId } else { $null } - } - } finally { - if ($fileContent) { $fileContent.Dispose() } - if ($content) { $content.Dispose() } - if ($client) { $client.Dispose() } - } -} - function Invoke-AgentTelegramCardSelfTest { $sample = [pscustomobject]@{ host = "192.168.0.110" @@ -1940,33 +1707,6 @@ function Send-AgentTelegram { $telegram = $Config.telegram $enabled = ($telegram -and $telegram.enabled) - $botTokenEnv = if ($telegram.botTokenEnv) { $telegram.botTokenEnv } else { "AGENT99_TELEGRAM_BOT_TOKEN" } - $chatIdEnv = if ($telegram.chatIdEnv) { $telegram.chatIdEnv } else { "AGENT99_TELEGRAM_CHAT_ID" } - $tokenEnvAliases = @($botTokenEnv) - if ($telegram.botTokenEnvAliases) { $tokenEnvAliases += @($telegram.botTokenEnvAliases) } - $chatIdEnvAliases = @($chatIdEnv) - if ($telegram.chatIdEnvAliases) { $chatIdEnvAliases += @($telegram.chatIdEnvAliases) } - - function Get-AgentEnvSecret { - param([string[]]$Names) - foreach ($name in ($Names | Where-Object { $_ } | Select-Object -Unique)) { - foreach ($scope in @("Machine", "User", "Process")) { - $value = [Environment]::GetEnvironmentVariable($name, $scope) - if ($value) { - return [pscustomobject]@{ value = $value; name = $name; scope = $scope } - } - } - } - return $null - } - - # The canonical gateway transport is not wired into Agent99 yet. Keep all - # credential bindings unset so the fail-closed receipt below cannot read a - # host-local bot token/chat identifier before returning. - $tokenSecret = $null - $chatSecret = $null - $token = $null - $chatId = $null $chatIdOverride = if ($Data -and $Data.PSObject.Properties["replyChatId"]) { [string]$Data.replyChatId } else { "" } $card = Get-AgentIncidentCardModel $Severity $EventType $Message $Data $incidentState = Get-AgentTelegramIncidentState $card @@ -1984,9 +1724,9 @@ function Send-AgentTelegram { fingerprint = $card.fingerprint lifecycle = $card.lifecycle stateKey = $card.stateKey - configured = [bool]($token -and $chatId) - tokenEnv = if ($tokenSecret) { $tokenSecret.name } else { $null } - chatIdEnv = if ($chatSecret) { $chatSecret.name } else { $null } + configured = $false + tokenEnv = $null + chatIdEnv = $null chatOverride = if ($chatIdOverride) { "telegram-origin" } else { $null } replyToMessageId = if ($replyToMessageId) { $replyToMessageId } else { $null } messageId = $null @@ -2019,80 +1759,6 @@ function Send-AgentTelegram { } $script:TelegramAttempts += $attempt return $attempt - - $text = Format-AgentTelegramText $Severity $EventType $Message $Data - $caption = Format-AgentTelegramCaption $Severity $EventType $Message $Data - $visualPath = Get-AgentTelegramVisualPath $Data - if (-not $visualPath) { - $visualPath = New-AgentTelegramCardImage $Severity $EventType $text $Data - } - $attempt.visualPath = $visualPath - if (-not ($token -and ($chatId -or $chatIdOverride))) { - $relayText = if ($visualPath) { $caption } else { $text } - $relayResult = Send-AgentTelegramRelay $relayText $telegram.relay $chatIdOverride $visualPath $replyToMessageId - $attempt.relay = $relayResult - if ($relayResult.ok) { - $attempt.sent = $true - $attempt.messageId = $relayResult.messageId - $attempt.error = $null - $attempt.visualSent = [bool]$relayResult.photoSent - if ($visualPath -and -not $relayResult.photoSent) { - $attempt.visualError = if ($relayResult.photoError) { $relayResult.photoError } else { "relay_sendPhoto_not_available" } - } - } else { - $attempt.error = "telegram_not_configured" - } - if ($attempt.sent) { - $attempt.incidentStatePath = Save-AgentTelegramIncidentState $card $incidentState $attempt.messageId $replyToMessageId - $attempt.incidentStateWritten = $true - } - $script:TelegramAttempts += $attempt - return $attempt - } - - try { - $targetChat = if ($chatIdOverride) { $chatIdOverride } else { $chatId } - if ($visualPath -and (Test-Path -LiteralPath $visualPath)) { - try { - $photoResult = Send-AgentTelegramPhotoDirect $token $targetChat $caption $visualPath $replyToMessageId - $attempt.sent = $true - $attempt.visualSent = $true - $attempt.messageId = $photoResult.messageId - } catch { - $attempt.visualError = $_.Exception.Message - Write-AgentLog "telegram_sendPhoto_failed event=$EventType fallback=sendMessage error=$($_.Exception.Message)" - $fallbackBody = @{ - chat_id = $targetChat - text = $text - disable_web_page_preview = "true" - } - if ($replyToMessageId) { $fallbackBody.reply_parameters = (@{ message_id = [long]$replyToMessageId } | ConvertTo-Json -Compress) } - $fallbackResult = Invoke-RestMethod -Method Post -Uri "blocked://canonical-telegram-gateway-required/sendMessage" -Body $fallbackBody - $attempt.sent = $true - $attempt.messageId = if ($fallbackResult.result) { $fallbackResult.result.message_id } else { $null } - } - } else { - $textBody = @{ - chat_id = $targetChat - text = $text - disable_web_page_preview = "true" - } - if ($replyToMessageId) { $textBody.reply_parameters = (@{ message_id = [long]$replyToMessageId } | ConvertTo-Json -Compress) } - $textResult = Invoke-RestMethod -Method Post -Uri "blocked://canonical-telegram-gateway-required/sendMessage" -Body $textBody - $attempt.sent = $true - $attempt.messageId = if ($textResult.result) { $textResult.result.message_id } else { $null } - } - } catch { - $attempt.error = $_.Exception.Message - } - - if ($attempt.sent) { - $attempt.incidentStatePath = Save-AgentTelegramIncidentState $card $incidentState $attempt.messageId $replyToMessageId - $attempt.incidentStateWritten = $true - } - - $script:TelegramAttempts += $attempt - return $attempt } function Record-AgentEvent { diff --git a/apps/api/src/services/notifications/telegram.py b/apps/api/src/services/notifications/telegram.py index 5eb7bb94b..52d865199 100644 --- a/apps/api/src/services/notifications/telegram.py +++ b/apps/api/src/services/notifications/telegram.py @@ -123,7 +123,10 @@ class TelegramWebhookProvider(NotificationProvider): if not self.enabled: return False try: - from src.services.telegram_gateway import get_telegram_gateway + from src.services.telegram_gateway import ( + _telegram_send_delivery_succeeded, + get_telegram_gateway, + ) gw = get_telegram_gateway() result = await gw.send_alert_notification( @@ -132,7 +135,7 @@ class TelegramWebhookProvider(NotificationProvider): signal_family="product_raw_monitoring", severity="P3", ) - return bool(result.get("ok")) + return _telegram_send_delivery_succeeded(result) except Exception as e: logger.error("telegram_connection_test_failed", error=str(e)) return False diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 00e382f00..25bc95342 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -28,6 +28,7 @@ import html import json import os import re +import secrets from collections.abc import Mapping from dataclasses import dataclass, field from datetime import UTC, datetime @@ -146,7 +147,26 @@ _AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY = "_awooop_source_envelope_extra" _CANONICAL_ROUTE_CONTEXT_KEY = "_awoooi_canonical_telegram_route" _LEGACY_NOTIFICATION_TYPE_KEY = "_awoooi_legacy_notification_type" _NON_MONITORING_INTERACTION_KEY = "_awoooi_non_monitoring_interaction" +_BOT_TOKEN_OVERRIDE_KEY = "_awoooi_bot_token_override" +_ACTUAL_BOT_ALIAS_KEY = "_awoooi_actual_bot_alias" +_DELIVERY_CONTEXT_KEY = "_awoooi_delivery_context" _CANONICAL_SRE_DESTINATION = "telegram_chat_alias:awoooi_sre_war_room" +_GUARDED_TELEGRAM_BOT_METHODS = frozenset( + { + "sendMessage", + "sendDocument", + "sendPhoto", + "sendMediaGroup", + "editMessageText", + "sendAnimation", + "sendVideo", + "sendAudio", + "sendVoice", + } +) +_GUARDED_TELEGRAM_REPLY_METHODS = _GUARDED_TELEGRAM_BOT_METHODS - { + "editMessageText" +} _NON_MONITORING_INTERACTION_KINDS = frozenset( { "operator_command_reply", @@ -154,6 +174,64 @@ _NON_MONITORING_INTERACTION_KINDS = frozenset( "bot_discussion_reply", } ) + + +def _telegram_destination_binding(chat_id: object) -> str: + """Return a durable, redaction-safe binding for one Telegram destination.""" + if isinstance(chat_id, bool) or chat_id is None: + return "" + normalized = str(chat_id).strip() + if not normalized: + return "" + return hashlib.sha256( + f"telegram-destination-v1:{normalized}".encode() + ).hexdigest() + + +def _telegram_provider_delivery_evidence( + provider_result: object, +) -> tuple[list[int], str]: + """Extract positive message ids and one consistent provider chat binding.""" + if isinstance(provider_result, Mapping): + messages = [provider_result] + elif isinstance(provider_result, list) and provider_result: + messages = provider_result + else: + return [], "" + + message_ids: list[int] = [] + destination_bindings: set[str] = set() + destination_binding_complete = True + for message in messages: + if not isinstance(message, Mapping): + return [], "" + message_id = message.get("message_id") + if ( + isinstance(message_id, bool) + or not isinstance(message_id, int) + or message_id <= 0 + ): + return [], "" + provider_chat = message.get("chat") + if not isinstance(provider_chat, Mapping): + destination_binding_complete = False + message_ids.append(message_id) + continue + destination_binding = _telegram_destination_binding( + provider_chat.get("id") + ) + if not destination_binding: + destination_binding_complete = False + message_ids.append(message_id) + continue + message_ids.append(message_id) + destination_bindings.add(destination_binding) + + if not destination_binding_complete or len(destination_bindings) != 1: + return message_ids, "" + return message_ids, destination_bindings.pop() + + _CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS = 3 _NO_WRITE_REPLAY_DIGEST_WINDOW_MINUTES = 30 _HOST_RESOURCE_ALERT_HEADER_RE = re.compile( @@ -853,15 +931,87 @@ def _telegram_send_delivery_succeeded(result: object) -> bool: return False delivery_status = str(result.get("_awooop_delivery_status") or "").strip() - if delivery_status and delivery_status not in {"sent", "sent_reused"}: + if delivery_status != "sent": return False - if result.get("_awooop_provider_send_performed") is False: + if result.get("_awooop_provider_send_performed") is not True: return False route_receipt = result.get("_awoooi_canonical_route_receipt") if ( - isinstance(route_receipt, Mapping) - and route_receipt.get("provider_send_performed") is False + not isinstance(route_receipt, Mapping) + or route_receipt.get("schema_version") + != "telegram_canonical_egress_receipt_v1" + or route_receipt.get("decision") != "allowed" + or route_receipt.get("provider_send_performed") is not True + ): + return False + + delivery_context = result.get(_DELIVERY_CONTEXT_KEY) + if ( + not isinstance(delivery_context, Mapping) + or delivery_context.get("schema_version") + != "telegram_delivery_context_v1" + or delivery_context.get("destination_binding_verified") is not True + ): + return False + + route_sender_bot_alias = str( + route_receipt.get("sender_bot_alias") or "" + ).strip() + final_exit_sender_bot_alias = str( + delivery_context.get("sender_bot_alias") or "" + ).strip() + if ( + not route_sender_bot_alias + or route_sender_bot_alias != final_exit_sender_bot_alias + ): + return False + + route_destination_alias = str( + route_receipt.get("destination_alias") or "" + ).strip() + final_exit_destination_alias = str( + delivery_context.get("destination_alias") or "" + ).strip() + if ( + not route_destination_alias + or route_destination_alias != final_exit_destination_alias + ): + return False + + route_destination_binding = str( + route_receipt.get("destination_binding") or "" + ).strip() + final_exit_destination_binding = str( + delivery_context.get("destination_binding") or "" + ).strip() + payload_destination_binding = str( + delivery_context.get("payload_destination_binding") or "" + ).strip() + provider_destination_binding = str( + delivery_context.get("provider_destination_binding") or "" + ).strip() + if ( + not route_destination_binding + or len( + { + route_destination_binding, + final_exit_destination_binding, + payload_destination_binding, + provider_destination_binding, + } + ) + != 1 + ): + return False + + provider_result = result.get("result") + provider_message_ids, observed_provider_binding = ( + _telegram_provider_delivery_evidence(provider_result) + ) + if ( + not provider_message_ids + or observed_provider_binding != provider_destination_binding ): return False return True @@ -5272,6 +5422,8 @@ class TelegramGateway: "signal_family": str(context.get("signal_family") or "unknown")[:80], "severity": str(context.get("severity") or "unknown")[:16], "destination_alias": "none", + "destination_binding": "", + "sender_bot_alias": "", "provider_send_performed": False, } @@ -5363,6 +5515,9 @@ class TelegramGateway: "classifier": "canonical_shared_route_allowed", **normalized_context, "destination_alias": "awoooi_sre_war_room", + "destination_binding": _telegram_destination_binding( + canonical_chat_id + ), "sender_bot_alias": actual_bot_alias, "provider_send_performed": False, } @@ -5370,6 +5525,7 @@ class TelegramGateway: def _authorize_non_monitoring_interaction( self, *, + method: str, payload: Mapping[str, object], interaction_context: object, actual_bot_alias: str, @@ -5412,10 +5568,18 @@ class TelegramGateway: route_context={"signal_family": interaction_kind}, ) - reply_parameters = payload.get("reply_parameters") - reply_message_raw = payload.get("reply_to_message_id") - if isinstance(reply_parameters, Mapping): - reply_message_raw = reply_parameters.get("message_id") + if method == "editMessageText": + reply_message_raw = payload.get("message_id") + elif method in _GUARDED_TELEGRAM_REPLY_METHODS: + reply_parameters = payload.get("reply_parameters") + reply_message_raw = payload.get("reply_to_message_id") + if isinstance(reply_parameters, Mapping): + reply_message_raw = reply_parameters.get("message_id") + else: + return None, self._blocked_route_receipt( + "interaction_method_not_guarded", + route_context={"signal_family": interaction_kind}, + ) try: reply_message_id = int(str(reply_message_raw)) except (TypeError, ValueError): @@ -5441,6 +5605,9 @@ class TelegramGateway: "signal_family": interaction_kind, "severity": "N/A", "destination_alias": "inbound_interaction_reply_target", + "destination_binding": _telegram_destination_binding( + target_chat_id + ), "sender_bot_alias": actual_bot_alias, "inbound_context_verified": True, "provider_send_performed": False, @@ -5973,6 +6140,38 @@ class TelegramGateway: dict: API 回應 """ payload = dict(payload) + actual_bot_alias = str( + payload.pop(_ACTUAL_BOT_ALIAS_KEY, "tsenyang_bot") + ).strip() + token_override = payload.pop(_BOT_TOKEN_OVERRIDE_KEY, None) + provider_bot_token = self.bot_token + if token_override is None: + bot_binding_valid = actual_bot_alias == "tsenyang_bot" + else: + expected_token = { + "openclaw_bot": settings.OPENCLAW_BOT_TOKEN, + "nemotron_bot": settings.NEMOTRON_BOT_TOKEN, + }.get(actual_bot_alias, "") + bot_binding_valid = bool( + isinstance(token_override, str) + and expected_token + and secrets.compare_digest(token_override, expected_token) + ) + if bot_binding_valid: + provider_bot_token = str(token_override) + + if not bot_binding_valid: + route_receipt = self._blocked_route_receipt( + "sender_bot_binding_mismatch" + ) + return { + "ok": False, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "blocked_no_egress", + "_awooop_provider_send_performed": False, + "_awoooi_canonical_route_receipt": route_receipt, + } + route_context = payload.pop(_CANONICAL_ROUTE_CONTEXT_KEY, None) legacy_notification_type = payload.pop( _LEGACY_NOTIFICATION_TYPE_KEY, @@ -5983,22 +6182,34 @@ class TelegramGateway: None, ) canonical_route_receipt: dict[str, object] | None = None - if method == "sendMessage": - if route_context is not None or legacy_notification_type is not None: + if method in _GUARDED_TELEGRAM_BOT_METHODS: + has_route_context = ( + route_context is not None + or legacy_notification_type is not None + ) + if has_route_context and interaction_context is not None: + canonical_chat_id, canonical_route_receipt = ( + None, + self._blocked_route_receipt( + "ambiguous_route_and_interaction_context" + ), + ) + elif has_route_context: canonical_chat_id, canonical_route_receipt = ( self._authorize_canonical_send_message( payload=payload, route_context=route_context, legacy_notification_type=legacy_notification_type, - actual_bot_alias="tsenyang_bot", + actual_bot_alias=actual_bot_alias, ) ) elif interaction_context is not None: canonical_chat_id, canonical_route_receipt = ( self._authorize_non_monitoring_interaction( + method=method, payload=payload, interaction_context=interaction_context, - actual_bot_alias="tsenyang_bot", + actual_bot_alias=actual_bot_alias, ) ) else: @@ -6121,7 +6332,7 @@ class TelegramGateway: source_envelope_extra = dict(source_envelope_extra or {}) source_envelope_extra["agent99_dispatch_receipt"] = agent99_receipt - url = f"{self.api_url}/{method}" + url = f"{self.TELEGRAM_API_BASE}/bot{provider_bot_token}/{method}" # OTEL Span: telegram.api.{method} with _tracer.start_as_current_span( @@ -6144,12 +6355,80 @@ class TelegramGateway: f"Telegram API error: {result.get('description', 'Unknown error')}" ) - # 成功: 記錄 message_id (result 可能是 dict 或 bool,需防禦) + # Guarded methods reach terminal delivery only when Telegram's + # result binds the provider message back to the exact sender + # and destination authorized before this single final exit. result_val = result.get("result") - if isinstance(result_val, dict) and "message_id" in result_val: - span.set_attribute("telegram.message_id", result_val["message_id"]) - provider_message_id = str(result_val["message_id"]) - if ( + provider_message_ids, provider_destination_binding = ( + _telegram_provider_delivery_evidence(result_val) + ) + has_provider_message_id = bool(provider_message_ids) + provider_message_id_value = ( + provider_message_ids[0] + if provider_message_ids + else None + ) + payload_destination_binding = ( + _telegram_destination_binding(payload.get("chat_id")) + ) + route_destination_binding = str( + (canonical_route_receipt or {}).get( + "destination_binding" + ) + or "" + ) + destination_binding_verified = bool( + canonical_route_receipt is not None + and route_destination_binding + and route_destination_binding + == payload_destination_binding + == provider_destination_binding + ) + delivery_context = { + "schema_version": "telegram_delivery_context_v1", + "method": method, + "sender_bot_alias": actual_bot_alias, + "destination_alias": str( + (canonical_route_receipt or {}).get( + "destination_alias" + ) + or "" + ), + "destination_binding": route_destination_binding, + "payload_destination_binding": ( + payload_destination_binding + ), + "provider_destination_binding": ( + provider_destination_binding + ), + "destination_binding_verified": ( + destination_binding_verified + ), + } + result[_DELIVERY_CONTEXT_KEY] = delivery_context + if has_provider_message_id: + span.set_attribute( + "telegram.message_id", + provider_message_id_value, + ) + provider_message_id = str(provider_message_id_value) + if canonical_route_receipt is not None: + canonical_route_receipt["provider_send_performed"] = True + result["_awoooi_canonical_route_receipt"] = ( + canonical_route_receipt + ) + if not destination_binding_verified: + if controlled_reservation is not None: + result[ + "_awooop_outbound_mirror_acknowledged" + ] = False + result["_awooop_delivery_status"] = ( + "provider_destination_mismatch" + if provider_destination_binding + else "provider_destination_unverified" + ) + result["_awooop_provider_send_performed"] = True + elif ( controlled_reservation is not None and controlled_delivery_identity is not None ): @@ -6166,8 +6445,8 @@ class TelegramGateway: ) result["_awooop_provider_send_performed"] = True else: - if canonical_route_receipt is not None: - canonical_route_receipt["provider_send_performed"] = True + result["_awooop_delivery_status"] = "sent" + result["_awooop_provider_send_performed"] = True source_envelope_extra = ( _acknowledge_callback_reply_source_envelope( source_envelope_extra, @@ -6181,13 +6460,20 @@ class TelegramGateway: provider_message_id=provider_message_id, source_envelope_extra=source_envelope_extra, ) - elif controlled_reservation is not None: - result["_awooop_outbound_mirror_acknowledged"] = False - result["_awooop_delivery_status"] = "pending_unknown" + else: + if controlled_reservation is not None: + result["_awooop_outbound_mirror_acknowledged"] = False + if canonical_route_receipt is not None: + canonical_route_receipt["provider_send_performed"] = True + result["_awooop_delivery_status"] = ( + "provider_message_id_missing" + ) result["_awooop_provider_send_performed"] = True - if canonical_route_receipt is not None: - canonical_route_receipt["provider_send_performed"] = True + if ( + canonical_route_receipt is not None + and "_awoooi_canonical_route_receipt" not in result + ): result["_awoooi_canonical_route_receipt"] = ( canonical_route_receipt ) @@ -9139,6 +9425,11 @@ class TelegramGateway: "parse_mode": "HTML", "reply_markup": {"inline_keyboard": []}, "disable_web_page_preview": True, + _NON_MONITORING_INTERACTION_KEY: { + "interaction_kind": "callback_reply", + "inbound_chat_id": inbound_chat_id, + "inbound_message_id": inbound_message_id, + }, }) except Exception: # 移除按鈕保底 @@ -9777,6 +10068,11 @@ class TelegramGateway: "parse_mode": "HTML", "reply_markup": {"inline_keyboard": []}, "disable_web_page_preview": True, + _NON_MONITORING_INTERACTION_KEY: { + "interaction_kind": "callback_reply", + "inbound_chat_id": inbound_chat_id, + "inbound_message_id": inbound_message_id, + }, }) except TelegramGatewayError as e: # 文字更新失敗不影響整體流程,按鈕已移除 @@ -11553,8 +11849,7 @@ class TelegramGateway: actual_bot_alias: str, inbound_chat_id: str | int | None = None, ) -> dict: - """ - 用指定 Bot Token 發訊息。 + """Route an alternate owned bot through the canonical final exit. Args: token: Bot Token @@ -11565,82 +11860,31 @@ class TelegramGateway: Returns: dict: Telegram API 回應 """ - route_context = ( - { + payload: dict[str, object] = { + "text": text[:4096], + "parse_mode": parse_mode, + _BOT_TOKEN_OVERRIDE_KEY: token, + _ACTUAL_BOT_ALIAS_KEY: actual_bot_alias, + } + if product_id and signal_family and severity: + payload[_CANONICAL_ROUTE_CONTEXT_KEY] = { "product_id": product_id, "signal_family": signal_family, "severity": severity, } - if product_id and signal_family and severity - else None - ) - candidate_payload: dict[str, object] = { - "chat_id": str(inbound_chat_id or settings.SRE_GROUP_CHAT_ID or ""), - "text": text[:4096], - "parse_mode": parse_mode, - } - if reply_to_message_id: - candidate_payload["reply_parameters"] = { + elif reply_to_message_id: + payload["chat_id"] = str(inbound_chat_id or "") + payload[_NON_MONITORING_INTERACTION_KEY] = { + "interaction_kind": "bot_discussion_reply", + "inbound_chat_id": str(inbound_chat_id or ""), + "inbound_message_id": reply_to_message_id, + } + if reply_to_message_id is not None: + payload["reply_parameters"] = { "message_id": reply_to_message_id, "allow_sending_without_reply": True, } - - if route_context is not None: - chat_id, route_receipt = self._authorize_canonical_send_message( - payload=candidate_payload, - route_context=route_context, - legacy_notification_type=None, - actual_bot_alias=actual_bot_alias, - ) - elif reply_to_message_id: - chat_id, route_receipt = self._authorize_non_monitoring_interaction( - payload=candidate_payload, - interaction_context={ - "interaction_kind": "bot_discussion_reply", - "inbound_chat_id": str(inbound_chat_id or ""), - "inbound_message_id": reply_to_message_id, - }, - actual_bot_alias=actual_bot_alias, - ) - else: - chat_id, route_receipt = ( - None, - self._blocked_route_receipt( - "missing_product_signal_severity_or_interaction_context" - ), - ) - if chat_id is None: - return { - "ok": False, - "_awooop_outbound_mirror_acknowledged": False, - "_awooop_delivery_status": "blocked_no_egress", - "_awooop_provider_send_performed": False, - "_awoooi_canonical_route_receipt": route_receipt, - } - - if not self._initialized: - await self.initialize() - if not self._http_client: - raise TelegramGatewayError("HTTP client not initialized") - - url = f"{self.TELEGRAM_API_BASE}/bot{token}/sendMessage" - payload = dict(candidate_payload) - payload["chat_id"] = chat_id - - response = await self._http_client.post(url, json=payload) - response.raise_for_status() - result = response.json() - route_receipt["provider_send_performed"] = True - if isinstance(result, dict): - result["_awoooi_canonical_route_receipt"] = route_receipt - result_val = result.get("result") if isinstance(result, dict) else None - if isinstance(result_val, dict) and "message_id" in result_val: - await self._mirror_outbound_message( - method="sendMessage", - payload=payload, - provider_message_id=str(result_val["message_id"]), - ) - return result + return await self._send_request("sendMessage", payload) async def send_as_openclaw( self, diff --git a/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py b/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py index 112cb8c9e..3ed3908d6 100644 --- a/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py +++ b/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py @@ -96,7 +96,7 @@ def test_agent99_v5_formatter_synthetic_gate_loads_model_dependencies() -> None: assert 'C:\\Wooo' in synthetic -def test_agent99_telegram_threads_source_alert_and_persists_lifecycle() -> None: +def test_agent99_telegram_lifecycle_fails_closed_without_direct_sender() -> None: source = CONTROL.read_text(encoding="utf-8") inbox = SRE_INBOX.read_text(encoding="utf-8") @@ -104,8 +104,6 @@ def test_agent99_telegram_threads_source_alert_and_persists_lifecycle() -> None: assert "replyMessageId = $replyMessageId" in inbox assert 'PSObject.Properties["replyMessageId"]' in source assert 'PSObject.Properties["correlationKey"]' in source - assert '"reply_parameters"' in source - assert 'telegram_receipt_b64=' in source assert 'schemaVersion = "agent99_telegram_incident_state_v1"' in source assert 'rootMessageId = $rootMessageId' in source assert 'reason = "same_lifecycle_state"' in source @@ -113,6 +111,12 @@ def test_agent99_telegram_threads_source_alert_and_persists_lifecycle() -> None: assert "lastRunKey = $Card.runKey" in source assert "$previousRunKey -eq [string]$incidentCard.runKey" in source assert '$dedupeParts += "run=$($incidentCard.runKey)"' in source + assert 'error = "canonical_telegram_gateway_transport_required"' in source + assert 'providerSendPerformed = $false' in source + assert 'routeStatus = "blocked_no_egress"' in source + assert "function Send-AgentTelegramRelay" not in source + assert "function Send-AgentTelegramPhotoDirect" not in source + assert "telegram_receipt_b64=" not in source def test_agent99_records_deduped_telegram_as_an_explicit_attempt() -> None: @@ -495,7 +499,7 @@ def test_agent99_bounds_stale_ssh_processes_without_touching_tunnels() -> None: source = CONTROL.read_text(encoding="utf-8") config = json.loads((ROOT / "agent99.config.99.example.json").read_text()) guard = source[source.index("function Invoke-AgentStaleSshProcessGuard") :] - guard = guard[: guard.index("function Send-AgentTelegramRelay")] + guard = guard[: guard.index("function Get-AgentObjectValue")] assert 'Get-CimInstance Win32_Process -Filter "Name=\'ssh.exe\'"' in guard assert 'CommandLine -cmatch "(^|\\s)-[NLRD](\\s|$)"' in guard diff --git a/apps/api/tests/test_heartbeat_dedup_p0_4.py b/apps/api/tests/test_heartbeat_dedup_p0_4.py index 691c3cf88..6e3d8970a 100644 --- a/apps/api/tests/test_heartbeat_dedup_p0_4.py +++ b/apps/api/tests/test_heartbeat_dedup_p0_4.py @@ -14,6 +14,37 @@ from unittest.mock import AsyncMock, patch import pytest +import src.services.telegram_gateway as gateway_module + +_SENT_CHAT_ID = -1001 +_SENT_DESTINATION_BINDING = gateway_module._telegram_destination_binding( + _SENT_CHAT_ID +) +_SENT_RECEIPT = { + "ok": True, + "_awooop_delivery_status": "sent", + "_awooop_provider_send_performed": True, + "result": {"message_id": 42, "chat": {"id": _SENT_CHAT_ID}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "payload_destination_binding": _SENT_DESTINATION_BINDING, + "provider_destination_binding": _SENT_DESTINATION_BINDING, + "destination_binding_verified": True, + }, +} + class FakeRedis: """模擬 Redis 行為,記錄 set/get/delete 呼叫""" @@ -78,7 +109,7 @@ def gateway_with_fake_redis(): gw = TelegramGateway.__new__(TelegramGateway) # 跳過 __init__ gw._initialized = True gw._last_message_time = None - gw.send_to_group = AsyncMock() + gw.send_to_group = AsyncMock(return_value=dict(_SENT_RECEIPT)) gw.send_notification = AsyncMock() fake_redis = FakeRedis() diff --git a/apps/api/tests/test_telegram_callback_context_routing.py b/apps/api/tests/test_telegram_callback_context_routing.py index 3eca42751..5317eeaef 100644 --- a/apps/api/tests/test_telegram_callback_context_routing.py +++ b/apps/api/tests/test_telegram_callback_context_routing.py @@ -20,6 +20,31 @@ from src.services.telegram_gateway import TelegramGateway _CHAT_ID = "verified-chat" _MESSAGE_ID = 77 +_DESTINATION_BINDING = gateway_module._telegram_destination_binding(_CHAT_ID) +_SENT_RECEIPT = { + "ok": True, + "_awooop_delivery_status": "sent", + "_awooop_provider_send_performed": True, + "result": {"message_id": _MESSAGE_ID, "chat": {"id": _CHAT_ID}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "inbound_interaction_reply_target", + "destination_binding": _DESTINATION_BINDING, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "inbound_interaction_reply_target", + "destination_binding": _DESTINATION_BINDING, + "payload_destination_binding": _DESTINATION_BINDING, + "provider_destination_binding": _DESTINATION_BINDING, + "destination_binding_verified": True, + }, +} def _assert_verified_callback_payload(payload: dict) -> None: @@ -174,13 +199,15 @@ async def test_llm_and_category_results_keep_verified_callback_context( class _Redis: async def get(self, _key: str) -> str: - return json.dumps({ - "name": "inspect", - "provider": "internal", - "tool": "status", - "risk": "low", - "incident_id": "INC-1", - }) + return json.dumps( + { + "name": "inspect", + "provider": "internal", + "tool": "status", + "risk": "low", + "incident_id": "INC-1", + } + ) monkeypatch.setattr(gateway_module, "get_redis", lambda: _Redis()) monkeypatch.setattr( @@ -213,13 +240,15 @@ async def test_llm_and_category_results_keep_verified_callback_context( monkeypatch.setattr( callback_dispatcher, "dispatch_action", - AsyncMock(return_value=DispatchResult( - success=True, - action="inspect", - incident_id="INC-1", - user_id=42, - result_text="result", - )), + AsyncMock( + return_value=DispatchResult( + success=True, + action="inspect", + incident_id="INC-1", + user_id=42, + result_text="result", + ) + ), ) class _IncidentRepo: @@ -264,7 +293,9 @@ async def test_write_category_callback_keeps_verified_context( "verify_callback", AsyncMock(return_value={"id": 42}), ) - monkeypatch.setattr(gateway, "_check_incident_state_guard", AsyncMock(return_value=None)) + monkeypatch.setattr( + gateway, "_check_incident_state_guard", AsyncMock(return_value=None) + ) monkeypatch.setattr(gateway, "_answer_callback", AsyncMock()) spec = SimpleNamespace( requires_multi_sig=False, @@ -278,13 +309,15 @@ async def test_write_category_callback_keeps_verified_context( monkeypatch.setattr( callback_dispatcher, "dispatch_action", - AsyncMock(return_value=DispatchResult( - success=True, - action="restart_safe", - incident_id="INC-1", - user_id=42, - result_text="done", - )), + AsyncMock( + return_value=DispatchResult( + success=True, + action="restart_safe", + incident_id="INC-1", + user_id=42, + result_text="done", + ) + ), ) class _IncidentRepo: @@ -335,9 +368,7 @@ async def test_approval_replies_keep_verified_callback_context( _assert_verified_callback_payload(status_payload) source_extra = status_payload[gateway_module._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY] merged = gateway_module._merge_outbound_source_envelope_extra({}, source_extra) - assert merged["callback_reply"]["parent_provider_message_id"] == str( - _MESSAGE_ID - ) + assert merged["callback_reply"]["parent_provider_message_id"] == str(_MESSAGE_ID) send_request.reset_mock() approval_id = UUID("44444444-4444-4444-4444-444444444444") @@ -367,8 +398,9 @@ async def test_approval_replies_keep_verified_callback_context( @pytest.mark.parametrize( ("result", "expected"), [ - ({"ok": True}, True), - ({"ok": True, "_awooop_delivery_status": "sent_reused"}, True), + (_SENT_RECEIPT, True), + ({"ok": True}, False), + ({"ok": True, "_awooop_delivery_status": "sent_reused"}, False), ({"ok": False}, False), ( {"ok": True, "_awooop_delivery_status": "blocked_no_egress"}, @@ -414,7 +446,7 @@ async def test_approval_structured_no_send_retries_plain_on_exact_parent( ) -> None: gateway = TelegramGateway() attempts: list[tuple[str, dict]] = [] - send_results = iter([primary_failure, {"ok": True}]) + send_results = iter([primary_failure, dict(_SENT_RECEIPT)]) async def structured_primary_failure(method: str, payload: dict) -> dict: attempts.append((method, payload)) @@ -423,7 +455,7 @@ async def test_approval_structured_no_send_retries_plain_on_exact_parent( return {"ok": True} record_failure = AsyncMock(return_value=None) - shared_fallback = AsyncMock(return_value={"ok": True}) + shared_fallback = AsyncMock(return_value=dict(_SENT_RECEIPT)) monkeypatch.setattr(gateway, "_send_request", structured_primary_failure) monkeypatch.setattr(gateway, "_record_callback_reply_failure", record_failure) monkeypatch.setattr(gateway, "send_alert_notification", shared_fallback) @@ -454,10 +486,12 @@ async def test_approval_structured_double_no_send_records_durable_failure( ) -> None: gateway = TelegramGateway() attempts: list[tuple[str, dict]] = [] - send_results = iter([ - {"ok": True, "_awooop_delivery_status": "blocked_no_egress"}, - {"ok": True, "_awooop_provider_send_performed": False}, - ]) + send_results = iter( + [ + {"ok": True, "_awooop_delivery_status": "blocked_no_egress"}, + {"ok": True, "_awooop_provider_send_performed": False}, + ] + ) async def structured_no_send(method: str, payload: dict) -> dict: attempts.append((method, payload)) @@ -501,7 +535,7 @@ async def test_callback_card_edits_use_only_verified_inbound_identity( action: str, ) -> None: gateway = TelegramGateway() - send_request = AsyncMock(return_value={"ok": True}) + send_request = AsyncMock(return_value=dict(_SENT_RECEIPT)) monkeypatch.setattr(gateway, "_send_request", send_request) await gateway._update_message_after_action( @@ -658,17 +692,20 @@ async def test_autonomous_lifecycle_threads_use_canonical_route_and_parent_recei monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = TelegramGateway() - send_request = AsyncMock(return_value={"ok": True}) + send_request = AsyncMock(return_value=dict(_SENT_RECEIPT)) monkeypatch.setattr(gateway, "_send_request", send_request) monkeypatch.setattr(gateway_module, "get_redis", lambda: _LifecycleRedis()) assert await gateway.mark_auto_repaired("INC-1", "playbook", 10) is True assert await gateway.append_incident_update("INC-2", "healthy") is True - assert await gateway.append_grouped_alert_digest( - incident_id="INC-3", - group_key="group-1", - digest_text="digest", - ) is True + assert ( + await gateway.append_grouped_alert_digest( + incident_id="INC-3", + group_key="group-1", + digest_text="digest", + ) + is True + ) send_payloads = [ call.args[1] diff --git a/apps/api/tests/test_telegram_canonical_sender_gate.py b/apps/api/tests/test_telegram_canonical_sender_gate.py index 2231e277a..02cc352f6 100644 --- a/apps/api/tests/test_telegram_canonical_sender_gate.py +++ b/apps/api/tests/test_telegram_canonical_sender_gate.py @@ -1,6 +1,8 @@ from __future__ import annotations +import runpy from contextlib import asynccontextmanager +from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock @@ -11,13 +13,24 @@ import src.services.approval_db as approval_db import src.services.telegram_gateway as gateway_module from src.services.telegram_gateway import TelegramGateway +_REPO_ROOT = Path(__file__).resolve().parents[3] + class _FakeResponse: + def __init__(self, payload: dict) -> None: + self._payload = payload + def raise_for_status(self) -> None: return None def json(self) -> dict: - return {"ok": True, "result": {"message_id": 11}} + return { + "ok": True, + "result": { + "message_id": 11, + "chat": {"id": self._payload.get("chat_id")}, + }, + } class _FakeHttpClient: @@ -26,10 +39,12 @@ class _FakeHttpClient: async def post(self, url: str, json: dict) -> _FakeResponse: self.posts.append((url, json)) - return _FakeResponse() + return _FakeResponse(json) -def _prepared_gateway(monkeypatch: pytest.MonkeyPatch) -> tuple[TelegramGateway, _FakeHttpClient]: +def _prepared_gateway( + monkeypatch: pytest.MonkeyPatch, +) -> tuple[TelegramGateway, _FakeHttpClient]: gateway = TelegramGateway() client = _FakeHttpClient() gateway._initialized = True @@ -115,6 +130,77 @@ async def test_notification_provider_requires_affirmative_delivery_ack( assert result.error == "delivery_not_acknowledged" +@pytest.mark.asyncio +async def test_notification_connection_probe_rejects_structured_no_provider_send( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services.notifications import telegram as provider_module + + monkeypatch.setattr(provider_module.settings, "OPENCLAW_TG_BOT_TOKEN", "configured") + monkeypatch.setattr(provider_module.settings, "SRE_GROUP_CHAT_ID", "test-room") + gateway = SimpleNamespace( + send_alert_notification=AsyncMock( + return_value={ + "ok": True, + "_awooop_delivery_status": "blocked_no_egress", + "_awooop_provider_send_performed": False, + "_awoooi_canonical_route_receipt": { + "decision": "blocked_no_egress", + "provider_send_performed": False, + }, + } + ) + ) + monkeypatch.setattr(gateway_module, "get_telegram_gateway", lambda: gateway) + + assert await provider_module.TelegramWebhookProvider().test_connection() is False + gateway.send_alert_notification.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_notification_connection_probe_accepts_complete_canonical_delivery( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services.notifications import telegram as provider_module + + monkeypatch.setattr(provider_module.settings, "OPENCLAW_TG_BOT_TOKEN", "configured") + monkeypatch.setattr(provider_module.settings, "SRE_GROUP_CHAT_ID", "test-room") + chat_id = -1001 + destination_binding = gateway_module._telegram_destination_binding(chat_id) + gateway = SimpleNamespace( + send_alert_notification=AsyncMock( + return_value={ + "ok": True, + "_awooop_delivery_status": "sent", + "_awooop_provider_send_performed": True, + "result": {"message_id": 42, "chat": {"id": chat_id}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": destination_binding, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": destination_binding, + "payload_destination_binding": destination_binding, + "provider_destination_binding": destination_binding, + "destination_binding_verified": True, + }, + } + ) + ) + monkeypatch.setattr(gateway_module, "get_telegram_gateway", lambda: gateway) + + assert await provider_module.TelegramWebhookProvider().test_connection() is True + gateway.send_alert_notification.assert_awaited_once() + + @pytest.mark.asyncio async def test_unclassified_send_message_is_blocked_before_provider( monkeypatch: pytest.MonkeyPatch, @@ -131,6 +217,92 @@ async def test_unclassified_send_message_is_blocked_before_provider( assert client.posts == [] +@pytest.mark.asyncio +@pytest.mark.parametrize( + "method", + sorted(gateway_module._GUARDED_TELEGRAM_BOT_METHODS), +) +async def test_every_guarded_method_without_context_is_blocked_before_provider( + monkeypatch: pytest.MonkeyPatch, + method: str, +) -> None: + gateway, client = _prepared_gateway(monkeypatch) + + result = await gateway._send_request( + method, + { + "chat_id": "test-room", + "text": "unclassified", + "caption": "unclassified", + "photo": "file-placeholder", + "document": "file-placeholder", + "media": [], + "message_id": 42, + }, + ) + + assert result["_awooop_delivery_status"] == "blocked_no_egress" + assert result["_awooop_provider_send_performed"] is False + assert client.posts == [] + + +@pytest.mark.asyncio +async def test_guarded_photo_accepts_only_a_valid_canonical_route( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway, client = _prepared_gateway(monkeypatch) + + result = await gateway._send_request( + "sendPhoto", + { + "chat_id": "test-room", + "photo": "file-placeholder", + gateway_module._CANONICAL_ROUTE_CONTEXT_KEY: { + "product_id": "awoooi", + "signal_family": "security_incident", + "severity": "P1", + }, + }, + ) + + assert result["_awooop_delivery_status"] == "sent" + assert gateway_module._telegram_send_delivery_succeeded(result) is True + assert len(client.posts) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("method", "method_payload"), + [ + ("sendDocument", {"document": "file-placeholder", "reply_to_message_id": 42}), + ("editMessageText", {"text": "updated", "message_id": 42}), + ], +) +async def test_guarded_interaction_method_requires_exact_inbound_binding( + monkeypatch: pytest.MonkeyPatch, + method: str, + method_payload: dict, +) -> None: + gateway, client = _prepared_gateway(monkeypatch) + + result = await gateway._send_request( + method, + { + "chat_id": "operator-room", + **method_payload, + gateway_module._NON_MONITORING_INTERACTION_KEY: { + "interaction_kind": "callback_reply", + "inbound_chat_id": "operator-room", + "inbound_message_id": 42, + }, + }, + ) + + assert result["_awooop_delivery_status"] == "sent" + assert gateway_module._telegram_send_delivery_succeeded(result) is True + assert len(client.posts) == 1 + + @pytest.mark.asyncio async def test_info_and_business_methods_are_blocked_before_provider( monkeypatch: pytest.MonkeyPatch, @@ -209,15 +381,77 @@ async def test_allowed_legacy_monitoring_methods_use_canonical_provider_once( assert result["ok"] is True assert result["_awoooi_canonical_route_receipt"]["decision"] == "allowed" assert result["_awoooi_canonical_route_receipt"]["provider_send_performed"] is True + assert gateway_module._telegram_send_delivery_succeeded(result) is True assert len(client.posts) == 1 assert client.posts[0][1]["chat_id"] == "test-room" +@pytest.mark.asyncio +async def test_provider_ok_without_message_id_is_non_terminal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway, client = _prepared_gateway(monkeypatch) + monkeypatch.setattr( + _FakeResponse, + "json", + lambda _self: {"ok": True, "result": {}}, + ) + + result = await gateway.send_escalation_card( + incident_id="INC-TEST", + original_alertname="Outage", + duration_min=15, + ) + + assert result["_awooop_delivery_status"] == "provider_message_id_missing" + assert result["_awooop_provider_send_performed"] is True + assert result["_awoooi_canonical_route_receipt"]["provider_send_performed"] is True + assert gateway_module._telegram_send_delivery_succeeded(result) is False + assert len(client.posts) == 1 + + +@pytest.mark.asyncio +async def test_provider_chat_mismatch_is_non_terminal_delivery( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway, client = _prepared_gateway(monkeypatch) + monkeypatch.setattr( + _FakeResponse, + "json", + lambda _self: { + "ok": True, + "result": {"message_id": 11, "chat": {"id": "wrong-room"}}, + }, + ) + + result = await gateway.send_escalation_card( + incident_id="INC-TEST", + original_alertname="Outage", + duration_min=15, + ) + + assert result["_awooop_delivery_status"] == "provider_destination_mismatch" + assert result["_awooop_provider_send_performed"] is True + assert ( + result[gateway_module._DELIVERY_CONTEXT_KEY][ + "destination_binding_verified" + ] + is False + ) + assert gateway_module._telegram_send_delivery_succeeded(result) is False + assert len(client.posts) == 1 + + @pytest.mark.asyncio async def test_openclaw_cannot_own_tsenyang_monitoring_route( monkeypatch: pytest.MonkeyPatch, ) -> None: gateway, client = _prepared_gateway(monkeypatch) + monkeypatch.setattr( + gateway_module.settings, + "OPENCLAW_BOT_TOKEN", + "test-token-placeholder", + ) result = await gateway._send_as_bot( token="test-token-placeholder", @@ -235,6 +469,100 @@ async def test_openclaw_cannot_own_tsenyang_monitoring_route( assert client.posts == [] +@pytest.mark.asyncio +async def test_alternate_bot_uses_only_canonical_send_request( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway = TelegramGateway() + canonical_send = AsyncMock( + return_value={ + "ok": True, + "_awooop_delivery_status": "sent", + "_awooop_provider_send_performed": True, + "result": {"message_id": 11}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "provider_send_performed": True, + }, + } + ) + monkeypatch.setattr(gateway, "_send_request", canonical_send) + + result = await gateway._send_as_bot( + token="test-token-placeholder", + text="bounded reply", + reply_to_message_id=11, + actual_bot_alias="openclaw_bot", + inbound_chat_id="verified-room", + ) + + assert result["_awooop_delivery_status"] == "sent" + canonical_send.assert_awaited_once() + method, payload = canonical_send.await_args.args + assert method == "sendMessage" + assert payload[gateway_module._ACTUAL_BOT_ALIAS_KEY] == "openclaw_bot" + assert payload[gateway_module._BOT_TOKEN_OVERRIDE_KEY] == ("test-token-placeholder") + assert payload[gateway_module._NON_MONITORING_INTERACTION_KEY] == { + "interaction_kind": "bot_discussion_reply", + "inbound_chat_id": "verified-room", + "inbound_message_id": 11, + } + + +@pytest.mark.asyncio +async def test_alternate_bot_binding_mismatch_blocks_before_http( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway, client = _prepared_gateway(monkeypatch) + monkeypatch.setattr( + gateway_module.settings, + "OPENCLAW_BOT_TOKEN", + "expected-placeholder", + ) + + result = await gateway._send_as_bot( + token="mismatched-placeholder", + text="must not send", + reply_to_message_id=11, + actual_bot_alias="openclaw_bot", + inbound_chat_id="test-room", + ) + + assert result["_awooop_delivery_status"] == "blocked_no_egress" + assert result["_awoooi_canonical_route_receipt"]["classifier"] == ( + "sender_bot_binding_mismatch" + ) + assert client.posts == [] + + +def test_readability_guard_uses_complete_ast_function_boundary() -> None: + guard = runpy.run_path( + str(_REPO_ROOT / "scripts" / "security" / "telegram-alert-readability-guard.py") + ) + long_body = "\n".join(f" value_{index} = {index}" for index in range(300)) + source = ( + 'DECOY = "async def _send_request"\n' + "class Gateway:\n" + " async def _send_request(self, payload):\n" + f"{long_body}\n" + " return normalize_telegram_send_message_payload(\n" + ' "sendMessage", payload\n' + " )\n\n" + " async def unrelated(self):\n" + " return False\n" + ) + + segment = guard["function_segment"]( + source, + "async def _send_request", + ) + + assert len(segment) > 2200 + assert "normalize_telegram_send_message_payload" in segment + assert "async def unrelated" not in segment + + @pytest.mark.asyncio async def test_operator_reply_lane_and_runtime_edit_binding_remain_available( monkeypatch: pytest.MonkeyPatch, diff --git a/apps/api/tests/test_telegram_delivery_truth_callers.py b/apps/api/tests/test_telegram_delivery_truth_callers.py index 604c6d904..8885d4f6c 100644 --- a/apps/api/tests/test_telegram_delivery_truth_callers.py +++ b/apps/api/tests/test_telegram_delivery_truth_callers.py @@ -13,6 +13,7 @@ from src.models.incident import Incident, Severity, Signal from src.repositories.alert_operation_log_repository import ALERT_EVENT_TYPES from src.services import decision_manager as decision_module from src.services import report_generation_service as report_module +from src.services import telegram_gateway as gateway_module from src.services.ai_rate_limiter import AIRateLimiter from src.services.approval_execution import ApprovalExecutionService from src.services.report_generation_service import ReportGenerationService @@ -27,11 +28,33 @@ NO_SEND_RECEIPT = { }, } +_SENT_CHAT_ID = -1001 +_SENT_DESTINATION_BINDING = gateway_module._telegram_destination_binding( + _SENT_CHAT_ID +) SENT_RECEIPT = { "ok": True, "_awooop_delivery_status": "sent", "_awooop_provider_send_performed": True, "result": {"message_id": 42, "chat": {"id": -1001}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "payload_destination_binding": _SENT_DESTINATION_BINDING, + "provider_destination_binding": _SENT_DESTINATION_BINDING, + "destination_binding_verified": True, + }, } @@ -260,9 +283,7 @@ async def test_approval_blocked_receipt_writes_no_send_not_sent(monkeypatch) -> repair_attempted=True, ) - assert [row["event_type"] for row in operations] == [ - "NOTIFICATION_CLASSIFIED" - ] + assert [row["event_type"] for row in operations] == ["NOTIFICATION_CLASSIFIED"] assert operations[0]["event_type"] in ALERT_EVENT_TYPES assert operations[0]["action_detail"] == "telegram_execution_result_no_send" assert operations[0]["success"] is False diff --git a/apps/api/tests/test_telegram_delivery_truth_callers_v2.py b/apps/api/tests/test_telegram_delivery_truth_callers_v2.py index 841305d6b..18904e837 100644 --- a/apps/api/tests/test_telegram_delivery_truth_callers_v2.py +++ b/apps/api/tests/test_telegram_delivery_truth_callers_v2.py @@ -24,6 +24,7 @@ from src.services import converged_alert_recurrence_notifier as recurrence_modul from src.services import decision_manager as decision_module from src.services import failover_alerter as failover_module from src.services import k3s_monitor_service as k3s_module +from src.services import telegram_gateway as gateway_module from src.services import weekly_report_service as weekly_module NO_PROVIDER_SEND_RECEIPT = { @@ -36,11 +37,33 @@ NO_PROVIDER_SEND_RECEIPT = { }, } +_SENT_CHAT_ID = -1001 +_SENT_DESTINATION_BINDING = gateway_module._telegram_destination_binding( + _SENT_CHAT_ID +) SENT_RECEIPT = { "ok": True, "_awooop_delivery_status": "sent", "_awooop_provider_send_performed": True, "result": {"message_id": 42, "chat": {"id": -1001}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "payload_destination_binding": _SENT_DESTINATION_BINDING, + "provider_destination_binding": _SENT_DESTINATION_BINDING, + "destination_binding_verified": True, + }, } @@ -170,7 +193,9 @@ async def test_watchdog_releases_dedup_and_does_not_log_sent_without_provider_ac logger = _Logger() monkeypatch.setattr(watchdog_module, "logger", logger) monkeypatch.setattr(watchdog_module, "get_redis", lambda: redis) - monkeypatch.setattr(watchdog_module, "_is_grace_active", AsyncMock(return_value=True)) + monkeypatch.setattr( + watchdog_module, "_is_grace_active", AsyncMock(return_value=True) + ) monkeypatch.setattr( watchdog_module, "_count_pending_no_tg_sent", diff --git a/apps/api/tests/test_telegram_delivery_truth_gateway_v2.py b/apps/api/tests/test_telegram_delivery_truth_gateway_v2.py index 82d5fd041..b92866146 100644 --- a/apps/api/tests/test_telegram_delivery_truth_gateway_v2.py +++ b/apps/api/tests/test_telegram_delivery_truth_gateway_v2.py @@ -25,11 +25,33 @@ NO_SEND_RECEIPT = { }, } +_SENT_CHAT_ID = -1001 +_SENT_DESTINATION_BINDING = gateway_module._telegram_destination_binding( + _SENT_CHAT_ID +) SENT_RECEIPT = { "ok": True, "_awooop_delivery_status": "sent", "_awooop_provider_send_performed": True, - "result": {"message_id": 42}, + "result": {"message_id": 42, "chat": {"id": _SENT_CHAT_ID}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "payload_destination_binding": _SENT_DESTINATION_BINDING, + "provider_destination_binding": _SENT_DESTINATION_BINDING, + "destination_binding_verified": True, + }, } @@ -73,9 +95,86 @@ class FakeRedis: return True -def test_delivery_helper_rejects_truthy_structured_no_send() -> None: - assert _telegram_send_delivery_succeeded(dict(NO_SEND_RECEIPT)) is False - assert _telegram_send_delivery_succeeded({"ok": True}) is True +@pytest.mark.parametrize( + ("receipt", "expected"), + [ + (SENT_RECEIPT, True), + (NO_SEND_RECEIPT, False), + ({"ok": True}, False), + ({**SENT_RECEIPT, "ok": 1}, False), + ({**SENT_RECEIPT, "_awooop_delivery_status": ""}, False), + ({**SENT_RECEIPT, "_awooop_provider_send_performed": None}, False), + ({**SENT_RECEIPT, "result": {}}, False), + ({**SENT_RECEIPT, "result": {"message_id": 0}}, False), + ({**SENT_RECEIPT, "result": {"message_id": True}}, False), + ({**SENT_RECEIPT, "result": {"message_id": "42"}}, False), + ({**SENT_RECEIPT, "_awoooi_canonical_route_receipt": {}}, False), + ({**SENT_RECEIPT, gateway_module._DELIVERY_CONTEXT_KEY: {}}, False), + ( + { + **SENT_RECEIPT, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "blocked_no_egress", + "provider_send_performed": True, + }, + }, + False, + ), + ( + { + **SENT_RECEIPT, + "_awoooi_canonical_route_receipt": { + **SENT_RECEIPT["_awoooi_canonical_route_receipt"], + "sender_bot_alias": "", + }, + }, + False, + ), + ( + { + **SENT_RECEIPT, + gateway_module._DELIVERY_CONTEXT_KEY: { + **SENT_RECEIPT[gateway_module._DELIVERY_CONTEXT_KEY], + "sender_bot_alias": "openclaw_bot", + }, + }, + False, + ), + ( + { + **SENT_RECEIPT, + "_awoooi_canonical_route_receipt": { + **SENT_RECEIPT["_awoooi_canonical_route_receipt"], + "destination_alias": "other_destination", + }, + }, + False, + ), + ( + { + **SENT_RECEIPT, + "_awoooi_canonical_route_receipt": { + **SENT_RECEIPT["_awoooi_canonical_route_receipt"], + "destination_binding": "", + }, + }, + False, + ), + ( + { + **SENT_RECEIPT, + "result": {"message_id": 42, "chat": {"id": -2002}}, + }, + False, + ), + ], +) +def test_delivery_helper_fails_closed_without_complete_canonical_receipt( + receipt: object, + expected: bool, +) -> None: + assert _telegram_send_delivery_succeeded(receipt) is expected @pytest.mark.asyncio @@ -244,7 +343,8 @@ async def test_html_callback_no_send_exhausts_fallback_and_records_failure( assert send_request.await_count == 3 record_failure.assert_awaited_once() statuses = [ - call.args[1].get(gateway_module._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY, {}) + call.args[1] + .get(gateway_module._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY, {}) .get("callback_reply", {}) .get("status") for call in send_request.await_args_list @@ -298,6 +398,7 @@ async def test_interactive_reply_no_send_logs_failure_not_sent( failed_event = "ai_advisory_group_reply_failed" sent_event = "ai_advisory_group_reply_sent" else: + class Repo: async def get_by_id(self, _incident_id: str): return None @@ -451,9 +552,7 @@ async def test_dev_test_push_returns_false_for_structured_no_send( monkeypatch.setattr(telegram.settings, "ENVIRONMENT", "dev") monkeypatch.setattr(telegram, "get_telegram_gateway", lambda: Gateway()) - response = await telegram.test_push( - telegram.TestPushRequest(approval_id="APR-1") - ) + response = await telegram.test_push(telegram.TestPushRequest(approval_id="APR-1")) assert response["ok"] is False assert response["message"] == "Test push no-send" diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py index e57038bb2..51b0570a3 100644 --- a/apps/api/tests/test_telegram_message_templates.py +++ b/apps/api/tests/test_telegram_message_templates.py @@ -26,6 +26,35 @@ from src.services.telegram_gateway import ( normalize_telegram_send_message_payload, ) +_SENT_CHAT_ID = -1001 +_SENT_DESTINATION_BINDING = ( + telegram_gateway_module._telegram_destination_binding(_SENT_CHAT_ID) +) +_CANONICAL_SENT_RECEIPT = { + "ok": True, + "_awooop_delivery_status": "sent", + "_awooop_provider_send_performed": True, + "result": {"message_id": 77, "chat": {"id": _SENT_CHAT_ID}}, + "_awoooi_canonical_route_receipt": { + "schema_version": "telegram_canonical_egress_receipt_v1", + "decision": "allowed", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "sender_bot_alias": "tsenyang_bot", + "provider_send_performed": True, + }, + telegram_gateway_module._DELIVERY_CONTEXT_KEY: { + "schema_version": "telegram_delivery_context_v1", + "method": "sendMessage", + "sender_bot_alias": "tsenyang_bot", + "destination_alias": "awoooi_sre_war_room", + "destination_binding": _SENT_DESTINATION_BINDING, + "payload_destination_binding": _SENT_DESTINATION_BINDING, + "provider_destination_binding": _SENT_DESTINATION_BINDING, + "destination_binding_verified": True, + }, +} + def test_auto_repair_status_line_distinguishes_ai_retry_queued() -> None: """自動化失敗 reply 必須明確標示 AI 續跑,且不把 raw error 當純文字噴出。""" @@ -289,7 +318,7 @@ async def test_send_alert_notification_normalizes_host_resource_raw_dump(monkeyp async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -329,7 +358,7 @@ async def test_send_alert_notification_normalizes_aiops_signal_alert(monkeypatch async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -384,7 +413,7 @@ async def test_send_alert_notification_forces_html_card_for_markdown_host_alert( async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -417,7 +446,7 @@ async def test_send_text_normalizes_host_resource_alert(monkeypatch) -> None: async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -482,7 +511,10 @@ async def test_send_request_mirrors_direct_ai_alert_card_to_agent99(monkeypatch) return None def json(self) -> dict: - return {"ok": True, "result": {"message_id": 991}} + return { + "ok": True, + "result": {"message_id": 991, "chat": {"id": "chat"}}, + } class FakeHttpClient: async def post(self, url, json): # type: ignore[no-untyped-def] @@ -541,7 +573,10 @@ async def test_send_request_mirrors_direct_ai_alert_card_to_agent99(monkeypatch) await asyncio.sleep(0) assert result["ok"] is True - assert result["result"] == {"message_id": 991} + assert result["result"] == { + "message_id": 991, + "chat": {"id": "chat"}, + } assert result["_awoooi_canonical_route_receipt"]["decision"] == "allowed" assert mirrored assert mirrored[0]["source"] == "send_request" @@ -1789,7 +1824,10 @@ async def test_send_request_strips_awooop_callback_metadata_before_telegram_api( return None def json(self): - return {"ok": True, "result": {"message_id": 456}} + return { + "ok": True, + "result": {"message_id": 456, "chat": {"id": "chat"}}, + } class FakeClient: async def post(self, url, json): @@ -1878,7 +1916,7 @@ async def test_send_html_line_message_falls_back_to_plain_text_on_parse_error(mo sent_requests.append((method, payload)) if payload.get("parse_mode") == "HTML": raise telegram_gateway_module.TelegramGatewayError("HTTP error: 400") - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -1910,7 +1948,7 @@ async def test_send_html_line_message_marks_callback_reply_evidence(monkeypatch) sent_requests.append((method, payload)) if payload.get("parse_mode") == "HTML": raise telegram_gateway_module.TelegramGatewayError("HTTP error: 400") - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -2002,7 +2040,13 @@ async def test_send_html_line_message_marks_callback_reply_evidence(monkeypatch) acknowledged = ( telegram_gateway_module._acknowledge_callback_reply_source_envelope( fallback_source_extra, - delivery_result={"ok": True}, + delivery_result={ + **_CANONICAL_SENT_RECEIPT, + "result": { + "message_id": 88, + "chat": {"id": _SENT_CHAT_ID}, + }, + }, provider_message_id="88", ) ) @@ -2042,7 +2086,7 @@ async def test_send_html_line_message_uses_rescue_when_markup_fallback_fails(mon sent_requests.append((method, payload)) if len(sent_requests) < 3: raise telegram_gateway_module.TelegramGatewayError("HTTP error: 400") - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -2070,7 +2114,7 @@ async def test_send_html_line_message_attaches_awooop_markup_to_first_chunk(monk async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -2160,7 +2204,7 @@ async def test_send_notification_long_html_uses_plain_text_without_cutting_tags( async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(gateway, "_send_request", fake_send_request) long_text = "📊 事件歷史統計\n告警鍵: " + ("node<188>&" * 90) + "" @@ -2446,7 +2490,7 @@ async def test_append_incident_update_deduplicates_same_status(monkeypatch): async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: fake_redis) monkeypatch.setattr(gateway, "_send_request", fake_send_request) @@ -2491,7 +2535,7 @@ async def test_append_incident_update_suppresses_duplicate_failure_across_incide async def fake_send_request(method, payload): sent_requests.append((method, payload)) - return {"ok": True} + return dict(_CANONICAL_SENT_RECEIPT) monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: fake_redis) monkeypatch.setattr(gateway, "_send_request", fake_send_request) diff --git a/apps/api/tests/test_telegram_notification_egress_scanners.py b/apps/api/tests/test_telegram_notification_egress_scanners.py new file mode 100644 index 000000000..3959d99d5 --- /dev/null +++ b/apps/api/tests/test_telegram_notification_egress_scanners.py @@ -0,0 +1,495 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +from types import ModuleType + +import pytest + +ROOT = Path(__file__).resolve().parents[3] + + +def _load_scanner(filename: str, module_name: str) -> ModuleType: + path = ROOT / "scripts/security" / filename + spec = importlib.util.spec_from_file_location(module_name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +INVENTORY = _load_scanner( + "telegram-notification-egress-inventory.py", + "telegram_notification_egress_inventory_test", +) +NO_NEW_BYPASS_GUARD = _load_scanner( + "telegram-notification-egress-no-new-bypass-guard.py", + "telegram_notification_egress_no_new_bypass_guard_test", +) +SCANNERS = (INVENTORY, NO_NEW_BYPASS_GUARD) + + +def _write_zero_baseline(root: Path) -> None: + snapshot = ( + root / "docs/security/telegram-notification-egress-inventory.snapshot.json" + ) + snapshot.parent.mkdir(parents=True, exist_ok=True) + snapshot.write_text( + json.dumps( + { + "summary": { + "direct_bot_api_call_count": 0, + "direct_bot_api_file_count": 0, + }, + "direct_bot_api_calls": [], + } + ), + encoding="utf-8", + ) + + +def _write_minimal_gateway(root: Path) -> None: + gateway = root / "apps/api/src/services/telegram_gateway.py" + gateway.parent.mkdir(parents=True, exist_ok=True) + gateway.write_text( + "normalize_telegram_send_message_payload = object()\n", + encoding="utf-8", + ) + + +@pytest.mark.parametrize("scanner", SCANNERS) +def test_scanner_detects_multiline_split_bot_api_url(scanner: ModuleType) -> None: + source = """ +import requests + +def send_telegram_alert(token: str) -> object: + base = "https://api.telegram.org/" + endpoint = ( + base + + f"bot{token}/" + + "sendMessage" + ) + return requests.post(endpoint, json={"text": "bounded"}) +""" + + findings = scanner.scan_direct_bot_api_surfaces("scripts/ops/split.py", source) + + assert len(findings) == 1 + assert findings[0]["method"] == "sendMessage" + assert findings[0]["detection_kind"] == "direct_bot_api_endpoint" + + +@pytest.mark.parametrize("scanner", SCANNERS) +@pytest.mark.parametrize("transport", ["Invoke-WebRequest", "Invoke-RestMethod"]) +def test_scanner_detects_powershell_direct_transport( + scanner: ModuleType, + transport: str, +) -> None: + source = f""" +function Send-AgentTelegramLegacy {{ + $base = "https://api.telegram.org/" + $uri = $base + "bot" + $Token + "/sendPhoto" + {transport} -Method Post -Uri $uri +}} +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "agent99-control-plane.ps1", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "sendPhoto" + assert findings[0]["detection_kind"] == "direct_bot_api_endpoint" + + +@pytest.mark.parametrize("scanner", SCANNERS) +@pytest.mark.parametrize( + "transport", + [ + "requests.post(target, json=payload)", + "urllib.request.urlopen(target, data=payload)", + "client.PostAsync(target, content)", + ], +) +def test_scanner_detects_custom_sender_without_literal_endpoint( + scanner: ModuleType, + transport: str, +) -> None: + source = f""" +function Send-AgentTelegramLegacy {{ + $target = $ConfiguredProviderUrl + "/sendMessage" + {transport} +}} +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "agent99-control-plane.ps1", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "sendMessage" + assert findings[0]["detection_kind"] == "custom_direct_sender" + + +@pytest.mark.parametrize("scanner", SCANNERS) +@pytest.mark.parametrize( + "transport", + [ + "httpx.Client().post(endpoint, json=payload)", + "requests.Session().post(endpoint, json=payload)", + "urllib.request.build_opener().open(endpoint, data=payload)", + "opener.open(endpoint, data=payload)", + ], +) +def test_scanner_detects_factory_clients_and_urllib_openers_with_dynamic_endpoint( + scanner: ModuleType, + transport: str, +) -> None: + source = f""" +def dispatch(endpoint: str, bot_token: str, chat_id: str, payload: bytes) -> object: + return {transport} +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "scripts/ops/provider_dispatch.py", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "dynamic" + assert findings[0]["detection_kind"] == "custom_direct_sender" + assert findings[0]["function"] == "dispatch" + + +@pytest.mark.parametrize("scanner", SCANNERS) +@pytest.mark.parametrize( + ("imports", "transport"), + [ + ("import requests as rq", "rq.post(endpoint, json=payload)"), + ("import httpx as hx", "hx.post(endpoint, json=payload)"), + ( + "import httpx as hx", + "hx.Client().post(endpoint, json=payload)", + ), + ( + "from httpx import Client as HC", + "HC().post(endpoint, json=payload)", + ), + ( + "from requests import Session as S", + "S().post(endpoint, json=payload)", + ), + ( + "from urllib.request import build_opener as make_opener", + "make_opener().open(endpoint, data=payload)", + ), + ], +) +def test_scanner_resolves_python_import_alias_transport_calls( + scanner: ModuleType, + imports: str, + transport: str, +) -> None: + source = f""" +{imports} + +def dispatch(endpoint: str, bot_token: str, chat_id: str, payload: bytes) -> object: + return {transport} +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "scripts/ops/aliased_provider_dispatch.py", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "dynamic" + assert findings[0]["detection_kind"] == "custom_direct_sender" + assert findings[0]["function"] == "dispatch" + + +@pytest.mark.parametrize("scanner", SCANNERS) +@pytest.mark.parametrize( + ("imports", "transport"), + [ + ("import requests as rq", "rq.post(endpoint, json=payload)"), + ( + "import httpx as hx", + "hx.Client().post(endpoint, json=payload)", + ), + ( + "from httpx import Client as HC", + "HC().post(endpoint, json=payload)", + ), + ( + "from requests import Session as S", + "S().post(endpoint, json=payload)", + ), + ( + "from urllib.request import build_opener as make_opener", + "make_opener().open(endpoint, data=payload)", + ), + ], +) +def test_scanner_resolves_module_scope_import_alias_transport_calls( + scanner: ModuleType, + imports: str, + transport: str, +) -> None: + source = f""" +{imports} + +bot_token = object() +chat_id = object() +endpoint = object() +payload = b"" +result = {transport} +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "scripts/ops/module_alias_provider.py", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "dynamic" + assert findings[0]["detection_kind"] == "custom_direct_sender" + assert findings[0]["function"] == "" + assert findings[0]["function_qualified"] == "" + + +@pytest.mark.parametrize("scanner", SCANNERS) +def test_scanner_resolves_imported_factory_assigned_to_local_client( + scanner: ModuleType, +) -> None: + source = """ +from httpx import Client as HC + +def dispatch(endpoint: str, bot_token: str, chat_id: str, payload: bytes) -> object: + sender = HC() + return sender.post(endpoint, content=payload) +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "scripts/ops/assigned_alias_provider.py", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "dynamic" + assert findings[0]["function"] == "dispatch" + + +@pytest.mark.parametrize("scanner", SCANNERS) +@pytest.mark.parametrize("upload_method", ["UploadString", "UploadData", "UploadFile"]) +def test_scanner_detects_powershell_webclient_uploads_with_dynamic_endpoint( + scanner: ModuleType, + upload_method: str, +) -> None: + source = f""" +function Publish-LegacyPayload {{ + param($BotToken, $ChatId, $Uri) + $webClient = New-Object System.Net.WebClient + $webClient.{upload_method}($Uri, $Payload) +}} +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "scripts/ops/legacy-provider.ps1", + source, + ) + + assert len(findings) == 1 + assert findings[0]["method"] == "dynamic" + assert findings[0]["detection_kind"] == "custom_direct_sender" + assert findings[0]["function"] == "Publish-LegacyPayload" + + +@pytest.mark.parametrize("scanner", SCANNERS) +def test_scanner_excludes_only_canonical_gateway_final_exit( + scanner: ModuleType, +) -> None: + source = """ +class TelegramGateway: + async def _send_request(self, token: str, payload: dict) -> object: + url = f"https://api.telegram.org/bot{token}/sendMessage" + return await self._http_client.post(url, json=payload) +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "apps/api/src/services/telegram_gateway.py", + source, + ) + + assert findings == [] + + +@pytest.mark.parametrize("scanner", SCANNERS) +def test_scanner_does_not_allowlist_same_function_name_outside_gateway( + scanner: ModuleType, +) -> None: + source = """ +import requests as rq + +async def _send_request( + endpoint: str, + bot_token: str, + chat_id: str, + payload: dict, +) -> object: + return rq.post(endpoint, json=payload) +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "scripts/ops/telegram_gateway.py", + source, + ) + + assert len(findings) == 1 + assert findings[0]["function"] == "_send_request" + assert findings[0]["detection_kind"] == "custom_direct_sender" + + +@pytest.mark.parametrize("scanner", SCANNERS) +def test_scanner_allowlists_only_class_qualified_gateway_final_exit( + scanner: ModuleType, +) -> None: + source = """ +import requests as rq + +async def _send_request( + endpoint: str, + bot_token: str, + chat_id: str, + payload: dict, +) -> object: + return rq.post(endpoint, json=payload) + +class RogueGateway: + async def _send_request( + self, + endpoint: str, + bot_token: str, + chat_id: str, + payload: dict, + ) -> object: + return rq.post(endpoint, json=payload) + +class TelegramGateway: + async def _send_request( + self, + endpoint: str, + bot_token: str, + chat_id: str, + payload: dict, + ) -> object: + return rq.post(endpoint, json=payload) +""" + + findings = scanner.scan_direct_bot_api_surfaces( + "apps/api/src/services/telegram_gateway.py", + source, + ) + + assert {item["function_qualified"] for item in findings} == { + "_send_request", + "RogueGateway._send_request", + } + + +def test_no_new_bypass_guard_fails_closed_on_split_endpoint(tmp_path: Path) -> None: + _write_zero_baseline(tmp_path) + _write_minimal_gateway(tmp_path) + sender = tmp_path / "scripts/ops/split_sender.py" + sender.parent.mkdir(parents=True, exist_ok=True) + sender.write_text( + """ +import requests + +def send_telegram(token: str) -> object: + base = "https://api.telegram.org/" + endpoint = base + f"bot{token}/" + "sendDocument" + return requests.post(endpoint, files={}) +""", + encoding="utf-8", + ) + + report = NO_NEW_BYPASS_GUARD.build_report(tmp_path) + + assert report["summary"]["current_direct_bot_api_call_count"] == 1 + assert report["summary"]["new_bypass_count"] == 1 + with pytest.raises(SystemExit, match="direct/custom bypass"): + NO_NEW_BYPASS_GUARD.validate(tmp_path) + + +def test_local_github_workflow_is_frozen_legacy_not_active_bypass( + tmp_path: Path, +) -> None: + _write_zero_baseline(tmp_path) + _write_minimal_gateway(tmp_path) + workflow = tmp_path / ".github/workflows/legacy-alert.yml" + workflow.parent.mkdir(parents=True, exist_ok=True) + workflow.write_text( + """ +steps: + - name: frozen legacy telegram sender + run: | + endpoint="https://api.telegram.org/bot${BOT_TOKEN}/sendMessage" + curl -X POST "$endpoint" -d "chat_id=${CHAT_ID}" +""", + encoding="utf-8", + ) + + inventory_report = INVENTORY.build_report(tmp_path) + inventory_item = inventory_report["direct_bot_api_calls"][0] + assert inventory_report["summary"]["direct_bot_api_call_count"] == 1 + assert inventory_report["summary"]["active_direct_bot_api_call_count"] == 0 + assert ( + inventory_report["summary"]["github_frozen_legacy_direct_bot_api_call_count"] + == 1 + ) + assert inventory_item["source_truth_classification"] == ( + "frozen_legacy_source_truth" + ) + assert inventory_item["runtime_execution_authorized"] is False + assert inventory_item["workflow_execution_authorized"] is False + assert ( + inventory_report["execution_boundaries"]["github_workflow_execution_authorized"] + is False + ) + + guard_report = NO_NEW_BYPASS_GUARD.build_report(tmp_path) + assert guard_report["status"] == "pass_no_direct_or_custom_bypass" + assert guard_report["summary"]["detected_direct_bot_api_call_count"] == 1 + assert guard_report["summary"]["current_direct_bot_api_call_count"] == 0 + assert guard_report["summary"]["new_bypass_count"] == 0 + assert ( + guard_report["summary"]["github_frozen_legacy_direct_bot_api_call_count"] == 1 + ) + frozen_item = guard_report["frozen_legacy_direct_bot_api_calls"][0] + assert frozen_item["source_truth_classification"] == ("frozen_legacy_source_truth") + assert frozen_item["runtime_execution_authorized"] is False + assert frozen_item["workflow_execution_authorized"] is False + assert ( + guard_report["execution_boundaries"]["github_workflow_execution_authorized"] + is False + ) + NO_NEW_BYPASS_GUARD.validate(tmp_path) + + +def test_agent99_has_no_direct_or_custom_telegram_sender_source() -> None: + source = (ROOT / "agent99-control-plane.ps1").read_text(encoding="utf-8") + + for scanner in SCANNERS: + assert ( + scanner.scan_direct_bot_api_surfaces("agent99-control-plane.ps1", source) + == [] + ) + assert "function Send-AgentTelegramRelay" not in source + assert "function Send-AgentTelegramPhotoDirect" not in source + assert "canonical_telegram_gateway_transport_required" in source + assert "providerSendPerformed = $false" in source + assert 'routeStatus = "blocked_no_egress"' in source diff --git a/scripts/security/telegram-alert-readability-guard.py b/scripts/security/telegram-alert-readability-guard.py index 34b4346e2..9d4041df2 100644 --- a/scripts/security/telegram-alert-readability-guard.py +++ b/scripts/security/telegram-alert-readability-guard.py @@ -8,6 +8,7 @@ Bot API、不讀 secret、不連線主機,也不啟動任何 runtime gate。 from __future__ import annotations import argparse +import ast import json import subprocess from datetime import datetime, timedelta, timezone @@ -164,11 +165,44 @@ def require_contains(label: str, text: str, marker: str) -> None: raise SystemExit(f"BLOCKED {label}: missing {marker!r}") -def function_segment(text: str, marker: str, *, limit: int = 2200) -> str: - start = text.find(marker) - if start == -1: - raise SystemExit(f"BLOCKED source function marker missing: {marker!r}") - return text[start : start + limit] +def function_segment(text: str, marker: str) -> str: + """Return the complete named function using Python AST source boundaries.""" + marker_prefix, separator, function_name = marker.partition("def ") + function_name = function_name.strip() + if not separator or not function_name.isidentifier(): + raise SystemExit(f"BLOCKED invalid source function marker: {marker!r}") + + expected_type: type[ast.FunctionDef] | type[ast.AsyncFunctionDef] + expected_type = ( + ast.AsyncFunctionDef + if marker_prefix.strip() == "async" + else ast.FunctionDef + ) + try: + tree = ast.parse(text) + except SyntaxError as exc: + raise SystemExit( + f"BLOCKED telegram gateway source is not valid Python: {exc.msg}" + ) from exc + + matches = [ + node + for node in ast.walk(tree) + if isinstance(node, expected_type) and node.name == function_name + ] + if len(matches) != 1: + raise SystemExit( + "BLOCKED source function marker must resolve exactly once: " + f"{marker!r}, matches={len(matches)}" + ) + + node = matches[0] + if node.end_lineno is None: + raise SystemExit( + f"BLOCKED source function boundary unavailable: {marker!r}" + ) + lines = text.splitlines(keepends=True) + return "".join(lines[node.lineno - 1 : node.end_lineno]) def git_commit(root: Path) -> str: diff --git a/scripts/security/telegram-notification-egress-inventory.py b/scripts/security/telegram-notification-egress-inventory.py index ec392c273..14e2a157c 100644 --- a/scripts/security/telegram-notification-egress-inventory.py +++ b/scripts/security/telegram-notification-egress-inventory.py @@ -1,14 +1,16 @@ #!/usr/bin/env python3 """Build a repo-only Telegram notification egress inventory. -This scanner identifies Telegram Bot API sendMessage paths that can bypass -TelegramGateway's final-exit formatter. It does not read secrets, call -Telegram, modify workflows, or send notifications. +This scanner identifies guarded Telegram Bot API paths that can bypass +TelegramGateway's final-exit formatter. Local ``.github/workflows`` files are +audited only as frozen legacy source truth. It does not read secrets, call +Telegram, modify or execute workflows, or send notifications. """ from __future__ import annotations import argparse +import ast import hashlib import json import re @@ -24,6 +26,7 @@ TAIPEI = timezone(timedelta(hours=8)) SCAN_ROOTS = ( Path("agent99-control-plane.ps1"), Path(".gitea/workflows"), + Path(".github/workflows"), Path("scripts/ops"), Path("scripts/ci"), Path("scripts/reboot-recovery"), @@ -42,17 +45,81 @@ GUARDED_BOT_METHODS = ( "sendAudio", "sendVoice", ) -DIRECT_BOT_API_RE = re.compile( +BOT_TOKEN_URL_RE = re.compile( r"api\.telegram\.org/bot.*?/(?P" + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + r")\b", re.IGNORECASE, ) +GUARDED_BOT_METHOD_RE = re.compile( + r"\b(?P" + + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + + r")\b", + re.IGNORECASE, +) +COMPACT_BOT_ENDPOINT_RE = re.compile( + r"api\.telegram\.org.{0,1024}?bot.{0,512}?/(?P" + + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + + r")", + re.IGNORECASE, +) +DIRECT_HTTP_TRANSPORT_RE = re.compile( + r"(?:" + r"requests\s*\.\s*Session\s*\([^)]*\)\s*\.\s*(?:post|request|send)\s*\(|" + r"requests\s*\.\s*(?:post|request)\s*\(|" + r"httpx\s*\.\s*(?:AsyncClient|Client)\s*\([^)]*\)\s*\.\s*(?:post|request|send)\s*\(|" + r"httpx\s*\.\s*(?:post|request)\s*\(|" + r"(?:urllib\s*\.\s*request\s*\.\s*)?build_opener\s*\([^)]*\)\s*\.\s*open\s*\(|" + r"urllib\s*\.\s*request\s*\.\s*(?:urlopen|Request)\s*\(|" + r"\burlopen\s*\(|" + r"\b(?:_?opener)\s*\.\s*open\s*\(|" + r"\bInvoke-(?:WebRequest|RestMethod)\b|" + r"\.\s*PostAsync\s*\(|" + r"\.\s*Upload(?:String|Data|File)\s*\(|" + r"\b(?:fetch|curl|wget)\b|" + r"\b(?:_http_client|client|session)\s*\.\s*(?:post|request|send)\s*\(" + r")", + re.IGNORECASE, +) +TELEGRAM_CONTEXT_RE = re.compile( + r"(?:telegram|bot[_-]?token|chat[_-]?id)", re.IGNORECASE +) +BOT_TOKEN_CONTEXT_RE = re.compile( + r"(?:\bbot[_-]?token\b|\btelegram[_-]?(?:bot[_-]?)?token\b|\$Token\b)", + re.IGNORECASE, +) +CHAT_CONTEXT_RE = re.compile( + r"(?:\bchat[_-]?id\b|\$ChatId\b)", + re.IGNORECASE, +) +POWERSHELL_FUNCTION_RE = re.compile( + r"^function\s+(?P[A-Za-z0-9_-]+)\s*\{", re.IGNORECASE | re.MULTILINE +) +CANONICAL_FINAL_EXITS = { + ( + "apps/api/src/services/telegram_gateway.py", + "TelegramGateway._send_request", + ), +} +_AST_IMPORT_NODE_TYPES = (ast.Import, ast.ImportFrom) +_AST_FUNCTION_NODE_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef) +_AST_SEQUENCE_TARGET_NODE_TYPES = (ast.Tuple, ast.List) +_AST_NON_MODULE_SCOPE_NODE_TYPES = ( + ast.ClassDef, + ast.FunctionDef, + ast.AsyncFunctionDef, + ast.Lambda, +) GATEWAY_CALLSITE_RE = re.compile( r"(?:send_alert_notification\(|\b(?:tg|gw|gateway|telegram)\.send_text\(|_send_request\(\s*[\"']sendMessage[\"'])" ) SECRET_INTERPOLATION_RE = re.compile(r"\$\{\{\s*secrets\.[^}]+\}\}") -BOT_TOKEN_URL_RE = DIRECT_BOT_API_RE +BOT_TOKEN_FRAGMENT_RE = re.compile( + r"bot[^/\s\"']+/(?P" + + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + + r")\b", + re.IGNORECASE, +) REQUIRED_OWNER_FIELDS = [ "egress_surface_id", @@ -167,10 +234,429 @@ def sanitize_excerpt(line: str) -> str: lambda match: f"api.telegram.org/bot/{match.group('method')}", excerpt, ) + excerpt = BOT_TOKEN_FRAGMENT_RE.sub( + lambda match: f"bot/{match.group('method')}", + excerpt, + ) return excerpt[:180] +def _compact_source(source: str) -> str: + without_string_prefixes = re.sub( + r"(?i)\b(?:fr|rf|f|r|u|b)(?=[\"'])", + "", + source, + ) + return re.sub(r"[\s\"'`+()]+", "", without_string_prefixes) + + +_PYTHON_HTTP_TRANSPORT_CALL_RE = re.compile( + r"^(?:" + r"requests\.(?:post|request)|" + r"requests\.Session\(\)\.(?:post|request|send)|" + r"httpx\.(?:post|request)|" + r"httpx\.(?:AsyncClient|Client)\(\)\.(?:post|request|send)|" + r"urllib\.request\.(?:urlopen|Request)|" + r"urllib\.request\.build_opener\(\)\.open" + r")$" +) + + +def _record_python_import_alias( + node: ast.AST, + aliases: dict[str, str], +) -> None: + if isinstance(node, ast.Import): + for imported in node.names: + bound_name = imported.asname or imported.name.split(".", 1)[0] + aliases[bound_name] = ( + imported.name if imported.asname else bound_name + ) + return + + module = str(node.module or "").strip(".") + if not module: + return + for imported in node.names: + if imported.name == "*": + continue + aliases[imported.asname or imported.name] = ( + f"{module}.{imported.name}" + ) + + +def _resolve_python_expression( + node: ast.AST, + aliases: dict[str, str], +) -> str | None: + if isinstance(node, ast.Name): + return aliases.get(node.id, node.id) + if isinstance(node, ast.Attribute): + owner = _resolve_python_expression(node.value, aliases) + return f"{owner}.{node.attr}" if owner else None + if isinstance(node, ast.Call): + callable_name = _resolve_python_expression(node.func, aliases) + return f"{callable_name}()" if callable_name else None + return None + + +def _assignment_target_names(node: ast.AST) -> list[str]: + if isinstance(node, ast.Name): + return [node.id] + if isinstance(node, _AST_SEQUENCE_TARGET_NODE_TYPES): + return [ + name + for child in node.elts + for name in _assignment_target_names(child) + ] + return [] + + +def _python_parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]: + return { + child: parent + for parent in ast.walk(tree) + for child in ast.iter_child_nodes(parent) + } + + +def _python_function_qualified_name( + function: ast.FunctionDef | ast.AsyncFunctionDef, + parents: dict[ast.AST, ast.AST], +) -> str: + parts = [function.name] + parent = parents.get(function) + while parent is not None: + if isinstance(parent, (ast.ClassDef, *_AST_FUNCTION_NODE_TYPES)): + parts.append(parent.name) + parent = parents.get(parent) + return ".".join(reversed(parts)) + + +def _nearest_python_function( + node: ast.AST, + parents: dict[ast.AST, ast.AST], +) -> ast.FunctionDef | ast.AsyncFunctionDef | None: + parent = parents.get(node) + while parent is not None: + if isinstance(parent, _AST_FUNCTION_NODE_TYPES): + return parent + parent = parents.get(parent) + return None + + +def _is_python_module_scope( + node: ast.AST, + parents: dict[ast.AST, ast.AST], +) -> bool: + parent = parents.get(node) + while parent is not None: + if isinstance(parent, _AST_NON_MODULE_SCOPE_NODE_TYPES): + return False + parent = parents.get(parent) + return True + + +def _resolve_python_transport_calls( + nodes: list[ast.AST], + aliases: dict[str, str], +) -> list[tuple[int, str]]: + resolved_calls: list[tuple[int, str]] = [] + for node in sorted( + nodes, + key=lambda item: ( + int(getattr(item, "lineno", 0)), + int(getattr(item, "col_offset", 0)), + 0 if isinstance(item, _AST_IMPORT_NODE_TYPES) else 1, + ), + ): + if isinstance(node, _AST_IMPORT_NODE_TYPES): + _record_python_import_alias(node, aliases) + continue + if isinstance(node, ast.Assign): + resolved_value = _resolve_python_expression(node.value, aliases) + for target in node.targets: + for name in _assignment_target_names(target): + if resolved_value: + aliases[name] = resolved_value + else: + aliases.pop(name, None) + continue + if isinstance(node, ast.AnnAssign): + resolved_value = ( + _resolve_python_expression(node.value, aliases) + if node.value is not None + else None + ) + for name in _assignment_target_names(node.target): + if resolved_value: + aliases[name] = resolved_value + else: + aliases.pop(name, None) + continue + if not isinstance(node, ast.Call): + continue + qualified_call = _resolve_python_expression(node.func, aliases) + if qualified_call and _PYTHON_HTTP_TRANSPORT_CALL_RE.fullmatch( + qualified_call + ): + resolved_calls.append((node.lineno, qualified_call)) + return sorted(set(resolved_calls)) + + +def _python_transport_calls_by_scope( + text: str, +) -> tuple[ + dict[tuple[str, int, int], list[tuple[int, str]]], + list[tuple[int, str]], +]: + """Resolve HTTP aliases in function and true module-level AST scopes.""" + try: + tree = ast.parse(text) + except SyntaxError: + return {}, [] + + parents = _python_parent_map(tree) + module_aliases: dict[str, str] = {} + module_nodes = [ + node + for node in ast.walk(tree) + if _is_python_module_scope(node, parents) + ] + module_calls = _resolve_python_transport_calls( + module_nodes, + module_aliases, + ) + + transports: dict[tuple[str, int, int], list[tuple[int, str]]] = {} + function_nodes = [ + node + for node in ast.walk(tree) + if isinstance(node, _AST_FUNCTION_NODE_TYPES) + and node.end_lineno is not None + ] + for function in function_nodes: + aliases = dict(module_aliases) + arguments = [ + *function.args.posonlyargs, + *function.args.args, + *function.args.kwonlyargs, + ] + if function.args.vararg is not None: + arguments.append(function.args.vararg) + if function.args.kwarg is not None: + arguments.append(function.args.kwarg) + for argument in arguments: + aliases.pop(argument.arg, None) + + scoped_nodes = [ + node + for node in ast.walk(function) + if node is function + or _nearest_python_function(node, parents) is function + ] + resolved_calls = _resolve_python_transport_calls(scoped_nodes, aliases) + + if resolved_calls: + qualified_name = _python_function_qualified_name(function, parents) + transports[ + ( + qualified_name, + function.lineno, + int(function.end_lineno), + ) + ] = resolved_calls + return transports, module_calls + + +def _transport_window_units( + text: str, + *, + excluded_line_ranges: list[tuple[int, int]], +) -> list[tuple[str, str, int, int, str]]: + lines = text.splitlines() + units: list[tuple[str, str, int, int, str]] = [] + for match in DIRECT_HTTP_TRANSPORT_RE.finditer(text): + line_number = text.count("\n", 0, match.start()) + 1 + if any(start <= line_number <= end for start, end in excluded_line_ranges): + continue + start_line = max(1, line_number - 20) + end_line = min(len(lines), line_number + 20) + units.append( + ( + "", + "", + start_line, + end_line, + "\n".join(lines[start_line - 1 : end_line]), + ) + ) + return units + + +def _source_units( + relative_path: str, + text: str, +) -> list[tuple[str, str, int, int, str]]: + lines = text.splitlines() + units: list[tuple[str, str, int, int, str]] = [] + ranges: list[tuple[int, int]] = [] + if relative_path.endswith(".py"): + try: + tree = ast.parse(text) + except SyntaxError: + tree = None + if tree is not None: + parents = _python_parent_map(tree) + nodes = [ + node + for node in ast.walk(tree) + if isinstance(node, _AST_FUNCTION_NODE_TYPES) + and getattr(node, "end_lineno", None) + ] + for node in sorted( + nodes, key=lambda item: (item.lineno, item.end_lineno or item.lineno) + ): + end_line = int(node.end_lineno or node.lineno) + ranges.append((node.lineno, end_line)) + units.append( + ( + node.name, + _python_function_qualified_name(node, parents), + node.lineno, + end_line, + "\n".join(lines[node.lineno - 1 : end_line]), + ) + ) + elif relative_path.endswith(".ps1"): + matches = list(POWERSHELL_FUNCTION_RE.finditer(text)) + for index, match in enumerate(matches): + start_line = text.count("\n", 0, match.start()) + 1 + end_offset = ( + matches[index + 1].start() if index + 1 < len(matches) else len(text) + ) + end_line = text.count("\n", 0, end_offset) + 1 + ranges.append((start_line, end_line)) + units.append( + ( + match.group("name"), + match.group("name"), + start_line, + end_line, + text[match.start() : end_offset], + ) + ) + return units + _transport_window_units(text, excluded_line_ranges=ranges) + + +def scan_direct_bot_api_surfaces( + relative_path: str, + text: str, +) -> list[dict[str, Any]]: + """Find direct Telegram transports, including split URLs and custom senders.""" + + source_lines = text.splitlines() + findings: list[dict[str, Any]] = [] + seen: set[tuple[int, str, str]] = set() + python_transports, module_transports = ( + _python_transport_calls_by_scope(text) + if relative_path.endswith(".py") + else ({}, []) + ) + source_units = _source_units(relative_path, text) + for line_number, _qualified_call in module_transports: + start_line = max(1, line_number - 20) + end_line = min(len(source_lines), line_number + 20) + source_units.append( + ( + "", + "", + start_line, + end_line, + "\n".join(source_lines[start_line - 1 : end_line]), + ) + ) + + for ( + function_name, + function_qualified, + start_line, + _end_line, + source, + ) in source_units: + if (relative_path, function_qualified) in CANONICAL_FINAL_EXITS: + continue + resolved_python_transports = ( + [ + item + for item in module_transports + if start_line <= item[0] <= _end_line + ] + if function_qualified == "" + else python_transports.get( + (function_qualified, start_line, _end_line), + [], + ) + ) + transport_match = DIRECT_HTTP_TRANSPORT_RE.search(source) + if transport_match is None and not resolved_python_transports: + continue + method_match = GUARDED_BOT_METHOD_RE.search(source) + endpoint_match = COMPACT_BOT_ENDPOINT_RE.search(_compact_source(source)) + telegram_named = bool( + re.search(r"(?:telegram|bot)", function_name, re.IGNORECASE) + ) + telegram_context = bool(TELEGRAM_CONTEXT_RE.search(source)) + token_and_chat_context = bool( + BOT_TOKEN_CONTEXT_RE.search(source) and CHAT_CONTEXT_RE.search(source) + ) + if ( + endpoint_match is None + and method_match is None + and not ((telegram_named and telegram_context) or token_and_chat_context) + ): + continue + + method = ( + endpoint_match.group("method") + if endpoint_match is not None + else method_match.group("method") + if method_match is not None + else "dynamic" + ) + line_number = ( + resolved_python_transports[0][0] + if resolved_python_transports + else start_line + source[: transport_match.start()].count("\n") + ) + detection_kind = ( + "direct_bot_api_endpoint" + if endpoint_match is not None + else "custom_direct_sender" + ) + key = (line_number, method.lower(), detection_kind) + if key in seen: + continue + seen.add(key) + source_line = ( + source_lines[line_number - 1] if line_number <= len(source_lines) else "" + ) + findings.append( + { + "line": line_number, + "method": method, + "detection_kind": detection_kind, + "function": function_name, + "function_qualified": function_qualified, + "sanitized_excerpt": sanitize_excerpt(source_line), + } + ) + return findings + + def surface_kind(relative_path: str) -> str: + if relative_path.startswith(".github/workflows/"): + return "github_frozen_legacy_workflow_direct_bot_api" if relative_path.startswith(".gitea/workflows/"): return "gitea_workflow_direct_bot_api" if relative_path.startswith("scripts/ops/"): @@ -186,6 +672,12 @@ def surface_kind(relative_path: str) -> str: return "other_direct_bot_api" +def source_truth_classification(relative_path: str) -> str: + if relative_path.startswith(".github/workflows/"): + return "frozen_legacy_source_truth" + return "active_repo_source_truth" + + def line_hash(relative_path: str, line_number: int, line: str) -> str: payload = f"{relative_path}:{line_number}:{line.strip()}".encode("utf-8") return hashlib.sha256(payload).hexdigest()[:16] @@ -200,42 +692,50 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: for path in files: relative_path = path.relative_to(root).as_posix() text = path.read_text(encoding="utf-8", errors="replace") - for line_number, line in enumerate(text.splitlines(), start=1): - direct_match = DIRECT_BOT_API_RE.search(line) - if direct_match: - kind = surface_kind(relative_path) - direct_calls.append( - { - "egress_surface_id": f"telegram_egress:{kind}:{relative_path}:{line_number}", - "surface_kind": kind, - "path": relative_path, - "line": line_number, - "line_hash": line_hash(relative_path, line_number, line), - "method": direct_match.group("method"), - "sanitized_excerpt": sanitize_excerpt(line), - "required_owner_fields": REQUIRED_OWNER_FIELDS, - "reviewer_checks": REVIEWER_CHECKS, - "outcome_lanes": OUTCOME_LANES, - "blocked_actions": BLOCKED_ACTIONS, - "owner_response_received": False, - "owner_response_accepted": False, - "formatter_convergence_accepted": False, - "redaction_contract_accepted": False, - "delivery_receipt_accepted": False, - "direct_bot_api_migration_authorized": False, - "telegram_send_authorized": False, - "bot_api_call_authorized": False, - "workflow_modification_authorized": False, - "script_modification_authorized": False, - "secret_value_collection_allowed": False, - "raw_payload_storage_allowed": False, - "production_write_authorized": False, - "runtime_gate": False, - "action_buttons_allowed": False, - "not_authorization": True, - } - ) + text_lines = text.splitlines() + for finding in scan_direct_bot_api_surfaces(relative_path, text): + line_number = int(finding["line"]) + line = text_lines[line_number - 1] if line_number <= len(text_lines) else "" + kind = surface_kind(relative_path) + truth_classification = source_truth_classification(relative_path) + direct_calls.append( + { + "egress_surface_id": f"telegram_egress:{kind}:{relative_path}:{line_number}", + "surface_kind": kind, + "source_truth_classification": truth_classification, + "path": relative_path, + "line": line_number, + "line_hash": line_hash(relative_path, line_number, line), + "method": finding["method"], + "detection_kind": finding["detection_kind"], + "function": finding["function"], + "sanitized_excerpt": finding["sanitized_excerpt"], + "required_owner_fields": REQUIRED_OWNER_FIELDS, + "reviewer_checks": REVIEWER_CHECKS, + "outcome_lanes": OUTCOME_LANES, + "blocked_actions": BLOCKED_ACTIONS, + "owner_response_received": False, + "owner_response_accepted": False, + "formatter_convergence_accepted": False, + "redaction_contract_accepted": False, + "delivery_receipt_accepted": False, + "direct_bot_api_migration_authorized": False, + "telegram_send_authorized": False, + "bot_api_call_authorized": False, + "workflow_modification_authorized": False, + "script_modification_authorized": False, + "secret_value_collection_allowed": False, + "raw_payload_storage_allowed": False, + "production_write_authorized": False, + "runtime_execution_authorized": False, + "workflow_execution_authorized": False, + "runtime_gate": False, + "action_buttons_allowed": False, + "not_authorization": True, + } + ) + for line_number, line in enumerate(text_lines, start=1): if GATEWAY_CALLSITE_RE.search(line): gateway_calls.append( { @@ -245,14 +745,56 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: } ) + active_direct_calls = [ + item + for item in direct_calls + if item["source_truth_classification"] == "active_repo_source_truth" + ] + frozen_legacy_direct_calls = [ + item + for item in direct_calls + if item["source_truth_classification"] == "frozen_legacy_source_truth" + ] direct_files = sorted({item["path"] for item in direct_calls}) - workflow_direct_calls = [item for item in direct_calls if item["surface_kind"] == "gitea_workflow_direct_bot_api"] - ops_direct_calls = [item for item in direct_calls if item["surface_kind"] == "ops_script_direct_bot_api"] - ci_direct_calls = [item for item in direct_calls if item["surface_kind"] == "ci_script_direct_bot_api"] - api_direct_calls = [item for item in direct_calls if item["surface_kind"] == "api_direct_bot_api"] + active_direct_files = sorted({item["path"] for item in active_direct_calls}) + frozen_legacy_direct_files = sorted( + {item["path"] for item in frozen_legacy_direct_calls} + ) + workflow_direct_calls = [ + item + for item in direct_calls + if item["surface_kind"] == "gitea_workflow_direct_bot_api" + ] + ops_direct_calls = [ + item + for item in direct_calls + if item["surface_kind"] == "ops_script_direct_bot_api" + ] + ci_direct_calls = [ + item + for item in direct_calls + if item["surface_kind"] == "ci_script_direct_bot_api" + ] + api_direct_calls = [ + item for item in direct_calls if item["surface_kind"] == "api_direct_bot_api" + ] + custom_direct_senders = [ + item + for item in direct_calls + if item["detection_kind"] == "custom_direct_sender" + ] + direct_endpoints = [ + item + for item in direct_calls + if item["detection_kind"] == "direct_bot_api_endpoint" + ] telegram_gateway_path = root / "apps/api/src/services/telegram_gateway.py" - telegram_gateway_text = telegram_gateway_path.read_text(encoding="utf-8", errors="replace") - gateway_formatter_present = "normalize_telegram_send_message_payload" in telegram_gateway_text + telegram_gateway_text = telegram_gateway_path.read_text( + encoding="utf-8", errors="replace" + ) + gateway_formatter_present = ( + "normalize_telegram_send_message_payload" in telegram_gateway_text + ) return { "schema_version": "telegram_notification_egress_inventory_v1", @@ -265,12 +807,24 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: "scanned_file_count": len(files), "direct_bot_api_file_count": len(direct_files), "direct_bot_api_call_count": len(direct_calls), + "active_direct_bot_api_file_count": len(active_direct_files), + "active_direct_bot_api_call_count": len(active_direct_calls), + "github_frozen_legacy_direct_bot_api_file_count": len( + frozen_legacy_direct_files + ), + "github_frozen_legacy_direct_bot_api_call_count": len( + frozen_legacy_direct_calls + ), + "direct_bot_api_endpoint_count": len(direct_endpoints), + "custom_direct_sender_count": len(custom_direct_senders), "workflow_direct_bot_api_call_count": len(workflow_direct_calls), "ops_script_direct_bot_api_call_count": len(ops_direct_calls), "ci_script_direct_bot_api_call_count": len(ci_direct_calls), "api_direct_bot_api_call_count": len(api_direct_calls), "gateway_normalized_callsite_count": len(gateway_calls), - "gateway_final_exit_formatter_present_count": 1 if gateway_formatter_present else 0, + "gateway_final_exit_formatter_present_count": 1 + if gateway_formatter_present + else 0, "required_owner_field_count": len(REQUIRED_OWNER_FIELDS), "reviewer_check_count": len(REVIEWER_CHECKS), "outcome_lane_count": len(OUTCOME_LANES), @@ -284,6 +838,7 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: "telegram_send_authorized_count": 0, "bot_api_call_authorized_count": 0, "workflow_modification_authorized_count": 0, + "workflow_execution_authorized_count": 0, "script_modification_authorized_count": 0, "secret_value_collection_allowed_count": 0, "raw_payload_storage_allowed_count": 0, @@ -296,6 +851,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: "telegram_send_authorized": False, "bot_api_call_authorized": False, "workflow_modification_authorized": False, + "workflow_execution_authorized": False, + "github_workflow_execution_authorized": False, "script_modification_authorized": False, "secret_value_collection_allowed": False, "secret_hash_collection_allowed": False, @@ -311,7 +868,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: "direct_bot_api_calls": direct_calls, "gateway_normalized_callsite_refs": gateway_calls, "operator_interpretation": [ - "direct_bot_api_call_count 大於 0 代表仍有 workflow / ops / API 旁路可能繞過 TelegramGateway formatter。", + "active_direct_bot_api_call_count 大於 0 代表仍有 active workflow / ops / API 旁路可能繞過 TelegramGateway formatter。", + ".github/workflows 僅列為 frozen_legacy_source_truth 供盤點,runtime_execution_authorized=false、workflow_execution_authorized=false,禁止執行。", "本清冊只建立 metadata-only egress surface,不送 Telegram、不修改 workflow / script、不讀 secret value。", "後續要收斂 direct Bot API 必須另走 owner response、formatter convergence、redaction contract、delivery receipt 與維護窗口。", ], @@ -321,11 +879,15 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: def validate(root: Path) -> None: report = build_report(root) if report["summary"]["gateway_final_exit_formatter_present_count"] != 1: - raise SystemExit("BLOCKED telegram egress inventory: gateway formatter not found") + raise SystemExit( + "BLOCKED telegram egress inventory: gateway formatter not found" + ) def main() -> None: - parser = argparse.ArgumentParser(description="Build Telegram notification egress inventory") + parser = argparse.ArgumentParser( + description="Build Telegram notification egress inventory" + ) parser.add_argument("--root", default=".", help="repository root") parser.add_argument("--output", help="write JSON snapshot") parser.add_argument("--generated-at", help="fixed generated_at timestamp") @@ -342,6 +904,9 @@ def main() -> None: print( "TELEGRAM_NOTIFICATION_EGRESS_INVENTORY_OK " f"direct_calls={report['summary']['direct_bot_api_call_count']} " + f"active={report['summary']['active_direct_bot_api_call_count']} " + f"frozen_legacy={report['summary']['github_frozen_legacy_direct_bot_api_call_count']} " + f"custom={report['summary']['custom_direct_sender_count']} " f"files={report['summary']['direct_bot_api_file_count']} " f"workflow={report['summary']['workflow_direct_bot_api_call_count']} " f"ops={report['summary']['ops_script_direct_bot_api_call_count']} " diff --git a/scripts/security/telegram-notification-egress-no-new-bypass-guard.py b/scripts/security/telegram-notification-egress-no-new-bypass-guard.py index bc2ebc4bc..098914e32 100644 --- a/scripts/security/telegram-notification-egress-no-new-bypass-guard.py +++ b/scripts/security/telegram-notification-egress-no-new-bypass-guard.py @@ -2,14 +2,16 @@ """檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路。 本 guard 只掃描 repo 原始碼與 committed snapshot,不讀 secret、不呼叫 -Telegram、不修改 workflow / script / API sender。既有 direct send 仍是待 -owner response 的基線;任何新增或變形的 direct Bot API endpoint 都必須先 -進 inventory / owner request / migration plan,而不是直接合併。 +Telegram、不修改或執行 workflow / script / API sender。Committed baseline +僅供 drift 比對;目前 active source 的 direct endpoint 或 custom sender 必須 +真正為零。Local ``.github/workflows`` 只以 frozen legacy source truth 稽核, +不得以 regex 漏掃、歷史基線或 frozen source 產生 false-green。 """ from __future__ import annotations import argparse +import ast import json import re import subprocess @@ -22,10 +24,13 @@ from typing import Any TAIPEI = timezone(timedelta(hours=8)) -SOURCE_SNAPSHOT = Path("docs/security/telegram-notification-egress-inventory.snapshot.json") +SOURCE_SNAPSHOT = Path( + "docs/security/telegram-notification-egress-inventory.snapshot.json" +) SCAN_ROOTS = ( Path("agent99-control-plane.ps1"), Path(".gitea/workflows"), + Path(".github/workflows"), Path("scripts/ops"), Path("scripts/ci"), Path("scripts/reboot-recovery"), @@ -50,6 +55,65 @@ BOT_ENDPOINT_RE = re.compile( + r")\b", re.IGNORECASE, ) +GUARDED_BOT_METHOD_RE = re.compile( + r"\b(?P" + + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + + r")\b", + re.IGNORECASE, +) +COMPACT_BOT_ENDPOINT_RE = re.compile( + r"api\.telegram\.org.{0,1024}?bot.{0,512}?/(?P" + + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + + r")", + re.IGNORECASE, +) +DIRECT_HTTP_TRANSPORT_RE = re.compile( + r"(?:" + r"requests\s*\.\s*Session\s*\([^)]*\)\s*\.\s*(?:post|request|send)\s*\(|" + r"requests\s*\.\s*(?:post|request)\s*\(|" + r"httpx\s*\.\s*(?:AsyncClient|Client)\s*\([^)]*\)\s*\.\s*(?:post|request|send)\s*\(|" + r"httpx\s*\.\s*(?:post|request)\s*\(|" + r"(?:urllib\s*\.\s*request\s*\.\s*)?build_opener\s*\([^)]*\)\s*\.\s*open\s*\(|" + r"urllib\s*\.\s*request\s*\.\s*(?:urlopen|Request)\s*\(|" + r"\burlopen\s*\(|" + r"\b(?:_?opener)\s*\.\s*open\s*\(|" + r"\bInvoke-(?:WebRequest|RestMethod)\b|" + r"\.\s*PostAsync\s*\(|" + r"\.\s*Upload(?:String|Data|File)\s*\(|" + r"\b(?:fetch|curl|wget)\b|" + r"\b(?:_http_client|client|session)\s*\.\s*(?:post|request|send)\s*\(" + r")", + re.IGNORECASE, +) +TELEGRAM_CONTEXT_RE = re.compile( + r"(?:telegram|bot[_-]?token|chat[_-]?id)", re.IGNORECASE +) +BOT_TOKEN_CONTEXT_RE = re.compile( + r"(?:\bbot[_-]?token\b|\btelegram[_-]?(?:bot[_-]?)?token\b|\$Token\b)", + re.IGNORECASE, +) +CHAT_CONTEXT_RE = re.compile( + r"(?:\bchat[_-]?id\b|\$ChatId\b)", + re.IGNORECASE, +) +POWERSHELL_FUNCTION_RE = re.compile( + r"^function\s+(?P[A-Za-z0-9_-]+)\s*\{", re.IGNORECASE | re.MULTILINE +) +CANONICAL_FINAL_EXITS = { + ( + "apps/api/src/services/telegram_gateway.py", + "TelegramGateway._send_request", + ), +} +_AST_IMPORT_NODE_TYPES = (ast.Import, ast.ImportFrom) +_AST_FUNCTION_NODE_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef) +_AST_SEQUENCE_TARGET_NODE_TYPES = (ast.Tuple, ast.List) +_AST_NON_MODULE_SCOPE_NODE_TYPES = ( + ast.ClassDef, + ast.FunctionDef, + ast.AsyncFunctionDef, + ast.Lambda, +) SECRET_INTERPOLATION_RE = re.compile(r"\$\{\{\s*secrets\.[^}]+\}\}") BOT_TOKEN_URL_RE = re.compile( r"api\.telegram\.org/bot.*?/(?P" @@ -57,6 +121,12 @@ BOT_TOKEN_URL_RE = re.compile( + r")\b", re.IGNORECASE, ) +BOT_TOKEN_FRAGMENT_RE = re.compile( + r"bot[^/\s\"']+/(?P" + + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS) + + r")\b", + re.IGNORECASE, +) def git_short_sha(root: Path) -> str: @@ -96,13 +166,436 @@ def sanitize_excerpt(line: str) -> str: lambda match: f"api.telegram.org/bot/{match.group('method')}", excerpt, ) + excerpt = BOT_TOKEN_FRAGMENT_RE.sub( + lambda match: f"bot/{match.group('method')}", + excerpt, + ) return excerpt[:180] +def _compact_source(source: str) -> str: + without_string_prefixes = re.sub( + r"(?i)\b(?:fr|rf|f|r|u|b)(?=[\"'])", + "", + source, + ) + return re.sub(r"[\s\"'`+()]+", "", without_string_prefixes) + + +_PYTHON_HTTP_TRANSPORT_CALL_RE = re.compile( + r"^(?:" + r"requests\.(?:post|request)|" + r"requests\.Session\(\)\.(?:post|request|send)|" + r"httpx\.(?:post|request)|" + r"httpx\.(?:AsyncClient|Client)\(\)\.(?:post|request|send)|" + r"urllib\.request\.(?:urlopen|Request)|" + r"urllib\.request\.build_opener\(\)\.open" + r")$" +) + + +def _record_python_import_alias( + node: ast.AST, + aliases: dict[str, str], +) -> None: + if isinstance(node, ast.Import): + for imported in node.names: + bound_name = imported.asname or imported.name.split(".", 1)[0] + aliases[bound_name] = ( + imported.name if imported.asname else bound_name + ) + return + + module = str(node.module or "").strip(".") + if not module: + return + for imported in node.names: + if imported.name == "*": + continue + aliases[imported.asname or imported.name] = ( + f"{module}.{imported.name}" + ) + + +def _resolve_python_expression( + node: ast.AST, + aliases: dict[str, str], +) -> str | None: + if isinstance(node, ast.Name): + return aliases.get(node.id, node.id) + if isinstance(node, ast.Attribute): + owner = _resolve_python_expression(node.value, aliases) + return f"{owner}.{node.attr}" if owner else None + if isinstance(node, ast.Call): + callable_name = _resolve_python_expression(node.func, aliases) + return f"{callable_name}()" if callable_name else None + return None + + +def _assignment_target_names(node: ast.AST) -> list[str]: + if isinstance(node, ast.Name): + return [node.id] + if isinstance(node, _AST_SEQUENCE_TARGET_NODE_TYPES): + return [ + name + for child in node.elts + for name in _assignment_target_names(child) + ] + return [] + + +def _python_parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]: + return { + child: parent + for parent in ast.walk(tree) + for child in ast.iter_child_nodes(parent) + } + + +def _python_function_qualified_name( + function: ast.FunctionDef | ast.AsyncFunctionDef, + parents: dict[ast.AST, ast.AST], +) -> str: + parts = [function.name] + parent = parents.get(function) + while parent is not None: + if isinstance(parent, (ast.ClassDef, *_AST_FUNCTION_NODE_TYPES)): + parts.append(parent.name) + parent = parents.get(parent) + return ".".join(reversed(parts)) + + +def _nearest_python_function( + node: ast.AST, + parents: dict[ast.AST, ast.AST], +) -> ast.FunctionDef | ast.AsyncFunctionDef | None: + parent = parents.get(node) + while parent is not None: + if isinstance(parent, _AST_FUNCTION_NODE_TYPES): + return parent + parent = parents.get(parent) + return None + + +def _is_python_module_scope( + node: ast.AST, + parents: dict[ast.AST, ast.AST], +) -> bool: + parent = parents.get(node) + while parent is not None: + if isinstance(parent, _AST_NON_MODULE_SCOPE_NODE_TYPES): + return False + parent = parents.get(parent) + return True + + +def _resolve_python_transport_calls( + nodes: list[ast.AST], + aliases: dict[str, str], +) -> list[tuple[int, str]]: + resolved_calls: list[tuple[int, str]] = [] + for node in sorted( + nodes, + key=lambda item: ( + int(getattr(item, "lineno", 0)), + int(getattr(item, "col_offset", 0)), + 0 if isinstance(item, _AST_IMPORT_NODE_TYPES) else 1, + ), + ): + if isinstance(node, _AST_IMPORT_NODE_TYPES): + _record_python_import_alias(node, aliases) + continue + if isinstance(node, ast.Assign): + resolved_value = _resolve_python_expression(node.value, aliases) + for target in node.targets: + for name in _assignment_target_names(target): + if resolved_value: + aliases[name] = resolved_value + else: + aliases.pop(name, None) + continue + if isinstance(node, ast.AnnAssign): + resolved_value = ( + _resolve_python_expression(node.value, aliases) + if node.value is not None + else None + ) + for name in _assignment_target_names(node.target): + if resolved_value: + aliases[name] = resolved_value + else: + aliases.pop(name, None) + continue + if not isinstance(node, ast.Call): + continue + qualified_call = _resolve_python_expression(node.func, aliases) + if qualified_call and _PYTHON_HTTP_TRANSPORT_CALL_RE.fullmatch( + qualified_call + ): + resolved_calls.append((node.lineno, qualified_call)) + return sorted(set(resolved_calls)) + + +def _python_transport_calls_by_scope( + text: str, +) -> tuple[ + dict[tuple[str, int, int], list[tuple[int, str]]], + list[tuple[int, str]], +]: + """Resolve HTTP aliases in function and true module-level AST scopes.""" + try: + tree = ast.parse(text) + except SyntaxError: + return {}, [] + + parents = _python_parent_map(tree) + module_aliases: dict[str, str] = {} + module_nodes = [ + node + for node in ast.walk(tree) + if _is_python_module_scope(node, parents) + ] + module_calls = _resolve_python_transport_calls( + module_nodes, + module_aliases, + ) + + transports: dict[tuple[str, int, int], list[tuple[int, str]]] = {} + function_nodes = [ + node + for node in ast.walk(tree) + if isinstance(node, _AST_FUNCTION_NODE_TYPES) + and node.end_lineno is not None + ] + for function in function_nodes: + aliases = dict(module_aliases) + arguments = [ + *function.args.posonlyargs, + *function.args.args, + *function.args.kwonlyargs, + ] + if function.args.vararg is not None: + arguments.append(function.args.vararg) + if function.args.kwarg is not None: + arguments.append(function.args.kwarg) + for argument in arguments: + aliases.pop(argument.arg, None) + + scoped_nodes = [ + node + for node in ast.walk(function) + if node is function + or _nearest_python_function(node, parents) is function + ] + resolved_calls = _resolve_python_transport_calls(scoped_nodes, aliases) + + if resolved_calls: + qualified_name = _python_function_qualified_name(function, parents) + transports[ + ( + qualified_name, + function.lineno, + int(function.end_lineno), + ) + ] = resolved_calls + return transports, module_calls + + +def _transport_window_units( + text: str, + *, + excluded_line_ranges: list[tuple[int, int]], +) -> list[tuple[str, str, int, int, str]]: + lines = text.splitlines() + units: list[tuple[str, str, int, int, str]] = [] + for match in DIRECT_HTTP_TRANSPORT_RE.finditer(text): + line_number = text.count("\n", 0, match.start()) + 1 + if any(start <= line_number <= end for start, end in excluded_line_ranges): + continue + start_line = max(1, line_number - 20) + end_line = min(len(lines), line_number + 20) + units.append( + ( + "", + "", + start_line, + end_line, + "\n".join(lines[start_line - 1 : end_line]), + ) + ) + return units + + +def _source_units( + relative_path: str, + text: str, +) -> list[tuple[str, str, int, int, str]]: + lines = text.splitlines() + units: list[tuple[str, str, int, int, str]] = [] + ranges: list[tuple[int, int]] = [] + if relative_path.endswith(".py"): + try: + tree = ast.parse(text) + except SyntaxError: + tree = None + if tree is not None: + parents = _python_parent_map(tree) + nodes = [ + node + for node in ast.walk(tree) + if isinstance(node, _AST_FUNCTION_NODE_TYPES) + and getattr(node, "end_lineno", None) + ] + for node in sorted( + nodes, key=lambda item: (item.lineno, item.end_lineno or item.lineno) + ): + end_line = int(node.end_lineno or node.lineno) + ranges.append((node.lineno, end_line)) + units.append( + ( + node.name, + _python_function_qualified_name(node, parents), + node.lineno, + end_line, + "\n".join(lines[node.lineno - 1 : end_line]), + ) + ) + elif relative_path.endswith(".ps1"): + matches = list(POWERSHELL_FUNCTION_RE.finditer(text)) + for index, match in enumerate(matches): + start_line = text.count("\n", 0, match.start()) + 1 + end_offset = ( + matches[index + 1].start() if index + 1 < len(matches) else len(text) + ) + end_line = text.count("\n", 0, end_offset) + 1 + ranges.append((start_line, end_line)) + units.append( + ( + match.group("name"), + match.group("name"), + start_line, + end_line, + text[match.start() : end_offset], + ) + ) + return units + _transport_window_units(text, excluded_line_ranges=ranges) + + +def scan_direct_bot_api_surfaces( + relative_path: str, + text: str, +) -> list[dict[str, Any]]: + """Find direct Telegram transports, including split URLs and custom senders.""" + + source_lines = text.splitlines() + findings: list[dict[str, Any]] = [] + seen: set[tuple[int, str, str]] = set() + python_transports, module_transports = ( + _python_transport_calls_by_scope(text) + if relative_path.endswith(".py") + else ({}, []) + ) + source_units = _source_units(relative_path, text) + for line_number, _qualified_call in module_transports: + start_line = max(1, line_number - 20) + end_line = min(len(source_lines), line_number + 20) + source_units.append( + ( + "", + "", + start_line, + end_line, + "\n".join(source_lines[start_line - 1 : end_line]), + ) + ) + + for ( + function_name, + function_qualified, + start_line, + _end_line, + source, + ) in source_units: + if (relative_path, function_qualified) in CANONICAL_FINAL_EXITS: + continue + resolved_python_transports = ( + [ + item + for item in module_transports + if start_line <= item[0] <= _end_line + ] + if function_qualified == "" + else python_transports.get( + (function_qualified, start_line, _end_line), + [], + ) + ) + transport_match = DIRECT_HTTP_TRANSPORT_RE.search(source) + if transport_match is None and not resolved_python_transports: + continue + method_match = GUARDED_BOT_METHOD_RE.search(source) + endpoint_match = COMPACT_BOT_ENDPOINT_RE.search(_compact_source(source)) + telegram_named = bool( + re.search(r"(?:telegram|bot)", function_name, re.IGNORECASE) + ) + telegram_context = bool(TELEGRAM_CONTEXT_RE.search(source)) + token_and_chat_context = bool( + BOT_TOKEN_CONTEXT_RE.search(source) and CHAT_CONTEXT_RE.search(source) + ) + if ( + endpoint_match is None + and method_match is None + and not ((telegram_named and telegram_context) or token_and_chat_context) + ): + continue + + method = ( + endpoint_match.group("method") + if endpoint_match is not None + else method_match.group("method") + if method_match is not None + else "dynamic" + ) + line_number = ( + resolved_python_transports[0][0] + if resolved_python_transports + else start_line + source[: transport_match.start()].count("\n") + ) + detection_kind = ( + "direct_bot_api_endpoint" + if endpoint_match is not None + else "custom_direct_sender" + ) + key = (line_number, method.lower(), detection_kind) + if key in seen: + continue + seen.add(key) + source_line = ( + source_lines[line_number - 1] if line_number <= len(source_lines) else "" + ) + findings.append( + { + "line": line_number, + "method": method, + "detection_kind": detection_kind, + "function": function_name, + "function_qualified": function_qualified, + "sanitized_excerpt": sanitize_excerpt(source_line), + } + ) + return findings + + def signature(path: str, method: str, sanitized_excerpt: str) -> str: return f"{path}::{method.lower()}::{sanitized_excerpt}" +def source_truth_classification(relative_path: str) -> str: + if relative_path.startswith(".github/workflows/"): + return "frozen_legacy_source_truth" + return "active_repo_source_truth" + + def load_source_snapshot(root: Path) -> dict[str, Any]: snapshot_path = root / SOURCE_SNAPSHOT return json.loads(snapshot_path.read_text(encoding="utf-8")) @@ -111,9 +604,16 @@ def load_source_snapshot(root: Path) -> dict[str, Any]: def build_baseline(source_snapshot: dict[str, Any]) -> Counter[str]: baseline: Counter[str] = Counter() for item in source_snapshot.get("direct_bot_api_calls", []): + if ( + source_truth_classification(str(item["path"])) + == "frozen_legacy_source_truth" + ): + continue excerpt = item.get("sanitized_excerpt", "") match = BOT_ENDPOINT_RE.search(excerpt) - method = match.group("method") if match else "sendMessage" + method = str( + item.get("method") or (match.group("method") if match else "dynamic") + ) baseline[signature(item["path"], method, excerpt)] += 1 return baseline @@ -123,19 +623,25 @@ def scan_current_direct_endpoints(root: Path) -> list[dict[str, Any]]: for path in iter_scannable_files(root): relative_path = path.relative_to(root).as_posix() text = path.read_text(encoding="utf-8", errors="replace") - for line_number, line in enumerate(text.splitlines(), start=1): - for match in BOT_ENDPOINT_RE.finditer(line): - method = match.group("method") - sanitized = sanitize_excerpt(line) - findings.append( - { - "path": relative_path, - "line": line_number, - "method": method, - "sanitized_excerpt": sanitized, - "signature": signature(relative_path, method, sanitized), - } - ) + for finding in scan_direct_bot_api_surfaces(relative_path, text): + method = finding["method"] + sanitized = finding["sanitized_excerpt"] + findings.append( + { + "path": relative_path, + "line": finding["line"], + "method": method, + "detection_kind": finding["detection_kind"], + "function": finding["function"], + "sanitized_excerpt": sanitized, + "signature": signature(relative_path, method, sanitized), + "source_truth_classification": source_truth_classification( + relative_path + ), + "runtime_execution_authorized": False, + "workflow_execution_authorized": False, + } + ) return findings @@ -153,7 +659,17 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: generated = generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds") source_snapshot = load_source_snapshot(root) baseline = build_baseline(source_snapshot) - current_findings = scan_current_direct_endpoints(root) + detected_findings = scan_current_direct_endpoints(root) + current_findings = [ + item + for item in detected_findings + if item["source_truth_classification"] == "active_repo_source_truth" + ] + frozen_legacy_findings = [ + item + for item in detected_findings + if item["source_truth_classification"] == "frozen_legacy_source_truth" + ] remaining_baseline = baseline.copy() new_bypass_findings: list[dict[str, Any]] = [] @@ -172,23 +688,48 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: current_files = sorted({item["path"] for item in current_findings}) new_bypass_files = sorted({item["path"] for item in new_bypass_findings}) counts_by_method = method_counts(current_findings) + custom_direct_senders = [ + item + for item in current_findings + if item["detection_kind"] == "custom_direct_sender" + ] + direct_endpoints = [ + item + for item in current_findings + if item["detection_kind"] == "direct_bot_api_endpoint" + ] source_summary = source_snapshot["summary"] return { "schema_version": "telegram_notification_egress_no_new_bypass_guard_v1", "generated_at": generated, "git_commit": git_short_sha(root), - "status": "pass_no_new_bypass" if not new_bypass_findings else "blocked_new_bypass_detected", + "status": "pass_no_direct_or_custom_bypass" + if not current_findings + else "blocked_direct_or_custom_bypass_detected", "mode": "repo_source_scan_no_secret_value_no_telegram_send", "source_snapshot": SOURCE_SNAPSHOT.as_posix(), "guarded_roots": [path.as_posix() for path in SCAN_ROOTS], "guarded_bot_methods": list(GUARDED_BOT_METHODS), "summary": { - "source_direct_bot_api_call_count": source_summary["direct_bot_api_call_count"], - "source_direct_bot_api_file_count": source_summary["direct_bot_api_file_count"], + "source_direct_bot_api_call_count": source_summary[ + "direct_bot_api_call_count" + ], + "source_direct_bot_api_file_count": source_summary[ + "direct_bot_api_file_count" + ], "baseline_signature_count": sum(baseline.values()), + "detected_direct_bot_api_call_count": len(detected_findings), "current_direct_bot_api_call_count": len(current_findings), "current_direct_bot_api_file_count": len(current_files), + "current_direct_bot_api_endpoint_count": len(direct_endpoints), + "current_custom_direct_sender_count": len(custom_direct_senders), + "github_frozen_legacy_direct_bot_api_call_count": len( + frozen_legacy_findings + ), + "github_frozen_legacy_direct_bot_api_file_count": len( + {item["path"] for item in frozen_legacy_findings} + ), "guarded_method_count": len(GUARDED_BOT_METHODS), "sendMessage_call_count": counts_by_method["sendMessage"], "sendDocument_call_count": counts_by_method["sendDocument"], @@ -209,7 +750,9 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: ), "new_bypass_count": len(new_bypass_findings), "new_bypass_file_count": len(new_bypass_files), - "removed_baseline_call_count": sum(item["removed_count"] for item in removed_baseline_signatures), + "removed_baseline_call_count": sum( + item["removed_count"] for item in removed_baseline_signatures + ), "runtime_gate_count": 0, "action_button_count": 0, }, @@ -218,6 +761,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: "telegram_send_authorized": False, "bot_api_call_authorized": False, "workflow_modification_authorized": False, + "workflow_execution_authorized": False, + "github_workflow_execution_authorized": False, "script_modification_authorized": False, "api_sender_refactor_authorized": False, "secret_value_collection_allowed": False, @@ -231,15 +776,18 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: "not_authorization": True, }, "current_direct_bot_api_calls": current_findings, + "detected_direct_bot_api_calls": detected_findings, + "frozen_legacy_direct_bot_api_calls": frozen_legacy_findings, "new_bypass_findings": new_bypass_findings, "removed_baseline_signatures": removed_baseline_signatures, "operator_interpretation": [ - "new_bypass_count 維持 0 才代表沒有新增未登記 Telegram Bot API 直送旁路。", + "current_direct_bot_api_call_count 必須為 0,才代表沒有 Telegram direct/custom bypass。", + ".github/workflows 僅以 frozen_legacy_source_truth 盤點,不計入 active/new failure;runtime 與 workflow execution 均未授權且禁止執行。", ( f"committed inventory 目前有 {source_summary['direct_bot_api_call_count']} 個 direct Bot API call site;" - "若數值大於 0 仍是待 controlled migration 的基線,不代表已批准保留。" + "baseline 只做 drift 比對,不能批准或隱藏目前 source 旁路。" ), - "sendDocument / sendPhoto / sendMediaGroup 等附件型出口若出現在 repo source,會被視為新增旁路並阻擋。", + "分段 URL、requests/httpx/urllib、Invoke-WebRequest/Invoke-RestMethod 與自訂 sender 都會被掃描。", "本 guard 只讀 repo source 與 committed snapshot,不送 Telegram、不讀 Bot token、不修改 workflow / script / API sender。", ], } @@ -248,10 +796,11 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: def validate(root: Path) -> None: report = build_report(root) errors: list[str] = [] - if report["summary"]["new_bypass_count"]: - for item in report["new_bypass_findings"]: + if report["summary"]["current_direct_bot_api_call_count"]: + for item in report["current_direct_bot_api_calls"]: errors.append( - f"{item['path']}:{item['line']}: 新增未登記 Telegram Bot API 旁路 {item['method']}" + f"{item['path']}:{item['line']}: Telegram direct/custom bypass " + f"{item['detection_kind']} {item['method']} ({item['function']})" ) if errors: @@ -262,10 +811,14 @@ def validate(root: Path) -> None: def main() -> None: - parser = argparse.ArgumentParser(description="檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路") + parser = argparse.ArgumentParser( + description="檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路" + ) parser.add_argument("--root", default=".", help="repository root") parser.add_argument("--output", help="寫出 JSON 報告") - parser.add_argument("--generated-at", help="固定 generated_at 時間,供 committed snapshot 使用") + parser.add_argument( + "--generated-at", help="固定 generated_at 時間,供 committed snapshot 使用" + ) args = parser.parse_args() root = Path(args.root).resolve() @@ -273,13 +826,18 @@ def main() -> None: if args.output: output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) - output.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + output.write_text( + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) validate(root) summary = report["summary"] print( "TELEGRAM_NOTIFICATION_EGRESS_NO_NEW_BYPASS_GUARD_OK " f"current={summary['current_direct_bot_api_call_count']} " + f"custom={summary['current_custom_direct_sender_count']} " + f"frozen_legacy={summary['github_frozen_legacy_direct_bot_api_call_count']} " f"baseline={summary['baseline_signature_count']} " f"new={summary['new_bypass_count']} " f"sendDocument={summary['sendDocument_call_count']} "