Reintroduce preresolved dependencies optimization

pip_audit/__init__.py

@@ -2,4 +2,4 @@
 The `pip_audit` APIs.
 """
 
-__version__ = "2.5.5"
+__version__ = "2.5.6"

pip_audit/_dependency_source/__init__.py

@@ -2,7 +2,13 @@
 Dependency source interfaces and implementations for `pip-audit`.
 """
 
-from .interface import PYPI_URL, DependencyFixError, DependencySource, DependencySourceError
+from .interface import (
+    PYPI_URL,
+    DependencyFixError,
+    DependencySource,
+    DependencySourceError,
+    InvalidRequirementSpecifier,
+)
 from .pip import PipSource, PipSourceError
 from .pyproject import PyProjectSource
 from .requirement import RequirementSource
@@ -12,6 +18,7 @@
     "DependencyFixError",
     "DependencySource",
     "DependencySourceError",
+    "InvalidRequirementSpecifier",
     "PipSource",
     "PipSourceError",
     "PyProjectSource",

pip_audit/_dependency_source/interface.py

@@ -57,3 +57,11 @@ class DependencyFixError(Exception):
     """
 
     pass
+
+class InvalidRequirementSpecifier(DependencySourceError):
+    """
+    A `DependencySourceError` specialized for the case of a non-PEP 440 requirements
+    specifier.
+    """
+
+    pass

pip_audit/_dependency_source/requirement.py

@@ -15,12 +15,22 @@
 
 from packaging.specifiers import SpecifierSet
 from packaging.utils import canonicalize_name
-from pip_requirements_parser import InstallRequirement, InvalidRequirementLine, RequirementsFile
-
-from pip_audit._dependency_source import DependencyFixError, DependencySource, DependencySourceError
+from packaging.version import Version
+from pip_requirements_parser import (
+    InstallRequirement,
+    InvalidRequirementLine,
+    RequirementsFile,
+)
+
+from pip_audit._dependency_source import (
+    DependencyFixError,
+    DependencySource,
+    DependencySourceError,
+    InvalidRequirementSpecifier,
+)
 from pip_audit._fix import ResolvedFixVersion
 from pip_audit._service import Dependency
-from pip_audit._service.interface import ResolvedDependency
+from pip_audit._service.interface import ResolvedDependency, SkippedDependency
 from pip_audit._state import AuditState
 from pip_audit._virtual_env import VirtualEnv, VirtualEnvError
 
@@ -123,11 +133,30 @@ def collect(self) -> Iterator[Dependency]:
                 t.unlink()
 
     def _collect_from_files(self, filenames: list[Path]) -> Iterator[Dependency]:
+        # Figure out whether we have a fully resolved set of dependencies.
+        reqs: list[InstallRequirement] = []
+        require_hashes: bool = self._require_hashes
+        for filename in filenames:
+            rf = RequirementsFile.from_file(filename)
+            if len(rf.invalid_lines) > 0:
+                invalid = rf.invalid_lines[0]
+                raise InvalidRequirementSpecifier(
+                    f"requirement file {filename} contains invalid specifier at "
+                    f"line {invalid.line_number}: {invalid.error_message}"
+                )
+
+            # If one or more requirements have a hash, this implies `--require-hashes`.
+            require_hashes = require_hashes or any(req.hash_options for req in rf.requirements)
+            reqs.extend(rf.requirements)
+
+        # If the user has supplied `--no-deps` or there are hashed requirements, we should assume
+        # that we have a fully resolved set of dependencies and we should waste time by invoking
+        # `pip`.
+        if self._no_deps or require_hashes:
+            yield from self._collect_preresolved_deps(iter(reqs), require_hashes)
+            return
+
         ve_args = []
-        if self._no_deps:
-            ve_args.append("--no-deps")
-        if self._require_hashes:
-            ve_args.append("--require-hashes")
         for filename in filenames:
             ve_args.extend(["-r", str(filename)])
 
@@ -247,6 +276,62 @@ def _recover_files(self, tmp_files: list[IO[str]]) -> None:
                 continue
 
 
+    def _collect_preresolved_deps(
+        self, reqs: Iterator[InstallRequirement], require_hashes: bool
+    ) -> Iterator[Dependency]:
+        """
+        Collect pre-resolved (pinned) dependencies.
+        """
+        req_names: set[str] = set()
+        for req in reqs:
+            if not req.hash_options and require_hashes:
+                raise RequirementSourceError(f"requirement {req.dumps()} does not contain a hash")
+            if req.req is None:
+                # PEP 508-style URL requirements don't have a pre-declared version, even
+                # when hashed; the `#egg=name==version` syntax is non-standard and not supported
+                # by `pip` itself.
+                #
+                # In this case, we can't audit the dependency so we should signal to the
+                # caller that we're skipping it.
+                yield SkippedDependency(
+                    name=req.requirement_line.line,
+                    skip_reason="could not deduce package version from URL requirement",
+                )
+                continue
+            if self._skip_editable and req.is_editable:
+                yield SkippedDependency(name=req.name, skip_reason="requirement marked as editable")
+            if req.marker is not None and not req.marker.evaluate():
+                continue  # pragma: no cover
+
+            # This means we have a duplicate requirement for the same package
+            if req.name in req_names:
+                raise RequirementSourceError(
+                    f"package {req.name} has duplicate requirements: {str(req)}"
+                )
+            req_names.add(req.name)
+
+            # NOTE: URL dependencies cannot be pinned, so skipping them
+            # makes sense (under the same principle of skipping dependencies
+            # that can't be found on PyPI). This is also consistent with
+            # what `pip --no-deps` does (installs the URL dependency, but
+            # not any subdependencies).
+            if req.is_url:
+                yield SkippedDependency(
+                    name=req.name,
+                    skip_reason="URL requirements cannot be pinned to a specific package version",
+                )
+            elif not req.specifier:
+                raise RequirementSourceError(f"requirement {req.name} is not pinned: {str(req)}")
+            else:
+                pinned_specifier = PINNED_SPECIFIER_RE.match(str(req.specifier))
+                if pinned_specifier is None:
+                    raise RequirementSourceError(
+                        f"requirement {req.name} is not pinned to an exact version: {str(req)}"
+                    )
+
+                yield ResolvedDependency(req.name, Version(pinned_specifier.group("version")))
+
+
 class RequirementSourceError(DependencySourceError):
     """A requirements-parsing specific `DependencySourceError`."""
 

pip_audit/_format/cyclonedx.py

@@ -11,7 +11,6 @@
 from cyclonedx.model.bom import Bom
 from cyclonedx.model.component import Component
 from cyclonedx.model.vulnerability import Vulnerability
-from cyclonedx.parser import BaseParser
 
 import pip_audit._fix as fix
 import pip_audit._service as service
@@ -21,28 +20,28 @@
 logger = logging.getLogger(__name__)
 
 
-class _PipAuditResultParser(BaseParser):
-    def __init__(self, result: dict[service.Dependency, list[service.VulnerabilityResult]]):
-        super().__init__()
+def _pip_audit_result_to_bom(
+    result: dict[service.Dependency, list[service.VulnerabilityResult]]
+) -> Bom:
+    vulnerabilities = []
+    components = []
 
-        for dep, vulns in result.items():
-            # TODO(alex): Is there anything interesting we can do with skipped dependencies in
-            # the CycloneDX format?
-            if dep.is_skipped():
-                continue
-            dep = cast(service.ResolvedDependency, dep)
+    for dep, vulns in result.items():
+        # TODO(alex): Is there anything interesting we can do with skipped dependencies in
+        # the CycloneDX format?
+        if dep.is_skipped():
+            continue
+        dep = cast(service.ResolvedDependency, dep)
 
-            c = Component(name=dep.name, version=str(dep.version))
-            for vuln in vulns:
-                c.add_vulnerability(
-                    Vulnerability(
-                        id=vuln.id,
-                        description=vuln.description,
-                        recommendation="Upgrade",
-                    )
-                )
+        c = Component(name=dep.name, version=str(dep.version))
+        for vuln in vulns:
+            vulnerabilities.append(
+                Vulnerability(id=vuln.id, description=vuln.description, recommendation="Upgrade")
+            )
 
-            self._components.append(c)
+        components.append(c)
+
+    return Bom(components=components, vulnerabilities=vulnerabilities)
 
 
 class CycloneDxFormat(VulnerabilityFormat):
@@ -90,9 +89,7 @@ def format(
         if fixes:
             logger.warning("--fix output is unsupported by CycloneDX formats")
 
-        parser = _PipAuditResultParser(result)
-        bom = Bom.from_parser(parser)
-
+        bom = _pip_audit_result_to_bom(result)
         formatter = output.get_instance(
             bom=bom,
             output_format=self._inner_format.value,

test/data/requirements.txt

@@ -0,0 +1 @@
+flask==2.0.1

test/dependency_source/test_requirement.py

@@ -17,7 +17,7 @@
     requirement,
 )
 from pip_audit._fix import ResolvedFixVersion
-from pip_audit._service import ResolvedDependency
+from pip_audit._service import ResolvedDependency, SkippedDependency
 from pip_audit._state import AuditState
 from pip_audit._virtual_env import VirtualEnv, VirtualEnvError
 
@@ -433,9 +433,8 @@ def test_requirement_source_require_hashes_not_fully_resolved(req_file):
         require_hashes=True,
     )
 
-    with pytest.raises(DependencySourceError):
-        list(source.collect())
-
+    specs = list(source.collect())
+    assert specs == [ResolvedDependency("flask", Version("2.0.1"))]
 
 def test_requirement_source_require_hashes_missing(req_file):
     source = _init_requirement([(req_file(), "wheel==0.38.1")], require_hashes=True)
@@ -499,12 +498,20 @@ def test_requirement_source_require_hashes_incorrect_hash(req_file):
         list(source.collect())
 
 
-@pytest.mark.online
-def test_requirement_source_no_deps(req_file):
-    source = _init_requirement([(req_file(), "flask==2.0.1")], no_deps=True)
+def test_requirement_source_no_deps_editable_skip(req_file):
+    source = _init_requirement(
+        [(req_file(), "-e file:flask.py#egg=flask==2.0.1")], no_deps=True, skip_editable=True
+    )
 
     specs = list(source.collect())
-    assert specs == [ResolvedDependency("Flask", Version("2.0.1"))]
+    assert SkippedDependency(name="flask", skip_reason="requirement marked as editable") in specs
+
+
+def test_requirement_source_no_deps_duplicate_dependencies(req_file):
+    source = _init_requirement([(req_file(), "flask==1.0\nflask==1.0")], no_deps=True)
+
+    with pytest.raises(DependencySourceError):
+        list(source.collect())
 
 
 def test_requirement_source_no_double_open(monkeypatch, req_file):
@@ -706,3 +713,71 @@ def test_requirement_source_fix_invalid_lines(req_file):
                 dep=ResolvedDependency(name="flask", version=Version("0.5")), version=Version("1.0")
             )
         )
+
+
+def test_requirement_source_no_deps(req_file):
+    source = _init_requirement([(req_file(), "flask==2.0.1")], no_deps=True)
+
+    specs = list(source.collect())
+    assert specs == [ResolvedDependency("flask", Version("2.0.1"))]
+
+
+def test_requirement_source_no_deps_unpinned(req_file):
+    source = _init_requirement([(req_file(), "flask\nrequests==1.0")], no_deps=True)
+
+    # When dependency resolution is disabled, all requirements must be pinned.
+    with pytest.raises(DependencySourceError):
+        list(source.collect())
+
+
+def test_requirement_source_no_deps_not_exact_version(req_file):
+    source = _init_requirement([(req_file(), "flask==1.0\nrequests>=1.0")], no_deps=True)
+
+    # When dependency resolution is disabled, all requirements must be pinned.
+    with pytest.raises(DependencySourceError):
+        list(source.collect())
+
+
+def test_requirement_source_no_deps_unpinned_url(req_file):
+    source = _init_requirement(
+        [
+            (
+                req_file(),
+                "https://github.com/pallets/flask/archive/refs/tags/2.0.1.tar.gz#egg=flask\n",
+            )
+        ],
+        no_deps=True,
+    )
+
+    assert list(source.collect()) == [
+        SkippedDependency(
+            name="flask",
+            skip_reason="URL requirements cannot be pinned to a specific package version",
+        )
+    ]
+
+
+def test_requirement_source_no_deps_editable_with_egg_fragment(req_file):
+    source = _init_requirement([(req_file(), "-e file:flask.py#egg=flask==2.0.1")], no_deps=True)
+
+    specs = list(source.collect())
+    assert (
+        SkippedDependency(
+            name="flask",
+            skip_reason="URL requirements cannot be pinned to a specific package version",
+        )
+        in specs
+    )
+
+
+def test_requirement_source_no_deps_editable_without_egg_fragment(req_file):
+    source = _init_requirement([(req_file(), "-e file:flask.py")], no_deps=True)
+
+    specs = list(source.collect())
+    assert (
+        SkippedDependency(
+            name="-e file:flask.py",
+            skip_reason="could not deduce package version from URL requirement",
+        )
+        in specs
+    )