Allow importing of custom keystore entroy blob by ursula init CLI and Keystore API.

nucypher/cli/commands/ursula.py

@@ -161,7 +161,7 @@ class UrsulaConfigOptions:
                 # TODO: Exit codes (not only for this, but for other exceptions)
                 return click.get_current_context().exit(1)
 
-    def generate_config(self, emitter, config_root, force):
+    def generate_config(self, emitter, config_root, force, secret_key):
 
         if self.dev:
             raise RuntimeError('Persistent configurations cannot be created in development mode.')
@@ -180,6 +180,7 @@ class UrsulaConfigOptions:
             self.rest_host = collect_worker_ip_address(emitter, network=self.domain, force=force)
 
         return UrsulaConfiguration.generate(password=get_nucypher_password(emitter=emitter, confirm=True),
+                                            secret_key=bytes.fromhex(secret_key),
                                             config_root=config_root,
                                             rest_host=self.rest_host,
                                             rest_port=self.rest_port,
@@ -295,7 +296,8 @@ def ursula():
 @option_force
 @option_config_root
 @group_general_config
-def init(general_config, config_options, force, config_root):
+@click.option('--secret-key', help="An custom pre-secured secret hex blob to use for key derivations", type=click.STRING)
+def init(general_config, config_options, force, config_root, secret_key):
     """Create a new Ursula node configuration."""
     emitter = setup_emitter(general_config, config_options.worker_address)
     _pre_launch_warnings(emitter, dev=None, force=force)
@@ -305,7 +307,7 @@ def init(general_config, config_options, force, config_root):
         raise click.BadOptionUsage('--provider', message="--provider is required to initialize a new ursula.")
     if not config_options.federated_only and not config_options.domain:
         config_options.domain = select_network(emitter)
-    ursula_config = config_options.generate_config(emitter, config_root, force)
+    ursula_config = config_options.generate_config(emitter, config_root, force, secret_key)
     filepath = ursula_config.to_configuration_file()
     paint_new_installation_help(emitter, new_configuration=ursula_config, filepath=filepath)
 

nucypher/config/base.py

@@ -579,10 +579,10 @@ class CharacterConfiguration(BaseConfiguration):
         return super().update(filepath=self.config_file_location, **kwargs)
 
     @classmethod
-    def generate(cls, password: str, *args, **kwargs):
+    def generate(cls, password: str, secret_key: bytes, *args, **kwargs):
         """Shortcut: Hook-up a new initial installation and configuration."""
         node_config = cls(dev_mode=False, *args, **kwargs)
-        node_config.initialize(password=password)
+        node_config.initialize(secret_key=secret_key, password=password)
         return node_config
 
     def cleanup(self) -> None:
@@ -769,7 +769,7 @@ class CharacterConfiguration(BaseConfiguration):
                 power_ups.append(power_up)
         return power_ups
 
-    def initialize(self, password: str) -> Path:
+    def initialize(self, password: str, secret_key: Optional[bytes] = None) -> Path:
         """Initialize a new configuration and write installation files to disk."""
 
         # Development
@@ -780,7 +780,7 @@ class CharacterConfiguration(BaseConfiguration):
         # Persistent
         else:
             self._ensure_config_root_exists()
-            self.write_keystore(password=password, interactive=self.MNEMONIC_KEYSTORE)
+            self.write_keystore(secret_key=secret_key, password=password, interactive=self.MNEMONIC_KEYSTORE)
 
         self._cache_runtime_filepaths()
         self.node_storage.initialize()
@@ -794,8 +794,15 @@ class CharacterConfiguration(BaseConfiguration):
         self.log.debug(message)
         return self.config_root
 
-    def write_keystore(self, password: str, interactive: bool = True) -> Keystore:
-        self.__keystore = Keystore.generate(password=password, keystore_dir=self.keystore_dir, interactive=interactive)
+    def write_keystore(self, password: str, secret_key: Optional[bytes] = None, interactive: bool = True) -> Keystore:
+        if secret_key:
+            self.__keystore = Keystore.import_secure(secret=secret_key,
+                                                     password=password,
+                                                     keystore_dir=self.keystore_dir)
+        else:
+            self.__keystore = Keystore.generate(password=password,
+                                                keystore_dir=self.keystore_dir,
+                                                interactive=interactive)
         return self.keystore
 
     @classmethod

nucypher/crypto/keystore.py

@@ -320,6 +320,17 @@ class Keystore:
         instance = cls(keystore_path=filepath)
         return instance
 
+    @classmethod
+    def import_secure(cls, secret: bytes, password: str, keystore_dir: Optional[Path] = None) -> 'Keystore':
+        """
+        Generate a Keystore using a a custom pre-secured entropy blob.
+        This method of keystore creation does not generate a mnemonic phrase - it is assumed
+        that the provided blob is recoverable and secure.
+        """
+        path = Keystore.__save(secret=secret, password=password, keystore_dir=keystore_dir)
+        keystore = cls(keystore_path=path)
+        return keystore
+
     @classmethod
     def restore(cls, words: str, password: str, keystore_dir: Optional[Path] = None) -> 'Keystore':
         """Restore a keystore from seed words"""

tests/unit/crypto/test_keystore.py

@@ -241,6 +241,33 @@ def test_restore_keystore_from_mnemonic(tmpdir, mocker):
     assert keystore._Keystore__secret == secret
 
 
+def test_import_custom_keystore(tmpdir):
+
+    # Too short - 32 bytes is required
+    custom_secret = b'tooshort'
+    with pytest.raises(ValueError, match="Entropy must be at least 32 bytes."):
+        _keystore = Keystore.import_secure(secret=custom_secret,
+                                           password=INSECURE_DEVELOPMENT_PASSWORD,
+                                           keystore_dir=tmpdir)
+
+    # Import private key
+    custom_secret = os.urandom(32)  # not secure but works
+    keystore = Keystore.import_secure(secret=custom_secret,
+                                      password=INSECURE_DEVELOPMENT_PASSWORD,
+                                      keystore_dir=tmpdir)
+    keystore.unlock(password=INSECURE_DEVELOPMENT_PASSWORD)
+    assert keystore._Keystore__secret == custom_secret
+    keystore.lock()
+
+    path = keystore.keystore_path
+    del keystore
+
+    # Restore custom secret from encrypted keystore file
+    keystore = Keystore(keystore_path=path)
+    keystore.unlock(password=INSECURE_DEVELOPMENT_PASSWORD)
+    assert keystore._Keystore__secret == custom_secret
+
+
 def test_derive_signing_power(tmpdir):
     keystore = Keystore.generate(INSECURE_DEVELOPMENT_PASSWORD, keystore_dir=tmpdir)
     keystore.unlock(password=INSECURE_DEVELOPMENT_PASSWORD)