Show PyPI Inspector link for remote PyPI findings (#797)

* feat: show PyPI Inspector link for remote findings

* feat: clickable per-finding PyPI Inspector deep links

* fix: populate inspector links on verify/project scan path

---------

Co-authored-by: Jensen Zane <jensenzane@Jensens-MacBook-Air.local>
Co-authored-by: Christophe Tafani-Dereeper <christophe.tafanidereeper@datadoghq.com>

Author: 3 people

guarddog/cli.py

@@ -443,14 +443,16 @@ def _scan_remote_sandboxed(scanner, name, version, rules):
             }
             for risk in risk_objects
         ]
-        return {
+        results = {
             "issues": metadata_results["issues"] + sourcecode_results["issues"],
             "errors": metadata_results["errors"] | sourcecode_results["errors"],
             "results": metadata_results["results"] | sourcecode_results["results"],
             "path": file_path,
             "risk_score": risk_score,
             "risks": formatted_risks,
         }
+        scanner._annotate_remote_results(results, package_info, name, version)
+        return results
     finally:
         shutil.rmtree(tmpdir, ignore_errors=True)
 

guarddog/reporters/human_readable.py

@@ -1,6 +1,7 @@
 import os
 import re
 from enum import Enum
+from urllib.parse import quote
 
 from termcolor import colored
 from guarddog.reporters import BaseReporter
@@ -201,7 +202,10 @@ def _rail(lines: List[str], color: str | None) -> List[str]:
 
     @staticmethod
     def _format_one_risk(
-        risk: dict, path_prefix: Optional[str], ceiling: str | None
+        risk: dict,
+        path_prefix: Optional[str],
+        ceiling: str | None,
+        deep_base: Optional[str] = None,
     ) -> List[str]:
         """Render a single risk as a railed block: rule, desc, location, code.
 
@@ -227,7 +231,16 @@ def _format_one_risk(
         ]
         if desc:
             block.append("  " + colored(desc, sev_color))
-        block.append("  " + colored(f"{loc_kw} {_sanitize(loc)}", "dark_grey"))
+
+        inspector_url = (
+            HumanReadableReporter._pypi_finding_inspector_url(deep_base, loc_raw)
+            if deep_base and loc_raw
+            else None
+        )
+        loc_line = colored(f"{loc_kw} {_sanitize(loc)}", "dark_grey")
+        if inspector_url:
+            loc_line = HumanReadableReporter._hyperlink(inspector_url, loc_line)
+        block.append("  " + loc_line)
 
         code = risk.get("threat_code", "")
         if code:
@@ -244,7 +257,10 @@ def _format_one_risk(
 
     @staticmethod
     def _format_findings(
-        risks: list, path_prefix: Optional[str], ceiling: str | None
+        risks: list,
+        path_prefix: Optional[str],
+        ceiling: str | None,
+        deep_base: Optional[str] = None,
     ) -> List[str]:
         """Per-tactic findings list, colored by rule severity clamped to the band."""
         lines: List[str] = []
@@ -278,7 +294,7 @@ def _format_findings(
 
                 for risk in tactic_risks:
                     lines += HumanReadableReporter._format_one_risk(
-                        risk, path_prefix, ceiling
+                        risk, path_prefix, ceiling, deep_base
                     )
 
                 lines.append("")
@@ -305,6 +321,62 @@ def _format_summary(risk_score: dict, num_risks: int) -> List[str]:
             colored(stats, "dark_grey"),
         ]
 
+    @staticmethod
+    def _hyperlink(url: str, label: str) -> str:
+        """Wrap `label` in an OSC 8 terminal hyperlink pointing to `url`.
+
+        Terminals that support OSC 8 render `label` as a clickable link; the rest
+        ignore the escape wrapper and show `label` unchanged. `url` must be free of
+        terminal control bytes, otherwise a crafted value could break out of the
+        escape sequence, so we fall back to the bare label when it isn't.
+        """
+        if _TERMINAL_CONTROL_RE.search(url):
+            return label
+        return f"\x1b]8;;{url}\x1b\\{label}\x1b]8;;\x1b\\"
+
+    @staticmethod
+    def _pypi_deep_link_base(results: dict) -> Optional[str]:
+        """Base for per-finding PyPI Inspector links, or None when unavailable.
+
+        Combines the project URL (already carries the escaped package/version) with
+        the scanned distribution's `/packages/...` path, giving a prefix that only
+        needs the in-archive file path and `#line.N` appended.
+        """
+        inspector_url = results.get("pypi_inspector_url")
+        dist_path = results.get("pypi_dist_path")
+        if not inspector_url or not dist_path or results.get("issues", 0) < 1:
+            return None
+        return inspector_url.rstrip("/") + dist_path
+
+    @staticmethod
+    def _pypi_finding_inspector_url(
+        deep_base: str, threat_location: str
+    ) -> Optional[str]:
+        """Deep link to a finding's file and line on PyPI Inspector.
+
+        `threat_location` is `<relpath-in-archive>:<line>`, relative to the extract
+        root, which is exactly the path Inspector expects after the distribution
+        filename.
+        """
+        file_part, _, line = threat_location.rpartition(":")
+        if not file_part or not line.isdigit():
+            return None
+        return f"{deep_base}/{quote(file_part, safe='/')}#line.{line}"
+
+    @staticmethod
+    def _format_pypi_inspector_url(results: dict) -> List[str]:
+        inspector_url = results.get("pypi_inspector_url")
+        if not inspector_url or results.get("issues", 0) < 1:
+            return []
+
+        link = HumanReadableReporter._hyperlink(
+            inspector_url, colored("view on PyPI Inspector", "dark_grey")
+        )
+        return [
+            "",
+            colored("Package files:", "dark_grey") + f" {link}",
+        ]
+
     @staticmethod
     def _format_header(identifier: str, num_risks: int) -> List[str]:
         bold = colored(_sanitize(identifier), None, attrs=["bold"])
@@ -333,10 +405,15 @@ def print_scan_results(identifier: str, results: dict) -> str:
         )
         ceiling = HumanReadableReporter._BAND_CEILING.get(label)
 
+        deep_base = HumanReadableReporter._pypi_deep_link_base(results)
+
         lines: List[str] = []
         lines += HumanReadableReporter._format_header(identifier, len(risks))
         if risks:
-            lines += HumanReadableReporter._format_findings(risks, path_prefix, ceiling)
+            lines += HumanReadableReporter._format_findings(
+                risks, path_prefix, ceiling, deep_base
+            )
+        lines += HumanReadableReporter._format_pypi_inspector_url(results)
         if risk_score:
             lines += HumanReadableReporter._format_summary(risk_score, len(risks))
 

guarddog/scanners/pypi_package_scanner.py

@@ -1,5 +1,6 @@
 import os
 import typing
+from urllib.parse import quote, urlparse
 
 from guarddog.analyzer.analyzer import Analyzer
 from guarddog.ecosystems import ECOSYSTEM
@@ -18,6 +19,33 @@ def download_and_get_package_info(
         extract_dir = self.download_package(package_name, directory, version)
         return get_package_info(package_name), extract_dir
 
+    def get_package_version(self, package_info: dict, requested_version=None):
+        if requested_version is not None:
+            return requested_version
+        return package_info.get("info", {}).get("version")
+
+    @staticmethod
+    def _select_distribution(package_info: dict, version) -> typing.Optional[dict]:
+        """Return the release file GuardDog scans for a version (first supported archive)."""
+        releases = package_info.get("releases", {})
+        if version is None:
+            version = package_info.get("info", {}).get("version")
+        for file in releases.get(version, []):
+            if is_supported_archive(file["filename"]):
+                return file
+        return None
+
+    def get_package_dist_path(self, package_info: dict, requested_version=None):
+        distribution = self._select_distribution(package_info, requested_version)
+        if distribution is None:
+            return None
+        return urlparse(distribution["url"]).path
+
+    def get_package_inspector_url(self, name, version):
+        package = quote(name, safe="")
+        version = quote(str(version), safe="")
+        return f"https://inspector.pypi.io/project/{package}/{version}/"
+
     def download_package(self, package_name, directory, version=None) -> str:
         """Downloads the PyPI distribution for a given package and version
 
@@ -46,20 +74,15 @@ def download_package(self, package_name, directory, version=None) -> str:
                 f"Version {version} for package {package_name} doesn't exist."
             )
 
-        files = releases[version]
-        url, file_extension = None, None
-
-        for file in files:
-            if is_supported_archive(file["filename"]):
-                url = file["url"]
-                _, file_extension = os.path.splitext(file["filename"])
-                break
-
-        if not (url and file_extension):
+        distribution = self._select_distribution(data, version)
+        if distribution is None:
             raise Exception(
                 f"Compressed file for {package_name} does not exist on PyPI."
             )
 
+        url = distribution["url"]
+        _, file_extension = os.path.splitext(distribution["filename"])
+
         # Path to compressed package
         zippath = os.path.join(directory, package_name + file_extension)
         unzippedpath = os.path.join(directory, package_name)

guarddog/scanners/scanner.py

@@ -217,6 +217,38 @@ def download_and_get_package_info(
     ) -> typing.Tuple[dict, str]:
         raise NotImplementedError("download_and_get_package_info is not implemented")
 
+    def get_package_version(self, package_info: dict, requested_version=None):
+        return requested_version
+
+    def get_package_dist_path(self, package_info: dict, requested_version=None):
+        """Return the PyPI-style `/packages/...` path of the scanned distribution.
+
+        Used to build PyPI Inspector deep links. Only PyPI packages have one.
+        """
+        return None
+
+    def get_package_inspector_url(self, name, version) -> typing.Optional[str]:
+        """URL to the package's files on PyPI Inspector, or None if unsupported."""
+        return None
+
+    def _annotate_remote_results(self, results, package_info, name, version):
+        """Attach PyPI Inspector metadata to a remote scan's results.
+
+        Shared by the plain and sandboxed remote scan paths so direct scans and
+        `verify` project scans surface the same version, distribution path, and
+        Inspector links.
+        """
+        scanned_version = self.get_package_version(package_info, version)
+        if scanned_version is not None:
+            results["package_version"] = scanned_version
+            if results.get("issues", 0) >= 1:
+                inspector_url = self.get_package_inspector_url(name, scanned_version)
+                if inspector_url is not None:
+                    results["pypi_inspector_url"] = inspector_url
+        dist_path = self.get_package_dist_path(package_info, version)
+        if dist_path is not None:
+            results["pypi_dist_path"] = dist_path
+
     def _scan_remote(
         self, name, base_dir, version=None, rules=None, write_package_info=False
     ):
@@ -233,6 +265,7 @@ def _scan_remote(
             return {"issues": 0, "errors": {"download-package": str(e)}}
 
         results = self.analyzer.analyze(file_path, package_info, rules, name, version)
+        self._annotate_remote_results(results, package_info, name, version)
         if write_package_info:
             package_name = name.replace("/", "-")
             suffix = (

tests/core/test_pypi_package_scanner.py

@@ -0,0 +1,92 @@
+from guarddog.scanners import PypiPackageScanner
+
+
+def _package_info():
+    return {
+        "info": {"version": "1.0"},
+        "releases": {
+            "1.0": [
+                {
+                    "filename": "aws-dd-forwarder-1.0-py3-none-any.whl",
+                    "url": (
+                        "https://files.pythonhosted.org/packages/ab/cd/"
+                        "ef01/aws-dd-forwarder-1.0-py3-none-any.whl"
+                    ),
+                },
+                {
+                    "filename": "aws-dd-forwarder-1.0.tar.gz",
+                    "url": (
+                        "https://files.pythonhosted.org/packages/f8/57/"
+                        "87be/aws-dd-forwarder-1.0.tar.gz"
+                    ),
+                },
+            ]
+        },
+    }
+
+
+def test_get_package_dist_path_returns_first_supported_archive_path():
+    scanner = PypiPackageScanner()
+    path = scanner.get_package_dist_path(_package_info(), "1.0")
+    assert path == "/packages/ab/cd/ef01/aws-dd-forwarder-1.0-py3-none-any.whl"
+
+
+def test_get_package_dist_path_defaults_to_latest_version():
+    scanner = PypiPackageScanner()
+    path = scanner.get_package_dist_path(_package_info())
+    assert path == "/packages/ab/cd/ef01/aws-dd-forwarder-1.0-py3-none-any.whl"
+
+
+def test_get_package_dist_path_none_when_no_supported_archive():
+    scanner = PypiPackageScanner()
+    package_info = {
+        "info": {"version": "1.0"},
+        "releases": {
+            "1.0": [{"filename": "aws-dd-forwarder-1.0.metadata", "url": "https://x/y"}]
+        },
+    }
+    assert scanner.get_package_dist_path(package_info, "1.0") is None
+
+
+def test_get_package_dist_path_none_for_unknown_version():
+    scanner = PypiPackageScanner()
+    assert scanner.get_package_dist_path(_package_info(), "9.9") is None
+
+
+def test_get_package_inspector_url_builds_project_url():
+    scanner = PypiPackageScanner()
+    url = scanner.get_package_inspector_url("requests", "2.28.1")
+    assert url == "https://inspector.pypi.io/project/requests/2.28.1/"
+
+
+def test_get_package_inspector_url_escapes_name_and_version():
+    scanner = PypiPackageScanner()
+    url = scanner.get_package_inspector_url("my package", "1.0.0+local")
+    assert url == "https://inspector.pypi.io/project/my%20package/1.0.0%2Blocal/"
+
+
+def test_annotate_remote_results_sets_links_when_findings_exist():
+    scanner = PypiPackageScanner()
+    results = {"issues": 1}
+    scanner._annotate_remote_results(
+        results, _package_info(), "aws-dd-forwarder", "1.0"
+    )
+    assert results["package_version"] == "1.0"
+    assert results["pypi_inspector_url"] == (
+        "https://inspector.pypi.io/project/aws-dd-forwarder/1.0/"
+    )
+    assert results["pypi_dist_path"] == (
+        "/packages/ab/cd/ef01/aws-dd-forwarder-1.0-py3-none-any.whl"
+    )
+
+
+def test_annotate_remote_results_omits_inspector_url_without_findings():
+    scanner = PypiPackageScanner()
+    results = {"issues": 0}
+    scanner._annotate_remote_results(
+        results, _package_info(), "aws-dd-forwarder", "1.0"
+    )
+    assert "pypi_inspector_url" not in results
+    # Version and distribution path are still recorded.
+    assert results["package_version"] == "1.0"
+    assert "pypi_dist_path" in results