Repository: DataDog/guarddog (python)

Issue:
Unused callback function argument in PackageScanner
The `PackageScanner.scan_local()` method accepts a `callback` function argument that it does not use.

```python
# guarddog/scanners/scanner.py:227

def scan_local(
        self, path, rules=None, callback: typing.Callable[[dict], None] = noop
) -> dict:
    if rules is not None:
        rules = set(rules)

    if not os.path.exists(path):
        raise Exception(f"Path {path} does not exist.")

    if any(path.endswith(ext) for ext in (".tar.gz", ".tgz", ".zip", ".whl")):
        with tempfile.TemporaryDirectory() as tmpdirname:
            safe_extract(path, tmpdirname)
            return self.analyzer.analyze_sourcecode(
                tmpdirname, rules=rules
            )

    if os.path.isdir(path):
        return self.analyzer.analyze_sourcecode(path, rules=rules)

    raise Exception(
        f"Path {path} is not a directory nor an archive type supported by GuardDog."
    )
```

It should be applied to the result of `analyze_sourcecode()` before returning it.

Pull request:
Simplify local target checks
This PR makes three principal changes:

1. It simplifies the conditions under which Guarddog will decide to perform a local scan: whenever the target is a directory or a regular file in the local filesystem.  It also eliminates as much as possible duplicate checks related to local targets

2. It adds functions in `guarddog/utils/archives.py` for checking whether a file extension is for a supported archive format and uses them pervasively

3. Closes #411

Patch:
diff --git a/guarddog/cli.py b/guarddog/cli.py
index f957994..e577d05 100644
--- a/guarddog/cli.py
+++ b/guarddog/cli.py
@@ -4,6 +4,7 @@ CLI command that scans a package version for user-specified malware flags.
 Includes rules based on package registry metadata and source code analysis.
 """
 
+from functools import reduce
 import json as js
 import logging
 import os
@@ -20,7 +21,6 @@ from guarddog.ecosystems import ECOSYSTEM
 from guarddog.reporters.sarif import report_verify_sarif
 from guarddog.scanners import get_scanner
 from guarddog.scanners.scanner import PackageScanner
-from functools import reduce
 
 EXIT_CODE_ISSUES_FOUND = 1
 
@@ -190,32 +190,6 @@ def _verify(
     return return_value  # this is mostly for testing
 
 
-def is_local_target(identifier: str) -> bool:
-    """
-    @param identifier:  The name/path of the package as passed to "guarddog ecosystem scan"
-    @return:            Whether the identifier should be considered a local path
-    """
-    if (
-        identifier.startswith("/")
-        or identifier.startswith("./")
-        or identifier.startswith("../")
-    ):
-        return True
-
-    if identifier == ".":
-        return True
-
-    # If this looks like an archive, consider it as a local target if the target exists on the local filesystem
-    if (
-        identifier.endswith(".tar.gz")
-        or identifier.endswith(".zip")
-        or identifier.endswith(".whl")
-    ):
-        return os.path.exists(identifier)
-
-    return False
-
-
 def _scan(
     identifier,
     version,
@@ -240,20 +214,17 @@ def _scan(
         sys.exit(1)
 
     results = []
-    if is_local_target(identifier):
-        log.debug(
-            f"Considering that '{identifier}' is a local target, scanning filesystem"
-        )
-        if os.path.isdir(identifier):
-            log.debug(f"Considering that '{identifier}' as a local directory")
-            for package in os.listdir(identifier):
-                result = scanner.scan_local(f"{identifier}/{package}", rule_param)
-                result["package"] = package
-                results.append(result)
-        else:
-            result = scanner.scan_local(identifier, rule_param)
-            result["package"] = identifier
+    if os.path.isdir(identifier):
+        log.debug(f"Considering that '{identifier}' is a local directory")
+        for package in os.listdir(identifier):
+            result = scanner.scan_local(f"{identifier}/{package}", rule_param)
+            result["package"] = package
             results.append(result)
+    elif os.path.isfile(identifier):
+        log.debug(f"Considering that '{identifier}' is a local file")
+        result = scanner.scan_local(identifier, rule_param)
+        result["package"] = identifier
+        results.append(result)
     else:
         log.debug(f"Considering that '{identifier}' is a remote target")
         try:
diff --git a/guarddog/scanners/pypi_package_scanner.py b/guarddog/scanners/pypi_package_scanner.py
index f2ad888..e6d7015 100644
--- a/guarddog/scanners/pypi_package_scanner.py
+++ b/guarddog/scanners/pypi_package_scanner.py
@@ -4,6 +4,7 @@ import typing
 from guarddog.analyzer.analyzer import Analyzer
 from guarddog.ecosystems import ECOSYSTEM
 from guarddog.scanners.scanner import PackageScanner
+from guarddog.utils.archives import is_supported_archive
 from guarddog.utils.package_info import get_package_info
 
 
@@ -42,25 +43,20 @@ class PypiPackageScanner(PackageScanner):
             raise Exception(f"Version {version} for package {package_name} doesn't exist.")
 
         files = releases[version]
-        url = None
-        file_extension = None
+        url, file_extension = None, None
 
         for file in files:
-            # Store url to compressed package and appropriate file extension
-            if file["filename"].endswith(".tar.gz"):
+            if is_supported_archive(file["filename"]):
                 url = file["url"]
-                file_extension = ".tar.gz"
+                _, file_extension = os.path.splitext(file["filename"])
+                break
 
-            if any(file["filename"].endswith(ext) for ext in (".egg", ".whl", ".zip")):
-                url = file["url"]
-                file_extension = ".zip"
-
-        if not (url or file_extension):
+        if not (url and file_extension):
             raise Exception(f"Compressed file for {package_name} does not exist on PyPI.")
 
         # Path to compressed package
         zippath = os.path.join(directory, package_name + file_extension)
-        unzippedpath = zippath.removesuffix(file_extension)
-
+        unzippedpath = os.path.join(directory, package_name)
         self.download_compressed(url, zippath, unzippedpath)
+
         return unzippedpath
diff --git a/guarddog/scanners/scanner.py b/guarddog/scanners/scanner.py
index 8026bdc..f23cdaf 100644
--- a/guarddog/scanners/scanner.py
+++ b/guarddog/scanners/scanner.py
@@ -233,35 +233,31 @@ class PackageScanner(Scanner):
         Args:
             path (str): path to package
             rules (set, optional): Set of rule names to use. Defaults to all rules.
+            callback (typing.Callable[[dict], None], optional): Callback to apply to Analyzer output
 
         Raises:
             Exception: Analyzer exception
 
         Returns:
             dict: Analyzer output with rules to results mapping
-            rules: rules to apply
-            callback: callback to call for each result
         """
 
         if rules is not None:
             rules = set(rules)
 
-        if not os.path.exists(path):
-            raise Exception(f"Path {path} does not exist.")
-
-        if any(path.endswith(ext) for ext in (".tar.gz", ".tgz", ".zip", ".whl")):
-            with tempfile.TemporaryDirectory() as tmpdirname:
-                safe_extract(path, tmpdirname)
-                return self.analyzer.analyze_sourcecode(
-                    tmpdirname, rules=rules
-                )
-
+        results = None
         if os.path.isdir(path):
-            return self.analyzer.analyze_sourcecode(path, rules=rules)
+            results = self.analyzer.analyze_sourcecode(path, rules=rules)
+        elif os.path.isfile(path):
+            with tempfile.TemporaryDirectory() as tempdir:
+                safe_extract(path, tempdir)
+                results = self.analyzer.analyze_sourcecode(tempdir, rules=rules)
+        else:
+            raise Exception(f"Local scan target {path} is neither a directory nor a file.")
 
-        raise Exception(
-            f"Path {path} is not a directory nor an archive type supported by GuardDog."
-        )
+        callback(results)
+
+        return results
 
     @abstractmethod
     def download_and_get_package_info(
diff --git a/guarddog/utils/archives.py b/guarddog/utils/archives.py
index 44b47aa..d23b37a 100644
--- a/guarddog/utils/archives.py
+++ b/guarddog/utils/archives.py
@@ -7,24 +7,68 @@ import tarsafe  # type:ignore
 log = logging.getLogger("guarddog")
 
 
+def is_supported_archive(path: str) -> bool:
+    """
+    Decide whether a file contains a supported archive.
+
+    Args:
+        path (str): The local filesystem path to examine
+
+    Returns:
+        bool: Represents the decision reached for the file
+    """
+    return is_tar_archive(path) or is_zip_archive(path)
+
+
+def is_tar_archive(path: str) -> bool:
+    """
+    Decide whether a file contains a tar archive.
+
+    Args:
+        path (str): The local filesystem path to examine
+
+    Returns:
+        bool: Represents the decision reached for the file
+    """
+    return any(path.endswith(ext) for ext in [".tar.gz", ".tgz"])
+
+
+def is_zip_archive(path: str) -> bool:
+    """
+    Decide whether a file contains a zip, whl or egg archive.
+
+    Args:
+        path (str): The local filesystem path to examine
+
+    Returns:
+        bool: Represents the decision reached for the file
+    """
+    return any(path.endswith(ext) for ext in [".zip", ".whl", ".egg"])
+
+
 def safe_extract(source_archive: str, target_directory: str) -> None:
     """
     safe_extract safely extracts archives to a target directory.
 
-    This function does not clean up the original archive, and does not create the target directory if it does not exist.
+    This function does not clean up the original archive and does not
+    create the target directory if it does not exist.  It also assumes
+    the source archive argument is a path to a regular file on the
+    local filesystem.
 
     @param source_archive:      The archive to extract
     @param target_directory:    The directory where to extract the archive to
     @raise ValueError           If the archive type is unsupported
+
     """
     log.debug(f"Extracting archive {source_archive} to directory {target_directory}")
-    if source_archive.endswith('.tar.gz') or source_archive.endswith('.tgz'):
+    if is_tar_archive(source_archive):
         tarsafe.open(source_archive).extractall(target_directory)
-    elif source_archive.endswith('.zip') or source_archive.endswith('.whl'):
+    elif is_zip_archive(source_archive):
         with zipfile.ZipFile(source_archive, 'r') as zip:
             for file in zip.namelist():
-                # Note: zip.extract cleans up any malicious file name such as directory traversal attempts
-                # This is not the case of zipfile.extractall
+                # Note: zip.extract cleans up any malicious file name
+                # such as directory traversal attempts This is not the
+                # case of zipfile.extractall
                 zip.extract(file, path=os.path.join(target_directory, file))
     else:
-        raise ValueError("unsupported archive extension: " + target_directory)
+        raise ValueError(f"unsupported archive extension: {source_archive}")


Test patch:
diff --git a/tests/core/test_cli.py b/tests/core/test_cli.py
index 404bc75..82f5f4a 100644
--- a/tests/core/test_cli.py
+++ b/tests/core/test_cli.py
@@ -1,45 +1,139 @@
-import unittest.mock
+import unittest
+import unittest.mock as mock
 
 import guarddog.cli
+import guarddog.scanners.scanner as scanner
 from guarddog.ecosystems import ECOSYSTEM
 
 
-def test_is_local_target():
-    assert guarddog.cli.is_local_target("/tmp/foo")
-    assert guarddog.cli.is_local_target("./foo")
-    assert guarddog.cli.is_local_target("../foo")
-    assert guarddog.cli.is_local_target(".")
-    assert not guarddog.cli.is_local_target("foo")
-
-    with unittest.mock.patch("os.path.exists") as mock:
-        mock.return_value = True
-        assert guarddog.cli.is_local_target("foo.tar.gz")
-
-    with unittest.mock.patch("os.path.exists") as mock:
-        mock.return_value = False
-        assert not guarddog.cli.is_local_target("foo.tar.gz")
-
-
-def test_get_rule_param_include():
-    """
-    Test the parsing function returns the included parameter
-    """
-    rules = guarddog.cli._get_rule_param(("shady-links",), (), ECOSYSTEM.NPM)
-    assert rules
-    assert len(rules) == 1
-
-def test_get_rule_param_exclude():
-    """
-    Test the parsing function returns a list without the excluded parameter
-    """
-    rules = guarddog.cli._get_rule_param((), ("shady-links",), ECOSYSTEM.PYPI)
-    assert rules
-    assert len(rules) != 1
-    assert "shady-links" not in rules
-
-def test_get_rule_param_empty():
-    """
-    Test the parsing function returns None when no rules are provided
-    """
-    rules = guarddog.cli._get_rule_param((), (), ECOSYSTEM.PYPI)
-    assert rules is None
+class TestCli(unittest.TestCase):
+
+    def test_local_directory(self):
+        """
+        Test that the CLI identifies local directories correctly
+        """
+        self._test_local_directory_template("/tmp/foo")
+        self._test_local_directory_template("./foo")
+        self._test_local_directory_template("../foo")
+        self._test_local_directory_template(".")
+
+    def test_local_file(self):
+        """
+        Test that the CLI identifies local files correctly
+        """
+        self._test_local_file_template("/tmp/foo")
+        self._test_local_file_template("./foo")
+        self._test_local_file_template("../foo")
+
+    def test_get_rule_param_include(self):
+        """
+        Test the parsing function returns the included parameter
+        """
+        rules = guarddog.cli._get_rule_param(("shady-links",), (), ECOSYSTEM.NPM)
+        assert rules
+        assert len(rules) == 1
+
+    def test_get_rule_param_exclude(self):
+        """
+        Test the parsing function returns a list without the excluded parameter
+        """
+        rules = guarddog.cli._get_rule_param((), ("shady-links",), ECOSYSTEM.PYPI)
+        assert rules
+        assert len(rules) != 1
+        assert "shady-links" not in rules
+
+    def test_get_rule_param_empty(self):
+        """
+        Test the parsing function returns None when no rules are provided
+        """
+        rules = guarddog.cli._get_rule_param((), (), ECOSYSTEM.PYPI)
+        assert rules is None
+
+    def _test_local_directory_template(self, directory: str):
+        # `directory` is a directory
+        with mock.patch("os.path.isdir") as isdir:
+            isdir.return_value = True
+            with mock.patch("os.listdir") as listdir:
+                listdir.return_value = []
+                with self.assertLogs("guarddog", level="DEBUG") as cm:
+                    guarddog.cli._scan(directory, "0.1.0", (), (), None, False, ECOSYSTEM.PYPI)
+                self.assertIn(
+                    f"DEBUG:guarddog:Considering that '{directory}' is a local directory",
+                    cm.output
+                )
+                self.assertNotIn(
+                    f"DEBUG:guarddog:Considering that '{directory}' is a local file",
+                    cm.output
+                )
+                self.assertNotIn(
+                    f"DEBUG:guarddog:Considering that '{directory}' is a remote target",
+                    cm.output
+                )
+
+        # `directory` is neither a directory nor a file
+        with mock.patch("os.path.isdir") as isdir:
+            isdir.return_value = False
+            with mock.patch("os.path.isfile") as isfile:
+                isfile.return_value = False
+                with mock.patch.object(scanner.PackageScanner, 'scan_local', return_value={}) as _:
+                    with self.assertLogs("guarddog", level="DEBUG") as cm:
+                        guarddog.cli._scan(directory, "0.1.0", (), (), None, False, ECOSYSTEM.PYPI)
+                    self.assertNotIn(
+                        f"DEBUG:guarddog:Considering that '{directory}' is a local directory",
+                        cm.output
+                    )
+                    self.assertNotIn(
+                        f"DEBUG:guarddog:Considering that '{directory}' is a local file",
+                        cm.output
+                    )
+                    self.assertIn(
+                        f"DEBUG:guarddog:Considering that '{directory}' is a remote target",
+                        cm.output
+                    )
+
+    def _test_local_file_template(self, filename: str):
+        # `filename` is a file
+        with mock.patch("os.path.isdir") as isdir:
+            isdir.return_value = False
+            with mock.patch("os.path.isfile") as isfile:
+                isfile.return_value = True
+                with mock.patch.object(scanner.PackageScanner, 'scan_local', return_value={}) as _:
+                    with self.assertLogs("guarddog", level="DEBUG") as cm:
+                        guarddog.cli._scan(filename, "0.1.0", (), (), None, False, ECOSYSTEM.PYPI)
+                    self.assertNotIn(
+                        f"DEBUG:guarddog:Considering that '{filename}' is a local directory",
+                        cm.output
+                    )
+                    self.assertIn(
+                        f"DEBUG:guarddog:Considering that '{filename}' is a local file",
+                        cm.output
+                    )
+                    self.assertNotIn(
+                        f"DEBUG:guarddog:Considering that '{filename}' is a remote target",
+                        cm.output
+                    )
+
+        # `filename` is neither a directory nor a file
+        with mock.patch("os.path.isdir") as isdir:
+            isdir.return_value = False
+            with mock.patch("os.path.isfile") as isfile:
+                isfile.return_value = False
+                with mock.patch.object(scanner.PackageScanner, 'scan_local', return_value={}) as _:
+                    with self.assertLogs("guarddog", level="DEBUG") as cm:
+                        guarddog.cli._scan(filename, "0.1.0", (), (), None, False, ECOSYSTEM.PYPI)
+                    self.assertNotIn(
+                        f"DEBUG:guarddog:Considering that '{filename}' is a local directory",
+                        cm.output
+                    )
+                    self.assertNotIn(
+                        f"DEBUG:guarddog:Considering that '{filename}' is a local file",
+                        cm.output
+                    )
+                    self.assertIn(
+                        f"DEBUG:guarddog:Considering that '{filename}' is a remote target",
+                        cm.output
+                    )
+
+
+if __name__ == "__main__":
+    unittest.main()