Cleanup implementation

Author: spoorcc

dfetch/commands/report.py

@@ -14,9 +14,9 @@
 from dfetch.manifest.project import ProjectEntry
 from dfetch.project import create_super_project
 from dfetch.project.metadata import Metadata
-from dfetch.project.subproject import SubProject
 from dfetch.reporting import REPORTERS, ReportTypes
 from dfetch.util.license import License, guess_license_in_file
+from dfetch.util.util import is_license_file
 
 logger = get_logger(__name__)
 
@@ -89,7 +89,7 @@ def _determine_licenses(project: ProjectEntry) -> list[License]:
 
         license_files = []
         with dfetch.util.util.in_directory(project.destination):
-            for license_file in filter(SubProject.is_license_file, glob.glob("*")):
+            for license_file in filter(is_license_file, glob.glob("*")):
                 logger.debug(f"Found license file {license_file} for {project.name}")
                 guessed_license = guess_license_in_file(license_file)
 

dfetch/manifest/project.py

@@ -336,8 +336,8 @@ class Integrity:
     Holds the ``hash:`` sub-field today and is designed to accommodate
     future signature-verification fields:
 
-    * ``sig`` – URL of a detached signature file (``.sig`` / ``.asc``).
-    * ``sig_key`` – URL or fingerprint of the signing key (``.p7s`` / ``.gpg``).
+    * ``sig`` - URL of a detached signature file (``.sig`` / ``.asc``).
+    * ``sig_key`` - URL or fingerprint of the signing key (``.p7s`` / ``.gpg``).
     """
 
     hash: str = field(default="")

dfetch/project/archivesubproject.py

@@ -2,15 +2,15 @@
 
 Archives are a third VCS type alongside ``git`` and ``svn``.  They represent
 versioned dependencies that are distributed as ``.tar.gz``, ``.tgz``,
-``.tar.bz2``, ``.tar.xz`` or ``.zip`` files reachable via any URL that Python's
-:mod:`urllib.request` understands (``http://``, ``https://``, ``file://``, …).
+``.tar.bz2``, ``.tar.xz`` or ``.zip`` files reachable via ``http://``,
+``https://``, or ``file://`` URLs.
 
 Unlike git and SVN, archives have no inherent "branching" or "tagging"
 concept.  Version identity is expressed through:
 
-* **No hash** – the URL itself acts as the identity.  The archive is
+* **No hash** - the URL itself acts as the identity.  The archive is
   considered up-to-date as long as the same URL is still reachable.
-* **``integrity.hash: <algorithm>:<hex>``** – the cryptographic hash of the
+* **``integrity.hash: <algorithm>:<hex>``** - the cryptographic hash of the
   archive file acts as the version identifier.  The fetch step verifies the
   downloaded archive against this hash and raises an error on mismatch.
 
@@ -41,22 +41,18 @@
 
 from __future__ import annotations
 
+import hmac
+import http.client
 import os
 import pathlib
 import tempfile
-import urllib.request as _ur
 
 from dfetch.log import get_logger
 from dfetch.manifest.project import ProjectEntry
 from dfetch.manifest.version import Version
 from dfetch.project.subproject import SubProject
 from dfetch.vcs.archive import (
-    _safe_compare_hex,  # private helper, intentionally imported for internal use
-)
-from dfetch.vcs.archive import (
-    _suffix_for_url,  # private helper, intentionally imported for internal use
-)
-from dfetch.vcs.archive import (
+    ARCHIVE_EXTENSIONS,
     SUPPORTED_HASH_ALGORITHMS,
     ArchiveLocalRepo,
     ArchiveRemote,
@@ -67,6 +63,24 @@
 logger = get_logger(__name__)
 
 
+def _safe_compare_hex(actual: str, expected: str) -> bool:
+    """Constant-time comparison of two hex digest strings.
+
+    Uses :func:`hmac.compare_digest` to avoid leaking timing information about
+    the expected hash value.
+    """
+    return hmac.compare_digest(actual.lower(), expected.lower())
+
+
+def _suffix_for_url(url: str) -> str:
+    """Return the archive file suffix for *url* (e.g. ``'.tar.gz'``, ``'.zip'``)."""
+    lower = url.lower()
+    for ext in sorted(ARCHIVE_EXTENSIONS, key=len, reverse=True):
+        if lower.endswith(ext):
+            return ext
+    return ".archive"
+
+
 class ArchiveSubProject(SubProject):
     """A project fetched from a tar/zip archive URL.
 
@@ -83,10 +97,6 @@ def __init__(self, project: ProjectEntry) -> None:
         self._project_entry = project
         self._remote_repo = ArchiveRemote(project.remote_url)
 
-    # ------------------------------------------------------------------
-    # SubProject abstract interface
-    # ------------------------------------------------------------------
-
     def check(self) -> bool:
         """Return *True* when the project URL looks like an archive."""
         return is_archive_url(self.remote)
@@ -98,16 +108,16 @@ def revision_is_enough() -> bool:
 
     @staticmethod
     def list_tool_info() -> None:
-        """Log information about the archive fetching tool (Python's urllib)."""
-        SubProject._log_tool("urllib", _ur.__doc__ or "built-in")
+        """Log information about the archive fetching tool (Python's http.client)."""
+        SubProject._log_tool("http.client", http.client.__doc__ or "built-in")
 
     def get_default_branch(self) -> str:
         """Archives have no branches; return an empty string."""
         return ""
 
     def _latest_revision_on_branch(self, branch: str) -> str:  # noqa: ARG002
         """For archives the 'latest revision' is always the URL (or hash)."""
-        return self._project_entry.remote_url
+        return self.remote
 
     def _download_and_compute_hash(self, algorithm: str = "sha256") -> str:
         """Download the archive to a temporary file and return its hash.
@@ -117,20 +127,16 @@ def _download_and_compute_hash(self, algorithm: str = "sha256") -> str:
         Raises:
             RuntimeError: On download failure or unsupported algorithm.
         """
-        tmp_path: str | None = None
+        fd, tmp_path = tempfile.mkstemp(suffix=_suffix_for_url(self.remote))
+        os.close(fd)
         try:
-            with tempfile.NamedTemporaryFile(
-                suffix=_suffix_for_url(self._project_entry.remote_url), delete=False
-            ) as tmp:
-                tmp_path = tmp.name
             self._remote_repo.download(tmp_path)
             return compute_hash(tmp_path, algorithm)
         finally:
-            if tmp_path:
-                try:
-                    os.remove(tmp_path)
-                except OSError:
-                    pass
+            try:
+                os.remove(tmp_path)
+            except OSError:
+                pass
 
     def _does_revision_exist(self, revision: str) -> bool:
         """Check whether *revision* (a hash or URL string) is still valid.
@@ -151,17 +157,13 @@ def _does_revision_exist(self, revision: str) -> bool:
                 except RuntimeError:
                     return False
 
-        # revision is the URL – just check accessibility
+        # revision is the URL - just check accessibility
         return self._remote_repo.is_accessible()
 
     def _list_of_tags(self) -> list[str]:
         """Archives have no tags; returns an empty list."""
         return []
 
-    # ------------------------------------------------------------------
-    # Version overrides
-    # ------------------------------------------------------------------
-
     @property
     def wanted_version(self) -> Version:
         """Version derived from the ``integrity.hash`` field or the archive URL.
@@ -174,11 +176,7 @@ def wanted_version(self) -> Version:
         """
         if self._project_entry.hash:
             return Version(revision=self._project_entry.hash)
-        return Version(revision=self._project_entry.remote_url)
-
-    # ------------------------------------------------------------------
-    # Fetch
-    # ------------------------------------------------------------------
+        return Version(revision=self.remote)
 
     def _fetch_impl(self, version: Version) -> Version:
         """Download and extract the archive to the local destination.
@@ -193,15 +191,12 @@ def _fetch_impl(self, version: Version) -> Version:
         Returns:
             The version that was actually fetched (hash string or URL).
         """
-        url = self._project_entry.remote_url
         expected_hash = self._project_entry.hash
 
         pathlib.Path(self.local_path).mkdir(parents=True, exist_ok=True)
 
-        suffix = _suffix_for_url(url)
-        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
-            tmp_path = tmp.name
-
+        fd, tmp_path = tempfile.mkstemp(suffix=_suffix_for_url(self.remote))
+        os.close(fd)
         try:
             self._remote_repo.download(tmp_path)
 
@@ -231,13 +226,7 @@ def _fetch_impl(self, version: Version) -> Version:
             except OSError:
                 pass
 
-        if expected_hash:
-            return Version(revision=expected_hash)
-        return Version(revision=url)
-
-    # ------------------------------------------------------------------
-    # Freeze support
-    # ------------------------------------------------------------------
+        return Version(revision=expected_hash if expected_hash else self.remote)
 
     def freeze_project(self, project: ProjectEntry) -> str | None:
         """Pin *project* to a cryptographic hash of the archive.
@@ -264,7 +253,7 @@ def freeze_project(self, project: ProjectEntry) -> str | None:
 
         revision = on_disk.revision
 
-        # Already hash-pinned – revision is "sha256:<hex>"
+        # Already hash-pinned - revision is "sha256:<hex>"
         if revision.startswith(tuple(f"{a}:" for a in SUPPORTED_HASH_ALGORITHMS)):
             if project.hash == revision:
                 return None

dfetch/project/gitsubproject.py

@@ -8,7 +8,7 @@
 from dfetch.manifest.project import ProjectEntry
 from dfetch.manifest.version import Version
 from dfetch.project.subproject import SubProject
-from dfetch.util.util import safe_rmtree
+from dfetch.util.util import LICENSE_GLOBS, safe_rmtree
 from dfetch.vcs.git import GitLocalRepo, GitRemote, get_git_version
 
 logger = get_logger(__name__)
@@ -64,8 +64,8 @@ def _fetch_impl(self, version: Version) -> Version:
         # When exporting a file, the destination directory must already exist
         pathlib.Path(self.local_path).mkdir(parents=True, exist_ok=True)
 
-        license_globs = [f"/{name.lower()}" for name in self.LICENSE_GLOBS] + [
-            f"/{name.upper()}" for name in self.LICENSE_GLOBS
+        license_globs = [f"/{name.lower()}" for name in LICENSE_GLOBS] + [
+            f"/{name.upper()}" for name in LICENSE_GLOBS
         ]
 
         local_repo = GitLocalRepo(self.local_path)

dfetch/project/subproject.py

@@ -1,6 +1,5 @@
 """SubProject."""
 
-import fnmatch
 import os
 import pathlib
 from abc import ABC, abstractmethod
@@ -26,7 +25,6 @@ class SubProject(ABC):
     """
 
     NAME = ""
-    LICENSE_GLOBS = ["licen[cs]e*", "copying*", "copyright*"]
 
     def __init__(self, project: ProjectEntry) -> None:
         """Create the subproject."""
@@ -413,11 +411,3 @@ def freeze_project(self, project: ProjectEntry) -> str | None:
                 on_disk_version.revision or on_disk_version.tag or str(on_disk_version)
             )
         return None
-
-    @staticmethod
-    def is_license_file(filename: str) -> bool:
-        """Check if the given filename is a license file."""
-        return any(
-            fnmatch.fnmatch(filename.lower(), pattern)
-            for pattern in SubProject.LICENSE_GLOBS
-        )

dfetch/project/svnsubproject.py

@@ -11,6 +11,7 @@
 from dfetch.util.util import (
     find_matching_files,
     find_non_matching_files,
+    is_license_file,
     safe_rm,
 )
 from dfetch.vcs.svn import SvnRemote, SvnRepo, get_svn_version
@@ -103,7 +104,7 @@ def _determine_what_to_fetch(self, version: Version) -> tuple[str, str, str]:
     def _remove_ignored_files(self) -> None:
         """Remove any ignored files, whilst keeping license files."""
         for file_or_dir in find_matching_files(self.local_path, self.ignore):
-            if not (file_or_dir.is_file() and self.is_license_file(file_or_dir.name)):
+            if not (file_or_dir.is_file() and is_license_file(file_or_dir.name)):
                 safe_rm(file_or_dir)
 
     def _fetch_impl(self, version: Version) -> Version:
@@ -168,9 +169,7 @@ def _get_info(self, branch: str) -> dict[str, str]:
     def _license_files(url_path: str) -> list[str]:
         return [
             str(license)
-            for license in filter(
-                SvnSubProject.is_license_file, SvnRepo.files_in_path(url_path)
-            )
+            for license in filter(is_license_file, SvnRepo.files_in_path(url_path))
         ]
 
     def _get_revision(self, branch: str) -> str:

dfetch/util/util.py

@@ -12,6 +12,72 @@
 
 from _hashlib import HASH
 
+#: Glob patterns used to identify license files by filename.
+LICENSE_GLOBS = ["licen[cs]e*", "copying*", "copyright*"]
+
+
+def is_license_file(filename: str) -> bool:
+    """Return *True* when *filename* matches a known license file pattern."""
+    return any(fnmatch.fnmatch(filename.lower(), pattern) for pattern in LICENSE_GLOBS)
+
+
+def _copy_entry(src_entry: str, dest_entry: str) -> None:
+    """Copy a single file or directory *src_entry* to *dest_entry*."""
+    if os.path.isdir(src_entry):
+        shutil.copytree(src_entry, dest_entry)
+    else:
+        shutil.copy2(src_entry, dest_entry)
+
+
+def copy_directory_contents(src_dir: str, dest_dir: str) -> None:
+    """Copy every entry in *src_dir* directly into *dest_dir*.
+
+    Directories are copied recursively; files are copied with metadata.
+    """
+    for entry_name in os.listdir(src_dir):
+        _copy_entry(
+            os.path.join(src_dir, entry_name),
+            os.path.join(dest_dir, entry_name),
+        )
+
+
+def copy_src_subset(
+    src_root: str, dest_dir: str, src: str, keep_licenses: bool
+) -> None:
+    """Copy a *src* sub-path from *src_root* into *dest_dir*.
+
+    When *src* is a directory, its contents are copied flat into *dest_dir*.
+    When *src* is a single file, that file is copied into *dest_dir*.
+    If *keep_licenses* is ``True``, any license files found directly in
+    *src_root* are also copied regardless of the *src* filter.
+
+    Raises:
+        RuntimeError: When *src* does not exist inside *src_root*.
+    """
+    src_path = os.path.join(src_root, src)
+    if os.path.isdir(src_path):
+        copy_directory_contents(src_path, dest_dir)
+    elif os.path.isfile(src_path):
+        shutil.copy2(src_path, os.path.join(dest_dir, os.path.basename(src_path)))
+    else:
+        raise RuntimeError(f"src {src!r} was not found in the extracted archive")
+
+    if keep_licenses:
+        for entry_name in os.listdir(src_root):
+            full_path = os.path.join(src_root, entry_name)
+            if os.path.isfile(full_path) and is_license_file(entry_name):
+                shutil.copy2(full_path, os.path.join(dest_dir, entry_name))
+
+
+def prune_files_by_pattern(directory: str, patterns: Sequence[str]) -> None:
+    """Remove files and directories in *directory* matching *patterns*.
+
+    License files are never removed even when they match a pattern.
+    """
+    for file_or_dir in find_matching_files(directory, patterns):
+        if not (file_or_dir.is_file() and is_license_file(file_or_dir.name)):
+            safe_rm(file_or_dir)
+
 
 def _remove_readonly(func: Any, path: str, _: Any) -> None:
     if not os.access(path, os.W_OK):