"""FIELD-AB-03: accept supported config changes without weakening code/storage trust.""" import base64 import copy import hashlib import json import os from pathlib import Path import shutil import subprocess import sys import time import connection import field_profiles as profiles import field_network import server_context as server from field_ui import print, input TOOLS = server.path('tools', profiles.RUNTIME / 'deploy-tools/provider-20260927-r01') ENV_KEYS = { 'analytics': {'CONTEXTRA_ARTIFACT_PUBLIC_BASE','AI_CONTROL_SERVER_ID'}, 'ai-control-sidecar': {'AI_CONTROL_MANAGER_BASE_URL','AI_CONTROL_EVENT_ENDPOINT','AI_CONTROL_SERVER_ID'}, 'ai-gateway': {'AI_CONTROL_HOST_PORT','AI_ARTIFACT_LAN_CIDR'}, } def sha(path): return hashlib.sha256(profiles.safe_path(path).read_bytes()).hexdigest() def run(args, timeout=30): from deployment_kit.compose_snapshot import command_environment from maintenance_process import run as execute return execute(args,check=True,capture_output=True,text=True,timeout=timeout,env=command_environment()).stdout def libraries(): if str(TOOLS) not in sys.path: sys.path.insert(0,str(TOOLS)) def masked(doc): from apply_source import normalize d=normalize(doc) for name,keys in ENV_KEYS.items(): env=d['services'][name].setdefault('environment',{}) for k in keys:env.pop(k,None) for name in ('redis','ai-gateway'): for item in d['services'][name].get('ports',[]): for key in ('published','host_ip'):item.pop(key,None) if name=='ai-gateway':item.pop('target',None) return d def check_drift(base, candidate): if masked(base) != masked(candidate): raise ValueError('지원 범위 밖의 Compose 변경입니다. image/mount/command/권한/DB는 긴급모드에서도 자동 승인하지 않습니다.') def remember_baseline(): state=profiles.RUNTIME/'state' a=profiles.read_json(state/'active.json',1048576) s=state/'site.json' if sha(s)!=a['site_sha256']:raise ValueError('기존 정상 site 지문이 아닙니다.') profiles.atomic(profiles.STATE/'accepted-sites'/a['site_sha256'],s.read_bytes()) def inspect(): if server.portable(): import compose_backend return compose_backend.inspect() libraries() from deployment_kit import site,release,compose_snapshot state=profiles.RUNTIME/'state';a=profiles.read_json(state/'active.json',1048576) inflight=profiles.read_json(state/'inflight.json') if inflight.get('status')!='DONE' or inflight.get('transaction')!=a['transaction']: raise ValueError('기존 Runtime 적용 미완료. ct config resume 또는 ct config restore로 확인하세요.') cp=connection.active_compose() if cp.parent.name!=a['transaction']:raise ValueError('서비스와 active transaction이 다릅니다.') normalized=cp.with_name('compose.normalized.json') if sha(normalized)!=a['normalized_sha256']:raise ValueError('정상 Compose 기준 손상. 무조건 SHA 재등록하지 않습니다.') base=profiles.read_json(normalized,8*1024**2) live=profiles.read_json(cp,8*1024**2);check_drift(base,live) value=site.load(state/'site.json') if sha(state/'site.json')!=a['site_sha256']: saved=profiles.STATE/'accepted-sites'/a['site_sha256'] if not saved.exists() or sha(saved)!=a['site_sha256']:raise ValueError('site 변경 비교 기준이 없습니다. root 백업을 확인하세요.') old=profiles.read_json(saved) one=copy.deepcopy(old);two=copy.deepcopy(value);one.pop('ai_ip');two.pop('ai_ip') if one!=two:raise ValueError('site의 경로/디스크/서비스 변경은 지원 범위 밖입니다.') args,_=release.compose_args(value,profiles.RUNTIME/'releases'/a['release_id']) rendered=json.loads(run(['docker',*args,'config','--format','json'])) check_drift(base,rendered) source=a.get('offline_source') if source: from source_prepare import inspect_tree if inspect_tree(Path(source['snapshot']))!=source['files']:raise ValueError('승인된 코드 snapshot이 변경됐습니다. ct apply 경로로 처리하세요.') # Reads only; bound model/source files and volumes are not silently re-authorized. compose_snapshot.verify_bindings(a.get('bound_files',{})) return a,value,cp,base,live,rendered def public_values(doc): s=doc['services'];side=s['ai-control-sidecar'].get('environment',{}) gateway=s['ai-gateway'].get('environment',{}) redis=s['redis']['ports'][0] gateway_port=s['ai-gateway']['ports'][0] control=int(gateway.get('AI_CONTROL_HOST_PORT',gateway_port['published'])) if control!=int(gateway_port['published']) or control!=int(gateway_port['target']): raise ValueError('Gateway env/published/target 포트가 서로 다릅니다. ct profile edit에서 사용할 포트 하나를 지정하세요.') from urllib.parse import urlsplit base=side['AI_CONTROL_MANAGER_BASE_URL'] return dict(ai_ip=redis.get('host_ip'),manager=base,web_host=urlsplit(base).hostname, event_url=side.get('AI_CONTROL_EVENT_ENDPOINT',base+'/api/v1/events'), control_port=control,redis_port=int(redis['published']), artifact_lan_cidr=gateway.get('AI_ARTIFACT_LAN_CIDR'),server_id=side.get('AI_CONTROL_SERVER_ID')) def import_changes(p,source): from ipaddress import IPv4Interface a,site,cp,base,live,rendered=inspect() if source not in ('compose','persistent','site'):raise ValueError('compose/persistent/site 중 선택하세요.') result=copy.deepcopy(p) if source=='site': prefix=IPv4Interface(result['network']['address']).network.prefixlen if result['network']['address'] else 24 result['network'].update(mode='static',address=site['ai_ip']+'/'+str(prefix)) else: values=public_values(live if source=='compose' else rendered) prefix=IPv4Interface(result['network']['address']).network.prefixlen if result['network']['address'] else 24 result['network'].update(mode='static',address=profiles.ipv4(values['ai_ip'])+'/'+str(prefix)) result.update(control_port=values['control_port'],redis_port=values['redis_port'],artifact_lan_cidr=values['artifact_lan_cidr']) w=profiles.web(result) w.update(manager=values['manager'],host=profiles.ipv4(values['web_host']),event_url=values['event_url'],server_id=values['server_id']) if w['host'] not in result['redis_allow']:result['redis_allow'].append(w['host']) return profiles.validate(result) def prepare(p): if server.portable(): import compose_backend return compose_backend.prepare(p) import network_prepare a,site,cp,base,live,rendered=inspect() pending=profiles.STATE/'pending.json' if pending.exists():raise ValueError('진행 중 준비가 있습니다. ct config status/resume/cancel을 사용하세요.') credentials=profiles.read_credentials(profiles.secret_path(p)) ca=profiles.ca_path(p) connection.verify_ca(ca) wanted=connection.addresses(field_network.ai_ip(p),profiles.web(p)['host'],profiles.web(p)['manager'],p['control_port'],p['redis_port']) wanted.update(event_endpoint=profiles.web(p)['event_url'],artifact_lan_cidr=p['artifact_lan_cidr'],redis_allow=p['redis_allow']) name='field-'+time.strftime('%Y%m%dT%H%M%S')+'-'+str(os.getpid()) directory=profiles.STATE/name directory.mkdir(parents=True,mode=0o700) plan=dict(desired=wanted,baseline=dict(compose=str(cp),compose_sha256=sha(cp),unit_sha256=sha(connection.UNIT))) profiles.atomic(directory/'plan.json',plan) # Credentials are explicit, profile-specific input; never fall back to a different Manager account. stage=network_prepare.prepare(directory/'plan.json',ca,sha(ca),credentials=credentials,server_id=profiles.web(p)['server_id'], connection_version=profiles.web(p).get('connection_version')) record=profiles.read_json(stage/'record.json') record.update(profile=p['id'],profile_stamp=profiles.stamp(p),profile_snapshot=p,redis_allow=p['redis_allow'], profile_credentials_sha256=sha(profiles.secret_path(p)),profile_ca_sha256=sha(ca)) profiles.atomic(stage/'record.json',record) record['prepared_files']={name:sha(stage/name) for name in ( 'compose.proposed.json','host-enrollment.json','bridge/login-body.json','bridge/password', 'bridge/manager-ca.crt','bridge/control-token','bridge/relay-token','tls/control-tls.key')} profiles.atomic(profiles.STATE/'stages'/(stage.name+'.json'),record) profiles.atomic(pending,{'stage':str(stage),'profile':p['id'],'phase':'PREPARED'}) print('프로필 준비 완료. 아직 로그인/Runtime 적용하지 않았습니다. 다음: ct config connect') return stage def pending_stage(): record=profiles.read_json(profiles.STATE/'pending.json') stage=profiles.safe_path(record['stage']) if stage.parent != server.path('data',profiles.RUNTIME.parent/'data')/'site' or not stage.name.startswith('field-'): raise ValueError('준비 경로 오류') return stage def verify_pending(stage): r=profiles.read_json(profiles.STATE/'stages'/(stage.name+'.json')) disk=profiles.read_json(stage/'record.json') for key in ('profile','profile_stamp','desired','base_active_sha256','profile_snapshot'): if disk.get(key)!=r.get(key):raise ValueError('준비본의 기준이 변경됐습니다. 원격 요청을 하지 않습니다.') for name,expected in r['prepared_files'].items(): if sha(stage/name)!=expected:raise ValueError('준비 이후 파일 변경: '+name) p=profiles.load(r['profile']) if profiles.stamp(p)!=r['profile_stamp'] or sha(profiles.secret_path(p))!=r['profile_credentials_sha256'] or sha(profiles.ca_path(p))!=r['profile_ca_sha256']: raise ValueError('준비 후 프로필/계정/CA가 변경됐습니다. 미발급이면 cancel 후 다시 prepare하세요.') return r,p def stage_ready(stage): import network_prepare sys.path.insert(0,str(network_prepare.VENDOR)) from tools.secure_bridge_tls import verify from control_sidecar.config import Config record=profiles.read_json(stage/'record.json') verify(stage/'tls/control-tls.key',stage/'tls/control-tls.crt',stage/'bridge/manager-ca.crt',record['desired']['ai_ip']) Config.from_env(profiles.read_json(stage/'host-enrollment.json')) if (stage/'tls/control-tls.crt').read_bytes()!=(stage/'bridge/control-tls.crt').read_bytes(): raise ValueError('Gateway/Sidecar 인증서가 다릅니다.') def export_receipt(stage,phase): record=profiles.read_json(profiles.STATE/'stages'/(stage.name+'.json')) out=profiles.SETUP/'outbox'/('field-change-'+stage.name.removeprefix('field-')) out.mkdir(mode=0o750,parents=True,exist_ok=True) if os.geteuid()==0: u=profiles.account();os.chown(out,u.pw_uid,u.pw_gid) verification={'phase':phase,'runtime_started':False,'web_acceptance':'USER_CHECK_PENDING','profile':record['profile'], 'credentials':'configured; values excluded','ca_file_sha256':record['profile_ca_sha256']} summary=('# 현장 설정 변경\n\n- 환경: '+record['profile_snapshot']['label']+'\n- 단계: '+phase+ '\n- 실제 Web 표시·카메라 수용: 사용자 확인 대기\n- 복구: ct config restore\n- 원장·미디어·비밀번호/token/private key는 반출하지 않습니다.\n') profiles.atomic(out/'SUMMARY.ko.md',summary.encode(),user=True) profiles.atomic(out/'changes.redacted.json',{'before':record.get('public_before'), 'after':record['profile_snapshot'],'credential_values':'EXCLUDED'},user=True) profiles.atomic(out/'verification.json',verification,user=True) print('반출 자료:',out,' ZIP: ct config export') return out def replay_writes(journal): # Root-owned journal is written only after scope validation. Each write is idempotent and CAS checked. expected={str(profiles.RUNTIME/'state'/n):0o600 for n in ('site.json','active.json','inflight.json')} expected[str(connection.UNIT)]=0o644 expected[str(profiles.RUNTIME.parent/'data/site/redis-access.sh')]=0o700 if len(journal['writes'])!=len(expected) or {x['path'] for x in journal['writes']}!=set(expected): raise ValueError('적용 journal의 경로 범위가 다릅니다.') for item in journal['writes']: if item['mode']!=expected[item['path']]:raise ValueError('적용 권한 범위 오류') path=profiles.safe_path(item['path']) before=base64.b64decode(item['before']) if item['before'] is not None else None if (path.read_bytes() if path.exists() else None) not in (before,base64.b64decode(item['after'])): raise ValueError('적용 중 다른 수정 발견: '+str(path)) for item in journal['writes']: path=profiles.safe_path(item['path']) after=base64.b64decode(item['after']);before=base64.b64decode(item['before']) if item['before'] is not None else None current=path.read_bytes() if path.exists() else None if current==after:continue if current!=before:raise ValueError('적용 중 다른 수정 발견: '+str(path)) profiles.atomic(path,after,mode=item['mode']) def prepare_transaction(tx, stage, active_transaction): intent=profiles.STATE/'transaction-intents'/(stage.name+'.json') expected={'stage':str(stage),'transaction':str(tx),'parent':active_transaction} if tx.exists(): if not intent.exists() or profiles.read_json(intent)!=expected or tx.name==active_transaction: raise ValueError('기존 transaction은 자동 교체하지 않습니다.') profiles.safe_path(tx) saved=tx.with_name(tx.name+'-incomplete-'+str(time.time_ns())) tx.rename(saved) profiles.atomic(intent,expected) tx.mkdir(mode=0o700) def network_overlay(doc): # Compose identifies ports by IP/target/published/protocol; a new IP otherwise appends. # JSON flow values preserve types/escaping without adding a YAML dependency. lines = [] for key, value in doc.items(): if key == 'name': continue if key != 'services': lines.append(json.dumps(key) + ': ' + json.dumps(value)) continue lines.append('services:') for name, service in value.items(): lines.append(' ' + json.dumps(name) + ':') for field, item in service.items(): tag = '!override ' if field == 'ports' and name in ('redis', 'ai-gateway') else '' lines.append(' ' + json.dumps(field) + ': ' + tag + json.dumps(item)) return ('\n'.join(lines) + '\n').encode() def apply_stage(stage): if server.portable(): import compose_backend return compose_backend.apply_stage(stage) import network_apply libraries() from deployment_kit import site,release,compose_snapshot from deployment_kit.core import lock from apply_source import normalize stage=profiles.safe_path(stage) record,p=verify_pending(stage) if shutil.disk_usage(profiles.RUNTIME).free < 15*1024**3:raise ValueError('root 여유15GiB 필요') journal_path=profiles.STATE/'apply-pending.json' state=profiles.RUNTIME/'state' with lock(state): if run(['docker','ps','--filter','label=com.docker.compose.project=contextra-icerti','--format','{{.ID}}']).strip(): raise ValueError('ct stop으로 정상 정지한 뒤 적용하세요.') if not journal_path.exists(): a,current,cp,base,live,rendered=inspect() if sha(state/'active.json')!=record['base_active_sha256']:raise ValueError('준비 후 배포가 바뀌었습니다.') if sha(profiles.RUNTIME.parent/'data/site/redis-access.sh')!=record['script_sha256']: raise ValueError('준비 후 Redis 접근 정책이 변경됐습니다.') stage_ready(stage) from deployment_kit.site_guard import verify verify(state/'site.json') if record['desired']['ai_ip'] not in field_network.addresses(p['network']['interface']):raise ValueError('프로필 AI 주소가 NIC에 없습니다. ct net status/apply 확인') doc=profiles.read_json(stage/'compose.proposed.json',8*1024**2) for service in ('analytics','ai-control-sidecar','ai-gateway','redis'): doc['services'][service].setdefault('labels',{})[release.TRANSACTION_LABEL]=stage.name new=copy.deepcopy(current);new['ai_ip']=record['desired']['ai_ip'] release.validate_rendered(doc,new,a['images']) tx=state/'transactions'/stage.name prepare_transaction(tx,stage,a['transaction']) compose_snapshot.seal(doc,tx/'compose.json',current['project']) overlay=stage/'network.compose.yaml' profiles.atomic(overlay,network_overlay(doc)) old=a.get('offline_network',{}).get('overlay') if old: if new['compose_files'][-1]!=old:raise ValueError('새 overlay가 뒤에 있어 검토가 필요합니다.') new['compose_files'][-1]=str(overlay) else:new['compose_files'].append(str(overlay)) site.validate(new) args,inputs=release.compose_args(new,profiles.RUNTIME/'releases'/a['release_id']) if normalize(json.loads(run(['docker',*args,'config','--format','json'])))!=normalize(doc):raise ValueError('최종 Compose 재해석 불일치') unit=connection.UNIT.read_bytes();needle=('/transactions/'+a['transaction']+'/').encode() if unit.count(needle)!=3:raise ValueError('알 수 없는 서비스 실행 경로') next_unit=unit.replace(needle,('/transactions/'+stage.name+'/').encode()) firewall=network_apply.firewall_script(new['ai_ip'],p['redis_allow'],p['redis_port']).encode() updated=copy.deepcopy(a);updated.update(transaction=stage.name,compose_sha256=sha(tx/'compose.json'),normalized_sha256=sha(tx/'compose.normalized.json'), site_sha256=hashlib.sha256((json.dumps(new,ensure_ascii=False,indent=2)+'\n').encode()).hexdigest(),config_inputs=inputs,bound_files=compose_snapshot.file_bindings(doc)) updated['offline_network']={'stage':str(stage),'overlay':str(overlay),'parent_transaction':a['transaction'],'desired':record['desired']} updated.setdefault('acceptance',{})['OFFLINE_NETWORK']='PENDING_START_AND_WEB_CHECK' records=[(profiles.RUNTIME.parent/'data/site/redis-access.sh',firewall,0o700),(state/'site.json',new,0o600), (connection.UNIT,next_unit,0o644),(state/'active.json',updated,0o600), (state/'inflight.json',{'status':'DONE','stage':'FIELD_APPLIED_STOPPED','transaction':stage.name},0o600)] writes=[] for path,data,mode in records: before=path.read_bytes() if path.exists() else None raw=data if isinstance(data,bytes) else (json.dumps(data,ensure_ascii=False,indent=2)+'\n').encode() writes.append({'path':str(path),'before':base64.b64encode(before).decode() if before is not None else None,'after':base64.b64encode(raw).decode(),'mode':mode}) journal={'stage':str(stage),'transaction':stage.name,'phase':'APPLYING','writes':writes} profiles.atomic(journal_path,journal) record['public_before']=public_values(base) profiles.atomic(profiles.STATE/'stages'/(stage.name+'.json'),record) profiles.atomic(tx/'accepted.json',updated) else: journal=profiles.read_json(journal_path,8*1024**2) if journal['stage']!=str(stage):raise ValueError('다른 적용 작업이 진행 중입니다.') stage_ready(stage) replay_writes(journal) run([sys.executable,'-c',network_apply.firewall_script(record['desired']['ai_ip'],p['redis_allow'],p['redis_port'])]) run(['systemctl','daemon-reload']) remember_baseline() profiles.atomic(profiles.STATE/'last-applied.json',{'profile':p['id'],'stamp':profiles.stamp(p),'stage':str(stage),'verification':'USER_CHECK_PENDING'}) profiles.atomic(profiles.STATE/'selected.json',{'profile':p['id']}) journal['phase']='APPLIED';profiles.atomic(profiles.STATE/'history'/(stage.name+'.json'),journal) journal_path.unlink() (profiles.STATE/'pending.json').unlink(missing_ok=True) export_receipt(stage,'APPLIED_SERVICES_STOPPED') print('설정 반영 완료·재부팅 유지. 서비스/카메라 자동 Start 없음. 다음: ct start') def restore_config(): if server.portable(): import compose_backend return compose_backend.restore_config() libraries() from deployment_kit.core import lock with lock(profiles.RUNTIME/'state'): _restore_config() def _restore_config(): pending=profiles.STATE/'apply-pending.json' if pending.exists():journal=profiles.read_json(pending,8*1024**2) else: last=profiles.read_json(profiles.STATE/'last-applied.json') journal=profiles.read_json(profiles.STATE/'history'/(Path(last['stage']).name+'.json'),8*1024**2) libraries() if run(['docker','ps','--filter','label=com.docker.compose.project=contextra-icerti','--format','{{.ID}}']).strip():raise ValueError('ct stop 후 복구하세요.') reversed_journal=copy.deepcopy(journal) # Credentials/session/media are never part of these writes. for row in reversed_journal['writes']: if row['before'] is None:raise ValueError('이전 파일 없는 복구는 수동 확인이 필요합니다.') row['before'],row['after']=row['after'],row['before'] for row in reversed_journal['writes']: path=profiles.safe_path(row['path']);raw=path.read_bytes() if raw not in (base64.b64decode(row['before']),base64.b64decode(row['after'])):raise ValueError('이후 변경 발견, 복구 중단: '+str(path)) replay_writes(reversed_journal) previous_script=next(base64.b64decode(x['after']).decode() for x in reversed_journal['writes'] if x['path'].endswith('/redis-access.sh')) run([sys.executable,'-c',previous_script]);run(['systemctl','daemon-reload']) pending.unlink(missing_ok=True) (profiles.STATE/'pending.json').unlink(missing_ok=True) profiles.atomic(profiles.STATE/'last-restored.json',{'stage':journal['stage'],'verification':'USER_CHECK_PENDING'}) (profiles.STATE/'last-applied.json').unlink(missing_ok=True) remember_baseline() print('이전 로컬 설정 복구. 이전 세션의 원격 유효성은 복구되지 않습니다. 자동 Start 없음.')