"""Korean standalone entrypoint. No installed ct command or internet prerequisite.""" import argparse import json import os from pathlib import Path import shutil import sys from ..core import KitError, absolute, approved, canonical, read_json, require, sha256, write_json from . import package from .record import Record, attempt_record def trusted_directory(path): p=absolute(path,no_symlinks=True) require(all(x.stat().st_uid==0 and not x.stat().st_mode & 0o022 for x in (p,*p.parents)), 'STAGE_OWNERSHIP','서버 staging과 상위 폴더는 root 소유이고 다른 계정이 변경할 수 없어야 합니다.') def log_failure(record, code, diagnostic, attempt=False): if record is None:return try: if attempt:record.step('BLOCKED',error_code=code,command_diagnostic=diagnostic) elif (record.root/'journal.json').exists(): r=record.read();r.update(error_code=code,command_diagnostic=diagnostic);record.write(r) except (OSError,ValueError,KitError): print('[LOG_WRITE_FAILED] 기존 로컬 기록을 보존했습니다. 디스크/권한 확인이 필요합니다.',file=sys.stderr) def inventory(): from ..core import docker p=Path('/etc/contextra-maintenance/server.json') result={'definition_present':p.is_file(),'backend':'unknown','scope':'READ_ONLY'} if p.is_file(): d=read_json(p) result.update(server=d.get('server'),backend=d.get('backend'),definition_sha256=sha256(p)) _,raw=docker('ps','-a','--filter','label=com.docker.compose.project='+d['project'], '--format','{{.Label "com.docker.compose.service"}} {{.Image}} {{.State}}') result['services']=raw.splitlines() else: legacy=Path('/workspace/contextra/runtime/state/active.json') result['legacy_active_present']=legacy.is_file() if legacy.is_file(): d=read_json(legacy) result['legacy_active_sha256']=sha256(legacy) result['legacy_transaction']=d.get('transaction') result['apply_supported']=False launcher=Path('/usr/local/bin/ct') result['ct_launcher_present']=launcher.is_file() if launcher.is_file() and not launcher.is_symlink():result['ct_launcher_sha256']=sha256(launcher) return result def stage(src, dst, digest): m=package.verify(src,digest) dst=absolute(dst,no_symlinks=True) require(str(dst).startswith(('/data/','/workspace/','/var/lib/contextra/')), 'STAGE_PATH','서버 로컬 데이터 경로에 새 폴더를 지정하세요. USB 자체는 사용하지 않습니다.') require(not dst.exists(),'OUTPUT_EXISTS','기존 staging을 덮어쓰지 않습니다.') ancestor=dst.parent while not ancestor.exists():ancestor=ancestor.parent trusted_directory(ancestor) required=sum(v['size'] for v in m['files'].values())+1024**3 require(shutil.disk_usage(ancestor).free>required,'STAGE_SPACE','서버 복사 공간이 부족합니다.') dst.parent.mkdir(parents=True,exist_ok=True) shutil.copytree(src,dst) dst.chmod(0o700) package.verify(dst,digest) return {'stage':str(dst),'next':'이 폴더의 ct-offline을 사용하세요. USB 원본은 보존됐습니다.'} def parser(): p=argparse.ArgumentParser(description=__doc__) p.add_argument('action',nargs='?',choices=('inventory','verify','stage','prepare','status','apply','resume','rollback','export','split','join')) p.add_argument('--package',type=Path,default=Path(__file__).resolve().parents[2]) p.add_argument('--sha256');p.add_argument('--destination',type=Path);p.add_argument('--file',type=Path) p.add_argument('--record',type=Path,help='패키지 없이 반출할 로컬 작업 기록 폴더') return p def main(argv=None): a=parser().parse_args(argv);record=None;attempt=None try: attempt=attempt_record() print('이번 명령의 로컬 기록:',attempt.root) if a.action is None: print('Contextra 오프라인 배포\n1 서버 정보(읽기) 2 패키지 검사 3 서버로 복사\n4 준비 기록 5 상태 6 적용 7 중단 후 재개 8 이전 정상 버전 복귀 9 로그 반출\n0 종료') choices={'1':'inventory','2':'verify','3':'stage','4':'prepare','5':'status','6':'apply','7':'resume','8':'rollback','9':'export'} n=input('번호: ').strip() if n=='0':attempt.step('CANCELLED');return 0 require(n in choices,'MENU_SELECTION','표시된 번호를 선택하세요.') a.action=choices[n] attempt.step('SELECTED',action=a.action) if a.action=='export': if a.record is None:a.record=Path(input('반출할 로컬 기록 폴더(상태 안내에 표시된 경로): ').strip()) if a.destination is None:a.destination=Path(input('반출할 기존 폴더 절대경로: ').strip()) require(any(a.record.is_relative_to(root) for root in (Path('/var/lib/contextra-maintenance'),Path.home()/'.local/state/contextra-offline')), 'RECORD_SCOPE','ct-offline이 만든 로컬 기록만 반출합니다.') if approved(True,phrase='EXPORT'):result=str(Record(a.record).export(a.destination)) elif a.action=='inventory':result=inventory() elif a.action in ('split','join'): from . import transport require(a.file is not None and a.destination is not None,'TRANSPORT_PATH','--file/--destination을 지정하세요.') result=transport.split(a.file,a.destination) if a.action=='split' else transport.join(a.file,a.destination,a.sha256) else: if not a.sha256:a.sha256=input('별도로 전달받은 manifest SHA256: ').strip() m=package.verify(a.package,a.sha256) attempt.step('PACKAGE_VERIFIED',related_release_id=m['release_id'],package_sha256=a.sha256) if a.action=='verify':result={'package':'PASS','release':m['release_id'],'runtime':'NOT_CHANGED'} elif a.action=='stage': require(os.geteuid()==0,'SUDO_REQUIRED','서버 staging은 sudo로 수행하세요.') if a.destination is None:a.destination=Path(input('서버 로컬의 새 staging 절대경로: ').strip()) result=stage(a.package,a.destination,a.sha256) else: from .host import Host from .workflow import Workflow require(os.geteuid()==0,'SUDO_REQUIRED','설치 기록과 적용은 sudo로 수행하세요.') host=Host();root=host.fp.STATE/'offline'/m['release_id'];record=Record(root) flow=Workflow(host,a.package,m,a.sha256,root) if a.action=='status':result=dict(record.public(),scope='RECORDED_STATE_NOT_LIVE_HEALTH') else: trusted_directory(a.package) require(str(a.package).startswith(('/data/','/workspace/','/var/lib/contextra/')) and a.package.stat().st_uid==0 and not a.package.stat().st_mode & 0o077, 'LOCAL_STAGE_REQUIRED','서버의 root 전용 staging에서 수행하세요. USB 원본에서 직접 적용하지 않습니다.') print('대상:',host.s['server'],'패키지:',m['release_id'],'로컬 기록:',root) print('prepare는 기록만 생성합니다. apply/resume/rollback은 정지된 서비스의 image와 설정을 바꾸고 기동합니다.') print('네트워크/SSH/DB/영상/인증정보를 초기화하지 않습니다. 채널 Start는 호출하지 않습니다.') if approved(True,phrase='PREPARE' if a.action=='prepare' else 'RESTORE' if a.action=='rollback' else 'APPLY'): with host.lock(): if a.action=='prepare':flow.prepare();result=record.public() elif a.action=='rollback':flow.rollback();result=record.public() else:flow.apply(resume=a.action=='resume');result=record.public() attempt.step('DONE') print(json.dumps(result,ensure_ascii=False,indent=2,default=str));return 0 except KitError as exc: log_failure(attempt,exc.code,exc.diagnostic,attempt=True) # Keep the last durable phase; a timeout may have changed Docker state. log_failure(record,exc.code,exc.diagnostic) print(exc.explain(),file=sys.stderr);return 2 except (OSError,ValueError,KeyError,TypeError) as exc: log_failure(attempt,'INPUT_OR_ENVIRONMENT_ERROR',{'type':type(exc).__name__},attempt=True) log_failure(record,'INPUT_OR_ENVIRONMENT_ERROR',{'type':type(exc).__name__}) print('[BLOCKED] INPUT_OR_ENVIRONMENT_ERROR ('+type(exc).__name__+')\n원문에는 비밀값이 있을 수 있어 출력하지 않습니다. 기존 기록을 보존하고 담당자에게 공유하세요.',file=sys.stderr) return 2 except (KeyboardInterrupt,EOFError): log_failure(attempt,'INTERRUPTED',{},attempt=True) print('중단됐습니다. 실제 상태와 로컬 기록을 먼저 확인하세요. 자동 재시도/복귀하지 않습니다.',file=sys.stderr);return 130 if __name__=='__main__':raise SystemExit(main())