#!/usr/bin/env python3 """ IwoooS DNS / TLS / certbot 只讀清冊。 本工具只讀取已提交的 Nginx repo-only drift snapshot，整理 domain、TLS certificate path、ACME challenge、upstream、admin route 與 WebSocket 影響面。 它不做 DNS 查詢、不連線 TLS、不執行 certbot、不 SSH、不 reload Nginx。 """ from __future__ import annotations import argparse import json import re import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any TAIPEI = timezone(timedelta(hours=8)) TIER_ORDER = {"C0": 0, "C1": 1, "C2": 2, "C3": 3} CERT_DOMAIN_RE = re.compile(r"/etc/letsencrypt/live/([^/]+)/") 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 cert_domain(path: str) -> str | None: match = CERT_DOMAIN_RE.search(path) if not match: return None return match.group(1) def cert_domain_matches(domain: str, candidate: str) -> bool: if candidate == domain: return True if candidate.startswith("*.") and domain.endswith(candidate[1:]): return True return False def strongest_tier(tiers: set[str]) -> str: if not tiers: return "C3" return sorted(tiers, key=lambda tier: TIER_ORDER.get(tier, 99))[0] def ensure_domain(domains: dict[str, dict[str, Any]], domain: str) -> dict[str, Any]: if domain not in domains: domains[domain] = { "domain": domain, "hosts": set(), "config_ids": set(), "source_paths": set(), "live_paths": set(), "control_tiers": set(), "server_block_indexes": set(), "listens": set(), "certificate_paths": set(), "certificate_key_paths": set(), "certificate_path_domains": set(), "acme_challenge_roots": set(), "upstreams": set(), "admin_route_count": 0, "websocket_route_count": 0, } return domains[domain] def add_server_to_domain( domain_item: dict[str, Any], config: dict[str, Any], server: dict[str, Any], ) -> None: domain_item["hosts"].add(config["host"]) domain_item["config_ids"].add(config["config_id"]) domain_item["source_paths"].add(config["repo_source_path"]) domain_item["live_paths"].add(config["live_path"]) domain_item["control_tiers"].add(config["control_tier"]) domain_item["server_block_indexes"].add(f"{config['config_id']}#{server['index']}") domain_item["listens"].update(server.get("listens", [])) domain_item["certificate_paths"].update(server.get("ssl_certificates", [])) domain_item["certificate_key_paths"].update(server.get("ssl_certificate_keys", [])) domain_item["upstreams"].update(server.get("proxy_passes", [])) for path in server.get("ssl_certificates", []): candidate = cert_domain(path) if candidate: domain_item["certificate_path_domains"].add(candidate) for location in server.get("locations", []): path = str(location.get("path", "")) if ".well-known/acme-challenge" in path: domain_item["acme_challenge_roots"].update(location.get("roots", [])) if "/admin" in path: domain_item["admin_route_count"] += 1 if location.get("websocket_upgrade"): domain_item["websocket_route_count"] += 1 def normalize_domain_item(item: dict[str, Any]) -> dict[str, Any]: domain = item["domain"] cert_domains = sorted(item["certificate_path_domains"]) certificate_owner_confirmation_required = bool(cert_domains) and not any( cert_domain_matches(domain, candidate) for candidate in cert_domains ) tls_certificate_path_present = bool(item["certificate_paths"]) if not tls_certificate_path_present: status = "repo_only_tls_path_missing" elif certificate_owner_confirmation_required: status = "repo_only_owner_confirmation_required" else: status = "repo_only_ready_for_owner_review" return { "domain": domain, "hosts": sorted(item["hosts"]), "config_ids": sorted(item["config_ids"]), "source_paths": sorted(item["source_paths"]), "live_paths": sorted(item["live_paths"]), "control_tier": strongest_tier(item["control_tiers"]), "server_block_refs": sorted(item["server_block_indexes"]), "listens": sorted(item["listens"]), "tls_certificate_path_present": tls_certificate_path_present, "certificate_paths": sorted(item["certificate_paths"]), "certificate_key_paths": sorted(item["certificate_key_paths"]), "certificate_path_domains": cert_domains, "certificate_owner_confirmation_required": certificate_owner_confirmation_required, "acme_challenge_present": bool(item["acme_challenge_roots"]), "acme_challenge_roots": sorted(item["acme_challenge_roots"]), "upstreams": sorted(item["upstreams"]), "admin_route_count": item["admin_route_count"], "websocket_route_count": item["websocket_route_count"], "live_tls_probe_status": "not_executed", "dns_resolution_status": "not_executed", "certbot_renewal_status": "not_executed", "owner_review_status": status, } def build_report(root: Path, nginx_report: dict[str, Any], generated_at: str | None) -> dict[str, Any]: domains: dict[str, dict[str, Any]] = {} for config in nginx_report.get("configs", []): parsed = config.get("repo_source", {}).get("parsed", {}) for server in parsed.get("servers", []): for domain in server.get("server_names", []): if not domain or domain == "_": continue item = ensure_domain(domains, domain) add_server_to_domain(item, config, server) managed_domains = [normalize_domain_item(item) for item in domains.values()] managed_domains.sort(key=lambda item: item["domain"]) certificate_paths = sorted( { path for item in managed_domains for path in item.get("certificate_paths", []) } ) owner_confirmation_required = [ item for item in managed_domains if item["certificate_owner_confirmation_required"] or not item["tls_certificate_path_present"] ] acme_domains = [item for item in managed_domains if item["acme_challenge_present"]] admin_domains = [item for item in managed_domains if item["admin_route_count"] > 0] websocket_domains = [item for item in managed_domains if item["websocket_route_count"] > 0] return { "schema_version": "domain_tls_certbot_inventory_v1", "generated_at": generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds"), "mode": "repo_only_from_nginx_source_of_truth", "git_commit": git_short_sha(root), "source_nginx_report": "docs/security/nginx-config-drift-repo.snapshot.json", "execution_boundaries": { "dns_query_executed": False, "live_tls_probe_executed": False, "certbot_renew_executed": False, "nginx_reload_executed": False, "host_write_executed": False, "runtime_gate_opened": False, "secret_value_collected": False, "action_buttons_allowed": False, }, "summary": { "source_config_count": nginx_report.get("summary", {}).get("source_config_count", 0), "managed_domain_count": len(managed_domains), "unique_certificate_path_count": len(certificate_paths), "acme_challenge_domain_count": len(acme_domains), "certificate_owner_confirmation_required_count": len(owner_confirmation_required), "admin_route_domain_count": len(admin_domains), "websocket_route_domain_count": len(websocket_domains), "owner_response_request_sent_count": 0, "owner_response_received_count": 0, "owner_response_accepted_count": 0, "runtime_gate_count": 0, "live_tls_probe_executed": False, "dns_change_executed": False, "certbot_renew_executed": False, "nginx_reload_executed": False, "action_buttons_allowed": False, }, "certificate_paths": certificate_paths, "managed_domains": managed_domains, "owner_confirmation_required_domains": [ { "domain": item["domain"], "certificate_path_domains": item["certificate_path_domains"], "tls_certificate_path_present": item["tls_certificate_path_present"], "owner_review_status": item["owner_review_status"], } for item in owner_confirmation_required ], "required_owner_fields": [ "owner_role_or_team", "decision", "decision_reason", "affected_scope", "redacted_evidence_refs", "followup_owner", "rollback_owner", "maintenance_window", "validation_plan", ], "next_steps": [ "請 owner 確認 certificate path 是否由 SAN 或 wildcard 合法覆蓋；未確認前不得 renew 或 reload。", "未來若要做 live TLS / DNS probe，需另行 scope approval；本清冊只保留 repo-only 證據。", "任何 certbot renew、Nginx reload 或 DNS 變更都必須另開維護窗口、rollback owner 與 post-check。", ], } def main() -> int: parser = argparse.ArgumentParser(description="IwoooS DNS / TLS / certbot 只讀清冊") parser.add_argument("--root", default=".", help="repo root") parser.add_argument( "--nginx-report", default="docs/security/nginx-config-drift-repo.snapshot.json", help="Nginx repo-only drift detector snapshot", ) parser.add_argument("--output", help="寫出 JSON 報告") parser.add_argument("--generated-at", help="固定報告時間，供 committed snapshot 使用") args = parser.parse_args() root = Path(args.root).resolve() nginx_report_path = root / args.nginx_report nginx_report = load_json(nginx_report_path) report = build_report(root, nginx_report, args.generated_at) payload = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) if args.output: output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(payload + "\n", encoding="utf-8") else: print(payload) summary = report["summary"] print( "DOMAIN_TLS_CERTBOT_INVENTORY_OK " f"domains={summary['managed_domain_count']} " f"cert_paths={summary['unique_certificate_path_count']} " f"owner_confirm={summary['certificate_owner_confirmation_required_count']} " f"runtime_gate={summary['runtime_gate_count']}", file=sys.stderr, ) return 0 if __name__ == "__main__": sys.exit(main())