Store secrets keys and SSH keys in Zookeeper

As a preparation for the HA scheduler, project secrets keys and SSH keys
will now also be stored in Zookeeper. All private data in Zookeeper will
be encrypted at rest.

Existing keys on the filesystem will be automatically imported into
Zookeeper and new keys will still be available as files for backup.

Change-Id: I2a7d1a555f1db1f2178d3bb2f06756ecc8bc7a81

doc/source/discussion/components.rst

@@ -115,6 +115,7 @@ An example ``zuul.conf``:
    status_url=https://zuul.example.com/status
 
    [scheduler]
+   key_store_password=MY_SECRET_PASSWORD
    log_config=/etc/zuul/scheduler-logging.yaml
 
 A minimal Zuul system may consist of a :ref:`scheduler` and
@@ -304,6 +305,11 @@ The following sections of ``zuul.conf`` are used by the scheduler:
 
 .. attr:: scheduler
 
+   .. attr:: key_store_password
+      :required:
+
+      Encryption password for private data stored in Zookeeper.
+
    .. attr:: command_socket
       :default: /var/lib/zuul/scheduler.socket
 

doc/source/examples/etc_zuul/zuul.conf

@@ -11,6 +11,7 @@ tls_key=/var/certs/keys/clientkey.pem
 tls_ca=/var/certs/certs/cacert.pem
 
 [scheduler]
+key_store_password=secret
 tenant_config=/etc/zuul/main.yaml
 
 [connection "gerrit"]

doc/source/howtos/installation.rst

@@ -132,6 +132,7 @@ service in Zuul, and a connection to Gerrit.
 **zuul.conf**::
 
     [scheduler]
+    key_store_password=secret
     tenant_config=/etc/zuul/main.yaml
 
     [gearman_server]

doc/source/howtos/zuul-from-scratch.rst

@@ -73,6 +73,7 @@ to appropriate values for your setup.
    listen_address=0.0.0.0
 
    [scheduler]
+   key_store_password=secret
    tenant_config=/etc/zuul/main.yaml
    EOF"
 

etc/zuul.conf-sample

@@ -19,6 +19,7 @@ start=true
 ;port=4730
 
 [scheduler]
+key_store_password=secret
 tenant_config=/etc/zuul/main.yaml
 log_config=/etc/zuul/logging.conf
 pidfile=/var/run/zuul/zuul.pid

releasenotes/notes/zookeeper-key-storage-a1ad32aa8d63b05f.yaml

@@ -0,0 +1,14 @@
+---
+features:
+  - |
+    Project secrets keys and SSH keys are now stored in Zookeeper. All private
+    data will be encrypted at rest, which requires a new mandatory setting
+    :attr:`scheduler.key_store_password` in ``zuul.conf``.
+
+    For backup purposes the secrets keys and SSH keys will still exist on the
+    local filesystem of the scheduler as before.
+upgrade:
+  - |
+    As project secrets keys and SSH keys are stored encrypted in Zookeeper the
+    new :attr:`scheduler.key_store_password` option in ``zuul.conf`` is
+    required. Please add it to your configuration.

tests/base.py

@@ -4254,6 +4254,9 @@ class ZuulTestCase(BaseTestCase):
         self.config.set(
             'scheduler', 'command_socket',
             os.path.join(self.test_root, 'scheduler.socket'))
+        if not self.config.has_option("scheduler", "key_store_password"):
+            self.config.set("scheduler", "key_store_password",
+                            uuid.uuid4().hex)
         self.config.set('merger', 'git_dir', self.merger_src_root)
         self.config.set('executor', 'git_dir', self.executor_src_root)
         self.config.set('executor', 'private_key_file', self.private_key_file)

tests/unit/test_keystorage.py

@@ -14,13 +14,16 @@
 
 import os
 import fixtures
+from unittest.mock import patch
 
+from zuul.lib import encryption
 from zuul.lib import keystorage
+from zuul.zk import ZooKeeperClient
 
 from tests.base import BaseTestCase
 
 
-class TestKeyStorage(BaseTestCase):
+class TestFileKeyStorage(BaseTestCase):
 
     def _setup_keys(self, root, connection_name, project_name):
         cn = os.path.join(root, connection_name)
@@ -50,7 +53,7 @@ class TestKeyStorage(BaseTestCase):
     def test_key_storage(self):
         root = self.useFixture(fixtures.TempDir()).path
         self._setup_keys(root, 'gerrit', 'org/example')
-        keystorage.KeyStorage(root)
+        keystorage.FileKeyStorage(root)
         self.assertFile(root, '.version', '1')
         self.assertPaths(root, [
             '.version',
@@ -65,4 +68,115 @@ class TestKeyStorage(BaseTestCase):
             'ssh/tenant',
         ])
         # It shouldn't need to upgrade this time
-        keystorage.KeyStorage(root)
+        keystorage.FileKeyStorage(root)
+
+
+class TestZooKeeperKeyStorage(BaseTestCase):
+
+    def setUp(self):
+        super().setUp()
+
+        self.setupZK()
+        self.zk_client = ZooKeeperClient(
+            self.zk_chroot_fixture.zk_hosts,
+            tls_cert=self.zk_chroot_fixture.zookeeper_cert,
+            tls_key=self.zk_chroot_fixture.zookeeper_key,
+            tls_ca=self.zk_chroot_fixture.zookeeper_ca)
+        self.addCleanup(self.zk_client.disconnect)
+        self.zk_client.connect()
+
+    def test_backup(self):
+        root = self.useFixture(fixtures.TempDir()).path
+        backup = keystorage.FileKeyStorage(root)
+        key_store = keystorage.ZooKeeperKeyStorage(
+            self.zk_client, password="DEADBEEF", backup=backup)
+
+        # Create keys in the backup keystore
+        backup_secrets_pk = encryption.serialize_rsa_private_key(
+            backup.getProjectSecretsKeys("github", "org/project")[0])
+        backup_ssh_keys = backup.getProjectSSHKeys("github", "org/project")
+
+        self.assertEqual(
+            encryption.serialize_rsa_private_key(
+                key_store.getProjectSecretsKeys("github", "org/project")[0]
+            ), backup_secrets_pk)
+        self.assertEqual(
+            key_store.getProjectSSHKeys("github", "org/project"),
+            backup_ssh_keys)
+
+        # Keys should initially not be in the backup keystore
+        self.assertFalse(
+            backup.hasProjectSecretsKeys("github", "org/project1"))
+        self.assertFalse(
+            backup.hasProjectSSHKeys("github", "org/project1"))
+
+        self.assertIsNotNone(
+            key_store.getProjectSecretsKeys("github", "org/project1"))
+        self.assertIsNotNone(
+            key_store.getProjectSSHKeys("github", "org/project1"))
+
+        # Keys should now also exist in the backup keystore
+        self.assertTrue(
+            backup.hasProjectSecretsKeys("github", "org/project1"))
+        self.assertTrue(
+            backup.hasProjectSSHKeys("github", "org/project1"))
+
+    def test_key_store_upgrade(self):
+        # Test that moving an unencrypted key on the file system to ZK
+        # (encrypted) works as expected and the backup keys stay the same.
+        root = self.useFixture(fixtures.TempDir()).path
+        backup = keystorage.FileKeyStorage(root)
+        key_store = keystorage.ZooKeeperKeyStorage(
+            self.zk_client, password="DECAFBAD", backup=backup)
+
+        # Create reference keys in the backup key store
+        ref_secrets_pk = encryption.serialize_rsa_private_key(
+            backup.getProjectSecretsKeys("github", "org/project")[0])
+        ref_ssh_keys = backup.getProjectSSHKeys("github", "org/project")
+
+        # Make sure we get the backup keys via the primary key store.
+        self.assertEqual(
+            encryption.serialize_rsa_private_key(
+                key_store.getProjectSecretsKeys("github", "org/project")[0]
+            ), ref_secrets_pk)
+        self.assertEqual(key_store.getProjectSSHKeys("github", "org/project"),
+                         ref_ssh_keys)
+
+        # Make sure we can still read the updates keys from the backup
+        # key store.
+        self.assertEqual(
+            encryption.serialize_rsa_private_key(
+                backup.getProjectSecretsKeys("github", "org/project")[0]
+            ), ref_secrets_pk)
+        self.assertEqual(backup.getProjectSSHKeys("github", "org/project"),
+                         ref_ssh_keys)
+
+        # Make sure that the backup key store is not used after keys have been
+        # written to the primary key store.
+        exc = AssertionError("Keys should not be loaded from backup store")
+        # with patch.object(backup, "getProjectSecretsKeys", side_effect=exc):
+        with patch.object(backup, "getProjectSecretsKeys", side_effect=exc):
+            self.assertEqual(
+                encryption.serialize_rsa_private_key(
+                    key_store.getProjectSecretsKeys("github", "org/project")[0]
+                ), ref_secrets_pk)
+
+        exc = AssertionError("SSH keys should not be loaded from backup store")
+        with patch.object(backup, "getProjectSSHKeys", side_effect=exc):
+            self.assertEqual(
+                key_store.getProjectSSHKeys("github", "org/project"),
+                ref_ssh_keys)
+
+    def test_without_backup(self):
+        key_store = keystorage.ZooKeeperKeyStorage(
+            self.zk_client, password="DECAFBAD")
+        secrets_pk = encryption.serialize_rsa_private_key(
+            key_store.getProjectSecretsKeys("github", "org/project")[0])
+        ssh_keys = key_store.getProjectSSHKeys("github", "org/project")
+
+        self.assertEqual(
+            encryption.serialize_rsa_private_key(
+                key_store.getProjectSecretsKeys("github", "org/project")[0]
+            ), secrets_pk)
+        self.assertEqual(key_store.getProjectSSHKeys("github", "org/project"),
+                         ssh_keys)

tests/unit/test_v3.py

@@ -25,7 +25,6 @@ from unittest import skip, skipIf
 import paramiko
 
 import zuul.configloader
-from zuul.lib import encryption
 from tests.base import (
     AnsibleZuulTestCase,
     ZuulTestCase,
@@ -3686,14 +3685,10 @@ class TestProjectKeys(ZuulTestCase):
             with open(os.path.join(FIXTURE_DIR, fn)) as i:
                 test_keys.append(i.read())
 
-        key_root = os.path.join(self.state_root, 'keys')
-        secrets_key_file = os.path.join(
-            key_root,
-            'secrets/project/gerrit/org/project/0.pem')
-        # Make sure that a proper key was created on startup
-        with open(secrets_key_file, "rb") as f:
-            private_secrets_key, public_secrets_key = \
-                encryption.deserialize_rsa_keypair(f.read())
+        keystore = self.scheds.first.sched.getKeyStorage()
+        private_secrets_key, public_secrets_key = (
+            keystore.getProjectSecretsKeys("gerrit", "org/project")
+        )
 
         # Make sure that we didn't just end up with the static fixture
         # key
@@ -3702,15 +3697,18 @@ class TestProjectKeys(ZuulTestCase):
         # Make sure it's the right length
         self.assertEqual(4096, private_secrets_key.key_size)
 
-        ssh_key_file = os.path.join(
-            key_root,
-            'ssh/project/gerrit/org/project/0.pem')
         # Make sure that a proper key was created on startup
-        ssh_key = paramiko.RSAKey.from_private_key_file(ssh_key_file)
+        private_ssh_key, public_ssh_key = (
+            keystore.getProjectSSHKeys("gerrit", "org/project")
+        )
 
         # Make sure that we didn't just end up with the static fixture
         # key
-        self.assertTrue(private_secrets_key not in test_keys)
+        self.assertTrue(private_ssh_key not in test_keys)
+
+        with io.StringIO(private_ssh_key) as o:
+            ssh_key = paramiko.RSAKey.from_private_key(
+                o, password=keystore.password)
 
         # Make sure it's the right length
         self.assertEqual(2048, ssh_key.get_bits())

zuul/configloader.py

@@ -32,7 +32,6 @@ import zuul.manager.independent
 import zuul.manager.supercedent
 import zuul.manager.serial
 from zuul.lib import encryption
-from zuul.lib.keystorage import KeyStorage
 from zuul.lib.logutil import get_annotated_logger
 from zuul.lib.re2util import filter_allowed_disallowed
 from zuul.zk.semaphore import SemaphoreHandler
@@ -2203,14 +2202,11 @@ class TenantParser(object):
 class ConfigLoader(object):
     log = logging.getLogger("zuul.ConfigLoader")
 
-    def __init__(self, connections, scheduler, merger, key_dir):
+    def __init__(self, connections, scheduler, merger, keystorage):
         self.connections = connections
         self.scheduler = scheduler
         self.merger = merger
-        if key_dir:
-            self.keystorage = KeyStorage(key_dir)
-        else:
-            self.keystorage = None
+        self.keystorage = keystorage
         self.tenant_parser = TenantParser(connections, scheduler,
                                           merger, self.keystorage)
         self.admin_rule_parser = AuthorizationRuleParser()

zuul/lib/encryption.py

@@ -37,22 +37,29 @@ def generate_rsa_keypair():
 
 
 # https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/#key-serialization
-def serialize_rsa_private_key(private_key):
+def serialize_rsa_private_key(private_key, password=None):
     """Serialize an RSA private key
 
     This returns a PEM-encoded serialized form of an RSA private key
-    suitable for storing on disk.  It is not password-protected.
+    suitable for storing on disk. In case a password is supplied the
+    encoded key will be encrypted.
 
     :arg private_key: A private key object as returned by
         :func:generate_rsa_keypair()
+    : arg password: A password in case the key should be encrypted
 
     :returns: A PEM-encoded string representation of the private key.
 
     """
+    if password is None:
+        encryption_algorithm = serialization.NoEncryption()
+    else:
+        encryption_algorithm = serialization.BestAvailableEncryption(password)
+
     return private_key.private_bytes(
         encoding=serialization.Encoding.PEM,
         format=serialization.PrivateFormat.TraditionalOpenSSL,
-        encryption_algorithm=serialization.NoEncryption()
+        encryption_algorithm=encryption_algorithm
     )
 
 
@@ -75,20 +82,22 @@ def serialize_rsa_public_key(public_key):
 
 
 # https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/#key-loading
-def deserialize_rsa_keypair(data):
+def deserialize_rsa_keypair(data, password=None):
     """Deserialize an RSA private key
 
     This deserializes an RSA private key and returns the keypair
-    (private and public) for use in decryption.
+    (private and public) for use in decryption. If the given key
+    is encrypted a password must be supplied.
 
     :arg data: A PEM-encoded serialized private key
+    :arg password: Optional password in case the private key is encrypted.
 
     :returns: A tuple (private_key, public_key)
 
     """
     private_key = serialization.load_pem_private_key(
         data,
-        password=None,
+        password=password,
         backend=default_backend()
     )
     public_key = private_key.public_key()

zuul/lib/keystorage.py

@@ -12,13 +12,18 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-import tempfile
+import abc
+import io
+import json
 import logging
 import os
+import tempfile
 
+import kazoo
 import paramiko
 
 from zuul.lib import encryption
+from zuul.zk import ZooKeeperBase
 
 RSA_KEY_SIZE = 2048
 
@@ -124,8 +129,33 @@ class MigrationV1(Migration):
         os.rmdir(tmpdir)
 
 
-class KeyStorage(object):
+class KeyStorage(abc.ABC):
     log = logging.getLogger("zuul.KeyStorage")
+
+    @abc.abstractmethod
+    def getProjectSSHKeys(self, connection_name, project_name):
+        """Return the private and public SSH keys for the project
+
+        A new key will be created if necessary.
+
+        :returns: A tuple containing the PEM encoded private key and
+                  base64 encoded public key.
+        """
+        pass
+
+    @abc.abstractmethod
+    def getProjectSecretsKeys(self, connection_name, project_name):
+        """Return the private and public secrets keys for the project
+
+        A new key will be created if necessary.
+
+        :returns: A tuple (private_key, public_key)
+        """
+        pass
+
+
+class FileKeyStorage(KeyStorage):
+    log = logging.getLogger("zuul.FileKeyStorage")
     current_version = MigrationV1
 
     def __init__(self, root):
@@ -149,74 +179,200 @@ class KeyStorage(object):
         return os.path.join(self.root, 'ssh', 'project',
                             connection, project, version + '.pem')
 
-    def getProjectSSHKeys(self, connection, project):
-        """Return the private and public SSH keys for the project
+    def hasProjectSSHKeys(self, connection_name, project_name):
+        return os.path.isfile(
+            self.getProjectSSHKeyFile(connection_name, project_name))
 
-        A new key will be created if necessary.
+    def setProjectSSHKey(self, connection_name, project_name, private_key):
+        private_key_file = self.getProjectSSHKeyFile(connection_name,
+                                                     project_name)
+        key_dir = os.path.dirname(private_key_file)
+        if not os.path.isdir(key_dir):
+            os.makedirs(key_dir, 0o700)
 
-        :returns: A tuple containing the PEM encoded private key and
-            base64 encoded public key.
+        private_key.write_private_key_file(private_key_file)
+        return private_key_file
 
-        """
+    def getProjectSSHKeys(self, connection_name, project_name):
+        private_key_file = self._ensureSSHKeyFile(connection_name,
+                                                  project_name)
 
-        private_key_file = self.getProjectSSHKeyFile(connection, project)
-        if not os.path.exists(private_key_file):
-            self.log.info(
-                "Generating SSH public key for project %s", project
-            )
-            self._createSSHKey(private_key_file)
         key = paramiko.RSAKey.from_private_key_file(private_key_file)
         with open(private_key_file, 'r') as f:
             private_key = f.read()
         public_key = key.get_base64()
+
         return (private_key, 'ssh-rsa ' + public_key)
 
-    def _createSSHKey(self, fn):
-        key_dir = os.path.dirname(fn)
+    def _ensureSSHKeyFile(self, connection_name, project_name):
+        private_key_file = self.getProjectSSHKeyFile(connection_name,
+                                                     project_name)
+        if os.path.exists(private_key_file):
+            return private_key_file
+
+        self.log.info("Generating SSH public key for project %s", project_name)
+        pk = paramiko.RSAKey.generate(bits=RSA_KEY_SIZE)
+        return self.setProjectSSHKey(connection_name, project_name, pk)
+
+    def hasProjectSecretsKeys(self, connection_name, project_name):
+        return os.path.isfile(
+            self.getProjectSecretsKeyFile(connection_name, project_name))
+
+    def setProjectSecretsKey(self, connection_name, project_name, private_key):
+        filename = self.getProjectSecretsKeyFile(connection_name, project_name)
+        key_dir = os.path.dirname(filename)
         if not os.path.isdir(key_dir):
             os.makedirs(key_dir, 0o700)
 
-        pk = paramiko.RSAKey.generate(bits=RSA_KEY_SIZE)
-        pk.write_private_key_file(fn)
+        # Dump keys to filesystem.  We only save the private key
+        # because the public key can be constructed from it.
+        self.log.info("Saving RSA keypair for project %s to %s",
+                      project_name, filename)
+
+        pem_private_key = encryption.serialize_rsa_private_key(private_key)
+        # Ensure private key is read/write for zuul user only.
+        with open(os.open(filename,
+                          os.O_CREAT | os.O_WRONLY, 0o600), 'wb') as f:
+            f.write(pem_private_key)
+        return filename
 
     def getProjectSecretsKeys(self, connection_name, project_name):
-        """Return the private and public secrets keys for the project
-
-        A new key will be created if necessary.
-
-        :returns: A tuple (private_key, public_key)
-        """
-
-        private_key_file = self._ensureKeyFile(connection_name, project_name)
+        private_key_file = self._ensureSecretsKeyFile(connection_name,
+                                                      project_name)
 
         # Load keypair
         with open(private_key_file, "rb") as f:
             return encryption.deserialize_rsa_keypair(f.read())
 
-    def _ensureKeyFile(self, connection_name, project_name):
-        filename = self.getProjectSecretsKeyFile(
-            connection_name, project_name
-        )
+    def _ensureSecretsKeyFile(self, connection_name, project_name):
+        filename = self.getProjectSecretsKeyFile(connection_name, project_name)
         if os.path.isfile(filename):
             return filename
 
-        key_dir = os.path.dirname(filename)
-        if not os.path.isdir(key_dir):
-            os.makedirs(key_dir, 0o700)
-
         self.log.info("Generating RSA keypair for project %s", project_name)
         private_key, public_key = encryption.generate_rsa_keypair()
-        pem_private_key = encryption.serialize_rsa_private_key(private_key)
 
-        # Dump keys to filesystem.  We only save the private key
-        # because the public key can be constructed from it.
-        self.log.info(
-            "Saving RSA keypair for project %s to %s", project_name, filename
+        return self.setProjectSecretsKey(connection_name, project_name,
+                                         private_key)
+
+
+class ZooKeeperKeyStorage(ZooKeeperBase, KeyStorage):
+    log = logging.getLogger("zuul.ZooKeeperKeyStorage")
+    SECRETS_PATH = "/keystorage/{}/{}/secrets"
+    SSH_PATH = "/keystorage/{}/{}/ssh"
+
+    def __init__(self, zookeeper_client, password, backup=None):
+        super().__init__(zookeeper_client)
+        self.password = password
+        self.password_bytes = password.encode("utf-8")
+        self.backup = backup
+
+    def getProjectSSHKeys(self, connection_name, project_name):
+        key_path = self.SSH_PATH.format(connection_name, project_name)
+
+        try:
+            key = self._getSSHKey(key_path)
+        except kazoo.exceptions.NoNodeError:
+            self.log.debug("Could not find existing SSH key for %s/%s",
+                           connection_name, project_name)
+            if self.backup and self.backup.hasProjectSSHKeys(
+                connection_name, project_name
+            ):
+                self.log.debug("Using SSH key for %s/%s from backup key store",
+                               connection_name, project_name)
+                pk, _ = self.backup.getProjectSSHKeys(connection_name,
+                                                      project_name)
+                with io.StringIO(pk) as o:
+                    key = paramiko.RSAKey.from_private_key(o)
+            else:
+                self.log.debug("Generating a new SSH key for %s/%s",
+                               connection_name, project_name)
+                key = paramiko.RSAKey.generate(bits=RSA_KEY_SIZE)
+
+            try:
+                self._storeSSHKey(key_path, key)
+            except kazoo.exceptions.NodeExistsError:
+                # Handle race condition between multiple schedulers
+                # creating the same SSH key.
+                key = self._getSSHKey(key_path)
+
+        # Make sure the SSH key is also stored in the backup keystore
+        if self.backup:
+            self.backup.setProjectSSHKey(connection_name, project_name, key)
+
+        with io.StringIO() as o:
+            key.write_private_key(o)
+            private_key = o.getvalue()
+        public_key = "ssh-rsa {}".format(key.get_base64())
+
+        return private_key, public_key
+
+    def _getSSHKey(self, key_path):
+        raw_pk, _ = self.kazoo_client.get(key_path)
+        encrypted_key = raw_pk.decode("utf-8")
+        with io.StringIO(encrypted_key) as o:
+            return paramiko.RSAKey.from_private_key(o, self.password)
+
+    def _storeSSHKey(self, key_path, key):
+        with io.StringIO() as o:
+            key.write_private_key(o, self.password)
+            private_key = o.getvalue()
+        raw_pk = private_key.encode("utf-8")
+        self.kazoo_client.create(key_path, value=raw_pk, makepath=True)
+
+    def getProjectSecretsKeys(self, connection_name, project_name):
+        key_path = self.SECRETS_PATH.format(connection_name, project_name)
+
+        try:
+            pem_private_key, _ = self._getSecretsKeys(key_path)
+        except kazoo.exceptions.NoNodeError:
+            self.log.debug("Could not find existing secrets key for %s/%s",
+                           connection_name, project_name)
+            if self.backup and self.backup.hasProjectSecretsKeys(
+                    connection_name, project_name):
+                self.log.debug(
+                    "Using secrets key for %s/%s from backup key store",
+                    connection_name, project_name)
+                private_key, public_key = self.backup.getProjectSecretsKeys(
+                    connection_name, project_name)
+            else:
+                self.log.debug("Generating a new secrets key for %s/%s",
+                               connection_name, project_name)
+                private_key, public_key = encryption.generate_rsa_keypair()
+
+            pem_private_key = encryption.serialize_rsa_private_key(
+                private_key, self.password_bytes)
+            pem_public_key = encryption.serialize_rsa_public_key(public_key)
+            try:
+                self._storeSecretsKeys(key_path, pem_private_key,
+                                       pem_public_key)
+            except kazoo.exceptions.NodeExistsError:
+                # Handle race condition between multiple schedulers
+                # creating the same secrets key.
+                pem_private_key, _ = self._getSecretsKeys(key_path)
+
+        private_key, public_key = encryption.deserialize_rsa_keypair(
+            pem_private_key, self.password_bytes)
+
+        # Make sure the private key is also stored in the backup keystore
+        if self.backup:
+            self.backup.setProjectSecretsKey(connection_name, project_name,
+                                             private_key)
+
+        return private_key, public_key
+
+    def _getSecretsKeys(self, key_path):
+        data, _ = self.kazoo_client.get(key_path)
+        keys = json.loads(data)
+        return (
+            keys["private_key"].encode("utf-8"),
+            keys["public_key"].encode("utf-8"),
         )
 
-        # Ensure private key is read/write for zuul user only.
-        with open(os.open(filename,
-                          os.O_CREAT | os.O_WRONLY, 0o600), 'wb') as f:
-            f.write(pem_private_key)
-
-        return filename
+    def _storeSecretsKeys(self, key_path, private_key, public_key):
+        keys = {
+            "private_key": private_key.decode("utf-8"),
+            "public_key": public_key.decode("utf-8"),
+        }
+        data = json.dumps(keys).encode("utf-8")
+        self.kazoo_client.create(key_path, value=data, makepath=True)

zuul/scheduler.py

@@ -35,6 +35,7 @@ from zuul.lib import commandsocket
 from zuul.lib.ansible import AnsibleManager
 from zuul.lib.config import get_default
 from zuul.lib.gear_utils import getGearmanFunctions
+from zuul.lib.keystorage import FileKeyStorage, ZooKeeperKeyStorage
 from zuul.lib.logutil import get_annotated_logger
 from zuul.lib.queue import NamedQueue
 from zuul.lib.statsd import get_statsd, normalize_statsd_name
@@ -592,6 +593,12 @@ class Scheduler(threading.Thread):
             os.mkdir(d)
         return d
 
+    def _get_key_store_password(self):
+        try:
+            return self.config["scheduler"]["key_store_password"]
+        except KeyError:
+            raise RuntimeError("No key store password configured!")
+
     def _get_key_dir(self):
         state_dir = get_default(self.config, 'scheduler', 'state_dir',
                                 '/var/lib/zuul', expand_user=True)
@@ -605,6 +612,12 @@ class Scheduler(threading.Thread):
                             "current mode is %o" % (key_dir, mode))
         return key_dir
 
+    def getKeyStorage(self):
+        file_key_store = FileKeyStorage(self._get_key_dir())
+        return ZooKeeperKeyStorage(self.zk_client,
+                                   password=self._get_key_store_password(),
+                                   backup=file_key_store)
+
     def _checkTenantSourceConf(self, config):
         tenant_config = None
         script = False
@@ -649,7 +662,7 @@ class Scheduler(threading.Thread):
 
             loader = configloader.ConfigLoader(
                 self.connections, self, self.merger,
-                self._get_key_dir())
+                self.getKeyStorage())
             tenant_config, script = self._checkTenantSourceConf(self.config)
             self.unparsed_abide = loader.readConfig(
                 tenant_config, from_script=script)
@@ -694,7 +707,7 @@ class Scheduler(threading.Thread):
 
             loader = configloader.ConfigLoader(
                 self.connections, self, self.merger,
-                self._get_key_dir())
+                self.getKeyStorage())
             tenant_config, script = self._checkTenantSourceConf(self.config)
             old_unparsed_abide = self.unparsed_abide
             self.unparsed_abide = loader.readConfig(
@@ -749,7 +762,7 @@ class Scheduler(threading.Thread):
             old_tenant = self.abide.tenants[event.tenant_name]
             loader = configloader.ConfigLoader(
                 self.connections, self, self.merger,
-                self._get_key_dir())
+                self.getKeyStorage())
             abide = loader.reloadTenant(
                 self.abide, old_tenant, self.ansible_manager)
             tenant = abide.tenants[event.tenant_name]