diff --git a/.getzilla/getzilla/_policy_legacy.py b/.getzilla/getzilla/_policy_legacy.py index 9ee103de..7208ba99 100644 --- a/.getzilla/getzilla/_policy_legacy.py +++ b/.getzilla/getzilla/_policy_legacy.py @@ -13,7 +13,7 @@ import time from dataclasses import dataclass from pathlib import Path, PureWindowsPath -from typing import Any +from typing import Any, Callable from . import fsx from .state import active_write_agents, get_active_route, has_valid_approval @@ -2994,6 +2994,18 @@ def _same_file(root: Path, word: str, rel: str) -> bool: return os.path.isfile(target) and os.path.realpath(os.path.join(str(root), word)) == target +def consumer_secret_paths(consumers: object, *, script_matches: Callable[[str], bool]) -> frozenset[str]: + """Select decoded registration metadata using the caller's identity predicate; no I/O.""" + if not isinstance(consumers, list): + return frozenset() + allowed: set[str] = set() + for item in consumers: + if (isinstance(item, dict) and isinstance(item.get('script'), str) and isinstance(item.get('secrets'), list) + and script_matches(item['script'])): + allowed.update(secret for secret in item['secrets'] if isinstance(secret, str) and secret) + return frozenset(allowed) + + def _consumer_handoffs(root: Path, command: str) -> dict[tuple[str, ...], frozenset[str]]: """Secret paths the owner lets an exact script receive (project policy ``secret_consumers``). @@ -3037,11 +3049,7 @@ def _consumer_handoffs(root: Path, command: str) -> dict[tuple[str, ...], frozen first = _executable_name(raw[0]) interpreter = _CONSUMER_INTERPRETER.fullmatch(first) is not None script = raw[1] if interpreter and len(raw) > 1 else raw[0] - allowed: set[str] = set() - for item in consumers: - if (isinstance(item, dict) and isinstance(item.get('script'), str) and isinstance(item.get('secrets'), list) - and _same_file(root, script, item['script'])): - allowed.update(secret for secret in item['secrets'] if isinstance(secret, str) and secret) + allowed = consumer_secret_paths(consumers, script_matches=lambda listed: _same_file(root, script, listed)) if not allowed: continue sinks = {*redirected, *written} diff --git a/.getzilla/getzilla/deploy_preflight.py b/.getzilla/getzilla/deploy_preflight.py new file mode 100644 index 00000000..c378a32d --- /dev/null +++ b/.getzilla/getzilla/deploy_preflight.py @@ -0,0 +1,511 @@ +"""Bounded local deployment observations. Nothing in this module grants execution.""" +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import stat +import time +import unicodedata +from dataclasses import dataclass +from pathlib import Path + +from . import fsx +from ._policy_legacy import DEFAULT_SECRET_READ, _secret_path_match, consumer_secret_paths +from .component_state import BoundRoot, ComponentStateError, FileSnapshot, path_parts +from .source_admission import sensitive_class + +SCHEMA_REVISION = 'getzilla-deploy-preflight-v1' +ALGORITHM_REVISION = 'getzilla-deploy-preflight-algorithm-v1' +POLICY_PATH = '.getzilla/config/project-policy.json' +MAX_INPUT_BYTES = 65536 +MAX_ARTIFACTS = 128 +MAX_FILE_BYTES = 16 * 1024 * 1024 +MAX_TOTAL_BYTES = 64 * 1024 * 1024 +TTL_SECONDS = 300 +WORK_SECONDS = 10 +_MAX_EPOCH = 253402300799 +DIAGNOSTICS = frozenset({ + 'INPUT_INVALID', 'INPUT_TOO_LARGE', 'PATH_UNSAFE', 'FILE_MISSING', 'FILE_UNSAFE', + 'FILE_CHANGED', 'HASH_MISMATCH', 'MODE_MISMATCH', 'MODE_UNSUPPORTED', 'POLICY_MISSING', + 'POLICY_INVALID', 'CONSUMER_UNREGISTERED', 'BUDGET_EXCEEDED', 'ROOT_CHANGED', + 'ALGORITHM_CHANGED', 'CLOCK_UNCERTAIN', 'OBSERVATION_EXPIRED', 'OBSERVATION_CHANGED', + 'PROVENANCE_UNVERIFIED', 'RECOVERY_UNVERIFIED', 'INTERNAL_ERROR', +}) +_IDENTIFIER = re.compile(r'[A-Za-z0-9][A-Za-z0-9._-]{0,63}') +_SHA256 = re.compile(r'[0-9a-f]{64}') +Revision = tuple[str, str | None, str | None, str | None] + + +class PreflightError(ValueError): + """Only fixed diagnostics cross this boundary, never underlying exceptions.""" + + def __init__(self, code: str) -> None: + self.code = code if type(code) is str and code in DIAGNOSTICS else 'INTERNAL_ERROR' + super().__init__(self.code) + + +@dataclass(frozen=True, repr=False) +class FileExpectation: + path: str + sha256: str + mode: int + revision: Revision | None = None + + +@dataclass(frozen=True, repr=False) +class DeployManifest: + project_id: str + operation: str + operator: FileExpectation + artifacts: tuple[FileExpectation, ...] + artifact_revision: Revision + target: tuple[str, str, str] + credential_selector: str + recovery: tuple[str, str, str] + + +def _fail(code: str = 'INPUT_INVALID') -> None: + raise PreflightError(code) from None + + +def _clean_text(value: object, maximum: int) -> bool: + return (type(value) is str and 0 < len(value) <= maximum + and not any(ord(c) < 32 or 127 <= ord(c) <= 159 or 0xD800 <= ord(c) <= 0xDFFF for c in value)) + + +def _object(value: object, keys: set[str]) -> dict: + if type(value) is not dict or value.keys() != keys: + _fail() + return value + + +def _identifier(value: object) -> str: + if type(value) is not str or _IDENTIFIER.fullmatch(value) is None: + _fail() + return value + + +def _sha256(value: object) -> str: + if type(value) is not str or _SHA256.fullmatch(value) is None: + _fail() + return value + + +def _strict_json(raw: bytes, *, invalid: str = 'INPUT_INVALID', allow_floats: bool = False) -> object: + if type(raw) is not bytes: + _fail(invalid) + if len(raw) > MAX_INPUT_BYTES: + _fail('INPUT_TOO_LARGE') + def unique(pairs): + result = {} + for key, value in pairs: + if key in result: + _fail(invalid) + result[key] = value + return result + def invalid_number(_value): + _fail(invalid) + def finite_float(value): + parsed = float(value) + if not math.isfinite(parsed): + _fail(invalid) + return parsed + try: + text = raw.decode('utf-8') + # Limit nesting before json.loads recurses, ignoring quoted/escaped braces. + depth, quoted, escaped = 0, False, False + for char in text: + if quoted: + if escaped: + escaped = False + elif char == '\\': + escaped = True + elif char == '"': + quoted = False + elif char == '"': + quoted = True + elif char in '{[': + depth += 1 + if depth > 8: + _fail(invalid) + elif char in '}]': + depth -= 1 + return json.loads(text, object_pairs_hook=unique, parse_float=finite_float if allow_floats else invalid_number, + parse_constant=invalid_number) + except (UnicodeError, ValueError, RecursionError): + _fail(invalid) + + +def _public_path(value: object) -> str: + if not _clean_text(value, 1024): + _fail('PATH_UNSAFE') + try: + parts = path_parts(value) + except ComponentStateError: + _fail('PATH_UNSAFE') + if (len(parts) > 32 or any(part.casefold() == '.git' for part in parts) + or any(char in value for char in '*?"<>|') + or any(sensitive_class('/'.join(parts[:i]), allow_examples=False) + or _secret_path_match('/'.join(parts[:i]), DEFAULT_SECRET_READ) + for i in range(1, len(parts) + 1))): + _fail('PATH_UNSAFE') + return value + + +def _overlap(first: str, second: str) -> bool: + a, b = _fold(first.split('/')), _fold(second.split('/')) + return a[:len(b)] == b or b[:len(a)] == a + + +def _distinct(paths: tuple[str, ...]) -> None: + for index, path in enumerate(paths): + if any(_overlap(path, previous) for previous in paths[:index]): + _fail('PATH_UNSAFE') + + +def _revision(value: object) -> Revision: + data = _object(value, {'label', 'git'}) + label = _identifier(data['label']) + if data['git'] is None: + return (label, None, None, None) + git = _object(data['git'], {'object_format', 'commit_oid', 'tree_oid'}) + fmt = git['object_format'] + if type(fmt) is not str or fmt not in {'sha1', 'sha256'}: + _fail() + size = 40 if fmt == 'sha1' else 64 + for key in ('commit_oid', 'tree_oid'): + if type(git[key]) is not str or re.fullmatch('[0-9a-f]{' + str(size) + '}', git[key]) is None: + _fail() + return (label, fmt, git['commit_oid'], git['tree_oid']) + + +def _file(value: object, *, operator: bool = False) -> FileExpectation: + keys = {'path', 'sha256', 'mode'} | ({'revision'} if operator else set()) + data = _object(value, keys) + if type(data['mode']) is not int or not 0 <= data['mode'] <= 511: + _fail() + return FileExpectation(_public_path(data['path']), _sha256(data['sha256']), data['mode'], + _revision(data['revision']) if operator else None) + + +def parse_manifest(raw: bytes) -> DeployManifest: + """Parse bounded declarations only; no filesystem, environment or policy access.""" + data = _object(_strict_json(raw), {'schema_version', 'project_id', 'operation', 'operator', 'artifacts', + 'artifact_revision', 'target', 'credential_selector', 'recovery'}) + if data['schema_version'] != SCHEMA_REVISION or data['operation'] != 'dev-delivery': + _fail() + if type(data['artifacts']) is not list or not 1 <= len(data['artifacts']) <= MAX_ARTIFACTS: + _fail() + operator = _file(data['operator'], operator=True) + artifacts = tuple(_file(value) for value in data['artifacts']) + _distinct((POLICY_PATH, operator.path, *(item.path for item in artifacts))) + target = _object(data['target'], {'identity', 'destination_id', 'transport'}) + if type(target['transport']) is not str or target['transport'] not in {'ssh-sftp', 'local-fixture'}: + _fail() + recovery = _object(data['recovery'], {'protocol_id', 'version', 'descriptor_sha256'}) + if not _clean_text(data['credential_selector'], 4096): + _fail() + return DeployManifest( + _identifier(data['project_id']), 'dev-delivery', operator, artifacts, _revision(data['artifact_revision']), + (_identifier(target['identity']), _identifier(target['destination_id']), target['transport']), + data['credential_selector'], + (_identifier(recovery['protocol_id']), _identifier(recovery['version']), _sha256(recovery['descriptor_sha256'])), + ) + + +def _fold(parts) -> tuple[str, ...]: + return tuple(unicodedata.normalize('NFC', part).casefold() for part in parts) + + +def _digest(value: object) -> str: + return hashlib.sha256(json.dumps(value, ensure_ascii=True, sort_keys=True, + separators=(',', ':'), allow_nan=False).encode()).hexdigest() + + +def _policy(raw: bytes) -> list[dict]: + data = _strict_json(raw, invalid='POLICY_INVALID', allow_floats=True) + if type(data) is not dict: + _fail('POLICY_INVALID') + records = data.get('secret_consumers') + if type(records) is not list or len(records) > 64: + _fail('POLICY_INVALID') + for item in records: + if (type(item) is not dict or not _clean_text(item.get('script'), 4096) + or type(item.get('secrets')) is not list or len(item['secrets']) > 64 + or any(not _clean_text(secret, 4096) for secret in item['secrets'])): + _fail('POLICY_INVALID') + return records + + +def _lexical_parts(path: str) -> tuple[str, ...]: + result: list[str] = [] + for part in path.split('/'): + if part in {'', '.'}: + continue + if part == '..': + if result: + result.pop() + else: + result.append(part) + return _fold(result) + + +def _selector_aliases(selectors: tuple[str, ...], roots: tuple[str, ...]) -> tuple[tuple[str, ...], ...]: + """Conservative lexical exclusions only. Never resolve or inspect a selector.""" + bases = tuple(_lexical_parts(root) for root in roots) + aliases = set() + for selector in selectors: + value = selector.replace('\\', '/') + if (not _clean_text(value, 4096) or value.startswith('//') + or any(char in value for char in ':~$%*?[]{}')): + _fail('PATH_UNSAFE') + candidates = (value,) if value.startswith('/') else tuple(root + '/' + value for root in roots) + for candidate in candidates: + parts = _lexical_parts(candidate) + for base in bases: + if parts[:len(base)] == base: + aliases.add(parts[len(base):]) + return tuple(sorted(aliases)) + + +def _exclude_aliases(paths: tuple[str, ...], selectors: tuple[str, ...], roots: tuple[str, ...]) -> None: + aliases = _selector_aliases(selectors, roots) + for path in paths: + parts = _fold(path.split('/')) + if any(parts[:len(alias)] == alias or alias[:len(parts)] == parts for alias in aliases): + _fail('PATH_UNSAFE') + + +def _script_matches(listed: str, operator: str) -> bool: + try: + return _public_path(listed) == operator + except PreflightError: + return False + + +@dataclass(frozen=True, repr=False) +class DeploymentObservation: + """In-memory consistency record; not authenticated and never deserialized.""" + schema_revision: str + algorithm_revision: str + root_identity: tuple[int, int, int] + manifest_path_digest: str + inputs_digest: str + registration: str + issued_at: int + expires_at: int + digest: str + + +@dataclass(frozen=True, repr=False) +class PreflightReport: + state: str + observation_digest: str | None + issued_at: int | None + expires_at: int | None + package: str + registration: str + recovery: str + diagnostics: tuple[str, ...] + + @classmethod + def blocked(cls, code: str) -> PreflightReport: + safe = code if type(code) is str and code in DIAGNOSTICS else 'INTERNAL_ERROR' + registration = ('missing' if safe in {'POLICY_MISSING', 'CONSUMER_UNREGISTERED'} else + 'invalid' if safe == 'POLICY_INVALID' else 'not_evaluated') + return cls('blocked', None, None, None, 'blocked', registration, 'not_evaluated', + (safe, 'PROVENANCE_UNVERIFIED', 'RECOVERY_UNVERIFIED')) + + def to_dict(self) -> dict[str, object]: + # Callers cannot turn a constructed/tampered report into arbitrary output. + valid = (type(self) is PreflightReport + and type(self.state) is str and self.state in {'ready_for_review', 'blocked'} + and type(self.package) is str and self.package in {'matched', 'blocked', 'not_evaluated'} + and type(self.registration) is str and self.registration in {'matched', 'missing', 'invalid', 'not_evaluated'} + and type(self.recovery) is str and self.recovery in {'declared_unverified', 'not_evaluated'} + and type(self.diagnostics) is tuple and len(self.diagnostics) <= 16 + and all(type(code) is str and code in DIAGNOSTICS for code in self.diagnostics) + and (self.observation_digest is None or type(self.observation_digest) is str + and _SHA256.fullmatch(self.observation_digest) is not None) + and all(value is None or type(value) is int and 0 <= value <= _MAX_EPOCH + for value in (self.issued_at, self.expires_at))) + if not valid: + return PreflightReport.blocked('INTERNAL_ERROR').to_dict() + return {'schema_version': 'getzilla-deploy-preflight-report-v1', 'state': self.state, + 'observation_digest': self.observation_digest, 'issued_at': self.issued_at, 'expires_at': self.expires_at, + 'package': self.package, 'registration': self.registration, 'provenance': 'unverified', + 'recovery': self.recovery, 'connection': 'not_attempted', 'delivery': 'not_attempted', + 'dispatch_permitted': False, 'operational_grant': False, 'diagnostics': list(self.diagnostics)} + + +def _clock() -> int: + value = time.time() + if type(value) not in (float, int) or not math.isfinite(value) or not 0 <= value <= _MAX_EPOCH - TTL_SECONDS: + _fail('CLOCK_UNCERTAIN') + return int(value) + + +def _budget(started: float) -> None: + now = time.monotonic() + if not math.isfinite(now) or now < started: + _fail('CLOCK_UNCERTAIN') + if now - started >= WORK_SECONDS: + _fail('BUDGET_EXCEEDED') + + +def _root_check(bound: BoundRoot) -> None: + try: + bound.check() + except (ComponentStateError, OSError): + _fail('ROOT_CHANGED') + + +def _read(bound: BoundRoot, path: str, limit: int, started: float, *, policy: bool = False) -> FileSnapshot: + _budget(started) + _root_check(bound) + try: + result = bound.read(path, limit=limit) + except ComponentStateError: + _fail('FILE_UNSAFE') + _budget(started) + if result is None: + _fail('POLICY_MISSING' if policy else 'FILE_MISSING') + return result + + +def _observe(root: Path, manifest_path: str, expected_root: tuple[int, int, int] | None = None): + _public_path(manifest_path) + _distinct((POLICY_PATH, manifest_path)) + if fsx.WINDOWS: + _fail('MODE_UNSUPPORTED') + if not isinstance(root, Path): + _fail() + started = time.monotonic() + if type(started) not in (float, int) or not math.isfinite(started): + _fail('CLOCK_UNCERTAIN') + try: + binding = BoundRoot(root) + except (ComponentStateError, OSError): + _fail('ROOT_CHANGED') + with binding as bound: + info = fsx.fstat_dir(bound.handle) + root_identity = (info.st_dev, info.st_ino, stat.S_IFMT(info.st_mode)) + if expected_root is not None and expected_root != root_identity: + _fail('ROOT_CHANGED') + roots = (str(bound.path), os.path.abspath(root)) + policy = _read(bound, POLICY_PATH, MAX_INPUT_BYTES, started, policy=True) + consumers = _policy(policy.content) + selectors = tuple(secret for record in consumers for secret in record['secrets']) + _exclude_aliases((manifest_path,), selectors, roots) + source = _read(bound, manifest_path, MAX_INPUT_BYTES, started) + manifest = parse_manifest(source.content) + payloads = (manifest.operator, *manifest.artifacts) + paths = (manifest_path, *(item.path for item in payloads)) + _distinct((POLICY_PATH, *paths)) + _exclude_aliases(paths, (*selectors, manifest.credential_selector), roots) + selected = consumer_secret_paths(consumers, script_matches=lambda listed: _script_matches(listed, manifest.operator.path)) + registration = 'matched' if manifest.credential_selector in selected else 'missing' + snapshots = [(POLICY_PATH, policy, MAX_INPUT_BYTES), (manifest_path, source, MAX_INPUT_BYTES)] + total = 0 + for expected in payloads: + remaining = MAX_TOTAL_BYTES - total + if remaining <= 0: + _fail('INPUT_TOO_LARGE') + limit = min(MAX_FILE_BYTES, remaining) + snapshot = _read(bound, expected.path, limit, started) + total += len(snapshot.content) + if snapshot.sha256 != expected.sha256: + _fail('HASH_MISMATCH') + if snapshot.mode != expected.mode: + _fail('MODE_MISMATCH') + snapshots.append((expected.path, snapshot, limit)) + # Re-read every input, not only ancestry. Stop at the first changed input; + # a changed policy cannot permit a newly named second-pass payload. + for path, before, limit in snapshots: + after = _read(bound, path, limit, started, policy=path == POLICY_PATH) + if after != before: + _fail('FILE_CHANGED') + _root_check(bound) + inputs = [(path, item.sha256, item.mode, item.identity) for path, item, _ in snapshots] + inputs_digest = _digest(['getzilla-deploy-inputs-v1', sorted(inputs)]) + _budget(started) + return root_identity, inputs_digest, registration + + +def _observation_values(observation: DeploymentObservation) -> tuple: + return (observation.schema_revision, observation.algorithm_revision, observation.root_identity, + observation.manifest_path_digest, observation.inputs_digest, observation.registration, + observation.issued_at, observation.expires_at) + + +def _checksum(values: tuple) -> str: + return _digest(['getzilla-deploy-observation-v1', values]) + + +def _validate_observation(observation: object) -> DeploymentObservation: + if type(observation) is not DeploymentObservation: + _fail() + if (not _clean_text(observation.schema_revision, 128) or not _clean_text(observation.algorithm_revision, 128) + or type(observation.root_identity) is not tuple or len(observation.root_identity) != 3 + or any(type(value) is not int or not 0 <= value <= 2**128 for value in observation.root_identity) + or type(observation.registration) is not str or observation.registration not in {'matched', 'missing'} + or any(type(value) is not int or not 0 <= value <= _MAX_EPOCH + for value in (observation.issued_at, observation.expires_at)) + or observation.expires_at != observation.issued_at + TTL_SECONDS): + _fail() + for value in (observation.manifest_path_digest, observation.inputs_digest, observation.digest): + _sha256(value) + if _checksum(_observation_values(observation)) != observation.digest: + _fail('OBSERVATION_CHANGED') + return observation + + +def _valid_time(observation: DeploymentObservation) -> None: + now = _clock() + if now < observation.issued_at: + _fail('CLOCK_UNCERTAIN') + if now >= observation.expires_at: + _fail('OBSERVATION_EXPIRED') + + +def prepare_preflight(root: Path, manifest_path: str) -> DeploymentObservation: + """Observe admitted public inputs twice. No secret, network or operational effects.""" + try: + _clock() + identity, inputs_digest, registration = _observe(root, manifest_path) + issued = _clock() + values = (SCHEMA_REVISION, ALGORITHM_REVISION, identity, _digest(manifest_path), inputs_digest, + registration, issued, issued + TTL_SECONDS) + return DeploymentObservation(*values, _checksum(values)) + except PreflightError: + raise + except Exception: + _fail('INTERNAL_ERROR') + + +def recheck_preflight(root: Path, observation: DeploymentObservation, manifest_path: str) -> PreflightReport: + """Re-observe inputs; output always keeps execution, provenance and delivery unavailable.""" + try: + observed = _validate_observation(observation) + if observed.schema_revision != SCHEMA_REVISION or observed.algorithm_revision != ALGORITHM_REVISION: + _fail('ALGORITHM_CHANGED') + _valid_time(observed) + _public_path(manifest_path) + if _digest(manifest_path) != observed.manifest_path_digest: + _fail('OBSERVATION_CHANGED') + _identity, digest, registration = _observe(root, manifest_path, observed.root_identity) + _valid_time(observed) + if digest != observed.inputs_digest or registration != observed.registration: + _fail('OBSERVATION_CHANGED') + codes = ('PROVENANCE_UNVERIFIED', 'RECOVERY_UNVERIFIED') + ready = registration == 'matched' + return PreflightReport('ready_for_review' if ready else 'blocked', observed.digest, observed.issued_at, + observed.expires_at, 'matched', registration, 'declared_unverified', + codes if ready else ('CONSUMER_UNREGISTERED', *codes)) + except PreflightError as error: + return PreflightReport.blocked(error.code) + except Exception: + return PreflightReport.blocked('INTERNAL_ERROR') diff --git a/README.md b/README.md index a1c6a7a1..0b99c2d0 100644 --- a/README.md +++ b/README.md @@ -173,6 +173,7 @@ All commands go through one launcher, `getzilla `. The Linux and macOS | | `getzilla entire --context status\|clone\|fetch\|push\|trails\|checks` | Optional Entire adapter; explicit grants for writes, no automated merge or session upload | | | `getzilla trust-ci init\|status\|github-app\|trust-key` | Sets up and checks a self-hosted Trust CI host | | | `getzilla deploy` | Prints, never runs, the publish commands a person runs | +| | `getzilla deploy preflight --root PATH --manifest REL --json` | [Observe a local deployment package](docs/deploy-preflight.md); grants no execution | | Keep clean | `getzilla cleanup inventory` | Lists what Getzilla keeps on disk, with sizes and repository copies; deletes nothing | | | `getzilla cleanup worktrees [--remove]` | Lists worktrees and the reasons to keep them; `--remove` removes only those with no unique work | | | `getzilla cleanup snapshot `, `snapshot --remove ` | Makes a review copy of one commit and removes it after the review | diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 08e59b1a..a3fc083c 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -425,7 +425,7 @@ For a fresh clone, bootstrap state comes from `START_HERE.md` / `PROJECT_STATE.j - [`scripts/getzilla_verify.py`](../scripts/getzilla_verify.py) - [`scripts/getzilla_review.py`](../scripts/getzilla_review.py) - [`scripts/getzilla_approve.py`](../scripts/getzilla_approve.py) — exact action/resource delegated local grant only -- [`scripts/getzilla_deploy.py`](../scripts/getzilla_deploy.py) +- [`scripts/getzilla_deploy.py`](../scripts/getzilla_deploy.py) — [local deployment preflight](deploy-preflight.md) and release preparation - [`scripts/getzilla_doctor.py`](../scripts/getzilla_doctor.py) - [`scripts/getzilla_demo.py`](../scripts/getzilla_demo.py) - [local demo assets and fixtures](../.getzilla/demo/index.html) @@ -641,7 +641,7 @@ Component writers share a persistent advisory guard. It coordinates cooperating ## Scripts -Local loop: measured route → change package when required → bounded committed-HEAD observations → independent selected reviews → persist reports outside Git → one final local gate selected by the delivery lane, parallel with the required exact-head external gate (public Actions or private Trust CI) after exact delegated UNVERIFIED branch transport → current receipts/protected PR eligibility. `scripts/getzilla_deploy.py` is prepare-only and must not bypass protected-branch or exact-SHA requirements. +Local loop: measured route → change package when required → bounded committed-HEAD observations → independent selected reviews → persist reports outside Git → one final local gate selected by the delivery lane, parallel with the required exact-head external gate (public Actions or private Trust CI) after exact delegated UNVERIFIED branch transport → current receipts/protected PR eligibility. The release branch of `scripts/getzilla_deploy.py` is prepare-only and must not bypass protected-branch or exact-SHA requirements. Its explicit-root [preflight](deploy-preflight.md) observes local package consistency and grants no execution or merge authority. MCP tools fail closed: a tool of any server is an external write, and needs an exact external-write grant, unless its name, in plain ASCII, starts or ends with a read verb (`get`, `list`, `search`, `read`, `lookup`, `describe`, `download`) and holds no write verb anywhere (`get_or_create_user` is a write); a name that starts with `run` or `request` is a write, and `query`/`fetch` tools are writes unless listed. A project lists its own read tools with other names, exactly, in `.getzilla/config/policy.json` under `"mcp_read_tools": {"Server": ["tool"]}`. A GitHub write is also checked against this repository; a GitHub proxy whose server name does not contain `github` gets the grant check only. @@ -684,7 +684,7 @@ Every command runs as `getzilla ` through `bin/getzilla` (`bin\getzilla | `protected-write` | `scripts/getzilla_protected_write.py` | Apply an approved protected-path write from a manifest (`--dry-run` first) | | `ci` | `scripts/getzilla_ci.py` | Select the CI gate by visibility; `--write` renders the pinned public workflow; `--protect-plan` shows branch protection | | `trust-ci` | `scripts/getzilla_trust_ci.py` | Trust CI host: `init`, `status`, `github-app`, `trust-key` | -| `deploy` | `scripts/getzilla_deploy.py` | Prepare-only delivery commands | +| `deploy` | `scripts/getzilla_deploy.py` | Prepare-only release commands; `preflight --root PATH --manifest REL --json` performs a [local observation](deploy-preflight.md) | | — | `scripts/getzilla_evidence.py` | Export bounded local workflow evidence, without approvals or secrets, with `--output ` | | `cleanup` | `scripts/getzilla_cleanup.py` | `inventory` (read-only disk report with repository copies), `worktrees [--remove]` (guarded removal), `snapshot ` / `snapshot --remove ` (review copies) | | `branches` | `scripts/branch_lifecycle.py` | Branch retention `report`; `apply` deletes only from a reviewed plan with `--allow-delete` | diff --git a/docs/deploy-preflight.md b/docs/deploy-preflight.md new file mode 100644 index 00000000..034ecc71 --- /dev/null +++ b/docs/deploy-preflight.md @@ -0,0 +1,74 @@ +# Local deployment preflight + +`getzilla deploy preflight` checks the consistency of a local public deployment package. It compares an operator, its artifacts and the project's `secret_consumers` registration. A successful result is an observation for review. It does not authorize or perform a deployment. + +```bash +getzilla deploy preflight --root /absolute/path/to/project --manifest deployment/preflight.json --json +``` + +The root is required. The manifest path is relative to that root, including when the command runs from a subdirectory. The CLI prepares and immediately rechecks one in-memory observation. It accepts no saved observation token. Its exit codes are: + +| Code | Meaning | +| --- | --- | +| `0` | `ready_for_review`: local inputs and registration match; execution is still unavailable | +| `1` | `blocked`: a fixed diagnostic explains an unsafe, missing, changed or unsupported input | +| `2` | Invalid command arguments; their values are not echoed | + +The existing flat `getzilla deploy`, `deploy --json` and `deploy --record --json` commands retain release preparation and receipt approval behavior. `preflight --record` is refused before inputs are read. + +## Manifest and registration + +The [versioned schema](../schemas/deploy-preflight-manifest-v1.schema.json) describes the JSON structure. This example uses synthetic identifiers. Its zero hashes are placeholders and deliberately do not qualify a real package. Replace them with hashes of reviewed public files and the reviewed recovery descriptor. Check the source changes before rebuilding expectations. + +```json +{ + "schema_version": "getzilla-deploy-preflight-v1", + "project_id": "example", + "operation": "dev-delivery", + "operator": { + "path": "deployment/operator.py", + "sha256": "0000000000000000000000000000000000000000000000000000000000000000", + "mode": 420, + "revision": {"label": "reviewed-operator-v1", "git": null} + }, + "artifacts": [{ + "path": "public/application.js", + "sha256": "0000000000000000000000000000000000000000000000000000000000000000", + "mode": 420 + }], + "artifact_revision": {"label": "reviewed-payload-v1", "git": null}, + "target": {"identity": "dev-example", "destination_id": "web-assets", "transport": "local-fixture"}, + "credential_selector": ".env.fixture", + "recovery": { + "protocol_id": "fixture-only", + "version": "v1", + "descriptor_sha256": "0000000000000000000000000000000000000000000000000000000000000000" + } +} +``` + +Mode `420` is POSIX `0644`. Modes are checked exactly on POSIX. Windows returns `MODE_UNSUPPORTED` before reading inputs; v1 does not claim portable mode verification. + +The fixed `.getzilla/config/project-policy.json` must contain a `secret_consumers` array. The operator needs an existing record with its exact portable relative script path and the exact selector string. Preflight never adds a record. Absolute, outside-root or nonportable script spellings cannot match preflight registration; existing command authorization keeps its own unchanged identity rules. A registration match proves only that this metadata exists. It neither admits a command nor approves the operator bytes. + +The selector is metadata only. Its credential file is not opened, expanded, resolved or inspected. Policy selectors from every record exclude possible local aliases, even when that record does not match the operator. Dot components, separators, absolute paths under the root and Unicode/case equivalents are normalized only for exclusion. Overlap, expansion/glob syntax, foreign roots and ambiguous selectors block public reads. Lexically outside absolute POSIX selectors are not resolved. Unknown physical aliases and secrets copied into otherwise public files are outside this path-based guarantee. + +Public admission combines the strict sensitive-file classifier with the full built-in `DEFAULT_SECRET_READ` path exclusions. This includes coding-agent and cloud credential stores, `secrets/**` and Trust CI runtime. Template or public-key exceptions do not weaken this floor. Preflight uses this built-in floor plus the registered and declared selector exclusions; it does not load or bind custom `secret_read_paths` from `.getzilla/config/policy.json`. It is therefore not an evaluation of the complete effective command-admission policy. Projects must keep their proposed inputs public and account for additional project-specific exclusions before running this local observation. + +## What the observation binds + +The implementation admits the manifest before opening it, then admits the complete payload inventory before opening any payload. It uses bounded regular single-link reads and rejects sensitive file classes, traversal, reserved paths, control-file overlap and path collisions. The fixed policy is the configuration bootstrap input. + +Both passes compare all input bytes, modes and file identities, including policy and manifest. The observation binds the filesystem root, manifest spelling, input digests, registration, schema and algorithm revision, and a 300-second lifetime. A recheck performs fresh reads. A same-path operator replacement, revoked policy, changed target or artifact, different root, clock rollback or expiry blocks reuse. Rebuilding a manifest creates different expectations and cannot establish reviewed provenance. + +Input JSON is limited to 64 KiB and eight nesting levels. Duplicate keys, nonfinite values and manifest floating-point numbers are refused. The schema covers structure; runtime checks also enforce lexical integer spelling, path relationships and read admission. Policy records are limited to 64, each with at most 64 selectors. There are at most 128 artifacts, 16 MiB per payload file and 64 MiB of payload per pass. A ten-second monotonic budget is checked at read boundaries. Filesystem refusals, including size limits enforced by the bounded reader, produce fixed diagnostics. This budget does not interrupt a stalled kernel I/O call. + +## Boundaries of a successful result + +Every result fixes `dispatch_permitted` and `operational_grant` to `false`, and `connection` and `delivery` to `not_attempted`. Git provenance always remains `unverified`, even when the manifest declares plausible commit/tree IDs. Recovery remains `declared_unverified`; a protocol name or descriptor hash proves no crash behavior. A missing registration may coexist with a matched public package, but the result is blocked. + +Preflight imports or executes no operator, starts no subprocess, uses no network and touches no target or lock. It creates no route, grant, receipt, recovery journal or import cache during ordinary CLI startup. The existing `getzilla` launcher still starts its normal Python dispatcher process. Reports contain fixed status fields, diagnostic codes, hashes and times; they omit paths, selectors, raw exceptions and command output. + +The observation checksum is unkeyed consistency evidence, not authentication. Installed Getzilla is trusted. Relevant algorithm or ABI changes must bump `ALGORITHM_REVISION`; v1 does not attest loaded executable code or inspect module files, so ZIP engines remain supported. Two passes do not provide a global atomic snapshot or isolation from a hostile concurrent writer. Changes after the observation can invalidate it. + +Exact Git provenance, execution admission, shared-lock crash recovery and independent DEV readback require separate qualification. A new manifest alone does not complete those steps or close the deployment-readiness issue. diff --git a/schemas/deploy-preflight-manifest-v1.schema.json b/schemas/deploy-preflight-manifest-v1.schema.json new file mode 100644 index 00000000..1d1e5b14 --- /dev/null +++ b/schemas/deploy-preflight-manifest-v1.schema.json @@ -0,0 +1,79 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://getzilla.dev/schemas/deploy-preflight-manifest-v1.schema.json", + "title": "Deployment preflight declarations (not authorization or provenance)", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "project_id", "operation", "operator", "artifacts", "artifact_revision", "target", "credential_selector", "recovery"], + "properties": { + "schema_version": {"const": "getzilla-deploy-preflight-v1"}, + "project_id": {"$ref": "#/$defs/identifier"}, + "operation": {"const": "dev-delivery"}, + "operator": { + "type": "object", "additionalProperties": false, + "required": ["path", "sha256", "mode", "revision"], + "properties": { + "path": {"$ref": "#/$defs/path"}, + "sha256": {"$ref": "#/$defs/sha256"}, + "mode": {"$ref": "#/$defs/mode"}, + "revision": {"$ref": "#/$defs/revision"} + } + }, + "artifacts": {"type": "array", "minItems": 1, "maxItems": 128, "items": {"$ref": "#/$defs/file"}}, + "artifact_revision": {"$ref": "#/$defs/revision"}, + "target": { + "type": "object", "additionalProperties": false, + "required": ["identity", "destination_id", "transport"], + "properties": { + "identity": {"$ref": "#/$defs/identifier"}, + "destination_id": {"$ref": "#/$defs/identifier"}, + "transport": {"enum": ["ssh-sftp", "local-fixture"]} + } + }, + "credential_selector": {"type": "string", "minLength": 1, "maxLength": 4096, "pattern": "^[^\u0000-\u001f\u007f-\u009f\ud800-\udfff]+$"}, + "recovery": { + "type": "object", "additionalProperties": false, + "required": ["protocol_id", "version", "descriptor_sha256"], + "properties": { + "protocol_id": {"$ref": "#/$defs/identifier"}, + "version": {"$ref": "#/$defs/identifier"}, + "descriptor_sha256": {"$ref": "#/$defs/sha256"} + } + } + }, + "$defs": { + "identifier": {"type": "string", "minLength": 1, "maxLength": 64, "pattern": "^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$"}, + "sha256": {"type": "string", "minLength": 64, "maxLength": 64, "pattern": "^[0-9a-f]{64}$"}, + "mode": {"type": "integer", "minimum": 0, "maximum": 511}, + "path": { + "type": "string", "minLength": 1, "maxLength": 1024, + "pattern": "^(?!/)(?!.*//)(?!.*(?:^|/)\\.{1,2}(?:/|$))(?!.*[ .](?:/|$))[^\\\\:\u0000-\u001f\u007f-\u009f\ud800-\udfff*?\"<>|]+$", + "description": "Portable relative spelling. Runtime additionally checks component count, reserved devices, sensitive classes, casefold collisions, control overlap and lexical credential aliases before opening files." + }, + "file": { + "type": "object", "additionalProperties": false, "required": ["path", "sha256", "mode"], + "properties": {"path": {"$ref": "#/$defs/path"}, "sha256": {"$ref": "#/$defs/sha256"}, "mode": {"$ref": "#/$defs/mode"}} + }, + "revision": { + "type": "object", "additionalProperties": false, "required": ["label", "git"], + "properties": { + "label": {"$ref": "#/$defs/identifier"}, + "git": { + "oneOf": [ + {"type": "null"}, + { + "type": "object", "additionalProperties": false, + "required": ["object_format", "commit_oid", "tree_oid"], + "properties": {"object_format": {"const": "sha1"}, "commit_oid": {"type": "string", "minLength": 40, "maxLength": 40, "pattern": "^[0-9a-f]{40}$"}, "tree_oid": {"type": "string", "minLength": 40, "maxLength": 40, "pattern": "^[0-9a-f]{40}$"}} + }, + { + "type": "object", "additionalProperties": false, + "required": ["object_format", "commit_oid", "tree_oid"], + "properties": {"object_format": {"const": "sha256"}, "commit_oid": {"$ref": "#/$defs/sha256"}, "tree_oid": {"$ref": "#/$defs/sha256"}} + } + ] + } + } + } + } +} diff --git a/scripts/getzilla_cli.py b/scripts/getzilla_cli.py index 5e7f2c3b..e90f5b5f 100644 --- a/scripts/getzilla_cli.py +++ b/scripts/getzilla_cli.py @@ -6,9 +6,13 @@ """ from __future__ import annotations +import sys + +if __name__ == '__main__' and sys.argv[1:3] == ['deploy', 'preflight']: + sys.dont_write_bytecode = True + import os import subprocess -import sys from pathlib import Path HOME = Path(__file__).resolve().parents[1] @@ -44,7 +48,7 @@ 'git': ('getzilla_git.py', 'Git platform capabilities and offline remote identity'), 'entire': ('getzilla_entire.py', 'optional Entire transport and Trails (extra:entire)'), 'trust-ci': ('getzilla_trust_ci.py', 'set up and check the self-hosted Trust CI host (init, status)'), - 'deploy': ('getzilla_deploy.py', 'prepare human-owned publish commands'), + 'deploy': ('getzilla_deploy.py', 'prepare publish commands or observe a local package with preflight'), 'harness': ('getzilla_harness.py', 'check or regenerate the selected agent harness files'), 'setup-agent': ('getzilla_setup_agent.py', 'configure the coding agent and its models'), 'install-into': ('install_into.py', 'install Getzilla into a project (--plan, --materialize-new)'), diff --git a/scripts/getzilla_deploy.py b/scripts/getzilla_deploy.py index ef4c8c55..bd66b0bb 100644 --- a/scripts/getzilla_deploy.py +++ b/scripts/getzilla_deploy.py @@ -1,6 +1,11 @@ from __future__ import annotations import sys + +# The read-only command must also avoid import-cache writes during ordinary startup. +if __name__ == '__main__' and sys.argv[1:2] == ['preflight']: + sys.dont_write_bytecode = True + from pathlib import Path ROOT = Path(__file__).resolve().parents[1] @@ -12,16 +17,60 @@ from getzilla.deploy import prepare_deploy from getzilla.util import find_root -parser = argparse.ArgumentParser(description='Prepare human-owned publish commands. Never executes tag, push, or release.') -parser.add_argument('--record', action='store_true', help='Write a deploy/prepared receipt. Requires production approval.') -parser.add_argument('--json', action='store_true') -args = parser.parse_args() -report = prepare_deploy(find_root(), record=args.record) -if args.json: - print(json.dumps(report, ensure_ascii=False, indent=2)) -elif report.get('ok'): - for command in report.get('commands', []): - print(command) -else: - print(report.get('error', 'deploy prepare failed'), file=sys.stderr) -raise SystemExit(0 if report.get('ok') else 1) +class _UsageError(ValueError): + pass + + +class _PreflightParser(argparse.ArgumentParser): + def error(self, _message: str) -> None: + raise _UsageError from None + + +def _preflight(argv: list[str]) -> int: + parser = _PreflightParser(prog='getzilla deploy preflight', allow_abbrev=False, + description='Observe a local package for review. Grants no execution or DEV readiness.') + parser.add_argument('--root', required=True, type=Path, help='Explicit root containing the public inputs.') + parser.add_argument('--manifest', required=True, help='Portable root-relative manifest path.') + parser.add_argument('--json', action='store_true') + try: + args = parser.parse_args(argv) + except _UsageError: + print('getzilla deploy preflight: invalid arguments', file=sys.stderr) + return 2 + from getzilla.deploy_preflight import PreflightError, PreflightReport, prepare_preflight, recheck_preflight + try: + observation = prepare_preflight(args.root, args.manifest) + report = recheck_preflight(args.root, observation, args.manifest) + except PreflightError as error: + report = PreflightReport.blocked(error.code) + except Exception: + report = PreflightReport.blocked('INTERNAL_ERROR') + data = report.to_dict() + if args.json: + print(json.dumps(data, ensure_ascii=True, indent=2)) + else: + print('Preflight: ' + data['state'] + '; ' + ', '.join(data['diagnostics'])) + return 0 if data['state'] == 'ready_for_review' else 1 + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + if argv[:1] == ['preflight']: + return _preflight(argv[1:]) + parser = argparse.ArgumentParser(description='Prepare human-owned publish commands. Never executes tag, push, or release.') + parser.add_argument('--record', action='store_true', help='Write a deploy/prepared receipt. Requires production approval.') + parser.add_argument('--json', action='store_true') + args = parser.parse_args(argv) + report = prepare_deploy(find_root(), record=args.record) + if args.json: + print(json.dumps(report, ensure_ascii=False, indent=2)) + elif report.get('ok'): + for command in report.get('commands', []): + print(command) + else: + print(report.get('error', 'deploy prepare failed'), file=sys.stderr) + return 0 if report.get('ok') else 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/scripts/install_into.py b/scripts/install_into.py index f79be965..627df171 100644 --- a/scripts/install_into.py +++ b/scripts/install_into.py @@ -83,6 +83,7 @@ "scripts/getzilla_status.py", "scripts/getzilla_agent.py", "scripts/getzilla_deploy.py", + "docs/deploy-preflight.md", "scripts/getzilla_evidence.py", "scripts/getzilla_ci_evidence.py", "scripts/getzilla_git.py", @@ -101,6 +102,7 @@ ".coveragerc", "schemas/change-spec.schema.json", "schemas/change-spec-v1.schema.json", + "schemas/deploy-preflight-manifest-v1.schema.json", "schemas/architecture-system.schema.json", "schemas/architecture-rules.schema.json", "schemas/governance-rule.schema.json", diff --git a/tests/test_deploy_preflight.py b/tests/test_deploy_preflight.py new file mode 100644 index 00000000..e670f7e4 --- /dev/null +++ b/tests/test_deploy_preflight.py @@ -0,0 +1,685 @@ +"""Deployment observations are bounded consistency checks, never execution grants.""" +from __future__ import annotations + +import copy +import dataclasses +import hashlib +import importlib.util +import io +import json +import os +import socket +import subprocess +import sys +import tempfile +import unittest +from contextlib import ExitStack, redirect_stderr, redirect_stdout +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / '.getzilla')) + +from getzilla import deploy_preflight as preflight +from getzilla import fsx +from getzilla.component_state import BoundRoot + + +def manifest() -> dict: + revision = {'label': 'fixture-v1', 'git': None} + return { + 'schema_version': 'getzilla-deploy-preflight-v1', + 'project_id': 'example', + 'operation': 'dev-delivery', + 'operator': {'path': 'deploy/operator.py', 'sha256': hashlib.sha256(b'operator\n').hexdigest(), + 'mode': 0o644, 'revision': revision}, + 'artifacts': [{'path': 'public/application.js', 'sha256': hashlib.sha256(b'payload\n').hexdigest(), + 'mode': 0o644}], + 'artifact_revision': copy.deepcopy(revision), + 'target': {'identity': 'dev-example', 'destination_id': 'web-assets', 'transport': 'local-fixture'}, + 'credential_selector': '.env.fixture', + 'recovery': {'protocol_id': 'fixture-only', 'version': 'v1', 'descriptor_sha256': 'a' * 64}, + } + + +def encoded(value: dict) -> bytes: + return json.dumps(value, ensure_ascii=True, separators=(',', ':')).encode() + + +class ManifestTests(unittest.TestCase): + def rejected(self, raw: bytes, code: str = 'INPUT_INVALID') -> None: + with self.assertRaises(preflight.PreflightError) as caught: + preflight.parse_manifest(raw) + self.assertEqual(caught.exception.code, code) + self.assertEqual(str(caught.exception), code) + + def test_valid_manifest_is_immutable_and_repr_hides_input(self) -> None: + parsed = preflight.parse_manifest(encoded(manifest())) + self.assertIs(type(parsed), preflight.DeployManifest) + self.assertIs(type(parsed.artifacts), tuple) + self.assertEqual(parsed.operator.path, 'deploy/operator.py') + with self.assertRaises(AttributeError): + parsed.project_id = 'changed' + for item in (parsed, parsed.operator, *parsed.artifacts): + self.assertNotIn('.env.fixture', repr(item)) + self.assertNotIn('operator.py', repr(item)) + + def test_unknown_and_missing_keys_are_rejected_at_every_object(self) -> None: + for key in manifest(): + value = manifest() + del value[key] + with self.subTest(missing=key): + self.rejected(encoded(value)) + locations = [(), ('operator',), ('operator', 'revision'), ('artifact_revision',), + ('target',), ('recovery',), ('artifacts', 0)] + for location in locations: + value = manifest() + target = value + for key in location: + target = target[key] + target['unexpected'] = 'private-value-must-not-escape' + with self.subTest(extra=location): + self.rejected(encoded(value)) + + def test_json_duplicates_invalid_utf8_and_noninteger_numbers(self) -> None: + good = encoded(manifest()) + self.rejected(good.replace(b'"project_id":"example"', b'"project_id":"x","project_id":"example"')) + self.rejected(good.replace(b'"mode":420', b'"mode":420,"mode":420', 1)) + self.rejected(b'\xff') + for token in (b'NaN', b'Infinity', b'-Infinity', b'420.0', b'4.2e2', b'true', b'null', b'"420"'): + with self.subTest(token=token): + self.rejected(good.replace(b'"mode":420', b'"mode":' + token, 1)) + for value in (None, [], 'private-value-must-not-escape', 1, True): + self.rejected(json.dumps(value).encode()) + + def test_size_depth_and_collection_bounds(self) -> None: + self.rejected(b' ' * (65536 + 1), 'INPUT_TOO_LARGE') + self.rejected(b'[' * 2000 + b'0' + b']' * 2000) + for count in (0, 129): + value = manifest() + value['artifacts'] = [dict(value['artifacts'][0], path=f'public/a{i}.js') for i in range(count)] + self.rejected(encoded(value)) + value = manifest() + value['artifacts'] = [dict(value['artifacts'][0], path=f'public/a{i}.js') for i in range(128)] + self.assertEqual(len(preflight.parse_manifest(encoded(value)).artifacts), 128) + + def test_scalar_constraints_and_modes(self) -> None: + invalids = [ + ('project_id', ''), ('project_id', '../bad'), ('project_id', 'é'), ('project_id', 'a' * 65), + ('schema_version', 'v2'), ('operation', 'production'), + ('credential_selector', ''), ('credential_selector', 'x' * 4097), + ('credential_selector', 'x\nprivate-value'), ('credential_selector', 'x\x7f'), + ] + for key, invalid in invalids: + value = manifest() + value[key] = invalid + with self.subTest(key=key, invalid=invalid[:40]): + self.rejected(encoded(value)) + for key, invalid in [('mode', -1), ('mode', 512), ('sha256', 'A' * 64), ('sha256', 'a' * 63)]: + value = manifest() + value['operator'][key] = invalid + self.rejected(encoded(value)) + for mode in (0, 511): + value = manifest() + value['operator']['mode'] = mode + self.assertEqual(preflight.parse_manifest(encoded(value)).operator.mode, mode) + + def test_revision_oids_are_bounded_declarations(self) -> None: + for fmt, size in [('sha1', 40), ('sha256', 64)]: + value = manifest() + value['artifact_revision']['git'] = {'object_format': fmt, 'commit_oid': 'a' * size, + 'tree_oid': 'b' * size} + preflight.parse_manifest(encoded(value)) + for key, bad in [('commit_oid', 'a' * (size - 1)), ('tree_oid', 'G' * size), + ('object_format', 'unknown'), ('other', 'value')]: + broken = copy.deepcopy(value) + broken['artifact_revision']['git'][key] = bad + self.rejected(encoded(broken)) + + def test_paths_refuse_traversal_devices_sensitive_inputs_and_collisions(self) -> None: + invalids = ['', '/absolute', '../outside', 'a/../b', 'a/./b', 'a//b', 'a\\b', 'C:drive', + 'a/nul.txt', 'a/CON', 'a/trailing.', 'a/trailing ', 'a\x00b', 'a\x7fb', + '.git/config', 'x/.GIT/HEAD', '.env.example', 'x/private.pem', + 'x/credentials.json', 'x/.ssh/id.pub', 'private.db', 'bundle.zip', + '/'.join(['a'] * 33), 'a' * 1025] + for path in invalids: + value = manifest() + value['operator']['path'] = path + with self.subTest(path=path[:60]): + self.rejected(encoded(value), 'PATH_UNSAFE') + for path in ('deploy/operator.py', 'DEPLOY/OPERATOR.PY', '.getzilla/config/project-policy.json', + 'deploy/operator.py/child'): + value = manifest() + value['artifacts'][0]['path'] = path + with self.subTest(collision=path): + self.rejected(encoded(value), 'PATH_UNSAFE') + + def test_schema_matches_structural_parser_boundaries(self) -> None: + import jsonschema + schema = json.loads((ROOT / 'schemas/deploy-preflight-manifest-v1.schema.json').read_text()) + validator = jsonschema.Draft202012Validator(schema) + validator.check_schema(schema) + value = manifest() + self.assertFalse(list(validator.iter_errors(value))) + variants = [] + for key in value: + bad = manifest() + del bad[key] + variants.append(bad) + for key, bad_value in [('mode', True), ('mode', 512), ('mode', -1), ('sha256', 'x')]: + bad = manifest() + bad['operator'][key] = bad_value + variants.append(bad) + bad = manifest() + bad['target']['extra'] = 'x' + variants.append(bad) + for bad in variants: + self.assertTrue(list(validator.iter_errors(bad))) + self.rejected(encoded(bad)) + + +@unittest.skipIf(os.name == 'nt', 'POSIX mode observation; Windows refusal has its own control') +class ObservationTests(unittest.TestCase): + def setUp(self) -> None: + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.value = manifest() + (self.root / 'deploy').mkdir() + (self.root / 'public').mkdir() + (self.root / '.getzilla/config').mkdir(parents=True) + (self.root / 'deploy/operator.py').write_bytes(b'operator\n') + (self.root / 'public/application.js').write_bytes(b'payload\n') + (self.root / 'deploy/operator.py').chmod(0o644) + (self.root / 'public/application.js').chmod(0o644) + self.policy = {'secret_consumers': [{'script': 'deploy/operator.py', 'secrets': ['.env.fixture']}]} + self.save() + + def save(self) -> None: + (self.root / 'fixture.json').write_bytes(encoded(self.value)) + (self.root / preflight.POLICY_PATH).write_bytes(encoded(self.policy)) + + def prepare(self): + return preflight.prepare_preflight(self.root, 'fixture.json') + + def report(self, observation=None): + if observation is None: + observation = self.prepare() + return preflight.recheck_preflight(self.root, observation, 'fixture.json').to_dict() + + def assert_blocked(self, report, code): + self.assertEqual(report['state'], 'blocked') + self.assertIn(code, report['diagnostics']) + self.assertIs(report['dispatch_permitted'], False) + self.assertIs(report['operational_grant'], False) + + def test_unchanged_observation_is_reviewable_only(self) -> None: + observation = self.prepare() + report = self.report(observation) + self.assertEqual(report['state'], 'ready_for_review') + self.assertEqual(report['package'], 'matched') + self.assertEqual(report['registration'], 'matched') + self.assertEqual(report['provenance'], 'unverified') + self.assertEqual(report['recovery'], 'declared_unverified') + self.assertEqual(report['connection'], 'not_attempted') + self.assertEqual(report['delivery'], 'not_attempted') + self.assertIs(report['dispatch_permitted'], False) + self.assertIs(report['operational_grant'], False) + self.assertEqual(report['expires_at'] - report['issued_at'], 300) + self.assertEqual(set(report['diagnostics']), {'PROVENANCE_UNVERIFIED', 'RECOVERY_UNVERIFIED'}) + for item in (observation, preflight.recheck_preflight(self.root, observation, 'fixture.json')): + self.assertNotIn('.env.fixture', repr(item)) + for private in (str(self.root), 'operator.py', '.env.fixture', 'dev-example'): + self.assertNotIn(private, json.dumps(report)) + + def test_same_registered_operator_path_does_not_authorize_replacement(self) -> None: + observation = self.prepare() + (self.root / 'deploy/operator.py').write_bytes(b'changed operator\n') + self.assert_blocked(self.report(observation), 'HASH_MISMATCH') + + def test_artifact_bytes_modes_and_identity_are_bound(self) -> None: + artifact = self.root / 'public/application.js' + for mutation, code in [('bytes', 'HASH_MISMATCH'), ('mode', 'MODE_MISMATCH'), ('replace', 'OBSERVATION_CHANGED')]: + with self.subTest(mutation=mutation): + artifact.write_bytes(b'payload\n') + artifact.chmod(0o644) + observation = self.prepare() + if mutation == 'bytes': + artifact.write_bytes(b'changed\n') + elif mutation == 'mode': + artifact.chmod(0o600) + else: + artifact.unlink() + artifact.write_bytes(b'payload\n') + artifact.chmod(0o644) + self.assert_blocked(self.report(observation), code) + + def test_manifest_target_selector_recovery_and_revisions_invalidate_binding(self) -> None: + edits = [('target', 'identity', 'different'), ('target', 'destination_id', 'other'), + ('target', 'transport', 'ssh-sftp'), ('recovery', 'version', 'v2'), + ('recovery', 'descriptor_sha256', 'b' * 64), ('artifact_revision', 'label', 'v2'), + ('operator', 'revision', {'label': 'v2', 'git': None})] + for group, key, value in edits: + self.value = manifest() + self.save() + observation = self.prepare() + self.value[group][key] = value + (self.root / 'fixture.json').write_bytes(encoded(self.value)) + with self.subTest(group=group, key=key): + self.assert_blocked(self.report(observation), 'OBSERVATION_CHANGED') + self.value = manifest() + self.save() + observation = self.prepare() + self.value['credential_selector'] = '.env.other' + self.policy['secret_consumers'][0]['secrets'].append('.env.other') + self.save() + self.assert_blocked(self.report(observation), 'OBSERVATION_CHANGED') + self.value['operation'] = 'production' + self.save() + self.assert_blocked(self.report(observation), 'INPUT_INVALID') + + def test_policy_changes_revoke_or_invalidate_observation(self) -> None: + observation = self.prepare() + self.policy['secret_consumers'] = [] + (self.root / preflight.POLICY_PATH).write_bytes(encoded(self.policy)) + self.assert_blocked(self.report(observation), 'OBSERVATION_CHANGED') + fresh = self.report() + self.assert_blocked(fresh, 'CONSUMER_UNREGISTERED') + self.assertEqual(fresh['registration'], 'missing') + self.assertEqual(fresh['package'], 'matched') + policy = self.root / preflight.POLICY_PATH + policy.unlink() + self.assert_blocked(self.report(observation), 'POLICY_MISSING') + for invalid in (b'{', b'{}', b'{"secret_consumers":[],"secret_consumers":[]}', + b'{"secret_consumers":[{"script":"x","secrets":3}]}'): + policy.write_bytes(invalid) + self.assert_blocked(self.report(observation), 'POLICY_INVALID') + + def test_exact_policy_bytes_are_bound_even_when_registration_is_unchanged(self) -> None: + observation = self.prepare() + self.policy['unrelated_metadata'] = 'different' + (self.root / preflight.POLICY_PATH).write_bytes(encoded(self.policy)) + self.assert_blocked(self.report(observation), 'OBSERVATION_CHANGED') + + def test_policy_metadata_is_allowed_but_bounded(self) -> None: + self.policy['unrelated'] = {'cost': 0.25, 'enabled': True} + self.policy['secret_consumers'][0]['owner_metadata'] = {'reviewed': True} + self.save() + self.assertEqual(self.report()['registration'], 'matched') + for key, value in [('script', ''), ('script', 'x' * 4097), ('secrets', ['x'] * 65), + ('secrets', ['']), ('secrets', ['x' * 4097]), ('secrets', [False])]: + self.policy['secret_consumers'] = [{'script': 'deploy/operator.py', 'secrets': ['.env.fixture']}] + self.policy['secret_consumers'][0][key] = value + self.save() + with self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'POLICY_INVALID') + + def test_manifest_alias_is_rejected_before_manifest_read(self) -> None: + aliases = ['fixture.json', './fixture.json', 'part/../fixture.json', '.\\fixture.json', + 'FIXTURE.JSON', str(self.root / 'fixture.json'), str(self.root / './fixture.json'), + '.', str(self.root), '$UNKNOWN/file', '~/file', 'C:relative', 'https://host/selector', + '//server/share', 'prefix/*'] + original = BoundRoot.read + for alias in aliases: + self.policy['secret_consumers'][0]['secrets'] = [alias] + self.save() + calls = [] + def observe(binding, relative, **kwargs): + calls.append(relative) + return original(binding, relative, **kwargs) + with patch.object(BoundRoot, 'read', observe), self.assertRaises(preflight.PreflightError) as error: + self.prepare() + with self.subTest(alias=alias): + self.assertEqual(error.exception.code, 'PATH_UNSAFE') + self.assertEqual(calls, [preflight.POLICY_PATH]) + + def test_entire_inventory_is_admitted_before_any_payload_open(self) -> None: + original = BoundRoot.read + for selector in ['public', 'public/application.js', 'PUBLIC/APPLICATION.JS', + 'public/application.js/descendant', str(self.root / 'public/application.js')]: + self.value['credential_selector'] = selector + self.save() + calls = [] + def observe(binding, relative, **kwargs): + calls.append(relative) + return original(binding, relative, **kwargs) + with patch.object(BoundRoot, 'read', observe), self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'PATH_UNSAFE') + self.assertEqual(calls, [preflight.POLICY_PATH, 'fixture.json']) + + def test_nonmatching_policy_selectors_still_exclude_reads(self) -> None: + self.policy['secret_consumers'].append({'script': '/outside/uninspected.py', + 'secrets': ['public/application.js']}) + self.save() + with self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'PATH_UNSAFE') + + def test_outside_or_nonportable_script_cannot_match_or_be_inspected(self) -> None: + for script in ['/outside/operator.py', '../operator.py', str(self.root / 'deploy/operator.py'), + './deploy/operator.py', 'DEPLOY/OPERATOR.PY']: + self.policy['secret_consumers'][0]['script'] = script + self.save() + with patch('os.path.realpath', side_effect=AssertionError('private-path')), \ + patch.object(Path, 'resolve', side_effect=AssertionError('private-path')): + report = self.report() + self.assert_blocked(report, 'CONSUMER_UNREGISTERED') + + def test_links_hardlinks_special_files_and_unsafe_root_fail(self) -> None: + artifact = self.root / 'public/application.js' + artifact.unlink() + artifact.symlink_to(self.root / 'deploy/operator.py') + with self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'FILE_UNSAFE') + artifact.unlink() + os.link(self.root / 'deploy/operator.py', artifact) + with self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'FILE_UNSAFE') + artifact.unlink() + os.mkfifo(artifact) + with self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'FILE_UNSAFE') + artifact.unlink() + artifact.write_bytes(b'payload\n') + alias = self.root / 'alias' + alias.symlink_to(self.root, target_is_directory=True) + # Root-owned system links may be canonicalized; this is an untrusted parent. + self.root.chmod(0o777) + with self.assertRaises(preflight.PreflightError) as error: + preflight.prepare_preflight(alias, 'fixture.json') + self.assertEqual(error.exception.code, 'ROOT_CHANGED') + + def test_other_root_or_manifest_spelling_rejects_old_observation(self) -> None: + observation = self.prepare() + with tempfile.TemporaryDirectory() as empty: + report = preflight.recheck_preflight(Path(empty), observation, 'fixture.json').to_dict() + self.assert_blocked(report, 'ROOT_CHANGED') + self.assert_blocked(preflight.recheck_preflight(self.root, observation, './fixture.json').to_dict(), 'PATH_UNSAFE') + (self.root / 'second.json').write_bytes(encoded(self.value)) + self.assert_blocked(preflight.recheck_preflight(self.root, observation, 'second.json').to_dict(), 'OBSERVATION_CHANGED') + + def test_later_payload_mutation_is_detected_by_second_pass(self) -> None: + original = BoundRoot.read + seen = [] + def mutate(binding, relative, **kwargs): + snapshot = original(binding, relative, **kwargs) + seen.append(relative) + if relative == 'public/application.js' and seen.count(relative) == 1: + (self.root / 'deploy/operator.py').write_bytes(b'changed after first read\n') + return snapshot + with patch.object(BoundRoot, 'read', mutate), self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'FILE_CHANGED') + self.assertEqual(seen.count('deploy/operator.py'), 2) + + def test_policy_change_stops_second_pass_before_new_manifest_or_payload_reads(self) -> None: + original = BoundRoot.read + calls = [] + def mutate(binding, relative, **kwargs): + snapshot = original(binding, relative, **kwargs) + calls.append(relative) + if relative == 'public/application.js': + (self.root / preflight.POLICY_PATH).write_bytes(b'{"secret_consumers":[]}') + return snapshot + with patch.object(BoundRoot, 'read', mutate), self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'FILE_CHANGED') + self.assertEqual(calls[-1], preflight.POLICY_PATH) + self.assertEqual(calls.count('fixture.json'), 1) + + def test_expiry_clock_and_implementation_revision_are_checked(self) -> None: + with patch.object(preflight.time, 'time', return_value=1000.0): + observation = self.prepare() + for clock, code in [(999.0, 'CLOCK_UNCERTAIN'), (1300.0, 'OBSERVATION_EXPIRED'), + (float('nan'), 'CLOCK_UNCERTAIN'), (float('inf'), 'CLOCK_UNCERTAIN'), + (True, 'CLOCK_UNCERTAIN')]: + with patch.object(preflight.time, 'time', return_value=clock): + self.assert_blocked(self.report(observation), code) + with patch.object(preflight.time, 'time', return_value=1299.0): + self.assertEqual(self.report(observation)['state'], 'ready_for_review') + for name in ('ALGORITHM_REVISION', 'SCHEMA_REVISION'): + with patch.object(preflight, name, 'different'): + self.assert_blocked(self.report(observation), 'ALGORITHM_CHANGED') + + def test_typed_observation_cannot_be_replayed_or_tampered_with(self) -> None: + observation = self.prepare() + for value in [dataclasses.asdict(observation), None, True, + dataclasses.replace(observation, root_identity=(True, 2, 3)), + dataclasses.replace(observation, issued_at=True), + dataclasses.replace(observation, registration='private-value')]: + self.assert_blocked(preflight.recheck_preflight(self.root, value, 'fixture.json').to_dict(), 'INPUT_INVALID') + changed = dataclasses.replace(observation, digest='f' * 64) + self.assert_blocked(self.report(changed), 'OBSERVATION_CHANGED') + + def test_report_rejects_untrusted_fields_and_has_a_closed_output(self) -> None: + report = preflight.recheck_preflight(self.root, self.prepare(), 'fixture.json') + changed = dataclasses.replace(report, diagnostics=('synthetic-private-value',)) + safe = changed.to_dict() + self.assert_blocked(safe, 'INTERNAL_ERROR') + self.assertNotIn('synthetic-private-value', json.dumps(safe)) + self.assertEqual(set(safe), {'schema_version', 'state', 'observation_digest', 'issued_at', 'expires_at', + 'package', 'registration', 'provenance', 'recovery', 'connection', 'delivery', + 'dispatch_permitted', 'operational_grant', 'diagnostics'}) + + def test_rebuilt_hashes_and_declared_oids_never_prove_provenance_or_recovery(self) -> None: + (self.root / 'deploy/operator.py').write_bytes(b'new unreviewed bytes\n') + self.value['operator']['sha256'] = hashlib.sha256(b'new unreviewed bytes\n').hexdigest() + self.value['operator']['revision']['git'] = {'object_format': 'sha1', 'commit_oid': 'a' * 40, 'tree_oid': 'b' * 40} + self.value['recovery']['protocol_id'] = 'unknown-live-legacy-owner' + self.save() + report = self.report() + self.assertEqual(report['state'], 'ready_for_review') + self.assertEqual(report['provenance'], 'unverified') + self.assertEqual(report['recovery'], 'declared_unverified') + self.assertIs(report['dispatch_permitted'], False) + + def test_no_writes_network_dispatch_private_reads_or_module_file_access(self) -> None: + before = {p.relative_to(self.root).as_posix(): p.read_bytes() for p in self.root.rglob('*') if p.is_file()} + original = fsx.open_at + admitted = {'project-policy.json', 'fixture.json', 'operator.py', 'application.js'} + opened = [] + def read_only(parent, name, flags, *args, **kwargs): + self.assertIn(name, admitted) + self.assertEqual(flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_APPEND), 0) + opened.append(name) + return original(parent, name, flags, *args, **kwargs) + with ExitStack() as stack: + for owner, name in [(socket, 'socket'), (subprocess, 'Popen'), (os, 'system'), + (os, 'unlink'), (os, 'mkdir'), (os, 'rename'), (os, 'replace'), + (Path, 'read_bytes'), (Path, 'read_text'), (Path, 'write_bytes'), (Path, 'write_text')]: + stack.enter_context(patch.object(owner, name, side_effect=AssertionError('forbidden-private-value'))) + stack.enter_context(patch.object(fsx, 'open_at', read_only)) + stack.enter_context(patch.object(preflight, '__file__', '/unavailable/engine.zip/getzilla/deploy_preflight.py')) + report = self.report() + self.assertEqual(report['state'], 'ready_for_review') + self.assertEqual(set(opened), admitted) + self.assertEqual(len(opened), 16) # two reads of four files in both prepare and recheck + after = {p.relative_to(self.root).as_posix(): p.read_bytes() for p in self.root.rglob('*') if p.is_file()} + self.assertEqual(before, after) + + def test_safe_error_output_and_resource_cleanup_on_injected_failure(self) -> None: + observation = self.prepare() + original_close = BoundRoot.close + closed = [] + def close(binding): + original_close(binding) + closed.append(binding) + self.assertFalse(binding._handles) + with patch.object(BoundRoot, 'read', side_effect=RuntimeError('synthetic-private-value')), \ + patch.object(BoundRoot, 'close', close): + report = self.report(observation) + self.assert_blocked(report, 'INTERNAL_ERROR') + self.assertEqual(len(closed), 1) + self.assertNotIn('synthetic-private-value', json.dumps(report)) + with patch.object(BoundRoot, 'read', side_effect=KeyboardInterrupt): + with self.assertRaises(KeyboardInterrupt): + self.report(observation) + + def test_size_and_monotonic_work_budgets_stop_reads(self) -> None: + with patch.object(preflight, 'MAX_FILE_BYTES', 4), self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'FILE_UNSAFE') + with patch.object(preflight, 'MAX_TOTAL_BYTES', 9), self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'INPUT_TOO_LARGE') + with patch.object(preflight.time, 'monotonic', side_effect=[0.0, 11.0]), \ + patch.object(BoundRoot, 'read') as read, self.assertRaises(preflight.PreflightError) as error: + self.prepare() + self.assertEqual(error.exception.code, 'BUDGET_EXCEEDED') + read.assert_not_called() + + +class CredentialFloorTests(unittest.TestCase): + paths = ( + '.codex/auth.json', '.CODEX/AUTH.JSON', 'nested/.codex/auth.json', + '.claude/.credentials.json', '.vault-token', '.config/gh/hosts.yml', '.config/gh/opaque', + '.config/gh', '.composer/auth.json', '.config/composer/auth.json', + '.config/gcloud/opaque', '.azure/opaque', '.terraform.d/credentials.tfrc.json', + '.cargo/credentials.toml', '.gem/credentials', 'credentials.other', 'nested/credentials-data', + 'secrets/opaque.txt', 'nested/secrets/opaque.txt', 'trust-ci/runtime/opaque', + '.codex/auth.json/child', '.env.example', '.ssh/known_hosts', + ) + + def test_manifest_is_rejected_before_any_bound_input_read(self) -> None: + for path in self.paths: + with self.subTest(path=path), patch.object(BoundRoot, 'read') as read: + with self.assertRaises(preflight.PreflightError) as error: + preflight.prepare_preflight(ROOT, path) + self.assertEqual(error.exception.code, 'PATH_UNSAFE') + read.assert_not_called() + + def test_operator_and_artifacts_reject_the_same_credential_floor(self) -> None: + for path in self.paths: + for role in ('operator', 'artifact'): + value = manifest() + target = value['operator'] if role == 'operator' else value['artifacts'][0] + target['path'] = path + with self.subTest(path=path, role=role): + with self.assertRaises(preflight.PreflightError) as error: + preflight.parse_manifest(encoded(value)) + self.assertEqual(error.exception.code, 'PATH_UNSAFE') + + +class PlatformTests(unittest.TestCase): + def test_windows_refuses_posix_mode_claim_before_any_input_open(self) -> None: + with patch.object(preflight.fsx, 'WINDOWS', True), patch.object(preflight, 'BoundRoot') as reader: + with self.assertRaises(preflight.PreflightError) as error: + preflight.prepare_preflight(Path('/uninspected'), 'fixture.json') + self.assertEqual(error.exception.code, 'MODE_UNSUPPORTED') + reader.assert_not_called() + + +class CliTests(unittest.TestCase): + def load_cli(self): + spec = importlib.util.spec_from_file_location('deploy_cli_fixture', ROOT / 'scripts/getzilla_deploy.py') + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + def invoke(self, module, args): + stdout, stderr = io.StringIO(), io.StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + code = module.main(args) + return code, stdout.getvalue(), stderr.getvalue() + + def test_import_is_safe_and_legacy_flat_arguments_are_preserved(self) -> None: + with patch('getzilla.deploy.prepare_deploy', side_effect=AssertionError('import side effect')): + module = self.load_cli() + for args, record in [([], False), (['--json'], False), (['--record', '--json'], True)]: + with patch.object(module, 'find_root', return_value=Path('/fixture-root')) as find, \ + patch.object(module, 'prepare_deploy', return_value={'ok': True, 'commands': ['fixture-command']}) as prepare: + code, stdout, stderr = self.invoke(module, args) + self.assertEqual(code, 0, stderr) + self.assertIn('fixture-command', stdout) + find.assert_called_once_with() + prepare.assert_called_once_with(Path('/fixture-root'), record=record) + + def test_preflight_rejects_bad_arguments_without_echo_or_input_read(self) -> None: + module = self.load_cli() + for args in [[], ['--root', '/fixture'], ['--manifest', 'fixture.json'], + ['--root', '/fixture', '--manifest', 'fixture.json', '--record'], + ['--root', '/fixture', '--manifest', 'fixture.json', 'synthetic-private-value'], + ['--root', '/fixture', '--manifest', 'fixture.json', '--unknown=synthetic-private-value']]: + with patch.object(preflight, 'prepare_preflight', side_effect=AssertionError('input read')), \ + patch.object(module, 'prepare_deploy', side_effect=AssertionError('release effect')), \ + patch.object(module, 'find_root', side_effect=AssertionError('root discovery')): + code, stdout, stderr = self.invoke(module, ['preflight', *args]) + self.assertEqual(code, 2) + self.assertEqual(stdout, '') + self.assertEqual(stderr, 'getzilla deploy preflight: invalid arguments\n') + + def test_help_is_static_and_preflight_failures_have_safe_json(self) -> None: + module = self.load_cli() + stdout = io.StringIO() + with redirect_stdout(stdout), self.assertRaises(SystemExit) as exit_status: + module.main(['preflight', '--help']) + self.assertEqual(exit_status.exception.code, 0) + self.assertIn('--root', stdout.getvalue()) + with patch.object(preflight, 'prepare_preflight', side_effect=preflight.PreflightError('HASH_MISMATCH')), \ + patch.object(module, 'prepare_deploy', side_effect=AssertionError('release effect')), \ + patch.object(module, 'find_root', side_effect=AssertionError('root discovery')): + code, stdout, stderr = self.invoke(module, ['preflight', '--root', '/fixture', '--manifest', 'fixture.json', '--json']) + self.assertEqual(code, 1) + self.assertEqual(stderr, '') + self.assertIn('HASH_MISMATCH', json.loads(stdout)['diagnostics']) + + def test_installed_files_and_zip_engines_from_root_and_subdirectory(self) -> None: + from tests.test_install_on_demand import INSTALL + environment = {key: value for key, value in os.environ.items() + if key not in {'PYTHONPATH', 'PYTHONDONTWRITEBYTECODE', 'PYTHONPYCACHEPREFIX'}} + with tempfile.TemporaryDirectory() as directory: + for archive in (False, True): + with self.subTest(archive=archive): + target = Path(directory) / ('archive' if archive else 'files') + INSTALL._materialize_new(ROOT, target, include_dependencies=False, include_optional=False, + selection=INSTALL.PayloadSelection((), engine_archive=archive)) + for relative in ('schemas/deploy-preflight-manifest-v1.schema.json', 'docs/deploy-preflight.md'): + self.assertEqual((target / relative).read_bytes(), (ROOT / relative).read_bytes()) + if archive: + self.assertFalse((target / '.getzilla/getzilla/deploy_preflight.py').exists()) + else: + self.assertTrue((target / '.getzilla/getzilla/deploy_preflight.py').is_file()) + (target / 'deploy').mkdir() + (target / 'public').mkdir() + (target / 'deploy/operator.py').write_bytes(b'operator\n') + (target / 'public/application.js').write_bytes(b'payload\n') + for relative in ('deploy/operator.py', 'public/application.js'): + (target / relative).chmod(0o644) + (target / preflight.POLICY_PATH).write_bytes(encoded({'secret_consumers': [ + {'script': 'deploy/operator.py', 'secrets': ['.env.fixture']}]})) + (target / 'fixture.json').write_bytes(encoded(manifest())) + nested = target / 'src/deep' + nested.mkdir(parents=True) + (nested / 'fixture.json').write_bytes(b'not the explicit root input') + def snapshot(): + return {p.relative_to(target).as_posix(): + (p.stat().st_mode, hashlib.sha256(p.read_bytes()).hexdigest() if p.is_file() else None) + for p in target.rglob('*')} + before = snapshot() + for cwd, script, prefix in [(target, target / 'scripts/getzilla_deploy.py', []), + (nested, ROOT / 'scripts/getzilla_cli.py', ['deploy'])]: + # -I removes checkout imports and ignores bytecode environment settings; + # deliberately no -B, so accidental import-cache writes are observable. + result = subprocess.run([sys.executable, '-I', str(script), *prefix, 'preflight', + '--root', str(target), '--manifest', 'fixture.json', '--json'], + cwd=cwd, env=environment, capture_output=True, text=True, timeout=30) + self.assertEqual(result.stderr, '') + report = json.loads(result.stdout) + if os.name == 'nt': + self.assertEqual(result.returncode, 1) + self.assertIn('MODE_UNSUPPORTED', report['diagnostics']) + else: + self.assertEqual(result.returncode, 0, result.stdout) + self.assertEqual(report['state'], 'ready_for_review') + self.assertIs(report['dispatch_permitted'], False) + self.assertEqual(report['provenance'], 'unverified') + self.assertEqual(snapshot(), before, 'preflight must not create caches, native state or locks') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_getzilla_cli.py b/tests/test_getzilla_cli.py index 5c843166..b64099e8 100644 --- a/tests/test_getzilla_cli.py +++ b/tests/test_getzilla_cli.py @@ -77,6 +77,18 @@ def test_route_help_exposes_task_and_session_options(self) -> None: for option in ('--session', '--continue', '--new-task'): self.assertIn(option, proc.stdout) + def test_deploy_preflight_arguments_pass_through_unchanged(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + project = Path(tmp) + (project / '.getzilla').mkdir() + (project / 'scripts').mkdir() + (project / 'scripts/getzilla_deploy.py').write_text( + 'import json, sys\nprint(json.dumps(sys.argv[1:]))\nraise SystemExit(23)\n', encoding='utf-8') + arguments = ('preflight', '--root', 'root with spaces', '--manifest', 'public.json', '--json') + proc = run('deploy', *arguments, cwd=project) + self.assertEqual(proc.returncode, 23, proc.stderr) + self.assertEqual(json.loads(proc.stdout), list(arguments)) + def test_every_command_answers_help_with_usage(self) -> None: for name in cli.COMMANDS: with self.subTest(command=name): diff --git a/tests/test_installer.py b/tests/test_installer.py index 3830ff97..67d11286 100644 --- a/tests/test_installer.py +++ b/tests/test_installer.py @@ -677,6 +677,9 @@ def test_payload_is_sorted_safe_duplicate_free_and_profile_explicit(self) -> Non self.assertIn("AGENTS.md", generic_paths) for expected in ( ".getzilla/getzilla/governance.py", + ".getzilla/getzilla/deploy_preflight.py", + "schemas/deploy-preflight-manifest-v1.schema.json", + "docs/deploy-preflight.md", ".getzilla/templates/change/architecture.md", ".getzilla/templates/change/requirements.md", "scripts/getzilla_governance.py", diff --git a/tests/test_manifest_package.py b/tests/test_manifest_package.py index 5192295e..0cea6310 100644 --- a/tests/test_manifest_package.py +++ b/tests/test_manifest_package.py @@ -266,6 +266,9 @@ def test_architecture_tooling_schemas_and_templates_are_packaged(self) -> None: rels = {path.relative_to(ROOT).as_posix() for path in included_files(ROOT)} required = { '.getzilla/getzilla/architecture.py', + '.getzilla/getzilla/deploy_preflight.py', + 'schemas/deploy-preflight-manifest-v1.schema.json', + 'docs/deploy-preflight.md', '.getzilla/getzilla/architecture_diagrams.py', '.getzilla/getzilla/architecture_diff.py', '.getzilla/getzilla/architecture_fitness.py', diff --git a/tests/test_owner_optins.py b/tests/test_owner_optins.py index 636a7d10..4c859a4b 100644 --- a/tests/test_owner_optins.py +++ b/tests/test_owner_optins.py @@ -7,6 +7,7 @@ import sys import unittest from pathlib import Path +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / '.getzilla')) @@ -14,6 +15,7 @@ from getzilla.policy import evaluate_pre_tool # noqa: E402 from getzilla.router import build_route # noqa: E402 from getzilla.state import add_approval +from getzilla import _policy_legacy as legacy from tests._support import activate_owned_route from tests._support import project_copy # noqa: E402 @@ -68,6 +70,36 @@ def test_opted_in_main_still_needs_the_exact_grant(self) -> None: self.assertFalse(allowed) +class ConsumerSelectionTests(unittest.TestCase): + def test_selection_is_pure_and_preserves_filtering_and_unions(self) -> None: + consumers = [None, {}, {'script': 1, 'secrets': ['a']}, {'script': 'a.py', 'secrets': 'bad'}, + {'script': 'a.py', 'secrets': ['a', '', 4, None], 'metadata': 'kept'}, + {'script': 'b.py', 'secrets': ['other']}, {'script': 'a.py', 'secrets': ['b', 'a']}] + called = [] + def matches(script): + called.append(script) + return script == 'a.py' + with patch.object(legacy.os.path, 'realpath', side_effect=AssertionError('unexpected I/O')), \ + patch.object(legacy.os.path, 'isfile', side_effect=AssertionError('unexpected I/O')), \ + patch('builtins.open', side_effect=AssertionError('unexpected I/O')): + result = legacy.consumer_secret_paths(consumers, script_matches=matches) + self.assertEqual(result, frozenset({'a', 'b'})) + self.assertEqual(called, ['a.py', 'b.py', 'a.py']) + for invalid in (None, {}, 'text', (), 2): + self.assertEqual(legacy.consumer_secret_paths(invalid, script_matches=matches), frozenset()) + + def test_legacy_caller_keeps_its_existing_identity_predicate(self) -> None: + root = Path('/synthetic-root') + records = [{'script': 'listed.py', 'secrets': ['selector']}, + {'script': 'other.py', 'secrets': ['another']}] + with patch.object(legacy, 'project_policy', return_value={'secret_consumers': records}), \ + patch.object(legacy, '_same_file', side_effect=[True, False]) as identity: + result = legacy._consumer_handoffs(root, 'python3 actual.py --option value') + self.assertEqual(result, {('python3', 'actual.py', '--option', 'value'): frozenset({'selector'})}) + self.assertEqual([call.args for call in identity.call_args_list], + [(root, 'actual.py', 'listed.py'), (root, 'actual.py', 'other.py')]) + + class SecretConsumerTests(unittest.TestCase): def test_a_listed_script_may_receive_its_secret_and_nothing_else_may(self) -> None: with project_copy(git=True) as root: