Verify packages index with best checksum algorithm

backend/common/repo.py

@@ -9,6 +9,7 @@
 import lzma
 from urllib import parse
 import hashlib
+from collections import namedtuple
 
 import requests
 
@@ -42,6 +43,8 @@ class Checksum:  # pylint: disable=R0903
             md5: str = ""
             sha1: str = ""
             sha256: str = ""
+            sha384: str = ""
+            sha512: str = ""
 
         def __init__(self, size: int, uri: str):
             self.checksum = DpkgRepo.ReleaseEntry.Checksum()
@@ -136,22 +139,41 @@ def _parse_release_index(self, release: str) -> "EntryDict":
         :param release: decoded content of the Release file
         :return: dictionary
         """
+        # Length of hexadecimal representation for each checksum algorithm
+        LEN_MD5 = 128 // 4
+        LEN_SHA1 = 160 // 4
+        LEN_SHA256 = 256 // 4
+        LEN_SHA384 = 384 // 4
+        LEN_SHA512 = 512 // 4
+        Entry = namedtuple("Entry", "checksum, size, path")
         for line in release.split(os.linesep):
-            cs_s_path = tuple(filter(None, line.strip().replace("\t", " ").split(" ")))
-            if len(cs_s_path) == 3 and len(cs_s_path[0]) in [0x20, 0x28, 0x40]:
-                try:
-                    int(cs_s_path[0], 0x10)
-                    rel_entry = DpkgRepo.ReleaseEntry(int(cs_s_path[1]), cs_s_path[2])
-                    self._release.setdefault(rel_entry.uri, rel_entry)
-                    if len(cs_s_path[0]) == 0x20:
-                        self._release[rel_entry.uri].checksum.md5 = cs_s_path[0]
-                    elif len(cs_s_path[0]) == 0x28:
-                        self._release[rel_entry.uri].checksum.sha1 = cs_s_path[0]
-                    elif len(cs_s_path[0]) == 0x40:
-                        self._release[rel_entry.uri].checksum.sha256 = cs_s_path[0]
-
-                except ValueError:
-                    pass
+            try:
+                entry = Entry._make(
+                    filter(None, line.strip().replace("\t", " ").split(" "))
+                )
+                int(entry.checksum, 0x10) # assert entry.checksum is hexadecimal
+                rel_entry = DpkgRepo.ReleaseEntry(int(entry.size), entry.path)
+            except (TypeError, ValueError):
+                continue
+
+            if len(entry.checksum) in (
+                LEN_MD5,
+                LEN_SHA1,
+                LEN_SHA256,
+                LEN_SHA384,
+                LEN_SHA512,
+            ):
+                self._release.setdefault(rel_entry.uri, rel_entry)
+                if len(entry.checksum) == LEN_MD5:
+                    self._release[rel_entry.uri].checksum.md5 = entry.checksum
+                elif len(entry.checksum) == LEN_SHA1:
+                    self._release[rel_entry.uri].checksum.sha1 = entry.checksum
+                elif len(entry.checksum) == LEN_SHA256:
+                    self._release[rel_entry.uri].checksum.sha256 = entry.checksum
+                elif len(entry.checksum) == LEN_SHA384:
+                    self._release[rel_entry.uri].checksum.sha384 = entry.checksum
+                elif len(entry.checksum) == LEN_SHA512:
+                    self._release[rel_entry.uri].checksum.sha512 = entry.checksum
 
         return self._release
 
@@ -209,9 +231,8 @@ def is_flat(self) -> bool:
 
     def verify_packages_index(self) -> bool:
         """
-        Verify Packages index against all listed checksum algorithms.
+        Verify Packages index with the best available checksum algorithm.
 
-        :param name: name of the packages index
         :return: result (boolean)
         """
         name, data = self.get_pkg_index_raw()
@@ -221,12 +242,16 @@ def verify_packages_index(self) -> bool:
            return True
 
         entry = self.get_release_index().get(name)
-        for algorithm in ["md5", "sha1", "sha256"]:
-            if entry is None:
-                res = False
-            else:
-                res = getattr(hashlib, algorithm)(data).hexdigest() == getattr(entry.checksum, algorithm)
-            if not res:
+        if entry is None:
+            return False
+
+        result = False
+        for algorithm in ("sha512", "sha384", "sha256", "sha1", "md5"):
+            entry_checksum = getattr(entry.checksum, algorithm, None)
+            if entry_checksum:
+                result = getattr(hashlib, algorithm)(data).hexdigest() == entry_checksum
                 break
+            else:
+                continue
 
-        return res
+        return result

backend/common/test/test_repo.py

@@ -73,6 +73,25 @@ def test_verify_packages_index(self):
             repo = DpkgRepo("http://mygreathost.com/ubuntu/dists/bionic/restricted/binary-amd64/")
             assert repo.verify_packages_index()
 
+    @patch("spacewalk.common.repo.DpkgRepo.get_pkg_index_raw", MagicMock(return_value=("Packages.gz", b"\x00")))
+    @patch("spacewalk.common.repo.DpkgRepo.is_flat", MagicMock(return_value=False))
+    def test_verify_packages_index_missing_some_checksums(self):
+        """
+        Test verify_packages_index method when only sha256 checksum is available.
+
+        :return:
+        """
+        gri = DpkgRepo.ReleaseEntry(size=999, uri="restricted/binary-amd64")
+        gri.checksum.md5 = ""
+        gri.checksum.sha1 = ""
+        gri.checksum.sha256 = "6e340b9cffb37a989ca544e6bb780a2c78901d3fb33738768511a30617afa01d"
+
+        release_index = MagicMock()
+        release_index().get = MagicMock(return_value=gri)
+        with patch("common.repo.DpkgRepo.get_release_index", release_index):
+            repo = DpkgRepo("http://mygreathost.com/ubuntu/dists/bionic/restricted/binary-amd64/")
+            assert repo.verify_packages_index()
+
     @patch("spacewalk.common.repo.DpkgRepo.get_release_index", mock_release_index)
     def test_is_flat(self):
         """
@@ -296,19 +315,6 @@ def test_decompress_pkg_index_gz_failure(self):
         assert zdcmp.called
         assert "hot" in str(exc.value)
 
-    @patch("spacewalk.common.repo.requests.get", MagicMock(
-        return_value=FakeRequests().conf(status_code=http.HTTPStatus.NOT_FOUND, content=b"")))
-    def test_get_pkg_index_raw_exception(self):
-        """
-        Test getting package index file exception handling
-
-        :return:
-        """
-        with pytest.raises(GeneralRepoException) as exc:
-            DpkgRepo("http://dummy/url").get_pkg_index_raw()
-
-        assert "No variants of package index has been found on http://dummy/url repo" == str(exc.value)
-
     def test_append_index_file_to_url(self):
         """
         Test append index files to the given url.

backend/spacewalk-backend.changes

@@ -1,3 +1,5 @@
+- Only check strongest available Ubuntu/Debian repository index checksum
+
 -------------------------------------------------------------------
 Wed Jun 10 12:14:57 CEST 2020 - jgonzalez@suse.com