283 lines
11 KiB
Python
283 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""檢查 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，而不是直接合併。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
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(".gitea/workflows"),
|
||
Path("scripts/ops"),
|
||
Path("scripts/ci"),
|
||
Path("apps/api/src"),
|
||
)
|
||
SCAN_SUFFIXES = {".py", ".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,
|
||
)
|
||
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,
|
||
)
|
||
|
||
|
||
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
|
||
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,
|
||
)
|
||
return excerpt[:180]
|
||
|
||
|
||
def signature(path: str, method: str, sanitized_excerpt: str) -> str:
|
||
return f"{path}::{method.lower()}::{sanitized_excerpt}"
|
||
|
||
|
||
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", []):
|
||
excerpt = item.get("sanitized_excerpt", "")
|
||
match = BOT_ENDPOINT_RE.search(excerpt)
|
||
method = match.group("method") if match else "sendMessage"
|
||
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 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),
|
||
}
|
||
)
|
||
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)
|
||
current_findings = scan_current_direct_endpoints(root)
|
||
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)
|
||
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",
|
||
"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()),
|
||
"current_direct_bot_api_call_count": len(current_findings),
|
||
"current_direct_bot_api_file_count": len(current_files),
|
||
"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,
|
||
"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,
|
||
"new_bypass_findings": new_bypass_findings,
|
||
"removed_baseline_signatures": removed_baseline_signatures,
|
||
"operator_interpretation": [
|
||
"new_bypass_count 維持 0 才代表沒有新增未登記 Telegram Bot API 直送旁路。",
|
||
"既有 18 個 sendMessage 旁路仍是待 owner response 的基線，不代表已批准或已收斂。",
|
||
"sendDocument / sendPhoto / sendMediaGroup 等附件型出口若出現在 repo source，會被視為新增旁路並阻擋。",
|
||
"本 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"]["new_bypass_count"]:
|
||
for item in report["new_bypass_findings"]:
|
||
errors.append(
|
||
f"{item['path']}:{item['line']}: 新增未登記 Telegram Bot API 旁路 {item['method']}"
|
||
)
|
||
|
||
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"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())
|