launchpad-reviewers team mailing list archive

[Andy Whitcroft <apw@xxxxxxxxxxxxx>]

Andy Whitcroft has proposed merging lp:~apw/launchpad/uefi-auto-key into lp:launchpad.

Requested reviews:
  Colin Watson (cjwatson)

For more details, see:
https://code.launchpad.net/~apw/launchpad/uefi-auto-key/+merge/293087

Update the uefi custom upload to optionally manage key generation.  Also in preparation for signing generification move key validation/generation to first use.
-- 
Your team Launchpad code reviewers is subscribed to branch lp:launchpad.

=== 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-04-27 10:58:39 +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_uefi.py'
--- lib/lp/archivepublisher/tests/test_uefi.py	2016-02-05 16:51:12 +0000
+++ lib/lp/archivepublisher/tests/test_uefi.py	2016-04-27 10:58:39 +0000
@@ -17,12 +17,27 @@
 from lp.testing import TestCase
 from lp.testing.fakemethod import FakeMethod
 
+class FakeMethodExecuteCmd(FakeMethod):
+    """A fake command executer."""
+    def __call__(self, *args, **kwargs):
+        super(FakeMethodExecuteCmd, self).__call__(*args, **kwargs)
+
+        cmdl = args[0]
+
+        # keygen command for UEFI keys:
+        if (cmdl[0] == 'openssl' and
+            cmdl[8] == '-keyout' and cmdl[9].startswith('/tmp/') and
+            cmdl[10] == '-out' and cmdl[11].startswith('/tmp/')):
+            
+            write_file(cmdl[9], "")
+            write_file(cmdl[11], "")
 
 class FakeConfig:
     """A fake publisher configuration."""
     def __init__(self, archiveroot, uefiroot):
         self.archiveroot = archiveroot
         self.uefiroot = uefiroot
+        self.uefiautokey = False
 
 
 class TestUefi(TestCase):
@@ -37,11 +52,44 @@
         old_umask = os.umask(0o022)
         self.addCleanup(os.umask, old_umask)
 
-    def setUpKeyAndCert(self):
+    def setUpAutoKey(self):
+        self.pubconf.uefiautokey = True
+
+    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 validateKeyAndCert(self):
+        if os.path.exists(self.key) and os.path.exists(self.cert):
+            return True
+        else:
+            return False
+
+    def validateCmdUefiKeygen(self, call):
+        args = call[0][0]
+
+        archive_name = os.path.basename(self.pubconf.archiveroot)
+        owner_name   = os.path.basename(os.path.dirname(self.pubconf.archiveroot))
+        common_name  = '/CN=PPA ' + owner_name + ' ' + archive_name + '/'
+        
+        cmd_gen  = ['openssl', 'req', '-new', '-x509', '-newkey', 'rsa:2048',
+                    '-subj', common_name,
+                    '-keyout', self.key, '-out', self.cert,
+                    '-days', '3650', '-nodes', '-sha256']
+        return args == cmd_gen
+
+    def validateCmdSbsign(self, call, filename):
+        args = call[0][0]
+
+        if len(args) >= 6 and args[5].startswith('/'):
+            args[5] = os.path.basename(args[5])
+
+        cmd_sign = ['sbsign', '--key', self.key, '--cert', self.cert, filename ]
+
+        return args == cmd_sign
 
     def openArchive(self, loader_type, version, arch):
         self.path = os.path.join(
@@ -53,7 +101,7 @@
         self.archive.close()
         self.buffer.close()
         upload = UefiUpload()
-        upload.sign = FakeMethod()
+        upload.execute_cmd = FakeMethodExecuteCmd()
         upload.process(self.pubconf, self.path, self.suite)
         return upload
 
@@ -69,7 +117,7 @@
         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(0, upload.execute_cmd.call_count)
 
     def test_missing_key_and_cert(self):
         # If the configured key/cert are missing, processing succeeds but
@@ -77,7 +125,7 @@
         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(0, upload.execute_cmd.call_count)
 
     def test_no_efi_files(self):
         # Tarballs containing no *.efi files are extracted without complaint.
@@ -120,10 +168,10 @@
         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(1, upload.execute_cmd.call_count)
+        self.assertEqual(1, len(upload.execute_cmd.calls[0][0]))
         self.assertEqual(
-            "empty.efi", os.path.basename(upload.sign.calls[0][0][0]))
+            "empty.efi", os.path.basename(upload.execute_cmd.calls[0][0][0][5]))
 
     def test_installed(self):
         # Files in the tarball are installed correctly.
@@ -133,3 +181,29 @@
         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(self.validateKeyAndCert())
+        self.openArchive("test", "1.0", "amd64")
+        self.archive.add_file("1.0/empty.efi", "")
+        self.process()
+        self.assertTrue(os.path.exists(os.path.join(
+            self.getUefiPath("test", "amd64"), "1.0", "empty.efi")))
+        self.assertFalse(self.validateKeyAndCert())
+
+    def test_create_uefi_keys_autokey_on(self):
+        # Keys are created as needed.
+        self.setUpAutoKey()
+        self.setUpKeyAndCert(create=False)
+        self.assertFalse(self.validateKeyAndCert())
+        self.openArchive("test", "1.0", "amd64")
+        self.archive.add_file("1.0/empty.efi", "")
+        upload = self.process()
+        self.assertTrue(os.path.exists(os.path.join(
+            self.getUefiPath("test", "amd64"), "1.0", "empty.efi")))
+        self.assertTrue(self.validateKeyAndCert())
+        self.assertEqual(2, upload.execute_cmd.call_count)
+        self.assertTrue(self.validateCmdUefiKeygen(upload.execute_cmd.calls[0]))
+        self.assertTrue(self.validateCmdSbsign(upload.execute_cmd.calls[1], "empty.efi"))

=== 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-04-27 10:58:39 +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,13 +95,71 @@
                 if filename.endswith(".efi"):
                     yield os.path.join(dirpath, filename)
 
+    def genUefiKeys(self):
+        old_mask = os.umask(0o022)
+        try:
+            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.basename(self.archiveroot)
+            owner_name   = os.path.basename(os.path.dirname(directory))
+            common_name  = '/CN=PPA ' + owner_name + ' ' + archive_name + '/'
+
+            os.umask(0o077)
+            cmdl = [ 'openssl', 'req', '-new', '-x509', '-newkey', 'rsa:2048',
+                     '-subj', common_name, '-keyout', self.key,
+                     '-out', self.cert, '-days', '3650', '-nodes', '-sha256' ]
+            self.execute_cmd(cmdl)
+
+            if os.path.exists(self.cert):
+                os.chmod(self.cert, 0o644)
+
+        finally:
+            os.umask(old_mask)
+
+    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.genUefiKeys()
+                
+            # 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 getSigningCommand(self, image):
         """Return the command used to sign an image."""
-        return ["sbsign", "--key", self.key, "--cert", self.cert, image]
+
+        (key, cert) = self.getUefiKeys()
+        if not key or not cert:
+            return None
+
+        return ["sbsign", "--key", key, "--cert", cert, image]
+
+    def execute_cmd(self, cmdl):
+        """Execute a signing command."""
+        return subprocess.call(cmdl)
 
     def sign(self, image):
         """Sign an image."""
-        if subprocess.call(self.getSigningCommand(image)) != 0:
+        cmdl = self.getSigningCommand(image)
+        if cmdl == None or self.execute_cmd(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:
@@ -120,11 +171,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.sign(efi_filename)
 
     def shouldInstall(self, filename):
         return filename.startswith("%s/" % self.version)