From 2f62b773ad4a05a5f1a08f54233763de645d556d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Stelmach?= Date: Sun, 17 Aug 2025 21:24:33 +0200 Subject: [PATCH 1/3] Extend handling of a new passphrases Enable reading of a new passphrase not only from BORG_NEW_PASSPHRASE environement variable but also from a passcommand or a file descriptor. If Passphrase.new() is called from FlexiKey.change_passphrase(), do not read a passphrase from regular sources (i.e. BORG_PASSPHRASE et al.) if it is available via one of the "new" ones (BORG_NEW_PASSPHRASE et al.). This makes it possible to change passphrases in fully non-interactive manner. --- src/borg/crypto/key.py | 2 +- src/borg/helpers/passphrase.py | 39 ++++++++++++++++++++++------------ 2 files changed, 26 insertions(+), 15 deletions(-) diff --git a/src/borg/crypto/key.py b/src/borg/crypto/key.py index b07a064f8e..b3fd276157 100644 --- a/src/borg/crypto/key.py +++ b/src/borg/crypto/key.py @@ -637,7 +637,7 @@ def _save(self, passphrase, algorithm, label=None): def change_passphrase(self, passphrase=None): if passphrase is None: - passphrase = Passphrase.new(allow_empty=True) + 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 diff --git a/src/borg/helpers/passphrase.py b/src/borg/helpers/passphrase.py index 7112c9257e..f572176b16 100644 --- a/src/borg/helpers/passphrase.py +++ b/src/borg/helpers/passphrase.py @@ -54,25 +54,33 @@ def _env_passphrase(cls, env_var, default=None): return cls(passphrase) @classmethod - def env_passphrase(cls, default=None, other=False): + def env_passphrase(cls, default=None, other=False, new=False): + if other and new: + raise ValueError("Only one of 'other' and 'new' may be true") if other: cls._check_ambiguity(["BORG_OTHER_PASSPHRASE", "BORG_OTHER_PASSCOMMAND", "BORG_OTHER_PASSPHRASE_FD"]) + elif new: + cls._check_ambiguity(["BORG_OTHER_PASSPHRASE", "BORG_OTHER_PASSCOMMAND", "BORG_OTHER_PASSPHRASE_FD"]) else: cls._check_ambiguity(["BORG_PASSPHRASE", "BORG_PASSCOMMAND", "BORG_PASSPHRASE_FD"]) env_var = "BORG_OTHER_PASSPHRASE" if other else "BORG_PASSPHRASE" + env_var = "BORG_NEW_PASSPHRASE" if new else env_var passphrase = cls._env_passphrase(env_var, default) if passphrase is not None: return passphrase - passphrase = cls.env_passcommand(other=other) + passphrase = cls.env_passcommand(other=other, new=new) if passphrase is not None: return passphrase - passphrase = cls.fd_passphrase(other=other) + passphrase = cls.fd_passphrase(other=other, new=new) if passphrase is not None: return passphrase @classmethod - def env_passcommand(cls, default=None, other=False): + def env_passcommand(cls, default=None, other=False, new=False): + if other and new: + raise ValueError("Only one of 'other' and 'new' may be true") env_var = "BORG_OTHER_PASSCOMMAND" if other else "BORG_PASSCOMMAND" + env_var = "BORG_NEW_PASSCOMMAND" if other else env_var passcommand = os.environ.get(env_var, None) if passcommand is not None: # passcommand is a system command (not inside pyinstaller env) @@ -84,8 +92,11 @@ def env_passcommand(cls, default=None, other=False): return cls(passphrase.rstrip("\n")) @classmethod - def fd_passphrase(cls, other=False): + def fd_passphrase(cls, other=False, new=False): + if other and new: + raise ValueError("Only one of 'other' and 'new' may be true") env_var = "BORG_OTHER_PASSPHRASE_FD" if other else "BORG_PASSPHRASE_FD" + env_var = "BORG_NEW_PASSPHRASE_FD" if new else env_var try: fd = int(os.environ.get(env_var)) except (ValueError, TypeError): @@ -94,10 +105,6 @@ def fd_passphrase(cls, other=False): passphrase = f.read() return cls(passphrase.rstrip("\n")) - @classmethod - def env_new_passphrase(cls, default=None): - return cls._env_passphrase("BORG_NEW_PASSPHRASE", default) - @classmethod def getpass(cls, prompt): try: @@ -158,6 +165,9 @@ def fmt_var(env_var): {fmt_var("BORG_PASSPHRASE")} {fmt_var("BORG_PASSCOMMAND")} {fmt_var("BORG_PASSPHRASE_FD")} + {fmt_var("BORG_NEW_PASSPHRASE")} + {fmt_var("BORG_NEW_PASSCOMMAND")} + {fmt_var("BORG_NEW_PASSPHRASE_FD")} {fmt_var("BORG_OTHER_PASSPHRASE")} {fmt_var("BORG_OTHER_PASSCOMMAND")} {fmt_var("BORG_OTHER_PASSPHRASE_FD")} @@ -166,13 +176,14 @@ def fmt_var(env_var): print(passphrase_info, file=sys.stderr) @classmethod - def new(cls, allow_empty=False): - passphrase = cls.env_new_passphrase() - if passphrase is not None: - return passphrase - passphrase = cls.env_passphrase() + def new(cls, allow_empty=False, only_new=False): + passphrase = cls.env_passphrase(new=True) if passphrase is not None: return passphrase + if not only_new: + passphrase = cls.env_passphrase() + if passphrase is not None: + return passphrase for retry in range(1, 11): passphrase = cls.getpass("Enter new passphrase: ") if allow_empty or passphrase: From 31609c7dfdee82a1855d48b741199dc7997a24a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Stelmach?= Date: Wed, 13 Aug 2025 11:52:15 +0200 Subject: [PATCH 2/3] 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. --- pyproject.toml | 1 + src/borg/archiver/key_cmds.py | 10 +- src/borg/archiver/repo_create_cmd.py | 7 ++ src/borg/constants.py | 2 + src/borg/crypto/fido2.py | 172 +++++++++++++++++++++++++++ src/borg/crypto/key.py | 78 +++++++++--- src/borg/helpers/passphrase.py | 9 +- src/borg/item.pyx | 3 +- 8 files changed, 264 insertions(+), 18 deletions(-) create mode 100644 src/borg/crypto/fido2.py diff --git a/pyproject.toml b/pyproject.toml index 8eb9fd7181..91380b4bf2 100644 --- a/pyproject.toml +++ b/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/" diff --git a/src/borg/archiver/key_cmds.py b/src/borg/archiver/key_cmds.py index 40fed2f711..1109dc9e6d 100644 --- a/src/borg/archiver/key_cmds.py +++ b/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,13 @@ 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,6 +295,7 @@ 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( diff --git a/src/borg/archiver/repo_create_cmd.py b/src/borg/archiver/repo_create_cmd.py index ac7fc0144a..258bd841c1 100644 --- a/src/borg/archiver/repo_create_cmd.py +++ b/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.", + ) diff --git a/src/borg/constants.py b/src/borg/constants.py index edb4a440d3..cb8de8a8b3 100644 --- a/src/borg/constants.py +++ b/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", } diff --git a/src/borg/crypto/fido2.py b/src/borg/crypto/fido2.py new file mode 100644 index 0000000000..dd5bbeff43 --- /dev/null +++ b/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 diff --git a/src/borg/crypto/key.py b/src/borg/crypto/key.py index b3fd276157..96863bbe7d 100644 --- a/src/borg/crypto/key.py +++ b/src/borg/crypto/key.py @@ -29,6 +29,7 @@ from ..repoobj import RepoObj, RepoObj1 +from .fido2 import Fido2Operations from .low_level import bytes_to_int, num_cipher_blocks, hmac_sha256 from .low_level import AES256_OCB, CHACHA20_POLY1305 from . import low_level @@ -508,6 +509,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 +565,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 +605,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,7 +639,20 @@ 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, @@ -631,17 +660,31 @@ def _save(self, passphrase, algorithm, label=None): 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 +718,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 +921,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 +1270,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): diff --git a/src/borg/helpers/passphrase.py b/src/borg/helpers/passphrase.py index f572176b16..8aac3cf728 100644 --- a/src/borg/helpers/passphrase.py +++ b/src/borg/helpers/passphrase.py @@ -176,7 +176,7 @@ def fmt_var(env_var): print(passphrase_info, file=sys.stderr) @classmethod - def new(cls, allow_empty=False, only_new=False): + def new(cls, allow_empty=False, only_new=False, pin_prompt=None): passphrase = cls.env_passphrase(new=True) if passphrase is not None: return passphrase @@ -184,6 +184,13 @@ def new(cls, allow_empty=False, only_new=False): passphrase = cls.env_passphrase() if passphrase is not None: return passphrase + if pin_prompt: + passphrase = cls.getpass(pin_prompt) + if passphrase is not None: + return passphrase + else: + print("PIN must not be blank.", file=sys.stderr) + raise PasswordRetriesExceeded for retry in range(1, 11): passphrase = cls.getpass("Enter new passphrase: ") if allow_empty or passphrase: diff --git a/src/borg/item.pyx b/src/borg/item.pyx index d2a94ce083..bb570b0316 100644 --- a/src/borg/item.pyx +++ b/src/borg/item.pyx @@ -407,7 +407,7 @@ cdef class EncryptedKey(PropDict): VALID_KEYS = {'version', 'algorithm', 'iterations', 'salt', 'hash', 'data', 'argon2_time_cost', 'argon2_memory_cost', 'argon2_parallelism', 'argon2_type', - 'label'} + 'label', 'fido2_credential_id'} version = PropDictProperty(int) algorithm = PropDictProperty(str) @@ -420,6 +420,7 @@ cdef class EncryptedKey(PropDict): argon2_parallelism = PropDictProperty(int) argon2_type = PropDictProperty(str) label = PropDictProperty(str) # optional human-readable borg key label, e.g. "admin" + fido2_credential_id = PropDictProperty(bytes) def update_internal(self, d): # legacy support for migration (data from old msgpacks comes in as bytes always, but sometimes we want str) From 8dc0f663029ca5e1a923bcedd0d505f2952ab1cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Stelmach?= Date: Mon, 22 Sep 2025 22:06:26 +0200 Subject: [PATCH 3/3] WIP: propagate args down to decrypt_key_file_fido2() Tested with repo-info only. More code paths need to be covered. To be continued. --- src/borg/archiver/_common.py | 9 ++++- src/borg/archiver/repo_create_cmd.py | 7 ---- src/borg/crypto/key.py | 51 ++++++++++++++++------------ src/borg/manifest.py | 4 +-- 4 files changed, 39 insertions(+), 32 deletions(-) diff --git a/src/borg/archiver/_common.py b/src/borg/archiver/_common.py index 64e91f5132..ec11243394 100644 --- a/src/borg/archiver/_common.py +++ b/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=()): diff --git a/src/borg/archiver/repo_create_cmd.py b/src/borg/archiver/repo_create_cmd.py index 258bd841c1..ac7fc0144a 100644 --- a/src/borg/archiver/repo_create_cmd.py +++ b/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.", - ) diff --git a/src/borg/crypto/key.py b/src/borg/crypto/key.py index 96863bbe7d..8a82d68a0d 100644 --- a/src/borg/crypto/key.py +++ b/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 @@ -223,7 +224,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) @@ -233,7 +234,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 @@ -506,7 +507,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 @@ -516,23 +517,23 @@ 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): @@ -540,7 +541,7 @@ def _load(self, key_data, passphrase): 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) @@ -553,7 +554,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() @@ -566,7 +567,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() @@ -605,8 +606,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) @@ -871,7 +878,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). @@ -883,7 +890,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 @@ -901,14 +908,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: @@ -917,9 +924,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); @@ -1248,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 @@ -1258,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 diff --git a/src/borg/manifest.py b/src/borg/manifest.py index d25070d8f4..eb1a8897ad 100644 --- a/src/borg/manifest.py +++ b/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)