Support --require-secure-hashes in verify

Signed-off-by: Michał Górny <mgorny@gentoo.org>

4 files changed, 130 insertions, 24 deletions

diff --git a/gemato/cli.py b/gemato/cli.py
index 42f4ced..5b2370d 100644
--- a/gemato/cli.py
+++ b/gemato/cli.py
@@ -176,6 +176,16 @@ class BaseManifestLoaderMixin:
             '-x', '--one-file-system', action='store_true',
             help='Do not cross filesystem boundaries (report an error '
                  'instead)')
+        secugroup = subp.add_mutually_exclusive_group()
+        secugroup.add_argument(
+            "--require-secure-hashes", action="store_true",
+            default=None,
+            help="Require using secure hashes (default if Manifest is signed)")
+        secugroup.add_argument(
+            "--no-require-secure-hashes", action="store_false",
+            dest="require_secure_hashes",
+            help="Do not require using secure hashes (default if Manifest "
+                 "is not signed)")
 
     def parse_args(self, args, argp):
         super().parse_args(args, argp)
@@ -187,6 +197,9 @@ class BaseManifestLoaderMixin:
             self.init_kwargs['max_jobs'] = args.jobs
         if args.one_file_system:
             self.init_kwargs['allow_xdev'] = False
+        if args.require_secure_hashes is not None:
+            self.init_kwargs["require_secure_hashes"] = (
+                args.require_secure_hashes)
 
 
 class VerifyCommand(BaseManifestLoaderMixin, VerifyingOpenPGPMixin,
@@ -302,16 +315,6 @@ class BaseUpdateMixin(BaseManifestLoaderMixin, BaseOpenPGPMixin):
             '-p', '--profile',
             help='Use the specified profile ("default", "ebuild", '
                  '"old-ebuild"...)')
-        secugroup = update.add_mutually_exclusive_group()
-        secugroup.add_argument(
-            "--require-secure-hashes", action="store_true",
-            default=None,
-            help="Require using secure hashes (default if Manifest is signed)")
-        secugroup.add_argument(
-            "--no-require-secure-hashes", action="store_false",
-            dest="require_secure_hashes",
-            help="Do not require using secure hashes (default if Manifest "
-                 "is not signed)")
         signgroup = update.add_mutually_exclusive_group()
         signgroup.add_argument(
             '-s', '--sign', action='store_true',
@@ -350,9 +353,6 @@ class BaseUpdateMixin(BaseManifestLoaderMixin, BaseOpenPGPMixin):
                 get_profile_by_name(args.profile))
         if args.sign is not None:
             self.init_kwargs['sign_openpgp'] = args.sign
-        if args.require_secure_hashes is not None:
-            self.init_kwargs["require_secure_hashes"] = (
-                args.require_secure_hashes)
 
 
 class UpdateCommand(BaseUpdateMixin, GematoCommand):
diff --git a/gemato/recursiveloader.py b/gemato/recursiveloader.py
index fa7acb9..93acb84 100644
--- a/gemato/recursiveloader.py
+++ b/gemato/recursiveloader.py
@@ -47,7 +47,8 @@ class ManifestLoader:
     Helper class to load Manifests in subprocesses.
     """
 
-    __slots__ = ['root_directory', 'verify_openpgp', 'openpgp_env']
+    __slots__ = ['root_directory', 'verify_openpgp', 'openpgp_env',
+                 'require_secure_hash']
 
     def __init__(self, root_directory, verify_openpgp, openpgp_env):
         """
@@ -60,6 +61,10 @@ class ManifestLoader:
         self.root_directory = root_directory
         self.verify_openpgp = verify_openpgp
         self.openpgp_env = openpgp_env
+        # this is set on the instance because the value may depend
+        # on whether the top-level Manifest is signed (and this class
+        # is used to load it)
+        self.require_secure_hash = False
 
     def verify_and_load(self, relpath, verify_entry=None):
         """
@@ -77,7 +82,9 @@ class ManifestLoader:
         path = os.path.join(self.root_directory, relpath)
 
         if verify_entry is not None:
-            ret, diff = verify_path(path, verify_entry)
+            ret, diff = verify_path(
+                path, verify_entry,
+                require_secure_hash=self.require_secure_hash)
             if not ret:
                 raise ManifestMismatch(relpath, verify_entry, diff)
 
@@ -107,19 +114,23 @@ class SubprocessVerifier:
     """
 
     __slots__ = ['top_level_manifest_filename',
-                 'manifest_device', 'fail_handler', 'last_mtime']
+                 'manifest_device', 'fail_handler', 'last_mtime',
+                 'require_secure_hashes']
 
     def __init__(self, top_level_manifest_filename,
-                 manifest_device, fail_handler, last_mtime):
+                 manifest_device, fail_handler, last_mtime,
+                 require_secure_hashes):
         self.top_level_manifest_filename = top_level_manifest_filename
         self.manifest_device = manifest_device
         self.fail_handler = fail_handler
         self.last_mtime = last_mtime
+        self.require_secure_hashes = require_secure_hashes
 
     def _verify_one_file(self, path, relpath, e):
         ret, diff = verify_path(path, e,
                                 expected_dev=self.manifest_device,
-                                last_mtime=self.last_mtime)
+                                last_mtime=self.last_mtime,
+                                require_secure_hash=self.require_secure_hashes)
 
         if not ret:
             err = ManifestMismatch(relpath, e, diff)
@@ -267,9 +278,12 @@ class ManifestRecursiveLoader:
         an exception upon crossing filesystem boundaries. It defaults
         to false.
 
-        If @require_secure_hashes is True, only secure hashes can be used.
-        If it is False, all hashes are permitted. If it is None, secure
-        hashes are required if top-level Manifest is going to be signed.
+        If @require_secure_hashes is True, strengthened hash security
+        rules are enforced. When verifying Manifest entries, at least
+        one secure hash must be present. When writing Manifest entries,
+        all used hashes must be secure. If it is False, all hashes are
+        permitted. If it is None, secure hashes are required
+        if top-level Manifest is signed or going to be signed.
         """
 
         self.root_directory = os.path.dirname(top_manifest_path)
@@ -313,6 +327,7 @@ class ManifestRecursiveLoader:
             else:
                 require_secure_hashes = self.sign_openpgp
         self.require_secure_hashes = require_secure_hashes
+        self.manifest_loader.require_secure_hash = require_secure_hashes
 
         if require_secure_hashes and hashes is not None:
             insecure = list(filter(lambda x: not is_hash_secure(x), hashes))
@@ -527,7 +542,8 @@ class ManifestRecursiveLoader:
         """
         real_path = os.path.join(self.root_directory, relpath)
         path_entry = self.find_path_entry(relpath)
-        return verify_path(real_path, path_entry)
+        return verify_path(real_path, path_entry,
+                           require_secure_hash=self.require_secure_hashes)
 
     def assert_path_verifies(self, relpath):
         """
@@ -538,7 +554,8 @@ class ManifestRecursiveLoader:
         real_path = os.path.join(self.root_directory, relpath)
         path_entry = self.find_path_entry(relpath)
         ret, diff = verify_path(real_path, path_entry,
-                                expected_dev=self.manifest_device)
+                                expected_dev=self.manifest_device,
+                                require_secure_hash=self.require_secure_hashes)
         if not ret:
             raise ManifestMismatch(relpath, path_entry, diff)
 
@@ -707,7 +724,8 @@ class ManifestRecursiveLoader:
         verifier = SubprocessVerifier(
                 self.top_level_manifest_filename,
                 self.manifest_device,
-                fail_handler, last_mtime)
+                fail_handler, last_mtime,
+                self.require_secure_hashes)
 
         with MultiprocessingPoolWrapper(self.max_jobs) as pool:
             # verify the directories in parallel
diff --git a/tests/test_openpgp.py b/tests/test_openpgp.py
index 3e30b65..602379e 100644
--- a/tests/test_openpgp.py
+++ b/tests/test_openpgp.py
@@ -585,6 +585,9 @@ def test_cli(base_tree, caplog, manifest_var, key_var, expected):
                               str(base_tree / '.key.bin'),
                               '--no-refresh-keys',
                               '--require-signed-manifest',
+                              # we verify this option separately
+                              # and our test data currently sucks
+                              '--no-require-secure-hashes',
                               str(base_tree)])
     if str(OpenPGPNoImplementation('install gpg')) in caplog.text:
         pytest.skip('OpenPGP implementation missing')
@@ -1030,3 +1033,27 @@ def test_update_require_secure_cli(base_tree, caplog, hashes_arg,
     assert retval == expected
     if expected == 1:
         assert str(ManifestInsecureHashes(insecure)) in caplog.text
+
+
+@pytest.mark.parametrize(
+    "require_secure", ["", "--no-require-secure-hashes"])
+def test_verify_require_secure_cli(base_tree, caplog, require_secure):
+    with open(base_tree / ".key.bin", "wb") as keyf:
+        keyf.write(VALID_PUBLIC_KEY)
+    with open(base_tree / "Manifest", "w") as f:
+        f.write(SIGNED_MANIFEST)
+
+    retval = gemato.cli.main(["gemato", "verify",
+                              "--no-refresh-keys",
+                              "--require-signed-manifest",
+                              "-K", str(base_tree / ".key.bin"),
+                              str(base_tree)]
+                             + require_secure.split())
+    if str(OpenPGPNoImplementation('install gpg')) in caplog.text:
+        pytest.skip('OpenPGP implementation missing')
+
+    expected = (1 if require_secure != "--no-require-secure-hashes"
+                else 0)
+    assert retval == expected
+    if expected == 1:
+        assert str(ManifestInsecureHashes(["MD5"])) in caplog.text
diff --git a/tests/test_recursiveloader.py b/tests/test_recursiveloader.py
index 5cbd4d8..6e2395b 100644
--- a/tests/test_recursiveloader.py
+++ b/tests/test_recursiveloader.py
@@ -817,6 +817,20 @@ DATA test 0 X-UNKNOWN 0123456789abcdef
     }
 
 
+class SecureHashLayout(BaseLayout):
+    """Layout using at least one cryptographically secure hash"""
+
+    MANIFESTS = {
+        "Manifest": """
+DATA test 0 MD5 d41d8cd98f00b204e9800998ecf8427e\
+ SHA256 e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
+""",
+    }
+    FILES = {
+        "test": "",
+    }
+
+
 FLAT_LAYOUTS = [
     DuplicateEntryLayout,
     DuplicateEbuildEntryLayout,
@@ -2490,6 +2504,12 @@ INSECURE_HASH_TESTS = [
     ("", []),
 ]
 
+INSECURE_HASH_VERIFY_TESTS = [
+    # layout, insecure
+    (UnknownHashLayout, ["MD5"]),
+    (SecureHashLayout, None),
+]
+
 
 @pytest.mark.parametrize("hashes_arg,insecure", INSECURE_HASH_TESTS)
 def test_insecure_hashes(layout_factory, hashes_arg, insecure):
@@ -2504,6 +2524,36 @@ def test_insecure_hashes(layout_factory, hashes_arg, insecure):
                                 require_secure_hashes=True)
 
 
+@pytest.mark.parametrize("layout,insecure", INSECURE_HASH_VERIFY_TESTS)
+@pytest.mark.parametrize(
+    "func,path",
+    [(ManifestRecursiveLoader.verify_path, "test"),
+     (ManifestRecursiveLoader.assert_path_verifies, "test"),
+     (ManifestRecursiveLoader.assert_directory_verifies, ""),
+     ])
+def test_insecure_hashes_verify(layout_factory, layout, insecure, func, path):
+    tmp_path = layout_factory.create(layout)
+    m = ManifestRecursiveLoader(tmp_path / layout.TOP_MANIFEST,
+                                allow_xdev=False,
+                                require_secure_hashes=True)
+
+    ctx = (pytest.raises(ManifestInsecureHashes) if insecure is not None
+           else contextlib.nullcontext())
+    with ctx:
+        func(m, path)
+
+
+def test_insecure_hashes_load(layout_factory):
+    layout = BasicTestLayout
+    tmp_path = layout_factory.create(layout)
+    m = ManifestRecursiveLoader(tmp_path / layout.TOP_MANIFEST,
+                                allow_xdev=False,
+                                require_secure_hashes=True)
+
+    with pytest.raises(ManifestInsecureHashes):
+        m.load_manifests_for_path("sub")
+
+
 @pytest.mark.parametrize("hashes_arg,insecure", INSECURE_HASH_TESTS)
 @pytest.mark.parametrize(
     "func,arg",
@@ -2534,6 +2584,17 @@ def test_insecure_hashes_update_no_arg(layout_factory):
         m.update_entry_for_path("sub/deeper/test")
 
 
+@pytest.mark.parametrize("layout,insecure", INSECURE_HASH_VERIFY_TESTS)
+def test_insecure_hashes_verify_cli(layout_factory, caplog, layout,
+                                    insecure):
+    tmp_path = layout_factory.create(layout)
+    expected = 1 if insecure is not None else 0
+    assert gemato.cli.main(["gemato", "verify", "--require-secure-hashes",
+                            str(tmp_path)]) == expected
+    if insecure is not None:
+        assert str(ManifestInsecureHashes(insecure)) in caplog.text
+
+
 @pytest.mark.parametrize("hashes_arg,insecure", INSECURE_HASH_TESTS)
 @pytest.mark.parametrize("command", ["create", "update"])
 def test_insecure_hashes_update_cli(layout_factory, caplog,