diff --git a/docs/development/ENGINEERING_PLATFORM_ROADMAP.md b/docs/development/ENGINEERING_PLATFORM_ROADMAP.md index babd06ff..cf9d0800 100644 --- a/docs/development/ENGINEERING_PLATFORM_ROADMAP.md +++ b/docs/development/ENGINEERING_PLATFORM_ROADMAP.md @@ -477,6 +477,7 @@ The lane is: schema-68 implementation + fixture qualification -> protected review/full EP gates -> 2.3.81 exact-artifact release and authorized installation + -> 2.3.82 installed-preview inventory correction and exact-artifact release -> installed read-only preview -> later explicit coordinated Forge+EP clean-CENTRAL preflight ``` diff --git a/docs/engineering/EP_CENTRAL_OPERATIONAL_RESET_V1.md b/docs/engineering/EP_CENTRAL_OPERATIONAL_RESET_V1.md index e7e35ef8..2e2423a1 100644 --- a/docs/engineering/EP_CENTRAL_OPERATIONAL_RESET_V1.md +++ b/docs/engineering/EP_CENTRAL_OPERATIONAL_RESET_V1.md @@ -2,7 +2,7 @@ **Owning product:** Engineering Platform. **Profile:** `EP_CENTRAL_OPERATIONAL_HISTORY_V1`. **Server schema:** 68. **Candidate -release:** 2.3.81. +release:** 2.3.82. This is a bounded local maintenance route for removing EP operational history while retaining installation identity, project and repository attachment, @@ -163,6 +163,45 @@ classification. An arbitrary file hidden inside `runtime/`, `operations/`, blocks prepare without being moved or deleted. A pending CENTRAL import is an explicit active-ingest blocker. +An exact `operations//candidate-venv/` below a parseable owning +`operation.json` and `candidate-runtime.json` is one opaque +`INSTALLATION_RUNTIME` preserve boundary only when both records bind the same +operation, the currently inspected CENTRAL installation identity, the exact +owning update steps/cleanup roots, plan and candidate path. The +preview records the directory boundary but does not descend into, hash, copy or +follow its contents; normal Python virtual-environment links therefore do not +become reset targets or false symlink findings. The `candidate-venv` directory +itself must be a real directory. A linked venv root, a venv-shaped directory +without its owning identity records, malformed or path-conflicting records, +symlinks in other staging trees and unclassified sibling data remain +fail-closed. + +The target identity is a non-empty canonical UUID and must match independently +in `runtime-identity.json`, `ep_installations` and +`engineering_metadata['installation.instance_id']`. Empty or malformed values +and any pairwise disagreement block preview before updater staging can be +classified. Marker, journal and prepared-candidate installation IDs must equal +that same verified target UUID. + +Candidate preparation intentionally precedes journal creation. In that exact +crash window, a closed-schema `candidate-runtime.json` may be the only durable +owner record. If it binds the parent operation ID, installation, exact +`candidate-venv`, contained staged wheel, target version/digest and source +revision—and the staged wheel's bytes and canonical filename verify against +that digest/identity—preview preserves the exact `candidate-venv`, `download` and +`pip-cache` boundaries opaquely as +`INSTALLATION_RUNTIME_STAGING_UNBOUND` and reports +`INCOMPLETE_NO_OPERATION_JOURNAL`. It does not follow or copy their contents. +This fallback applies only while `operation.json` is genuinely absent; +malformed, linked or conflicting records fail closed. Any sibling outside +those exact updater staging boundaries remains unknown. + +Every accepted opaque directory is held through a non-following directory +descriptor. Immediately before the inventory receipt returns, preview requires +both that pinned descriptor and the current path to remain the same directory +device/inode. A directory-to-link replacement, including one after initial +classification, fails closed without publishing a stale preserve receipt. + ## Protected backup and recovery `prepare` requires an operator-selected backup root outside the active data @@ -462,8 +501,11 @@ decision. ## Delivery status -At source-candidate creation this slice is `IMPLEMENTED` and locally fixture -qualified, while protected PR review, hosted full gates, release publication, -artifact-byte qualification and installed live preview remain separate +The installed 2.3.81 read-only preview proved schema 68, `quick_check`, foreign +keys and counts, but exposed a false `EXTERNAL_SYMLINK_UNSAFE` blocker by +descending into a preserved updater `candidate-venv`. Release 2.3.82 contains +the targeted opaque-boundary correction and synthetic positive/negative +qualification. Protected PR review, hosted full gates, release publication, +artifact-byte qualification and a new installed live preview remain separate evidence. Installation may activate schema 68 and the chat relationship repair but must not automatically prepare/apply a reset or delete historical rows. diff --git a/docs/engineering/EP_SERVER_PRODUCTION_RELEASE.md b/docs/engineering/EP_SERVER_PRODUCTION_RELEASE.md index 59b86279..345c0d3a 100644 --- a/docs/engineering/EP_SERVER_PRODUCTION_RELEASE.md +++ b/docs/engineering/EP_SERVER_PRODUCTION_RELEASE.md @@ -17,6 +17,24 @@ maintenance controls; neither publication nor installation authorizes artifact qualification and an installed read-only preview are required before the capability can be reported installed. +The 2.3.82 corrective patch keeps the same schema and reset semantics. It +prevents read-only maintenance preview from recursively entering an exact +updater-owned `operations//candidate-venv` and falsely rejecting normal +virtual-environment symlinks. The venv is preserved as an opaque runtime +boundary only when the journal and candidate marker bind its exact operation, +current CENTRAL installation, closed updater step/cleanup contract and path. +Opaque directories stay descriptor-pinned until the final inventory receipt, +so a path swap cannot publish stale classification. All malformed, +unrecognized and active-route symlink protections remain in force. Publication +and installation still do not authorize a reset. + +The preview also recognizes the updater's bounded pre-journal crash window. A +closed candidate marker may preserve only its exact candidate/download/cache +staging boundaries as opaque `INSTALLATION_RUNTIME_STAGING_UNBOUND` after its +canonical staged wheel filename and digest are verified, with an explicit +incomplete-state readback. It never turns an arbitrary operation sibling, +malformed marker or linked boundary into known product data. + ## Release-operation lifecycle V1 After both the production-wheel and dashboard qualifications succeed, the diff --git a/package-lock.json b/package-lock.json index e336ceab..120f6935 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "engineering-platform-browser-validation", - "version": "2.3.81", + "version": "2.3.82", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "engineering-platform-browser-validation", - "version": "2.3.81", + "version": "2.3.82", "devDependencies": { "@playwright/test": "1.63.0" } diff --git a/package.json b/package.json index 703e3b40..ac073ab3 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "engineering-platform-browser-validation", "private": true, - "version": "2.3.81", + "version": "2.3.82", "scripts": { "test:engineering-dashboard": "PYTHONPATH=src python3 -m engineering_platform.dashboard_browser_validation", "test:engineering-dashboard-logic": "node --test tests/engineering/dashboard_status_store.test.mjs tests/engineering/ui_localization_contract.test.mjs tests/engineering/dashboard_translation.test.mjs", diff --git a/pyproject.toml b/pyproject.toml index 4965dbdc..d61d6430 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "engineering-platform" -version = "2.3.81" +version = "2.3.82" description = "Local-first Engineering Platform execution operations runtime" readme = "README.md" requires-python = ">=3.14,<3.15" diff --git a/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json b/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json index ab357d7e..a8af789e 100644 --- a/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json +++ b/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json @@ -3,7 +3,7 @@ "platform": { "id": "engineering-platform", "name": "Engineering Platform", - "version": "2.3.81", + "version": "2.3.82", "generation": 2, "documentation_namespace": "engineering-platform", "capability_registry_version": 1 diff --git a/src/engineering_platform/ENGINEERING_PLATFORM_STATUS.md b/src/engineering_platform/ENGINEERING_PLATFORM_STATUS.md index d04fc99f..35a6971e 100644 --- a/src/engineering_platform/ENGINEERING_PLATFORM_STATUS.md +++ b/src/engineering_platform/ENGINEERING_PLATFORM_STATUS.md @@ -75,6 +75,24 @@ changes execution semantics. It targets the installation-owned CENTRAL database, executes `PRAGMA optimize` and `VACUUM` only when no CENTRAL lifecycle is active, and never opens, compacts or configures a project-local database. +- The installation-owner maintenance preview inventories preserved updater + operations without entering a `candidate-venv` whose owning journal and + candidate marker bind the exact operation, current CENTRAL installation, + owning update steps/cleanup roots and path. That venv remains a + descriptor-pinned opaque installation-runtime boundary through final receipt + publication, while malformed or unbound records, path swaps, linked venv + roots, symlinks elsewhere, active ingest roots and unknown sibling data + remain fail-closed. +- Operational-reset identity readback independently requires the same non-empty + canonical UUID in the runtime identity, installation table and installation + metadata. Empty, malformed or pairwise-conflicting values cannot authorize + opaque updater classification. +- A candidate prepared before its update journal was durably created is + read back as `INSTALLATION_RUNTIME_STAGING_UNBOUND` only from its exact + closed-schema candidate marker. Its candidate/download/cache boundaries are + preserved opaquely only after the canonical staged wheel identity and digest + verify, with an explicit incomplete state; arbitrary siblings and malformed, + missing or conflicting markers remain blocking. - During an active run, a fully successful specialist review remains visible as compact historical evidence after the review phase. Partial, failed or running reviewer projections remain phase-scoped and are never presented as diff --git a/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json b/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json index 133f80c7..07cf3642 100644 --- a/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json +++ b/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json @@ -1,15 +1,15 @@ { "bootstrap_contract": "2026.12", "checkpoint_format": 1, - "dashboard_version": "2.3.81", + "dashboard_version": "2.3.82", "handoff_protocol": 1, "memory_format": 2, "minimum_codex_cli": "0.146.0", "inbox_protocol": 1, - "platform_version": "2.3.81", + "platform_version": "2.3.82", "report_format": 2, - "runner_version": "2.3.81", + "runner_version": "2.3.82", "status_model": 1, "storage_schema": 45, - "watcher_version": "2.3.81" + "watcher_version": "2.3.82" } diff --git a/src/engineering_platform/central_operational_reset.py b/src/engineering_platform/central_operational_reset.py index cb58adae..57173487 100644 --- a/src/engineering_platform/central_operational_reset.py +++ b/src/engineering_platform/central_operational_reset.py @@ -36,6 +36,14 @@ PLAN_VERSION = 1 SCHEMA_VERSION = 68 _OPERATION = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{7,127}") +_INSTANCE_ID = re.compile( + r"[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}" +) +_UPDATE_STEPS = ( + "INSTALLATION_LOCK", "INVENTORY_AND_COMPATIBILITY", "EXACT_ARTIFACT", + "QUIESCE", "BACKUP_AND_MIGRATION", "ACTIVATE", "VERIFY", "CLEANUP", +) +_UPDATE_CLEANUP_DIRECTORIES = ("build", "download", "pip-cache") _ACTIVE_STATES = frozenset({ "PREPARING", "AUTHORIZED", "ARTIFACTS_ARCHIVING", "ARTIFACTS_ARCHIVED", "DB_APPLIED", "VERIFIED", "FAILED", @@ -469,6 +477,10 @@ def _actor(data_root: Path) -> str: return f"uid:{uid}:{name}" +def _valid_instance_id(value: object) -> bool: + return isinstance(value, str) and _INSTANCE_ID.fullmatch(value) is not None + + def _identity(data_root: Path, connection: sqlite3.Connection) -> dict[str, object]: try: file_identity = json.loads((data_root / "runtime-identity.json").read_text(encoding="utf-8")) @@ -478,16 +490,22 @@ def _identity(data_root: Path, connection: sqlite3.Connection) -> dict[str, obje metadata = connection.execute( "SELECT value FROM engineering_metadata WHERE key='installation.instance_id'" ).fetchone() + runtime_instance = file_identity.get("instance_id") if isinstance(file_identity, dict) else None + database_instance = row[0] if row is not None else None + metadata_instance = metadata[0] if metadata is not None else None if ( - not isinstance(file_identity, dict) or not isinstance(file_identity.get("instance_id"), str) - or row is None or metadata is None or file_identity["instance_id"] != row[0] != "" - or row[0] != metadata[0] + not _valid_instance_id(runtime_instance) + or not _valid_instance_id(database_instance) + or not _valid_instance_id(metadata_instance) + or runtime_instance != database_instance + or runtime_instance != metadata_instance + or database_instance != metadata_instance ): raise OperationalResetError("TARGET_IDENTITY_CONFLICT") database = (data_root / central_database.DATABASE_FILENAME).resolve() stat = database.stat() return { - "product": "engineering-platform", "instance_id": str(row[0]), + "product": "engineering-platform", "instance_id": str(database_instance), "data_root": str(data_root), "database": str(database), "schema_version": int(row[1]), "database_device": stat.st_dev, "database_inode": stat.st_ino, } @@ -584,11 +602,408 @@ def _safe_files(directory: Path, logical_root: str) -> list[dict[str, object]]: return rows +def _candidate_runtime_marker( + operation_root: Path, *, expected_installation_id: str, + require_staged_artifact: bool, +) -> dict[str, object] | None: + """Read the updater's exact, closed, token-free candidate identity.""" + if not _valid_instance_id(expected_installation_id): + return None + operation_id = operation_root.name + marker = _regular_json_object(operation_root / "candidate-runtime.json") + if marker is None or _OPERATION.fullmatch(operation_id) is None: + return None + expected_candidate = str(operation_root / "candidate-venv") + if ( + set(marker) != { + "schema_version", "operation_id", "installation_id", "target_version", + "target_digest", "target_source_revision", "staged_artifact", "candidate_venv", + } + or marker.get("schema_version") != 1 + or marker.get("operation_id") != operation_id + or not _valid_instance_id(marker.get("installation_id")) + or marker.get("installation_id") != expected_installation_id + or marker.get("candidate_venv") != expected_candidate + or re.fullmatch(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", + str(marker.get("target_version", ""))) is None + or re.fullmatch(r"sha256:[0-9a-f]{64}", str(marker.get("target_digest", ""))) is None + or re.fullmatch(r"[0-9a-f]{40}", str(marker.get("target_source_revision", ""))) is None + ): + return None + staged = marker.get("staged_artifact") + if not isinstance(staged, str): + return None + staged_path = Path(staged) + if ( + not staged_path.is_absolute() + or staged_path != Path(os.path.normpath(staged)) + or staged_path.parent != operation_root / "download" + or staged_path.name != ( + f"engineering_platform-{marker['target_version']}-py3-none-any.whl" + ) + ): + return None + if require_staged_artifact: + try: + metadata = staged_path.lstat() + except OSError: + return None + if ( + not stat.S_ISREG(metadata.st_mode) + or _regular_file_digest(staged_path) != marker["target_digest"] + ): + return None + return marker + + +def _path_is_absent(path: Path) -> bool: + """Distinguish an absent path from a dangling link or unreadable entry.""" + try: + path.lstat() + except FileNotFoundError: + return True + except OSError: + return False + return False + + +def _unbound_candidate_runtime_marker( + operation_root: Path, *, expected_installation_id: str, +) -> dict[str, object] | None: + """Return a closed candidate marker only in the pre-journal crash window.""" + if not _path_is_absent(operation_root / "operation.json"): + return None + return _candidate_runtime_marker( + operation_root, expected_installation_id=expected_installation_id, + require_staged_artifact=True, + ) + + +def _opaque_update_directory_classification( + root_name: str, directory: Path, relative: str, *, expected_installation_id: str, +) -> str | None: + """Recognize only an identity-bound updater runtime/staging directory. + + Merely finding ``operation.json`` beside a venv-shaped directory is not + ownership evidence: an arbitrary file must not turn unknown application + data into an opaque preserve boundary. The journal and the updater's + candidate marker therefore have to bind the same operation, installation, + plan and exact paths before normal Python venv symlinks become opaque. + """ + parts = Path(relative).parts + if ( + root_name != "operations" or len(parts) != 2 + or _OPERATION.fullmatch(parts[0]) is None + or parts[1] not in {"candidate-venv", "download", "pip-cache"} + ): + return None + + operation_id = parts[0] + operation_root = directory.parent + unbound_marker = _unbound_candidate_runtime_marker( + operation_root, expected_installation_id=expected_installation_id, + ) + if unbound_marker is not None: + return "INSTALLATION_RUNTIME_STAGING_UNBOUND" + if parts[1] != "candidate-venv": + return None + data_root = operation_root.parent.parent + journal = _regular_json_object(operation_root / "operation.json") + marker = _candidate_runtime_marker( + operation_root, expected_installation_id=expected_installation_id, + require_staged_artifact=False, + ) + if journal is None: + return None + if marker is None: + return None + schema = journal.get("schema_version") + base_fields = { + "schema_version", "operation_id", "plan", "plan_digest", "state", "events", + } + fields_by_schema = { + 1: base_fields, + 2: base_fields | {"prepared_candidate", "prepared_candidate_digest"}, + 3: base_fields | { + "prepared_candidate", "prepared_candidate_digest", + "prepared_record_provenance", "prepared_record_provenance_digest", + }, + 4: base_fields | { + "prepared_candidate", "prepared_candidate_digest", + "prepared_record_provenance", "prepared_record_provenance_digest", + "execution_admission", "execution_admission_digest", + }, + } + if ( + type(schema) is not int + or set(journal) != fields_by_schema.get(schema) + or not isinstance(journal.get("events"), list) + or journal.get("state") not in { + "PREPARED", "INVENTORIED", "QUIESCING", "QUIESCED", "BACKED_UP", + "MIGRATED", "ACTIVATED", "VERIFIED", "CLEANUP_PENDING", "COMPLETE", + } + ): + return None + plan = journal.get("plan") + if not isinstance(plan, dict): + return None + plan_fields = { + "operation_id", "installation_id", "data_root", "current_version", + "current_digest", "target_version", "target_digest", "target_source_revision", + "artifact", "cleanup_targets", "steps", + } + installation_id = plan.get("installation_id") + expected_cleanup = [ + str(operation_root / name) for name in _UPDATE_CLEANUP_DIRECTORIES + ] + if ( + frozenset(plan) not in {frozenset(plan_fields), frozenset(plan_fields | {"legacy_adoption"})} + or journal.get("operation_id") != operation_id + or plan.get("operation_id") != operation_id + or installation_id != expected_installation_id + or plan.get("data_root") != str(data_root) + or plan.get("cleanup_targets") != expected_cleanup + or plan.get("steps") != list(_UPDATE_STEPS) + or re.fullmatch(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", + str(plan.get("current_version", ""))) is None + or re.fullmatch(r"sha256:[0-9a-f]{64}", str(plan.get("current_digest", ""))) is None + ): + return None + try: + canonical_plan = json.dumps( + plan, sort_keys=True, separators=(",", ":"), + ).encode("utf-8") + except (TypeError, ValueError): + return None + if journal.get("plan_digest") != "sha256:" + hashlib.sha256(canonical_plan).hexdigest(): + return None + + expected_candidate = str(directory) + expected_operation = str(operation_root) + expected_interpreter = str(directory / "bin" / "python") + if ( + set(marker) != { + "schema_version", "operation_id", "installation_id", "target_version", + "target_digest", "target_source_revision", "staged_artifact", "candidate_venv", + } + or marker.get("schema_version") != 1 + or marker.get("operation_id") != operation_id + or marker.get("installation_id") != installation_id + or marker.get("candidate_venv") != expected_candidate + or marker.get("target_version") != plan.get("target_version") + or marker.get("target_digest") != plan.get("target_digest") + or marker.get("target_source_revision") != plan.get("target_source_revision") + or marker.get("staged_artifact") != plan.get("artifact") + ): + return None + staged_artifact = marker.get("staged_artifact") + if ( + not isinstance(staged_artifact, str) + or Path(staged_artifact).parent != operation_root / "download" + ): + return None + + prepared = journal.get("prepared_candidate") + if prepared is None: + return ( + "INSTALLATION_RUNTIME" + if journal.get("prepared_candidate_digest") is None else None + ) + if not isinstance(prepared, dict): + return None + package = prepared.get("package") + if ( + set(prepared) != { + "operation_id", "installation_id", "operation_root", "staged_artifact", + "artifact_digest", "candidate_venv", "interpreter", "pip_cache", "package", + } + or prepared.get("operation_id") != operation_id + or prepared.get("installation_id") != installation_id + or prepared.get("operation_root") != expected_operation + or prepared.get("candidate_venv") != expected_candidate + or prepared.get("interpreter") != expected_interpreter + or prepared.get("staged_artifact") != staged_artifact + or prepared.get("artifact_digest") != plan.get("target_digest") + or prepared.get("pip_cache") != str(operation_root / "pip-cache") + or not isinstance(package, dict) + or set(package) != {"interpreter", "version", "metadata", "package"} + or package.get("interpreter") != expected_interpreter + or package.get("version") != plan.get("target_version") + or not all(isinstance(package.get(field), str) and package[field] + for field in ("version", "metadata", "package")) + or not all(_lexically_within(directory, Path(str(package[field]))) + for field in ("metadata", "package")) + ): + return None + try: + canonical_prepared = json.dumps( + prepared, sort_keys=True, separators=(",", ":"), + ).encode("utf-8") + except (TypeError, ValueError): + return None + if journal.get("prepared_candidate_digest") != ( + "sha256:" + hashlib.sha256(canonical_prepared).hexdigest() + ): + return None + return "INSTALLATION_RUNTIME" + + +def _lexically_within(root: Path, candidate: Path) -> bool: + """Check an already persisted absolute path without resolving any link.""" + if ( + not candidate.is_absolute() + or candidate != Path(os.path.normpath(str(candidate))) + ): + return False + try: + candidate.relative_to(root) + except ValueError: + return False + return candidate != root + + +def _regular_json_object(path: Path) -> dict[str, object] | None: + """Read one bounded regular JSON file without following its final link.""" + descriptor = -1 + try: + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 4 * 1024 * 1024: + return None + with os.fdopen(descriptor, "r", encoding="utf-8") as stream: + descriptor = -1 + value = json.load(stream) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + finally: + if descriptor >= 0: + os.close(descriptor) + return value if isinstance(value, dict) else None + + +def _regular_file_digest(path: Path) -> str | None: + """Hash one regular file while refusing a final-component link swap.""" + descriptor = -1 + try: + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + return None + digest = hashlib.sha256() + with os.fdopen(descriptor, "rb") as stream: + descriptor = -1 + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return "sha256:" + digest.hexdigest() + except OSError: + return None + finally: + if descriptor >= 0: + os.close(descriptor) + + +def _walk_preserved_files( + directory: Path, *, root_name: str, code: str, expected_installation_id: str, + opaque_bindings: list[tuple[Path, str, int, int, int]], +) -> tuple[list[Path], list[tuple[Path, str, int, int, int]]]: + """Walk preserved product data while treating exact updater venvs as opaque.""" + if not directory.exists(): + return [], [] + if directory.is_symlink() or not directory.is_dir(): + raise OperationalResetError(code) + files: list[Path] = [] + opaque: list[tuple[Path, str, int, int, int]] = [] + pending = [directory] + while pending: + current = pending.pop() + try: + entries = sorted(os.scandir(current), key=lambda entry: entry.name) + except OSError as error: + raise OperationalResetError(code) from error + for entry in entries: + try: + path = Path(entry.path) + metadata = path.lstat() + if stat.S_ISLNK(metadata.st_mode): + raise OperationalResetError(code) + if stat.S_ISDIR(metadata.st_mode): + relative = path.relative_to(directory).as_posix() + classification = _opaque_update_directory_classification( + root_name, path, relative, + expected_installation_id=expected_installation_id, + ) + if classification is not None: + descriptor = _open_opaque_directory( + path, device=metadata.st_dev, inode=metadata.st_ino, + ) + binding = ( + path, classification, metadata.st_dev, metadata.st_ino, + descriptor, + ) + opaque.append(binding) + opaque_bindings.append(binding) + else: + pending.append(path) + elif stat.S_ISREG(metadata.st_mode): + files.append(path) + else: + raise OperationalResetError(code) + except OSError as error: + raise OperationalResetError(code) from error + return sorted(files), sorted(opaque, key=lambda item: item[0]) + + +def _open_opaque_directory(path: Path, *, device: int, inode: int) -> int: + """Pin the accepted real directory until the inventory is published.""" + descriptor = -1 + try: + descriptor = os.open( + path, + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), + ) + metadata = os.fstat(descriptor) + if ( + not stat.S_ISDIR(metadata.st_mode) + or metadata.st_dev != device + or metadata.st_ino != inode + ): + raise OperationalResetError("EXTERNAL_SYMLINK_UNSAFE") + return descriptor + except OSError as error: + if descriptor >= 0: + os.close(descriptor) + raise OperationalResetError("EXTERNAL_SYMLINK_UNSAFE") from error + except BaseException: + if descriptor >= 0: + os.close(descriptor) + raise + + +def _recheck_opaque_directory( + path: Path, *, device: int, inode: int, descriptor: int, +) -> None: + """Bind the final opaque receipt to the still-pinned original directory.""" + try: + pinned = os.fstat(descriptor) + current = path.lstat() + except OSError as error: + raise OperationalResetError("EXTERNAL_SYMLINK_UNSAFE") from error + if ( + not stat.S_ISDIR(pinned.st_mode) + or not stat.S_ISDIR(current.st_mode) + or pinned.st_dev != device or pinned.st_ino != inode + or current.st_dev != device or current.st_ino != inode + ): + raise OperationalResetError("EXTERNAL_SYMLINK_UNSAFE") + + def _known_top_level(name: str) -> bool: return name in _KNOWN_TOP_LEVEL or _VERSIONED_RECOVERY_BACKUP.fullmatch(name) is not None -def _nested_preserved_classification(root_name: str, relative: str, path: Path) -> str | None: +def _nested_preserved_classification( + root_name: str, relative: str, path: Path, *, expected_installation_id: str, +) -> str | None: """Classify only product-owned shapes inside preserved top-level roots.""" parts = Path(relative).parts if root_name == "runtime": @@ -605,12 +1020,22 @@ def _nested_preserved_classification(root_name: str, relative: str, path: Path) return "INSTALLATION_OPERATION_AUDIT" if parts and _OPERATION.fullmatch(parts[0]) is not None: operation_root = path.parents[len(parts) - 2] + if _unbound_candidate_runtime_marker( + operation_root, expected_installation_id=expected_installation_id, + ) is not None: + if parts[1:] == ("candidate-runtime.json",): + return "INSTALLATION_RUNTIME_STAGING_UNBOUND" + # The exact staging subtrees are normally opaque. This branch + # also keeps a race/retry readback classified if a leaf was + # enumerated just before the boundary recognition. + if len(parts) >= 3 and parts[1] in {"download", "pip-cache"}: + return "INSTALLATION_RUNTIME_STAGING_UNBOUND" if (operation_root / "operation.json").is_file() and len(parts) >= 2: if parts[1] in {"operation.json", "candidate-runtime.json"} and len(parts) == 2: return "INSTALLATION_OPERATION_AUDIT" if parts[1:] == ("backup", "central.sqlite"): return "FORENSIC_OR_RECOVERY" - if parts[1] in {"build", "download", "pip-cache", "candidate-venv"}: + if parts[1] in {"build", "download", "pip-cache"}: return "INSTALLATION_RUNTIME_STAGING" return None if root_name == "migration": @@ -651,8 +1076,9 @@ def _nested_preserved_classification(root_name: str, relative: str, path: Path) return None -def _external_inventory( - data_root: Path, +def _external_inventory_bound( + data_root: Path, *, expected_installation_id: str, + opaque_bindings: list[tuple[Path, str, int, int, int]], ) -> tuple[list[dict[str, object]], list[dict[str, object]], list[str], list[str]]: unknown = sorted(path.name for path in data_root.iterdir() if not _known_top_level(path.name)) rows: list[dict[str, object]] = [] @@ -693,11 +1119,30 @@ def _external_inventory( "runtime", "operations", "recovery", "migration", "backups", "operational-reset-archive", }: - for nested in _walk_regular_files(path, code="EXTERNAL_SYMLINK_UNSAFE"): + nested_files, opaque_boundaries = _walk_preserved_files( + path, root_name=path.name, code="EXTERNAL_SYMLINK_UNSAFE", + expected_installation_id=expected_installation_id, + opaque_bindings=opaque_bindings, + ) + for boundary, classification, _device, _inode, _descriptor in opaque_boundaries: + relative = boundary.relative_to(path).as_posix() + entry = { + "path": f"{path.name}/{relative}", + "classification": classification, + "effect": "PRESERVE", "kind": "directory", + "opaque_boundary": True, "symlinks_followed": False, + } + if classification == "INSTALLATION_RUNTIME_STAGING_UNBOUND": + entry["installation_update_state"] = "INCOMPLETE_NO_OPERATION_JOURNAL" + preserved.append(entry) + for nested in nested_files: relative = nested.relative_to(path).as_posix() if path.name == "runtime" and relative.startswith("central-data-imports/"): continue - classification = _nested_preserved_classification(path.name, relative, nested) + classification = _nested_preserved_classification( + path.name, relative, nested, + expected_installation_id=expected_installation_id, + ) if classification is None: unknown.append(f"{path.name}/{relative}") continue @@ -711,6 +1156,34 @@ def _external_inventory( return rows, preserved, sorted(set(unknown)), active_ingest_controls +def _external_inventory( + data_root: Path, *, expected_installation_id: str, +) -> tuple[list[dict[str, object]], list[dict[str, object]], list[str], list[str]]: + """Inventory external data while pinning every published opaque boundary.""" + if not _valid_instance_id(expected_installation_id): + raise OperationalResetError("TARGET_IDENTITY_CONFLICT") + bindings: list[tuple[Path, str, int, int, int]] = [] + try: + result = _external_inventory_bound( + data_root, expected_installation_id=expected_installation_id, + opaque_bindings=bindings, + ) + # This is intentionally the final work before returning the receipt. + # A path swapped after initial acceptance cannot be represented as the + # pinned directory that was actually classified. + for path, _classification, device, inode, descriptor in bindings: + _recheck_opaque_directory( + path, device=device, inode=inode, descriptor=descriptor, + ) + return result + finally: + for _path, _classification, _device, _inode, descriptor in bindings: + try: + os.close(descriptor) + except OSError: + pass + + def _runtime_activity(data_root: Path) -> dict[str, object]: try: payload = json.loads((data_root / "runtime.json").read_text(encoding="utf-8")) @@ -807,7 +1280,9 @@ def preview(data_root: Path) -> dict[str, object]: raise OperationalResetError("CENTRAL_INSPECTION_FAILED") from error external_error: str | None = None try: - external, preserved_external, unknown_paths, active_ingest_controls = _external_inventory(root) + external, preserved_external, unknown_paths, active_ingest_controls = _external_inventory( + root, expected_installation_id=str(identity["instance_id"]), + ) except OperationalResetError as error: external, preserved_external, unknown_paths, active_ingest_controls = [], [], [], [] external_error = error.code @@ -1725,6 +2200,7 @@ def _verify_under_lock( credentials = int(connection.execute("SELECT COUNT(*) FROM ep_consumer_credentials").fetchone()[0]) projects = int(connection.execute("SELECT COUNT(*) FROM ep_project_registrations").fetchone()[0]) repositories = int(connection.execute("SELECT COUNT(*) FROM ep_repository_registrations").fetchone()[0]) + expected_installation_id = str(_identity(root, connection)["instance_id"]) archive = root / "operational-reset-archive" / operation_id boundary: dict[str, object] | None = None boundary_path_value = row.get("finish_boundary_path") @@ -1750,7 +2226,9 @@ def _verify_under_lock( raise OperationalResetError("FINISH_BOUNDARY_INVALID") if not allow_boundary_recovery: frozen_modes = _verify_frozen_empty_active_roots(root) - external, _preserved, unknown, active_controls = _external_inventory(root) + external, _preserved, unknown, active_controls = _external_inventory( + root, expected_installation_id=expected_installation_id, + ) backup = verify_backup( Path(str(row["backup_path"])), operation_id=operation_id, plan_digest=plan_digest, expected_manifest_digest=str(row["backup_manifest_digest"]), diff --git a/src/engineering_platform/platform_version.py b/src/engineering_platform/platform_version.py index 3228ad1d..7db9d044 100644 --- a/src/engineering_platform/platform_version.py +++ b/src/engineering_platform/platform_version.py @@ -12,7 +12,7 @@ SEMVER = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$") CONTRACT = re.compile(r"^(\d{4})\.(0[1-9]|1[0-2])$") -CURRENT_PLATFORM_VERSION = "2.3.81" +CURRENT_PLATFORM_VERSION = "2.3.82" MANIFEST_FIELDS = frozenset( { "platform_version", diff --git a/src/engineering_platform/templates/workspace-config.json b/src/engineering_platform/templates/workspace-config.json index fc44f87f..fc681316 100644 --- a/src/engineering_platform/templates/workspace-config.json +++ b/src/engineering_platform/templates/workspace-config.json @@ -3,7 +3,7 @@ "platform": { "id": "engineering-platform", "name": "Engineering Platform", - "version": "2.3.81", + "version": "2.3.82", "generation": 2, "documentation_namespace": "engineering-platform", "capability_registry_version": 1 diff --git a/tests/engineering/test_central_operational_reset.py b/tests/engineering/test_central_operational_reset.py index 3150a0fb..4f7fb34f 100644 --- a/tests/engineering/test_central_operational_reset.py +++ b/tests/engineering/test_central_operational_reset.py @@ -83,6 +83,73 @@ def _prepared(self, *, operation_id: str = "reset-fixture-0001") -> tuple[str, s ) return operation_id, str(plan["plan_digest"]) + def _bind_update_candidate_fixture(self, operation: Path) -> None: + """Write the exact token-free identities that own one candidate venv.""" + operation_id = operation.name + installation_id = str(json.loads( + (self.root / "runtime-identity.json").read_text(encoding="utf-8") + )["instance_id"]) + staged = operation / "download" / "engineering_platform-2.3.82-py3-none-any.whl" + candidate = operation / "candidate-venv" + plan = { + "operation_id": operation_id, + "installation_id": installation_id, + "data_root": str(self.root), + "current_version": "2.3.81", + "current_digest": "sha256:" + "a" * 64, + "target_version": "2.3.82", + "target_digest": "sha256:" + "b" * 64, + "target_source_revision": "c" * 40, + "artifact": str(staged), + "cleanup_targets": [ + str(operation / name) for name in ("build", "download", "pip-cache") + ], + "steps": [ + "INSTALLATION_LOCK", "INVENTORY_AND_COMPATIBILITY", "EXACT_ARTIFACT", + "QUIESCE", "BACKUP_AND_MIGRATION", "ACTIVATE", "VERIFY", "CLEANUP", + ], + } + prepared = { + "operation_id": operation_id, + "installation_id": installation_id, + "operation_root": str(operation), + "staged_artifact": str(staged), + "artifact_digest": plan["target_digest"], + "candidate_venv": str(candidate), + "interpreter": str(candidate / "bin" / "python"), + "pip_cache": str(operation / "pip-cache"), + "package": { + "interpreter": str(candidate / "bin" / "python"), + "version": "2.3.82", + "metadata": str(candidate / "lib" / "python3.14" / "site-packages" / "engineering_platform-2.3.82.dist-info"), + "package": str(candidate / "lib" / "python3.14" / "site-packages" / "engineering_platform"), + }, + } + canonical_plan = json.dumps(plan, sort_keys=True, separators=(",", ":")).encode() + canonical_prepared = json.dumps( + prepared, sort_keys=True, separators=(",", ":"), + ).encode() + (operation / "operation.json").write_text(json.dumps({ + "schema_version": 2, + "operation_id": operation_id, + "plan": plan, + "plan_digest": "sha256:" + hashlib.sha256(canonical_plan).hexdigest(), + "state": "PREPARED", + "events": [{"state": "PREPARED", "evidence": {}}], + "prepared_candidate": prepared, + "prepared_candidate_digest": "sha256:" + hashlib.sha256(canonical_prepared).hexdigest(), + }), encoding="utf-8") + (operation / "candidate-runtime.json").write_text(json.dumps({ + "schema_version": 1, + "operation_id": operation_id, + "installation_id": installation_id, + "target_version": plan["target_version"], + "target_digest": plan["target_digest"], + "target_source_revision": plan["target_source_revision"], + "staged_artifact": str(staged), + "candidate_venv": str(candidate), + }), encoding="utf-8") + def test_schema_owned_inventory_is_exhaustive_and_preview_is_read_only(self) -> None: before = hashlib.sha256((self.root / "epdata.sqlite").read_bytes()).hexdigest() plan = reset.preview(self.root) @@ -807,6 +874,376 @@ def test_nested_preserved_external_inventory_is_explicit_and_unknown_appdata_blo self.assertIn("runtime/user-payload.bin", blocked["unknown_external_paths"]) self.assertEqual(unknown.read_bytes(), b"keep") + def test_updater_candidate_venv_is_an_opaque_preserved_runtime_boundary(self) -> None: + operation = self.root / "operations" / "update-live-preview-0001" + operation.mkdir(parents=True) + venv = operation / "candidate-venv" + (venv / "bin").mkdir(parents=True) + (venv / "lib" / "python3.14" / "site-packages").mkdir(parents=True) + (venv / "bin" / "python").symlink_to("python3.14") + (venv / "bin" / "python3.14").symlink_to("/usr/bin/python3") + (venv / "lib" / "python3.14" / "site-packages" / "installed.py").write_text( + "# updater-owned fixture\n", encoding="utf-8", + ) + self._bind_update_candidate_fixture(operation) + + plan = reset.preview(self.root) + + self.assertNotIn("EXTERNAL_SYMLINK_UNSAFE", plan["blocking_codes"]) + self.assertNotIn("EXTERNAL_CLASSIFICATION_INCOMPLETE", plan["blocking_codes"]) + boundary = next( + item for item in plan["preserved_external"] + if item["path"] == "operations/update-live-preview-0001/candidate-venv" + ) + self.assertEqual("INSTALLATION_RUNTIME", boundary["classification"]) + self.assertEqual("directory", boundary["kind"]) + self.assertTrue(boundary["opaque_boundary"]) + self.assertFalse(boundary["symlinks_followed"]) + self.assertFalse(any( + str(item["path"]).startswith( + "operations/update-live-preview-0001/candidate-venv/" + ) for item in plan["preserved_external"] + )) + + # A sibling staging symlink is not covered by the opaque venv contract. + outside = Path(self.temporary.name) / "outside-runtime" + outside.mkdir() + (operation / "build").mkdir() + unsafe = operation / "build" / "unsafe-link" + unsafe.symlink_to(outside, target_is_directory=True) + self.assertIn("EXTERNAL_SYMLINK_UNSAFE", reset.preview(self.root)["blocking_codes"]) + unsafe.unlink() + + # A candidate-venv is opaque only below an exact journalled operation; + # the same symlink shape at an unowned path remains fail-closed. + unowned = self.root / "operations" / "update-no-journal-0002" / "candidate-venv" + unowned.mkdir(parents=True) + (unowned / "python").symlink_to("/usr/bin/python3") + self.assertIn("EXTERNAL_SYMLINK_UNSAFE", reset.preview(self.root)["blocking_codes"]) + (unowned / "python").unlink() + unowned.rmdir() + unowned.parent.rmdir() + + linked_operation = self.root / "operations" / "update-venv-link-0003" + linked_operation.mkdir() + self._bind_update_candidate_fixture(linked_operation) + linked_venv = linked_operation / "candidate-venv" + linked_venv.symlink_to(outside, target_is_directory=True) + self.assertIn("EXTERNAL_SYMLINK_UNSAFE", reset.preview(self.root)["blocking_codes"]) + linked_venv.unlink() + (linked_operation / "candidate-runtime.json").unlink() + (linked_operation / "operation.json").unlink() + linked_operation.rmdir() + + # A malformed journal or a candidate marker bound to another path + # cannot hide regular unknown content behind a venv-shaped directory. + malformed = self.root / "operations" / "update-malformed-0004" + malformed_venv = malformed / "candidate-venv" + malformed_venv.mkdir(parents=True) + (malformed_venv / "unknown.bin").write_bytes(b"keep") + (malformed / "operation.json").write_text("{}", encoding="utf-8") + (malformed / "candidate-runtime.json").write_text("{}", encoding="utf-8") + malformed_plan = reset.preview(self.root) + self.assertIn("EXTERNAL_CLASSIFICATION_INCOMPLETE", malformed_plan["blocking_codes"]) + self.assertIn( + "operations/update-malformed-0004/candidate-venv/unknown.bin", + malformed_plan["unknown_external_paths"], + ) + (malformed_venv / "unknown.bin").unlink() + malformed_venv.rmdir() + (malformed / "candidate-runtime.json").unlink() + (malformed / "operation.json").unlink() + malformed.rmdir() + + unbound = self.root / "operations" / "update-unbound-0005" + unbound_venv = unbound / "candidate-venv" + unbound_venv.mkdir(parents=True) + (unbound_venv / "unknown.bin").write_bytes(b"keep") + self._bind_update_candidate_fixture(unbound) + marker_path = unbound / "candidate-runtime.json" + marker = json.loads(marker_path.read_text(encoding="utf-8")) + marker["candidate_venv"] = str(unbound / "some-other-venv") + marker_path.write_text(json.dumps(marker), encoding="utf-8") + unbound_plan = reset.preview(self.root) + self.assertIn("EXTERNAL_CLASSIFICATION_INCOMPLETE", unbound_plan["blocking_codes"]) + self.assertIn( + "operations/update-unbound-0005/candidate-venv/unknown.bin", + unbound_plan["unknown_external_paths"], + ) + + unknown = operation / "operator-notes.txt" + unknown.write_text("preserve", encoding="utf-8") + blocked = reset.preview(self.root) + self.assertIn("EXTERNAL_CLASSIFICATION_INCOMPLETE", blocked["blocking_codes"]) + self.assertIn( + "operations/update-live-preview-0001/operator-notes.txt", + blocked["unknown_external_paths"], + ) + self.assertEqual("preserve", unknown.read_text(encoding="utf-8")) + + def test_marker_bound_pre_journal_updater_staging_is_opaque_but_incomplete(self) -> None: + operation = self.root / "operations" / "ep-update-2358-release-20260915-001" + operation.mkdir(parents=True) + candidate = operation / "candidate-venv" + (candidate / "bin").mkdir(parents=True) + (candidate / "bin" / "python").symlink_to("python3.14") + (candidate / "bin" / "python3.14").symlink_to("/usr/bin/python3") + self._bind_update_candidate_fixture(operation) + marker = json.loads((operation / "candidate-runtime.json").read_text(encoding="utf-8")) + staged = Path(str(marker["staged_artifact"])) + staged.parent.mkdir() + staged.write_bytes(b"synthetic wheel") + marker["target_digest"] = "sha256:" + hashlib.sha256(staged.read_bytes()).hexdigest() + (operation / "candidate-runtime.json").write_text( + json.dumps(marker), encoding="utf-8", + ) + pip_cache = operation / "pip-cache" + pip_cache.mkdir() + # Exact updater-owned subtrees are opaque in this explicit crash + # window, so their internal links are neither followed nor reset. + (pip_cache / "cached-link").symlink_to(Path(self.temporary.name) / "outside-cache") + (operation / "operation.json").unlink() + + plan = reset.preview(self.root) + + self.assertNotIn("EXTERNAL_SYMLINK_UNSAFE", plan["blocking_codes"]) + self.assertNotIn("EXTERNAL_CLASSIFICATION_INCOMPLETE", plan["blocking_codes"]) + preserved = {str(item["path"]): item for item in plan["preserved_external"]} + marker_path = f"operations/{operation.name}/candidate-runtime.json" + self.assertEqual( + preserved[marker_path]["classification"], + "INSTALLATION_RUNTIME_STAGING_UNBOUND", + ) + for name in ("candidate-venv", "download", "pip-cache"): + boundary = preserved[f"operations/{operation.name}/{name}"] + self.assertEqual( + boundary["classification"], "INSTALLATION_RUNTIME_STAGING_UNBOUND", + ) + self.assertTrue(boundary["opaque_boundary"]) + self.assertFalse(boundary["symlinks_followed"]) + self.assertEqual( + boundary["installation_update_state"], + "INCOMPLETE_NO_OPERATION_JOURNAL", + ) + self.assertFalse(any( + path.startswith(f"operations/{operation.name}/candidate-venv/") + or path.startswith(f"operations/{operation.name}/download/") + or path.startswith(f"operations/{operation.name}/pip-cache/") + for path in preserved + )) + + sibling = operation / "operator-notes.txt" + sibling.write_text("preserve", encoding="utf-8") + blocked = reset.preview(self.root) + self.assertIn("EXTERNAL_CLASSIFICATION_INCOMPLETE", blocked["blocking_codes"]) + self.assertIn( + f"operations/{operation.name}/operator-notes.txt", + blocked["unknown_external_paths"], + ) + self.assertEqual(sibling.read_text(encoding="utf-8"), "preserve") + + def test_unbound_staging_marker_mismatch_missing_and_linked_root_fail_closed(self) -> None: + outside = Path(self.temporary.name) / "outside-unbound" + outside.mkdir() + for suffix, mutation in ( + ("mismatch", "MISMATCH"), + ("malformed", "MALFORMED"), + ("missing", "MISSING"), + ("identity", "WRONG_INSTALLATION"), + ("digest", "DIGEST_MISMATCH"), + ("wheel", "WHEEL_IDENTITY"), + ("linked", "LINKED_ROOT"), + ): + operation = self.root / "operations" / f"ep-update-unbound-{suffix}-0001" + operation.mkdir(parents=True) + self._bind_update_candidate_fixture(operation) + marker_path = operation / "candidate-runtime.json" + marker = json.loads(marker_path.read_text(encoding="utf-8")) + staged = Path(str(marker["staged_artifact"])) + staged.parent.mkdir() + staged.write_bytes(b"synthetic wheel") + marker["target_digest"] = "sha256:" + hashlib.sha256(staged.read_bytes()).hexdigest() + marker_path.write_text(json.dumps(marker), encoding="utf-8") + (operation / "operation.json").unlink() + candidate = operation / "candidate-venv" + if mutation == "LINKED_ROOT": + candidate.symlink_to(outside, target_is_directory=True) + else: + candidate.mkdir() + (candidate / "python").symlink_to("/usr/bin/python3") + if mutation == "MISMATCH": + marker["candidate_venv"] = str(operation / "different-venv") + marker_path.write_text(json.dumps(marker), encoding="utf-8") + elif mutation == "MALFORMED": + marker_path.write_text("{}", encoding="utf-8") + elif mutation == "WRONG_INSTALLATION": + marker["installation_id"] = "forged-installation-id" + marker_path.write_text(json.dumps(marker), encoding="utf-8") + elif mutation == "DIGEST_MISMATCH": + marker["target_digest"] = "sha256:" + "d" * 64 + marker_path.write_text(json.dumps(marker), encoding="utf-8") + elif mutation == "WHEEL_IDENTITY": + foreign = staged.with_name("foreign-candidate.whl") + staged.rename(foreign) + staged = foreign + marker["staged_artifact"] = str(staged) + marker_path.write_text(json.dumps(marker), encoding="utf-8") + else: + marker_path.unlink() + + blocked = reset.preview(self.root) + self.assertIn("EXTERNAL_SYMLINK_UNSAFE", blocked["blocking_codes"]) + + if candidate.is_symlink(): + candidate.unlink() + else: + (candidate / "python").unlink() + candidate.rmdir() + staged.unlink() + staged.parent.rmdir() + marker_path.unlink(missing_ok=True) + operation.rmdir() + + def test_journalled_candidate_rejects_forged_identity_cleanup_and_steps(self) -> None: + operation = self.root / "operations" / "update-adversarial-0001" + operation.mkdir(parents=True) + candidate = operation / "candidate-venv" + (candidate / "bin").mkdir(parents=True) + (candidate / "bin" / "python").symlink_to("/usr/bin/python3") + self._bind_update_candidate_fixture(operation) + journal_path = operation / "operation.json" + marker_path = operation / "candidate-runtime.json" + original_journal = journal_path.read_text(encoding="utf-8") + original_marker = marker_path.read_text(encoding="utf-8") + + def persist(journal: dict[str, object], marker: dict[str, object]) -> None: + plan = journal["plan"] + prepared = journal["prepared_candidate"] + assert isinstance(plan, dict) and isinstance(prepared, dict) + journal["plan_digest"] = "sha256:" + hashlib.sha256(json.dumps( + plan, sort_keys=True, separators=(",", ":"), + ).encode()).hexdigest() + journal["prepared_candidate_digest"] = "sha256:" + hashlib.sha256(json.dumps( + prepared, sort_keys=True, separators=(",", ":"), + ).encode()).hexdigest() + journal_path.write_text(json.dumps(journal), encoding="utf-8") + marker_path.write_text(json.dumps(marker), encoding="utf-8") + + for case in ("WRONG_INSTALLATION", "EXTERNAL_CLEANUP", "UNKNOWN_STEP"): + journal = json.loads(original_journal) + marker = json.loads(original_marker) + plan = journal["plan"] + prepared = journal["prepared_candidate"] + assert isinstance(plan, dict) and isinstance(prepared, dict) + if case == "WRONG_INSTALLATION": + plan["installation_id"] = "forged-installation-id" + prepared["installation_id"] = "forged-installation-id" + marker["installation_id"] = "forged-installation-id" + elif case == "EXTERNAL_CLEANUP": + plan["cleanup_targets"] = [ + str(Path(self.temporary.name) / "outside-cleanup"), + str(operation / "download"), + str(operation / "pip-cache"), + ] + else: + plan["steps"] = [*plan["steps"], "UNKNOWN_SIDE_EFFECT"] + persist(journal, marker) + with self.subTest(case=case): + blocked = reset.preview(self.root) + self.assertIn("EXTERNAL_SYMLINK_UNSAFE", blocked["blocking_codes"]) + + journal_path.write_text(original_journal, encoding="utf-8") + marker_path.write_text(original_marker, encoding="utf-8") + allowed = reset.preview(self.root) + self.assertNotIn("EXTERNAL_SYMLINK_UNSAFE", allowed["blocking_codes"]) + + def test_empty_database_identity_and_empty_updater_bindings_never_become_opaque(self) -> None: + runtime_identity = json.loads( + (self.root / "runtime-identity.json").read_text(encoding="utf-8") + )["instance_id"] + self.assertRegex( + runtime_identity, + r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", + ) + operation = self.root / "operations" / "update-empty-identity-0001" + operation.mkdir(parents=True) + candidate = operation / "candidate-venv" + candidate.mkdir() + unknown = candidate / "unknown-application-data.bin" + unknown.write_bytes(b"preserve") + self._bind_update_candidate_fixture(operation) + + journal_path = operation / "operation.json" + marker_path = operation / "candidate-runtime.json" + journal = json.loads(journal_path.read_text(encoding="utf-8")) + marker = json.loads(marker_path.read_text(encoding="utf-8")) + plan = journal["plan"] + prepared = journal["prepared_candidate"] + assert isinstance(plan, dict) and isinstance(prepared, dict) + plan["installation_id"] = "" + prepared["installation_id"] = "" + marker["installation_id"] = "" + journal["plan_digest"] = "sha256:" + hashlib.sha256(json.dumps( + plan, sort_keys=True, separators=(",", ":"), + ).encode()).hexdigest() + journal["prepared_candidate_digest"] = "sha256:" + hashlib.sha256(json.dumps( + prepared, sort_keys=True, separators=(",", ":"), + ).encode()).hexdigest() + journal_path.write_text(json.dumps(journal), encoding="utf-8") + marker_path.write_text(json.dumps(marker), encoding="utf-8") + with reset._central_connection(self.root / "epdata.sqlite") as connection: + connection.execute("UPDATE ep_installations SET instance_id=''") + connection.execute( + "UPDATE engineering_metadata SET value='' " + "WHERE key='installation.instance_id'" + ) + + with self.assertRaisesRegex(reset.OperationalResetError, "TARGET_IDENTITY_CONFLICT"): + reset.preview(self.root) + with self.assertRaisesRegex(reset.OperationalResetError, "TARGET_IDENTITY_CONFLICT"): + reset._external_inventory(self.root, expected_installation_id="") + self.assertIsNone(reset._opaque_update_directory_classification( + "operations", candidate, f"{operation.name}/candidate-venv", + expected_installation_id="", + )) + self.assertEqual(unknown.read_bytes(), b"preserve") + + def test_opaque_directory_swap_before_receipt_fails_closed(self) -> None: + operation = self.root / "operations" / "update-swap-race-0001" + operation.mkdir(parents=True) + candidate = operation / "candidate-venv" + (candidate / "bin").mkdir(parents=True) + (candidate / "bin" / "python").symlink_to("/usr/bin/python3") + self._bind_update_candidate_fixture(operation) + displaced = operation / "candidate-venv-displaced" + outside = Path(self.temporary.name) / "outside-swap-target" + outside.mkdir() + original_recheck = reset._recheck_opaque_directory + swapped = False + + def swap_then_recheck( + path: Path, *, device: int, inode: int, descriptor: int, + ) -> None: + nonlocal swapped + if path == candidate and not swapped: + candidate.rename(displaced) + candidate.symlink_to(outside, target_is_directory=True) + swapped = True + original_recheck( + path, device=device, inode=inode, descriptor=descriptor, + ) + + with patch.object( + reset, "_recheck_opaque_directory", side_effect=swap_then_recheck, + ): + blocked = reset.preview(self.root) + self.assertTrue(swapped) + self.assertIn("EXTERNAL_SYMLINK_UNSAFE", blocked["blocking_codes"]) + self.assertFalse(any( + item.get("path") == "operations/update-swap-race-0001/candidate-venv" + for item in blocked["preserved_external"] + )) + def test_cli_argument_and_internal_errors_are_stable_secret_free_contracts(self) -> None: def invoke(arguments: list[str]) -> tuple[int, dict[str, object]]: output = io.StringIO()