Improved teh sarif file parsing

Author: Orlando Barrera II

socketsecurity/core/messages.py

@@ -15,16 +15,12 @@ class Messages:
     @staticmethod
     def map_severity_to_sarif(severity: str) -> str:
         """
-        Map Socket severity levels to SARIF levels (GitHub code scanning).
-        
-        'low' -> 'note'
-        'medium' or 'middle' -> 'warning'
-        'high' or 'critical' -> 'error'
+        Map Socket Security severity levels to SARIF levels.
         """
         severity_mapping = {
             "low": "note",
             "medium": "warning",
-            "middle": "warning",  # older data might say "middle"
+            "middle": "warning",
             "high": "error",
             "critical": "error",
         }
@@ -33,82 +29,67 @@ def map_severity_to_sarif(severity: str) -> str:
     @staticmethod
     def find_line_in_file(packagename: str, packageversion: str, manifest_file: str) -> tuple:
         """
-        Finds the line number and snippet of code for the given package/version in a manifest file.
-        Returns a 2-tuple: (line_number, snippet_or_message).
-
-        Supports:
-          1) JSON-based manifest files (package-lock.json, Pipfile.lock, composer.lock)
-             - Locates a dictionary entry with the matching package & version
-             - Does a rough line-based search to find the actual line in the raw text
-          2) Text-based (requirements.txt, package.json, yarn.lock, etc.)
-             - Uses compiled regex patterns to detect a match line by line
+        Given a manifest file, find the line number and snippet where the package is declared.
+        For JSON-based manifests (package-lock.json, Pipfile.lock, composer.lock, package.json), 
+        we attempt to parse the JSON to verify the package is present, then search for the key.
+        For text-based manifests, we use a regex search.
         """
-        # Extract just the file name to detect manifest type
         file_type = Path(manifest_file).name
 
-        # ----------------------------------------------------
-        # 1) JSON-based manifest files
-        # ----------------------------------------------------
-        if file_type in ["package-lock.json", "Pipfile.lock", "composer.lock"]:
+        # Handle JSON-based files.
+        if file_type in ["package-lock.json", "Pipfile.lock", "composer.lock", "package.json"]:
             try:
-                # Read entire file so we can parse JSON and also do raw line checks
                 with open(manifest_file, "r", encoding="utf-8") as f:
                     raw_text = f.read()
-
-                # Attempt JSON parse
-                data = json.loads(raw_text)
-
-                # In practice, you may need to check data["dependencies"], data["default"], etc.
-                # This is an example approach.
-                packages_dict = (
-                    data.get("packages")
-                    or data.get("default")
-                    or data.get("dependencies")
-                    or {}
-                )
-
-                found_key = None
-                found_info = None
-                # Locate a dictionary entry whose 'version' matches
-                for key, value in packages_dict.items():
-                    # For NPM package-lock, keys might look like "node_modules/axios"
-                    if key.endswith(packagename) and "version" in value:
-                        if value["version"] == packageversion:
-                            found_key = key
-                            found_info = value
-                            break
-
-                if found_key and found_info:
-                    # Search lines to approximate the correct line number
-                    needle_key = f'"{found_key}":'               # e.g. "node_modules/axios":
-                    needle_version = f'"version": "{packageversion}"'
-                    lines = raw_text.splitlines()
-                    best_line = 1
-                    snippet = None
-
-                    for i, line in enumerate(lines, start=1):
-                        if (needle_key in line) or (needle_version in line):
-                            best_line = i
-                            snippet = line.strip()
-                            break  # On first match, stop
-
-                    # If we found an approximate line, return it; else fallback to line 1
-                    if best_line > 0 and snippet:
-                        return best_line, snippet
-                    else:
-                        return 1, f'"{found_key}": {found_info}'
+                try:
+                    data = json.loads(raw_text)
+                except json.JSONDecodeError:
+                    data = {}
+
+                found = False
+                # For package.json, check dependencies and devDependencies.
+                if file_type == "package.json":
+                    deps = data.get("dependencies", {})
+                    deps_dev = data.get("devDependencies", {})
+                    all_deps = {**deps, **deps_dev}
+                    if packagename in all_deps:
+                        actual_version = all_deps[packagename]
+                        # Allow for versions with caret/tilde prefixes.
+                        if actual_version == packageversion or actual_version.lstrip("^~") == packageversion:
+                            found = True
                 else:
-                    return 1, f"{packagename} {packageversion} (not found in {manifest_file})"
-
-            except (FileNotFoundError, json.JSONDecodeError):
-                return 1, f"Error reading {manifest_file}"
+                    # For other JSON-based manifests, look into common keys.
+                    for key in ["packages", "default", "dependencies"]:
+                        if key in data:
+                            packages_dict = data[key]
+                            # In package-lock.json, keys can be paths (e.g. "node_modules/axios")
+                            for key_item, info in packages_dict.items():
+                                if key_item.endswith(packagename):
+                                    # info may be a dict (with "version") or a simple version string.
+                                    ver = info if isinstance(info, str) else info.get("version", "")
+                                    if ver == packageversion:
+                                        found = True
+                                        break
+                        if found:
+                            break
 
-        # ----------------------------------------------------
-        # 2) Text-based / line-based manifests
-        # ----------------------------------------------------
-        # Define a dictionary of patterns for common manifest types
+                if not found:
+                    return 1, f'"{packagename}": not found in {manifest_file}'
+
+                # Now search the raw text to locate the declaration line.
+                needle = f'"{packagename}":'
+                lines = raw_text.splitlines()
+                for i, line in enumerate(lines, start=1):
+                    if needle in line:
+                        return i, line.strip()
+                return 1, f'"{packagename}": declaration not found'
+            except FileNotFoundError:
+                return 1, f"{manifest_file} not found"
+            except Exception as e:
+                return 1, f"Error reading {manifest_file}: {e}"
+
+        # For text-based files, define regex search patterns for common manifest types.
         search_patterns = {
-            "package.json":         rf'"{packagename}":\s*"{packageversion}"',
             "yarn.lock":            rf'{packagename}@{packageversion}',
             "pnpm-lock.yaml":       rf'"{re.escape(packagename)}"\s*:\s*\{{[^}}]*"version":\s*"{re.escape(packageversion)}"',
             "requirements.txt":     rf'^{re.escape(packagename)}\s*(?:==|===|!=|>=|<=|~=|\s+)?\s*{re.escape(packageversion)}(?:\s*;.*)?$',
@@ -132,33 +113,25 @@ def find_line_in_file(packagename: str, packageversion: str, manifest_file: str)
             "conanfile.txt":        rf'{re.escape(packagename)}/{re.escape(packageversion)}',
             "vcpkg.json":           rf'"{re.escape(packagename)}":\s*"{re.escape(packageversion)}"',
         }
-
-        # If no specific pattern is found for this file name, fallback to a naive approach
         searchstring = search_patterns.get(file_type, rf'{re.escape(packagename)}.*{re.escape(packageversion)}')
-
         try:
-            # Read file lines and search for a match
             with open(manifest_file, 'r', encoding="utf-8") as file:
                 lines = [line.rstrip("\n") for line in file]
                 for line_number, line_content in enumerate(lines, start=1):
-                    # For Python conditional dependencies, ignore everything after first ';'
+                    # For cases where dependencies have conditionals (e.g. Python), only consider the main part.
                     line_main = line_content.split(";", 1)[0].strip()
-
-                    # Use a case-insensitive regex search
                     if re.search(searchstring, line_main, re.IGNORECASE):
                         return line_number, line_content.strip()
-
         except FileNotFoundError:
             return 1, f"{manifest_file} not found"
         except Exception as e:
             return 1, f"Error reading {manifest_file}: {e}"
-
         return 1, f"{packagename} {packageversion} (not found)"
 
     @staticmethod
     def get_manifest_type_url(manifest_file: str, pkg_name: str, pkg_version: str) -> str:
         """
-        Determine the correct URL path based on the manifest file type.
+        Determine the URL prefix based on the manifest file.
         """
         manifest_to_url_prefix = {
             "package.json": "npm",
@@ -181,18 +154,20 @@ def get_manifest_type_url(manifest_file: str, pkg_name: str, pkg_version: str) -
             "composer.json": "composer",
             "vcpkg.json": "vcpkg",
         }
-
         file_type = Path(manifest_file).name
         url_prefix = manifest_to_url_prefix.get(file_type, "unknown")
         return f"https://socket.dev/{url_prefix}/package/{pkg_name}/alerts/{pkg_version}"
 
     @staticmethod
     def create_security_comment_sarif(diff) -> dict:
         """
-        Create SARIF-compliant output from the diff report, including dynamic URL generation
-        based on manifest type and improved <br/> formatting for GitHub SARIF display.
+        Create a SARIF-compliant JSON object for alerts. This function now:
+         - Accepts multiple manifest files (from alert.introduced_by or alert.manifests)
+         - Generates one SARIF location per manifest file.
+         - Supports various language-specific manifest types.
         """
         scan_failed = False
+        # (Optional: handle scan failure based on alert.error flags)
         if len(diff.new_alerts) == 0:
             for alert in diff.new_alerts:
                 if alert.error:
@@ -225,27 +200,30 @@ def create_security_comment_sarif(diff) -> dict:
             rule_id = f"{pkg_name}=={pkg_version}"
             severity = alert.severity
 
-            # --- NEW LOGIC: Determine the list of manifest files ---
-            if alert.introduced_by and isinstance(alert.introduced_by[0], list):
-                # Extract file names from each introduced_by entry
-                manifest_files = [entry[1] for entry in alert.introduced_by]
-            elif alert.manifests:
-                # Split semicolon-delimited manifest string if necessary
-                manifest_files = [mf.strip() for mf in alert.manifests.split(";")]
+            # --- Determine manifest files from alert data ---
+            manifest_files = []
+            if alert.introduced_by and isinstance(alert.introduced_by, list):
+                for entry in alert.introduced_by:
+                    if isinstance(entry, list) and len(entry) >= 2:
+                        manifest_files.append(entry[1])
+                    elif isinstance(entry, str):
+                        manifest_files.extend([m.strip() for m in entry.split(";") if m.strip()])
+            elif hasattr(alert, 'manifests') and alert.manifests:
+                manifest_files = [mf.strip() for mf in alert.manifests.split(";") if mf.strip()]
             else:
                 manifest_files = ["requirements.txt"]
 
-            # Use the first file for generating the help URL.
+            # Use the first manifest for URL generation.
             socket_url = Messages.get_manifest_type_url(manifest_files[0], pkg_name, pkg_version)
 
-            # Prepare the description messages.
+            # Prepare descriptions with HTML <br/> for GitHub display.
             short_desc = (
                 f"{alert.props.get('note', '')}<br/><br/>Suggested Action:<br/>"
                 f"{alert.suggestion}<br/><a href=\"{socket_url}\">{socket_url}</a>"
             )
             full_desc = "{} - {}".format(alert.title, alert.description.replace('\r\n', '<br/>'))
 
-            # Create the rule if not already defined.
+            # Create or reuse the rule definition.
             if rule_id not in rules_map:
                 rules_map[rule_id] = {
                     "id": rule_id,
@@ -258,12 +236,12 @@ def create_security_comment_sarif(diff) -> dict:
                     },
                 }
 
-            # --- NEW LOGIC: Create separate locations for each manifest file ---
+            # --- Build SARIF locations for each manifest file ---
             locations = []
             for mf in manifest_files:
                 line_number, line_content = Messages.find_line_in_file(pkg_name, pkg_version, mf)
                 if line_number < 1:
-                    line_number = 1  # Ensure SARIF compliance.
+                    line_number = 1
                 locations.append({
                     "physicalLocation": {
                         "artifactLocation": {"uri": mf},
@@ -274,15 +252,13 @@ def create_security_comment_sarif(diff) -> dict:
                     }
                 })
 
-            # Add the SARIF result.
             result_obj = {
                 "ruleId": rule_id,
                 "message": {"text": short_desc},
                 "locations": locations,
             }
             results_list.append(result_obj)
 
-        # Attach rules and results.
         sarif_data["runs"][0]["tool"]["driver"]["rules"] = list(rules_map.values())
         sarif_data["runs"][0]["results"] = results_list