"""Deterministic allowlist bundles. Checksums establish integrity, not publisher identity.""" from __future__ import annotations import hashlib import gzip import io import os from pathlib import Path import re import shutil import tarfile import tempfile from .core import absolute, relative, identifier, require, KitError, sha256, canonical, parse_json, read_json, run SCHEMA = 1 MAX_FILES = 20000 MAX_TOTAL = 128 * 1024**3 MAX_META = 8 * 1024**2 DOCKER_FILES = frozenset({ 'Dockerfile', 'entrypoint.sh', 'requirements-runtime.txt', 'constraints-runtime.txt', 'requirements-tracking.txt', 'init-db.sql', 'redis.conf', 'secure-gateway-entrypoint.sh', 'secure-gateway.conf.template', }) DEPLOY_FILES = frozenset({'runtime.env.example','secure-bridge.compose.yml','secure-bridge.env.example', 'web-start.compose.yml'}) PUBLIC_DOCS = frozenset({'README.ko.md','QUICK_START.ko.md','HOST_BOOTSTRAP.ko.md','STORAGE.ko.md', 'OFFLINE_UPDATE.ko.md','ROLLBACK.ko.md','MODEL_TOOLING.ko.md','TROUBLESHOOTING.ko.md', 'ERROR_CATALOG.ko.md','ARCHITECTURE.ko.md','RELEASE_FORMAT.ko.md','FIRST_SITE.ko.md','LOCAL_PREPARATION.ko.md'}) def allowed_source(path): """Only release inputs, never all of docs/tools/configs. Extension gates also apply.""" parts = Path(path).parts if any(x.startswith('.') or x in ('__pycache__','AGENTS.md','node_modules') for x in parts): return False if path in ('deployctl','docker-compose.yml','LICENSE','LICENSE.txt','NOTICE','NOTICE.txt'): return True if len(parts) >= 2 and parts[0] in ('app','control_sidecar'): return Path(path).suffix in ('.py','.tmpl','.txt','.json') or path == 'control_sidecar/Dockerfile' if parts and parts[0] == 'deployment_kit': return Path(path).suffix in ('.py','.json','.sh','.cpp') or path=='deployment_kit/ModelTooling.Dockerfile' if len(parts) == 2 and parts[0] == 'docker': return parts[1] in DOCKER_FILES if len(parts) == 2 and parts[0] == 'deploy': return parts[1] in DEPLOY_FILES if len(parts) == 3 and parts[:2] == ('deploy','configs'): return parts[2].endswith('.example') if len(parts) == 3 and parts[:2] == ('configs','control_templates'): return parts[2] in ('medium.json','pose.json','fire.json') if len(parts) == 3 and parts[:2] == ('docs','deployment'): return parts[2] in PUBLIC_DOCS if len(parts) == 2 and parts[0] == 'tools': return parts[1] in ('secure_bridge_tls.py','secure_bridge_enroll.py','secure_bridge_provision.py') return False def destination(path): if path == 'deployctl' or path.startswith(('deployment_kit/', 'docs/deployment/')): return path if path in ('LICENSE','LICENSE.txt','NOTICE','NOTICE.txt'): return path return 'source/' + path def forbidden_payload(path): p = Path(path) return any(x in ('.git','AGENTS.md','__pycache__') for x in p.parts) or any( x in path for x in ('docs/profiles/', 'docs/deployment-review/')) def validate_manifest(m): require(isinstance(m, dict) and m.get('schema') == SCHEMA, 'BUNDLE_SCHEMA', '지원하지 않는 배포 형식입니다.') require(set(m)<= {'schema','release_id','kind','source_sha','compatibility_id','parent_release', 'files','images','app_patch','build_contract','requirements','snapshot'},'BUNDLE_FIELDS','manifest에 미지원/비공개 필드가 있습니다.') if 'snapshot' in m: s=m['snapshot'] require(isinstance(s,dict) and set(s)=={'kind','sha256','candidate_sha'} and s['kind']=='EXACT_WORKTREE' and re.fullmatch(r'[0-9a-f]{64}',s['sha256']) and re.fullmatch(r'[0-9a-f]{40}',s['candidate_sha']), 'SNAPSHOT_SPEC','미커밋 snapshot의 출처와 해시가 필요합니다.') req=m.get('requirements',{}) require(isinstance(req,dict) and set(req)<= {'driver_min','compose_min','tensorrt_version'}, 'REQUIREMENTS_FIELDS','호환성 requirement 필드만 허용합니다.') for value in req.values(): require(isinstance(value,str) and re.fullmatch(r'[0-9]+(?:\.[0-9]+){1,3}',value), 'REQUIREMENTS_VERSION','요구 버전은 검증한 숫자 버전으로 기록하세요.') identifier(m.get('release_id')) require(m.get('kind') in ('CONFIG','APP','FULL'), 'PATCH_KIND', 'CONFIG/APP/FULL 중 하나를 지정하세요.') require(re.fullmatch(r'[0-9a-f]{40}', m.get('source_sha','')), 'SOURCE_SHA', '정확한 소스 commit SHA가 필요합니다.') identifier(m.get('compatibility_id')) if m.get('parent_release'): identifier(m['parent_release']) files = m.get('files') require(isinstance(files, dict) and 0 < len(files) <= MAX_FILES, 'BUNDLE_FILES', '배포 파일 목록이 잘못됐습니다.') require(re.fullmatch(r'[0-9a-f]{64}', m.get('build_contract','')), 'BUILD_CONTRACT_MISSING', '빌드 계약 지문이 필요합니다.') total = 0 for name, spec in files.items(): relative(name) require(name not in ('manifest.json','SHA256SUMS') and not forbidden_payload(name), 'BUNDLE_PRIVATE_FILE', '개발용/개인 파일 또는 예약된 이름이 포함됐습니다.') require(isinstance(spec, dict) and set(spec) == {'sha256','size','mode'} and re.fullmatch(r'[0-9a-f]{64}', spec['sha256']) and type(spec['size']) is int and 0 <= spec['size'] <= MAX_TOTAL and spec['mode'] in (0o644,0o755), 'BUNDLE_FILE_SPEC', '파일 해시/크기/권한이 잘못됐습니다.') total += spec['size'] require(total <= MAX_TOTAL, 'BUNDLE_SIZE', '배포물 해제 크기가 상한을 초과합니다.') images = m.get('images', []) require(isinstance(images, list) and len(images) <= 12, 'IMAGE_SPEC', '이미지 목록을 확인하세요.') seen = set() for image in images: require(isinstance(image,dict) and set(image)<= {'service','id','reference','archive'}, 'IMAGE_SPEC','이미지 metadata에는 경로/식별자만 허용합니다.') identifier(image['service']) require(image['service'] not in seen, 'IMAGE_SPEC', '같은 서비스 이미지가 중복됐습니다.') seen.add(image['service']) require(re.fullmatch(r'sha256:[0-9a-f]{64}', image.get('id','')), 'IMAGE_SPEC', '이미지의 immutable ID가 필요합니다.') require(isinstance(image.get('reference'),str) and image['reference'] and not image['reference'].startswith('-') and not any(c.isspace() for c in image['reference']), 'IMAGE_SPEC', '이미지 reference가 잘못됐습니다.') if image.get('archive'): require(image['archive'] in files and image['archive'].startswith('images/'), 'IMAGE_ARCHIVE', '이미지 tar가 bundle 파일 목록에 없습니다.') if m['kind'] == 'APP': patch = m.get('app_patch', {}) require(isinstance(patch,dict) and set(patch)=={'service','base_id','base_contract'}, 'APP_PATCH_CONTRACT','APP patch의 base 계약 필드를 확인하세요.') require(patch.get('service') == 'analytics' and re.fullmatch(r'sha256:[0-9a-f]{64}', patch.get('base_id','')) and re.fullmatch(r'[0-9a-f]{64}', patch.get('base_contract','')) and m.get('parent_release'), 'APP_PATCH_CONTRACT', 'APP patch에는 이전 release와 승인된 base image/의존성 지문이 필요합니다.') require(any(x.startswith('source/app/') for x in files), 'APP_PATCH_CONTRACT', '전체 app snapshot이 필요합니다.') for name in files: require(all(parent.as_posix() not in files for parent in Path(name).parents if parent.as_posix()!='.'), 'ARCHIVE_PATH_CONFLICT','파일과 디렉터리 이름이 겹칩니다.') return m def committed_files(repo, source_sha): require(re.fullmatch(r'[0-9a-f]{40}',source_sha or ''),'SOURCE_SHA','소스 commit을 고정하세요.') _, raw = run(['git','-C',str(repo),'ls-tree','-rz',source_sha],limit=16*1024**2) result={} for line in raw.split('\0'): if not line: continue meta, name=line.split('\t',1) mode, kind, blob=meta.split() relative(name) if kind!='blob': continue result[name]=(mode,blob) require(result,'SOURCE_SHA','source commit의 파일을 읽지 못했습니다.') return result def checked_source(repo, name, record): mode, blob=record if mode == 'snapshot': p=absolute(repo/name,no_symlinks=True) require(p.is_file() and sha256(p)==blob,'SOURCE_DIRTY','승인 snapshot 이후 파일이 바뀌었습니다.') return p require(mode in ('100644','100755'),'BUNDLE_SYMLINK','배포 source의 링크/특수파일을 거부합니다.') p=absolute(repo/name,no_symlinks=True) require(p.is_file(),'SOURCE_MISSING','commit에 있는 배포 파일이 작업공간에 없습니다.') h=hashlib.sha1(('blob '+str(p.stat().st_size)+'\0').encode()) with p.open('rb') as f: for block in iter(lambda:f.read(1024*1024),b''):h.update(block) require(h.hexdigest()==blob,'SOURCE_DIRTY','배포 source가 지정 commit과 다릅니다. 사용자 변경을 덮지 않았습니다.') return p def contract(repo, tracked=None): """Fingerprint committed build inputs, not arbitrary untracked worktree files.""" repo=Path(repo) if tracked is None: _,ref=run(['git','-C',str(repo),'rev-parse','HEAD']) tracked=committed_files(repo,ref.strip()) found={} for name, record in sorted(tracked.items()): if name.startswith('native/') or name in ('docker/Dockerfile','docker/requirements-runtime.txt', 'docker/constraints-runtime.txt','docker/requirements-tracking.txt','docker/entrypoint.sh'): found[name]=sha256(checked_source(repo,name,record)) require(found,'BUILD_CONTRACT_MISSING','빌드/의존성 파일이 없는 소스입니다.') return hashlib.sha256(canonical(found)).hexdigest() def create(repo, spec_path, output): repo, output = absolute(repo, no_symlinks=True), absolute(output, no_symlinks=True) require(not output.exists() and not output.is_relative_to(repo), 'OUTPUT_EXISTS', 'bundle 출력은 repo 밖의 새 파일로 지정하세요.') spec = read_json(spec_path) require(isinstance(spec,dict) and set(spec)<= {'release_id','kind','source_sha','compatibility_id', 'parent_release','images','app_patch','requirements','assets','snapshot'},'BUNDLE_FIELDS','release spec에는 배포 필드만 넣으세요.') tracked=committed_files(repo,spec.get('source_sha')) snapshot=spec.get('snapshot') if snapshot is not None: require(isinstance(snapshot,dict) and set(snapshot)=={'files','candidate_sha'} and isinstance(snapshot['files'],dict),'SNAPSHOT_SPEC','명시적 파일별 snapshot 승인이 필요합니다.') require(all(n in snapshot['files'] for n in tracked if allowed_source(n)), 'SNAPSHOT_INCOMPLETE','기존 배포 파일을 snapshot에서 조용히 제외하지 않습니다.') for name,digest in snapshot['files'].items(): relative(name) require(allowed_source(name) and isinstance(digest,str) and re.fullmatch(r'[0-9a-f]{64}',digest), 'SNAPSHOT_SCOPE','snapshot에도 동일한 공개 파일 허용 목록을 적용합니다.') tracked[name]=('snapshot',digest) payload = {} for rel,record in sorted(tracked.items()): if allowed_source(rel): payload[destination(rel)] = checked_source(repo,rel,record) for item in spec.get('assets', []): name = relative(item['destination']) require(name.startswith(('images/','models/')) and name not in payload, 'ASSET_PATH', 'asset은 images/ 또는 models/ 아래 새 파일이어야 합니다.') p = absolute(item['source'], no_symlinks=True) require(p.is_file(), 'ASSET_MISSING', 'asset 파일이 없습니다.') require(sha256(p) == item['sha256'], 'ASSET_HASH', '승인한 asset 해시와 다릅니다.') payload[name] = p require('deployctl' in payload and 'source/docker-compose.yml' in payload, 'BUNDLE_INCOMPLETE', 'deployctl/기본 Compose가 없습니다.') m = {k:v for k,v in spec.items() if k != 'assets'} if snapshot is not None: m['snapshot']={'kind':'EXACT_WORKTREE','sha256':hashlib.sha256(canonical(snapshot['files'])).hexdigest(), 'candidate_sha':snapshot['candidate_sha']} m.update(schema=SCHEMA, build_contract=contract(repo,tracked), files={}) for name,p in payload.items(): m['files'][name] = dict(sha256=sha256(p), size=p.stat().st_size, mode=0o755 if name=='deployctl' or p.suffix=='.sh' else 0o644) validate_manifest(m) if m['kind']=='APP': require(m['app_patch']['base_contract'] == m['build_contract'], 'FULL_BUNDLE_REQUIRED', 'native/SDK/의존성/entrypoint가 달라 APP patch로 처리할 수 없습니다.') meta = canonical(m) require(len(meta) <= MAX_META, 'MANIFEST_SIZE', 'manifest가 너무 큽니다.') sums = ''.join(f'{s["sha256"]} {n}\n' for n,s in sorted(m['files'].items())).encode() output.parent.mkdir(parents=True,exist_ok=True) with output.open('xb') as raw: # FULL already contains large image layers; favor bounded preparation time. level = 1 if m['kind'] == 'FULL' else 9 with gzip.GzipFile(filename='', fileobj=raw, mode='wb', mtime=0, compresslevel=level) as gz, tarfile.open(fileobj=gz, mode='w', format=tarfile.PAX_FORMAT) as tar: for name,data in (('manifest.json',meta),('SHA256SUMS',sums)): info=tarfile.TarInfo(name); info.size=len(data); info.mode=0o644; info.mtime=0 tar.addfile(info,io.BytesIO(data)) for name,p in sorted(payload.items()): info=tarfile.TarInfo(name); info.size=m['files'][name]['size']; info.mode=m['files'][name]['mode']; info.mtime=0 with p.open('rb') as f: tar.addfile(info,f) # Re-read bytes rather than assuming files did not change during packaging. verify(output) return {'archive':str(output),'sha256':sha256(output),'release_id':m['release_id'], 'files':len(payload),'note':'archive SHA256는 USB와 별도의 신뢰 경로로 전달하세요.'} def verify(archive, *, expected_hash=None, extract_to=None): archive=absolute(archive,no_symlinks=True) require(archive.is_file(), 'BUNDLE_MISSING', '배포 압축파일이 없습니다.') if expected_hash is not None: require(re.fullmatch(r'[0-9a-f]{64}',expected_hash) and sha256(archive)==expected_hash, 'BUNDLE_HASH', '승인받은 배포물 SHA256와 다릅니다. 실행하지 않았습니다.') target=None if extract_to is not None: target=absolute(extract_to,no_symlinks=True) require(not target.exists(), 'RELEASE_EXISTS', '새 staging 폴더만 허용합니다.') try: with tarfile.open(archive,mode='r:*') as tar: first=tar.next() require(first is not None and first.name=='manifest.json' and first.isfile() and first.size<=MAX_META, 'MANIFEST_FIRST', '첫 파일은 크기가 제한된 manifest.json이어야 합니다.') raw=tar.extractfile(first).read(MAX_META+1) m=validate_manifest(parse_json(raw)) expected_sums=''.join(f'{s["sha256"]} {n}\n' for n,s in sorted(m['files'].items())).encode() seen={'manifest.json'}; total=0 if target is not None: target.mkdir(parents=True,mode=0o700) (target/'manifest.json').write_bytes(raw) for member in tar: # tar iteration may return the cached first member; skip exactly that object. if member is first: continue name=relative(member.name) require(name not in seen and member.isfile() and not member.issym() and not member.islnk(), 'ARCHIVE_ENTRY', '중복/링크/특수파일은 배포물에서 허용하지 않습니다.') seen.add(name) require(len(seen)<=MAX_FILES+2,'BUNDLE_FILES','배포 파일 수 상한을 초과했습니다.') if name=='SHA256SUMS': require(member.size==len(expected_sums),'CHECKSUM_LIST','체크섬 목록 크기가 다릅니다.') data=tar.extractfile(member).read(len(expected_sums)+1) require(data==expected_sums,'CHECKSUM_LIST','manifest와 checksum 목록이 다릅니다.') if target is not None: (target/name).write_bytes(data) continue require(name in m['files'],'UNLISTED_FILE','manifest에 없는 파일이 들어 있습니다.') entry=m['files'][name] require(member.size==entry['size'],'FILE_SIZE','파일 크기가 manifest와 다릅니다.') total+=member.size require(total<=MAX_TOTAL,'BUNDLE_SIZE','압축 해제 크기 상한을 초과했습니다.') src=tar.extractfile(member); h=hashlib.sha256() dest=None if target is not None: path=target/name; path.parent.mkdir(parents=True,exist_ok=True) dest=path.open('xb') try: while True: block=src.read(1024*1024) if not block: break h.update(block) if dest: dest.write(block) finally: if dest: dest.close() require(h.hexdigest()==entry['sha256'],'FILE_HASH','배포 파일 checksum이 다릅니다.') if target is not None: os.chmod(target/name,entry['mode']) require(seen==set(m['files'])|{'manifest.json','SHA256SUMS'},'BUNDLE_INCOMPLETE','배포 파일이 누락됐습니다.') except (tarfile.TarError, OSError, EOFError) as exc: raise KitError('ARCHIVE_INVALID','압축파일을 읽거나 해제하지 못했습니다. staging은 보존합니다.') from exc return m def verify_directory(root): root=absolute(root,no_symlinks=True) m=validate_manifest(read_json(root/'manifest.json',MAX_META)) actual=set() for path in root.rglob('*'): require(not path.is_symlink(),'RELEASE_TAMPERED','release 안에 링크가 있습니다.') if path.is_file(): actual.add(path.relative_to(root).as_posix()) require(actual==set(m['files'])|{'manifest.json','SHA256SUMS'},'RELEASE_TAMPERED','release의 파일 목록이 달라졌습니다.') sums=''.join(f'{v["sha256"]} {n}\n' for n,v in sorted(m['files'].items())).encode() require((root/'SHA256SUMS').read_bytes()==sums,'CHECKSUM_LIST','checksum 목록이 달라졌습니다.') for name,s in m['files'].items(): p=absolute(root/name,no_symlinks=True) require(p.is_file() and p.stat().st_size==s['size'] and sha256(p)==s['sha256'] and (p.stat().st_mode & 0o7777)==s['mode'], 'RELEASE_TAMPERED','release 파일이 바뀌었습니다. 그대로 적용하지 않습니다.') return m