Merge lp:~apw/launchpad/uefi-auto-key into lp:launchpad

[Andy Whitcroft]

Move the existing raw-uefi custom upload to validating the uefi signing keys on first use. For PPAs generate the keys on first use if they are missing.

Update the uefi custom upload to optionally manage key generation. Also in preparation for signing generification move key validation/generation to first use.

[Colin Watson (cjwatson)]

review: Needs Fixing

[Andy Whitcroft (apw)]

I have pushed up a number of updates on top of the previous code. I believe this should reflect your review comments. In particular I have revamps the testing to split command generation checks from flow testing. Oh and the lint roller has been applied.

review: Needs Resubmitting

[Andy Whitcroft (apw)]

review: Abstain

[Colin Watson (cjwatson)]

Definitely getting there, but still some things to fix, mostly in the tests.

review: Needs Fixing

[Andy Whitcroft (apw)]

Ok. I pushed up another attempt. I hope this one covers your remaining review comments. As expected the majority of the diff is in the tests. Looking at the diff against mainline there is actually very little delta for the operational code (also as expected).

[Colin Watson (cjwatson)]

Looks good now!

review: Approve

=== modified file 'lib/lp/archivepublisher/config.py'
--- lib/lp/archivepublisher/config.py	2016-02-05 16:51:12 +0000
+++ lib/lp/archivepublisher/config.py	2016-05-03 11:13:13 +0000
@@ -79,12 +79,15 @@
 
     if archive.is_main:
         pubconf.uefiroot = pubconf.archiveroot + '-uefi'
+        pubconf.uefiautokey = False
     elif archive.is_ppa:
         pubconf.uefiroot = os.path.join(
             ppa_config.signing_keys_root, "uefi",
             archive.owner.name, archive.name)
+        pubconf.uefiautokey = True
     else:
         pubconf.uefiroot = None
+        pubconf.uefiautokey = False
 
     pubconf.poolroot = os.path.join(pubconf.archiveroot, 'pool')
     pubconf.distsroot = os.path.join(pubconf.archiveroot, 'dists')
 
=== modified file 'lib/lp/archivepublisher/tests/test_config.py'
--- lib/lp/archivepublisher/tests/test_config.py	2016-01-13 17:09:34 +0000
+++ lib/lp/archivepublisher/tests/test_config.py	2016-05-03 11:13:13 +0000
@@ -47,6 +47,7 @@
         self.assertEqual(
             self.root + "/ubuntutest-temp", primary_config.temproot)
         self.assertEqual(archiveroot + "-uefi", primary_config.uefiroot)
+        self.assertFalse(primary_config.uefiautokey)
         self.assertIs(None, primary_config.metaroot)
         self.assertEqual(archiveroot + "-staging", primary_config.stagingroot)
 
@@ -70,6 +71,7 @@
         self.assertEqual(
             self.root + "/ubuntutest-temp", partner_config.temproot)
         self.assertEqual(archiveroot + "-uefi", partner_config.uefiroot)
+        self.assertFalse(partner_config.uefiautokey)
         self.assertIs(None, partner_config.metaroot)
         self.assertEqual(archiveroot + "-staging", partner_config.stagingroot)
 
@@ -92,6 +94,7 @@
         self.assertEqual(archiveroot + "-misc", copy_config.miscroot)
         self.assertEqual(archiveroot + "-temp", copy_config.temproot)
         self.assertIsNone(copy_config.uefiroot)
+        self.assertFalse(copy_config.uefiautokey)
         self.assertIs(None, copy_config.metaroot)
         self.assertIs(None, copy_config.stagingroot)
 
@@ -131,6 +134,7 @@
         uefiroot = "/var/tmp/ppa-signing-keys.test/uefi/%s/%s" % (
             self.ppa.owner.name, self.ppa.name)
         self.assertEqual(uefiroot, self.ppa_config.uefiroot)
+        self.assertTrue(self.ppa_config.uefiautokey)
         self.assertIs(None, self.ppa_config.metaroot)
         self.assertIs(None, self.ppa_config.stagingroot)
 
@@ -165,6 +169,7 @@
         uefiroot = "/var/tmp/ppa-signing-keys.test/uefi/%s/%s" % (
             p3a.owner.name, p3a.name)
         self.assertEqual(uefiroot, p3a_config.uefiroot)
+        self.assertTrue(self.ppa_config.uefiautokey)
         self.assertIs(None, p3a_config.metaroot)
         self.assertIs(None, p3a_config.stagingroot)
 
 
=== modified file 'lib/lp/archivepublisher/tests/test_uefi.py'
--- lib/lp/archivepublisher/tests/test_uefi.py	2016-02-05 16:51:12 +0000
+++ lib/lp/archivepublisher/tests/test_uefi.py	2016-05-03 11:13:13 +0000
@@ -7,6 +7,8 @@
 
 import os
 
+from fixtures import MonkeyPatch
+
 from lp.archivepublisher.customupload import (
     CustomUploadAlreadyExists,
     CustomUploadBadUmask,
@@ -18,11 +20,35 @@
 from lp.testing.fakemethod import FakeMethod
 
 
+class FakeMethodGenUefiKeys(FakeMethod):
+    """Fake execution of generation of Uefi keys pairs."""
+    def __init__(self, upload=None, *args, **kwargs):
+        super(FakeMethodGenUefiKeys, self).__init__(*args, **kwargs)
+        self.upload = upload
+
+    def __call__(self, *args, **kwargs):
+        super(FakeMethodGenUefiKeys, self).__call__(*args, **kwargs)
+
+        write_file(self.upload.key, "")
+        write_file(self.upload.cert, "")
+
+
 class FakeConfig:
-    """A fake publisher configuration."""
-    def __init__(self, archiveroot, uefiroot):
-        self.archiveroot = archiveroot
-        self.uefiroot = uefiroot
+    """A fake publisher configuration for the main archive."""
+    def __init__(self, distroroot, uefiroot):
+        self.distroroot = distroroot
+        self.uefiroot = uefiroot
+        self.archiveroot = os.path.join(self.distroroot, 'ubuntu')
+        self.uefiautokey = False
+
+
+class FakeConfigPPA:
+    """A fake publisher configuration for a PPA."""
+    def __init__(self, distroroot, uefiroot, owner, ppa):
+        self.distroroot = distroroot
+        self.uefiroot = uefiroot
+        self.archiveroot = os.path.join(self.distroroot, owner, ppa, 'ubuntu')
+        self.uefiautokey = True
 
 
 class TestUefi(TestCase):
@@ -37,11 +63,17 @@
         old_umask = os.umask(0o022)
         self.addCleanup(os.umask, old_umask)
 
-    def setUpKeyAndCert(self):
+    def setUpPPA(self):
+        self.pubconf = FakeConfigPPA(self.temp_dir, self.uefi_dir,
+            'ubuntu-archive', 'testing')
+        self.testcase_cn = '/CN=PPA ubuntu-archive testing/'
+
+    def setUpKeyAndCert(self, create=True):
         self.key = os.path.join(self.uefi_dir, "uefi.key")
         self.cert = os.path.join(self.uefi_dir, "uefi.crt")
-        write_file(self.key, "")
-        write_file(self.cert, "")
+        if create:
+            write_file(self.key, "")
+            write_file(self.cert, "")
 
     def openArchive(self, loader_type, version, arch):
         self.path = os.path.join(
@@ -52,41 +84,48 @@
     def process(self):
         self.archive.close()
         self.buffer.close()
+        fake_call = FakeMethod()
         upload = UefiUpload()
-        upload.sign = FakeMethod()
-        upload.process(self.pubconf, self.path, self.suite)
+        upload.signUefi = FakeMethod()
+        with self.useFixture(MonkeyPatch("subprocess.call", fake_call)):
+            upload.process(self.pubconf, self.path, self.suite)
+        # Under no circumstances is it safe to execute actual commands.
+        self.assertEqual(0, fake_call.call_count)
+
         return upload
 
     def getUefiPath(self, loader_type, arch):
         return os.path.join(
-            self.temp_dir, "dists", self.suite, "main", "uefi",
+            self.pubconf.archiveroot, "dists", self.suite, "main", "uefi",
             "%s-%s" % (loader_type, arch))
 
     def test_unconfigured(self):
         # If there is no key/cert configuration, processing succeeds but
-        # nothing is signed.
+        # nothing is signed.  Signing is attempted.
         self.pubconf = FakeConfig(self.temp_dir, None)
         self.openArchive("test", "1.0", "amd64")
         self.archive.add_file("1.0/empty.efi", "")
         upload = self.process()
-        self.assertEqual(0, upload.sign.call_count)
+        self.assertEqual(1, upload.signUefi.call_count)
 
     def test_missing_key_and_cert(self):
         # If the configured key/cert are missing, processing succeeds but
-        # nothing is signed.
+        # nothing is signed.  Signing is attempted.
         self.openArchive("test", "1.0", "amd64")
         self.archive.add_file("1.0/empty.efi", "")
         upload = self.process()
-        self.assertEqual(0, upload.sign.call_count)
+        self.assertEqual(1, upload.signUefi.call_count)
 
     def test_no_efi_files(self):
         # Tarballs containing no *.efi files are extracted without complaint.
+        # Nothing is signed.
         self.setUpKeyAndCert()
         self.openArchive("empty", "1.0", "amd64")
         self.archive.add_file("1.0/hello", "world")
-        self.process()
+        upload = self.process()
         self.assertTrue(os.path.exists(os.path.join(
             self.getUefiPath("empty", "amd64"), "1.0", "hello")))
+        self.assertEqual(0, upload.signUefi.call_count)
 
     def test_already_exists(self):
         # If the target directory already exists, processing fails.
@@ -104,15 +143,60 @@
         os.umask(0o002)  # cleanup already handled by setUp
         self.assertRaises(CustomUploadBadUmask, self.process)
 
-    def test_correct_signing_command(self):
-        # getSigningCommand returns the correct command.
+    def test_correct_uefi_signing_command_executed(self):
+        # Check that calling signUefi() will generate the expected command
+        # when appropriate keys are present.
         self.setUpKeyAndCert()
-        upload = UefiUpload()
-        upload.setTargetDirectory(
-            self.pubconf, "test_1.0_amd64.tar.gz", "distroseries")
-        expected_command = [
-            "sbsign", "--key", self.key, "--cert", self.cert, "t.efi"]
-        self.assertEqual(expected_command, upload.getSigningCommand("t.efi"))
+        fake_call = FakeMethod()
+        with self.useFixture(MonkeyPatch("subprocess.call", fake_call)):
+            upload = UefiUpload()
+            upload.generateUefiKeys = FakeMethod()
+            upload.setTargetDirectory(
+                self.pubconf, "test_1.0_amd64.tar.gz", "distroseries")
+            upload.signUefi('t.efi')
+        self.assertEqual(1, fake_call.call_count)
+        # Assert command form.
+        args = fake_call.calls[0][0][0]
+        expected_cmd = [
+            'sbsign', '--key', self.key, '--cert', self.cert, 't.efi',
+        ]
+        self.assertEqual(expected_cmd, args)
+        self.assertEqual(0, upload.generateUefiKeys.call_count)
+
+    def test_correct_uefi_signing_command_executed_no_keys(self):
+        # Check that calling signUefi() will generate no commands when
+        # no keys are present.
+        self.setUpKeyAndCert(create=False)
+        fake_call = FakeMethod()
+        with self.useFixture(MonkeyPatch("subprocess.call", fake_call)):
+            upload = UefiUpload()
+            upload.generateUefiKeys = FakeMethod()
+            upload.setTargetDirectory(
+                self.pubconf, "test_1.0_amd64.tar.gz", "distroseries")
+            upload.signUefi('t.efi')
+        self.assertEqual(0, fake_call.call_count)
+        self.assertEqual(0, upload.generateUefiKeys.call_count)
+
+    def test_correct_uefi_keygen_command_executed(self):
+        # Check that calling generateUefiKeys() will generate the
+        # expected command.
+        self.setUpPPA()
+        self.setUpKeyAndCert(create=False)
+        fake_call = FakeMethod()
+        with self.useFixture(MonkeyPatch("subprocess.call", fake_call)):
+            upload = UefiUpload()
+            upload.setTargetDirectory(
+                self.pubconf, "test_1.0_amd64.tar.gz", "distroseries")
+            upload.generateUefiKeys()
+        self.assertEqual(1, fake_call.call_count)
+        # Assert the actual command matches.
+        args = fake_call.calls[0][0][0]
+        expected_cmd = [
+            'openssl', 'req', '-new', '-x509', '-newkey', 'rsa:2048',
+            '-subj', self.testcase_cn, '-keyout', self.key, '-out', self.cert,
+            '-days', '3650', '-nodes', '-sha256',
+            ]
+        self.assertEqual(expected_cmd, args)
 
     def test_signs_image(self):
         # Each image in the tarball is signed.
@@ -120,10 +204,7 @@
         self.openArchive("test", "1.0", "amd64")
         self.archive.add_file("1.0/empty.efi", "")
         upload = self.process()
-        self.assertEqual(1, upload.sign.call_count)
-        self.assertEqual(1, len(upload.sign.calls[0][0]))
-        self.assertEqual(
-            "empty.efi", os.path.basename(upload.sign.calls[0][0][0]))
+        self.assertEqual(1, upload.signUefi.call_count)
 
     def test_installed(self):
         # Files in the tarball are installed correctly.
@@ -133,3 +214,36 @@
         self.process()
         self.assertTrue(os.path.exists(os.path.join(
             self.getUefiPath("test", "amd64"), "1.0", "empty.efi")))
+
+    def test_create_uefi_keys_autokey_off(self):
+        # Keys are not created.
+        self.setUpKeyAndCert(create=False)
+        self.assertFalse(os.path.exists(self.key))
+        self.assertFalse(os.path.exists(self.cert))
+        fake_call = FakeMethod()
+        with self.useFixture(MonkeyPatch("subprocess.call", fake_call)):
+            upload = UefiUpload()
+            upload.generateUefiKeys = FakeMethodGenUefiKeys(upload=upload)
+            upload.setTargetDirectory(
+                self.pubconf, "test_1.0_amd64.tar.gz", "distroseries")
+            upload.signUefi('t.efi')
+        self.assertEqual(0, upload.generateUefiKeys.call_count)
+        self.assertFalse(os.path.exists(self.key))
+        self.assertFalse(os.path.exists(self.cert))
+
+    def test_create_uefi_keys_autokey_on(self):
+        # Keys are created on demand.
+        self.setUpPPA()
+        self.setUpKeyAndCert(create=False)
+        self.assertFalse(os.path.exists(self.key))
+        self.assertFalse(os.path.exists(self.cert))
+        fake_call = FakeMethod()
+        with self.useFixture(MonkeyPatch("subprocess.call", fake_call)):
+            upload = UefiUpload()
+            upload.generateUefiKeys = FakeMethodGenUefiKeys(upload=upload)
+            upload.setTargetDirectory(
+                self.pubconf, "test_1.0_amd64.tar.gz", "distroseries")
+            upload.signUefi('t.efi')
+        self.assertEqual(1, upload.generateUefiKeys.call_count)
+        self.assertTrue(os.path.exists(self.key))
+        self.assertTrue(os.path.exists(self.cert))
 
=== modified file 'lib/lp/archivepublisher/uefi.py'
--- lib/lp/archivepublisher/uefi.py	2013-07-25 16:45:20 +0000
+++ lib/lp/archivepublisher/uefi.py	2016-05-03 11:13:13 +0000
@@ -68,24 +68,17 @@
                 self.logger.warning("No UEFI root configured for this archive")
             self.key = None
             self.cert = None
+            self.autokey = False
         else:
             self.key = os.path.join(pubconf.uefiroot, "uefi.key")
             self.cert = os.path.join(pubconf.uefiroot, "uefi.crt")
-            if not os.access(self.key, os.R_OK):
-                if self.logger is not None:
-                    self.logger.warning(
-                        "UEFI private key %s not readable" % self.key)
-                self.key = None
-            if not os.access(self.cert, os.R_OK):
-                if self.logger is not None:
-                    self.logger.warning(
-                        "UEFI certificate %s not readable" % self.cert)
-                self.cert = None
+            self.autokey = pubconf.uefiautokey
 
         self.setComponents(tarfile_path)
         self.targetdir = os.path.join(
             pubconf.archiveroot, "dists", distroseries, "main", "uefi",
             "%s-%s" % (self.loader_type, self.arch))
+        self.archiveroot = pubconf.archiveroot
 
     @classmethod
     def getSeriesKey(cls, tarfile_path):
@@ -102,17 +95,74 @@
                 if filename.endswith(".efi"):
                     yield os.path.join(dirpath, filename)
 
-    def getSigningCommand(self, image):
-        """Return the command used to sign an image."""
-        return ["sbsign", "--key", self.key, "--cert", self.cert, image]
-
-    def sign(self, image):
-        """Sign an image."""
-        if subprocess.call(self.getSigningCommand(image)) != 0:
+    def generateUefiKeys(self):
+        """Generate new UEFI Keys for this archive."""
+        directory = os.path.dirname(self.key)
+        if not os.path.exists(directory):
+            os.makedirs(directory)
+
+        # XXX: pull out the PPA owner and name to seed key CN
+        archive_name = os.path.dirname(self.archiveroot)
+        owner_name = os.path.basename(os.path.dirname(archive_name))
+        archive_name = os.path.basename(archive_name)
+        common_name = '/CN=PPA ' + owner_name + ' ' + archive_name + '/'
+
+        old_mask = os.umask(0o077)
+        try:
+            new_key_cmd = [
+                'openssl', 'req', '-new', '-x509', '-newkey', 'rsa:2048',
+                '-subj', common_name, '-keyout', self.key, '-out', self.cert,
+                '-days', '3650', '-nodes', '-sha256',
+                ]
+            if subprocess.call(new_key_cmd) != 0:
+                # Just log this rather than failing, since custom upload errors
+                # tend to make the publisher rather upset.
+                if self.logger is not None:
+                    self.logger.warning(
+                        "Failed to generate UEFI signing keys for %s" %
+                        common_name)
+        finally:
+            os.umask(old_mask)
+
+        if os.path.exists(self.cert):
+            os.chmod(self.cert, 0o644)
+
+    def getUefiKeys(self):
+        """Validate and return the uefi key and cert for encryption."""
+
+        if self.key and self.cert:
+            # If neither of the key files exists then attempt to
+            # generate them.
+            if (self.autokey and not os.path.exists(self.key)
+                and not os.path.exists(self.cert)):
+                self.generateUefiKeys()
+
+            # If we have keys, but cannot read them they are dead to us.
+            if not os.access(self.key, os.R_OK):
+                if self.logger is not None:
+                    self.logger.warning(
+                        "UEFI private key %s not readable" % self.key)
+                self.key = None
+            if not os.access(self.cert, os.R_OK):
+                if self.logger is not None:
+                    self.logger.warning(
+                        "UEFI certificate %s not readable" % self.cert)
+                self.cert = None
+
+        return (self.key, self.cert)
+
+    def signUefi(self, image):
+        """Attempt to sign an image."""
+        (key, cert) = self.getUefiKeys()
+        if not key or not cert:
+            return
+        cmdl = ["sbsign", "--key", key, "--cert", cert, image]
+        if subprocess.call(cmdl) != 0:
             # Just log this rather than failing, since custom upload errors
             # tend to make the publisher rather upset.
             if self.logger is not None:
-                self.logger.warning("Failed to sign %s" % image)
+                self.logger.warning("UEFI Signing Failed '%s'" %
+                    " ".join(cmdl))
 
     def extract(self):
         """Copy the custom upload to a temporary directory, and sign it.
@@ -120,11 +170,10 @@
         No actual extraction is required.
         """
         super(UefiUpload, self).extract()
-        if self.key is not None and self.cert is not None:
-            efi_filenames = list(self.findEfiFilenames())
-            for efi_filename in efi_filenames:
-                remove_if_exists("%s.signed" % efi_filename)
-                self.sign(efi_filename)
+        efi_filenames = list(self.findEfiFilenames())
+        for efi_filename in efi_filenames:
+            remove_if_exists("%s.signed" % efi_filename)
+            self.signUefi(efi_filename)
 
     def shouldInstall(self, filename):
         return filename.startswith("%s/" % self.version)