Merge pull request #6244 from hexagonrecursion/bp-keyfile

Backport: Disallow overwriting of existing keyfiles on init

src/borg/crypto/key.py

@@ -687,12 +687,12 @@ def create(cls, repository, args):
         key.init_from_random_data()
         key.init_ciphers()
         target = key.get_new_target(args)
-        key.save(target, passphrase)
+        key.save(target, passphrase, create=True)
         logger.info('Key in "%s" created.' % target)
         logger.info('Keep this key safe. Your data will be inaccessible without it.')
         return key
 
-    def save(self, target, passphrase):
+    def save(self, target, passphrase, create=False):
         raise NotImplementedError
 
     def get_new_target(self, args):
@@ -753,7 +753,12 @@ def load(self, target, passphrase):
             self.target = target
         return success
 
-    def save(self, target, passphrase):
+    def save(self, target, passphrase, create=False):
+        if create and os.path.isfile(target):
+            # if a new keyfile key repository is created, ensure that an existing keyfile of another
+            # keyfile key repo is not accidentally overwritten by careless use of the BORG_KEY_FILE env var.
+            # see issue #6036
+            raise Error('Aborting because key in "%s" already exists.' % target)
         key_data = self._save(passphrase)
         with SaveFile(target) as fd:
             fd.write('%s %s\n' % (self.FILE_ID, bin_to_hex(self.repository_id)))
@@ -793,7 +798,7 @@ def load(self, target, passphrase):
             self.target = target
         return success
 
-    def save(self, target, passphrase):
+    def save(self, target, passphrase, create=False):
         self.logically_encrypted = passphrase != ''
         key_data = self._save(passphrase)
         key_data = key_data.encode('utf-8')  # remote repo: msgpack issue #99, giving bytes
@@ -831,8 +836,8 @@ def load(self, target, passphrase):
         self.logically_encrypted = False
         return success
 
-    def save(self, target, passphrase):
-        super().save(target, passphrase)
+    def save(self, target, passphrase, create=False):
+        super().save(target, passphrase, create=create)
         self.logically_encrypted = False
 
     def extract_nonce(self, payload):

src/borg/testsuite/archiver.py

@@ -32,6 +32,7 @@
     pass
 
 import borg
+import borg.helpers
 from .. import xattr, helpers, platform
 from ..archive import Archive, ChunkBuffer, flags_noatime, flags_normal
 from ..archiver import Archiver, parse_storage_quota, PURE_PYTHON_MSGPACK_WARNING
@@ -2572,6 +2573,25 @@ def test_init_nested_repositories(self):
             with pytest.raises(Repository.AlreadyExists):
                 self.cmd('init', '--encryption=repokey', self.repository_location + '/nested')
 
+    def test_init_refuse_to_overwrite_keyfile(self):
+        """BORG_KEY_FILE=something borg init should quit if "something" already exists.
+
+        See https://github.com/borgbackup/borg/pull/6046"""
+        keyfile = os.path.join(self.tmpdir, 'keyfile')
+        with environment_variable(BORG_KEY_FILE=keyfile):
+            self.cmd('init', '--encryption=keyfile', self.repository_location + '0')
+            with open(keyfile) as file:
+                before = file.read()
+            arg = ('init', '--encryption=keyfile', self.repository_location + '1')
+            if self.FORK_DEFAULT:
+                self.cmd(*arg, exit_code=2)
+            else:
+                with pytest.raises(borg.helpers.Error):
+                    self.cmd(*arg)
+            with open(keyfile) as file:
+                after = file.read()
+            assert before == after
+
     def check_cache(self):
         # First run a regular borg check
         self.cmd('check', self.repository_location)