"""OFFLINE-FULL-01: explicit apply/resume/rollback with before/after file CAS.""" import copy import json import os from pathlib import Path import shutil import stat from ..core import absolute, canonical, read_json, require, sha256, write_json, atomic_write from .record import Record def private_once(path, data): if path.exists(): require(path.read_bytes()==data,'STAGE_CONFLICT','보존된 준비 파일과 현재 내용이 다릅니다.') else:atomic_write(path,data,new=True) def validate_entries(entries): for e in entries: p=absolute(e['path'],no_symlinks=True) current=sha256(p) if p.exists() else None require(current in (e['before_sha256'],e['after_sha256']),'FILE_CONFLICT','다른 작업의 변경을 보존했습니다.') if p.exists(): st=p.stat() require((st.st_uid,st.st_gid,stat.S_IMODE(st.st_mode))==(e['uid'],e['gid'],e['mode']) and not os.listxattr(p), 'FILE_METADATA_CONFLICT','파일 권한/소유권/확장 속성이 바뀌었습니다. 자동 덮어쓰기하지 않습니다.') for side in ('before','after'): if e[side+'_sha256'] is not None: require(sha256(absolute(e[side+'_file'],no_symlinks=True))==e[side+'_sha256'], 'BACKUP_CHANGED','준비/복귀 파일 지문이 다릅니다.') def commit_file(entry, reverse=False): path = absolute(entry['path'], no_symlinks=True) before, after = entry['before_sha256'], entry['after_sha256'] desired, accepted = (before, after) if reverse else (after, before) actual = sha256(path) if path.exists() else None if actual == desired: return require(actual == accepted, 'FILE_CONFLICT', '계획 이후 파일이 바뀌었습니다. 사용자 변경을 덮지 않습니다.') if desired is None: path.unlink() fd=os.open(path.parent,os.O_RDONLY|os.O_DIRECTORY) try:os.fsync(fd) finally:os.close(fd) return src=absolute(entry['before_file' if reverse else 'after_file'],no_symlinks=True) require(sha256(src)==desired, 'BACKUP_CHANGED', '보존/후보 파일 지문이 다릅니다.') atomic_write(path, src.read_bytes(), expected=actual, new=actual is None, mode=entry['mode']) os.chown(path, entry['uid'], entry['gid']) def bindings_after(records, entries): result=copy.deepcopy(records) for e in entries: if e['before_sha256']==e['after_sha256']:continue path=Path(e['path']) for parent, value in result.items(): if parent==str(path): require(value==e['before_sha256'],'BINDING_CONFLICT','설정 보호 지문이 다릅니다.') result[parent]=e['after_sha256'] elif isinstance(value,dict) and path.is_relative_to(parent): rel=str(path.relative_to(parent)) require(value['directory'].get(rel)==e['before_sha256'],'BINDING_CONFLICT','설정 보호 항목이 다릅니다.') value['directory'][rel]=e['after_sha256'] return result class Workflow: def __init__(self, host, package, manifest, digest, root): self.host, self.package, self.m, self.digest = host, Path(package), manifest, digest self.record=Record(root) def target(self, snapshot): t=self.m['target'] require(all(snapshot[k]==t[k] for k in ('server','backend','definition_sha256')), 'WRONG_TARGET', '이 패키지의 승인 대상 서버가 아닙니다.') def space(self): require(shutil.disk_usage('/').free>=15*1024**3,'ROOT_RESERVE','root 여유15GiB를 확보하세요. 자동 정리는 하지 않습니다.') def prepare(self): require(not (self.record.root/'journal.json').exists(),'EXISTING_RUN','기존 기록이 있습니다. 상태/재개/복귀를 선택하세요.') self.space(); self.host.pending_clear() s=self.host.snapshot(); self.target(s) require(s['images']==self.m['before_images'],'BASE_IMAGE','기반 image가 다릅니다. 맞는 패키지를 준비하세요.') self.record.write(dict(release_id=self.m['release_id'],package_sha256=self.digest,phase='PREPARED',history=[], baseline=s,images_before=s['images'],images_after=self.m['after_images'], server=s['server'],backend=s['backend'],tool_source_commit=self.m['source_commit'])) def existing(self): r=self.record.read() require(r['release_id']==self.m['release_id'] and r['package_sha256']==self.digest,'RUN_PACKAGE','다른 패키지의 작업 기록입니다.') return r def base_cas(self, r): s=self.host.snapshot(); self.target(s) require(s==r['baseline'],'BASE_CHANGED','준비 후 기준이 바뀌었습니다. 기존 기록을 보존하고 다시 검토하세요.') def entry(self, path, data, index): path=absolute(path,no_symlinks=True) require(not path.exists() or path.is_file(),'TARGET_TYPE','일반 파일만 변경합니다.') private=self.record.root/'private';private.mkdir(mode=0o700,exist_ok=True) before=private/(str(index)+'.before');after=private/(str(index)+'.after') exists=path.exists();info=path.stat() if exists else None require(not exists or not os.listxattr(path),'FILE_METADATA_UNSUPPORTED','확장 ACL/속성 파일은 별도 검토가 필요합니다.') if exists:private_once(before,path.read_bytes()) private_once(after,data) return dict(path=str(path),before_sha256=sha256(before) if exists else None,after_sha256=sha256(after), before_file=str(before),after_file=str(after),mode=stat.S_IMODE(info.st_mode) if info else 0o644, uid=info.st_uid if info else 0,gid=info.st_gid if info else 0) def stage(self, r): private=self.record.root/'private';private.mkdir(mode=0o700,exist_ok=True) paths,policy=self.host.settings_paths();settings=[] for p in paths: name='input-'+p['kind'];private_once(private/name,Path(p['path']).read_bytes()) settings.append(dict(p,before_file=name)) private_once(private/'settings-inputs.json',canonical(settings)) self.host.stage_settings(private,self.m['after_images']['analytics'],self.package) entries=[self.entry(p['path'],(private/('after-'+p['kind'])).read_bytes(),i) for i,p in enumerate(settings)] if policy.exists(): require(read_json(policy)=={'version':1,'hold_seconds':3},'DISPLAY_POLICY_CONFLICT','기존 표시 정책이3초와 다릅니다. 자동 변경하지 않습니다.') else:entries.append(self.entry(policy,canonical({'version':1,'hold_seconds':3}),len(entries))) bindings=self.host.state/'protected-bindings.json' entries.append(self.entry(bindings,canonical(bindings_after(read_json(bindings,8*1024**2),entries)),len(entries))) tx,docs,active=self.host.transaction(self.m['release_id'],self.m['after_images']['analytics']) require(not tx.exists(),'TRANSACTION_EXISTS','기존 transaction을 덮어쓰지 않습니다.') proposal=private/'compose.proposed.json';private_once(proposal,canonical(docs['compose.physical.json'])) self.host.validate_transaction(proposal,docs['compose.json']) for name,value in docs.items():entries.append(self.entry(tx/name,canonical(value),len(entries))) entries.append(self.entry(self.host.state/'active.json',canonical(active),len(entries))) self.record.step('STAGED', entries=entries, candidate_active=active, settings_result=read_json(private/'settings-result.json')) def marker(self): path=self.host.fp.STATE/'apply-pending.json' expected={'owner':'ct-offline','release_id':self.m['release_id'],'journal':str(self.record.root/'journal.json')} if path.exists():require(read_json(path)==expected,'OTHER_PENDING','다른 유지보수 작업이 진행 중입니다.') else:write_json(path,expected,new=True) return path,expected def clear_marker(self): path=self.host.fp.STATE/'apply-pending.json' expected={'owner':'ct-offline','release_id':self.m['release_id'],'journal':str(self.record.root/'journal.json')} require(read_json(path)==expected,'OTHER_PENDING','유지보수 잠금 소유권이 바뀌었습니다.') path.unlink() def apply(self, resume=False): r=self.existing();self.space() if r['phase']=='VERIFIED': s=self.host.snapshot();self.target(s) require(s['images']==self.m['after_images'],'COMPLETED_STATE_DRIFT','완료 후 실행 image가 바뀌었습니다.') require(all(Path(e['path']).is_file() and sha256(e['path'])==e['after_sha256'] for e in r['entries']), 'COMPLETED_STATE_DRIFT','완료 후 설정이 바뀌었습니다. 기존 완료 기록은 보존합니다.') return r require(r['phase'] not in ('ROLLED_BACK','ROLLBACK_INTENT'),'ROLLED_BACK','복귀한 작업은 자동 재적용하지 않습니다. 새 작업ID를 준비하세요.') require(resume or r['phase']=='PREPARED','RESUME_REQUIRED','중단 기록을 먼저 읽고 명시적으로 재개하세요.') self.host.stopped() if 'entries' not in r: self.base_cas(r);self.host.pending_clear() self.record.step('LOAD_INTENT') self.host.load_images(self.package,self.m['archives'],self.m['after_images']) self.record.step('IMAGE_READY') self.stage(r);r=self.existing() else: # Only files owned by this transaction may be partially applied. changed={e['path'] for e in r['entries']} require(all(sha256(p)==v for p,v in r['baseline']['files'].items() if p not in changed), 'PROTECTED_DRIFT','배포 범위 밖의 기준이 바뀌었습니다.') validate_entries(r['entries']) self.marker();self.record.step('SWITCH_INTENT') for e in r['entries']:commit_file(e) self.record.step('FILES_APPLIED') self.host.create_containers();self.record.step('CONTAINERS_CREATED') self.record.step('START_INTENT') readiness=self.host.start_and_verify(self.m['after_images']) self.clear_marker();self.record.step('VERIFIED',readiness=readiness,error_code=None,command_diagnostic={}) return self.existing() def rollback(self): r=self.existing();self.space() if r['phase']=='ROLLED_BACK': self.base_cas(r) return r require('entries' in r,'NO_RUNTIME_CHANGE','실행 설정을 바꾸기 전입니다. image 반입/로컬 준비만 했습니다.') self.host.stopped() changed={e['path'] for e in r['entries']} require(all(sha256(p)==v for p,v in r['baseline']['files'].items() if p not in changed), 'PROTECTED_DRIFT','복귀 범위 밖의 기준이 바뀌었습니다.') validate_entries(r['entries']) from ..core import docker for image in self.m['before_images'].values():docker('image','inspect','--format','{{.Id}}',image) self.marker();self.record.step('ROLLBACK_INTENT') for e in reversed(r['entries']):commit_file(e,reverse=True) self.host.create_containers() readiness=self.host.start_and_verify(self.m['before_images']) self.clear_marker();self.record.step('ROLLED_BACK',readiness=readiness,rollback_verified=True,error_code=None,command_diagnostic={}) return self.existing()