borg key change-location

src/borg/archiver.py

@@ -44,7 +44,8 @@
     from .cache import Cache, assert_secure, SecurityManager
     from .constants import *  # NOQA
     from .compress import CompressionSpec
-    from .crypto.key import key_creator, key_argument_names, tam_required_file, tam_required, RepoKey
+    from .crypto.key import key_creator, key_argument_names, tam_required_file, tam_required
+    from .crypto.key import RepoKey, KeyfileKey, Blake2RepoKey, Blake2KeyfileKey
     from .crypto.keymanager import KeyManager
     from .helpers import EXIT_SUCCESS, EXIT_WARNING, EXIT_ERROR, EXIT_SIGNAL_BASE
     from .helpers import Error, NoManifestError, set_ec
@@ -363,6 +364,62 @@ def do_change_passphrase(self, args, repository, manifest, key):
             logger.info('Key location: %s', key.find_key())
         return EXIT_SUCCESS
 
+    @with_repository(exclusive=True, manifest=True, cache=True, compatibility=(Manifest.Operation.CHECK,))
+    def do_change_location(self, args, repository, manifest, key, cache):
+        """Change repository key location"""
+        if not hasattr(key, 'change_passphrase'):
+            print('This repository is not encrypted, cannot change the key location.')
+            return EXIT_ERROR
+
+        if args.key_mode == 'keyfile':
+            if isinstance(key, RepoKey):
+                key_new = KeyfileKey(repository)
+            elif isinstance(key, Blake2RepoKey):
+                key_new = Blake2KeyfileKey(repository)
+            elif isinstance(key, (KeyfileKey, Blake2KeyfileKey)):
+                print(f"Location already is {args.key_mode}")
+                return EXIT_SUCCESS
+            else:
+                raise Error("Unsupported key type")
+        if args.key_mode == 'repokey':
+            if isinstance(key, KeyfileKey):
+                key_new = RepoKey(repository)
+            elif isinstance(key, Blake2KeyfileKey):
+                key_new = Blake2RepoKey(repository)
+            elif isinstance(key, (RepoKey, Blake2RepoKey)):
+                print(f"Location already is {args.key_mode}")
+                return EXIT_SUCCESS
+            else:
+                raise Error("Unsupported key type")
+
+        for name in ('repository_id', 'enc_key', 'enc_hmac_key', 'id_key', 'chunk_seed',
+                     'tam_required', 'nonce_manager', 'cipher'):
+            value = getattr(key, name)
+            setattr(key_new, name, value)
+
+        key_new.target = key_new.get_new_target(args)
+        key_new.save(key_new.target, key._passphrase, create=True)  # save with same passphrase
+
+        # rewrite the manifest with the new key, so that the key-type byte of the manifest changes
+        manifest.key = key_new
+        manifest.write()
+        repository.commit(compact=False)
+
+        # we need to rewrite cache config and security key-type info,
+        # so that the cached key-type will match the repo key-type.
+        cache.begin_txn()  # need to start a cache transaction, otherwise commit() does nothing.
+        cache.key = key_new
+        cache.commit()
+
+        loc = key_new.find_key() if hasattr(key_new, 'find_key') else None
+        if args.keep:
+            logger.info(f'Key copied to {loc}')
+        else:
+            key.remove(key.target)  # remove key from current location
+            logger.info(f'Key moved to {loc}')
+
+        return EXIT_SUCCESS
+
     @with_repository(lock=False, exclusive=False, manifest=False, cache=False)
     def do_key_export(self, args, repository):
         """Export the repository key for backup"""
@@ -4250,6 +4307,28 @@ def define_borg_mount(parser):
         subparser.add_argument('location', metavar='REPOSITORY', nargs='?', default='',
                                type=location_validator(archive=False))
 
+        change_location_epilog = process_epilog("""
+        Change the location of a borg key. The key can be stored at different locations:
+
+        keyfile: locally, usually in the home directory
+        repokey: inside the repo (in the repo config)
+
+        Note: this command does NOT change the crypto algorithms, just the key location,
+              thus you must ONLY give the key location (keyfile or repokey).
+        """)
+        subparser = key_parsers.add_parser('change-location', parents=[common_parser], add_help=False,
+                                          description=self.do_change_location.__doc__,
+                                          epilog=change_location_epilog,
+                                          formatter_class=argparse.RawDescriptionHelpFormatter,
+                                          help='change key location')
+        subparser.set_defaults(func=self.do_change_location)
+        subparser.add_argument('location', metavar='REPOSITORY', nargs='?', default='',
+                               type=location_validator(archive=False))
+        subparser.add_argument('key_mode', metavar='KEY_LOCATION', choices=('repokey', 'keyfile'),
+                               help='select key location')
+        subparser.add_argument('--keep', dest='keep', action='store_true',
+                               help='keep the key also at the current location (default: remove it)')
+
         # borg list
         list_epilog = process_epilog("""
         This command lists the contents of a repository or an archive.

src/borg/crypto/key.py

@@ -634,6 +634,13 @@ def save(self, target, passphrase, create=False):
             target.save_key(key_data)
             self.target = target
 
+    def remove(self, target):
+        if self.STORAGE == KeyBlobStorage.KEYFILE:
+            os.remove(target)
+
+        if self.STORAGE == KeyBlobStorage.REPO:
+            target.save_key(b'')  # save empty key (no new api at remote repo necessary)
+
 
 class PassphraseKey:
     # this is only a stub, repos with this mode could not be created any more since borg 1.0, see #97.

src/borg/repository.py

@@ -334,11 +334,13 @@ def save_config(self, path, config):
     def save_key(self, keydata):
         assert self.config
         keydata = keydata.decode('utf-8')  # remote repo: msgpack issue #99, getting bytes
+        # note: saving an empty key means that there is no repokey any more
         self.config.set('repository', 'key', keydata)
         self.save_config(self.path, self.config)
 
     def load_key(self):
         keydata = self.config.get('repository', 'key')
+        # note: if we return an empty string, it means there is no repo key
         return keydata.encode('utf-8')  # remote repo: msgpack issue #99, returning bytes
 
     def get_free_nonce(self):

src/borg/testsuite/archiver.py

@@ -2490,6 +2490,38 @@ def test_change_passphrase(self):
         os.environ['BORG_PASSPHRASE'] = 'newpassphrase'
         self.cmd('list', self.repository_location)
 
+    def test_change_location_to_keyfile(self):
+        self.cmd('init', '--encryption=repokey', self.repository_location)
+        log = self.cmd('info', self.repository_location)
+        assert '(repokey)' in log
+        self.cmd('key', 'change-location', self.repository_location, 'keyfile')
+        log = self.cmd('info', self.repository_location)
+        assert '(key file)' in log
+
+    def test_change_location_to_b2keyfile(self):
+        self.cmd('init', '--encryption=repokey-blake2', self.repository_location)
+        log = self.cmd('info', self.repository_location)
+        assert '(repokey BLAKE2b)' in log
+        self.cmd('key', 'change-location', self.repository_location, 'keyfile')
+        log = self.cmd('info', self.repository_location)
+        assert '(key file BLAKE2b)' in log
+
+    def test_change_location_to_repokey(self):
+        self.cmd('init', '--encryption=keyfile', self.repository_location)
+        log = self.cmd('info', self.repository_location)
+        assert '(key file)' in log
+        self.cmd('key', 'change-location', self.repository_location, 'repokey')
+        log = self.cmd('info', self.repository_location)
+        assert '(repokey)' in log
+
+    def test_change_location_to_b2repokey(self):
+        self.cmd('init', '--encryption=keyfile-blake2', self.repository_location)
+        log = self.cmd('info', self.repository_location)
+        assert '(key file BLAKE2b)' in log
+        self.cmd('key', 'change-location', self.repository_location, 'repokey')
+        log = self.cmd('info', self.repository_location)
+        assert '(repokey BLAKE2b)' in log
+
     def test_break_lock(self):
         self.cmd('init', '--encryption=repokey', self.repository_location)
         self.cmd('break-lock', self.repository_location)