"""Small, explicit safety primitives. All host changes have a separate apply gate.""" from __future__ import annotations import contextlib import fcntl import hashlib import json import os from pathlib import Path, PurePosixPath import re import signal import subprocess import tempfile import time class KitError(RuntimeError): def __init__(self, code, message, action='deployctl doctor', diagnostic=None): self.code, self.message, self.action = code, message, action self.diagnostic = diagnostic or {} super().__init__(message) def explain(self): return (f'[BLOCKED] {self.code}\n쉽게 말하면: {self.message}\n' f'다음 행동: {self.action}\n안내: docs/deployment/ERROR_CATALOG.ko.md\n' '실패 후 자동 재시도·데이터 삭제·자동 복원은 하지 않습니다.') def require(condition, code, message, action='deployctl doctor'): if not condition: raise KitError(code, message, action) def identifier(value): require(isinstance(value, str) and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,79}', value) and value not in ('.', '..'), 'INVALID_ID', '이름은 영문/숫자/._- 80자 이내여야 합니다.') return value def absolute(value, *, no_symlinks=False): p = Path(value) require(p.is_absolute() and p != Path('/') and '..' not in p.parts and not any(ord(c) < 32 for c in str(p)), 'UNSAFE_PATH', '절대경로와 전용 하위 폴더를 사용하세요.') if no_symlinks: require(not any(x.is_symlink() for x in (p, *p.parents)), 'PATH_SYMLINK', '실제 저장/상태 경로에는 심볼릭 링크를 사용하지 마세요.') return p def relative(value): require(isinstance(value, str) and value and '\\' not in value and not any(ord(c) < 32 for c in value), 'ARCHIVE_PATH', '배포물에 잘못된 경로가 있습니다.') p = PurePosixPath(value) require(not p.is_absolute() and all(x not in ('', '.', '..') for x in value.split('/')), 'ARCHIVE_PATH', '배포 폴더 밖으로 나가는 경로를 거부했습니다.') return p.as_posix() def sha256(path): h = hashlib.sha256() with Path(path).open('rb') as f: for block in iter(lambda: f.read(1024 * 1024), b''): h.update(block) return h.hexdigest() def canonical(value): return (json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False) + '\n').encode() def _pairs(items): result = {} for k, v in items: require(k not in result, 'DUPLICATE_JSON_KEY', '같은 설정 키가 두 번 있습니다.') result[k] = v return result def parse_json(raw): try: return json.loads(raw, object_pairs_hook=_pairs, parse_constant=lambda _: (_ for _ in ()).throw(ValueError())) except (ValueError, UnicodeError, RecursionError) as exc: raise KitError('INVALID_JSON', 'JSON 형식을 확인하세요.') from exc def read_json(path, limit=4 * 1024 * 1024): p = absolute(path, no_symlinks=True) require(p.is_file() and p.stat().st_size <= limit, 'CONFIG_MISSING', '설정 파일이 없거나 너무 큽니다.') return parse_json(p.read_bytes()) def atomic_write(path, data, *, expected=None, new=False, mode=0o600): p = absolute(path, no_symlinks=True) require(not (new and p.exists()), 'FILE_EXISTS', '기존 파일을 덮어쓰지 않았습니다.') if expected is not None: require(p.is_file() and sha256(p) == expected, 'STATE_CHANGED', '확인 후 파일이 바뀌었습니다. 다시 계획하세요.') p.parent.mkdir(parents=True, exist_ok=True, mode=0o700) fd, name = tempfile.mkstemp(prefix='.' + p.name + '.', dir=p.parent) try: with os.fdopen(fd, 'wb') as f: os.fchmod(f.fileno(), mode) f.write(data if isinstance(data, bytes) else data.encode()) f.flush(); os.fsync(f.fileno()) if new: os.link(name, p) os.unlink(name) else: os.replace(name, p) dfd = os.open(p.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(dfd) finally: os.close(dfd) finally: if os.path.exists(name): os.unlink(name) def write_json(path, data, **kwargs): atomic_write(path, canonical(data), **kwargs) @contextlib.contextmanager def lock(directory): d = absolute(directory, no_symlinks=True) d.mkdir(parents=True, exist_ok=True, mode=0o700) fd = os.open(d / '.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise KitError('DEPLOY_BUSY', '다른 배포가 진행 중입니다. 강제로 잠금을 지우지 마세요.') from exc yield finally: os.close(fd) def approved(apply, yes=False, phrase='APPLY'): if not apply: return False require(os.geteuid() == 0, 'SUDO_REQUIRED', '변경 작업은 sudo로 다시 실행하세요.') if not yes: require(input(f'변경하려면 {phrase} 입력: ').strip() == phrase, 'CANCELLED', '사용자가 적용을 취소했습니다.') return True def command_diagnostic(exit_code, stderr): # Only fixed classifications and digests escape; arbitrary subprocess text may contain secrets. lowered = stderr.lower() causes = {'NO_SPACE': ('no space left', 'disk full'), 'PERMISSION': ('permission denied', 'access denied'), 'NOT_FOUND': ('no such file', 'not found'), 'CONNECTION': ('connection refused', 'connection timed out'), 'TLS': ('certificate verify failed', 'unknown authority')} return {'exit_code': exit_code, 'cause': next((k for k, terms in causes.items() if any(t in lowered for t in terms)), 'UNCLASSIFIED'), 'stderr_sha256': hashlib.sha256(stderr.encode()).hexdigest()} def run(argv, *, timeout=30, limit=1024 * 1024, check=True, environment=None, merge_output=False): """No shell; bounded output/time. Never include captured command output in public errors.""" require(isinstance(argv, list) and argv and all(isinstance(x, str) for x in argv), 'COMMAND_INVALID', '명령 형식이 잘못됐습니다.') with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err: try: proc = subprocess.Popen(argv, stdout=out, stderr=subprocess.STDOUT if merge_output else err, start_new_session=True, env=dict(os.environ if environment is None else environment, LC_ALL='C')) except OSError as exc: raise KitError('COMMAND_UNAVAILABLE', f'{Path(argv[0]).name} 명령을 실행할 수 없습니다.') from exc end = time.monotonic() + timeout try: while proc.poll() is None: if time.monotonic() >= end or sum(os.fstat(f.fileno()).st_size for f in (out, err)) > limit: raise KitError('COMMAND_LIMIT', '명령 시간/출력 상한에 도달했습니다. 실제 상태를 확인하세요.') time.sleep(.05) finally: if proc.poll() is None: with contextlib.suppress(ProcessLookupError): os.killpg(proc.pid, signal.SIGTERM) try: proc.wait(timeout=2) except subprocess.TimeoutExpired: with contextlib.suppress(ProcessLookupError): os.killpg(proc.pid, signal.SIGKILL) proc.wait(timeout=2) require(sum(os.fstat(f.fileno()).st_size for f in (out, err)) <= limit, 'COMMAND_LIMIT', '명령 출력 상한을 초과했습니다.') out.seek(0); text = out.read(limit).decode('utf-8', errors='replace') err.seek(0); error = err.read(limit).decode('utf-8', errors='replace') if check and proc.returncode != 0: raise KitError('COMMAND_FAILED', f'{Path(argv[0]).name} 실행 실패(exit={proc.returncode}). 비공개 원장과 현장 가이드를 확인하세요.', diagnostic=command_diagnostic(proc.returncode, error)) # Non-checking callers inspect the status first (e.g. helper-not-found cleanup). return proc.returncode, text if proc.returncode == 0 else error or text def docker(*args, **kwargs): return run(['docker', '--host', 'unix:///var/run/docker.sock', *args], **kwargs) def image_id(reference): _, raw = docker('image', 'inspect', '--format', '{{.Id}}', reference) value = raw.strip() require(re.fullmatch(r'sha256:[0-9a-f]{64}', value), 'IMAGE_ID_INVALID', '로컬 이미지 식별자를 확인할 수 없습니다.') return value def now(): import datetime return datetime.datetime.now(datetime.timezone.utc).isoformat()