Merge pull request #4079 from aboutcode-org/fix-npm-workspace-parsing

Fix pnpm workspace parsing and udpate package detection

Author: AyanSinhaMahapatra

src/packagedcode/npm.py

@@ -444,7 +444,8 @@ def get_workspace_members(cls, workspaces, codebase, workspace_root_path):
             # Case 3: This is a complex glob pattern, we are doing a full codebase walk
             # and glob matching each resource
             else:
-                for resource in workspace_root_path:
+                workspace_root = codebase.get_resource(path=workspace_root_path)
+                for resource in workspace_root.walk(codebase):
                     if NpmPackageJsonHandler.is_datafile(resource.location) and fnmatch.fnmatch(
                         name=resource.location, pat=workspace_path,
                     ):
@@ -1155,6 +1156,10 @@ def parse(cls, location, package_only=False):
         yield models.PackageData.from_data(package_data, package_only)
 
 
+class UnknownPnpmLockFormat(Exception):
+    pass
+
+
 class BasePnpmLockHandler(BaseNpmHandler):
 
     @classmethod
@@ -1181,31 +1186,62 @@ def parse(cls, location, package_only=False):
         }
         major_v, minor_v = lockfile_version.split(".")
 
-        resolved_packages = lock_data.get("packages", [])
+        resolved_packages = lock_data.get("packages", {})
+        dependency_relations = lock_data.get("snapshots", {})
+        dependency_relations_by_purl = {}
+        if dependency_relations:
+            for purl_fields, relations in dependency_relations.items():
+                clean_purl_fields = purl_fields.split("(")[0]
+                sections = clean_purl_fields.split("/")
+                namespace = None
+                if len(sections) == 2:
+                    namespace, name_version = sections
+                elif len(sections) == 1:
+                    name_version, = sections
+                name, version = name_version.split("@")
+
+                purl = PackageURL(
+                    type=cls.default_package_type,
+                    name=name,
+                    namespace=namespace,
+                    version=version,
+                ).to_string()
+                dependency_relations_by_purl[purl] = relations
+
         dependencies_by_purl = {}
 
         for purl_fields, data in resolved_packages.items():
             if major_v == "6":
                 clean_purl_fields = purl_fields.split("(")[0]
             elif major_v == "5" or is_shrinkwrap:
                 clean_purl_fields = purl_fields.split("_")[0]
-            else:
+            elif major_v == "9":
                 clean_purl_fields = purl_fields
-                raise Exception(lockfile_version, purl_fields)
+            else:
+                message = f"Unknown pnpm lockfile format: {lockfile_version}"
+                raise UnknownPnpmLockFormat(message, purl_fields)
 
             sections = clean_purl_fields.split("/")
-            name_version= None
+            name_version = None
+            namespace = None
             if major_v == "6":
                 if len(sections) == 2:
-                    namespace = None
                     _, name_version = sections
                 elif len(sections) == 3:
                     _, namespace, name_version = sections
+                elif len(sections) == 1:
+                    name_version, = sections
+
+                name, version = name_version.split("@")
+            elif major_v == "9":
+                if len(sections) == 2:
+                    namespace, name_version = sections
+                elif len(sections) == 1:
+                    name_version, = sections
 
                 name, version = name_version.split("@")
             elif major_v == "5" or is_shrinkwrap:
                 if len(sections) == 3:
-                    namespace = None
                     _, name, version = sections
                 elif len(sections) == 4:
                     _, namespace, name, version = sections
@@ -1223,7 +1259,21 @@ def parse(cls, location, package_only=False):
 
             dependencies = data.get('dependencies') or {}
             optional_dependencies = data.get('optionalDependencies') or {}
-            transitive_peer_dependencies = data.get('transitivePeerDependencies') or {}
+            transitive_peer_dependencies = data.get('transitivePeerDependencies') or []
+
+            if purl in dependency_relations_by_purl:
+                dependency_relations = dependency_relations_by_purl.get(purl)
+                if dependency_relations:
+                    deps = dependency_relations.get('dependencies')
+                    if deps:
+                        dependencies.update(deps)
+                    optional_deps = dependency_relations.get('optionalDependencies')
+                    if optional_deps:
+                        optional_dependencies.update(optional_deps)
+                    transitive_peer_deps = dependency_relations.get('transitivePeerDependencies')
+                    if transitive_peer_deps:
+                        transitive_peer_dependencies.extend(transitive_peer_deps)
+
             peer_dependencies = data.get('peerDependencies') or {}
             peer_dependencies_meta = data.get('peerDependenciesMeta') or {}