850 lines
29 KiB
Python
850 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
"""檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路。
|
||
|
||
本 guard 只掃描 repo 原始碼與 committed snapshot，不讀 secret、不呼叫
|
||
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
|
||
import sys
|
||
from collections import Counter
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
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(".github/workflows"),
|
||
Path("scripts/ops"),
|
||
Path("scripts/ci"),
|
||
Path("scripts/reboot-recovery"),
|
||
Path("apps/api/src"),
|
||
)
|
||
SCAN_SUFFIXES = {".py", ".ps1", ".sh", ".js", ".yml", ".yaml"}
|
||
GUARDED_BOT_METHODS = (
|
||
"sendMessage",
|
||
"sendDocument",
|
||
"sendPhoto",
|
||
"sendMediaGroup",
|
||
"editMessageText",
|
||
"sendAnimation",
|
||
"sendVideo",
|
||
"sendAudio",
|
||
"sendVoice",
|
||
)
|
||
|
||
BOT_ENDPOINT_RE = re.compile(
|
||
r"api\.telegram\.org/bot.*?/(?P<method>"
|
||
+ "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
|
||
+ r")\b",
|
||
re.IGNORECASE,
|
||
)
|
||
GUARDED_BOT_METHOD_RE = re.compile(
|
||
r"\b(?P<method>"
|
||
+ "|".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<method>"
|
||
+ "|".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<name>[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<method>"
|
||
+ "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
|
||
+ r")\b",
|
||
re.IGNORECASE,
|
||
)
|
||
BOT_TOKEN_FRAGMENT_RE = re.compile(
|
||
r"bot[^/\s\"']+/(?P<method>"
|
||
+ "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
|
||
+ r")\b",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def git_short_sha(root: Path) -> str:
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "rev-parse", "--short", "HEAD"],
|
||
cwd=root,
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
return result.stdout.strip()
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
|
||
def iter_scannable_files(root: Path) -> list[Path]:
|
||
files: list[Path] = []
|
||
for scan_root in SCAN_ROOTS:
|
||
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)
|
||
return sorted(files)
|
||
|
||
|
||
def sanitize_excerpt(line: str) -> str:
|
||
excerpt = line.strip()
|
||
excerpt = SECRET_INTERPOLATION_RE.sub("${{ secrets.<redacted> }}", excerpt)
|
||
excerpt = BOT_TOKEN_URL_RE.sub(
|
||
lambda match: f"api.telegram.org/bot<redacted>/{match.group('method')}",
|
||
excerpt,
|
||
)
|
||
excerpt = BOT_TOKEN_FRAGMENT_RE.sub(
|
||
lambda match: f"bot<redacted>/{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(
|
||
(
|
||
"<module>",
|
||
"<module>",
|
||
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(
|
||
(
|
||
"<module>",
|
||
"<module>",
|
||
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 == "<module>"
|
||
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"))
|
||
|
||
|
||
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 = str(
|
||
item.get("method") or (match.group("method") if match else "dynamic")
|
||
)
|
||
baseline[signature(item["path"], method, excerpt)] += 1
|
||
return baseline
|
||
|
||
|
||
def scan_current_direct_endpoints(root: Path) -> list[dict[str, Any]]:
|
||
findings: 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 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
|
||
|
||
|
||
def method_counts(findings: list[dict[str, Any]]) -> dict[str, int]:
|
||
counts = {method: 0 for method in GUARDED_BOT_METHODS}
|
||
for item in findings:
|
||
for method in GUARDED_BOT_METHODS:
|
||
if item["method"].lower() == method.lower():
|
||
counts[method] += 1
|
||
break
|
||
return counts
|
||
|
||
|
||
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)
|
||
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]] = []
|
||
|
||
for item in current_findings:
|
||
item_signature = item["signature"]
|
||
if remaining_baseline[item_signature] > 0:
|
||
remaining_baseline[item_signature] -= 1
|
||
continue
|
||
new_bypass_findings.append(item)
|
||
|
||
removed_baseline_signatures = [
|
||
{"signature": item_signature, "removed_count": count}
|
||
for item_signature, count in sorted(remaining_baseline.items())
|
||
if count > 0
|
||
]
|
||
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_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"
|
||
],
|
||
"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"],
|
||
"sendPhoto_call_count": counts_by_method["sendPhoto"],
|
||
"sendMediaGroup_call_count": counts_by_method["sendMediaGroup"],
|
||
"editMessageText_call_count": counts_by_method["editMessageText"],
|
||
"other_guarded_method_call_count": sum(
|
||
count
|
||
for method, count in counts_by_method.items()
|
||
if method
|
||
not in {
|
||
"sendMessage",
|
||
"sendDocument",
|
||
"sendPhoto",
|
||
"sendMediaGroup",
|
||
"editMessageText",
|
||
}
|
||
),
|
||
"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
|
||
),
|
||
"runtime_gate_count": 0,
|
||
"action_button_count": 0,
|
||
},
|
||
"execution_boundaries": {
|
||
"runtime_execution_authorized": False,
|
||
"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,
|
||
"secret_hash_collection_allowed": False,
|
||
"partial_token_collection_allowed": False,
|
||
"chat_route_change_authorized": False,
|
||
"bot_token_change_authorized": False,
|
||
"raw_payload_storage_allowed": False,
|
||
"production_write_authorized": False,
|
||
"action_buttons_allowed": False,
|
||
"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": [
|
||
"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；"
|
||
"baseline 只做 drift 比對，不能批准或隱藏目前 source 旁路。"
|
||
),
|
||
"分段 URL、requests/httpx/urllib、Invoke-WebRequest/Invoke-RestMethod 與自訂 sender 都會被掃描。",
|
||
"本 guard 只讀 repo source 與 committed snapshot，不送 Telegram、不讀 Bot token、不修改 workflow / script / API sender。",
|
||
],
|
||
}
|
||
|
||
|
||
def validate(root: Path) -> None:
|
||
report = build_report(root)
|
||
errors: list[str] = []
|
||
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 direct/custom bypass "
|
||
f"{item['detection_kind']} {item['method']} ({item['function']})"
|
||
)
|
||
|
||
if errors:
|
||
raise SystemExit(
|
||
"BLOCKED telegram notification egress no-new-bypass guard:\n"
|
||
+ "\n".join(f"- {error}" for error in errors)
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
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 使用"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
root = Path(args.root).resolve()
|
||
report = build_report(root, args.generated_at)
|
||
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",
|
||
)
|
||
|
||
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']} "
|
||
f"runtime_gate={summary['runtime_gate_count']}"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|