Merge pull request #6556 from hexagonrecursion/kdf-refactor

Move the key derivation code from helpers.Passphrase to crypto.FlexiKey

src/borg/archiver.py

@@ -46,7 +46,7 @@
     from .constants import *  # NOQA
     from .compress import CompressionSpec
     from .crypto.key import key_creator, key_argument_names, tam_required_file, tam_required
-    from .crypto.key import RepoKey, KeyfileKey, Blake2RepoKey, Blake2KeyfileKey
+    from .crypto.key import RepoKey, KeyfileKey, Blake2RepoKey, Blake2KeyfileKey, FlexiKey
     from .crypto.keymanager import KeyManager
     from .helpers import EXIT_SUCCESS, EXIT_WARNING, EXIT_ERROR, EXIT_SIGNAL_BASE
     from .helpers import Error, NoManifestError, set_ec
@@ -622,12 +622,11 @@ def chunkit(chunker_name, *args, **kwargs):
         for spec, func in tests:
             print(f"{spec:<24} {size:<10} {timeit(func, number=100):.3f}s")
 
-        from borg.helpers.passphrase import Passphrase
         print("KDFs (slow is GOOD, use argon2!) ===============================")
         count = 5
         for spec, func in [
-            ("pbkdf2", lambda: Passphrase('mypassphrase').kdf(b'salt'*8, PBKDF2_ITERATIONS, 32)),
-            ("argon2", lambda: Passphrase('mypassphrase').argon2(64, b'S' * ARGON2_SALT_BYTES, **ARGON2_ARGS)),
+            ("pbkdf2", lambda: FlexiKey.pbkdf2('mypassphrase', b'salt'*8, PBKDF2_ITERATIONS, 32)),
+            ("argon2", lambda: FlexiKey.argon2('mypassphrase', 64, b'S' * ARGON2_SALT_BYTES, **ARGON2_ARGS)),
         ]:
             print(f"{spec:<24} {count:<10} {timeit(func, number=count):.3f}s")
 

src/borg/crypto/key.py

@@ -3,12 +3,15 @@
 import os
 import textwrap
 from binascii import a2b_base64, b2a_base64, hexlify
-from hashlib import sha256
+from hashlib import sha256, pbkdf2_hmac
+from typing import Literal
 
 from ..logger import create_logger
 
 logger = create_logger()
 
+import argon2.low_level
+
 from ..constants import *  # NOQA
 from ..compress import Compressor
 from ..helpers import StableDict
@@ -447,15 +450,53 @@ def decrypt_key_file(self, data, passphrase):
             else:
                 raise UnsupportedKeyFormatError()
 
+    @staticmethod
+    def pbkdf2(passphrase, salt, iterations, output_len_in_bytes):
+        if os.environ.get("BORG_TESTONLY_WEAKEN_KDF") == "1":
+            iterations = 1
+        return pbkdf2_hmac('sha256', passphrase.encode('utf-8'), salt, iterations, output_len_in_bytes)
+
+    @staticmethod
+    def argon2(
+        passphrase: str,
+        output_len_in_bytes: int,
+        salt: bytes,
+        time_cost: int,
+        memory_cost: int,
+        parallelism: int,
+        type: Literal['i', 'd', 'id']
+    ) -> bytes:
+        if os.environ.get("BORG_TESTONLY_WEAKEN_KDF") == "1":
+            time_cost = 1
+            parallelism = 1
+            # 8 is the smallest value that avoids the "Memory cost is too small" exception
+            memory_cost = 8
+        type_map = {
+            'i': argon2.low_level.Type.I,
+            'd': argon2.low_level.Type.D,
+            'id': argon2.low_level.Type.ID,
+        }
+        key = argon2.low_level.hash_secret_raw(
+            secret=passphrase.encode("utf-8"),
+            hash_len=output_len_in_bytes,
+            salt=salt,
+            time_cost=time_cost,
+            memory_cost=memory_cost,
+            parallelism=parallelism,
+            type=type_map[type],
+        )
+        return key
+
     def decrypt_key_file_pbkdf2(self, encrypted_key, passphrase):
-        key = passphrase.kdf(encrypted_key.salt, encrypted_key.iterations, 32)
+        key = self.pbkdf2(passphrase, encrypted_key.salt, encrypted_key.iterations, 32)
         data = AES(key, b'\0'*16).decrypt(encrypted_key.data)
         if hmac.compare_digest(hmac_sha256(key, data), encrypted_key.hash):
             return data
         return None
 
     def decrypt_key_file_argon2(self, encrypted_key, passphrase):
-        key = passphrase.argon2(
+        key = self.argon2(
+            passphrase,
             output_len_in_bytes=64,
             salt=encrypted_key.salt,
             time_cost=encrypted_key.argon2_time_cost,
@@ -485,7 +526,7 @@ def encrypt_key_file(self, data, passphrase, algorithm):
     def encrypt_key_file_pbkdf2(self, data, passphrase):
         salt = os.urandom(32)
         iterations = PBKDF2_ITERATIONS
-        key = passphrase.kdf(salt, iterations, 32)
+        key = self.pbkdf2(passphrase, salt, iterations, 32)
         hash = hmac_sha256(key, data)
         cdata = AES(key, b'\0'*16).encrypt(data)
         enc_key = EncryptedKey(
@@ -500,7 +541,8 @@ def encrypt_key_file_pbkdf2(self, data, passphrase):
 
     def encrypt_key_file_argon2(self, data, passphrase):
         salt = os.urandom(ARGON2_SALT_BYTES)
-        key = passphrase.argon2(
+        key = self.argon2(
+            passphrase,
             output_len_in_bytes=64,
             salt=salt,
             **ARGON2_ARGS,

src/borg/helpers/passphrase.py

@@ -3,8 +3,6 @@
 import shlex
 import subprocess
 import sys
-from hashlib import pbkdf2_hmac
-from typing import Literal
 
 from . import bin_to_hex
 from . import Error
@@ -13,8 +11,6 @@
 
 from ..logger import create_logger
 
-import argon2.low_level
-
 logger = create_logger()
 
 
@@ -139,38 +135,3 @@ def new(cls, allow_empty=False):
 
     def __repr__(self):
         return '<Passphrase "***hidden***">'
-
-    def kdf(self, salt, iterations, length):
-        if os.environ.get("BORG_TESTONLY_WEAKEN_KDF") == "1":
-            iterations = 1
-        return pbkdf2_hmac('sha256', self.encode('utf-8'), salt, iterations, length)
-
-    def argon2(
-        self,
-        output_len_in_bytes: int,
-        salt: bytes,
-        time_cost,
-        memory_cost,
-        parallelism,
-        type: Literal['i', 'd', 'id']
-    ) -> bytes:
-        if os.environ.get("BORG_TESTONLY_WEAKEN_KDF") == "1":
-            time_cost = 1
-            parallelism = 1
-            # 8 is the smallest value that avoids the "Memory cost is too small" exception
-            memory_cost = 8
-        type_map = {
-            'i': argon2.low_level.Type.I,
-            'd': argon2.low_level.Type.D,
-            'id': argon2.low_level.Type.ID,
-        }
-        key = argon2.low_level.hash_secret_raw(
-            secret=self.encode("utf-8"),
-            hash_len=output_len_in_bytes,
-            salt=salt,
-            time_cost=time_cost,
-            memory_cost=memory_cost,
-            parallelism=parallelism,
-            type=type_map[type],
-        )
-        return key

src/borg/testsuite/crypto.py

@@ -10,8 +10,7 @@
 from ..crypto.low_level import bytes_to_long, bytes_to_int, long_to_bytes
 from ..crypto.low_level import hkdf_hmac_sha512
 from ..crypto.low_level import AES, hmac_sha256
-from ..crypto.key import KeyfileKey, UnsupportedKeyFormatError, RepoKey
-from ..helpers.passphrase import Passphrase
+from ..crypto.key import KeyfileKey, UnsupportedKeyFormatError, RepoKey, FlexiKey
 from ..helpers import msgpack
 from ..constants import KEY_ALGORITHMS
 
@@ -260,7 +259,7 @@ def test_hkdf_hmac_sha512_5(self):
         assert okm == bytes.fromhex('1407d46013d98bc6decefcfee55f0f90b0c7f63d68eb1a80eaf07e953cfc0a3a5240a155d6e4daa965bb')
 
 
-def test_decrypt_key_file_argon2_aes256_ctr_hmac_sha256(monkeypatch):
+def test_decrypt_key_file_argon2_aes256_ctr_hmac_sha256():
     plain = b'hello'
     # echo -n "hello, pass phrase" | argon2 saltsaltsaltsalt -id -t 1 -k 8 -p 1 -l 64 -r
     key = bytes.fromhex('d07cc7f9cfb483303e0b9fec176b2a9c559bb70c3a9fb0d5f9c0c23527cd09570212449f09f8cd28c1a41b73fa0098e889c3f2642e87c392e51f95d70d248d9d')
@@ -282,21 +281,18 @@ def test_decrypt_key_file_argon2_aes256_ctr_hmac_sha256(monkeypatch):
         'algorithm': 'argon2 aes256-ctr hmac-sha256',
         'data': envelope,
     })
-    monkeypatch.setenv('BORG_PASSPHRASE', "hello, pass phrase")
-    passphrase = Passphrase.new()
     key = KeyfileKey(None)
 
-    decrypted = key.decrypt_key_file(encrypted, passphrase)
+    decrypted = key.decrypt_key_file(encrypted, "hello, pass phrase")
 
     assert decrypted == plain
 
 
-def test_decrypt_key_file_pbkdf2_sha256_aes256_ctr_hmac_sha256(monkeypatch):
+def test_decrypt_key_file_pbkdf2_sha256_aes256_ctr_hmac_sha256():
     plain = b'hello'
     salt = b'salt'*4
-    monkeypatch.setenv('BORG_PASSPHRASE', "hello, pass phrase")
-    passphrase = Passphrase.new()
-    key = passphrase.kdf(salt, iterations=1, length=32)
+    passphrase = "hello, pass phrase"
+    key = FlexiKey.pbkdf2(passphrase, salt, 1, 32)
     hash = hmac_sha256(key, plain)
     data = AES(key, b'\0'*16).encrypt(plain)
     encrypted = msgpack.packb({
@@ -314,31 +310,27 @@ def test_decrypt_key_file_pbkdf2_sha256_aes256_ctr_hmac_sha256(monkeypatch):
     assert decrypted == plain
 
 
-def test_decrypt_key_file_unsupported_algorithm(monkeypatch):
+def test_decrypt_key_file_unsupported_algorithm():
     """We will add more algorithms in the future. We should raise a helpful error."""
-    monkeypatch.setenv('BORG_PASSPHRASE', "hello, pass phrase")
-    passphrase = Passphrase.new()
     key = KeyfileKey(None)
     encrypted = msgpack.packb({
         'algorithm': 'THIS ALGORITHM IS NOT SUPPORTED',
         'version': 1,
     })
 
     with pytest.raises(UnsupportedKeyFormatError):
-        key.decrypt_key_file(encrypted, passphrase)
+        key.decrypt_key_file(encrypted, "hello, pass phrase")
 
 
-def test_decrypt_key_file_v2_is_unsupported(monkeypatch):
+def test_decrypt_key_file_v2_is_unsupported():
     """There may eventually be a version 2 of the format. For now we should raise a helpful error."""
-    monkeypatch.setenv('BORG_PASSPHRASE', "hello, pass phrase")
-    passphrase = Passphrase.new()
     key = KeyfileKey(None)
     encrypted = msgpack.packb({
         'version': 2,
     })
 
     with pytest.raises(UnsupportedKeyFormatError):
-        key.decrypt_key_file(encrypted, passphrase)
+        key.decrypt_key_file(encrypted, "hello, pass phrase")
 
 
 @pytest.mark.parametrize('cli_argument, expected_algorithm', KEY_ALGORITHMS.items())