CM-73389: Stop generating a lockfile for an npm workspace member

cycode/cli/files_collector/sca/npm/restore_bun_dependencies.py

@@ -6,15 +6,19 @@
 import typer
 
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path
+from cycode.cli.files_collector.sca.npm.workspace import (
+    BUN_LOCK_FILE_NAME,
+    MANIFEST_FILE_NAME,
+    is_covered_workspace_member,
+)
 from cycode.cli.models import Document
-from cycode.cli.utils.path_utils import get_file_content
+from cycode.cli.utils.path_utils import get_file_content, get_scan_roots_from_context
 from cycode.cli.utils.shell_executor import shell
 from cycode.logger import get_logger
 
 logger = get_logger('Bun Restore Dependencies')
 
-BUN_MANIFEST_FILE_NAME = 'package.json'
-BUN_LOCK_FILE_NAME = 'bun.lock'
+BUN_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME
 
 # Only Bun >=1.2 produces the text-based `bun.lock` lockfile that we parse.
 # Older Bun versions emit a binary `bun.lockb`, which is not supported.
@@ -61,6 +65,9 @@ def is_project(self, document: Document) -> bool:
         if manifest_dir and (Path(manifest_dir) / BUN_LOCK_FILE_NAME).is_file():
             return True
 
+        if is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)):
+            return False
+
         return _indicates_bun(document.content)
 
     def _is_supported_bun_version(self) -> bool:

cycode/cli/files_collector/sca/npm/restore_deno_dependencies.py

@@ -4,14 +4,14 @@
 import typer
 
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path
+from cycode.cli.files_collector.sca.npm.workspace import DENO_LOCK_FILE_NAME
 from cycode.cli.models import Document
 from cycode.cli.utils.path_utils import get_file_content
 from cycode.logger import get_logger
 
 logger = get_logger('Deno Restore Dependencies')
 
 DENO_MANIFEST_FILE_NAMES = ('deno.json', 'deno.jsonc')
-DENO_LOCK_FILE_NAME = 'deno.lock'
 
 
 class RestoreDenoDependencies(BaseRestoreDependencies):

cycode/cli/files_collector/sca/npm/restore_npm_dependencies.py

@@ -3,15 +3,25 @@
 import typer
 
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies
+from cycode.cli.files_collector.sca.npm.workspace import (
+    BUN_LOCK_FILE_NAME,
+    DENO_LOCK_FILE_NAME,
+    MANIFEST_FILE_NAME,
+    NPM_LOCK_FILE_NAME,
+    NPM_SHRINKWRAP_FILE_NAME,
+    PNPM_LOCK_FILE_NAME,
+    YARN_LOCK_FILE_NAME,
+    is_covered_workspace_member,
+)
 from cycode.cli.models import Document
+from cycode.cli.utils.path_utils import get_scan_roots_from_context
 from cycode.logger import get_logger
 
 logger = get_logger('NPM Restore Dependencies')
 
-NPM_MANIFEST_FILE_NAME = 'package.json'
-NPM_LOCK_FILE_NAME = 'package-lock.json'
+NPM_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME
 # These lockfiles indicate another package manager owns the project — NPM should not run
-_ALTERNATIVE_LOCK_FILES = ('yarn.lock', 'pnpm-lock.yaml', 'deno.lock', 'bun.lock')
+_ALTERNATIVE_LOCK_FILES = (YARN_LOCK_FILE_NAME, PNPM_LOCK_FILE_NAME, DENO_LOCK_FILE_NAME, BUN_LOCK_FILE_NAME)
 
 
 class RestoreNpmDependencies(BaseRestoreDependencies):
@@ -24,6 +34,9 @@ def is_project(self, document: Document) -> bool:
         Yarn and pnpm projects are handled by their dedicated handlers, which run before
         this one in the handler list. This handler is the npm fallback.
 
+        A manifest is also declined when a lockfile further up already resolves it, whichever
+        package manager wrote that lockfile; see the workspace package for how that is decided.
+
         NOTE: this guard only excludes a project when an alternative lockfile is *physically
         present on disk*. It does not inspect the `packageManager`/`engines` signal in
         package.json. So a project that declares e.g. `packageManager: "bun@..."` (or pnpm)
@@ -37,16 +50,18 @@ def is_project(self, document: Document) -> bool:
             return False
 
         manifest_dir = self.get_manifest_dir(document)
-        if manifest_dir:
-            for lock_file in _ALTERNATIVE_LOCK_FILES:
-                if (Path(manifest_dir) / lock_file).is_file():
-                    logger.debug(
-                        'Skipping npm restore: alternative lockfile detected, %s',
-                        {'path': document.path, 'lockfile': lock_file},
-                    )
-                    return False
+        if not manifest_dir:
+            return True
+
+        for lock_file in _ALTERNATIVE_LOCK_FILES:
+            if (Path(manifest_dir) / lock_file).is_file():
+                logger.debug(
+                    'Skipping npm restore: alternative lockfile detected, %s',
+                    {'path': document.path, 'lockfile': lock_file},
+                )
+                return False
 
-        return True
+        return not is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx))
 
     def get_commands(self, manifest_file_path: str) -> list[list[str]]:
         return [
@@ -65,7 +80,11 @@ def get_lock_file_name(self) -> str:
         return NPM_LOCK_FILE_NAME
 
     def get_lock_file_names(self) -> list[str]:
-        return [NPM_LOCK_FILE_NAME]
+        return [NPM_LOCK_FILE_NAME, NPM_SHRINKWRAP_FILE_NAME]
+
+    def get_restored_lock_file_name(self, restore_file_path: str) -> str:
+        name = Path(restore_file_path).name
+        return name if name in self.get_lock_file_names() else self.get_lock_file_name()
 
     @staticmethod
     def prepare_manifest_file_path_for_command(manifest_file_path: str) -> str:

cycode/cli/files_collector/sca/npm/restore_pnpm_dependencies.py

@@ -5,14 +5,18 @@
 import typer
 
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path
+from cycode.cli.files_collector.sca.npm.workspace import (
+    MANIFEST_FILE_NAME,
+    PNPM_LOCK_FILE_NAME,
+    is_covered_workspace_member,
+)
 from cycode.cli.models import Document
-from cycode.cli.utils.path_utils import get_file_content
+from cycode.cli.utils.path_utils import get_file_content, get_scan_roots_from_context
 from cycode.logger import get_logger
 
 logger = get_logger('Pnpm Restore Dependencies')
 
-PNPM_MANIFEST_FILE_NAME = 'package.json'
-PNPM_LOCK_FILE_NAME = 'pnpm-lock.yaml'
+PNPM_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME
 
 
 def _indicates_pnpm(package_json_content: Optional[str]) -> bool:
@@ -44,6 +48,9 @@ def is_project(self, document: Document) -> bool:
         if manifest_dir and (Path(manifest_dir) / PNPM_LOCK_FILE_NAME).is_file():
             return True
 
+        if is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)):
+            return False
+
         return _indicates_pnpm(document.content)
 
     def try_restore_dependencies(self, document: Document) -> Optional[Document]:

cycode/cli/files_collector/sca/npm/restore_yarn_dependencies.py

@@ -5,14 +5,18 @@
 import typer
 
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path
+from cycode.cli.files_collector.sca.npm.workspace import (
+    MANIFEST_FILE_NAME,
+    YARN_LOCK_FILE_NAME,
+    is_covered_workspace_member,
+)
 from cycode.cli.models import Document
-from cycode.cli.utils.path_utils import get_file_content
+from cycode.cli.utils.path_utils import get_file_content, get_scan_roots_from_context
 from cycode.logger import get_logger
 
 logger = get_logger('Yarn Restore Dependencies')
 
-YARN_MANIFEST_FILE_NAME = 'package.json'
-YARN_LOCK_FILE_NAME = 'yarn.lock'
+YARN_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME
 
 
 def _indicates_yarn(package_json_content: Optional[str]) -> bool:
@@ -44,6 +48,9 @@ def is_project(self, document: Document) -> bool:
         if manifest_dir and (Path(manifest_dir) / YARN_LOCK_FILE_NAME).is_file():
             return True
 
+        if is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)):
+            return False
+
         return _indicates_yarn(document.content)
 
     def try_restore_dependencies(self, document: Document) -> Optional[Document]:

cycode/cli/files_collector/sca/npm/workspace/__init__.py

@@ -0,0 +1,49 @@
+"""Deciding whether a lockfile above a manifest already resolves that manifest's dependencies.
+
+The lockfile is the authority wherever it can name its members; the manifest's workspaces globs
+are a fallback for the formats that cannot. See resolvers.py for the per-format readers.
+"""
+
+from cycode.cli.files_collector.sca.npm.workspace import coverage as _coverage
+from cycode.cli.files_collector.sca.npm.workspace import globs as _globs
+from cycode.cli.files_collector.sca.npm.workspace import resolvers as _resolvers
+from cycode.cli.files_collector.sca.npm.workspace.coverage import (
+    WorkspaceCoverage,
+    find_covering_workspace,
+    is_covered_workspace_member,
+)
+from cycode.cli.files_collector.sca.npm.workspace.names import (
+    BUN_BINARY_LOCK_FILE_NAME,
+    BUN_LOCK_FILE_NAME,
+    DENO_LOCK_FILE_NAME,
+    MANIFEST_FILE_NAME,
+    NPM_LOCK_FILE_NAME,
+    NPM_SHRINKWRAP_FILE_NAME,
+    PNPM_LOCK_FILE_NAME,
+    YARN_LOCK_FILE_NAME,
+)
+from cycode.cli.files_collector.sca.npm.workspace.resolvers import MEMBER_RESOLVERS
+
+
+def clear_cache() -> None:
+    """Drop every memo, so one scan never inherits another scan's view of the filesystem."""
+    _resolvers.clear_cache()
+    _globs.clear_cache()
+    _coverage.clear_cache()
+
+
+__all__ = [
+    'BUN_BINARY_LOCK_FILE_NAME',
+    'BUN_LOCK_FILE_NAME',
+    'DENO_LOCK_FILE_NAME',
+    'MANIFEST_FILE_NAME',
+    'MEMBER_RESOLVERS',
+    'NPM_LOCK_FILE_NAME',
+    'NPM_SHRINKWRAP_FILE_NAME',
+    'PNPM_LOCK_FILE_NAME',
+    'YARN_LOCK_FILE_NAME',
+    'WorkspaceCoverage',
+    'clear_cache',
+    'find_covering_workspace',
+    'is_covered_workspace_member',
+]

cycode/cli/files_collector/sca/npm/workspace/coverage.py

@@ -0,0 +1,142 @@
+"""Walking up from a manifest to the lockfile that already resolves it."""
+
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, NamedTuple, Optional
+
+from cycode.cli.files_collector.sca.npm.workspace.files import logger, resolved_path
+from cycode.cli.files_collector.sca.npm.workspace.globs import declares_workspace_member
+from cycode.cli.files_collector.sca.npm.workspace.resolvers import MEMBER_RESOLVERS
+from cycode.cli.utils.path_utils import is_sub_path
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+_GIT_DIR_NAME = '.git'
+
+_reported_unscanned_roots: set = set()
+
+
+def clear_cache() -> None:
+    _reported_unscanned_roots.clear()
+
+
+class WorkspaceCoverage(NamedTuple):
+    package_manager: str
+    lock_file: Path
+
+
+def _scan_root_directories(scan_roots: tuple) -> list:
+    """The directory each scanned path stands for; scanning a file scans its directory."""
+    directories = []
+    for scan_root in scan_roots:
+        resolved = resolved_path(scan_root)
+        if os.path.isfile(resolved):
+            resolved = os.path.dirname(resolved)
+
+        if resolved:
+            directories.append(resolved)
+
+    return directories
+
+
+def _containing_scan_roots(manifest_dir: Path, scan_roots: tuple) -> list:
+    resolved_manifest_dir = resolved_path(manifest_dir)
+    return [
+        Path(directory)
+        for directory in _scan_root_directories(scan_roots)
+        if is_sub_path(directory, resolved_manifest_dir)
+    ]
+
+
+def _resolve_walk_boundary(manifest_dir: Path, scan_roots: tuple) -> Optional[Path]:
+    for root_dir in manifest_dir.parents:
+        if (root_dir / _GIT_DIR_NAME).exists():
+            return root_dir
+
+    containing = _containing_scan_roots(manifest_dir, scan_roots)
+    if containing:
+        return min(containing, key=lambda scan_root: len(scan_root.parts))
+
+    if scan_roots:
+        return manifest_dir
+
+    return None
+
+
+def _workspace_root_candidates(manifest_dir: Path, scan_roots: tuple) -> 'Iterator[Path]':
+    boundary = _resolve_walk_boundary(manifest_dir, scan_roots)
+    resolved_boundary = resolved_path(boundary) if boundary is not None else None
+    if resolved_boundary == resolved_path(manifest_dir):
+        return
+
+    for root_dir in manifest_dir.parents:
+        yield root_dir
+        if resolved_boundary is not None and resolved_path(root_dir) == resolved_boundary:
+            return
+
+
+def _find_covering_workspace(manifest_dir: Path, scan_roots: tuple) -> Optional[WorkspaceCoverage]:
+    for root_dir in _workspace_root_candidates(manifest_dir, scan_roots):
+        member_path = manifest_dir.relative_to(root_dir).as_posix()
+
+        for resolver in MEMBER_RESOLVERS:
+            for lock_file_name in resolver.lock_file_names:
+                lock_file = root_dir / lock_file_name
+                if not lock_file.is_file():
+                    continue
+
+                member_names = resolver.resolve(lock_file)
+                if member_names is not None:
+                    if member_path in member_names:
+                        return WorkspaceCoverage(resolver.package_manager, lock_file)
+                    continue
+
+                if resolver.may_use_workspace_globs and declares_workspace_member(root_dir, member_path):
+                    return WorkspaceCoverage(resolver.package_manager, lock_file)
+
+    return None
+
+
+def find_covering_workspace(manifest_dir: Optional[str], scan_roots: tuple = ()) -> Optional[WorkspaceCoverage]:
+    if not manifest_dir:
+        return None
+
+    return _find_covering_workspace(Path(manifest_dir), scan_roots)
+
+
+def _is_inside_scanned_paths(scan_roots: tuple, root_dir: Path) -> bool:
+    directories = _scan_root_directories(scan_roots)
+    if not directories:
+        logger.debug('No scanned paths in context; treating the workspace root as scanned, %s', {'root': str(root_dir)})
+        return True
+
+    resolved_root_dir = resolved_path(root_dir)
+    return any(is_sub_path(directory, resolved_root_dir) for directory in directories)
+
+
+def is_covered_workspace_member(manifest_dir: Optional[str], document_path: str, scan_roots: tuple = ()) -> bool:
+    coverage = find_covering_workspace(manifest_dir, scan_roots)
+    if coverage is None:
+        return False
+
+    details = {
+        'path': document_path,
+        'root_lockfile': str(coverage.lock_file),
+        'workspace': coverage.package_manager,
+    }
+    if _is_inside_scanned_paths(scan_roots, coverage.lock_file.parent):
+        logger.debug('Skipping restore: the workspace root lockfile already covers this member, %s', details)
+        return True
+
+    report_key = (document_path, str(coverage.lock_file))
+    if report_key not in _reported_unscanned_roots:
+        _reported_unscanned_roots.add(report_key)
+        logger.warning(
+            'The workspace root lockfile is outside the scanned path and will not be collected, '
+            'so this member is restored on its own. Scan the workspace root for the versions it '
+            'actually installs, %s',
+            details,
+        )
+
+    return False