#!/usr/bin/env python3 """Build a no-runtime migration plan draft for Telegram notification egress.""" from __future__ import annotations import argparse import json import subprocess import sys 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-owner-request-draft.snapshot.json") PLAN_FIELDS = [ "migration_candidate_id", "source_request_draft_id", "source_path", "surface_kind", "direct_call_count", "proposed_wave", "proposed_target", "proposed_change_summary", "required_owner_response_ref", "required_maintenance_window", "required_rollback_owner", "required_postcheck_ref", "required_delivery_receipt_ref", "required_no_secret_value_attestation", "required_no_raw_payload_attestation", "required_no_false_green_attestation", "not_authorization", ] REVIEWER_CHECKS = [ "source_owner_request_draft_current", "owner_response_required_before_change", "maintenance_window_required_before_change", "rollback_owner_required_before_change", "delivery_receipt_plan_required", "postcheck_plan_required", "redaction_contract_required", "break_glass_fallback_explicit", "no_secret_value_required", "no_raw_payload_required", "no_false_green_required", "workflow_changes_separate_from_docs", "script_changes_separate_from_docs", "api_sender_refactor_separate_from_docs", "runtime_gate_stays_zero", ] OUTCOME_LANES = [ "draft_waiting_owner_response", "ready_for_workflow_migration_review", "ready_for_ops_script_migration_review", "ready_for_api_sender_migration_review", "request_missing_owner_response", "request_missing_maintenance_or_rollback", "reject_secret_or_raw_payload", "reject_false_green_claim", "waiting_runtime_gate", ] BLOCKED_ACTIONS = [ "modify_workflow", "modify_ops_script", "refactor_api_sender", "send_telegram", "call_bot_api", "dispatch_workflow", "trigger_cd", "deploy_production", "read_secret_store", "collect_secret_value", "collect_secret_hash", "collect_partial_token", "store_raw_payload", "store_unredacted_log", "change_chat_route", "change_bot_token", "rotate_secret", "accept_cd_success_as_delivery_receipt", "accept_route_200_as_notification_delivery", "open_runtime_gate", "add_action_button", ] 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 load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def target_for(surface_kind: str) -> tuple[str, str, str]: if surface_kind == "gitea_workflow_direct_bot_api": return ( "wave_1_workflow_notification_wrapper", "scripts/ci/notify-awoooi-cicd.sh or AWOOI Alertmanager webhook", "Replace direct workflow Bot API send with normalized CI/CD notification wrapper after owner approval.", ) if surface_kind == "ops_script_direct_bot_api": return ( "wave_2_ops_notification_wrapper", "scripts/ops/notify-awoooi-ops.sh or AWOOI Alertmanager webhook", "Replace direct ops fallback send with normalized ops notification wrapper or documented break-glass fallback.", ) return ( "wave_3_api_sender_gateway", "TelegramGateway final-exit formatter", "Route API interim sender through TelegramGateway or equivalent final-exit normalization and mirror contract.", ) def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]: generated = generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds") source = load_json(root / SOURCE_SNAPSHOT) candidates: list[dict[str, Any]] = [] for item in source["request_drafts"]: wave, target, summary = target_for(item["surface_kind"]) candidates.append( { "migration_candidate_id": f"telegram_notification_egress_migration:{item['source_path']}", "source_request_draft_id": item["request_draft_id"], "source_path": item["source_path"], "surface_kind": item["surface_kind"], "direct_call_count": item["direct_call_count"], "proposed_wave": wave, "proposed_target": target, "proposed_change_summary": summary, "plan_fields": PLAN_FIELDS, "reviewer_checks": REVIEWER_CHECKS, "outcome_lanes": OUTCOME_LANES, "blocked_actions": BLOCKED_ACTIONS, "owner_response_required": True, "maintenance_window_required": True, "rollback_owner_required": True, "postcheck_required": True, "delivery_receipt_required": True, "owner_response_received": False, "owner_response_accepted": False, "migration_authorized": False, "workflow_modification_authorized": False, "script_modification_authorized": False, "api_sender_refactor_authorized": False, "telegram_send_authorized": False, "bot_api_call_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, } ) workflow = [item for item in candidates if item["surface_kind"] == "gitea_workflow_direct_bot_api"] ops = [item for item in candidates if item["surface_kind"] == "ops_script_direct_bot_api"] api = [item for item in candidates if item["surface_kind"] == "api_direct_bot_api"] waves = sorted({item["proposed_wave"] for item in candidates}) return { "schema_version": "telegram_notification_egress_migration_plan_draft_v1", "generated_at": generated, "git_commit": git_short_sha(root), "status": "migration_plan_draft_ready_no_runtime_action", "mode": "metadata_only_no_workflow_script_api_change_no_telegram_send", "source_snapshot": SOURCE_SNAPSHOT.as_posix(), "source_schema_version": source["schema_version"], "source_status": source["status"], "summary": { "source_request_draft_count": source["summary"]["request_draft_count"], "source_direct_bot_api_call_count": source["summary"]["source_direct_bot_api_call_count"], "migration_candidate_count": len(candidates), "workflow_migration_candidate_count": len(workflow), "ops_script_migration_candidate_count": len(ops), "api_direct_migration_candidate_count": len(api), "proposed_wave_count": len(waves), "plan_field_count": len(PLAN_FIELDS), "reviewer_check_count": len(REVIEWER_CHECKS), "outcome_lane_count": len(OUTCOME_LANES), "blocked_action_count": len(BLOCKED_ACTIONS), "owner_response_required_count": len(candidates), "maintenance_window_required_count": len(candidates), "rollback_owner_required_count": len(candidates), "postcheck_required_count": len(candidates), "delivery_receipt_required_count": len(candidates), "owner_response_received_count": 0, "owner_response_accepted_count": 0, "migration_authorized_count": 0, "workflow_modification_authorized_count": 0, "script_modification_authorized_count": 0, "api_sender_refactor_authorized_count": 0, "telegram_send_authorized_count": 0, "bot_api_call_authorized_count": 0, "secret_value_collection_allowed_count": 0, "raw_payload_storage_allowed_count": 0, "production_write_authorized_count": 0, "runtime_gate_count": 0, "action_button_count": 0, }, "execution_boundaries": { "runtime_execution_authorized": False, "workflow_modification_authorized": False, "script_modification_authorized": False, "api_sender_refactor_authorized": False, "telegram_send_authorized": False, "bot_api_call_authorized": False, "secret_value_collection_allowed": False, "raw_payload_storage_allowed": False, "production_write_authorized": False, "action_buttons_allowed": False, "not_authorization": True, }, "proposed_waves": waves, "migration_candidates": candidates, "operator_interpretation": [ "This is a migration plan draft only; it does not authorize workflow, script, API, Telegram, or production changes.", "Every candidate still requires owner response, maintenance window, rollback owner, receipt plan, and post-check evidence.", "Direct Bot API convergence remains 0 until a separate runtime-approved change is implemented and verified.", ], } def validate(root: Path) -> None: report = build_report(root) if report["summary"]["migration_candidate_count"] != report["summary"]["source_request_draft_count"]: raise SystemExit("BLOCKED telegram egress migration plan: candidate/draft count mismatch") if report["summary"]["runtime_gate_count"] != 0: raise SystemExit("BLOCKED telegram egress migration plan: runtime gate must stay 0") def main() -> None: parser = argparse.ArgumentParser(description="Build Telegram notification egress migration plan draft") 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") args = parser.parse_args() root = Path(args.root).resolve() report = build_report(root, args.generated_at) payload = json.dumps(report, ensure_ascii=False, indent=2) + "\n" if args.output: Path(args.output).write_text(payload, encoding="utf-8") else: sys.stdout.write(payload) print( "TELEGRAM_NOTIFICATION_EGRESS_MIGRATION_PLAN_DRAFT_OK " f"candidates={report['summary']['migration_candidate_count']} " f"waves={report['summary']['proposed_wave_count']} " f"authorized={report['summary']['migration_authorized_count']} " f"runtime_gate={report['summary']['runtime_gate_count']}", file=sys.stderr, ) if __name__ == "__main__": main()