Introduce support for FIDO2 to protect keys

Use FIDO2 devices with hmac-secret extension to generate
key encryption key (KEK) instead of passphrase processed
with argon2.

Author: steelman

pyproject.toml

@@ -59,6 +59,7 @@ s3 = ["borgstore[rest,blake3,s3] ~= 0.6.1"]
 sftp = ["borgstore[rest,blake3,sftp] ~= 0.6.1"]
 rclone = ["borgstore[rest,blake3,rclone] ~= 0.6.1"]
 cockpit = ["textual>=6.8.0"]  # might also work with older versions, untested
+fido2 = ["fido2 >= 0.9.1"]
 
 [project.urls]
 "Homepage" = "https://borgbackup.org/"

src/borg/archiver/key_cmds.py

@@ -21,7 +21,7 @@ def do_key_change_passphrase(self, args, repository, manifest):
         key = manifest.key
         if not hasattr(key, "change_passphrase"):
             raise CommandError("This repository is not encrypted, cannot change the passphrase.")
-        key.change_passphrase()
+        key.change_passphrase(args)
         logger.info("Key updated")
         if hasattr(key, "find_key"):
             # print key location to make backing it up easier
@@ -270,6 +270,10 @@ def build_parser_keys(self, subparsers, common_parser, mid_common_parser):
             parents=[common_parser], description=self.do_key_change_passphrase.__doc__, epilog=change_passphrase_epilog
         )
         key_parsers.add_subcommand("change-passphrase", subparser, help="change the repository passphrase")
+        subparser.add_argument(
+            "--fido2-device", metavar="DEVICE", dest="fido2_device", default=None,
+            help="select fido2 device to protect the repository key, use ``fido2-token -L`` to list available devices."
+        )
 
         add_epilog = process_epilog(
             """
@@ -288,12 +292,14 @@ def build_parser_keys(self, subparsers, common_parser, mid_common_parser):
         keys require a unique, user-defined ``--label``.
         """
         )
+
         subparser = ArgumentParser(parents=[common_parser], description=self.do_key_add.__doc__, epilog=add_epilog)
         key_parsers.add_subcommand("add", subparser, help="add a borg key (independent passphrase)")
         subparser.add_argument(
             "--label", metavar="LABEL", dest="label", required=True, help="label for the new borg key (must be unique)"
         )
 
+
         remove_epilog = process_epilog(
             """
         Remove a borg key from the repository.

src/borg/archiver/repo_create_cmd.py

@@ -261,3 +261,10 @@ 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/constants.py

@@ -244,6 +244,8 @@
     "pbkdf2": "sha256",
     # encrypt-then-MAC, kdf: argon2, encryption: chacha20, authentication: poly1305
     "argon2": "argon2 chacha20-poly1305",
+    # Fido2 hmac-secret
+    "fido2": "fido2 hmac-secret chacha20-poly1305",
 }
 
 

src/borg/crypto/fido2.py

@@ -0,0 +1,172 @@
+import os
+import sys
+
+from binascii import b2a_hex
+from ..logger import create_logger
+
+logger = create_logger()
+
+try:
+    from fido2.ctap2 import Ctap2, ClientPin
+    from fido2.ctap import CtapError
+    from fido2.hid import CtapHidDevice, get_descriptor, open_connection
+    from fido2.cose import ES256
+
+    has_fido2 = True
+except ImportError:
+    has_fido2 = False
+
+
+class Fido2Operations:
+    @classmethod
+    def find_device(cls, credential_id, rp_id="org.borgbackup.fido2"):
+        if not has_fido2:
+            raise ValueError("No FIDO2 support found. Install the 'fido2' module.")
+        for d in CtapHidDevice.list_devices():
+            ctap2 = Ctap2(d)
+
+            # It's not our device
+            if "hmac-secret" not in ctap2.info.extensions:
+                continue
+
+            # According to CTAP 2.1 specification, to do pre-flight we
+            # need to set up option to false with optionally
+            # pinUvAuthParam in assertion[1]. But for authenticator
+            # that doesn't support user presence, once up option is
+            # present, the authenticator may return
+            # CTAP2_ERR_UNSUPPORTED_OPTION[2].  So we simplely omit
+            # the option in that case.
+            # Reference:
+            # 1: https://fidoalliance.org/specs/fido-v2.1-ps-20210615/fido-client-to-authenticator-protocol-v2.1-ps-20210615.html#pre-flight
+            # 2: https://fidoalliance.org/specs/fido-v2.0-ps-20190130/fido-client-to-authenticator-protocol-v2.0-ps-20190130.html#authenticatorGetAssertion
+            #    (in step 5)
+            options = None
+            if ctap2.info.options.get("up", True):
+                options = {"up": False}
+            try:
+                ctap2.get_assertion(
+                    rp_id=rp_id,
+                    client_data_hash=b"\x00" * 32,
+                    allow_list=[{"type": "public-key", "id": credential_id}],
+                    extensions=None,
+                    options=options,
+                    pin_uv_param=None,
+                    pin_uv_protocol=None,
+                    event=None,
+                    on_keepalive=None,
+                )
+            except CtapError as e:
+                if CtapError.ERR.NO_CREDENTIALS == e.code:
+                    continue
+                raise e
+            logger.info(f"Found the FIDO2 device matching the credential: {d.descriptor.path}.")
+            return d.descriptor.path
+        else:
+            logger.error("No matching FIDO2 device found.")
+
+    def __init__(self, device=None, pin=None):
+        if not has_fido2:
+            raise ValueError("No FIDO2 support found. Install the 'fido2' module.")
+        if not device:
+            raise ValueError("FIDO2 device not specified.")
+        self._device_path = device
+        self._pin = pin
+
+        descriptor = get_descriptor(self._device_path)
+        hid_device = CtapHidDevice(descriptor, open_connection(descriptor))
+        self._ctap2 = Ctap2(hid_device)
+        self._client_pin = ClientPin(self._ctap2)
+
+        # TODO: verify that the device supports hmac-secret
+        # if not 'hmac-secret' in self._ctap2.info.extensions:
+        #     # Oh no!
+
+        #  Defaults are per table in 5.4 in FIDO2 spec
+        self.has_rk = self._ctap2.info.options.get("rk", False)
+        self.has_client_pin = self._ctap2.info.options.get("clientPin", False)
+        self.has_up = self._ctap2.info.options.get("up", True)
+        self.has_uv = self._ctap2.info.options.get("uv", False)
+
+    def _hmac_secret_input(self, salt1):
+        key_agreement, self._shared_secret = self._client_pin._get_shared_secret()
+        salt_enc = self._client_pin.protocol.encrypt(self._shared_secret, salt1)
+        salt_auth = self._client_pin.protocol.authenticate(self._shared_secret, salt_enc)
+        return {1: key_agreement, 2: salt_enc, 3: salt_auth, 4: self._client_pin.protocol.VERSION}
+
+    def _hmac_secret_output(self, data):
+        decrypted = self._client_pin.protocol.decrypt(self._shared_secret, data)
+        return decrypted[:32]
+
+    def _get_assertion(self, salt, credential_id, rp_id="org.borgbackup.fido2"):
+        return self._ctap2.get_assertion(
+            rp_id=rp_id,
+            client_data_hash=b"\x00" * 32,
+            allow_list=[{"type": "public-key", "id": credential_id}],
+            extensions={"hmac-secret": self._hmac_secret_input(salt)},
+            options=None,
+            pin_uv_param=None,
+            pin_uv_protocol=self._client_pin.protocol.VERSION,
+            event=None,
+            on_keepalive=None,
+        )
+
+    def use_hmac_hash(self, salt, credential_id):
+
+        # TODO: replace with…
+        print("\nTouch your authenticator device now...\n", file=sys.stderr)
+        assertion = self._get_assertion(salt, credential_id)
+        if not assertion.auth_data.extensions.get("hmac-secret"):
+            raise Exception("Failed to get assertion with hmac-secret")
+
+        secret = self._hmac_secret_output(assertion.auth_data.extensions["hmac-secret"])
+        return secret
+
+    def generate_hmac_hash(self, user, rp_id="org.borgbackup.fido2"):
+        # TODO: decide whether to use or not credentialProtectionPolicy
+        if self._pin:
+            pin_token = self._client_pin.get_pin_token(self._pin, ClientPin.PERMISSION.MAKE_CREDENTIAL, rp_id)
+            pin_auth = self._client_pin.protocol.authenticate(pin_token, b"\x00" * 32)
+        elif self.has_client_pin:
+            raise ValueError("PIN required but not provided")
+
+        if not (self.has_rk or self.has_uv):
+            cred_options = None
+        else:
+            cred_options = {}
+            if self.has_rk:
+                cred_options["rk"] = False
+            if self.has_uv:
+                cred_options["uv"] = False
+
+        print("\nTouch your authenticator device now...\n", file=sys.stderr)
+        result = self._ctap2.make_credential(
+            client_data_hash=b"\x00" * 32,
+            rp={"id": rp_id, "name": "Borg Repository"},
+            user={"id": user, "name": b2a_hex(user).decode("ascii")},
+            key_params=[{"type": "public-key", "alg": ES256.ALGORITHM}],
+            exclude_list=None,
+            extensions={"hmac-secret": True},
+            options=cred_options,
+            pin_uv_param=pin_auth,
+            pin_uv_protocol=self._client_pin.protocol.VERSION,
+            event=None,
+            on_keepalive=None,
+        )
+
+        if result.auth_data.extensions.get("hmac-secret") is None:
+            raise Exception("Failed to create credential with hmac-secret")
+        logger.info("New credential created with the hmac-secret extension.")
+
+        credential_id = result.auth_data.credential_data.credential_id
+
+        salt = os.urandom(32)
+        print("\nTouch your authenticator device now...\n", file=sys.stderr)
+        assertion = self._get_assertion(salt, credential_id)
+
+        if not assertion.auth_data.extensions.get("hmac-secret"):
+            raise Exception("Failed to get assertion with hmac-secret")
+        logger.info("An assertion with hmac-secret value created.")
+
+        secret = self._hmac_secret_output(assertion.auth_data.extensions["hmac-secret"])
+
+        return credential_id, salt, secret

src/borg/crypto/key.py

@@ -32,7 +32,7 @@
 from .low_level import bytes_to_int, num_cipher_blocks, hmac_sha256
 from .low_level import AES256_OCB, CHACHA20_POLY1305
 from . import low_level
-
+from .fido2 import Fido2Operations
 
 def keyfile_name_for(content: bytes) -> str:
     return sha256(content).hexdigest()
@@ -508,6 +508,7 @@ class FlexiKey:
     def detect(cls, repository, manifest_data, *, other=False):
         key = cls(repository)
         target = key.find_key()
+        # TODO: ask for "PIN" when applicable
         prompt = "Enter passphrase for key %s: " % target
         passphrase = Passphrase.env_passphrase(other=other)
         # a repository may have multiple borg keys, one per passphrase; try the
@@ -563,6 +564,8 @@ def decrypt_key_file(self, data, passphrase):
             self._encrypted_key_label = encrypted_key.get("label")
             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)
             else:
                 raise UnsupportedKeyFormatError()
 
@@ -601,9 +604,21 @@ 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)
+        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)
+        try:
+            return ae_cipher.decrypt(encrypted_key.data)
+        except low_level.IntegrityError:
+            return None
+
     def encrypt_key_file(self, data, passphrase, algorithm, label=None):
         if algorithm == "argon2 chacha20-poly1305":
             return self.encrypt_key_file_argon2(data, passphrase, label=label)
+        elif algorithm == "fido2 hmac-secret chacha20-poly1305":
+            return self.encrypt_key_file_fido2(data, passphrase, args)
         else:
             raise ValueError(f"Unexpected algorithm: {algorithm}")
 
@@ -623,25 +638,52 @@ def encrypt_key_file_argon2(self, data, passphrase, label=None):
         encrypted_key = EncryptedKey(**kw)
         return msgpack.packb(encrypted_key.as_dict())
 
-    def _save(self, passphrase, algorithm, label=None):
+    def encrypt_key_file_fido2(self, data, pin, args):
+        operations = Fido2Operations(args.fido2_device, pin)
+        credential_id, salt, secret = operations.generate_hmac_hash(user=self.repository_id)
+        ae_cipher = CHACHA20_POLY1305(key=secret, iv=0, header_len=0, aad_offset=0)
+        encrypted_key = EncryptedKey(
+            version=1,
+            algorithm="fido2 hmac-secret chacha20-poly1305",
+            salt=salt,
+            data=ae_cipher.encrypt(data),
+            fido2_credential_id=credential_id,
+        )
+        return msgpack.packb(encrypted_key.as_dict())
+
+    def _save(self, passphrase, algorithm, args, label=None):
         key = Key(
             version=2,
             repository_id=self.repository_id,
             crypt_key=self.crypt_key,
             id_key=self.id_key,
             chunk_seed=self.chunk_seed,
         )
-        data = self.encrypt_key_file(msgpack.packb(key.as_dict()), passphrase, algorithm, label=label)
+        data = self.encrypt_key_file(msgpack.packb(key.as_dict()), passphrase, algorithm, args, label=label)
         key_data = "\n".join(textwrap.wrap(binascii.b2a_base64(data).decode("ascii")))
         return key_data
 
-    def change_passphrase(self, passphrase=None):
-        if passphrase is None:
-            passphrase = Passphrase.new(allow_empty=True, only_new=True)
-        # replace the borg key we unlocked with: keep its label, write the new borg key, then
-        # (for repokey) delete the previously-loaded borg key (keyfile mode auto-erases it in save()).
-        old_id = self._loaded_key_id
-        self.save(self.target, passphrase, algorithm=self._encrypted_key_algorithm, label=self._loaded_label)
+    def change_passphrase(self, args, passphrase=None):
+        if args.fido2_device:
+            operations = Fido2Operations(args.fido2_device)
+            if operations.has_client_pin:
+                # TODO: try to be more descriptive about the device
+                passphrase = Passphrase.new(only_new=True, pin_prompt=f"Enter PIN for {args.fido2_device}: ")
+            else:
+                passphrase = Passphrase("")
+            key_algorithm = KEY_ALGORITHMS["fido2"]
+        else:
+            if passphrase is None:
+                passphrase = Passphrase.new(allow_empty=True, only_new=True)
+            # replace the borg key we unlocked with: keep its label, write the new borg key, then
+            # (for repokey) delete the previously-loaded borg key (keyfile mode auto-erases it in save()).
+            old_id = self._loaded_key_id
+
+            key_algorithm = self._encrypted_key_algorithm
+            # If fido2 was used before change it to argon2
+            if key_algorithm == KEY_ALGORITHMS["fido2"]:
+                key_algorithm = KEY_ALGORITHMS["argon2"]
+        self.save(self.target, passphrase, algorithm=key_algorithm, args=args, label=self._loaded_label))
         if self.storage == KeyBlobStorage.REPO and old_id and hasattr(self.repository, "delete_key"):
             if self._loaded_key_id != old_id:
                 self.repository.delete_key(old_id)
@@ -675,11 +717,16 @@ def create(cls, repository, args, *, other_key=None):
             key.init_from_given_data(crypt_key=crypt_key, id_key=id_key, chunk_seed=chunk_seed)
         else:
             key.init_from_random_data()
-        passphrase = Passphrase.new(allow_empty=True)
+        if args.fido2_device:
+            key_algorithm = KEY_ALGORITHMS["fido2"]
+            passphrase = Passphrase.new(pin_prompt="Enter PIN for {args.fido2_device}: ")
+        else:
+            key_algorithm = KEY_ALGORITHMS["argon2"]
+            passphrase = Passphrase.new(allow_empty=True)
         key.init_ciphers()
         target = key.get_new_target(args)
         # the first borg key of a repository is the protected "admin" key.
-        key.save(target, passphrase, create=True, algorithm=KEY_ALGORITHMS["argon2"], label=ADMIN_LABEL)
+        key.save(target, passphrase, key_algorithm, args, create=True, label=ADMIN_LABEL)
         logger.info('Key in "%s" created.' % key.target)
         logger.info("Keep this key safe. Your data will be inaccessible without it.")
         return key
@@ -873,10 +920,10 @@ def load(self, target, passphrase):
         else:
             return self.load_any(passphrase)
 
-    def save(self, target, passphrase, algorithm, create=False, label=None, replace=True):
+    def save(self, target, passphrase, algorithm, args, create=False, label=None, replace=True):
         # replace=True replaces the previously-loaded borg key (change-passphrase semantics);
         # replace=False adds an additional borg key, keeping the existing ones (key add).
-        key_data = self._save(passphrase, algorithm, label=label)
+        key_data = self._save(passphrase, algorithm, args, label=label)
         if self.storage == KeyBlobStorage.KEYFILE:
             old_target = getattr(self, "target", None)
             keys_dir = get_keys_dir()
@@ -1222,8 +1269,8 @@ def load_any(self, passphrase):
         self.logically_encrypted = False
         return success
 
-    def save(self, target, passphrase, algorithm, create=False, label=None, replace=True):
-        super().save(target, passphrase, algorithm, create=create, label=label, replace=replace)
+    def save(self, target, passphrase, algorithm, args, create=False, label=None, replace=True):
+        super().save(target, passphrase, algorithm, args, create=create, label=label, replace=replace)
         self.logically_encrypted = False
 
     def init_from_given_data(self, *, crypt_key, id_key, chunk_seed):