recursiveloader: Support verifying Manifest signatures explicitly

2 files changed, 90 insertions, 3 deletions

diff --git a/gemato/recursiveloader.py b/gemato/recursiveloader.py
index 0fa42ad..659049b 100644
--- a/gemato/recursiveloader.py
+++ b/gemato/recursiveloader.py
@@ -19,14 +19,24 @@ class ManifestRecursiveLoader(object):
     and provides methods to access the entries in them.
     """
 
-    def __init__(self, top_manifest_path):
+    def __init__(self, top_manifest_path,
+            verify_openpgp=True, openpgp_env=None):
         """
         Instantiate the loader for a Manifest tree starting at top-level
         Manifest @top_manifest_path.
+
+        @verify_openpgp and @openpgp_env are passed down
+        to ManifestFile. If the top-level Manifest is OpenPGP-signed
+        and the verification succeeds, openpgp_signed property
+        is set to True.
         """
         self.root_directory = os.path.dirname(top_manifest_path)
         self.loaded_manifests = {}
-        self.load_manifest(os.path.basename(top_manifest_path))
+        self.verify_openpgp = verify_openpgp
+        self.openpgp_env = openpgp_env
+        # TODO: allow catching OpenPGP exceptions somehow?
+        m = self.load_manifest(os.path.basename(top_manifest_path))
+        self.openpgp_signed = m.openpgp_signed
 
     def load_manifest(self, relpath, verify_entry=None):
         """
@@ -44,10 +54,11 @@ class ManifestRecursiveLoader(object):
                         relpath, verify_entry, diff)
         with gemato.compression.open_potentially_compressed_path(
                 path, 'r', encoding='utf8') as f:
-            m.load(f)
+            m.load(f, self.verify_openpgp, self.openpgp_env)
             st = os.fstat(f.fileno())
             self.manifest_device = st.st_dev
         self.loaded_manifests[relpath] = m
+        return m
 
     def _iter_manifests_for_path(self, path, recursive=False):
         """
diff --git a/tests/test_openpgp.py b/tests/test_openpgp.py
index 9d37716..5d55565 100644
--- a/tests/test_openpgp.py
+++ b/tests/test_openpgp.py
@@ -4,10 +4,15 @@
 # Licensed under the terms of 2-clause BSD license
 
 import io
+import os.path
+import shutil
+import tempfile
 import unittest
 
+import gemato.compression
 import gemato.manifest
 import gemato.openpgp
+import gemato.recursiveloader
 
 
 PUBLIC_KEY = u'''
@@ -192,6 +197,19 @@ class SignedManifestTest(unittest.TestCase):
             self.assertRaises(gemato.exceptions.ManifestSyntaxError,
                     m.load, f, verify_openpgp=False)
 
+    def test_recursive_manifest_loader(self):
+        d = tempfile.mkdtemp()
+        try:
+            with io.open(os.path.join(d, 'Manifest'), 'w') as f:
+                f.write(MODIFIED_SIGNED_MANIFEST)
+
+            m = gemato.recursiveloader.ManifestRecursiveLoader(
+                    os.path.join(d, 'Manifest'),
+                    verify_openpgp=False)
+            self.assertFalse(m.openpgp_signed)
+        finally:
+            shutil.rmtree(d)
+
 
 class OpenPGPCorrectKeyTest(unittest.TestCase):
     """
@@ -246,6 +264,35 @@ class OpenPGPCorrectKeyTest(unittest.TestCase):
             self.assertRaises(gemato.exceptions.OpenPGPVerificationFailure,
                 m.load, f, openpgp_env=self.env)
 
+    def test_recursive_manifest_loader(self):
+        d = tempfile.mkdtemp()
+        try:
+            with io.open(os.path.join(d, 'Manifest'), 'w') as f:
+                f.write(SIGNED_MANIFEST)
+
+            m = gemato.recursiveloader.ManifestRecursiveLoader(
+                    os.path.join(d, 'Manifest'),
+                    verify_openpgp=True,
+                    openpgp_env=self.env)
+            self.assertTrue(m.openpgp_signed)
+        finally:
+            shutil.rmtree(d)
+
+    def test_recursive_manifest_loader_compressed(self):
+        d = tempfile.mkdtemp()
+        try:
+            with gemato.compression.open_potentially_compressed_path(
+                    os.path.join(d, 'Manifest.gz'), 'w') as cf:
+                cf.write(SIGNED_MANIFEST)
+
+            m = gemato.recursiveloader.ManifestRecursiveLoader(
+                    os.path.join(d, 'Manifest.gz'),
+                    verify_openpgp=True,
+                    openpgp_env=self.env)
+            self.assertTrue(m.openpgp_signed)
+        finally:
+            shutil.rmtree(d)
+
 
 class OpenPGPNoKeyTest(unittest.TestCase):
     """
@@ -291,6 +338,35 @@ class OpenPGPNoKeyTest(unittest.TestCase):
         self.assertIsNotNone(m.find_path_entry('myebuild-0.ebuild'))
         self.assertFalse(m.openpgp_signed)
 
+    def test_recursive_manifest_loader(self):
+        d = tempfile.mkdtemp()
+        try:
+            with io.open(os.path.join(d, 'Manifest'), 'w') as f:
+                f.write(SIGNED_MANIFEST)
+
+            self.assertRaises(gemato.exceptions.OpenPGPVerificationFailure,
+                    gemato.recursiveloader.ManifestRecursiveLoader,
+                    os.path.join(d, 'Manifest'),
+                    verify_openpgp=True,
+                    openpgp_env=self.env)
+        finally:
+            shutil.rmtree(d)
+
+    def test_recursive_manifest_loader_compressed(self):
+        d = tempfile.mkdtemp()
+        try:
+            with gemato.compression.open_potentially_compressed_path(
+                    os.path.join(d, 'Manifest.gz'), 'w') as cf:
+                cf.write(SIGNED_MANIFEST)
+
+            self.assertRaises(gemato.exceptions.OpenPGPVerificationFailure,
+                    gemato.recursiveloader.ManifestRecursiveLoader,
+                    os.path.join(d, 'Manifest.gz'),
+                    verify_openpgp=True,
+                    openpgp_env=self.env)
+        finally:
+            shutil.rmtree(d)
+
 
 class OpenPGPContextManagerTest(unittest.TestCase):
     """