rewrite check_files_md5 in pure Python

apport/fileutils.py

@@ -16,11 +16,13 @@
 import contextlib
 import functools
 import glob
+import hashlib
 import http.client
 import io
 import json
 import operator
 import os
+import pathlib
 import pwd
 import re
 import socket
@@ -446,22 +448,21 @@ def check_files_md5(sumfile: str) -> list[str]:
     Return a list of files that don't match.
     """
     assert os.path.exists(sumfile)
-    md5sum = subprocess.run(
-        ["/usr/bin/md5sum", "-c", sumfile],
-        check=False,
-        capture_output=True,
-        cwd="/",
-        env={},
-    )
-
-    # if md5sum succeeded, don't bother parsing the output
-    if md5sum.returncode == 0:
-        return []
-
+    root = pathlib.Path("/")
+    sums = pathlib.Path(sumfile).read_text(encoding="utf-8")
     mismatches = []
-    for line in md5sum.stdout.decode().splitlines():
-        if line.endswith("FAILED"):
-            mismatches.append(line.rsplit(":", 1)[0])
+    for line in sums.splitlines():
+        # md5sum format: <32 hex hash><2 spaces><filepath>
+        expected_hash = line[:32]
+        filepath = line[34:]
+
+        try:
+            with (root / filepath).open("rb") as target_file:
+                md5sum = hashlib.file_digest(target_file, "md5").hexdigest()
+            if md5sum.lower() != expected_hash.lower():
+                mismatches.append(filepath)
+        except OSError:
+            pass
 
     return mismatches
 

apport/packaging_impl/apt_dpkg.py

@@ -686,7 +686,7 @@ def get_modified_files(self, package: str) -> list[str]:
             return []
 
         # create a list of files with a newer timestamp for md5sum'ing
-        sums = b""
+        modified = []
         sumfile = (
             f"/var/lib/dpkg/info/{package}:{self.get_system_architecture()}.md5sums"
         )
@@ -696,11 +696,11 @@ def get_modified_files(self, package: str) -> list[str]:
                 # some packages do not ship md5sums
                 return []
 
-        with open(sumfile, "rb") as fd:
+        with open(sumfile, "r", encoding="utf-8", errors="replace") as fd:
             for line in fd:
                 try:
                     # ignore lines with NUL bytes (happens, LP#96050)
-                    if b"\0" in line:
+                    if "\0" in line:
                         apport.logging.warning(
                             "%s contains NUL character, ignoring line", sumfile
                         )
@@ -711,17 +711,19 @@ def get_modified_files(self, package: str) -> list[str]:
                             "%s contains empty line, ignoring line", sumfile
                         )
                         continue
-                    s = os.stat(f"/{words[-1].decode('UTF-8')}".encode())
+                    expected_hash = words[0]
+                    filepath = pathlib.Path(f"/{words[-1]}")
+                    s = filepath.stat()
                     if max(s.st_mtime, s.st_ctime) <= max_time:
                         continue
+                    with filepath.open("rb") as target_file:
+                        md5sum = hashlib.file_digest(target_file, "md5").hexdigest()
+                    if md5sum.lower() != expected_hash.lower():
+                        modified.append(words[-1])
                 except OSError:
                     pass
 
-                sums += line
-
-        if sums:
-            return self._check_files_md5(sums)
-        return []
+        return modified
 
     def get_modified_conffiles(self, package: str) -> dict[str, bytes | str]:
         """Return modified configuration files of a package.
@@ -1544,35 +1546,6 @@ def _call_dpkg(args: list[str]) -> str:
             return dpkg.stdout.decode("UTF-8")
         raise ValueError("package does not exist")
 
-    @staticmethod
-    def _check_files_md5(sumfile: bytes) -> list[str]:
-        """Call md5sum.
-
-        This is separate from get_modified_files so that it is automatically
-        testable.
-        """
-        env: dict[str, str] = {}
-        md5sum = subprocess.run(
-            ["/usr/bin/md5sum", "-c"],
-            check=False,
-            input=sumfile,
-            capture_output=True,
-            cwd="/",
-            env=env,
-        )
-
-        # if md5sum succeeded, don't bother parsing the output
-        if md5sum.returncode == 0:
-            return []
-        out = md5sum.stdout.decode("UTF-8", errors="replace")
-
-        mismatches = []
-        for line in out.splitlines():
-            if line.endswith("FAILED"):
-                mismatches.append(line.rsplit(":", 1)[0])
-
-        return mismatches
-
     @staticmethod
     def _get_primary_mirror_from_apt_sources(apt_dir: str) -> str:
         """Heuristically determine primary mirror from an apt sources.list."""

tests/integration/test_fileutils.py

@@ -325,6 +325,7 @@ def test_check_files_md5(self) -> None:
             fd.write(f"""\
 2e41290da2fa3f68bd3313174467e3b5  {f1[1:]}
 f6423dfbc4faf022e58b4d3f5ff71a70  {f2}
+deadbeef000001111110000011110000  /non-existing/file
 """)
         self.assertEqual(
             apport.fileutils.check_files_md5(sumfile), [], "correct md5sums"

tests/integration/test_packaging_apt_dpkg.py

@@ -37,39 +37,6 @@ def tearDown(self) -> None:
         os.environ.update(self.orig_environ)
         shutil.rmtree(self.workdir)
 
-    def test_check_files_md5(self) -> None:
-        """_check_files_md5()."""
-        td = tempfile.mkdtemp()
-        try:
-            f1 = os.path.join(td, "test 1.txt")
-            f2 = os.path.join(td, "test:2.txt")
-            with open(f1, "w", encoding="utf-8") as fd:
-                fd.write("Some stuff")
-            with open(f2, "w", encoding="utf-8") as fd:
-                fd.write("More stuff")
-            # use one relative and one absolute path in checksums file
-            sumfile = (
-                b"2e41290da2fa3f68bd3313174467e3b5  " + f1[1:].encode() + b"\n"
-                b"f6423dfbc4faf022e58b4d3f5ff71a70  " + f2.encode() + b"\n"
-                b"deadbeef000001111110000011110000  /bin/\xc3\xa4"
-            )
-            self.assertEqual(impl._check_files_md5(sumfile), [], "correct md5sums")
-
-            with open(f1, "w", encoding="utf-8") as fd:
-                fd.write("Some stuff!")
-            self.assertEqual(impl._check_files_md5(sumfile), [f1[1:]], "file 1 wrong")
-            with open(f2, "w", encoding="utf-8") as fd:
-                fd.write("More stuff!")
-            self.assertEqual(
-                impl._check_files_md5(sumfile), [f1[1:], f2], "files 1 and 2 wrong"
-            )
-            with open(f1, "w", encoding="utf-8") as fd:
-                fd.write("Some stuff")
-            self.assertEqual(impl._check_files_md5(sumfile), [f2], "file 2 wrong")
-
-        finally:
-            shutil.rmtree(td)
-
     def test_get_version(self) -> None:
         """get_version()."""
         self.assertTrue(impl.get_version("libc6").startswith("2"))