docs(security): add Telegram egress migration plan draft [skip ci]
This commit is contained in:
Your Name
@@ -234,6 +234,9 @@ def validate(root: Path) -> None:
|
||||
telegram_notification_egress_owner_request_draft = load_json(
|
||||
security_dir / "telegram-notification-egress-owner-request-draft.snapshot.json"
|
||||
)
|
||||
telegram_notification_egress_migration_plan_draft = load_json(
|
||||
security_dir / "telegram-notification-egress-migration-plan-draft.snapshot.json"
|
||||
)
|
||||
public_runtime_config_change_evidence_acceptance = load_json(
|
||||
security_dir / "public-runtime-config-change-evidence-acceptance.snapshot.json"
|
||||
)
|
||||
@@ -21773,6 +21776,124 @@ def validate(root: Path) -> None:
|
||||
f"telegram_notification_egress_owner_request_draft.{item['request_draft_id']}.{false_key}",
|
||||
item[false_key],
|
||||
)
|
||||
assert_equal(
|
||||
"telegram_notification_egress_migration_plan_draft.schema",
|
||||
telegram_notification_egress_migration_plan_draft["schema_version"],
|
||||
"telegram_notification_egress_migration_plan_draft_v1",
|
||||
)
|
||||
assert_equal(
|
||||
"telegram_notification_egress_migration_plan_draft.status",
|
||||
telegram_notification_egress_migration_plan_draft["status"],
|
||||
"migration_plan_draft_ready_no_runtime_action",
|
||||
)
|
||||
assert_equal(
|
||||
"telegram_notification_egress_migration_plan_draft.mode",
|
||||
telegram_notification_egress_migration_plan_draft["mode"],
|
||||
"metadata_only_no_workflow_script_api_change_no_telegram_send",
|
||||
)
|
||||
expected_telegram_egress_migration_plan_summary = {
|
||||
"source_request_draft_count": 11,
|
||||
"source_direct_bot_api_call_count": 18,
|
||||
"migration_candidate_count": 11,
|
||||
"workflow_migration_candidate_count": 6,
|
||||
"ops_script_migration_candidate_count": 4,
|
||||
"api_direct_migration_candidate_count": 1,
|
||||
"proposed_wave_count": 3,
|
||||
"plan_field_count": 17,
|
||||
"reviewer_check_count": 15,
|
||||
"outcome_lane_count": 9,
|
||||
"blocked_action_count": 21,
|
||||
"owner_response_required_count": 11,
|
||||
"maintenance_window_required_count": 11,
|
||||
"rollback_owner_required_count": 11,
|
||||
"postcheck_required_count": 11,
|
||||
"delivery_receipt_required_count": 11,
|
||||
"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,
|
||||
}
|
||||
for key, expected in expected_telegram_egress_migration_plan_summary.items():
|
||||
assert_equal(
|
||||
f"telegram_notification_egress_migration_plan_draft.summary.{key}",
|
||||
telegram_notification_egress_migration_plan_draft["summary"][key],
|
||||
expected,
|
||||
)
|
||||
assert_equal(
|
||||
"telegram_notification_egress_migration_plan_draft.proposed_waves",
|
||||
telegram_notification_egress_migration_plan_draft["proposed_waves"],
|
||||
[
|
||||
"wave_1_workflow_notification_wrapper",
|
||||
"wave_2_ops_notification_wrapper",
|
||||
"wave_3_api_sender_gateway",
|
||||
],
|
||||
)
|
||||
for key, value in telegram_notification_egress_migration_plan_draft["execution_boundaries"].items():
|
||||
if key == "not_authorization":
|
||||
assert_true(f"telegram_notification_egress_migration_plan_draft.execution_boundaries.{key}", value)
|
||||
else:
|
||||
assert_false(f"telegram_notification_egress_migration_plan_draft.execution_boundaries.{key}", value)
|
||||
for item in telegram_notification_egress_migration_plan_draft["migration_candidates"]:
|
||||
assert_equal(
|
||||
f"telegram_notification_egress_migration_plan_draft.{item['migration_candidate_id']}.plan_fields",
|
||||
len(item["plan_fields"]),
|
||||
17,
|
||||
)
|
||||
assert_equal(
|
||||
f"telegram_notification_egress_migration_plan_draft.{item['migration_candidate_id']}.reviewer_checks",
|
||||
len(item["reviewer_checks"]),
|
||||
15,
|
||||
)
|
||||
assert_equal(
|
||||
f"telegram_notification_egress_migration_plan_draft.{item['migration_candidate_id']}.outcome_lanes",
|
||||
len(item["outcome_lanes"]),
|
||||
9,
|
||||
)
|
||||
assert_equal(
|
||||
f"telegram_notification_egress_migration_plan_draft.{item['migration_candidate_id']}.blocked_actions",
|
||||
len(item["blocked_actions"]),
|
||||
21,
|
||||
)
|
||||
for true_key in [
|
||||
"owner_response_required",
|
||||
"maintenance_window_required",
|
||||
"rollback_owner_required",
|
||||
"postcheck_required",
|
||||
"delivery_receipt_required",
|
||||
"not_authorization",
|
||||
]:
|
||||
assert_true(
|
||||
f"telegram_notification_egress_migration_plan_draft.{item['migration_candidate_id']}.{true_key}",
|
||||
item[true_key],
|
||||
)
|
||||
for false_key in [
|
||||
"owner_response_received",
|
||||
"owner_response_accepted",
|
||||
"migration_authorized",
|
||||
"workflow_modification_authorized",
|
||||
"script_modification_authorized",
|
||||
"api_sender_refactor_authorized",
|
||||
"telegram_send_authorized",
|
||||
"bot_api_call_authorized",
|
||||
"secret_value_collection_allowed",
|
||||
"raw_payload_storage_allowed",
|
||||
"production_write_authorized",
|
||||
"runtime_gate",
|
||||
"action_buttons_allowed",
|
||||
]:
|
||||
assert_false(
|
||||
f"telegram_notification_egress_migration_plan_draft.{item['migration_candidate_id']}.{false_key}",
|
||||
item[false_key],
|
||||
)
|
||||
assert_equal(
|
||||
"public_runtime_config_change_evidence_acceptance.schema",
|
||||
public_runtime_config_change_evidence_acceptance["schema_version"],
|
||||
|
||||
@@ -0,0 +1,275 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user