diff --git a/.github/workflows/npm-nested-metadata-validation-quality-ci.yml b/.github/workflows/npm-nested-metadata-validation-quality-ci.yml new file mode 100644 index 000000000..c73493291 --- /dev/null +++ b/.github/workflows/npm-nested-metadata-validation-quality-ci.yml @@ -0,0 +1,132 @@ +name: npm Nested Metadata Validation Quality CI + +on: + pull_request: + branches: [main] + paths: + - ".github/workflows/npm-nested-metadata-validation-quality-ci.yml" + - "scripts/ci/materialize_base_javascript_packages.py" + - "tests/conftest.py" + - "tests/test_javascript_materializer_descriptor_ancestry.py" + - "tests/test_javascript_materializer_existing_directory_coverage.py" + - "tests/test_javascript_materializer_output_edge_coverage.py" + - "tests/test_javascript_materializer_output_security.py" + - "tests/test_materialize_base_javascript_packages.py" + - "tests/test_npm_nested_metadata_lock_validation.py" + - "docs/doctoring/npm-nested-metadata-canonical-pins.md" + - "requirements-opencode-review-ci-hashes.txt" + - "CHANGELOG.md" + push: + branches: [main] + paths: + - ".github/workflows/npm-nested-metadata-validation-quality-ci.yml" + - "scripts/ci/materialize_base_javascript_packages.py" + - "tests/conftest.py" + - "tests/test_javascript_materializer_descriptor_ancestry.py" + - "tests/test_javascript_materializer_existing_directory_coverage.py" + - "tests/test_javascript_materializer_output_edge_coverage.py" + - "tests/test_javascript_materializer_output_security.py" + - "tests/test_materialize_base_javascript_packages.py" + - "tests/test_npm_nested_metadata_lock_validation.py" + - "docs/doctoring/npm-nested-metadata-canonical-pins.md" + - "requirements-opencode-review-ci-hashes.txt" + - "CHANGELOG.md" + +concurrency: + group: npm-nested-metadata-quality-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true + +jobs: + python-310-compatibility: + name: Python 3.10 compatibility + runs-on: ubuntu-24.04 + timeout-minutes: 10 + steps: + - name: Harden runner + uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0 + with: + egress-policy: audit + - name: Checkout exact source revision + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Set up Python 3.10 + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.10" + - name: Compile implementation and contracts + run: | + python -m compileall -q \ + scripts/ci/materialize_base_javascript_packages.py \ + tests/conftest.py \ + tests/test_javascript_materializer_descriptor_ancestry.py \ + tests/test_javascript_materializer_existing_directory_coverage.py \ + tests/test_javascript_materializer_output_edge_coverage.py \ + tests/test_javascript_materializer_output_security.py \ + tests/test_materialize_base_javascript_packages.py \ + tests/test_npm_nested_metadata_lock_validation.py + + python-314-quality: + name: Python 3.14 complete quality + runs-on: ubuntu-24.04 + timeout-minutes: 30 + steps: + - name: Harden runner + uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0 + with: + egress-policy: audit + - name: Checkout exact source revision + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Set up Python 3.14 + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + cache: pip + cache-dependency-path: requirements-opencode-review-ci-hashes.txt + - name: Install exact hash-locked quality tooling + run: >- + python -m pip install --disable-pip-version-check --require-hashes + -r requirements-opencode-review-ci-hashes.txt + - name: Run focused tests with complete production branch coverage + run: | + python -m coverage erase + python -m coverage run --branch -m pytest -q \ + tests/test_javascript_materializer_descriptor_ancestry.py \ + tests/test_javascript_materializer_existing_directory_coverage.py \ + tests/test_javascript_materializer_output_edge_coverage.py \ + tests/test_javascript_materializer_output_security.py \ + tests/test_materialize_base_javascript_packages.py \ + tests/test_npm_nested_metadata_lock_validation.py + python -m coverage report \ + --include=scripts/ci/materialize_base_javascript_packages.py \ + --show-missing \ + --fail-under=100 + - name: Enforce complete production docstrings and compilation + run: | + python -m interrogate --fail-under 100 scripts/ci/materialize_base_javascript_packages.py + python -m compileall -q \ + scripts/ci/materialize_base_javascript_packages.py \ + tests/conftest.py \ + tests/test_javascript_materializer_descriptor_ancestry.py \ + tests/test_javascript_materializer_existing_directory_coverage.py \ + tests/test_javascript_materializer_output_edge_coverage.py \ + tests/test_javascript_materializer_output_security.py \ + tests/test_materialize_base_javascript_packages.py \ + tests/test_npm_nested_metadata_lock_validation.py + - name: Run complete central regression suite + run: | + python -m coverage erase + python -m coverage run -m pytest tests -q + python -m coverage report --show-missing --fail-under=100 + - name: Verify clean patches + run: git diff --check diff --git a/CHANGELOG.md b/CHANGELOG.md index c993bf7cb..c3871690e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ Semantic Versioning where the repository publishes a release. ### Fixed +- Accepted npm v2/v3 metadata-only nested workspace and peer locations only when one exact scoped or unscoped canonical root package carries the same version, HTTPS npm-registry tarball, and canonical SHA-512 integrity, while continuing to reject malformed identities, partial pins, metadata-only roots, alternate origins, and version drift. - Bound each review-agent invocation key to the wrapper's complete canonical payload, including the base branch and requesting actor; altered fields with a valid-format key now fail before durable-leader election or forwarding, and wrapper write permission is job-scoped. - Bound both trusted-uv quality jobs to `github.event.pull_request.head.sha` and added a permanent two-checkout regression contract so exact-head compatibility, coverage, docstring, and compilation claims cannot silently measure GitHub's generated pull-request merge revision. - Made Strix treat only a single LiteLLM provider-error line containing NVIDIA NIM context and model-catalog 404 evidence as cross-model fallback evidence, rejecting cross-line signal assembly and provider-like target source literals; moved the public default to Nemotron 3 Super 120B and added a second NVIDIA hosted candidate before GitHub Models without neutralizing reported vulnerabilities. diff --git a/docs/doctoring/npm-nested-metadata-canonical-pins.md b/docs/doctoring/npm-nested-metadata-canonical-pins.md new file mode 100644 index 000000000..0f6b76802 --- /dev/null +++ b/docs/doctoring/npm-nested-metadata-canonical-pins.md @@ -0,0 +1,74 @@ +# npm nested metadata canonical pins + +## Decision + +Changed npm lockfiles remain untrusted pull-request inputs. The central JavaScript dependency materializer accepts npm lockfile versions 2 and 3 only after validating the complete `packages` map. Every non-link package location under a `node_modules` segment must declare a nonempty exact `version`. + +npm can serialize a nested workspace or peer location with version and classification metadata while the canonical root location carries the registry tarball and integrity fields. The validator therefore distinguishes two safe forms: + +1. **Complete pin** — the location declares both `resolved` and `integrity`. The URL must be an HTTPS tarball on `registry.npmjs.org` with no user information, explicit port, query, or fragment, and the integrity value must be one canonical SHA-512 SRI value. +2. **Metadata-only nested location** — the location declares neither field. It is accepted only when `node_modules/` contains one complete pin for the same scoped or unscoped package identity and the exact same version. + +A metadata-only canonical root entry is forbidden. A nested location that declares only one of `resolved` or `integrity` is also forbidden. Independently complete nested pins remain valid and may carry a different version because their bytes and integrity are self-contained. + +## Package identity + +Identity is derived from the path segments after the final `node_modules` component: + +- unscoped: exactly one segment, such as `react`; +- scoped: exactly two segments, such as `@types/react-dom`. + +Incomplete scopes, additional identity segments, absolute paths, backslashes, and parent traversal fail closed. Workspace links retain their separate bounded relative-link validation and never inherit registry metadata. + +```mermaid +flowchart TD + A[npm packages map entry] --> B{link is true?} + B -->|yes| C[Validate bounded relative workspace target] + B -->|no| D[Derive exact package identity and require version] + D --> E{resolved and integrity} + E -->|both present| F[Validate exact npm registry tarball and SHA-512 SRI] + E -->|one present| G[Reject partial pin] + E -->|both absent| H{canonical root?} + H -->|yes| I[Reject metadata-only root] + H -->|no| J[Require same identity and version at complete root pin] +``` + +## Security and compatibility boundary + +The policy does not repair, synthesize, or mutate lockfile metadata. It consumes the validated lock unchanged. It preserves the existing lockfile version, path, link, URL, origin, tarball suffix, and SHA-512 controls while admitting npm's location-keyed metadata representation. + +The canonical root pin is a provenance anchor for metadata-only locations, not a claim that all nested locations share one physical installation. A complete nested record is validated independently and does not depend on the root. Missing roots, version drift, malformed identity, partial fields, alternate registries, malformed URLs, and invalid integrity remain blocking. + +## Verification + +The permanent regression suite includes: + +- the BandScope `apps/desktop/node_modules/@types/react-dom` peer-location shape; +- unscoped metadata-only locations; +- independently pinned nested versions; +- missing canonical pins; +- canonical-version mismatch; +- metadata-only canonical roots; +- partial `resolved` or `integrity` declarations; +- malformed scoped identities; +- nonempty-version enforcement; +- alternate origins and invalid SHA-512 SRI values; and +- all pre-existing npm path, link, lockfile, URL, and integrity cases. + +The dedicated quality workflow runs Python 3.10 compilation, Python 3.14 focused tests with 100% production statement and branch coverage, 100% production docstrings, the complete central test suite, and a clean-patch check. + +## Incident recovery and rollback + +1. Preserve the exact pull-request head SHA, lockfile blob SHA, validation error, and quality-run ID. +2. Determine whether the changed lock is malformed or whether npm produced a supported metadata-only nested location. +3. Never add missing tarball or integrity values by hand. Regenerate the lock with the repository's pinned npm version when the lock is invalid. +4. Roll back only by restoring the prior fail-closed validator or another reviewed implementation that keeps the same identity, version, origin, and integrity controls. +5. Rerun the complete exact-head quality, security, and supply-chain matrix after any repair. + +## References + +npm, Inc. (2026). *package-lock.json*. npm Docs. https://docs.npmjs.com/cli/v11/configuring-npm/package-lock-json + +npm, Inc. (2026). *npm install*. npm Docs. https://docs.npmjs.com/cli/v11/commands/npm-install + +World Wide Web Consortium. (2016). *Subresource Integrity*. https://www.w3.org/TR/SRI/ diff --git a/scripts/ci/materialize_base_javascript_packages.py b/scripts/ci/materialize_base_javascript_packages.py old mode 100644 new mode 100755 index 407c17aa1..3ad466914 --- a/scripts/ci/materialize_base_javascript_packages.py +++ b/scripts/ci/materialize_base_javascript_packages.py @@ -12,8 +12,10 @@ import argparse import json +import os import pathlib import re +import stat import subprocess import sys import urllib.parse @@ -26,6 +28,9 @@ NPM_LOCK_NAMES = ("npm-shrinkwrap.json", "package-lock.json") NPM_REGISTRY_HOST = "registry.npmjs.org" SHA512_SRI_RE = re.compile(r"^sha512-[A-Za-z0-9+/]{86}==$") +_DIRECTORY_OPEN_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW +_NEW_FILE_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW +_REQUIRED_DIR_FD_FUNCTIONS = (os.open, os.mkdir, os.stat, os.unlink, os.rmdir) def _git(repo_root: pathlib.Path, *args: str) -> bytes: @@ -112,12 +117,6 @@ def base_pnpm_projects( str(project_root / lock_name) in regular_paths for lock_name in NPM_LOCK_NAMES ): - # A sibling npm lock means npm owns this project and the - # pnpm-lock.yaml is a vestigial second lockfile. Skip pnpm - # materialization so the downstream npm install path handles - # it, instead of failing the whole coverage-evidence job. A - # genuine pnpm-only project (no sibling npm lock) still must - # pin an exact pnpm packageManager. continue raise ValueError( f"trusted base package manifest {package_path} must declare an exact pnpm packageManager version" @@ -194,8 +193,6 @@ def base_npm_projects( ) package_manager = package_data.get("packageManager") if isinstance(package_manager, str) and PNPM_SPEC_RE.fullmatch(package_manager): - # An exact pnpm declaration owns this project. A sibling npm lock - # is vestigial and must not create a second dependency cache. continue lock_content = _git(repo_root, "show", f"{base_sha}:{lock_path}") @@ -249,6 +246,73 @@ def _lock_blob_sha(repo_root: pathlib.Path, revision_sha: str, lock_path: str) - return blob_sha.lower() +def _npm_package_identity( + lock_path: str, + package_path: str, + candidate: pathlib.PurePosixPath, +) -> str: + """Return the exact npm identity after the final ``node_modules`` segment.""" + final_node_modules = max( + index for index, part in enumerate(candidate.parts) if part == "node_modules" + ) + identity_parts = candidate.parts[final_node_modules + 1 :] + if ( + len(identity_parts) == 1 + and identity_parts[0] + and not identity_parts[0].startswith("@") + ): + return identity_parts[0] + if ( + len(identity_parts) == 2 + and identity_parts[0].startswith("@") + and len(identity_parts[0]) > 1 + and identity_parts[1] + and not identity_parts[1].startswith("@") + ): + return "/".join(identity_parts) + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} has a malformed npm package identity" + ) + + +def _validate_npm_registry_pin( + lock_path: str, + package_path: str, + resolved: Any, + integrity: Any, +) -> None: + """Validate one exact public-registry tarball and SHA-512 integrity pair.""" + if not isinstance(resolved, str) or not isinstance(integrity, str): + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must pin a registry tarball and SHA-512 integrity" + ) + parsed = urllib.parse.urlsplit(resolved) + try: + parsed_port = parsed.port + except ValueError as exc: + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} has an invalid registry URL" + ) from exc + if ( + parsed.scheme != "https" + or parsed.hostname != NPM_REGISTRY_HOST + or parsed.username is not None + or parsed.password is not None + or parsed_port is not None + or parsed.query + or parsed.fragment + or not parsed.path.startswith("/") + or not parsed.path.endswith(".tgz") + ): + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must resolve from https://{NPM_REGISTRY_HOST}/" + ) + if not SHA512_SRI_RE.fullmatch(integrity): + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must use one SHA-512 integrity value" + ) + + def validate_head_npm_lock(lock_path: str, lock_content: bytes) -> None: """Fail closed unless a changed HEAD npm lock is registry- and hash-bounded.""" try: @@ -274,6 +338,8 @@ def validate_head_npm_lock(lock_path: str, lock_content: bytes) -> None: f"current-head npm lock {lock_path} must contain an object-valued packages map" ) + canonical_versions: dict[str, str] = {} + metadata_only_locations: list[tuple[str, str, str]] = [] for package_path, metadata in sorted(packages.items()): if not isinstance(package_path, str) or not isinstance(metadata, dict): raise ValueError( @@ -291,6 +357,7 @@ def validate_head_npm_lock(lock_path: str, lock_content: bytes) -> None: if not package_path or "node_modules" not in candidate.parts: continue + identity = _npm_package_identity(lock_path, package_path, candidate) resolved = metadata.get("resolved") if metadata.get("link") is True: if not isinstance(resolved, str) or not resolved or "\\" in resolved: @@ -308,36 +375,325 @@ def validate_head_npm_lock(lock_path: str, lock_content: bytes) -> None: ) continue - integrity = metadata.get("integrity") - if not isinstance(resolved, str) or not isinstance(integrity, str): + has_resolved = "resolved" in metadata + has_integrity = "integrity" in metadata + if has_resolved != has_integrity: + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must not partially declare a registry tarball and SHA-512 integrity" + ) + if has_resolved: + _validate_npm_registry_pin( + lock_path, + package_path, + metadata.get("resolved"), + metadata.get("integrity"), + ) + + version = metadata.get("version") + canonical_path = f"node_modules/{identity}" + is_canonical_root = package_path == canonical_path + if not isinstance(version, str) or not version: + if is_canonical_root and not has_resolved: + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must pin a registry tarball and SHA-512 integrity" + ) raise ValueError( - f"current-head npm lock {lock_path} package {package_path} must pin a registry tarball and SHA-512 integrity" + f"current-head npm lock {lock_path} package {package_path} must declare a nonempty exact version" ) - parsed = urllib.parse.urlsplit(resolved) + + if has_resolved: + if is_canonical_root: + canonical_versions[identity] = version + continue + + if is_canonical_root: + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must be a complete canonical root pin" + ) + metadata_only_locations.append((package_path, identity, version)) + + for package_path, identity, version in metadata_only_locations: + canonical_version = canonical_versions.get(identity) + if canonical_version is None: + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} has no complete canonical root pin" + ) + if canonical_version != version: + raise ValueError( + f"current-head npm lock {lock_path} package {package_path} must match the exact canonical version {canonical_version}" + ) + + +def _require_descriptor_relative_capabilities() -> None: + """Fail before mutation when required descriptor-relative filesystem APIs are absent.""" + supported = getattr(os, "supports_dir_fd", set()) + if any(function not in supported for function in _REQUIRED_DIR_FD_FUNCTIONS): + raise ValueError("descriptor-relative output operations are unavailable") + if not all(hasattr(os, name) for name in ("O_DIRECTORY", "O_NOFOLLOW")): + raise ValueError("descriptor-relative output operations are unavailable") + + +def _reject_symlinked_output_components(output_dir: pathlib.Path) -> None: + """Reject existing symlink components before materialization writes begin.""" + candidate = output_dir.absolute() + if candidate == pathlib.Path(candidate.anchor): + raise ValueError("output directory must not be the filesystem root") + current = pathlib.Path(candidate.anchor) + for component in candidate.parts[1:]: + current /= component + if current.is_symlink(): + raise ValueError( + "output directory must not be a symlink; " + f"path must not contain symlinks: {current}" + ) + if not current.exists(): + break + if not current.is_dir(): + raise ValueError( + f"output directory path component must be a directory: {current}" + ) + + +def _directory_identity(metadata: os.stat_result) -> tuple[int, int]: + """Return one directory device/inode identity after validating its file type.""" + if not stat.S_ISDIR(metadata.st_mode): + raise ValueError("output directory binding changed during secure materialization") + return metadata.st_dev, metadata.st_ino + + +def _open_output_directory(output_dir: pathlib.Path) -> tuple[int, tuple[int, int]]: + """Open one no-follow output directory while detecting ancestor replacement races.""" + candidate = output_dir.absolute() + _reject_symlinked_output_components(candidate) + candidate.parent.mkdir(parents=True, exist_ok=True) + candidate.mkdir(exist_ok=True) + _reject_symlinked_output_components(candidate) + + expected_parent = _directory_identity( + os.stat(candidate.parent, follow_symlinks=False) + ) + expected_output = _directory_identity(os.stat(candidate, follow_symlinks=False)) + parent_fd = os.open(candidate.parent, _DIRECTORY_OPEN_FLAGS) + try: + if _directory_identity(os.fstat(parent_fd)) != expected_parent: + raise ValueError( + "output directory ancestor changed during secure materialization" + ) + output_fd = os.open( + candidate.name, + _DIRECTORY_OPEN_FLAGS, + dir_fd=parent_fd, + ) try: - parsed_port = parsed.port - except ValueError as exc: + if _directory_identity(os.fstat(output_fd)) != expected_output: + raise ValueError( + "output directory changed during secure materialization" + ) + os.fsync(parent_fd) + os.fsync(output_fd) + metadata = os.fstat(output_fd) + return output_fd, (metadata.st_dev, metadata.st_ino) + except BaseException: + os.close(output_fd) + raise + finally: + os.close(parent_fd) + + +def _verify_output_directory_binding( + output_dir: pathlib.Path, + output_fd: int, + identity: tuple[int, int], +) -> None: + """Fail closed if the published output pathname no longer names the opened directory.""" + descriptor_metadata = os.fstat(output_fd) + try: + path_metadata = os.stat(output_dir.absolute(), follow_symlinks=False) + except OSError as exc: + raise ValueError("output directory changed during secure materialization") from exc + if ( + not stat.S_ISDIR(path_metadata.st_mode) + or (descriptor_metadata.st_dev, descriptor_metadata.st_ino) != identity + or (path_metadata.st_dev, path_metadata.st_ino) != identity + ): + raise ValueError("output directory changed during secure materialization") + + +def _safe_relative_parts(relative_path: str) -> tuple[str, ...]: + """Return one normalized relative POSIX output path or fail closed.""" + candidate = pathlib.PurePosixPath(relative_path) + if ( + not relative_path + or "\\" in relative_path + or candidate.is_absolute() + or ".." in candidate.parts + or candidate.as_posix() != relative_path + or not candidate.parts + ): + raise ValueError(f"unsafe relative output path: {relative_path!r}") + return candidate.parts + + +def _open_relative_directory(root_fd: int, parts: tuple[str, ...]) -> int: + """Open or create child directories and bind each name to its observed inode.""" + current_fd = os.dup(root_fd) + try: + for part in parts: + created = False + try: + os.mkdir(part, mode=0o700, dir_fd=current_fd) + created = True + except FileExistsError: + pass + expected_identity = _directory_identity( + os.stat(part, dir_fd=current_fd, follow_symlinks=False) + ) + if created: + os.fsync(current_fd) + next_fd = os.open(part, _DIRECTORY_OPEN_FLAGS, dir_fd=current_fd) + try: + if _directory_identity(os.fstat(next_fd)) != expected_identity: + raise ValueError( + "output directory binding changed during secure materialization" + ) + os.fsync(next_fd) + except BaseException: + os.close(next_fd) + raise + os.close(current_fd) + current_fd = next_fd + return current_fd + except BaseException: + os.close(current_fd) + raise + + +def _create_project_directory(output_fd: int, directory: str) -> int: + """Create and bind a fresh project directory beneath the pinned output descriptor.""" + try: + os.mkdir(directory, mode=0o700, dir_fd=output_fd) + except FileExistsError as exc: + raise ValueError( + f"generated output path must not pre-exist: {directory}" + ) from exc + expected_identity = _directory_identity( + os.stat(directory, dir_fd=output_fd, follow_symlinks=False) + ) + os.fsync(output_fd) + project_fd = os.open(directory, _DIRECTORY_OPEN_FLAGS, dir_fd=output_fd) + try: + if _directory_identity(os.fstat(project_fd)) != expected_identity: raise ValueError( - f"current-head npm lock {lock_path} package {package_path} has an invalid registry URL" - ) from exc - if ( - parsed.scheme != "https" - or parsed.hostname != NPM_REGISTRY_HOST - or parsed.username is not None - or parsed.password is not None - or parsed_port is not None - or parsed.query - or parsed.fragment - or not parsed.path.startswith("/") - or not parsed.path.endswith(".tgz") - ): + "output directory binding changed during secure materialization" + ) + os.fsync(project_fd) + return project_fd + except BaseException: + os.close(project_fd) + raise + + +def _unlink_owned_file( + parent_fd: int, + filename: str, + identity: tuple[int, int], +) -> None: + """Remove one failed file only when its published name still identifies our inode.""" + try: + path_metadata = os.stat(filename, dir_fd=parent_fd, follow_symlinks=False) + except OSError: + return + if (path_metadata.st_dev, path_metadata.st_ino) != identity: + return + try: + os.unlink(filename, dir_fd=parent_fd) + except OSError: + return + os.fsync(parent_fd) + + +def _remove_owned_empty_directory( + parent_fd: int, + directory: str, + identity: tuple[int, int], +) -> None: + """Remove one empty generated directory only while its original inode is published.""" + try: + path_metadata = os.stat(directory, dir_fd=parent_fd, follow_symlinks=False) + except OSError: + return + if ( + not stat.S_ISDIR(path_metadata.st_mode) + or (path_metadata.st_dev, path_metadata.st_ino) != identity + ): + return + try: + os.rmdir(directory, dir_fd=parent_fd) + except OSError: + return + os.fsync(parent_fd) + + +def _write_new_file(parent_fd: int, filename: str, content: bytes) -> None: + """Create, synchronize, revalidate, and clean up one descriptor-pinned file.""" + try: + file_fd = os.open( + filename, + _NEW_FILE_FLAGS, + 0o600, + dir_fd=parent_fd, + ) + except FileExistsError as exc: + raise ValueError( + f"generated output file must not pre-exist: {filename}" + ) from exc + initial_metadata = os.fstat(file_fd) + identity = (initial_metadata.st_dev, initial_metadata.st_ino) + try: + if not stat.S_ISREG(initial_metadata.st_mode) or initial_metadata.st_nlink != 1: raise ValueError( - f"current-head npm lock {lock_path} package {package_path} must resolve from https://{NPM_REGISTRY_HOST}/" + "generated output files must be singly linked regular files" ) - if not SHA512_SRI_RE.fullmatch(integrity): + view = memoryview(content) + offset = 0 + while offset < len(view): + written = os.write(file_fd, view[offset:]) + if written <= 0: + raise OSError("output write made no progress") + offset += written + os.fsync(file_fd) + final_metadata = os.fstat(file_fd) + path_metadata = os.stat(filename, dir_fd=parent_fd, follow_symlinks=False) + if ( + not stat.S_ISREG(path_metadata.st_mode) + or (final_metadata.st_dev, final_metadata.st_ino) + != (path_metadata.st_dev, path_metadata.st_ino) + ): + raise ValueError("output file changed during secure materialization") + if final_metadata.st_nlink != 1 or path_metadata.st_nlink != 1: raise ValueError( - f"current-head npm lock {lock_path} package {package_path} must use one SHA-512 integrity value" + "generated output files must remain singly linked regular files" ) + os.fsync(parent_fd) + except BaseException: + _unlink_owned_file(parent_fd, filename, identity) + raise + finally: + os.close(file_fd) + + +def _write_relative_file( + project_fd: int, + relative_path: str, + content: bytes, +) -> None: + """Write one validated project-relative input through pinned directories.""" + parts = _safe_relative_parts(relative_path) + parent_fd = _open_relative_directory(project_fd, tuple(parts[:-1])) + try: + _write_new_file(parent_fd, parts[-1], content) + finally: + os.close(parent_fd) def materialize( @@ -347,81 +703,90 @@ def materialize( head_sha: str | None = None, ) -> list[dict[str, str]]: """Write trusted base and bounded HEAD inputs under Docker-context-safe paths.""" - if output_dir.exists() and output_dir.is_symlink(): - raise ValueError("output directory must not be a symlink") - output_dir.mkdir(parents=True, exist_ok=True) - - manifest: list[dict[str, str]] = [] - projects: list[tuple[str, str, dict[str, bytes], str, str]] = [] - base_npm = base_npm_projects(repo_root, base_sha) - base_npm_paths = {source_path for source_path, _manager, _inputs in base_npm} - base_npm_blobs: dict[str, str] = {} - for source_path, package_manager, base_inputs in ( - base_pnpm_projects(repo_root, base_sha) + base_npm - ): - lock_blob = _lock_blob_sha(repo_root, base_sha, source_path) - projects.append( - ( - source_path, - package_manager, - base_inputs, - base_sha.lower(), - lock_blob, - ) - ) - if source_path in base_npm_paths: - base_npm_blobs[source_path] = lock_blob - - if head_sha is not None: - if not SHA_RE.fullmatch(head_sha): - raise ValueError("head SHA must be exactly 40 hexadecimal characters") - for source_path, package_manager, head_inputs in base_npm_projects( - repo_root, head_sha + _require_descriptor_relative_capabilities() + output_fd, output_identity = _open_output_directory(output_dir) + try: + manifest: list[dict[str, str]] = [] + projects: list[tuple[str, str, dict[str, bytes], str, str]] = [] + base_npm = base_npm_projects(repo_root, base_sha) + base_npm_paths = {source_path for source_path, _manager, _inputs in base_npm} + base_npm_blobs: dict[str, str] = {} + for source_path, package_manager, base_inputs in ( + base_pnpm_projects(repo_root, base_sha) + base_npm ): - head_blob = _lock_blob_sha(repo_root, head_sha, source_path) - if base_npm_blobs.get(source_path) == head_blob: - continue - lock_name = pathlib.PurePosixPath(source_path).name - validate_head_npm_lock(source_path, head_inputs[lock_name]) + lock_blob = _lock_blob_sha(repo_root, base_sha, source_path) projects.append( ( source_path, package_manager, - head_inputs, - head_sha.lower(), - head_blob, + base_inputs, + base_sha.lower(), + lock_blob, ) ) + if source_path in base_npm_paths: + base_npm_blobs[source_path] = lock_blob + + if head_sha is not None: + if not SHA_RE.fullmatch(head_sha): + raise ValueError("head SHA must be exactly 40 hexadecimal characters") + for source_path, package_manager, head_inputs in base_npm_projects( + repo_root, head_sha + ): + head_blob = _lock_blob_sha(repo_root, head_sha, source_path) + if base_npm_blobs.get(source_path) == head_blob: + continue + lock_name = pathlib.PurePosixPath(source_path).name + validate_head_npm_lock(source_path, head_inputs[lock_name]) + projects.append( + ( + source_path, + package_manager, + head_inputs, + head_sha.lower(), + head_blob, + ) + ) - for index, ( - source_path, - package_manager, - base_inputs, - revision_sha, - lock_blob, - ) in enumerate(sorted(projects, key=lambda project: (project[0], project[3]))): - directory = f"project-{index:03d}" - project_dir = output_dir / directory - project_dir.mkdir() - for relative_path, content in sorted(base_inputs.items()): - destination = project_dir / relative_path - destination.parent.mkdir(parents=True, exist_ok=True) - destination.write_bytes(content) - manifest.append( - { - "directory": directory, - "lock_blob": lock_blob, - "package_manager": package_manager, - "revision_sha": revision_sha, - "source": source_path, - } - ) + for index, ( + source_path, + package_manager, + base_inputs, + revision_sha, + lock_blob, + ) in enumerate(sorted(projects, key=lambda project: (project[0], project[3]))): + directory = f"project-{index:03d}" + project_fd = _create_project_directory(output_fd, directory) + project_metadata = os.fstat(project_fd) + project_identity = (project_metadata.st_dev, project_metadata.st_ino) + try: + for relative_path, content in sorted(base_inputs.items()): + _write_relative_file(project_fd, relative_path, content) + os.fsync(project_fd) + except BaseException: + _remove_owned_empty_directory(output_fd, directory, project_identity) + raise + finally: + os.close(project_fd) + manifest.append( + { + "directory": directory, + "lock_blob": lock_blob, + "package_manager": package_manager, + "revision_sha": revision_sha, + "source": source_path, + } + ) - (output_dir / "manifest.json").write_text( - json.dumps(manifest, indent=2, sort_keys=True) + "\n", - encoding="utf-8", - ) - return manifest + manifest_content = ( + json.dumps(manifest, indent=2, sort_keys=True) + "\n" + ).encode("utf-8") + _write_new_file(output_fd, "manifest.json", manifest_content) + os.fsync(output_fd) + _verify_output_directory_binding(output_dir, output_fd, output_identity) + return manifest + finally: + os.close(output_fd) def main(argv: list[str] | None = None) -> int: diff --git a/tests/test_javascript_materializer_creation_cleanup_security.py b/tests/test_javascript_materializer_creation_cleanup_security.py new file mode 100644 index 000000000..09d1043b1 --- /dev/null +++ b/tests/test_javascript_materializer_creation_cleanup_security.py @@ -0,0 +1,149 @@ +"""Adversarial creation and rollback contracts for JavaScript lock materialization.""" + +from __future__ import annotations + +import os +from pathlib import Path +import pathlib + +import pytest + +from scripts.ci import materialize_base_javascript_packages as materializer + + +_BASE_SHA = "a" * 40 +_LOCK_BLOB_SHA = "b" * 40 + + +def _stub_projects( + monkeypatch: pytest.MonkeyPatch, + inputs: dict[str, bytes] | None = None, +) -> None: + """Replace Git discovery with one bounded npm project or an empty queue.""" + projects = [] + if inputs is not None: + projects = [("package-lock.json", "npm", inputs)] + monkeypatch.setattr( + materializer, + "base_npm_projects", + lambda *_args: projects, + ) + monkeypatch.setattr(materializer, "base_pnpm_projects", lambda *_args: []) + monkeypatch.setattr(materializer, "_lock_blob_sha", lambda *_args: _LOCK_BLOB_SHA) + + +def test_forwarding_open_instrumentation_does_not_change_platform_capability( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Capability checks use immutable CPython callables, not test wrappers.""" + output_directory = tmp_path / "generated_locks" + _stub_projects(monkeypatch) + real_open = os.open + + def forwarding_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", forwarding_open) + + manifest = materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert manifest == [] + assert (output_directory / "manifest.json").read_text(encoding="utf-8") == "[]\n" + + +def test_materializer_rejects_missing_follow_symlink_capability_before_mutation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """No-follow stat support is mandatory before any output path is created.""" + output_directory = tmp_path / "generated_locks" + _stub_projects(monkeypatch) + monkeypatch.setattr(os, "supports_follow_symlinks", set()) + + with pytest.raises(ValueError, match="descriptor-relative.*unavailable"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert not output_directory.exists() + + +def test_missing_ancestor_swap_never_creates_output_through_attacker_symlink( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pathname creation cannot be redirected while an ancestor is replaced.""" + trusted_root = tmp_path / "trusted_root" + trusted_root.mkdir() + pinned_root = tmp_path / "pinned_root" + attacker_root = tmp_path / "attacker_root" + attacker_parent = attacker_root / "missing_parent" + attacker_parent.mkdir(parents=True) + output_directory = trusted_root / "missing_parent" / "generated_locks" + attacker_output = attacker_parent / "generated_locks" + _stub_projects(monkeypatch) + + real_mkdir = pathlib.Path.mkdir + swapped = False + + def swap_after_parent_creation( + path: pathlib.Path, + *args: object, + **kwargs: object, + ) -> None: + nonlocal swapped + real_mkdir(path, *args, **kwargs) + if not swapped and path == output_directory.parent.absolute(): + trusted_root.rename(pinned_root) + trusted_root.symlink_to(attacker_root, target_is_directory=True) + swapped = True + + monkeypatch.setattr(pathlib.Path, "mkdir", swap_after_parent_creation) + + with pytest.raises(ValueError, match="ancestor|symlink|changed"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert swapped is True + assert not attacker_output.exists() + + +def test_late_write_failure_rolls_back_every_owned_file_and_directory( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Rollback removes all earlier generated entries while preserving operator data.""" + output_directory = tmp_path / "generated_locks" + output_directory.mkdir() + operator_note = output_directory / "operator-note.txt" + operator_note.write_text("preserve\n", encoding="utf-8") + _stub_projects( + monkeypatch, + { + "a-first.json": b"first\n", + "b-second.json": b"second\n", + }, + ) + real_write = os.write + write_calls = 0 + + def fail_second_file_write(file_descriptor: int, content: object) -> int: + nonlocal write_calls + write_calls += 1 + if write_calls == 2: + return 0 + return real_write(file_descriptor, content) + + monkeypatch.setattr(os, "write", fail_second_file_write) + + with pytest.raises(OSError, match="made no progress"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert write_calls == 2 + assert operator_note.read_text(encoding="utf-8") == "preserve\n" + assert sorted(path.name for path in output_directory.iterdir()) == [ + "operator-note.txt" + ] diff --git a/tests/test_javascript_materializer_descriptor_ancestry.py b/tests/test_javascript_materializer_descriptor_ancestry.py new file mode 100644 index 000000000..3778a6e09 --- /dev/null +++ b/tests/test_javascript_materializer_descriptor_ancestry.py @@ -0,0 +1,189 @@ +"""Adversarial contracts for descriptor-anchored materializer ancestry and cleanup.""" + +from __future__ import annotations + +import os +from pathlib import Path +import stat + +import pytest + +from scripts.ci import materialize_base_javascript_packages as materializer + + +_BASE_SHA = "a" * 40 +_LOCK_BLOB_SHA = "b" * 40 + + +def _projects(relative_path: str = "package-lock.json") -> list[tuple[str, str, dict[str, bytes]]]: + """Return one deterministic npm project with one optionally nested lock input.""" + return [ + ( + "package-lock.json", + "npm", + { + "package.json": b'{"name":"fixture"}\n', + relative_path: b'{"lockfileVersion":3,"packages":{}}\n', + }, + ) + ] + + +def _stub_project_discovery( + monkeypatch: pytest.MonkeyPatch, + projects: list[tuple[str, str, dict[str, bytes]]] | None = None, +) -> None: + """Replace Git-backed project discovery with bounded in-memory fixtures.""" + monkeypatch.setattr( + materializer, + "base_npm_projects", + lambda *_args: _projects() if projects is None else projects, + ) + monkeypatch.setattr(materializer, "base_pnpm_projects", lambda *_args: []) + monkeypatch.setattr(materializer, "_lock_blob_sha", lambda *_args: _LOCK_BLOB_SHA) + + +def test_materializer_rejects_intermediate_ancestor_swap_before_parent_open( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An intermediate ancestor swap cannot redirect the initially opened output tree.""" + trusted_root = tmp_path / "trusted_root" + trusted_parent = trusted_root / "nested_parent" + output_directory = trusted_parent / "generated_locks" + trusted_parent.mkdir(parents=True) + + pinned_root = tmp_path / "pinned_root" + attacker_root = tmp_path / "attacker_root" + attacker_output = attacker_root / "nested_parent" / "generated_locks" + attacker_output.mkdir(parents=True) + _stub_project_discovery(monkeypatch) + + real_open = os.open + swapped = False + + def swap_intermediate_ancestor( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + nonlocal swapped + if ( + not swapped + and Path(path) == trusted_parent.absolute() + and kwargs.get("dir_fd") is None + ): + trusted_root.rename(pinned_root) + trusted_root.symlink_to(attacker_root, target_is_directory=True) + swapped = True + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", swap_intermediate_ancestor) + + with pytest.raises(ValueError, match="ancestor|symlink|changed"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert swapped is True + assert list(attacker_output.iterdir()) == [] + assert list((pinned_root / "nested_parent" / "generated_locks").iterdir()) == [] + + +def test_materializer_rejects_new_nested_directory_replacement_before_open( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A directory created beneath a held descriptor must retain its original inode.""" + output_directory = tmp_path / "generated_locks" + saved_directory = tmp_path / "saved_nested_directory" + _stub_project_discovery(monkeypatch, _projects("nested_directory/package-lock.json")) + + real_open = os.open + swapped = False + + def swap_nested_directory( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + nonlocal swapped + if ( + not swapped + and path == "nested_directory" + and kwargs.get("dir_fd") is not None + ): + nested_directory = output_directory / "project-000" / "nested_directory" + nested_directory.rename(saved_directory) + nested_directory.mkdir() + swapped = True + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", swap_nested_directory) + + with pytest.raises(ValueError, match="directory.*changed|binding|inode"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert swapped is True + assert list(saved_directory.iterdir()) == [] + replacement = output_directory / "project-000" / "nested_directory" + assert not (replacement / "package-lock.json").exists() + + +def test_materializer_fsyncs_files_and_every_published_directory( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Durable evidence requires file bytes and directory entries to be synchronized.""" + output_directory = tmp_path / "generated_locks" + _stub_project_discovery(monkeypatch, _projects("nested_directory/package-lock.json")) + real_fsync = os.fsync + synchronized_modes: list[int] = [] + + def track_fsync(file_descriptor: int) -> None: + synchronized_modes.append(stat.S_IFMT(os.fstat(file_descriptor).st_mode)) + real_fsync(file_descriptor) + + monkeypatch.setattr(os, "fsync", track_fsync) + + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert stat.S_IFREG in synchronized_modes + assert stat.S_IFDIR in synchronized_modes + assert synchronized_modes.count(stat.S_IFDIR) >= 3 + + +def test_materializer_fails_closed_without_descriptor_relative_capabilities( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Unsupported runtimes must fail before creating any output path.""" + output_directory = tmp_path / "generated_locks" + _stub_project_discovery(monkeypatch, []) + monkeypatch.setattr(os, "supports_dir_fd", set()) + + with pytest.raises(ValueError, match="descriptor-relative.*unavailable"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert not output_directory.exists() + + +def test_failed_write_removes_only_owned_outputs_and_preserves_existing_entries( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Failure cleanup removes partial generated evidence without deleting prior content.""" + output_directory = tmp_path / "generated_locks" + output_directory.mkdir() + existing_file = output_directory / "operator-note.txt" + existing_file.write_text("preserve\n", encoding="utf-8") + _stub_project_discovery(monkeypatch) + monkeypatch.setattr(os, "write", lambda *_args: 0) + + with pytest.raises(OSError, match="made no progress"): + materializer.materialize(tmp_path, _BASE_SHA, output_directory) + + assert existing_file.read_text(encoding="utf-8") == "preserve\n" + assert sorted(path.name for path in output_directory.iterdir()) == [ + "operator-note.txt" + ] diff --git a/tests/test_javascript_materializer_existing_directory_coverage.py b/tests/test_javascript_materializer_existing_directory_coverage.py new file mode 100644 index 000000000..71e8df5c1 --- /dev/null +++ b/tests/test_javascript_materializer_existing_directory_coverage.py @@ -0,0 +1,25 @@ +"""Existing-directory branch coverage for the JavaScript lock materializer.""" + +from __future__ import annotations + +import os +from pathlib import Path + +from scripts.ci import materialize_base_javascript_packages as materializer + + +def test_relative_directory_reuses_existing_directory(tmp_path: Path) -> None: + """A pre-existing nested directory is opened without the creation-only sync path.""" + + nested_directory = tmp_path / "nested_directory" + nested_directory.mkdir() + root_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + nested_fd = materializer._open_relative_directory(root_fd, (nested_directory.name,)) + try: + assert os.path.samestat( + os.fstat(nested_fd), + os.stat(nested_directory, follow_symlinks=False), + ) + finally: + os.close(nested_fd) + os.close(root_fd) diff --git a/tests/test_javascript_materializer_output_edge_coverage.py b/tests/test_javascript_materializer_output_edge_coverage.py new file mode 100644 index 000000000..e7a009bc9 --- /dev/null +++ b/tests/test_javascript_materializer_output_edge_coverage.py @@ -0,0 +1,345 @@ +"""Branch-complete edge contracts for JavaScript materializer output hardening.""" + +from __future__ import annotations + +import errno +import os +from pathlib import Path + +import pytest + +from scripts.ci import materialize_base_javascript_packages as materializer + + +def _different_inode(metadata: os.stat_result) -> os.stat_result: + """Return metadata with the inode changed while retaining all other fields.""" + + values = list(metadata) + values[1] = metadata.st_ino + 1 + return os.stat_result(values) + + +def test_capability_gate_rejects_missing_no_follow_flag( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Secure output publication fails when a required open flag is unavailable.""" + + monkeypatch.delattr(materializer.os, "O_NOFOLLOW") + + with pytest.raises(ValueError, match="descriptor-relative output operations"): + materializer._require_descriptor_relative_capabilities() + + +def test_component_scan_rejects_existing_regular_file(tmp_path: Path) -> None: + """A regular file cannot become an intermediate output-directory component.""" + + blocking_file = tmp_path / "blocking_file" + blocking_file.write_bytes(b"not a directory") + + with pytest.raises(ValueError, match="path component must be a directory"): + materializer._reject_symlinked_output_components( + blocking_file / "generated_locks" + ) + + +def test_directory_identity_rejects_non_directory_metadata(tmp_path: Path) -> None: + """Directory identities reject regular-file metadata before inode comparison.""" + + regular_file = tmp_path / "regular_file" + regular_file.write_bytes(b"content") + + with pytest.raises(ValueError, match="binding changed"): + materializer._directory_identity(regular_file.stat()) + + +def test_output_open_detects_parent_descriptor_replacement( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The opened parent descriptor must retain the pre-open parent identity.""" + + output_directory = tmp_path / "generated_locks" + expected_parent = os.fspath(output_directory.parent) + real_open = materializer.os.open + real_fstat = materializer.os.fstat + parent_descriptors: list[int] = [] + + def capture_parent_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + descriptor = real_open(path, flags, *args, **kwargs) + if os.fspath(path) == expected_parent and kwargs.get("dir_fd") is None: + parent_descriptors.append(descriptor) + return descriptor + + def replace_parent_identity(descriptor: int) -> os.stat_result: + metadata = real_fstat(descriptor) + if descriptor in parent_descriptors: + return _different_inode(metadata) + return metadata + + monkeypatch.setattr(materializer.os, "open", capture_parent_open) + monkeypatch.setattr(materializer.os, "fstat", replace_parent_identity) + + with pytest.raises(ValueError, match="ancestor changed"): + materializer._open_output_directory(output_directory) + + assert len(parent_descriptors) == 1 + with pytest.raises(OSError) as raised: + os.fstat(parent_descriptors[0]) + assert raised.value.errno == errno.EBADF + + +def test_output_open_detects_output_descriptor_replacement_and_closes_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The opened output descriptor is closed when its inode mismatches the path.""" + + output_directory = tmp_path / "generated_locks" + real_open = materializer.os.open + real_fstat = materializer.os.fstat + output_descriptors: list[int] = [] + + def capture_output_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + descriptor = real_open(path, flags, *args, **kwargs) + if path == output_directory.name and kwargs.get("dir_fd") is not None: + output_descriptors.append(descriptor) + return descriptor + + def replace_output_identity(descriptor: int) -> os.stat_result: + metadata = real_fstat(descriptor) + if descriptor in output_descriptors: + return _different_inode(metadata) + return metadata + + monkeypatch.setattr(materializer.os, "open", capture_output_open) + monkeypatch.setattr(materializer.os, "fstat", replace_output_identity) + + with pytest.raises(ValueError, match="output directory changed"): + materializer._open_output_directory(output_directory) + + assert len(output_descriptors) == 1 + with pytest.raises(OSError) as raised: + os.fstat(output_descriptors[0]) + assert raised.value.errno == errno.EBADF + + +def test_relative_directory_creation_synchronizes_new_directory(tmp_path: Path) -> None: + """A newly created nested directory returns a live pinned descriptor.""" + + root_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + nested_fd = materializer._open_relative_directory(root_fd, ("nested_directory",)) + try: + assert (tmp_path / "nested_directory").is_dir() + assert os.path.samestat( + os.fstat(nested_fd), + os.stat(tmp_path / "nested_directory", follow_symlinks=False), + ) + finally: + os.close(nested_fd) + os.close(root_fd) + + +def test_relative_directory_detects_descriptor_replacement_and_closes_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A child descriptor is closed when it differs from the pre-open child inode.""" + + root_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + real_open = materializer.os.open + real_fstat = materializer.os.fstat + child_descriptors: list[int] = [] + + def capture_child_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + descriptor = real_open(path, flags, *args, **kwargs) + if path == "nested_directory" and kwargs.get("dir_fd") is not None: + child_descriptors.append(descriptor) + return descriptor + + def replace_child_identity(descriptor: int) -> os.stat_result: + metadata = real_fstat(descriptor) + if descriptor in child_descriptors: + return _different_inode(metadata) + return metadata + + monkeypatch.setattr(materializer.os, "open", capture_child_open) + monkeypatch.setattr(materializer.os, "fstat", replace_child_identity) + try: + with pytest.raises(ValueError, match="binding changed"): + materializer._open_relative_directory(root_fd, ("nested_directory",)) + finally: + os.close(root_fd) + + assert len(child_descriptors) == 1 + with pytest.raises(OSError) as raised: + os.fstat(child_descriptors[0]) + assert raised.value.errno == errno.EBADF + + +def test_project_directory_detects_descriptor_replacement_and_closes_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A fresh project descriptor is closed when its inode fails revalidation.""" + + output_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + real_open = materializer.os.open + real_fstat = materializer.os.fstat + project_descriptors: list[int] = [] + + def capture_project_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + descriptor = real_open(path, flags, *args, **kwargs) + if path == "project-000" and kwargs.get("dir_fd") == output_fd: + project_descriptors.append(descriptor) + return descriptor + + def replace_project_identity(descriptor: int) -> os.stat_result: + metadata = real_fstat(descriptor) + if descriptor in project_descriptors: + return _different_inode(metadata) + return metadata + + monkeypatch.setattr(materializer.os, "open", capture_project_open) + monkeypatch.setattr(materializer.os, "fstat", replace_project_identity) + try: + with pytest.raises(ValueError, match="binding changed"): + materializer._create_project_directory(output_fd, "project-000") + finally: + os.close(output_fd) + + assert len(project_descriptors) == 1 + with pytest.raises(OSError) as raised: + os.fstat(project_descriptors[0]) + assert raised.value.errno == errno.EBADF + + +def test_unlink_owned_file_ignores_missing_name(tmp_path: Path) -> None: + """Cleanup is a no-op when the generated filename no longer exists.""" + + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + materializer._unlink_owned_file(parent_fd, "missing_file", (1, 1)) + finally: + os.close(parent_fd) + + +def test_unlink_owned_file_ignores_replaced_identity(tmp_path: Path) -> None: + """Cleanup never unlinks a path that no longer names the generated inode.""" + + destination = tmp_path / "generated_file" + destination.write_bytes(b"replacement") + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + materializer._unlink_owned_file(parent_fd, destination.name, (1, 1)) + finally: + os.close(parent_fd) + assert destination.read_bytes() == b"replacement" + + +def test_unlink_owned_file_ignores_unlink_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Cleanup remains fail-safe when the owned filename cannot be unlinked.""" + + destination = tmp_path / "generated_file" + destination.write_bytes(b"content") + metadata = destination.stat() + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + + def deny_unlink(*_args: object, **_kwargs: object) -> None: + raise PermissionError(errno.EACCES, "synthetic") + + monkeypatch.setattr(materializer.os, "unlink", deny_unlink) + try: + materializer._unlink_owned_file( + parent_fd, + destination.name, + (metadata.st_dev, metadata.st_ino), + ) + finally: + os.close(parent_fd) + assert destination.read_bytes() == b"content" + + +def test_remove_owned_directory_ignores_missing_name(tmp_path: Path) -> None: + """Directory cleanup is a no-op when the generated directory disappeared.""" + + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + materializer._remove_owned_empty_directory( + parent_fd, + "missing_directory", + (1, 1), + ) + finally: + os.close(parent_fd) + + +def test_remove_owned_directory_ignores_regular_file(tmp_path: Path) -> None: + """Directory cleanup never removes a regular file at the generated name.""" + + destination = tmp_path / "project-000" + destination.write_bytes(b"content") + metadata = destination.stat() + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + materializer._remove_owned_empty_directory( + parent_fd, + destination.name, + (metadata.st_dev, metadata.st_ino), + ) + finally: + os.close(parent_fd) + assert destination.read_bytes() == b"content" + + +def test_remove_owned_directory_ignores_replaced_identity(tmp_path: Path) -> None: + """Directory cleanup preserves a directory whose inode no longer matches.""" + + destination = tmp_path / "project-000" + destination.mkdir() + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + materializer._remove_owned_empty_directory( + parent_fd, + destination.name, + (1, 1), + ) + finally: + os.close(parent_fd) + assert destination.is_dir() + + +def test_remove_owned_directory_ignores_rmdir_failure(tmp_path: Path) -> None: + """Nonempty owned directories remain available for forensic inspection.""" + + destination = tmp_path / "project-000" + destination.mkdir() + (destination / "retained_file").write_bytes(b"content") + metadata = destination.stat() + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + materializer._remove_owned_empty_directory( + parent_fd, + destination.name, + (metadata.st_dev, metadata.st_ino), + ) + finally: + os.close(parent_fd) + assert (destination / "retained_file").read_bytes() == b"content" diff --git a/tests/test_javascript_materializer_output_security.py b/tests/test_javascript_materializer_output_security.py new file mode 100644 index 000000000..06e70809c --- /dev/null +++ b/tests/test_javascript_materializer_output_security.py @@ -0,0 +1,387 @@ +"""Security regressions for descriptor-pinned JavaScript lock materialization.""" + +from __future__ import annotations + +import errno +import os +from pathlib import Path + +import pytest + +from scripts.ci import materialize_base_javascript_packages as materializer + + +def _one_project(relative_path: str = "package-lock.json") -> list[tuple[str, str, dict[str, bytes]]]: + """Return one deterministic trusted npm project fixture.""" + + return [ + ( + "package-lock.json", + "npm", + { + "package.json": b'{"name":"fixture"}\n', + relative_path: b'{"lockfileVersion":3,"packages":{}}\n', + }, + ) + ] + + +def _stub_project_discovery( + monkeypatch: pytest.MonkeyPatch, + projects: list[tuple[str, str, dict[str, bytes]]] | None = None, +) -> None: + """Replace Git-backed discovery with one bounded in-memory project.""" + + monkeypatch.setattr( + materializer, + "base_npm_projects", + lambda *_args: _one_project() if projects is None else projects, + ) + monkeypatch.setattr(materializer, "base_pnpm_projects", lambda *_args: []) + monkeypatch.setattr(materializer, "_lock_blob_sha", lambda *_args: "b" * 40) + + +def test_materializer_rejects_symlinked_output_parent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """No intermediate symlink may redirect descriptor-relative output creation.""" + + target_directory = tmp_path / "target_directory" + target_directory.mkdir() + linked_parent = tmp_path / "linked_parent" + linked_parent.symlink_to(target_directory, target_is_directory=True) + _stub_project_discovery(monkeypatch, []) + + with pytest.raises(ValueError, match="must not contain symlinks"): + materializer.materialize( + tmp_path, + "a" * 40, + linked_parent / "generated_locks", + ) + + assert list(target_directory.iterdir()) == [] + + +def test_materializer_fails_closed_when_output_binding_is_replaced( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Replacing the published pathname cannot receive trusted lock inputs.""" + + output_directory = tmp_path / "generated_locks" + pinned_directory = tmp_path / "pinned_locks" + replacement_directory = tmp_path / "replacement_locks" + + def replace_output_before_return( + *_args: object, + ) -> list[tuple[str, str, dict[str, bytes]]]: + output_directory.rename(pinned_directory) + replacement_directory.mkdir() + replacement_directory.rename(output_directory) + return _one_project() + + monkeypatch.setattr( + materializer, + "base_npm_projects", + replace_output_before_return, + ) + monkeypatch.setattr(materializer, "base_pnpm_projects", lambda *_args: []) + monkeypatch.setattr(materializer, "_lock_blob_sha", lambda *_args: "b" * 40) + + with pytest.raises(ValueError, match="changed during secure materialization"): + materializer.materialize(tmp_path, "a" * 40, output_directory) + + assert (pinned_directory / "project-000" / "package-lock.json").is_file() + assert list(output_directory.iterdir()) == [] + + +def test_materializer_anchors_writes_when_output_path_becomes_symlink( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A post-open output symlink cannot redirect the first generated file.""" + + output_directory = tmp_path / "generated_locks" + pinned_directory = tmp_path / "pinned_locks" + attacker_directory = tmp_path / "attacker_directory" + attacker_directory.mkdir() + _stub_project_discovery(monkeypatch) + real_open = os.open + attacked = False + + def swap_before_first_file_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + nonlocal attacked + if not attacked and path == "package-lock.json" and flags & os.O_CREAT: + attacked = True + output_directory.rename(pinned_directory) + output_directory.symlink_to(attacker_directory, target_is_directory=True) + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", swap_before_first_file_open) + + with pytest.raises(ValueError, match="changed during secure materialization"): + materializer.materialize(tmp_path, "a" * 40, output_directory) + + assert attacked is True + assert ( + pinned_directory / "project-000" / "package-lock.json" + ).read_bytes() == _one_project()[0][2]["package-lock.json"] + assert list(attacker_directory.iterdir()) == [] + + +@pytest.mark.parametrize("relative_path", ["", "../escape", "/absolute", "nested\\escape"]) +def test_materializer_rejects_unsafe_relative_input_paths( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + relative_path: str, +) -> None: + """Trusted inputs still require one lexical relative POSIX output path.""" + + _stub_project_discovery(monkeypatch, _one_project(relative_path)) + + with pytest.raises(ValueError, match="unsafe relative output path"): + materializer.materialize( + tmp_path, + "a" * 40, + tmp_path / "generated_locks", + ) + + assert not (tmp_path / "escape").exists() + + +def test_materializer_rejects_preexisting_generated_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A pre-existing generated name cannot be truncated or reinterpreted.""" + + output_directory = tmp_path / "generated_locks" + project_directory = output_directory / "project-000" + project_directory.mkdir(parents=True) + destination = project_directory / "package-lock.json" + destination.write_bytes(b"unchanged") + _stub_project_discovery(monkeypatch) + + with pytest.raises(ValueError, match="must not pre-exist"): + materializer.materialize(tmp_path, "a" * 40, output_directory) + + assert destination.read_bytes() == b"unchanged" + + +def test_materializer_detects_hard_link_added_during_pinned_write( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A hard link added after file creation must fail before success evidence.""" + + output_directory = tmp_path / "generated_locks" + outside_link = tmp_path / "captured_output" + _stub_project_discovery(monkeypatch) + real_fsync = os.fsync + linked = False + + def link_after_file_sync(file_descriptor: int) -> None: + nonlocal linked + real_fsync(file_descriptor) + destination = output_directory / "project-000" / "package-lock.json" + if linked or not destination.exists(): + return + descriptor_metadata = os.fstat(file_descriptor) + path_metadata = os.stat(destination, follow_symlinks=False) + if (descriptor_metadata.st_dev, descriptor_metadata.st_ino) != ( + path_metadata.st_dev, + path_metadata.st_ino, + ): + return + os.link(destination, outside_link) + linked = True + + monkeypatch.setattr(os, "fsync", link_after_file_sync) + + with pytest.raises(ValueError, match="singly linked regular files"): + materializer.materialize(tmp_path, "a" * 40, output_directory) + + assert linked is True + assert outside_link.read_bytes() == _one_project()[0][2]["package-lock.json"] + + +def test_materializer_detects_destination_swap_after_pinned_write( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A generated name swapped after open cannot become accepted evidence.""" + + output_directory = tmp_path / "generated_locks" + outside_file = tmp_path / "outside_file" + outside_file.write_bytes(b"unchanged") + _stub_project_discovery(monkeypatch) + real_fsync = os.fsync + swapped = False + + def swap_after_file_sync(file_descriptor: int) -> None: + nonlocal swapped + real_fsync(file_descriptor) + destination = output_directory / "project-000" / "package-lock.json" + if swapped or not destination.exists(): + return + swapped = True + destination.unlink() + destination.symlink_to(outside_file) + + monkeypatch.setattr(os, "fsync", swap_after_file_sync) + + with pytest.raises(ValueError, match="output file changed"): + materializer.materialize(tmp_path, "a" * 40, output_directory) + + assert outside_file.read_bytes() == b"unchanged" + + +def test_materializer_fails_when_descriptor_write_makes_no_progress( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A zero-length descriptor write is an error, not truncated success.""" + + _stub_project_discovery(monkeypatch) + monkeypatch.setattr(os, "write", lambda *_args: 0) + + with pytest.raises(OSError, match="made no progress"): + materializer.materialize( + tmp_path, + "a" * 40, + tmp_path / "generated_locks", + ) + + +def test_materializer_rejects_filesystem_root_output( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The filesystem root is never a generated-lock output directory.""" + + _stub_project_discovery(monkeypatch, []) + + with pytest.raises(ValueError, match="must not be the filesystem root"): + materializer.materialize(tmp_path, "a" * 40, Path("/")) + + +def test_materializer_preserves_bounded_directory_open_failures( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Descriptor-relative ENOTDIR remains fail-closed without a full-path leak.""" + + _stub_project_discovery(monkeypatch, []) + real_open = os.open + + def fail_output_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + if path == "generated_locks": + raise OSError(errno.ENOTDIR, "synthetic") + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", fail_output_open) + + with pytest.raises(NotADirectoryError, match="synthetic") as raised: + materializer.materialize( + tmp_path, + "a" * 40, + tmp_path / "generated_locks", + ) + assert raised.value.errno == errno.ENOTDIR + assert raised.value.filename is None + + +def test_output_binding_rejects_removed_published_path(tmp_path: Path) -> None: + """A removed output pathname cannot validate against its still-open descriptor.""" + + output_directory = tmp_path / "generated_locks" + output_directory.mkdir() + output_fd = os.open(output_directory, materializer._DIRECTORY_OPEN_FLAGS) + metadata = os.fstat(output_fd) + try: + output_directory.rmdir() + with pytest.raises(ValueError, match="changed during secure materialization"): + materializer._verify_output_directory_binding( + output_directory, + output_fd, + (metadata.st_dev, metadata.st_ino), + ) + finally: + os.close(output_fd) + + +def test_relative_directory_open_failure_closes_fail_closed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A child directory that cannot be opened propagates a bounded hard failure.""" + + root_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + real_open = os.open + + def fail_child_open( + path: object, + flags: int, + *args: object, + **kwargs: object, + ) -> int: + if path == "nested_directory": + raise OSError(errno.EACCES, "synthetic") + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", fail_child_open) + try: + with pytest.raises(OSError, match="synthetic"): + materializer._open_relative_directory(root_fd, ("nested_directory",)) + finally: + os.close(root_fd) + + +def test_project_directory_must_be_fresh(tmp_path: Path) -> None: + """A pre-existing numbered project directory is rejected before any file write.""" + + (tmp_path / "project-000").mkdir() + output_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + with pytest.raises(ValueError, match="must not pre-exist"): + materializer._create_project_directory(output_fd, "project-000") + finally: + os.close(output_fd) + + +def test_descriptor_file_must_be_fresh(tmp_path: Path) -> None: + """A pre-existing file name cannot be reopened through the descriptor helper.""" + + (tmp_path / "manifest.json").write_bytes(b"unchanged") + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + try: + with pytest.raises(ValueError, match="must not pre-exist"): + materializer._write_new_file(parent_fd, "manifest.json", b"replacement") + finally: + os.close(parent_fd) + assert (tmp_path / "manifest.json").read_bytes() == b"unchanged" + + +def test_new_file_rejects_non_single_link_initial_descriptor( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An unexpected initial link count fails before trusted bytes are written.""" + + parent_fd = os.open(tmp_path, materializer._DIRECTORY_OPEN_FLAGS) + real_fstat = os.fstat + + def force_multiple_links(file_descriptor: int) -> os.stat_result: + metadata = real_fstat(file_descriptor) + if file_descriptor == parent_fd: + return metadata + values = list(metadata) + values[3] = 2 + return os.stat_result(values) + + monkeypatch.setattr(os, "fstat", force_multiple_links) + try: + with pytest.raises(ValueError, match="singly linked regular files"): + materializer._write_new_file(parent_fd, "new-lock.json", b"trusted") + finally: + os.close(parent_fd) diff --git a/tests/test_npm_nested_metadata_lock_validation.py b/tests/test_npm_nested_metadata_lock_validation.py new file mode 100644 index 000000000..81367069b --- /dev/null +++ b/tests/test_npm_nested_metadata_lock_validation.py @@ -0,0 +1,272 @@ +"""Contracts for npm v2/v3 metadata-only nested package locations.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from scripts.ci import materialize_base_javascript_packages as materializer + + +_VALID_INTEGRITY = "sha512-" + ("A" * 86) + "==" + + +def _pinned(version: str, package_name: str) -> dict[str, str]: + """Return one exact public-registry package pin.""" + + archive_name = package_name.rsplit("/", 1)[-1] + return { + "version": version, + "resolved": ( + f"https://registry.npmjs.org/{package_name}/-/" + f"{archive_name}-{version}.tgz" + ), + "integrity": _VALID_INTEGRITY, + } + + +def _lock(packages: dict[str, object]) -> bytes: + """Serialize one npm lock fixture as UTF-8 JSON bytes.""" + + return json.dumps( + {"lockfileVersion": 3, "packages": packages}, + sort_keys=True, + ).encode("utf-8") + + +def test_accepts_bandscope_scoped_metadata_through_exact_root_pin() -> None: + """A BandScope-shaped peer location may reuse one exact canonical pin.""" + + packages = { + "": {"name": "bandscope"}, + "node_modules/@types/react-dom": _pinned("19.1.7", "@types/react-dom"), + "apps/desktop/node_modules/@types/react-dom": { + "version": "19.1.7", + "dev": True, + "peer": True, + }, + } + + materializer.validate_head_npm_lock("package-lock.json", _lock(packages)) + + +def test_accepts_unscoped_metadata_and_independently_pinned_nested_version() -> None: + """Metadata reuse and an independently complete nested pin can coexist.""" + + packages = { + "node_modules/react": _pinned("19.1.1", "react"), + "apps/web/node_modules/react": {"version": "19.1.1", "peer": True}, + "node_modules/legacy/node_modules/react": _pinned("18.3.1", "react"), + } + + materializer.validate_head_npm_lock("package-lock.json", _lock(packages)) + + +@pytest.mark.parametrize( + ("packages", "message"), + [ + ( + {"apps/web/node_modules/react": {"version": "19.1.1"}}, + "canonical root pin", + ), + ( + { + "node_modules/react": _pinned("19.1.1", "react"), + "apps/web/node_modules/react": {"version": "19.1.0"}, + }, + "exact canonical version", + ), + ( + { + "node_modules/react": { + "version": "19.1.1", + "resolved": _pinned("19.1.1", "react")["resolved"], + }, + "apps/web/node_modules/react": {"version": "19.1.1"}, + }, + "registry tarball and SHA-512 integrity", + ), + ( + { + "node_modules/react": _pinned("19.1.1", "react"), + "apps/web/node_modules/react": { + "version": "19.1.1", + "resolved": _pinned("19.1.1", "react")["resolved"], + }, + }, + "must not partially declare", + ), + ( + { + "node_modules/react": _pinned("19.1.1", "react"), + "apps/web/node_modules/react": { + "version": "19.1.1", + "integrity": _VALID_INTEGRITY, + }, + }, + "must not partially declare", + ), + ( + { + "node_modules/react": { + **_pinned("19.1.1", "react"), + "resolved": "https://example.invalid/react-19.1.1.tgz", + }, + "apps/web/node_modules/react": {"version": "19.1.1"}, + }, + "must resolve from https://registry.npmjs.org/", + ), + ( + { + "node_modules/react": { + **_pinned("19.1.1", "react"), + "integrity": "sha512-invalid", + }, + "apps/web/node_modules/react": {"version": "19.1.1"}, + }, + "must use one SHA-512 integrity value", + ), + ( + {"apps/web/node_modules/@types": {"version": "1.0.0"}}, + "malformed npm package identity", + ), + ( + {"apps/web/node_modules/@types/react/extra": {"version": "1.0.0"}}, + "malformed npm package identity", + ), + ( + { + "node_modules/react": { + "version": "19.1.1", + "dev": True, + } + }, + "canonical root pin", + ), + ( + { + "node_modules/react": _pinned("19.1.1", "react"), + "apps/web/node_modules/react": {"version": ""}, + }, + "nonempty exact version", + ), + ], +) +def test_rejects_untrusted_metadata_only_nested_locations( + packages: dict[str, object], + message: str, +) -> None: + """Every metadata-only location must close through one exact safe root pin.""" + + with pytest.raises(ValueError, match=message): + materializer.validate_head_npm_lock("package-lock.json", _lock(packages)) + + +def test_regular_base_path_filter_covers_every_rejection_branch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Tree parsing ignores trees, symlinks, absolute paths, and traversal paths.""" + + def git_stub(_repo_root: Path, *args: str) -> bytes: + assert args[:4] == ("ls-tree", "-r", "-z", "--full-tree") + return b"".join( + ( + b"040000 tree " + (b"0" * 40) + b"\tdirectory\0", + b"120000 blob " + (b"1" * 40) + b"\tsymlink\0", + b"100644 blob " + (b"2" * 40) + b"\t/absolute\0", + b"100644 blob " + (b"3" * 40) + b"\t../escape\0", + b"100644 blob " + (b"4" * 40) + b"\tpackage.json\0", + ) + ) + + monkeypatch.setattr(materializer, "_git", git_stub) + assert materializer._regular_base_paths(tmp_path, "a" * 40) == {"package.json"} + + +@pytest.mark.parametrize( + "lock_document", + [ + {"lockfileVersion": 3}, + { + "lockfileVersion": 3, + "packages": {"packages/missing": {"version": "1.0.0"}}, + }, + ], +) +def test_base_npm_materialization_covers_optional_workspace_metadata( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + lock_document: dict[str, object], +) -> None: + """Missing packages maps and absent workspace manifests stay non-fatal.""" + + monkeypatch.setattr( + materializer, + "_regular_base_paths", + lambda _repo_root, _base_sha: {"package.json", "package-lock.json"}, + ) + + def git_stub(_repo_root: Path, *args: str) -> bytes: + assert args[0] == "show" + target = args[1].split(":", 1)[1] + if target == "package.json": + return b'{"name":"fixture"}\n' + if target == "package-lock.json": + return json.dumps(lock_document).encode("utf-8") + raise AssertionError(target) + + monkeypatch.setattr(materializer, "_git", git_stub) + projects = materializer.base_npm_projects(tmp_path, "a" * 40) + assert len(projects) == 1 + assert set(projects[0][2]) == {"package.json", "package-lock.json"} + + +def test_registry_pin_rejects_non_string_metadata() -> None: + """Registry provenance fields must be exact strings before URL parsing.""" + + with pytest.raises(ValueError, match="must pin a registry tarball"): + materializer._validate_npm_registry_pin( + "package-lock.json", + "node_modules/react", + 123, + _VALID_INTEGRITY, + ) + + +def test_materialize_rejects_symlinked_output_parent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A parent symlink must never redirect materialized lockfile writes.""" + + trusted_parent = tmp_path / "trusted-parent" + redirected_parent = tmp_path / "redirected-parent" + trusted_parent.mkdir() + redirected_parent.mkdir() + symlink_parent = trusted_parent / "attacker-controlled" + symlink_parent.symlink_to(redirected_parent, target_is_directory=True) + output_dir = symlink_parent / "materialized-locks" + + monkeypatch.setattr(materializer, "base_npm_projects", lambda *_args: []) + monkeypatch.setattr(materializer, "base_pnpm_projects", lambda *_args: []) + + with pytest.raises(ValueError, match="symlink"): + materializer.materialize(tmp_path, "a" * 40, output_dir) + assert not (redirected_parent / "materialized-locks").exists() + + +def test_materialize_rejects_regular_file_output_parent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A regular-file path component must not be traversed as an output directory.""" + + regular_parent = tmp_path / "regular-parent" + regular_parent.write_text("not a directory\n", encoding="utf-8") + output_dir = regular_parent / "materialized-locks" + + monkeypatch.setattr(materializer, "base_npm_projects", lambda *_args: []) + monkeypatch.setattr(materializer, "base_pnpm_projects", lambda *_args: []) + + with pytest.raises(ValueError, match="path component must be a directory"): + materializer.materialize(tmp_path, "a" * 40, output_dir)