requirement: Close temporary files before passing them to pip (#551)

* requirement: Close temporary files before passing them to `pip`

* test: Add unit test

* requirement: Fix lint

* CHANGELOG: record changes

Signed-off-by: William Woodruff <[email protected]>

---------

Signed-off-by: William Woodruff <[email protected]>
Co-authored-by: William Woodruff <[email protected]>

CHANGELOG.md

@@ -8,6 +8,11 @@ All versions prior to 0.0.9 are untracked.
 
 ## [Unreleased]
 
+### Fixed
+
+* Fixed a crash on Windows caused by multiple open file handles to
+  input requirements ([#551](https://github.com/pypa/pip-audit/pull/551))
+
 ## [2.5.0]
 
 ### Changed

pip_audit/_dependency_source/requirement.py

@@ -84,7 +84,8 @@ def collect(self) -> Iterator[Dependency]:
         Raises a `RequirementSourceError` on any errors.
         """
 
-        with ExitStack() as stack:
+        tmp_files = []
+        try:
             # We need to handle process substitution inputs so we can invoke
             # `pip-audit` like so:
             #
@@ -97,20 +98,26 @@ def collect(self) -> Iterator[Dependency]:
             # In order to get around this, we're going to copy each input into a
             # a corresponding temporary file and then pass that set of files
             # into `pip`.
-            tmp_files = []
-
-            # For each input file, copy it to one of our temporary files.
-            # Ensure we flush so our writes are visible to `pip`.
             for filename in self._filenames:
-                tmp_file = stack.enter_context(NamedTemporaryFile(mode="w"))
+                # Deliberately pass `delete=False` so that our temporary file doesn't get
+                # automatically deleted on close. We need to close it so that `pip` can
+                # use it however, we obviously want it to persist.
+                tmp_file = NamedTemporaryFile(mode="w", delete=False)
                 with filename.open("r") as f:
                     shutil.copyfileobj(f, tmp_file)
-                    tmp_file.flush()
-                tmp_files.append(tmp_file)
+
+                # Close the file since it's going to get re-opened by `pip`
+                tmp_file.close()
+                tmp_files.append(tmp_file.name)
 
             # Now pass the list of temporary filenames into the rest of our
             # logic.
-            yield from self._collect_from_files([Path(f.name) for f in tmp_files])
+            yield from self._collect_from_files([Path(f) for f in tmp_files])
+        finally:
+            # Since we disabled automatically deletion for these temporary files, we need to
+            # manually delete them on the way out.
+            for t in tmp_files:
+                os.unlink(t)
 
     def _collect_from_files(self, filenames: list[os.PathLike]) -> Iterator[Dependency]:
         # Figure out whether we have a fully resolved set of dependencies.

test/dependency_source/test_requirement.py

@@ -3,6 +3,7 @@
 import os
 from email.message import EmailMessage
 from pathlib import Path
+from tempfile import NamedTemporaryFile
 
 import pretend  # type: ignore
 import pytest
@@ -17,7 +18,7 @@
 from pip_audit._fix import ResolvedFixVersion
 from pip_audit._service import ResolvedDependency, SkippedDependency
 from pip_audit._state import AuditState
-from pip_audit._virtual_env import VirtualEnvError
+from pip_audit._virtual_env import VirtualEnv, VirtualEnvError
 
 
 def get_metadata_mock():
@@ -483,6 +484,38 @@ def test_requirement_source_no_deps_duplicate_dependencies(req_file):
         list(source.collect())
 
 
+def test_requirement_source_no_double_open(monkeypatch, req_file):
+    source = _init_requirement([(req_file(), "flask==2.0.1")])
+
+    # Intercept the calls to `NamedTemporaryFile` to get a handle on each file object.
+    tmp_files = []
+
+    def named_temp_file(*args, **kwargs):
+        tmp_file = NamedTemporaryFile(*args, **kwargs)
+        tmp_files.append(tmp_file)
+        return tmp_file
+
+    monkeypatch.setattr(
+        requirement,
+        "NamedTemporaryFile",
+        named_temp_file,
+    )
+
+    # Intercept the `VirtualEnv` constructor to check that all file handles are closed prior to
+    # the `pip` invocation.
+    #
+    # `pip` will open the file so we need to ensure that we've closed it.
+    def virtual_env(*args, **kwargs):
+        for tmp_file in tmp_files:
+            assert tmp_file.closed
+        return VirtualEnv(*args, **kwargs)
+
+    monkeypatch.setattr(requirement, "VirtualEnv", virtual_env)
+
+    specs = list(source.collect())
+    assert ResolvedDependency("Flask", Version("2.0.1")) in specs
+
+
 def test_requirement_source_fix_explicit_subdep(monkeypatch, req_file):
     logger = pretend.stub(warning=pretend.call_recorder(lambda s: None))
     monkeypatch.setattr(requirement, "logger", logger)