"""Plan site-address changes from the active guarded transaction; never guesses IPs.""" import argparse import hashlib import ipaddress import json import os from pathlib import Path import shlex import subprocess import time from urllib.parse import urlsplit from field_ui import print, input import server_context as server STATE = server.path('state', '/var/lib/contextra-maintenance') UNIT = Path("/etc/systemd/system/contextra-runtime.service") def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def active_compose(unit=UNIT): if server.portable(): from compose_backend import current_path return current_path() for line in unit.read_text().splitlines(): if line.startswith("ExecStart="): args = shlex.split(line.split("=", 1)[1]) if "deployment_kit.site_guard" not in args or "--compose" not in args: raise ValueError("Unknown Runtime launcher") path = Path(args[args.index("--compose") + 1]) if not path.is_relative_to(server.path('runtime','/workspace/contextra/runtime')/'state/transactions') or path.is_symlink(): raise ValueError("Unexpected active Compose") return path raise ValueError("Active Compose not found") def addresses(ai_ip, web_ip, manager, control_port, redis_port): ai = ipaddress.ip_address(ai_ip) web = ipaddress.ip_address(web_ip) if ai.version != 4 or web.version != 4 or any(x.is_unspecified or x.is_multicast or x.is_loopback for x in (ai,web)): raise ValueError("Unicast IPv4 addresses required") url = urlsplit(manager) if url.scheme != "https" or url.username or url.password or url.query or url.fragment or url.path not in ("", "/"): raise ValueError("Manager HTTPS origin only; credentials/path are not allowed") if not url.hostname or any(c.isspace() for c in manager): raise ValueError("Manager hostname required") if not all(1024 <= port <= 65535 for port in (control_port, redis_port)): raise ValueError("Ports must be 1024..65535") return dict(ai_ip=str(ai), web_ip=str(web), manager_base_url=manager.rstrip("/"), control_port=control_port, redis_port=redis_port) def plan(args): public = addresses(args.ai_ip, args.web_ip, args.manager, args.control_port, args.redis_port) network=ipaddress.ip_network(args.lan_cidr,strict=False) if network.version!=4 or network.prefixlen<8 or ipaddress.ip_address(public["web_ip"]) not in network: raise ValueError("Explicit media LAN CIDR must contain Web IP") public["artifact_lan_cidr"]=str(network) event=urlsplit(args.event_url) if (event.scheme not in ("https","http") or event.hostname!=urlsplit(public["manager_base_url"]).hostname or event.path!="/api/v1/events" or event.query or event.fragment or event.username or event.password): raise ValueError("Explicit Event endpoint on the same Manager host is required") public["event_endpoint"]=args.event_url path = active_compose() compose = json.loads(path.read_text()) expected = {"analytics", "ai-control-sidecar", "ai-gateway", "redis", "postgresql"} if set(compose["services"]) != expected: raise ValueError("Service ownership changed; no plan generated") STATE.mkdir(mode=0o700, parents=True, exist_ok=True) output = STATE / ("connection-" + time.strftime("%Y%m%dT%H%M%S") + "-" + str(os.getpid())) output.mkdir(mode=0o700) result = dict(schema=1, state="PLANNED_NOT_APPLIED", desired=public, baseline=dict(compose=str(path), compose_sha256=digest(path), unit_sha256=digest(UNIT)), preserve=["camera settings", "DB/ledger/media", "server ID", "Redis ACL passwords", "private keys"], required=["OS IP/routing assigned on physical console", "Manager CA fingerprint approval", "AI leaf IP SAN matching new IP", "login-body advertisedIp/controlPort/redis host and port", "Manager base URL + event/artifact endpoint reconciliation", "Redis bind + host allowlist", "single explicit login/enrolment", "Gateway/WSS/Redis readback"]) (output / "plan.json").write_text(json.dumps(result, ensure_ascii=False, indent=2)) print(json.dumps(dict(status=result["state"], plan=str(output / "plan.json"), **public), ensure_ascii=False, indent=2)) print("현재 서비스/인증/방화벽/OS IP는 변경하지 않았습니다. 실제 적용은 배포 완료 뒤 이 계획과 함께 검증합니다.") def verify_ca(source): source = Path(source) if not source.is_file() or source.is_symlink() or source.stat().st_size > 128*1024: raise ValueError("Public CA file invalid") raw = source.read_bytes() if b"PRIVATE KEY" in raw or raw.count(b"-----BEGIN CERTIFICATE-----") != 1: raise ValueError("One public CA certificate only") for args in (["-noout", "-text"], ["-noout", "-checkend", "300"]): result = subprocess.run(["openssl", "x509", "-in", str(source), *args], check=True, capture_output=True, timeout=10) if "-text" in args and b"CA:TRUE" not in result.stdout: raise ValueError("Not a CA") result = subprocess.run(["openssl", "x509", "-in", str(source), "-noout", "-fingerprint", "-sha256", "-dates"], check=True, capture_output=True, text=True, timeout=10) print(result.stdout) print("위 fingerprint를 Web 담당자가 별도 전달한 값과 대조하세요. 설치나 trust 변경은 하지 않았습니다.") def main(argv): parser = argparse.ArgumentParser(description="주소 변경 계획/공개 CA 확인. 자동 로그인/서비스 변경 없음") sub = parser.add_subparsers(dest="action", required=True) p = sub.add_parser("plan") for flag in ("ai-ip", "web-ip", "manager"): p.add_argument("--"+flag, required=True) p.add_argument("--lan-cidr", required=True) p.add_argument("--event-url", required=True) p.add_argument("--control-port", type=int, default=38443) p.add_argument("--redis-port", type=int, default=16379) p = sub.add_parser("check-ca") p.add_argument("source") sub.add_parser("status") p=sub.add_parser("prepare") p.add_argument("plan") p.add_argument("--ca",required=True) p.add_argument("--ca-sha256",required=True) p=sub.add_parser("enroll") p.add_argument("stage") p=sub.add_parser("apply") p.add_argument("stage") p.add_argument("--yes",action="store_true") args = parser.parse_args(argv) if args.action == "plan": plan(args) elif args.action == "check-ca": verify_ca(args.source) elif args.action == "prepare": import network_prepare network_prepare.prepare(args.plan,args.ca,args.ca_sha256) elif args.action == "enroll": import network_prepare network_prepare.enroll(args.stage) elif args.action == "apply": import network_apply network_apply.apply(args.stage,confirmed=args.yes) else: path = active_compose() print(json.dumps(dict(active_compose=str(path), sha256=digest(path), apply_performed=False)))