Merge pull request #1001 from ThomasWaldmann/env-borg-key-file

add BORG_KEY_FILE

borg/key.py

@@ -35,6 +35,14 @@ class KeyfileNotFoundError(Error):
     """No key file for repository {} found in {}."""
 
 
+class KeyfileInvalidError(Error):
+    """Invalid key file for repository {} found in {}."""
+
+
+class KeyfileMismatchError(Error):
+    """Mismatch between repository {} and key file {}."""
+
+
 class RepoKeyNotFoundError(Error):
     """No key entry found in the config of repository {}."""
 
@@ -404,17 +412,33 @@ class KeyfileKey(KeyfileKeyBase):
     TYPE = 0x00
     FILE_ID = 'BORG_KEY'
 
+    def sanity_check(self, filename, id):
+        with open(filename, 'r') as fd:
+            line = fd.readline().strip()
+            if not line.startswith(self.FILE_ID):
+                raise KeyfileInvalidError(self.repository._location.canonical_path(), filename)
+            if line[len(self.FILE_ID) + 1:] != id:
+                raise KeyfileMismatchError(self.repository._location.canonical_path(), filename)
+            return filename
+
     def find_key(self):
+        id = self.repository.id_str
+        keyfile = os.environ.get('BORG_KEY_FILE')
+        if keyfile:
+            return self.sanity_check(keyfile, id)
         keys_dir = get_keys_dir()
         for name in os.listdir(keys_dir):
             filename = os.path.join(keys_dir, name)
-            with open(filename, 'r') as fd:
-                line = fd.readline().strip()
-                if line.startswith(self.FILE_ID) and line[len(self.FILE_ID) + 1:] == self.repository.id_str:
-                    return filename
+            try:
+                return self.sanity_check(filename, id)
+            except (KeyfileInvalidError, KeyfileMismatchError):
+                pass
         raise KeyfileNotFoundError(self.repository._location.canonical_path(), get_keys_dir())
 
     def get_new_target(self, args):
+        keyfile = os.environ.get('BORG_KEY_FILE')
+        if keyfile:
+            return keyfile
         filename = args.location.to_key_filename()
         path = filename
         i = 1

borg/testsuite/key.py

@@ -7,7 +7,7 @@ from binascii import hexlify, unhexlify
 from ..crypto import bytes_to_long, num_aes_blocks
 from ..key import PlaintextKey, PassphraseKey, KeyfileKey
 from ..helpers import Location, Chunk, bin_to_hex
-from . import BaseTestCase
+from . import BaseTestCase, environment_variable
 
 
 class KeyTestCase(BaseTestCase):
@@ -34,9 +34,11 @@ class KeyTestCase(BaseTestCase):
     def setUp(self):
         self.tmppath = tempfile.mkdtemp()
         os.environ['BORG_KEYS_DIR'] = self.tmppath
+        self.tmppath2 = tempfile.mkdtemp()
 
     def tearDown(self):
         shutil.rmtree(self.tmppath)
+        shutil.rmtree(self.tmppath2)
 
     class MockRepository:
         class _Location:
@@ -71,6 +73,20 @@ class KeyTestCase(BaseTestCase):
         chunk = Chunk(b'foo')
         self.assert_equal(chunk, key2.decrypt(key.id_hash(chunk.data), key.encrypt(chunk)))
 
+    def test_keyfile_kfenv(self):
+        keyfile = os.path.join(self.tmppath2, 'keyfile')
+        with environment_variable(BORG_KEY_FILE=keyfile, BORG_PASSPHRASE='testkf'):
+            assert not os.path.exists(keyfile)
+            key = KeyfileKey.create(self.MockRepository(), self.MockArgs())
+            assert os.path.exists(keyfile)
+            chunk = Chunk(b'XXX')
+            chunk_id = key.id_hash(chunk.data)
+            chunk_cdata = key.encrypt(chunk)
+            key = KeyfileKey.detect(self.MockRepository(), chunk_cdata)
+            self.assert_equal(chunk, key.decrypt(chunk_id, chunk_cdata))
+            os.unlink(keyfile)
+            self.assert_raises(FileNotFoundError, KeyfileKey.detect, self.MockRepository(), chunk_cdata)
+
     def test_keyfile2(self):
         with open(os.path.join(os.environ['BORG_KEYS_DIR'], 'keyfile'), 'w') as fd:
             fd.write(self.keyfile2_key_file)
@@ -78,6 +94,14 @@ class KeyTestCase(BaseTestCase):
         key = KeyfileKey.detect(self.MockRepository(), self.keyfile2_cdata)
         self.assert_equal(key.decrypt(self.keyfile2_id, self.keyfile2_cdata).data, b'payload')
 
+    def test_keyfile2_kfenv(self):
+        keyfile = os.path.join(self.tmppath2, 'keyfile')
+        with open(keyfile, 'w') as fd:
+            fd.write(self.keyfile2_key_file)
+        with environment_variable(BORG_KEY_FILE=keyfile, BORG_PASSPHRASE='passphrase'):
+            key = KeyfileKey.detect(self.MockRepository(), self.keyfile2_cdata)
+            self.assert_equal(key.decrypt(self.keyfile2_id, self.keyfile2_cdata).data, b'payload')
+
     def test_passphrase(self):
         os.environ['BORG_PASSPHRASE'] = 'test'
         key = PassphraseKey.create(self.MockRepository(), None)

docs/usage.rst

@@ -101,9 +101,11 @@ Some automatic "answerers" (if set, they automatically answer confirmation quest
     answer or ask you interactively, depending on whether retries are allowed (they by default are
     allowed). So please test your scripts interactively before making them a non-interactive script.
 
-Directories:
+Directories and files:
     BORG_KEYS_DIR
         Default to '~/.config/borg/keys'. This directory contains keys for encrypted repositories.
+    BORG_KEY_FILE
+        When set, use the given filename as repository key file.
     BORG_CACHE_DIR
         Default to '~/.cache/borg'. This directory contains the local cache and might need a lot
         of space for dealing with big repositories).