WIP: propagate args down to decrypt_key_file_fido2()

Tested with repo-info only. More code paths need to be covered.

To be continued.

Author: steelman

src/borg/archiver/_common.py

@@ -154,7 +154,7 @@ def wrapper(self, args, **kwargs):
                         from ..legacy.repoobj import RepoObj1
 
                         ro_cls = RepoObj1
-                    manifest_ = Manifest.load(repository, compatibility, other=False, ro_cls=ro_cls)
+                    manifest_ = Manifest.load(repository, compatibility, args, other=False, ro_cls=ro_cls)
                     kwargs["manifest"] = manifest_
                     if "compression" in args:
                         manifest_.repo_objs.compressor = args.compression.compressor
@@ -562,6 +562,13 @@ def define_common_options(add_common_option):
         action=Highlander,
         help="repository to use",
     )
+    add_common_option(
+        "--fido2-device",
+        metavar="DEVICE",
+        dest="fido2_device",
+        default=os.environ.get("BORG_FIDO2_DEVICE", "none"),
+        help="select fido2 device to protect the repository key, use ``fido2-token -L`` to list available devices.",
+    )
 
 
 def build_matcher(inclexcl_patterns, include_paths, pattern_roots=()):

src/borg/archiver/repo_create_cmd.py

@@ -261,10 +261,3 @@ def build_parser_repo_create(self, subparsers, common_parser, mid_common_parser)
             help="copy the crypt_key (used for authenticated encryption) from the key of the other repository "
             "(default: new random key).",
         )
-        subparser.add_argument(
-            "--fido2-device",
-            metavar="DEVICE",
-            dest="fido2_device",
-            help="select fido2 device to protect the repository key, use ``fido2-token -L`` "
-            "to list available devices.",
-        )

src/borg/crypto/key.py

@@ -1,6 +1,7 @@
 import binascii
 import hmac
 import os
+import stat
 import textwrap
 from hashlib import sha256
 from math import ceil
@@ -222,7 +223,7 @@ def identify_key(manifest_data):
     raise UnsupportedPayloadError(key_type)
 
 
-def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj):
+def key_factory(repository, manifest_chunk, args, *, other=False, ro_cls=RepoObj):
     manifest_data = ro_cls.extract_crypted_data(manifest_chunk)
     assert manifest_data, "manifest data must not be zero bytes long"
     key_cls = identify_key(manifest_data)
@@ -232,7 +233,7 @@ def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj):
         # tagged envelope modes (see MACKeyBase). The legacy key classes only exist to read borg
         # 1.x repositories (ro_cls is RepoObj1 then), e.g. for "borg transfer --from-borg1".
         raise UnsupportedPayloadError(manifest_data[0])
-    key = key_cls.detect(repository, manifest_data, other=other)
+    key = key_cls.detect(repository, manifest_data, args, other=other)
     key.stored_type = manifest_data[0]
     return key
 
@@ -505,7 +506,7 @@ class FlexiKey:
     _loaded_label = None  # label of the borg key we unlocked
 
     @classmethod
-    def detect(cls, repository, manifest_data, *, other=False):
+    def detect(cls, repository, manifest_data, args, *, other=False):
         key = cls(repository)
         target = key.find_key()
         # TODO: ask for "PIN" when applicable
@@ -515,31 +516,31 @@ def detect(cls, repository, manifest_data, *, other=False):
         # passphrase against all of them.
         if passphrase is None:
             passphrase = Passphrase()
-            if not key.load_any(passphrase):
+            if not key.load_any(passphrase, args):
                 for retry in range(0, 3):
                     passphrase = Passphrase.getpass(prompt)
-                    if key.load_any(passphrase):
+                    if key.load_any(passphrase, args):
                         break
                     Passphrase.display_debug_info(passphrase)
                 else:
                     raise PasswordRetriesExceeded
         else:
-            if not key.load_any(passphrase):
+            if not key.load_any(passphrase, args):
                 Passphrase.display_debug_info(passphrase)
                 raise PassphraseWrong
         key.init_ciphers(manifest_data)
         key._passphrase = passphrase
         return key
 
-    def _load(self, key_data, passphrase):
+    def _load(self, key_data, passphrase, args):
         try:
             key = binascii.a2b_base64(key_data)
         except (ValueError, binascii.Error):
             raise KeyfileInvalidError(self.repository._location.canonical_path(), "(repokey)") from None
         if len(key) < 20:
             # this is in no way a precise check, usually we have about 400b key data.
             raise KeyfileInvalidError(self.repository._location.canonical_path(), "(repokey)")
-        data = self.decrypt_key_file(key, passphrase)
+        data = self.decrypt_key_file(key, passphrase, args)
         if data:
             data = msgpack.unpackb(data)
             key = Key(internal_dict=data)
@@ -552,7 +553,7 @@ def _load(self, key_data, passphrase):
             return True
         return False
 
-    def decrypt_key_file(self, data, passphrase):
+    def decrypt_key_file(self, data, passphrase, args):
         unpacker = get_limited_unpacker("key")
         unpacker.feed(data)
         data = unpacker.unpack()
@@ -565,7 +566,7 @@ def decrypt_key_file(self, data, passphrase):
             if encrypted_key.algorithm == "argon2 chacha20-poly1305":
                 return self.decrypt_key_file_argon2(encrypted_key, passphrase)
             elif encrypted_key.algorithm == "fido2 hmac-secret chacha20-poly1305":
-                return self.decrypt_key_file_fido2(encrypted_key, passphrase)
+                return self.decrypt_key_file_fido2(encrypted_key, passphrase, args)
             else:
                 raise UnsupportedKeyFormatError()
 
@@ -604,8 +605,14 @@ def decrypt_key_file_argon2(self, encrypted_key, passphrase):
         except low_level.IntegrityError:
             return None
 
-    def decrypt_key_file_fido2(self, encrypted_key, pin):
-        device = Fido2Operations.find_device(encrypted_key.fido2_credential_id)
+    def decrypt_key_file_fido2(self, encrypted_key, pin, args):
+        device = args.fido2_device
+        if device == "auto":
+            device = Fido2Operations.find_device(encrypted_key.fido2_credential_id)
+        if device == "none" or not (os.access(device, os.F_OK) and stat.S_ISCHR(os.stat(device).st_mode)):
+            # The device may be invalid despite passing this check, but if we are here
+            # it is definitely invalid.
+            raise ValueError(f"Invalid or unspecified FIDO2 device: {device}")
         operations = Fido2Operations(device, pin)
         secret = operations.use_hmac_hash(encrypted_key.salt, encrypted_key.fido2_credential_id)
         ae_cipher = CHACHA20_POLY1305(key=secret, iv=0, header_len=0, aad_offset=0)
@@ -870,7 +877,7 @@ def _key_envelope(self, blob_text):
         unpacker.feed(raw)
         return EncryptedKey(internal_dict=unpacker.unpack())
 
-    def _try_key(self, key_id, blob_text, keyfile_path, passphrase):
+    def _try_key(self, key_id, blob_text, keyfile_path, passphrase, args):
         # try to unlock a single borg key with the given passphrase; on success, remember it.
         if is_keyfile(blob_text):
             # keyfile / modern repokey: data is wrapped in keyfile_format (BORG_KEY header).
@@ -882,7 +889,7 @@ def _try_key(self, key_id, blob_text, keyfile_path, passphrase):
             # borg 1.x repokey: stored as raw base64 without the BORG_KEY header.
             key_data = blob_text
         try:
-            loaded = self._load(key_data, passphrase)
+            loaded = self._load(key_data, passphrase, args)
         except Exception as exc:  # noqa: BLE001 - a corrupted borg key must not break unlocking via the others
             logger.debug("Borg key %s could not be loaded (corrupted?), skipping it: %s", key_id[:12], exc)
             return False
@@ -900,14 +907,14 @@ def _try_key(self, key_id, blob_text, keyfile_path, passphrase):
             return True
         return False
 
-    def load_any(self, passphrase):
+    def load_any(self, passphrase, args):
         """Try the passphrase against every borg key of this repository."""
         for key_id, blob_text, keyfile_path in self._iter_keys():
-            if self._try_key(key_id, blob_text, keyfile_path, passphrase):
+            if self._try_key(key_id, blob_text, keyfile_path, passphrase, args):
                 return True
         return False
 
-    def load(self, target, passphrase):
+    def load(self, target, passphrase, args):
         # load a specific borg key: for keyfiles, the explicit file given as target; for repokey,
         # any of the repository's borg keys (which are addressed by passphrase, not by target).
         if self.storage == KeyBlobStorage.KEYFILE:
@@ -916,9 +923,9 @@ def load(self, target, passphrase):
                     blob = fd.read()
             except OSError:
                 return False
-            return self._try_key(sha256(blob).hexdigest(), blob.decode("utf-8"), str(target), passphrase)
+            return self._try_key(sha256(blob).hexdigest(), blob.decode("utf-8"), str(target), passphrase, args)
         else:
-            return self.load_any(passphrase)
+            return self.load_any(passphrase, args)
 
     def save(self, target, passphrase, algorithm, args, create=False, label=None, replace=True):
         # replace=True replaces the previously-loaded borg key (change-passphrase semantics);
@@ -1228,6 +1235,7 @@ class AuthenticatedKeyBase(MACKeyBase, FlexiKey):
     # It's only authenticated, not encrypted.
     logically_encrypted = False
 
+<<<<<<< HEAD
     # every read is authenticated by the envelope tag (which covers the chunk id via the AAD),
     # independently of assert_id() - the same reasoning as for the AEAD keys, see
     # AEADKeyBase.assert_id about what verifying the chunk id adds on top of that.
@@ -1247,7 +1255,7 @@ def tag_key(self):
             self._tag_key = self.derive_key(salt=b"", domain=self.MAC_KEY_DOMAIN, size=32)
         return self._tag_key
 
-    def _load(self, key_data, passphrase):
+    def _load(self, key_data, passphrase, args):
         if AUTHENTICATED_NO_KEY:
             # fake _load if we have no key or passphrase. The key material is all-zero and thus
             # worthless, but these modes do not encrypt, so reading still works - decrypt() skips
@@ -1257,9 +1265,9 @@ def _load(self, key_data, passphrase):
             self.id_key = bytes(32)
             self.chunk_seed = 0
             return True
-        return super()._load(key_data, passphrase)
+        return super()._load(key_data, passphrase, args)
 
-    def load(self, target, passphrase):
+    def load(self, target, passphrase, args):
         success = super().load(target, passphrase)
         self.logically_encrypted = False
         return success

src/borg/manifest.py

@@ -500,13 +500,13 @@ def last_timestamp(self):
         return parse_timestamp(self.timestamp)
 
     @classmethod
-    def load(cls, repository, operations, key=None, *, other=False, ro_cls=RepoObj):
+    def load(cls, repository, operations, args, key=None, *, other=False, ro_cls=RepoObj):
         from .item import ManifestItem
         from .crypto.key import key_factory
 
         cdata = repository.get_manifest()
         if not key:
-            key = key_factory(repository, cdata, other=other, ro_cls=ro_cls)
+            key = key_factory(repository, cdata, args, other=other, ro_cls=ro_cls)
         manifest = cls(key, repository, ro_cls=ro_cls)
         _, data = manifest.repo_objs.parse(cls.MANIFEST_ID, cdata, ro_type=ROBJ_MANIFEST)
         manifest_dict = key.unpack_manifest(data)