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

@@ -607,6 +607,29 @@ 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:
+                self.assert_raises(Cache.CacheInitAbortedError, lambda: 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))