"""Stage new address/CA material without mutating active settings or logging in.""" import copy import hashlib import json import os from pathlib import Path import shutil import subprocess import sys from urllib.parse import urlsplit import connection import server_context as server from field_ui import print RUNTIME = server.path('runtime', '/workspace/contextra/runtime') DATA = server.path('data', '/workspace/contextra/data') VENDOR = Path(__file__).resolve().parent / "bridge_tools" def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def rewrite(doc, wanted, bridge, tls, session): doc = copy.deepcopy(doc) services = doc["services"] for name, cfg in services.items(): for mount in cfg.get("volumes",[]): if mount.get("type") != "bind": continue target = mount["target"] if target == "/run/bridge": mount["source"] = str(bridge) elif target.startswith("/run/bridge/"): mount["source"] = str(bridge / Path(target).name) elif target == "/run/ai-tls": mount["source"] = str(tls) elif target == "/run/bridge-session": mount["source"] = str(session) services["ai-control-sidecar"]["environment"]["AI_CONTROL_MANAGER_BASE_URL"] = wanted["manager_base_url"] services["ai-control-sidecar"]["environment"]["AI_CONTROL_EVENT_ENDPOINT"] = wanted["event_endpoint"] services["analytics"]["environment"]["CONTEXTRA_ARTIFACT_PUBLIC_BASE"] = f'https://{wanted["ai_ip"]}:{wanted["control_port"]}/artifacts' services["ai-gateway"]["environment"]["AI_CONTROL_HOST_PORT"] = str(wanted["control_port"]) services["ai-gateway"]["environment"]["AI_ARTIFACT_LAN_CIDR"] = wanted["artifact_lan_cidr"] services["ai-gateway"]["ports"][0]["published"] = str(wanted["control_port"]) services["ai-gateway"]["ports"][0]["target"] = wanted["control_port"] services["redis"]["ports"][0].update(host_ip=wanted["ai_ip"],published=str(wanted["redis_port"])) return doc def prepare(plan_file, ca_file, expected_ca_sha256, *, credentials=None, server_id=None, connection_version=None): if os.geteuid() != 0: raise ValueError("sudo required") plan_file = Path(plan_file) if any(p.is_symlink() for p in (plan_file,*plan_file.parents)) or not plan_file.is_relative_to(connection.STATE): raise ValueError("Use a ct connection plan") plan = json.loads(plan_file.read_text()) wanted = plan["desired"] connection.addresses(wanted["ai_ip"],wanted["web_ip"],wanted["manager_base_url"],wanted["control_port"],wanted["redis_port"]) if server.portable(): import compose_backend old_path = compose_backend.current_path() compose_backend.verify_plan(plan) else: old_path = connection.active_compose() if sha(old_path) != plan["baseline"]["compose_sha256"] or sha(connection.UNIT) != plan["baseline"]["unit_sha256"]: raise ValueError("Deployment changed since plan") ca = Path(ca_file) if sha(ca) != expected_ca_sha256: raise ValueError("Approved CA file SHA256 mismatch") sys.path.insert(0,str(VENDOR)) from tools.secure_bridge_tls import prepare as tls_prepare from tools.secure_bridge_provision import validate_ca validate_ca(ca) active = json.loads((RUNTIME/"state/active.json").read_text()) doc = json.loads(old_path.read_text()) name = plan_file.parent.name stage = DATA/"site"/name if stage.exists(): raise ValueError("Existing staged network material preserved") stage.mkdir(mode=0o700) bridge,tls,session = (stage/x for x in ("bridge","tls","bridge-session")) old_bridge = Path(next(x["source"] for x in doc["services"]["ai-control-sidecar"]["volumes"] if x["target"]=="/run/bridge")) old_tls = Path(next(x["source"] for x in doc["services"]["ai-gateway"]["volumes"] if x["target"]=="/run/ai-tls")) for directory in (bridge,tls,session): directory.mkdir(mode=0o700) os.chown(directory,10001,10001) for filename in ("password","control-token","relay-token"): original = old_bridge/filename if original.is_symlink(): raise ValueError("Private input symlink") shutil.copyfile(original,bridge/filename) (bridge/filename).chmod(0o600) os.chown(bridge/filename,10001,10001) shutil.copyfile(ca,bridge/"manager-ca.crt") (bridge/"manager-ca.crt").chmod(0o644) body = json.loads((old_bridge/"login-body.json").read_text()) old_body = copy.deepcopy(body) if credentials is not None: from auth_control import login_id body['loginId'] = login_id(credentials['loginId']) password = credentials['password'] if not isinstance(password, str) or not password or '\x00' in password or len(password.encode()) > 8192: raise ValueError('Invalid profile password') (bridge/'password').write_text(password+'\n') body.update(advertisedIp=wanted["ai_ip"],controlPort=wanted["control_port"],aiControlCa=ca.read_text()) body["redis"].update(host=wanted["ai_ip"],port=wanted["redis_port"]) import connection_versions as versions previous_env = doc['services']['ai-control-sidecar']['environment'] identity = versions.context(wanted['manager_base_url'], server_id or previous_env.get('AI_CONTROL_SERVER_ID')) same_peer = (previous_env.get('AI_CONTROL_MANAGER_BASE_URL','').rstrip('/') == identity['manager'] and previous_env.get('AI_CONTROL_SERVER_ID') == identity['server_id']) baseline = old_body if same_peer and 'connectionConfigVersion' in old_body else None body['connectionConfigVersion'] = versions.choose(identity, body, connection_version, baseline) (bridge/"login-body.json").write_text(json.dumps(body)) (bridge/"login-body.json").chmod(0o600) os.chown(bridge/"login-body.json",10001,10001) shutil.copyfile(old_tls/"control-tls.key",tls/"control-tls.key") (tls/"control-tls.key").chmod(0o600) tls_prepare(tls,wanted["ai_ip"]) for p in tls.iterdir(): os.chown(p,10001,10001) new_doc = rewrite(doc,wanted,bridge,tls,session) if server_id is not None: if not isinstance(server_id,str) or not server_id.strip() or len(server_id)>256: raise ValueError('Invalid profile server ID') for service in ('ai-control-sidecar','analytics'): new_doc['services'][service]['environment']['AI_CONTROL_SERVER_ID']=server_id env = dict(new_doc["services"]["ai-control-sidecar"]["environment"]) for key,name in (("AI_CONTROL_MANAGER_CA_FILE","manager-ca.crt"),("AI_CONTROL_LOGIN_BODY_FILE","login-body.json"), ("AI_CONTROL_PASSWORD_FILE","password"),("CONTEXTRA_CONTROL_TOKEN_FILE","control-token"), ("CONTEXTRA_RELAY_TOKEN_FILE","relay-token")): env[key] = str(bridge/name) env["AI_CONTROL_SESSION_FILE"] = str(session/Path(env["AI_CONTROL_SESSION_FILE"]).name) env["AI_CONTROL_CERTIFICATE_FILE"] = str(tls/"control-tls.crt") env.update(AI_BRIDGE_SECRET_ROOT=str(bridge),AI_BRIDGE_TLS_ROOT=str(tls),AI_CONTROL_HOST_PORT=str(wanted["control_port"]), CONTEXTRA_ARTIFACT_PUBLIC_BASE=new_doc["services"]["analytics"]["environment"]["CONTEXTRA_ARTIFACT_PUBLIC_BASE"]) (stage/"host-enrollment.json").write_text(json.dumps(env)) (stage/"compose.proposed.json").write_text(json.dumps(new_doc)) record=dict(status="PREPARED_NOT_APPLIED",stage=str(stage),desired=wanted,base_active_sha256=sha(RUNTIME/"state/active.json"), parent_transaction=active["transaction"],parent_compose=str(old_path),old_ai_ip=doc["services"]["redis"]["ports"][0].get("host_ip"), old_redis_port=doc["services"]["redis"]["ports"][0]["published"], connection_identity=identity,connection_version=body['connectionConfigVersion']) if not server.portable():record['script_sha256']=sha(DATA/'site/redis-access.sh') else:record['input_stamp']=plan['baseline']['input_stamp'] (stage/"record.json").write_text(json.dumps(record,indent=2)) for p in stage.glob("*.json"):p.chmod(0o600) from field_profiles import SETUP, account outbox=(SETUP if server.CURRENT else DATA.parent/'setup')/'outbox'/(stage.name+".csr") shutil.copyfile(tls/"control-tls.csr",outbox) user=account();os.chown(outbox,user.pw_uid,user.pw_gid) outbox.chmod(0o644) versions.remember(identity,stage,body,'prepared') print('Prepared connectionConfigVersion:',body['connectionConfigVersion'],'Web current version: UNKNOWN') print(json.dumps(dict(status=record["status"],stage=str(stage),csr=str(outbox)))) return stage def enroll(stage): stage = Path(stage) if not stage.is_relative_to(DATA/"site") or stage.is_symlink(): raise ValueError("Invalid staged directory") record=json.loads((stage/"record.json").read_text()) if sha(RUNTIME/"state/active.json")!=record["base_active_sha256"]: raise ValueError("Active deployment changed since prepare") if server.portable(): import compose_backend compose_backend.verify_enrollment(record) running=subprocess.run(["docker","ps","--filter","label=com.docker.compose.project="+server.project(),"--format","{{.ID}}"], check=True,capture_output=True,text=True,timeout=10).stdout.strip() if running:raise ValueError("ct stop before enrolment; existing sessions must not be disturbed") env=dict(os.environ,**json.loads((stage/"host-enrollment.json").read_text())) print("명시적 로그인/CSR 요청을1회 수행합니다. 결과가 불명확하면 자동 반복하지 않습니다.",flush=True) result_path=stage/'enroll-result.json' previous=json.loads(result_path.read_text()) if result_path.exists() else {} from field_profiles import atomic attempt=int(previous.get('attempt_count',0))+1 legacy=previous.get('prior_attempts_unknown',bool(previous) and 'attempt_count' not in previous) atomic(result_path,{'result':'REQUEST_INTENT_OUTCOME_UNKNOWN','attempt_count':attempt,'prior_attempts_unknown':legacy}) from maintenance_process import run p=run([sys.executable,str(VENDOR/"tools/secure_bridge_enroll.py"),"--directory",str(stage/"tls"), "--ip",record["desired"]["ai_ip"],"--journal",str(stage/"enroll-intent.json")],env=env,timeout=45,capture_output=True,text=True) summary={'returncode':p.returncode,'result':'CERTIFICATE_VERIFIED' if p.returncode==0 else 'UNCONFIRMED', 'attempt_count':attempt,'prior_attempts_unknown':legacy} for line in p.stdout[-65536:].splitlines(): try: item=json.loads(line) if isinstance(item,dict): code=item.get('error') if isinstance(code,str) and __import__('re').fullmatch('[A-Z0-9_]{1,80}',code):summary['error']=code if type(item.get('status')) is int:summary['http_status']=item['status'] for key in ('phase','remote_code','remote_field','remote_reason'): val=item.get(key) if isinstance(val,str) and __import__('re').fullmatch('[A-Za-z0-9_.]{1,80}',val):summary[key]=val except ValueError:pass from field_profiles import atomic atomic(stage/'enroll-result.json',summary) print(json.dumps(summary)) for directory in (stage/"bridge-session",stage/"tls"): for path in directory.iterdir(): if path.is_file():os.chown(path,10001,10001) if p.returncode:raise RuntimeError("Enrolment not confirmed; inspect intent before any retry") shutil.copyfile(stage/"tls/control-tls.crt",stage/"bridge/control-tls.crt") (stage/"bridge/control-tls.crt").chmod(0o644) os.chown(stage/"bridge/control-tls.crt",10001,10001) if record.get('connection_identity'): import connection_versions connection_versions.remember(record['connection_identity'],stage, json.loads((stage/'bridge/login-body.json').read_text()),'successful')