AWS Inspector 2 Line number bug + other changes (#14616)

* Fixed line number bug and added multiple fields to Package Vulnerability finding type

* changed startLine and endLine to keep descriptions consistent with past imports

* Add LocationData.dependency() support for package vulnerability findings

Populate unsaved_locations with dependency location data from
vulnerablePackages, gated behind V3_FEATURE_LOCATIONS. Also fix
process_endpoints to extend rather than overwrite unsaved_locations
so dependency locations are preserved.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Remove validate_locations call from metadata test

The test fixture contains Lambda ARNs with "$LATEST" which produces
invalid hostnames for endpoint validation. Endpoint validation is
already covered by the other parser tests.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Gate LocationData test assertions behind V3_FEATURE_LOCATIONS

Finding.unsaved_locations only exists when V3_FEATURE_LOCATIONS is
enabled, so the dependency location assertions must be conditional.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Fix import sorting for ruff linting

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Cody Maffucci <46459665+Maffooch@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

dojo/tools/aws_inspector2/parser.py

@@ -1,3 +1,4 @@
+import contextlib
 import json
 from datetime import UTC, datetime
 
@@ -114,6 +115,7 @@ def get_cvss_details(self, finding: Finding, raw_finding: dict) -> Finding:
 
     def get_package_vulnerability(self, finding: Finding, raw_finding: dict) -> Finding:
         vulnerability_details = raw_finding.get("packageVulnerabilityDetails", {})
+        vulnerable_packages = vulnerability_details.get("vulnerablePackages", [])
         vulnerability_packages_descriptions = "\n".join(
             [
                 (
@@ -123,14 +125,40 @@ def get_package_vulnerability(self, finding: Finding, raw_finding: dict) -> Find
                     f"\tfixed version: {vulnerability_package.get('fixedInVersion', 'N/A')}\n"
                     f"\tremediation: {vulnerability_package.get('remediation', 'N/A')}\n"
                 )
-                for vulnerability_package in vulnerability_details.get("vulnerablePackages", [])
+                for vulnerability_package in vulnerable_packages
             ],
         )
         if (vulnerability_id := vulnerability_details.get("vulnerabilityId", None)) is not None:
             finding.unsaved_vulnerability_ids = [vulnerability_id]
         vulnerability_source = vulnerability_details.get("source")
         vulnerability_source_url = vulnerability_details.get("sourceUrl")
-        # populate fields
+        # component name/version/file_path from the first vulnerable package
+        if vulnerable_packages:
+            finding.component_name = vulnerable_packages[0].get("name")
+            finding.component_version = vulnerable_packages[0].get("version")
+            finding.file_path = vulnerable_packages[0].get("filePath")
+            if settings.V3_FEATURE_LOCATIONS and finding.component_name:
+                finding.unsaved_locations.append(
+                    LocationData.dependency(
+                        name=finding.component_name,
+                        version=finding.component_version or "",
+                        file_path=finding.file_path or "",
+                    ),
+                )
+        # reference URLs from the advisory
+        reference_urls = vulnerability_details.get("referenceUrls", [])
+        if reference_urls:
+            finding.references = "\n".join(reference_urls)
+        # publish date from when the vendor first created the advisory
+        if vendor_created_at := vulnerability_details.get("vendorCreatedAt"):
+            with contextlib.suppress(ValueError):
+                finding.publish_date = date_parser.parse(vendor_created_at).date()
+        # CVSS v3 base score from the vendor-supplied CVSS entries
+        for cvss_entry in vulnerability_details.get("cvss", []):
+            if str(cvss_entry.get("version", "")).startswith("3") and cvss_entry.get("baseScore") is not None:
+                finding.cvssv3_score = float(cvss_entry["baseScore"])
+                break
+        # populate description fields
         if vulnerability_source is not None and vulnerability_source_url is not None:
             finding.url = vulnerability_source_url
             finding.description += (
@@ -149,8 +177,8 @@ def get_code_vulnerability(self, finding: Finding, raw_finding: dict) -> Finding
         file_path_info = raw_finding.get("filePath", {})
         file_name = file_path_info.get("fileName", "N/A")
         file_path = file_path_info.get("filePath", "N/A")
-        start_line = file_path_info.get("startLine", "N/A")
-        end_line = file_path_info.get("endLine", "N/A")
+        start_line = file_path_info.get("startLine", None)
+        end_line = file_path_info.get("endLine", None)
         detector_tags = ", ".join(raw_finding.get("detectorTags", []))
         reference_urls = ", ".join(raw_finding.get("referenceUrls", []))
         rule_id = raw_finding.get("ruleId", "N/A")
@@ -162,6 +190,10 @@ def get_code_vulnerability(self, finding: Finding, raw_finding: dict) -> Finding
         finding.sast_source_file_path = f"{file_path}{file_name}"
         finding.line = start_line
         finding.sast_source_line = start_line
+        if start_line is None:
+            start_line = "N/A"
+        if end_line is None:
+            end_line = "N/A"
         finding.description += (
             "\n**Additional info**\n"
             f"CWEs: {string_cwes}\n"
@@ -270,9 +302,9 @@ def process_endpoints(self, finding: Finding, raw_finding: dict) -> Finding:
                 endpoints.append(Endpoint(host=endpoint_host))
         finding.impact = "\n".join(impact)
         if settings.V3_FEATURE_LOCATIONS:
-            finding.unsaved_locations = endpoints
+            finding.unsaved_locations.extend(endpoints)
         else:
             # TODO: Delete this after the move to Locations
-            finding.unsaved_endpoints = endpoints
+            finding.unsaved_endpoints.extend(endpoints)
 
         return finding