Fixes

Author: spoorcc

.github/workflows/release.yml

@@ -50,7 +50,7 @@ jobs:
 
       - name: Update latest tag
         if: ${{ steps.release_info.outputs.tag == 'latest' }}
-        uses: EndBug/latest-tag@fabb56bc8d15d5937c76719060da2226f5c3ffa8 
+        uses: EndBug/latest-tag@fabb56bc8d15d5937c76719060da2226f5c3ffa8
         with:
           ref: latest
           description: Last state in main

dfetch/manifest/manifest.py

@@ -97,7 +97,9 @@ class ManifestDict(TypedDict, total=True):  # pylint: disable=too-many-ancestors
 
     version: int | str
     remotes: NotRequired[Sequence[RemoteDict | Remote]]
-    projects: Sequence[ProjectEntryDict | ProjectEntry | dict[str, str | list[str]]]
+    projects: Sequence[
+        ProjectEntryDict | ProjectEntry | dict[str, str | list[str] | dict[str, str]]
+    ]
 
 
 class Manifest:
@@ -140,14 +142,16 @@ def __init__(
     def _init_projects(
         self,
         projects: Sequence[
-            ProjectEntryDict | ProjectEntry | dict[str, str | list[str]]
+            ProjectEntryDict
+            | ProjectEntry
+            | dict[str, str | list[str] | dict[str, str]]
         ],
     ) -> dict[str, ProjectEntry]:
         """Iterate over projects from manifest and initialize ProjectEntries from it.
 
         Args:
             projects (Sequence[
-                Union[ProjectEntryDict, ProjectEntry, Dict[str, Union[str, list[str]]]]
+                Union[ProjectEntryDict, ProjectEntry, Dict[str, Union[str, list[str], dict[str, str]]]]
             ]): Iterable with projects
 
         Raises:
@@ -304,9 +308,11 @@ def _as_dict(self) -> dict[str, ManifestDict]:
         if len(remotes) == 1:
             remotes[0].pop("default", None)
 
-        projects: list[dict[str, str | list[str]]] = []
+        projects: list[dict[str, str | list[str] | dict[str, str]]] = []
         for project in self.projects:
-            project_yaml: dict[str, str | list[str]] = project.as_yaml()
+            project_yaml: dict[str, str | list[str] | dict[str, str]] = (
+                project.as_yaml()
+            )
             if len(remotes) == 1:
                 project_yaml.pop("remote", None)
             projects.append(project_yaml)

dfetch/manifest/project.py

@@ -370,7 +370,7 @@ def as_yaml(self) -> dict[str, str]:
         "repo-path": str,
         "vcs": str,
         "ignore": Sequence[str],
-        "integrity": dict,
+        "integrity": dict[str, str],
         "default_remote": str,
     },
     total=False,
@@ -398,7 +398,7 @@ def __init__(self, kwargs: ProjectEntryDict) -> None:
         self._tag: str = kwargs.get("tag", "")
         self._vcs: str = kwargs.get("vcs", "")
         self._ignore: Sequence[str] = kwargs.get("ignore", [])
-        integrity_data: dict = kwargs.get("integrity", {})
+        integrity_data: dict[str, str] = kwargs.get("integrity", {})
         self._integrity = Integrity(hash=integrity_data.get("hash", ""))
 
         if not self._remote and not self._url:
@@ -407,7 +407,7 @@ def __init__(self, kwargs: ProjectEntryDict) -> None:
     @classmethod
     def from_yaml(
         cls,
-        yamldata: dict[str, str | list[str]] | ProjectEntryDict,
+        yamldata: dict[str, str | list[str] | dict[str, str]] | ProjectEntryDict,
         default_remote: str = "",
     ) -> "ProjectEntry":
         """Create a Project Entry from yaml data.

dfetch/project/__init__.py

@@ -15,7 +15,9 @@
 from dfetch.project.svnsuperproject import SvnSuperProject
 from dfetch.util.util import resolve_absolute_path
 
-SUPPORTED_SUBPROJECT_TYPES = [ArchiveSubProject, GitSubProject, SvnSubProject]
+SUPPORTED_SUBPROJECT_TYPES: list[
+    type[ArchiveSubProject] | type[GitSubProject] | type[SvnSubProject]
+] = [ArchiveSubProject, GitSubProject, SvnSubProject]
 SUPPORTED_SUPERPROJECT_TYPES = [GitSuperProject, SvnSuperProject]
 
 logger = get_logger(__name__)

dfetch/project/archivesubproject.py

@@ -44,17 +44,22 @@
 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 (
     SUPPORTED_HASH_ALGORITHMS,
     ArchiveLocalRepo,
     ArchiveRemote,
-    _safe_compare_hex,  # private helper, intentionally imported for internal use
-    _suffix_for_url,    # private helper, intentionally imported for internal use
     compute_hash,
     is_archive_url,
 )
@@ -94,8 +99,6 @@ def revision_is_enough() -> bool:
     @staticmethod
     def list_tool_info() -> None:
         """Log information about the archive fetching tool (Python's urllib)."""
-        import urllib.request as _ur  # noqa: PLC0415
-
         SubProject._log_tool("urllib", _ur.__doc__ or "built-in")
 
     def get_default_branch(self) -> str:

dfetch/project/subproject.py

@@ -409,7 +409,9 @@ def freeze_project(self, project: ProjectEntry) -> str | None:
             return None
         if on_disk_version:
             project.version = on_disk_version
-            return on_disk_version.revision or on_disk_version.tag or str(on_disk_version)
+            return (
+                on_disk_version.revision or on_disk_version.tag or str(on_disk_version)
+            )
         return None
 
     @staticmethod

dfetch/reporting/sbom_reporter.py

@@ -109,11 +109,11 @@
 from cyclonedx.schema import OutputFormat, SchemaVersion
 
 import dfetch.util.purl
-from dfetch.util.purl import DFETCH_TO_CDX_HASH_ALGORITHM
 from dfetch.manifest.manifest import Manifest
 from dfetch.manifest.project import ProjectEntry
 from dfetch.reporting.reporter import Reporter
 from dfetch.util.license import License
+from dfetch.util.purl import DFETCH_TO_CDX_HASH_ALGORITHM
 
 # PyRight is pedantic with decorators see https://github.com/madpah/serializable/issues/8
 # It might be fixable with https://github.com/microsoft/pyright/discussions/4426, would prefer

dfetch/vcs/archive.py

@@ -31,23 +31,21 @@
 import tarfile
 import tempfile
 import urllib.error
+import urllib.parse
 import urllib.request
 import zipfile
 from collections.abc import Sequence
 
+from dfetch.log import get_logger
+from dfetch.project.subproject import SubProject
+from dfetch.util.util import find_matching_files, safe_rm
+
 #: Archive file extensions recognised by DFetch.
-#: Defined before any intra-package imports to avoid partial-initialisation
-#: issues when other modules (e.g. dfetch.util.purl) import this symbol while
-#: the module is still being initialised.
 ARCHIVE_EXTENSIONS = (".tar.gz", ".tgz", ".tar.bz2", ".tar.xz", ".zip")
 
 #: Hash algorithms supported by the ``integrity.hash`` manifest field.
 SUPPORTED_HASH_ALGORITHMS = ("sha256",)
 
-from dfetch.log import get_logger  # noqa: E402
-from dfetch.project.subproject import SubProject  # noqa: E402
-from dfetch.util.util import find_matching_files, safe_rm  # noqa: E402
-
 logger = get_logger(__name__)
 
 # Safety limits applied during extraction to prevent decompression bombs.
@@ -112,12 +110,15 @@ def is_accessible(self) -> bool:
         * ``http``/``https`` URLs first try a ``HEAD`` request.  If the server
           rejects it (405/501) a partial ``GET`` (``Range: bytes=0-0``) is
           attempted instead.  Returns *False* on any final failure.
+        * Any other URL scheme returns *False*.
         """
-        from urllib.parse import urlparse as _urlparse  # noqa: PLC0415
+        parsed = urllib.parse.urlparse(self.url)
+
+        if parsed.scheme == "file":
+            return os.path.exists(parsed.path)
 
-        if _urlparse(self.url).scheme == "file":
-            path = _urlparse(self.url).path
-            return os.path.exists(path)
+        if parsed.scheme not in ("http", "https"):
+            return False
 
         for method, headers in [
             ("HEAD", {}),
@@ -142,14 +143,27 @@ def download(self, dest_path: str) -> None:
             dest_path: Local file path to write the archive to.
 
         Raises:
-            RuntimeError: On download failure.
+            RuntimeError: On download failure or unsupported URL scheme.
         """
-        try:
-            urllib.request.urlretrieve(self.url, dest_path)
-        except (urllib.error.URLError, OSError) as exc:
+        parsed = urllib.parse.urlparse(self.url)
+        if parsed.scheme == "file":
+            try:
+                shutil.copy(parsed.path, dest_path)
+            except OSError as exc:
+                raise RuntimeError(
+                    f"'{self.url}' is not a valid URL or unreachable: {exc}"
+                ) from exc
+        elif parsed.scheme in ("http", "https"):
+            try:
+                urllib.request.urlretrieve(self.url, dest_path)
+            except (urllib.error.URLError, OSError) as exc:
+                raise RuntimeError(
+                    f"'{self.url}' is not a valid URL or unreachable: {exc}"
+                ) from exc
+        else:
             raise RuntimeError(
-                f"'{self.url}' is not a valid URL or unreachable: {exc}"
-            ) from exc
+                f"'{self.url}' uses unsupported scheme '{parsed.scheme}'."
+            )
 
 
 class ArchiveLocalRepo:
@@ -223,9 +237,13 @@ def _check_archive_limits(member_count: int, total_bytes: int) -> None:
             )
 
     @staticmethod
-    def _check_zip_members(zf: zipfile.ZipFile) -> None:
+    def check_zip_members(zf: zipfile.ZipFile) -> list[zipfile.ZipInfo]:
         """Validate all ZIP member paths against path-traversal attacks.
 
+        Returns:
+            The validated list of members, safe to pass to
+            :meth:`zipfile.ZipFile.extract`.
+
         Raises:
             RuntimeError: When any member contains an absolute path, a ``..``
                 component, or when the archive exceeds the size/count limits.
@@ -242,6 +260,7 @@ def _check_zip_members(zf: zipfile.ZipFile) -> None:
                 raise RuntimeError(
                     f"Archive contains an unsafe member path: {info.filename!r}"
                 )
+        return members
 
     @staticmethod
     def _check_tar_members(tf: tarfile.TarFile) -> None:
@@ -289,13 +308,13 @@ def _extract_raw(archive_path: str, dest_dir: str) -> None:
             with tarfile.open(archive_path, "r:*") as tf:
                 ArchiveLocalRepo._check_tar_members(tf)
                 if sys.version_info >= (3, 11, 4):
-                    tf.extractall(dest_dir, filter="tar")
+                    tf.extractall(dest_dir, filter="tar")  # nosec B202
                 else:
-                    tf.extractall(dest_dir)  # noqa: S202
+                    tf.extractall(dest_dir)  # nosec B202
         elif lower.endswith(".zip") or zipfile.is_zipfile(archive_path):
             with zipfile.ZipFile(archive_path) as zf:
                 ArchiveLocalRepo._check_zip_members(zf)
-                zf.extractall(dest_dir)
+                zf.extractall(dest_dir)  # nosec B202
         else:
             raise RuntimeError(
                 f"Unsupported archive format: '{archive_path}'. "
@@ -319,6 +338,8 @@ def _copy_with_src(
                     shutil.copy2(s, d)
         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 archive")
 
         if keep_licenses:
             for item in os.listdir(extract_root):

tests/test_archive.py

@@ -37,7 +37,10 @@ def test_compute_hash_empty_file():
     try:
         digest = compute_hash(path, "sha256")
         # SHA-256 of empty string
-        assert digest == "e3b0c44298fc1c149afbf4c8996fb924" "27ae41e4649b934ca495991b7852b855"
+        assert (
+            digest == "e3b0c44298fc1c149afbf4c8996fb924"
+            "27ae41e4649b934ca495991b7852b855"
+        )
     finally:
         os.remove(path)
 

tests/test_integrity.py

@@ -2,7 +2,6 @@
 
 from dfetch.manifest.project import Integrity, ProjectEntry
 
-
 # ---------------------------------------------------------------------------
 # Integrity dataclass
 # ---------------------------------------------------------------------------
@@ -54,7 +53,14 @@ def test_projectentry_integrity_falsy_without_hash():
 
 def test_projectentry_as_yaml_includes_integrity():
     h = "sha256:" + "d" * 64
-    project = ProjectEntry({"name": "lib", "url": "https://example.com/lib.tar.gz", "vcs": "archive", "integrity": {"hash": h}})
+    project = ProjectEntry(
+        {
+            "name": "lib",
+            "url": "https://example.com/lib.tar.gz",
+            "vcs": "archive",
+            "integrity": {"hash": h},
+        }
+    )
     yaml_data = project.as_yaml()
     assert yaml_data["integrity"] == {"hash": h}
 
@@ -66,7 +72,9 @@ def test_projectentry_as_yaml_omits_empty_integrity():
 
 
 def test_projectentry_hash_setter():
-    project = ProjectEntry({"name": "lib", "url": "https://example.com/lib.tar.gz", "vcs": "archive"})
+    project = ProjectEntry(
+        {"name": "lib", "url": "https://example.com/lib.tar.gz", "vcs": "archive"}
+    )
     h = "sha256:" + "e" * 64
     project.hash = h
     assert project.hash == h