diff --git a/autoform_cli/project/create.py b/autoform_cli/project/create.py index d13ed6c0..8e1c30be 100644 --- a/autoform_cli/project/create.py +++ b/autoform_cli/project/create.py @@ -16,6 +16,7 @@ from urllib.parse import urlsplit from .. import scaffold +from ..claims import _reject_json_constant, _strict_json_object from ..graph import _parse_node from ..scaffold import ( DEFAULT_AUTOFORM_SOURCE, @@ -29,11 +30,11 @@ ) from .catalog import SupportedRelease, load_release_catalog +PROJECT_CREATION_SCHEMA = "autoform-project-creation/v1" +_CREATION_RELEASE_SCHEMA = "autoform-project-creation-release/v1" _PACKAGE_NAME = re.compile(r"[A-Z][A-Za-z0-9]*") _FULL_SHA = re.compile(r"[0-9a-f]{40}") _RESERVED_PACKAGE_NAMES = frozenset({"Prop", "Sort", "Type"}) -_CREATION_RELEASE_SCHEMA = "autoform-project-creation-release/v1" -PROJECT_CREATION_SCHEMA = "autoform-project-creation/v1" _RELEASE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*") _STAGE_ATTEMPTS = 32 _TOOLCHAIN_MODULE_ROOTS = frozenset({"Init", "Lake", "Lean", "Std"}) @@ -44,38 +45,32 @@ r"\.(?P0|[1-9][0-9]{0,8})(?:-rc[1-9][0-9]{0,8})?)" ) # Safe inside a TOML basic string, never a Git option, and free of the -# `: ^ ~ @ { ? * [` revision operators. Additional checks below enforce the +# `: ^ ~ @ { ? * [` revision operators. `_valid_mathlib_revision` enforces the # structural restrictions Git applies to ref names. _MATHLIB_REV = re.compile(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,254}") _LONGEST_LAKE_ARTIFACT_SUFFIX = ".olean.private.hash" _MINIMUM_LEAN = (4, 27, 0) _LOCK_WAIT_SECONDS = 30.0 _LOCK_POLL_SECONDS = 0.05 -_UNSAFE_PARENT_MESSAGE = ( - "The target parent is group- or world-writable and is not a sticky directory owned by you " - "or root. Remove group and world write access (chmod g-w,o-w) or choose another parent." -) # Mathlib library roots that no creation descriptor lists: `docs` (every tag from # v4.27.0, and outside the descriptor root grammar), LongestPole (v4.27.0), and # Wanted (v4.34.1 on). _MATHLIB_EXTRA_ROOTS = frozenset({"docs", "LongestPole", "Wanted"}) -_MANIFEST_FIELDS = frozenset( - {"version", "packagesDir", "packages", "name", "lakeDir", "fixedToolchain"} -) +_MANIFEST_FIELDS = frozenset({"version", "packagesDir", "packages", "name", "lakeDir", "fixedToolchain"}) _MANIFEST_PACKAGE_FIELDS = frozenset( - { - "url", - "type", - "subDir", - "scope", - "rev", - "name", - "manifestFile", - "inputRev", - "inherited", - "configFile", - } + {"url", "type", "subDir", "scope", "rev", "name", "manifestFile", "inputRev", "inherited", "configFile"} ) +_UNSAFE_PARENT_MESSAGE = ( + "The target parent is group- or world-writable and is not a sticky directory owned by you " + "or root. Remove group and world write access (chmod g-w,o-w) or choose another parent." +) +_FAILED_MESSAGE = "Project creation failed; no project was created." +_EXISTS_MESSAGE = "The target already exists; project new never overwrites it." +_CONTRACTS_MESSAGE = "The generated project did not satisfy Autoform's project contracts." +_STAGED_MESSAGE = "The staged project did not satisfy Autoform's project contracts." +_DESCRIPTOR_MESSAGE = "The bundled project-creation release metadata is invalid." +_MANIFEST_MESSAGE = "The bundled release manifest is invalid." +_NO_RENAME_MESSAGE = "This platform cannot atomically publish a new project without replacement." class ProjectCreateError(ValueError): @@ -90,9 +85,7 @@ def as_dict(self) -> dict[str, object]: return {"error": {"code": self.code, "message": self.message}, "ok": False} def to_json(self) -> str: - return json.dumps( - self.as_dict(), ensure_ascii=True, sort_keys=True, separators=(",", ":") - ) + return json.dumps(self.as_dict(), ensure_ascii=True, sort_keys=True, separators=(",", ":")) @dataclass(frozen=True, slots=True) @@ -121,9 +114,7 @@ def as_dict(self) -> dict[str, object]: } def to_json(self) -> str: - return json.dumps( - self.as_dict(), ensure_ascii=True, sort_keys=True, separators=(",", ":") - ) + return json.dumps(self.as_dict(), ensure_ascii=True, sort_keys=True, separators=(",", ":")) @dataclass(frozen=True, slots=True) @@ -180,94 +171,59 @@ def create_project( # these retained bytes rather than reopening the template tree. templates = _read_templates(_TEMPLATES) _require_complete_templates(templates) - workflow_source, workflow_ref = _resolve_workflow_pin( - autoform_source, - autoform_ref, - templates=templates, - ) - plan, workflows_pinned = _build_project_plan( - package_name, - version, - release_bundle, - templates=templates, - autoform_source=workflow_source, - autoform_ref=workflow_ref, - ) - _plan_tree(plan) + workflow_source, workflow_ref = _resolve_workflow_pin(autoform_source, autoform_ref, templates=templates) + plan = _build_project_plan(package_name, version, release_bundle, templates, workflow_source, workflow_ref) + tree = _plan_tree(plan) _validate_roadmap_plan(plan) except (OSError, ScaffoldError, UnicodeError): - raise ProjectCreateError( - "project-create-validation-failed", - "The generated project did not satisfy Autoform's project contracts.", - ) from None - parent = requested.parent - parent_descriptor = _open_parent(parent) + raise ProjectCreateError("project-create-validation-failed", _CONTRACTS_MESSAGE) from None + parent_descriptor = _open_parent(requested.parent) parent_identity = _descriptor_identity(parent_descriptor) - publish_parent_descriptor: int | None = None - confirmed_parent_descriptor: int | None = None - stage_name: str | None = None + publish_descriptor: int | None = None + confirmed_descriptor: int | None = None stage_descriptor: int | None = None - publication_started = False - published = False - parent_synced = False - parent_rebound = False - parent_recheck_failed = False + stage_name: str | None = None + publication_started = published = parent_synced = False + parent_recheck: str | None = None try: _require_package_filename_fit(package_name, parent_descriptor) _lock_parent(parent_descriptor) _require_absent(parent_descriptor, requested.name) stage_name = _create_stage(parent_descriptor) - stage_metadata = os.stat(stage_name, dir_fd=parent_descriptor, follow_symlinks=False) - if not stat.S_ISDIR(stage_metadata.st_mode): - raise OSError(errno.ENOTDIR, "staging path is not a directory") - stage_descriptor = _open_stage(parent_descriptor, stage_name) + stage_descriptor = _open_directory(parent_descriptor, stage_name) _require_stage_identity(parent_descriptor, stage_name, stage_descriptor) - _materialize_project(stage_descriptor, plan) - _require_stage_identity(parent_descriptor, stage_name, stage_descriptor) - _validate_staged_project( - stage_descriptor, - plan, - package_name, - version, - release_bundle, - ) + # A directory renamed into the stage name before the open is not the private, empty stage we made. + if stat.S_IMODE(os.fstat(stage_descriptor).st_mode) != 0o700 or _list_directory(stage_descriptor): + raise ProjectCreateError("project-create-failed", _FAILED_MESSAGE) + _materialize_project(stage_descriptor, tree) _require_stage_identity(parent_descriptor, stage_name, stage_descriptor) + # Templates may carry a root manifest; only a catalog release may publish one. + # Refuse it before the chmod, while the populated stage is still private. + if release_bundle is None and "lake-manifest.json" in tree: + raise ProjectCreateError("project-create-validation-failed", _STAGED_MESSAGE) os.fchmod(stage_descriptor, 0o755) os.fsync(stage_descriptor) - _verify_project_plan(stage_descriptor, plan, root_mode=0o755) - _require_stage_identity(parent_descriptor, stage_name, stage_descriptor) - # Reopen the original requested parent path with O_NOFOLLOW at the last - # possible moment. Publication uses this fresh descriptor only after - # its device, inode, and owner match the directory we locked and staged - # in. A mismatch leaves the complete stage untouched. - publish_parent_descriptor = _reopen_bound_parent(parent, parent_identity) - _require_stage_identity(publish_parent_descriptor, stage_name, stage_descriptor) + _verify_project_tree(stage_descriptor, tree) + # Reopen the requested parent path with O_NOFOLLOW at the last possible + # moment, and publish through it only if it still names the directory + # we locked and staged in. A mismatch leaves the complete stage untouched. + publish_descriptor = _reopen_bound_parent(requested.parent, parent_identity) + _require_stage_identity(publish_descriptor, stage_name, stage_descriptor) publication_started = True try: - _rename_noreplace( - parent_descriptor, - stage_name, - publish_parent_descriptor, - requested.name, - ) + _rename_noreplace(parent_descriptor, stage_name, publish_descriptor, requested.name) except FileExistsError: - raise ProjectCreateError( - "project-target-exists", - "The target already exists; project new never overwrites it.", - ) from None + raise ProjectCreateError("project-target-exists", _EXISTS_MESSAGE) from None published = True - _require_stage_identity(publish_parent_descriptor, requested.name, stage_descriptor) - os.fsync(publish_parent_descriptor) + _require_stage_identity(publish_descriptor, requested.name, stage_descriptor) + os.fsync(publish_descriptor) parent_synced = True try: - confirmed_parent_descriptor = _reopen_bound_parent(parent, parent_identity) + confirmed_descriptor = _reopen_bound_parent(requested.parent, parent_identity) except ProjectCreateError as error: - if error.code == "project-parent-changed": - parent_rebound = True - else: - parent_recheck_failed = True + parent_recheck = error.code raise - _require_stage_identity(confirmed_parent_descriptor, requested.name, stage_descriptor) + _require_stage_identity(confirmed_descriptor, requested.name, stage_descriptor) return ProjectCreateResult( package=package_name, release=None if version.release is None else version.release.id, @@ -275,214 +231,87 @@ def create_project( mathlib_rev=version.mathlib_rev, target=requested.name, written=tuple(item.relative for item in plan), - workflows_pinned=workflows_pinned, + workflows_pinned=bool(workflow_ref), warnings=warnings, ) - except ProjectCreateError as error: - state = _publication_state( - parent_descriptor, - requested.name, - stage_name, - stage_descriptor, - ) + except BaseException as error: + state = _publication_state(parent_descriptor, requested.name, stage_name, stage_descriptor) if published or (publication_started and state != "stage"): - raise ProjectCreateError( - "project-create-commit-uncertain", - _commit_uncertain_message( - state, - parent_synced=parent_synced, - parent_rebound=parent_rebound, - parent_recheck_failed=parent_recheck_failed, - ), - ) from None - if stage_name is not None: - raise ProjectCreateError(error.code, _with_preserved_stage(error.message)) from None - raise - except OSError: - state = _publication_state( - parent_descriptor, - requested.name, - stage_name, - stage_descriptor, - ) - if published or (publication_started and state != "stage"): - raise ProjectCreateError( - "project-create-commit-uncertain", - _commit_uncertain_message( - state, - parent_synced=parent_synced, - parent_rebound=parent_rebound, - parent_recheck_failed=parent_recheck_failed, - ), - ) from None - message = "Project creation failed; no project was created." + raise _commit_uncertain(state, parent_synced, parent_recheck) from None + if isinstance(error, OSError): + code, message = "project-create-failed", _FAILED_MESSAGE + elif isinstance(error, ProjectCreateError): + code, message = error.code, error.message + else: + raise if stage_name is not None: - message = _with_preserved_stage(message) - raise ProjectCreateError("project-create-failed", message) from None - except BaseException: - state = _publication_state( - parent_descriptor, - requested.name, - stage_name, - stage_descriptor, - ) - if published or (publication_started and state != "stage"): - raise ProjectCreateError( - "project-create-commit-uncertain", - _commit_uncertain_message( - state, - parent_synced=parent_synced, - parent_rebound=parent_rebound, - parent_recheck_failed=parent_recheck_failed, - ), - ) from None - raise + message += " An .autoform-new-* stage may remain; inspect it before removal." + raise ProjectCreateError(code, message) from None finally: - state = _publication_state( - parent_descriptor, - requested.name, - stage_name, - stage_descriptor, - ) - publication_uncertain = not published and publication_started and state != "stage" + state = _publication_state(parent_descriptor, requested.name, stage_name, stage_descriptor) close_failed = False - for descriptor in ( - confirmed_parent_descriptor, - publish_parent_descriptor, - stage_descriptor, - parent_descriptor, - ): - if descriptor is None: - continue - try: - os.close(descriptor) - except OSError: - close_failed = True - if publication_uncertain: - raise ProjectCreateError( - "project-create-commit-uncertain", - _commit_uncertain_message( - state, - parent_synced=parent_synced, - parent_rebound=parent_rebound, - parent_recheck_failed=parent_recheck_failed, - ), - ) + for descriptor in (confirmed_descriptor, publish_descriptor, stage_descriptor, parent_descriptor): + if descriptor is not None: + try: + os.close(descriptor) + except OSError: + close_failed = True + if not published and publication_started and state != "stage": + raise _commit_uncertain(state, parent_synced, parent_recheck) if close_failed and published: - raise ProjectCreateError( - "project-create-commit-uncertain", - _commit_uncertain_message( - state, - parent_synced=parent_synced, - parent_rebound=parent_rebound, - parent_recheck_failed=parent_recheck_failed, - cleanup_failed=True, - ), - ) - - -def _with_preserved_stage(message: str) -> str: - return f"{message} An .autoform-new-* stage may remain; inspect it before removal." + raise _commit_uncertain(state, parent_synced, parent_recheck, cleanup_failed=True) -def _commit_uncertain_message( - state: str, - *, - parent_synced: bool, - parent_rebound: bool, - parent_recheck_failed: bool, - cleanup_failed: bool = False, -) -> str: - if parent_rebound: - return ( - "The project was published and its original parent directory was synced, but the " - "requested parent path no longer names that directory. The requested target may not " - "name the project; locate it before retrying." +def _commit_uncertain( + state: str, parent_synced: bool, parent_recheck: str | None, cleanup_failed: bool = False +) -> ProjectCreateError: + if parent_recheck == "project-parent-changed": + message = ( + "The project was published and its original parent directory was synced, but the requested parent " + "path no longer names that directory. The requested target may not name the project; locate it " + "before retrying." ) - if parent_recheck_failed: - return ( - "The project was published and its original parent directory was synced, but Autoform " - "could not reopen the requested parent path to confirm that it still names that " - "directory. The target was observed through the original parent descriptor; verify " - "the requested path before retrying." + elif parent_recheck is not None: + message = ( + "The project was published and its original parent directory was synced, but Autoform could not " + "reopen the requested parent path to confirm that it still names that directory. The target was " + "observed through the original parent descriptor; verify the requested path before retrying." ) - if state == "target": - durability = ( - "its parent directory was synced" - if parent_synced - else "the parent-directory sync was not confirmed" - ) - suffix = " and final descriptor cleanup failed" if cleanup_failed else "" - return ( - f"The target names the published project and {durability}{suffix}. " - "Do not retry project creation." + elif state == "target": + synced = "its parent directory was synced" if parent_synced else "the parent-directory sync was not confirmed" + cleanup = " and final descriptor cleanup failed" if cleanup_failed else "" + message = f"The target names the published project and {synced}{cleanup}. Do not retry project creation." + else: + message = ( + "Publication started, but neither the target nor the preserved stage names the project directory " + "held open by Autoform. It may have been moved; locate it before retrying." ) - return ( - "Publication started, but neither the target nor the preserved stage names the project " - "directory held open by Autoform. It may have been moved; locate it before retrying." - ) + return ProjectCreateError("project-create-commit-uncertain", message) def _validate_package(package: str | None, parent: Path) -> str: - if ( - not isinstance(package, str) - or _PACKAGE_NAME.fullmatch(package) is None - or package in _RESERVED_PACKAGE_NAMES - ): - raise ProjectCreateError( - "project-name-invalid", - "Project name must be an UpperCamelCase Lean identifier.", - ) - try: - name_limit = os.pathconf(parent, "PC_NAME_MAX") - except (AttributeError, OSError, ValueError): - raise ProjectCreateError( - "project-create-safety-unavailable", - "This platform cannot validate the generated Lean filename safely.", - ) from None - if not _package_artifact_fits(package, name_limit): - raise ProjectCreateError( - "project-name-invalid", - "Project name is too long for Lean and Lake artifacts on the target filesystem.", - ) + if not isinstance(package, str) or _PACKAGE_NAME.fullmatch(package) is None or package in _RESERVED_PACKAGE_NAMES: + raise ProjectCreateError("project-name-invalid", "Project name must be an UpperCamelCase Lean identifier.") + _require_package_filename_fit(package, parent) return package -def _package_artifact_fits(package: str, name_limit: int) -> bool: - return len(f"{package}{_LONGEST_LAKE_ARTIFACT_SUFFIX}".encode("ascii")) <= name_limit +def _require_package_filename_fit(package: str, directory: Path | int) -> None: + """Leave room in *directory* for the longest artifact name Lake derives from *package*.""" - -def _require_package_filename_fit(package: str, parent_descriptor: int) -> None: try: - name_limit = os.fpathconf(parent_descriptor, "PC_NAME_MAX") - except (AttributeError, OSError, ValueError): + name_limit = os.pathconf(directory, "PC_NAME_MAX") + except (AttributeError, OSError, TypeError, ValueError): raise ProjectCreateError( - "project-create-safety-unavailable", - "This platform cannot validate the generated Lean filename safely.", + "project-create-safety-unavailable", "This platform cannot validate the generated Lean filename safely." ) from None - if not _package_artifact_fits(package, name_limit): - raise ProjectCreateError( - "project-name-invalid", - "Project name is too long for Lean and Lake artifacts on the target filesystem.", - ) - - -def _find_release(release_id: str | None) -> SupportedRelease: - catalog = load_release_catalog() - release = next((item for item in catalog.releases if item.id == release_id), None) - if release is None: + if len(package) + len(_LONGEST_LAKE_ARTIFACT_SUFFIX) > name_limit: raise ProjectCreateError( - "project-release-unknown", - "The requested release is not in the bundled release catalog.", + "project-name-invalid", "Project name is too long for Lean and Lake artifacts on the target filesystem." ) - return release -def _resolve_version( - release_id: str | None, - lean_toolchain: str | None, - mathlib_rev: str | None, -) -> _ProjectVersion: +def _resolve_version(release_id: str | None, lean_toolchain: str | None, mathlib_rev: str | None) -> _ProjectVersion: """Choose the catalog release or the unlisted version pair the options name. A toolchain and revision equal to a catalog entry resolve to that entry, so @@ -491,34 +320,35 @@ def _resolve_version( if release_id is not None and (lean_toolchain is not None or mathlib_rev is not None): raise ProjectCreateError( - "project-version-invalid", - "Choose a catalog release or a Lean toolchain and Mathlib revision, not both.", + "project-version-invalid", "Choose a catalog release or a Lean toolchain and Mathlib revision, not both." ) if lean_toolchain is None: if mathlib_rev is not None: - raise ProjectCreateError( - "project-version-invalid", - "A Mathlib revision requires a Lean toolchain.", - ) + raise ProjectCreateError("project-version-invalid", "A Mathlib revision requires a Lean toolchain.") + catalog = load_release_catalog() release = ( - load_release_catalog().recommended if release_id is None else _find_release(release_id) + catalog.recommended + if release_id is None + else next((item for item in catalog.releases if item.id == release_id), None) ) + if release is None: + raise ProjectCreateError( + "project-release-unknown", "The requested release is not in the bundled release catalog." + ) return _ProjectVersion(release.lean_toolchain, release.mathlib_git, release.mathlib_rev, release) match = _LEAN_TOOLCHAIN.fullmatch(lean_toolchain) if isinstance(lean_toolchain, str) else None if match is None: raise ProjectCreateError( "project-version-invalid", - "The Lean toolchain must be a Lean release tag such as v4.30.0, v4.30.0-rc1, " - "or leanprover/lean4:v4.30.0.", + "The Lean toolchain must be a Lean release tag such as v4.30.0, v4.30.0-rc1, or leanprover/lean4:v4.30.0.", ) toolchain = f"leanprover/lean4:{match['tag']}" revision = match["tag"] if mathlib_rev is None else mathlib_rev if not isinstance(revision, str) or not _valid_mathlib_revision(revision): raise ProjectCreateError( "project-version-invalid", - "The Mathlib revision must be a valid Git tag, branch, or commit spelling: 1 to 255 " - "ASCII letters, digits, dots, underscores, hyphens, or slashes, starting with a " - "letter or digit.", + "The Mathlib revision must be a valid Git tag, branch, or commit spelling: 1 to 255 ASCII letters, " + "digits, dots, underscores, hyphens, or slashes, starting with a letter or digit.", ) catalog = load_release_catalog() for release in catalog.releases: @@ -533,19 +363,12 @@ def _resolve_version( def _valid_mathlib_revision(revision: str) -> bool: """Whether *revision* fits the safe grammar and Git's ref-name rules.""" - if ( - _MATHLIB_REV.fullmatch(revision) is None - or ".." in revision - or "@{" in revision - or revision.endswith(".") - ): - return False - parts = revision.split("/") - return all( - part - and not part.startswith(".") - and not part.endswith(".lock") - for part in parts + return ( + _MATHLIB_REV.fullmatch(revision) is not None + and ".." not in revision + and "@{" not in revision + and not revision.endswith(".") + and all(part and not part.startswith(".") and not part.endswith(".lock") for part in revision.split("/")) ) @@ -568,9 +391,8 @@ def _version_warnings(version: _ProjectVersion) -> tuple[tuple[str, str], ...]: "project-release-unlisted", f"{version.lean_toolchain} with Mathlib {version.mathlib_rev} does not name a bundled " "known-good release by its tag or full commit, so no lake-manifest.json was written. " - "Run `lake update` in " - "the project to resolve and lock Mathlib; it needs network access and also " - "downloads the Mathlib build cache. The project's lean-toolchain must match the " + "Run `lake update` in the project to resolve and lock Mathlib; it needs network access and " + "also downloads the Mathlib build cache. The project's lean-toolchain must match the " "lean-toolchain of that Mathlib revision.", ) ) @@ -578,20 +400,14 @@ def _version_warnings(version: _ProjectVersion) -> tuple[tuple[str, str], ...]: def _validate_package_for_release(package: str, bundle: _ReleaseBundle) -> None: - reserved = bundle.module_roots | _MATHLIB_EXTRA_ROOTS - if package.casefold() in {root.casefold() for root in reserved}: + if package.casefold() in {root.casefold() for root in bundle.module_roots | _MATHLIB_EXTRA_ROOTS}: raise ProjectCreateError( "project-name-invalid", "Project name must not shadow a module root used by Lean, Mathlib, or Mathlib's dependencies.", ) -def _resolve_workflow_pin( - source: str, - ref: str, - *, - templates: tuple[tuple[str, bytes, int], ...], -) -> tuple[str, str]: +def _resolve_workflow_pin(source: str, ref: str, *, templates: tuple[tuple[str, bytes, int], ...]) -> tuple[str, str]: """Choose the Autoform source and commit the generated workflows install. The rules are `autoform init`'s and run before any filesystem state @@ -604,8 +420,7 @@ def _resolve_workflow_pin( if not isinstance(source, str) or not isinstance(ref, str): raise ProjectCreateError( - "project-workflow-pin-invalid", - "The Autoform workflow source and ref must be strings.", + "project-workflow-pin-invalid", "The Autoform workflow source and ref must be strings." ) given_ref = ref.strip().lower() if given_ref and _FULL_SHA.fullmatch(given_ref) is None: @@ -635,14 +450,14 @@ def _validate_target(target: str | Path | None) -> Path: if target is None: raise ValueError encoded = os.fspath(target) - if not isinstance(encoded, str) or "\0" in encoded: - raise ValueError # Reject strings that the host filesystem codec cannot round-trip and # all surrogate spellings. Some POSIX kernels accept surrogate-escaped # bytes for lookup but reject them in atomic rename syscalls; rejecting # them here avoids leaving a complete stage after that late failure. if ( - any(0xD800 <= ord(character) <= 0xDFFF for character in encoded) + not isinstance(encoded, str) + or "\0" in encoded + or any(0xD800 <= ord(character) <= 0xDFFF for character in encoded) or os.fsdecode(os.fsencode(encoded)) != encoded ): raise ValueError @@ -652,11 +467,8 @@ def _validate_target(target: str | Path | None) -> Path: raw = selected.absolute() except (OSError, RuntimeError, TypeError, ValueError): raise ProjectCreateError("project-target-invalid", "The project target cannot be resolved safely.") from None - if raw.name in {"", ".", ".."}: - raise ProjectCreateError("project-target-invalid", "The project target must name a new directory.") - parent = raw.parent try: - metadata = parent.stat() + metadata = raw.parent.stat() except OSError as error: raise _parent_access_error(error) from None except ValueError: @@ -686,46 +498,38 @@ def _parent_access_error(error: OSError) -> ProjectCreateError: def _unsafe_parent_metadata(mode: int, owner: int) -> bool: - shared_writable = bool(mode & (stat.S_IWGRP | stat.S_IWOTH)) - if not shared_writable: - return False - if not mode & stat.S_ISVTX: - return True # Sticky directories protect entries only from peers, not from their owner. # Trust the invoking user and the system administrator, as conventional - # root-owned temporary directories require. - return not hasattr(os, "geteuid") or owner not in {0, os.geteuid()} + # root-owned temporary directories require. ACLs are not read: on macOS, + # an inheritable parent ACL can grant another uid access to the stage. + return bool(mode & (stat.S_IWGRP | stat.S_IWOTH)) and ( + not mode & stat.S_ISVTX or not hasattr(os, "geteuid") or owner not in {0, os.geteuid()} + ) def _open_parent(parent: Path) -> int: if ( - not hasattr(os, "O_NOFOLLOW") - or not hasattr(os, "O_DIRECTORY") - or not hasattr(os, "O_NONBLOCK") + not all(hasattr(os, name) for name in ("O_NOFOLLOW", "O_DIRECTORY", "O_NONBLOCK", "geteuid")) or any(function not in os.supports_dir_fd for function in (os.mkdir, os.open, os.stat)) or os.stat not in os.supports_follow_symlinks or os.listdir not in os.supports_fd - or not hasattr(os, "geteuid") ): raise ProjectCreateError( "project-create-safety-unavailable", "This platform cannot create the project with the required path safety.", ) - flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) - absolute = parent.absolute() try: - descriptor = os.open(absolute.anchor, flags) + descriptor = _open_directory(None, parent.anchor) try: - for part in absolute.parts[1:]: + for part in parent.parts[1:]: try: - child = os.open(part, flags, dir_fd=descriptor) + child = _open_directory(descriptor, part) except NotADirectoryError: # macOS reports a symbolic link as ENOTDIR too, so look # before calling the component a link. if not stat.S_ISLNK(os.stat(part, dir_fd=descriptor, follow_symlinks=False).st_mode): raise ProjectCreateError( - "project-parent-invalid", - "The target parent or one of its ancestors is not a directory.", + "project-parent-invalid", "The target parent or one of its ancestors is not a directory." ) from None raise os.close(descriptor) @@ -740,12 +544,10 @@ def _open_parent(parent: Path) -> int: # Depending on the platform, O_DIRECTORY|O_NOFOLLOW reports a symbolic # link as ELOOP, EMLINK, or ENOTDIR (macOS). if error.errno in {errno.ELOOP, errno.EMLINK, errno.ENOTDIR}: - raise ProjectCreateError( - "project-path-is-symlink", "The target path contains a symbolic link." - ) from None + raise ProjectCreateError("project-path-is-symlink", "The target path contains a symbolic link.") from None if error.errno in {errno.ENOENT, errno.EACCES, errno.EPERM}: raise _parent_access_error(error) from None - raise ProjectCreateError("project-create-failed", "Project creation failed; no project was created.") from None + raise ProjectCreateError("project-create-failed", _FAILED_MESSAGE) from None except UnicodeError: raise ProjectCreateError("project-target-invalid", "The project target cannot be resolved safely.") from None return descriptor @@ -758,14 +560,12 @@ def _reopen_bound_parent(parent: Path, expected_identity: tuple[int, int, int]) descriptor = _open_parent(parent) except ProjectCreateError: raise ProjectCreateError( - "project-parent-unverifiable", - "The requested parent path could not be reverified safely.", + "project-parent-unverifiable", "The requested parent path could not be reverified safely." ) from None try: if _descriptor_identity(descriptor) != expected_identity: raise ProjectCreateError( - "project-parent-changed", - "The requested parent path changed while the project was being created.", + "project-parent-changed", "The requested parent path changed while the project was being created." ) except BaseException: os.close(descriptor) @@ -782,8 +582,8 @@ def _require_absent(parent_descriptor: int, name: str) -> None: # Opening the parent needs only read permission; a lookup inside it also needs search. raise _parent_access_error(error) from None except OSError: - raise ProjectCreateError("project-create-failed", "Project creation failed; no project was created.") from None - raise ProjectCreateError("project-target-exists", "The target already exists; project new never overwrites it.") + raise ProjectCreateError("project-create-failed", _FAILED_MESSAGE) from None + raise ProjectCreateError("project-target-exists", _EXISTS_MESSAGE) def _create_stage(parent_descriptor: int) -> str: @@ -799,12 +599,12 @@ def _create_stage(parent_descriptor: int) -> str: "project-parent-inaccessible", "The target parent is not writable; project new needs write permission on it.", ) from None - raise ProjectCreateError("project-create-failed", "Project creation failed; no project was created.") + raise ProjectCreateError("project-create-failed", _FAILED_MESSAGE) -def _open_stage(parent_descriptor: int, stage_name: str) -> int: +def _open_directory(parent_descriptor: int | None, name: str) -> int: flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) - return os.open(stage_name, flags, dir_fd=parent_descriptor) + return os.open(name, flags, dir_fd=parent_descriptor) def _open_planned_file(parent_descriptor: int, name: str) -> int: @@ -816,8 +616,7 @@ def _lock_parent(parent_descriptor: int) -> None: """Serialize creation in the parent, giving up if another holder keeps the lock.""" unavailable = ProjectCreateError( - "project-create-safety-unavailable", - "This platform cannot serialize concurrent project creation safely.", + "project-create-safety-unavailable", "This platform cannot serialize concurrent project creation safely." ) try: import fcntl @@ -840,61 +639,44 @@ def _lock_parent(parent_descriptor: int) -> None: time.sleep(_LOCK_POLL_SECONDS) -def _list_directory(directory_descriptor: int) -> list[str]: +def _list_directory(directory_descriptor: int) -> set[str]: """List through a fresh descriptor so earlier scans cannot leave it at EOF.""" - fresh = _open_stage(directory_descriptor, ".") + fresh = _open_directory(directory_descriptor, ".") try: - return os.listdir(fresh) + return set(os.listdir(fresh)) finally: os.close(fresh) def _descriptor_identity(descriptor: int) -> tuple[int, int, int]: metadata = os.fstat(descriptor) - if not stat.S_ISDIR(metadata.st_mode): - raise OSError(errno.ENOTDIR, "staging path is not a directory") return metadata.st_dev, metadata.st_ino, metadata.st_uid -def _require_stage_identity(workspace_descriptor: int, stage_name: str, stage_descriptor: int) -> None: - expected = _descriptor_identity(stage_descriptor) - metadata = os.stat(stage_name, dir_fd=workspace_descriptor, follow_symlinks=False) - if ( - not stat.S_ISDIR(metadata.st_mode) - or metadata.st_uid != os.geteuid() - or (metadata.st_dev, metadata.st_ino, metadata.st_uid) != expected - ): - raise ProjectCreateError("project-create-failed", "Project creation failed; no project was created.") - - def _entry_matches_descriptor(parent_descriptor: int, name: str, descriptor: int) -> bool: try: expected = _descriptor_identity(descriptor) metadata = os.stat(name, dir_fd=parent_descriptor, follow_symlinks=False) except (OSError, UnicodeError): return False - return ( - stat.S_ISDIR(metadata.st_mode) - and metadata.st_uid == os.geteuid() - and (metadata.st_dev, metadata.st_ino, metadata.st_uid) == expected - ) + return metadata.st_uid == os.geteuid() and (metadata.st_dev, metadata.st_ino, metadata.st_uid) == expected + + +def _require_stage_identity(parent_descriptor: int, name: str, stage_descriptor: int) -> None: + if not _entry_matches_descriptor(parent_descriptor, name, stage_descriptor): + raise ProjectCreateError("project-create-failed", _FAILED_MESSAGE) def _publication_state( - parent_descriptor: int, - target_name: str, - stage_name: str | None, - stage_descriptor: int | None, + parent_descriptor: int, target_name: str, stage_name: str | None, stage_descriptor: int | None ) -> str: if stage_name is None or stage_descriptor is None: return "none" target_matches = _entry_matches_descriptor(parent_descriptor, target_name, stage_descriptor) stage_matches = _entry_matches_descriptor(parent_descriptor, stage_name, stage_descriptor) - if target_matches and not stage_matches: - return "target" - if stage_matches and not target_matches: - return "stage" + if target_matches != stage_matches: + return "target" if target_matches else "stage" return "detached" @@ -902,106 +684,68 @@ def _build_project_plan( package: str, version: _ProjectVersion, release_bundle: _ReleaseBundle | None, - *, templates: tuple[tuple[str, bytes, int], ...], autoform_source: str, autoform_ref: str, -) -> tuple[tuple[_ScaffoldFile, ...], bool]: - files = list(_core_project_plan(package, version, release_bundle)) - scaffold_files, _ = _scaffold_plan( - templates, - title=package, - repository_url="", - autoform_source=autoform_source or DEFAULT_AUTOFORM_SOURCE, - autoform_ref=autoform_ref, - ) - # The shared plan has already reduced installer modes to Git's executable - # bit, so `init` and `project new` publish the same canonical modes. - files.extend(scaffold_files) - return tuple(sorted(files, key=lambda item: item.relative)), bool(autoform_ref) - - -def _core_project_plan( - package: str, - version: _ProjectVersion, - release_bundle: _ReleaseBundle | None, ) -> tuple[_ScaffoldFile, ...]: - # Only a catalog release has a resolved manifest; `lake update` writes one - # for any other pair. - manifest = ( - () - if release_bundle is None - else (_ScaffoldFile("lake-manifest.json", _lake_manifest(package, release_bundle), 0o644),) + lakefile = ( + f'name = "{package}"\n' + 'version = "0.1.0"\n' + f'defaultTargets = ["{package}"]\n\n' + "[[require]]\n" + 'name = "mathlib"\n' + f'git = "{version.mathlib_git}"\n' + f'rev = "{version.mathlib_rev}"\n\n' + "[[lean_lib]]\n" + f'name = "{package}"\n' + 'srcDir = "src"\n' ) - return ( + module = ( + "import Mathlib\n\n" + f"namespace {package}\n\n" + "/-- Marker declaration for the initial project build. -/\n" + "def autoformProjectInitialized : Bool := true\n\n" + f"end {package}\n" + ) + files = [ _ScaffoldFile("lean-toolchain", f"{version.lean_toolchain}\n".encode(), 0o644), - _ScaffoldFile( - "lakefile.toml", - ( - f'name = "{package}"\n' - 'version = "0.1.0"\n' - f'defaultTargets = ["{package}"]\n\n' - "[[require]]\n" - 'name = "mathlib"\n' - f'git = "{version.mathlib_git}"\n' - f'rev = "{version.mathlib_rev}"\n\n' - "[[lean_lib]]\n" - f'name = "{package}"\n' - 'srcDir = "src"\n' - ).encode(), - 0o644, - ), - *manifest, - _ScaffoldFile( - f"src/{package}.lean", - ( - "import Mathlib\n\n" - f"namespace {package}\n\n" - "/-- Marker declaration for the initial project build. -/\n" - "def autoformProjectInitialized : Bool := true\n\n" - f"end {package}\n" - ).encode(), - 0o644, - ), + _ScaffoldFile("lakefile.toml", lakefile.encode(), 0o644), + _ScaffoldFile(f"src/{package}.lean", module.encode(), 0o644), + ] + # Only a catalog release has a resolved manifest; `lake update` writes one + # for any other pair. + if release_bundle is not None: + files.append(_ScaffoldFile("lake-manifest.json", _lake_manifest(package, release_bundle), 0o644)) + # The shared plan has already reduced installer modes to Git's executable + # bit, so `init` and `project new` publish the same canonical modes. + scaffold_files, _ = _scaffold_plan( + templates, title=package, repository_url="", autoform_source=autoform_source, autoform_ref=autoform_ref ) + files.extend(scaffold_files) + return tuple(sorted(files, key=lambda item: item.relative)) def _load_release_bundle(release: SupportedRelease) -> _ReleaseBundle: try: descriptor = _load_creation_release_descriptor(release) - manifest_bytes = ( - files("autoform_cli.project").joinpath(descriptor.manifest_resource).read_bytes() - ) + manifest_bytes = files("autoform_cli.project").joinpath(descriptor.manifest_resource).read_bytes() return _parse_release_bundle(manifest_bytes, release, descriptor.module_roots) except ProjectCreateError: raise except (OSError, TypeError, UnicodeError, ValueError, RecursionError, MemoryError): - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled release manifest is invalid.", - ) from None + raise ProjectCreateError("project-create-validation-failed", _MANIFEST_MESSAGE) from None -def _load_creation_release_descriptor( - release: SupportedRelease, -) -> _CreationReleaseDescriptor: +def _load_creation_release_descriptor(release: SupportedRelease) -> _CreationReleaseDescriptor: + invalid = ProjectCreateError("project-create-validation-failed", _DESCRIPTOR_MESSAGE) if _RELEASE_ID.fullmatch(release.id) is None: - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled project-creation release metadata is invalid.", - ) - resource = f"creation-release-{release.id}.json" + raise invalid try: - payload = json.loads( - files("autoform_cli.project").joinpath(resource).read_bytes(), - object_pairs_hook=_reject_duplicate_object, - parse_constant=_reject_json_constant, + payload = _load_strict_json( + files("autoform_cli.project").joinpath(f"creation-release-{release.id}.json").read_bytes() ) except (OSError, TypeError, UnicodeError, ValueError, RecursionError, MemoryError): - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled project-creation release metadata is invalid.", - ) from None + raise invalid from None expected_release = { "id": release.id, "lean_toolchain": release.lean_toolchain, @@ -1009,102 +753,64 @@ def _load_creation_release_descriptor( "mathlib_rev": release.mathlib_rev, "mathlib_commit": release.mathlib_commit, } - roots = payload.get("production_module_roots") if type(payload) is dict else None - manifest_resource = payload.get("lake_manifest") if type(payload) is dict else None + if type(payload) is not dict or set(payload) != {"schema", "release", "lake_manifest", "production_module_roots"}: + raise invalid + resource, roots = payload["lake_manifest"], payload["production_module_roots"] if ( - type(payload) is not dict - or set(payload) - != {"schema", "release", "lake_manifest", "production_module_roots"} - or payload.get("schema") != _CREATION_RELEASE_SCHEMA - or payload.get("release") != expected_release - or type(manifest_resource) is not str - or not _safe_resource_name(manifest_resource) + payload["schema"] != _CREATION_RELEASE_SCHEMA + or payload["release"] != expected_release + or type(resource) is not str + or not resource.endswith(".json") + or not _safe_relative(resource) + or any(ord(character) >= 0xD800 for character in resource) + or "/" in resource or type(roots) is not list or not roots - or any( - type(root) is not str or _PACKAGE_NAME.fullmatch(root) is None - for root in roots - ) + or any(type(root) is not str or _PACKAGE_NAME.fullmatch(root) is None for root in roots) or len({root.casefold() for root in roots}) != len(roots) ): - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled project-creation release metadata is invalid.", - ) - return _CreationReleaseDescriptor(manifest_resource, tuple(roots)) - - -def _safe_resource_name(value: str) -> bool: - return ( - value.endswith(".json") - and value not in {".", ".."} - and "/" not in value - and "\\" not in value - and "\0" not in value - and all(0x20 <= ord(character) < 0xD800 for character in value) - ) + raise invalid + return _CreationReleaseDescriptor(resource, tuple(roots)) def _parse_release_bundle( - manifest_bytes: bytes, - release: SupportedRelease, - module_roots: tuple[str, ...], + manifest_bytes: bytes, release: SupportedRelease, module_roots: tuple[str, ...] ) -> _ReleaseBundle: - try: - payload = json.loads( - manifest_bytes, - object_pairs_hook=_reject_duplicate_object, - parse_constant=_reject_json_constant, - ) - except (OSError, TypeError, UnicodeError, ValueError, RecursionError, MemoryError): - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled release manifest is invalid.", - ) from None - roots = frozenset(module_roots) - packages = payload.get("packages") if type(payload) is dict else None + invalid = ProjectCreateError("project-create-validation-failed", _MANIFEST_MESSAGE) + payload = _load_strict_json(manifest_bytes) if ( type(payload) is not dict or set(payload) != _MANIFEST_FIELDS - or payload.get("version") != "1.2.0" - or payload.get("packagesDir") != ".lake/packages" - or payload.get("name") != "" - or payload.get("lakeDir") != ".lake" - or payload.get("fixedToolchain") is not False - or type(packages) is not list - or not packages + or payload["version"] != "1.2.0" + or payload["packagesDir"] != ".lake/packages" + or payload["name"] != "" + or payload["lakeDir"] != ".lake" + or payload["fixedToolchain"] is not False + or type(payload["packages"]) is not list + or not payload["packages"] + or not all(_valid_manifest_package(entry) for entry in payload["packages"]) ): - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled release manifest is invalid.", - ) - names: list[str] = [] - for entry in packages: - if not _valid_manifest_package(entry): - raise ProjectCreateError( - "project-create-validation-failed", - "The bundled release manifest is invalid.", - ) - names.append(entry["name"]) - direct = [entry for entry in packages if entry.get("inherited") is False] + raise invalid + packages = payload["packages"] + roots = frozenset(module_roots) folded_roots = {root.casefold() for root in roots} + names = [entry["name"].casefold() for entry in packages] + direct = [entry for entry in packages if entry["inherited"] is False] if ( - len({name.casefold() for name in names}) != len(names) + len(set(names)) != len(names) + or any(name not in folded_roots for name in names) + or not _TOOLCHAIN_MODULE_ROOTS | _MATHLIB_PRODUCTION_ROOTS <= roots or len(direct) != 1 - or direct[0].get("name") != "mathlib" - or direct[0].get("url") != release.mathlib_git - or direct[0].get("inputRev") != release.mathlib_rev - or direct[0].get("rev") != release.mathlib_commit - or direct[0].get("subDir") is not None # releases load Mathlib from its repository root - or direct[0].get("configFile") != "lakefile.lean" - or direct[0].get("manifestFile") != "lake-manifest.json" - or not _TOOLCHAIN_MODULE_ROOTS <= roots - or not _MATHLIB_PRODUCTION_ROOTS <= roots - or any(name.casefold() not in folded_roots for name in names) + or direct[0]["name"] != "mathlib" + or direct[0]["url"] != release.mathlib_git + or direct[0]["inputRev"] != release.mathlib_rev + or direct[0]["rev"] != release.mathlib_commit + or direct[0]["subDir"] is not None # releases load Mathlib from its repository root + or direct[0]["configFile"] != "lakefile.lean" + or direct[0]["manifestFile"] != "lake-manifest.json" ): raise ProjectCreateError( - "project-create-validation-failed", - "The bundled release manifest does not match the release catalog.", + "project-create-validation-failed", "The bundled release manifest does not match the release catalog." ) return _ReleaseBundle(manifest_bytes=manifest_bytes, module_roots=roots) @@ -1112,30 +818,23 @@ def _parse_release_bundle( def _valid_manifest_package(entry: object) -> bool: if type(entry) is not dict or set(entry) != _MANIFEST_PACKAGE_FIELDS: return False - url = entry["url"] - name = entry["name"] - scope = entry["scope"] - revision = entry["rev"] input_revision = entry["inputRev"] - config_file = entry["configFile"] - manifest_file = entry["manifestFile"] - subdirectory = entry["subDir"] return ( - type(url) is str - and _safe_https_git_url(url) + type(entry["url"]) is str + and _safe_https_git_url(entry["url"]) and entry["type"] == "git" - and type(name) is str - and re.fullmatch(r"[A-Za-z][A-Za-z0-9]*", name) is not None - and type(scope) is str - and type(revision) is str - and _FULL_SHA.fullmatch(revision) is not None + and type(entry["name"]) is str + and re.fullmatch(r"[A-Za-z][A-Za-z0-9]*", entry["name"]) is not None + and type(entry["scope"]) is str + and type(entry["rev"]) is str + and _FULL_SHA.fullmatch(entry["rev"]) is not None and type(input_revision) is str and bool(input_revision) and all(ord(character) >= 0x20 for character in input_revision) and type(entry["inherited"]) is bool - and _safe_manifest_relative(config_file) - and _safe_manifest_relative(manifest_file) - and (subdirectory is None or _safe_manifest_relative(subdirectory)) + and _safe_relative(entry["configFile"]) + and _safe_relative(entry["manifestFile"]) + and (entry["subDir"] is None or _safe_relative(entry["subDir"])) ) @@ -1155,37 +854,28 @@ def _safe_https_git_url(value: str) -> bool: ) -def _safe_manifest_relative(value: object) -> bool: +def _safe_relative(value: object) -> bool: + """Whether *value* is a normalized relative POSIX path with printable components.""" + if ( type(value) is not str - or value in {"", ".", ".."} or "\\" in value - or "\0" in value - or any( - ord(character) < 0x20 or 0xD800 <= ord(character) <= 0xDFFF - for character in value - ) + or any(ord(character) < 0x20 or 0xD800 <= ord(character) <= 0xDFFF for character in value) ): return False path = PurePosixPath(value) return ( - not path.is_absolute() + value not in {"", ".", ".."} + and not path.is_absolute() and path.as_posix() == value and all(part not in {"", ".", ".."} for part in path.parts) ) -def _reject_duplicate_object(pairs: list[tuple[str, object]]) -> dict[str, object]: - result: dict[str, object] = {} - for key, value in pairs: - if key in result: - raise ValueError("duplicate JSON key") - result[key] = value - return result - +def _load_strict_json(data: bytes) -> object: + """Parse JSON, rejecting duplicate keys and NaN or Infinity.""" -def _reject_json_constant(value: str) -> None: - raise ValueError(f"invalid JSON constant: {value}") + return json.loads(data, object_pairs_hook=_strict_json_object, parse_constant=_reject_json_constant) def _lake_manifest(package: str, bundle: _ReleaseBundle) -> bytes: @@ -1195,298 +885,149 @@ def _lake_manifest(package: str, bundle: _ReleaseBundle) -> bytes: def _plan_tree(plan: tuple[_ScaffoldFile, ...]) -> dict[str, object]: - if type(plan) is not tuple: - raise ProjectCreateError( - "project-create-validation-failed", - "The generated project did not satisfy Autoform's project contracts.", - ) + """Nest the plan by path component, rejecting unsafe paths and file/directory collisions.""" + tree: dict[str, object] = {} for item in plan: - if ( - type(item) is not _ScaffoldFile - or type(item.relative) is not str - or type(item.content) is not bytes - or type(item.mode) is not int - or item.mode != item.mode & 0o777 - or item.mode & 0o022 - or not item.mode & 0o400 - ): - raise ProjectCreateError( - "project-create-validation-failed", - "The generated project did not satisfy Autoform's project contracts.", - ) - path = PurePosixPath(item.relative) - if ( - not item.relative - or "\\" in item.relative - or "\0" in item.relative - or any( - ord(character) < 0x20 or 0xD800 <= ord(character) <= 0xDFFF - for character in item.relative - ) - or path.is_absolute() - or path.as_posix() != item.relative - or any(part in {"", ".", ".."} for part in path.parts) - ): - raise ProjectCreateError( - "project-create-validation-failed", - "The generated project did not satisfy Autoform's project contracts.", - ) + if not _safe_relative(item.relative) or item.mode not in {0o644, 0o755}: + raise ProjectCreateError("project-create-validation-failed", _CONTRACTS_MESSAGE) + *directories, name = PurePosixPath(item.relative).parts branch = tree - for part in path.parts[:-1]: - child = branch.setdefault(part, {}) - if not isinstance(child, dict): - raise ProjectCreateError( - "project-create-validation-failed", - "The generated project did not satisfy Autoform's project contracts.", - ) - branch = child - if path.name in branch: - raise ProjectCreateError( - "project-create-validation-failed", - "The generated project did not satisfy Autoform's project contracts.", - ) - branch[path.name] = item + for part in directories: + branch = branch.setdefault(part, {}) + if not isinstance(branch, dict): + raise ProjectCreateError("project-create-validation-failed", _CONTRACTS_MESSAGE) + if name in branch: + raise ProjectCreateError("project-create-validation-failed", _CONTRACTS_MESSAGE) + branch[name] = item return tree -def _write_all(descriptor: int, content: bytes) -> None: - offset = 0 - while offset < len(content): - written = os.write(descriptor, content[offset:]) - if written <= 0: - raise OSError(errno.EIO, "short project file write") - offset += written - - -def _materialize_project(root_descriptor: int, plan: tuple[_ScaffoldFile, ...]) -> None: - tree = _plan_tree(plan) - +def _materialize_project(root_descriptor: int, tree: dict[str, object]) -> None: def write_directory(descriptor: int, entries: dict[str, object]) -> None: for name, entry in sorted(entries.items()): if isinstance(entry, dict): os.mkdir(name, mode=0o700, dir_fd=descriptor) - metadata = os.stat(name, dir_fd=descriptor, follow_symlinks=False) - child = _open_stage(descriptor, name) + child = _open_directory(descriptor, name) try: - if _descriptor_identity(child) != ( - metadata.st_dev, - metadata.st_ino, - metadata.st_uid, - ): - raise OSError(errno.ESTALE, "project directory changed") write_directory(child, entry) os.fchmod(child, 0o755) os.fsync(child) - current = os.stat(name, dir_fd=descriptor, follow_symlinks=False) - if (current.st_dev, current.st_ino, current.st_uid) != ( - metadata.st_dev, - metadata.st_ino, - metadata.st_uid, - ): - raise OSError(errno.ESTALE, "project directory changed") finally: os.close(child) continue - if not isinstance(entry, _ScaffoldFile): - raise OSError(errno.EINVAL, "invalid project plan") flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) child = os.open(name, flags, 0o600, dir_fd=descriptor) try: - _write_all(child, entry.content) + content = memoryview(entry.content) + while content: + written = os.write(child, content) + if written <= 0: + raise OSError(errno.EIO, "short project file write") + content = content[written:] os.fchmod(child, entry.mode) os.fsync(child) finally: os.close(child) - if set(_list_directory(descriptor)) != set(entries): - raise OSError(errno.ESTALE, "project directory changed") write_directory(root_descriptor, tree) -def _verify_project_plan( - root_descriptor: int, - plan: tuple[_ScaffoldFile, ...], - *, - root_mode: int = 0o700, -) -> None: - tree = _plan_tree(plan) +def _stable_metadata(metadata: os.stat_result) -> tuple[int, ...]: + return ( + metadata.st_dev, + metadata.st_ino, + metadata.st_mode, + metadata.st_nlink, + metadata.st_uid, + metadata.st_gid, + metadata.st_size, + metadata.st_mtime_ns, + metadata.st_ctime_ns, + ) + + +def _verify_project_tree(root_descriptor: int, tree: dict[str, object]) -> None: + """Re-read the stage through descriptors and require exactly the planned entries, modes, and bytes.""" + root = os.fstat(root_descriptor) - if ( - not stat.S_ISDIR(root.st_mode) - or root.st_uid != os.geteuid() - or stat.S_IMODE(root.st_mode) != root_mode - ): + if root.st_uid != os.geteuid() or stat.S_IMODE(root.st_mode) != 0o755: raise OSError(errno.ESTALE, "project root changed") - def stable_metadata(metadata: os.stat_result) -> tuple[int, ...]: - return ( - metadata.st_dev, - metadata.st_ino, - metadata.st_mode, - metadata.st_nlink, - metadata.st_uid, - metadata.st_gid, - metadata.st_size, - metadata.st_mtime_ns, - metadata.st_ctime_ns, - ) - def verify_directory(descriptor: int, entries: dict[str, object]) -> None: - directory_before = os.fstat(descriptor) - if set(_list_directory(descriptor)) != set(entries): + before = os.fstat(descriptor) + if _list_directory(descriptor) != set(entries): raise OSError(errno.ESTALE, "project directory changed") for name, entry in sorted(entries.items()): metadata = os.stat(name, dir_fd=descriptor, follow_symlinks=False) - if isinstance(entry, dict): - if not stat.S_ISDIR(metadata.st_mode): - raise OSError(errno.ESTALE, "project directory changed") - child = _open_stage(descriptor, name) - try: - opened = os.fstat(child) - if ( - stable_metadata(opened) != stable_metadata(metadata) - or opened.st_uid != os.geteuid() - or stat.S_IMODE(opened.st_mode) != 0o755 - ): - raise OSError(errno.ESTALE, "project directory changed") - verify_directory(child, entry) - after = os.fstat(child) - current = os.stat(name, dir_fd=descriptor, follow_symlinks=False) - if not (stable_metadata(opened) == stable_metadata(after) == stable_metadata(current)): - raise OSError(errno.ESTALE, "project directory changed") - finally: - os.close(child) - continue - if not isinstance(entry, _ScaffoldFile) or not stat.S_ISREG(metadata.st_mode): - raise OSError(errno.ESTALE, "project file changed") - child = _open_planned_file(descriptor, name) + is_directory = isinstance(entry, dict) + if not (stat.S_ISDIR if is_directory else stat.S_ISREG)(metadata.st_mode): + raise OSError(errno.ESTALE, "project entry changed") + child = (_open_directory if is_directory else _open_planned_file)(descriptor, name) try: opened = os.fstat(child) - content = bytearray() - while len(content) <= len(entry.content): - chunk = os.read( - child, - min(1024 * 1024, len(entry.content) + 1 - len(content)), - ) - if not chunk: - break - content.extend(chunk) + if is_directory: + verify_directory(child, entry) + else: + content = b"" + while len(content) <= len(entry.content): + chunk = os.read(child, len(entry.content) + 1 - len(content)) + if not chunk: + break + content += chunk after = os.fstat(child) current = os.stat(name, dir_fd=descriptor, follow_symlinks=False) if ( - not stat.S_ISREG(opened.st_mode) - or stable_metadata(metadata) != stable_metadata(opened) - or stable_metadata(opened) != stable_metadata(after) - or stable_metadata(after) != stable_metadata(current) - or after.st_nlink != 1 + len({_stable_metadata(item) for item in (metadata, opened, after, current)}) != 1 or after.st_uid != os.geteuid() - or stat.S_IMODE(after.st_mode) != entry.mode - or bytes(content) != entry.content + or stat.S_IMODE(after.st_mode) != (0o755 if is_directory else entry.mode) + or (not is_directory and (after.st_nlink != 1 or content != entry.content)) ): - raise OSError(errno.ESTALE, "project file changed") + raise OSError(errno.ESTALE, "project entry changed") finally: os.close(child) - directory_after = os.fstat(descriptor) - if stable_metadata(directory_before) != stable_metadata(directory_after) or set( - _list_directory(descriptor) - ) != set(entries): + unchanged = _stable_metadata(os.fstat(descriptor)) == _stable_metadata(before) + if not unchanged or _list_directory(descriptor) != set(entries): raise OSError(errno.ESTALE, "project directory changed") verify_directory(root_descriptor, tree) -def _validate_staged_project( - stage_descriptor: int, - plan: tuple[_ScaffoldFile, ...], - package: str, - version: _ProjectVersion, - release_bundle: _ReleaseBundle | None, -) -> None: - _verify_project_plan(stage_descriptor, plan) - indexed = {item.relative: item for item in plan} - expected_core = _core_project_plan(package, version, release_bundle) - if ("lake-manifest.json" in indexed) != (release_bundle is not None) or any( - indexed.get(expected.relative) != expected for expected in expected_core - ): - raise ProjectCreateError( - "project-create-validation-failed", - "The staged project did not satisfy Autoform's project contracts.", - ) - _validate_roadmap_plan(plan) - _verify_project_plan(stage_descriptor, plan) - - def _validate_roadmap_plan(plan: tuple[_ScaffoldFile, ...]) -> None: + invalid = ProjectCreateError("project-create-validation-failed", _STAGED_MESSAGE) roadmap = [ item for item in plan if item.relative.startswith("blueprint/roadmap/") and item.relative.endswith(".md") ] if len(roadmap) != 1 or roadmap[0].relative != "blueprint/roadmap/README.md": - raise ProjectCreateError( - "project-create-validation-failed", - "The staged project did not satisfy Autoform's project contracts.", - ) + raise invalid try: text = roadmap[0].content.decode("utf-8") except UnicodeError: - raise ProjectCreateError( - "project-create-validation-failed", - "The staged project did not satisfy Autoform's project contracts.", - ) from None + raise invalid from None parsed, issues = _parse_node("roadmap", Path("roadmap/README.md"), text) if issues or parsed is None or parsed.statement_targets or parsed.proof_targets: - raise ProjectCreateError( - "project-create-validation-failed", - "The staged project did not satisfy Autoform's project contracts.", - ) + raise invalid -def _rename_noreplace( - source_parent_descriptor: int, - source: str, - target_parent_descriptor: int, - target: str, -) -> None: +def _rename_noreplace(source_parent_descriptor: int, source: str, target_parent_descriptor: int, target: str) -> None: libc = ctypes.CDLL(None, use_errno=True) - source_bytes = os.fsencode(source) - target_bytes = os.fsencode(target) - if hasattr(libc, "renameatx_np"): - function = libc.renameatx_np - function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] - function.restype = ctypes.c_int - result = function( - source_parent_descriptor, - source_bytes, - target_parent_descriptor, - target_bytes, - 0x00000004, - ) - elif hasattr(libc, "renameat2"): - function = libc.renameat2 - function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] - function.restype = ctypes.c_int - result = function( - source_parent_descriptor, - source_bytes, - target_parent_descriptor, - target_bytes, - 1, - ) - else: - raise ProjectCreateError( - "project-create-safety-unavailable", - "This platform cannot atomically publish a new project without replacement.", - ) - if result == 0: + # The no-replace flag is RENAME_EXCL for macOS renameatx_np and RENAME_NOREPLACE for Linux renameat2. + name, flag = ("renameatx_np", 0x4) if hasattr(libc, "renameatx_np") else ("renameat2", 0x1) + if not hasattr(libc, name): + raise ProjectCreateError("project-create-safety-unavailable", _NO_RENAME_MESSAGE) + function = getattr(libc, name) + function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] + function.restype = ctypes.c_int + if ( + function(source_parent_descriptor, os.fsencode(source), target_parent_descriptor, os.fsencode(target), flag) + == 0 + ): return error = ctypes.get_errno() if error in {errno.EEXIST, errno.ENOTEMPTY}: raise FileExistsError(error, os.strerror(error), target) if error in {errno.EINVAL, errno.ENOSYS, errno.ENOTSUP}: - raise ProjectCreateError( - "project-create-safety-unavailable", - "This platform cannot atomically publish a new project without replacement.", - ) + raise ProjectCreateError("project-create-safety-unavailable", _NO_RENAME_MESSAGE) raise OSError(error, os.strerror(error), target) diff --git a/tests/test_project_create.py b/tests/test_project_create.py index 6ab210d9..83a3d42e 100644 --- a/tests/test_project_create.py +++ b/tests/test_project_create.py @@ -192,6 +192,46 @@ def test_incomplete_local_templates_are_not_published(tmp_path: Path, monkeypatc _refused(tmp_path / "Project", "project-create-validation-failed") +@pytest.mark.parametrize("relative", ["lake-manifest.json", "lake-manifest.json/note.md"], ids=["file", "directory"]) +def test_an_unlisted_pair_never_publishes_a_template_manifest( + relative: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + templates = tmp_path / "templates" + shutil.copytree(create_module._TEMPLATES, templates) + (templates / relative).parent.mkdir(exist_ok=True) + (templates / relative).write_text('{"version": "1.1.0", "packages": []}\n', encoding="utf-8") + target = tmp_path / "Project" + monkeypatch.setattr(create_module, "_TEMPLATES", templates) + + with pytest.raises(ProjectCreateError) as raised: + create_project(target, package="Project", release_id=None, lean_toolchain="v4.30.0") + + assert raised.value.code == "project-create-validation-failed" + assert raised.value.message == ( + create_module._STAGED_MESSAGE + " An .autoform-new-* stage may remain; inspect it before removal." + ) + assert not target.exists() + stages = list(tmp_path.glob(".autoform-new-*")) + assert len(stages) == 1 + assert (stages[0] / relative).is_file() + assert stat.S_IMODE(stages[0].stat().st_mode) == 0o700 + + +@pytest.mark.parametrize("relative", _CORE_FILES) +def test_a_template_cannot_replace_a_core_project_file( + relative: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + templates = tmp_path / "templates" + shutil.copytree(create_module._TEMPLATES, templates) + (templates / relative).parent.mkdir(exist_ok=True) + (templates / relative).write_text("template\n", encoding="utf-8") + monkeypatch.setattr(create_module, "_TEMPLATES", templates) + + error = _refused(tmp_path / "Project", "project-create-validation-failed") + + assert error.message == create_module._CONTRACTS_MESSAGE + + def test_group_writable_installed_templates_publish_canonical_modes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -361,6 +401,30 @@ def test_every_release_has_creation_contracts() -> None: } +@pytest.mark.parametrize("resource", ["manifest\ue000.json", "manifest\U0001f600.json"]) +def test_release_metadata_names_its_manifest_below_the_surrogate_range( + monkeypatch: pytest.MonkeyPatch, resource: str +) -> None: + release = load_release_catalog().recommended + name = f"creation-release-{release.id}.json" + payload = json.loads(create_module.files("autoform_cli.project").joinpath(name).read_bytes()) + payload["lake_manifest"] = resource + + class Resources: + def joinpath(self, _name: str) -> Resources: + return self + + def read_bytes(self) -> bytes: + return json.dumps(payload).encode() + + monkeypatch.setattr(create_module, "files", lambda _package: Resources()) + + with pytest.raises(ProjectCreateError) as raised: + create_module._load_creation_release_descriptor(release) + + assert raised.value.code == "project-create-validation-failed" + + @pytest.mark.parametrize("missing", ["Aesop", "Archive", "Counterexamples"]) def test_release_metadata_must_cover_manifest_and_mathlib_production_roots( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, missing: str @@ -628,28 +692,6 @@ def test_catalog_releases_meet_the_lean_floor() -> None: assert create_module._version_warnings(version) == () -def test_unlisted_plan_must_not_carry_a_manifest(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - target = tmp_path / "Project" - original = create_module._build_project_plan - bundle = create_module._load_release_bundle(load_release_catalog().recommended) - - def add_manifest(*args, **kwargs): - plan, pinned = original(*args, **kwargs) - manifest = type(plan[0])("lake-manifest.json", create_module._lake_manifest("Project", bundle), 0o644) - return tuple(sorted((*plan, manifest), key=lambda item: item.relative)), pinned - - monkeypatch.setattr(create_module, "_build_project_plan", add_manifest) - - with pytest.raises(ProjectCreateError) as raised: - create_project(target, package="Project", release_id=None, lean_toolchain="v4.30.0") - - assert raised.value.code == "project-create-validation-failed" - assert not target.exists() - stages = list(tmp_path.glob(".autoform-new-*")) - assert len(stages) == 1 - assert (stages[0] / "lake-manifest.json").is_file() - - def test_long_valid_target_name_does_not_expand_the_stage_name(tmp_path: Path) -> None: name_limit = os.pathconf(tmp_path, "PC_NAME_MAX") if name_limit < 64: @@ -805,38 +847,17 @@ def fail(*args, **kwargs): assert not list(stages[0].iterdir()) -def test_injected_validation_failure_preserves_stage_for_safe_recovery( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - target = tmp_path / "project" - - def fail(*args, **kwargs): - raise ProjectCreateError("project-create-validation-failed", "invalid") - - monkeypatch.setattr(create_module, "_validate_staged_project", fail) - with pytest.raises(ProjectCreateError) as raised: - create_project(target, package="Project", release_id=_RELEASE) - assert raised.value.code == "project-create-validation-failed" - assert ".autoform-new-* stage may remain" in raised.value.message - assert not target.exists() - stages = list(tmp_path.glob(".autoform-new-*")) - assert len(stages) == 1 - assert (stages[0] / "lean-toolchain").is_file() - - def test_invalid_planned_roadmap_is_never_published(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: target = tmp_path / "project" original = create_module._build_project_plan def corrupt(*args, **kwargs): - plan, pinned = original(*args, **kwargs) - changed = tuple( + return tuple( type(item)(item.relative, b"No H1 title.\n", item.mode) if item.relative == "blueprint/roadmap/README.md" else item - for item in plan + for item in original(*args, **kwargs) ) - return changed, pinned monkeypatch.setattr(create_module, "_build_project_plan", corrupt) @@ -848,23 +869,19 @@ def corrupt(*args, **kwargs): assert not list(tmp_path.glob(".autoform-new-*")) -@pytest.mark.parametrize("corruption", ["container", "relative", "content", "mode"]) -def test_plan_requires_exact_types_and_safe_file_modes_before_writing( +@pytest.mark.parametrize("corruption", ["relative", "mode"]) +def test_plan_requires_safe_paths_and_file_modes_before_writing( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, corruption: str ) -> None: original = create_module._build_project_plan def corrupt(*args, **kwargs): - plan, pinned = original(*args, **kwargs) - if corruption == "container": - return list(plan), pinned - first, *rest = plan + first, *rest = original(*args, **kwargs) changed = { "relative": type(first)(Path(first.relative), first.content, first.mode), - "content": type(first)(first.relative, memoryview(first.content), first.mode), "mode": type(first)(first.relative, first.content, 0o666), }[corruption] - return (changed, *rest), pinned + return (changed, *rest) monkeypatch.setattr(create_module, "_build_project_plan", corrupt) @@ -1109,7 +1126,7 @@ def test_requested_parent_rebind_before_publish_preserves_the_stage( parent.mkdir(mode=0o700) parent.chmod(0o755) target = parent / "Project" - original = create_module._validate_staged_project + original = create_module._materialize_project def rebind(*args, **kwargs) -> None: original(*args, **kwargs) @@ -1117,7 +1134,7 @@ def rebind(*args, **kwargs) -> None: parent.mkdir(mode=0o700) parent.chmod(0o755) - monkeypatch.setattr(create_module, "_validate_staged_project", rebind) + monkeypatch.setattr(create_module, "_materialize_project", rebind) with pytest.raises(ProjectCreateError) as raised: create_project(target, package="Project", release_id=_RELEASE) @@ -1190,9 +1207,20 @@ def fail_second(path: Path, expected_identity: tuple[int, int, int]) -> int: assert inspect_project(target).ok -def test_workspace_substitution_fails_before_publication(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +@pytest.mark.parametrize("template_manifest", [False, True], ids=["catalog-release", "unlisted-template-manifest"]) +def test_workspace_substitution_fails_before_publication( + template_manifest: bool, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: target = tmp_path / "project" - original = create_module._validate_staged_project + versions: dict[str, str | None] = {"release_id": _RELEASE} + if template_manifest: + # The identity check runs before the manifest refusal, so the code stays main's. + templates = tmp_path / "templates" + shutil.copytree(create_module._TEMPLATES, templates) + (templates / "lake-manifest.json").write_text('{"version": "1.1.0", "packages": []}\n', encoding="utf-8") + monkeypatch.setattr(create_module, "_TEMPLATES", templates) + versions = {"release_id": None, "lean_toolchain": "v4.30.0"} + original = create_module._materialize_project def substitute(*args, **kwargs) -> None: original(*args, **kwargs) @@ -1202,35 +1230,14 @@ def substitute(*args, **kwargs) -> None: stage.mkdir(mode=0o700) (stage / "FOREIGN").write_text("foreign\n", encoding="utf-8") - monkeypatch.setattr(create_module, "_validate_staged_project", substitute) + monkeypatch.setattr(create_module, "_materialize_project", substitute) with pytest.raises(ProjectCreateError) as raised: - create_project(target, package="Project", release_id=_RELEASE) + create_project(target, package="Project", **versions) assert raised.value.code == "project-create-failed" assert not target.exists() assert any(path.name == "FOREIGN" for path in tmp_path.rglob("FOREIGN")) - - -def test_corrupt_core_plan_is_rejected_without_path_based_inspection( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - target = tmp_path / "project" - original = create_module._build_project_plan - - def corrupt(*args, **kwargs): - plan, pinned = original(*args, **kwargs) - changed = tuple( - type(item)(item.relative, b"leanprover/lean4:v0.0.0\n", item.mode) - if item.relative == "lean-toolchain" - else item - for item in plan - ) - return changed, pinned - - monkeypatch.setattr(create_module, "_build_project_plan", corrupt) - with pytest.raises(ProjectCreateError) as raised: - create_project(target, package="Project", release_id=_RELEASE) - assert raised.value.code == "project-create-validation-failed" - assert not target.exists() + # The renamed stage is refused before its chmod, so the written tree stays private. + assert stat.S_IMODE(next(tmp_path.glob(".autoform-new-*-owned")).stat().st_mode) == 0o700 def test_stage_path_substitution_never_writes_to_symlink_target( @@ -1261,19 +1268,102 @@ def substitute(stage_descriptor, plan) -> None: def test_stage_open_failure_preserves_the_owned_empty_stage(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: target = tmp_path / "project" - original = create_module._open_stage + original = create_module._open_directory def fail_first_open(parent_descriptor: int, stage_name: str) -> int: if stage_name.startswith(".autoform-new-"): raise OSError("injected stage open failure") return original(parent_descriptor, stage_name) - monkeypatch.setattr(create_module, "_open_stage", fail_first_open) + monkeypatch.setattr(create_module, "_open_directory", fail_first_open) + + with pytest.raises(ProjectCreateError) as raised: + create_project(target, package="Project", release_id=_RELEASE) + + assert raised.value.code == "project-create-failed" + assert not target.exists() + stages = list(tmp_path.glob(".autoform-new-*")) + assert len(stages) == 1 + assert not list(stages[0].iterdir()) + + +def test_stage_substitution_is_refused_before_writing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + target = tmp_path / "project" + original = create_module._open_directory + + def substitute_after_open(parent_descriptor: int, name: str) -> int: + descriptor = original(parent_descriptor, name) + if name.startswith(".autoform-new-"): + (tmp_path / name).rename(tmp_path / f"{name}-owned") + (tmp_path / name).mkdir(mode=0o700) + return descriptor + + monkeypatch.setattr(create_module, "_open_directory", substitute_after_open) + + with pytest.raises(ProjectCreateError) as raised: + create_project(target, package="Project", release_id=_RELEASE) + + assert raised.value.code == "project-create-failed" + assert ".autoform-new-* stage may remain" in raised.value.message + assert not target.exists() + # Neither the replacement nor the opened stage, now under its -owned name, received a file. + stages = list(tmp_path.glob(".autoform-new-*")) + assert len(stages) == 2 + assert not any(list(stage.iterdir()) for stage in stages) + + +@pytest.mark.parametrize( + ("mode", "entries"), [(0o700, {"private.txt"}), (0o755, set())], ids=["nonempty-0700", "empty-0755"] +) +def test_swapped_in_stage_directory_is_refused_before_writing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mode: int, entries: set[str] +) -> None: + target = tmp_path / "project" + decoy = tmp_path / "decoy" + decoy.mkdir() + for name in entries: + (decoy / name).write_text("private\n", encoding="utf-8") + decoy.chmod(mode) + original = create_module._open_directory + + def swap_before_open(parent_descriptor: int, name: str) -> int: + if name.startswith(".autoform-new-") and decoy.exists(): + (tmp_path / name).rename(tmp_path / "original-stage") + decoy.rename(tmp_path / name) + return original(parent_descriptor, name) + + monkeypatch.setattr(create_module, "_open_directory", swap_before_open) + + with pytest.raises(ProjectCreateError) as raised: + create_project(target, package="Project", release_id=_RELEASE) + + assert raised.value.code == "project-create-failed" + assert ".autoform-new-* stage may remain" in raised.value.message + assert not target.exists() + (swapped,) = tmp_path.glob(".autoform-new-*") + assert stat.S_IMODE(swapped.stat().st_mode) == mode + assert {path.name for path in swapped.iterdir()} == entries + assert not list((tmp_path / "original-stage").iterdir()) + + +def test_foreign_owned_stage_is_refused_before_writing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + target = tmp_path / "project" + original = create_module._create_stage + euid = os.geteuid() + + def foreign_stage(parent_descriptor: int) -> str: + name = original(parent_descriptor) + # From here on the stage looks like a directory another uid put in its place. + monkeypatch.setattr(create_module.os, "geteuid", lambda: euid + 1) + return name + + monkeypatch.setattr(create_module, "_create_stage", foreign_stage) with pytest.raises(ProjectCreateError) as raised: create_project(target, package="Project", release_id=_RELEASE) assert raised.value.code == "project-create-failed" + assert ".autoform-new-* stage may remain" in raised.value.message assert not target.exists() stages = list(tmp_path.glob(".autoform-new-*")) assert len(stages) == 1 @@ -1282,14 +1372,16 @@ def fail_first_open(parent_descriptor: int, stage_name: str) -> int: def test_failure_path_never_attempts_recursive_deletion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: target = tmp_path / "project" + original = create_module._materialize_project def fail(*args, **kwargs): + original(*args, **kwargs) raise OSError("injected") def forbidden(*args, **kwargs): raise AssertionError("project creation attempted destructive cleanup") - monkeypatch.setattr(create_module, "_validate_staged_project", fail) + monkeypatch.setattr(create_module, "_materialize_project", fail) monkeypatch.setattr(create_module.os, "unlink", forbidden) monkeypatch.setattr(create_module.os, "rmdir", forbidden) @@ -1307,7 +1399,7 @@ def test_failure_cleanup_never_recurses_into_a_foreign_directory( victim = tmp_path / "victim" victim.mkdir() (victim / "KEEP").write_text("keep\n", encoding="utf-8") - original = create_module._validate_staged_project + original = create_module._materialize_project def substitute(*args, **kwargs): original(*args, **kwargs) @@ -1316,7 +1408,7 @@ def substitute(*args, **kwargs): victim.rename(stage / "blueprint") raise OSError("injected") - monkeypatch.setattr(create_module, "_validate_staged_project", substitute) + monkeypatch.setattr(create_module, "_materialize_project", substitute) with pytest.raises(ProjectCreateError) as raised: create_project(target, package="Project", release_id=_RELEASE) @@ -1369,16 +1461,16 @@ def make_world_writable(*args, **kwargs): assert stat.S_IMODE((stage / "blueprint").stat().st_mode) == 0o777 -def test_mutation_after_validation_is_not_published(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_mutated_stage_is_not_published(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: target = tmp_path / "project" - original = create_module._validate_staged_project + original = create_module._materialize_project def mutate(*args, **kwargs): original(*args, **kwargs) stage = next(tmp_path.glob(".autoform-new-*")) (stage / "lean-toolchain").write_text("mutated\n", encoding="utf-8") - monkeypatch.setattr(create_module, "_validate_staged_project", mutate) + monkeypatch.setattr(create_module, "_materialize_project", mutate) with pytest.raises(ProjectCreateError) as raised: create_project(target, package="Project", release_id=_RELEASE)