feat(telegram): enforce canonical product routing

scripts/security/telegram-notification-egress-inventory.py

@@ -22,21 +22,37 @@ from typing import Any
 TAIPEI = timezone(timedelta(hours=8))
 
 SCAN_ROOTS = (
+    Path("agent99-control-plane.ps1"),
     Path(".gitea/workflows"),
     Path("scripts/ops"),
     Path("scripts/ci"),
+    Path("scripts/reboot-recovery"),
     Path("apps/api/src"),
 )
-SCAN_SUFFIXES = {".py", ".sh", ".js", ".yml", ".yaml"}
+SCAN_SUFFIXES = {".py", ".ps1", ".sh", ".js", ".yml", ".yaml"}
 
+GUARDED_BOT_METHODS = (
+    "sendMessage",
+    "sendDocument",
+    "sendPhoto",
+    "sendMediaGroup",
+    "editMessageText",
+    "sendAnimation",
+    "sendVideo",
+    "sendAudio",
+    "sendVoice",
+)
 DIRECT_BOT_API_RE = re.compile(
-    r"api\.telegram\.org/bot.*sendMessage|sendMessage.*api\.telegram\.org/bot"
+    r"api\.telegram\.org/bot.*?/(?P<method>"
+    + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
+    + r")\b",
+    re.IGNORECASE,
 )
 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 = re.compile(r"api\.telegram\.org/bot.*?/sendMessage")
+BOT_TOKEN_URL_RE = DIRECT_BOT_API_RE
 
 REQUIRED_OWNER_FIELDS = [
     "egress_surface_id",
@@ -134,6 +150,10 @@ def iter_scannable_files(root: Path) -> list[Path]:
         absolute_root = root / scan_root
         if not absolute_root.exists():
             continue
+        if absolute_root.is_file():
+            if absolute_root.suffix in SCAN_SUFFIXES:
+                files.append(absolute_root)
+            continue
         for path in absolute_root.rglob("*"):
             if path.is_file() and path.suffix in SCAN_SUFFIXES:
                 files.append(path)
@@ -143,7 +163,10 @@ def iter_scannable_files(root: Path) -> list[Path]:
 def sanitize_excerpt(line: str) -> str:
     excerpt = line.strip()
     excerpt = SECRET_INTERPOLATION_RE.sub("${{ secrets.<redacted> }}", excerpt)
-    excerpt = BOT_TOKEN_URL_RE.sub("api.telegram.org/bot<redacted>/sendMessage", excerpt)
+    excerpt = BOT_TOKEN_URL_RE.sub(
+        lambda match: f"api.telegram.org/bot<redacted>/{match.group('method')}",
+        excerpt,
+    )
     return excerpt[:180]
 
 
@@ -154,6 +177,10 @@ def surface_kind(relative_path: str) -> str:
         return "ops_script_direct_bot_api"
     if relative_path.startswith("scripts/ci/"):
         return "ci_script_direct_bot_api"
+    if relative_path == "agent99-control-plane.ps1":
+        return "agent99_control_plane_direct_bot_api"
+    if relative_path.startswith("scripts/reboot-recovery/"):
+        return "reboot_recovery_direct_bot_api"
     if relative_path.startswith("apps/api/src/"):
         return "api_direct_bot_api"
     return "other_direct_bot_api"
@@ -174,7 +201,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
         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):
-            if DIRECT_BOT_API_RE.search(line):
+            direct_match = DIRECT_BOT_API_RE.search(line)
+            if direct_match:
                 kind = surface_kind(relative_path)
                 direct_calls.append(
                     {
@@ -183,6 +211,7 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
                         "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,

scripts/security/telegram-notification-egress-no-new-bypass-guard.py

@@ -24,12 +24,14 @@ TAIPEI = timezone(timedelta(hours=8))
 
 SOURCE_SNAPSHOT = Path("docs/security/telegram-notification-egress-inventory.snapshot.json")
 SCAN_ROOTS = (
+    Path("agent99-control-plane.ps1"),
     Path(".gitea/workflows"),
     Path("scripts/ops"),
     Path("scripts/ci"),
+    Path("scripts/reboot-recovery"),
     Path("apps/api/src"),
 )
-SCAN_SUFFIXES = {".py", ".sh", ".js", ".yml", ".yaml"}
+SCAN_SUFFIXES = {".py", ".ps1", ".sh", ".js", ".yml", ".yaml"}
 GUARDED_BOT_METHODS = (
     "sendMessage",
     "sendDocument",
@@ -77,6 +79,10 @@ def iter_scannable_files(root: Path) -> list[Path]:
         absolute_root = root / scan_root
         if not absolute_root.exists():
             continue
+        if absolute_root.is_file():
+            if absolute_root.suffix in SCAN_SUFFIXES:
+                files.append(absolute_root)
+            continue
         for path in absolute_root.rglob("*"):
             if path.is_file() and path.suffix in SCAN_SUFFIXES:
                 files.append(path)