Merge pull request #2489 from enkore/issue/2169

consider repokey w/o passphrase == unencrypted

docs/changes.rst

@@ -128,6 +128,15 @@ The best check that everything is ok is to run a dry-run extraction::
 Changelog
 =========
 
+Version 1.1.0b6 (unreleased)
+----------------------------
+
+Compatibility notes:
+
+- Repositories in a repokey mode with a blank passphrase are now treated
+  as unencrypted repositories for security checks
+  (e.g. BORG_UNKNOWN_UNENCRYPTED_REPO_ACCESS_IS_OK).
+
 Version 1.1.0b5 (2017-04-30)
 ----------------------------
 

src/borg/cache.py

@@ -130,7 +130,7 @@ class SecurityManager:
             self.save(manifest, key, cache)
 
     def assert_access_unknown(self, warn_if_unencrypted, key):
-        if warn_if_unencrypted and isinstance(key, PlaintextKey) and not self.known():
+        if warn_if_unencrypted and not key.passphrase_protected and not self.known():
             msg = ("Warning: Attempting to access a previously unknown unencrypted repository!\n" +
                    "Do you want to continue? [yN] ")
             if not yes(msg, false_msg="Aborting.", invalid_msg="Invalid answer, aborting.",

src/borg/crypto/key.py

@@ -234,6 +234,7 @@ class PlaintextKey(KeyBase):
     STORAGE = KeyBlobStorage.NO_STORAGE
 
     chunk_seed = 0
+    passphrase_protected = False
 
     def __init__(self, repository):
         super().__init__(repository)
@@ -329,6 +330,8 @@ class AESKeyBase(KeyBase):
 
     MAC = hmac_sha256
 
+    passphrase_protected = True
+
     def encrypt(self, chunk):
         data = self.compressor.compress(chunk)
         self.nonce_manager.ensure_reservation(num_aes_blocks(len(data)))
@@ -700,6 +703,10 @@ class RepoKey(ID_HMAC_SHA_256, KeyfileKeyBase):
         return self.repository
 
     def load(self, target, passphrase):
+        # While the repository is encrypted, we consider a repokey repository with a blank
+        # passphrase an unencrypted repository.
+        self.passphrase_protected = passphrase != ''
+
         # what we get in target is just a repo location, but we already have the repo obj:
         target = self.repository
         key_data = target.load_key()
@@ -710,6 +717,7 @@ class RepoKey(ID_HMAC_SHA_256, KeyfileKeyBase):
         return success
 
     def save(self, target, passphrase):
+        self.passphrase_protected = passphrase != ''
         key_data = self._save(passphrase)
         key_data = key_data.encode('utf-8')  # remote repo: msgpack issue #99, giving bytes
         target.save_key(key_data)

src/borg/testsuite/archiver.py

@@ -560,7 +560,8 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         if self.FORK_DEFAULT:
             self.cmd('create', self.repository_location + '::test.2', 'input', exit_code=EXIT_ERROR)
         else:
-            self.assert_raises(Cache.EncryptionMethodMismatch, lambda: self.cmd('create', self.repository_location + '::test.2', 'input'))
+            with pytest.raises(Cache.EncryptionMethodMismatch):
+                self.cmd('create', self.repository_location + '::test.2', 'input')
 
     def test_repository_swap_detection2(self):
         self.create_test_files()
@@ -573,7 +574,8 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         if self.FORK_DEFAULT:
             self.cmd('create', self.repository_location + '_encrypted::test.2', 'input', exit_code=EXIT_ERROR)
         else:
-            self.assert_raises(Cache.RepositoryAccessAborted, lambda: self.cmd('create', self.repository_location + '_encrypted::test.2', 'input'))
+            with pytest.raises(Cache.RepositoryAccessAborted):
+                self.cmd('create', self.repository_location + '_encrypted::test.2', 'input')
 
     def test_repository_swap_detection_no_cache(self):
         self.create_test_files()
@@ -589,7 +591,8 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         if self.FORK_DEFAULT:
             self.cmd('create', self.repository_location + '::test.2', 'input', exit_code=EXIT_ERROR)
         else:
-            self.assert_raises(Cache.EncryptionMethodMismatch, lambda: self.cmd('create', self.repository_location + '::test.2', 'input'))
+            with pytest.raises(Cache.EncryptionMethodMismatch):
+                self.cmd('create', self.repository_location + '::test.2', 'input')
 
     def test_repository_swap_detection2_no_cache(self):
         self.create_test_files()
@@ -607,6 +610,30 @@ class ArchiverTestCase(ArchiverTestCaseBase):
             with pytest.raises(Cache.RepositoryAccessAborted):
                 self.cmd('create', self.repository_location + '_encrypted::test.2', 'input')
 
+    def test_repository_swap_detection_repokey_blank_passphrase(self):
+        # Check that a repokey repo with a blank passphrase is considered like a plaintext repo.
+        self.create_test_files()
+        # User initializes her repository with her passphrase
+        self.cmd('init', '--encryption=repokey', self.repository_location)
+        self.cmd('create', self.repository_location + '::test', 'input')
+        # Attacker replaces it with her own repository, which is encrypted but has no passphrase set
+        shutil.rmtree(self.repository_path)
+        with environment_variable(BORG_PASSPHRASE=''):
+            self.cmd('init', '--encryption=repokey', self.repository_location)
+            # Delete cache & security database, AKA switch to user perspective
+            self.cmd('delete', '--cache-only', self.repository_location)
+            repository_id = bin_to_hex(self._extract_repository_id(self.repository_path))
+            shutil.rmtree(get_security_dir(repository_id))
+        with environment_variable(BORG_PASSPHRASE=None):
+            # This is the part were the user would be tricked, e.g. she assumes that BORG_PASSPHRASE
+            # is set, while it isn't. Previously this raised no warning,
+            # since the repository is, technically, encrypted.
+            if self.FORK_DEFAULT:
+                self.cmd('create', self.repository_location + '::test.2', 'input', exit_code=EXIT_ERROR)
+            else:
+                with pytest.raises(Cache.CacheInitAbortedError):
+                    self.cmd('create', self.repository_location + '::test.2', 'input')
+
     def test_repository_move(self):
         self.cmd('init', '--encryption=repokey', self.repository_location)
         repository_id = bin_to_hex(self._extract_repository_id(self.repository_path))
@@ -2254,7 +2281,8 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         if self.FORK_DEFAULT:
             self.cmd('key', 'import', self.repository_location, export_file, exit_code=2)
         else:
-            self.assert_raises(NotABorgKeyFile, lambda: self.cmd('key', 'import', self.repository_location, export_file))
+            with pytest.raises(NotABorgKeyFile):
+                self.cmd('key', 'import', self.repository_location, export_file)
 
         with open(export_file, 'w') as fd:
             fd.write('BORG_KEY a0a0a0\n')
@@ -2262,7 +2290,8 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         if self.FORK_DEFAULT:
             self.cmd('key', 'import', self.repository_location, export_file, exit_code=2)
         else:
-            self.assert_raises(RepoIdMismatch, lambda: self.cmd('key', 'import', self.repository_location, export_file))
+            with pytest.raises(RepoIdMismatch):
+                self.cmd('key', 'import', self.repository_location, export_file)
 
     def test_key_export_paperkey(self):
         repo_id = 'e294423506da4e1ea76e8dcdf1a3919624ae3ae496fddf905610c351d3f09239'
@@ -2675,13 +2704,13 @@ class RemoteArchiverTestCase(ArchiverTestCase):
             self.cmd('init', '--encryption=repokey', self.repository_location)
         # restricted to repo directory itself, fail for other directories with same prefix:
         with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', self.repository_path]):
-            self.assert_raises(PathNotAllowed,
-                               lambda: self.cmd('init', '--encryption=repokey', self.repository_location + '_0'))
+            with pytest.raises(PathNotAllowed):
+                self.cmd('init', '--encryption=repokey', self.repository_location + '_0')
 
         # restricted to a completely different path:
         with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', '/foo']):
-            self.assert_raises(PathNotAllowed,
-                               lambda: self.cmd('init', '--encryption=repokey', self.repository_location + '_1'))
+            with pytest.raises(PathNotAllowed):
+                self.cmd('init', '--encryption=repokey', self.repository_location + '_1')
         path_prefix = os.path.dirname(self.repository_path)
         # restrict to repo directory's parent directory:
         with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', path_prefix]):