Implement framework for flexible 2FA

README.rst

@@ -475,6 +475,57 @@ TOTP URI which can be passed to an OTP library to generate codes
    >>> pyotp.parse_uri(e.otp).now()
    799270
 
+multifactor authentication
+--------------------------
+
+PyKeePass supports securing a database using an arbitrary combination of "authentication factors".
+A single factor could be something like a password, a file, or a hardware authenticator.
+
+Factors are arranged into "factor groups". In order to open the database, *one* factor from
+each group must be provided.
+
+Example using a single FIDO2 authenticator to unlock a database:
+
+.. code:: python
+
+    # Import things
+    >>> from pykeepass import PyKeePass, FactorInfo, FactorGroup, FIDO2Factor, create_database
+    # Create new DB
+    >>> db = create_database()
+    # Create a FIDO2 factor
+    >>> fido2_factor = FIDO2Factor(name="MyCoolFIDO")
+    # Create a single factor group with that one factor in it
+    >>> group = FactorGroup(factors=[fido2_factor])
+    # Set PIN to use for the FIDO2 credential
+    >>> factor_data = {"fido2_pin": "my_pin"}
+    # Declare the one factor group is the only contributor to the database composite key
+    >>> factor_info = FactorInfo(comprehensive=True, factor_groups=[group])
+    # Save database
+    >>> db.factor_data = factor_data
+    >>> db.authentication_factors = factor_info
+    >>> db.save()
+    # Reopen database easily later - factor_info is stored inside it
+    >>> kp = PyKeePass(filename, factor_data=factor_data)
+
+Example using a password *and* one of two different keyfiles (the password will always be required):
+
+.. code:: python
+
+    >>> from pykeepass import PyKeePass, FactorInfo, FactorGroup, PasswordFactor, KeyFileFactor, create_database
+    # Password factor
+    >>> password_factor = PasswordFactor(name="MyCoolPassword")
+    # Keyfile factor
+    >>> kf_factor_1 = KeyFileFactor(name="First KF")
+    >>> kf_factor_2 = KeyFileFactor(name="Second KF")
+    # First factor group, password only
+    >>> group_1 = FactorGroup(factors=[password_factor])
+    # Second factor group, either of two key files
+    >>> group_2 = FactorGroup(factors=[kf_factor_1, kf_factor_2])
+    >>> factor_data = {"password": "my_pass", "keyfile": {"First KF": "/kf1", "Second KF": "/kf2"}}
+    >>> factor_info = FactorInfo(comprehensive=True, factor_groups=[group_1, group_2])
+
+It's okay to mix factors of different types within a group.
+
 
 Tests and Debugging
 -------------------

pykeepass/__init__.py

@@ -1,4 +1,5 @@
 from .pykeepass import PyKeePass, create_database
 from .version import __version__
+from .kdbx_parsing.factorinfo import FactorInfo, FactorGroup, FIDO2Factor, PasswordFactor, KeyFileFactor
 
-__all__ = ["PyKeePass", "create_database", "__version__"]
+__all__ = ["PyKeePass", "create_database", "__version__", 'FactorInfo', 'FactorGroup', 'FIDO2Factor', 'KeyFileFactor']

pykeepass/fido2.py

@@ -0,0 +1,147 @@
+import logging
+import random
+
+from fido2.cose import ES256
+from fido2.ctap import CtapError
+from fido2.ctap2.extensions import HmacSecretExtension, CredProtectExtension
+from fido2.hid import CtapHidDevice
+from fido2.client import Fido2Client, UserInteraction
+from fido2.webauthn import PublicKeyCredentialCreationOptions, PublicKeyCredentialRpEntity, \
+    PublicKeyCredentialUserEntity, PublicKeyCredentialParameters, PublicKeyCredentialType, \
+    PublicKeyCredentialDescriptor, PublicKeyCredentialRequestOptions, UserVerificationRequirement
+
+log = logging.getLogger(__name__)
+
+try:
+    from fido2.pcsc import CtapPcscDevice
+except ImportError:
+    CtapPcscDevice = None
+
+FIDO2_FACTOR_RPID = "fido2.keepass.nodomain"
+
+
+class NonInteractive(UserInteraction):
+
+    def __init__(self, fixed_pin):
+        self.fixed_pin = fixed_pin
+
+    def request_pin(self, permissions, rp_id):
+        return self.fixed_pin
+
+
+def _get_all_authenticators():
+    for dev in CtapHidDevice.list_devices():
+        yield dev
+    if CtapPcscDevice:
+        for dev in CtapPcscDevice.list_devices():
+            yield dev
+
+
+def _get_suitable_clients(pin_data):
+    for authenticator in _get_all_authenticators():
+        authenticator_path_string = repr(authenticator)
+
+        if isinstance(pin_data, str):
+            pin_to_use = pin_data
+        else:
+            pin_to_use = pin_data.get(authenticator_path_string, pin_data.get("*", None))
+
+        client = Fido2Client(
+            authenticator,
+            "https://{}".format(FIDO2_FACTOR_RPID),
+            user_interaction=NonInteractive(pin_to_use),
+            extension_types=[
+                HmacSecretExtension,
+                CredProtectExtension
+            ]
+        )
+
+        if "hmac-secret" in client.info.extensions and "credProtect" in client.info.extensions:
+            yield client
+
+
+class FIDOException(Exception):
+    pass
+
+
+def fido2_enroll(pin_data, already_enrolled_credentials):
+    log.info("Enrolling new FIDO2 authenticator")
+
+    # We don't care about the user ID
+    # So long as it doesn't collide with another one for the same authenticator, it's all good
+    user_id = random.randbytes(16)
+
+    chosen_client = next(_get_suitable_clients(pin_data), None)
+    if chosen_client is None:
+        raise FIDOException("Could not find an authenticator supporting the hmac-secret and credProtect extensions")
+
+    credential = chosen_client.make_credential(PublicKeyCredentialCreationOptions(
+        rp=PublicKeyCredentialRpEntity(
+            name="pykeepass",
+            id=FIDO2_FACTOR_RPID
+        ),
+        user=PublicKeyCredentialUserEntity(
+            name="keepass",
+            id=user_id,
+            display_name="KeePass"
+        ),
+        challenge=random.randbytes(32),
+        pub_key_cred_params=[
+            PublicKeyCredentialParameters(
+                type=PublicKeyCredentialType.PUBLIC_KEY,
+                alg=ES256.ALGORITHM
+            )
+        ],
+        exclude_credentials=[
+            PublicKeyCredentialDescriptor(
+                type=PublicKeyCredentialType.PUBLIC_KEY,
+                id=credential_id
+            ) for credential_id in already_enrolled_credentials
+        ],
+        extensions={
+            "hmacCreateSecret": True,
+            "credentialProtectionPolicy": CredProtectExtension.POLICY.REQUIRED,
+            "enforceCredentialProtectionPolicy": True
+        }
+    ))
+
+    if not credential.extension_results.get("hmacCreateSecret", False):
+        raise FIDOException("Authenticator didn't create an HMAC secret!")
+
+    return credential.attestation_object.auth_data.credential_data.credential_id
+
+
+def fido2_get_key_material(pin_data, credential_ids, salt1, salt2, verify_user=True):
+    log.info("Getting keying material from FIDO2 authenticator (with {} potential credentials)".format(len(credential_ids)))
+
+    user_verification = UserVerificationRequirement.REQUIRED if verify_user else UserVerificationRequirement.DISCOURAGED
+    for client in _get_suitable_clients(pin_data):
+        try:
+            assertion_response = client.get_assertion(
+                PublicKeyCredentialRequestOptions(
+                    challenge=random.randbytes(32),
+                    rp_id=FIDO2_FACTOR_RPID,
+                    allow_credentials=[
+                        PublicKeyCredentialDescriptor(
+                            type=PublicKeyCredentialType.PUBLIC_KEY,
+                            id=credential_id
+                        ) for credential_id in credential_ids
+                    ],
+                    user_verification=user_verification,
+                    extensions={
+                        "hmacGetSecret": {
+                            "salt1": salt1,
+                            "salt2": salt2
+                        }
+                    }
+                )
+            )
+            assertion = assertion_response.get_response(0)
+            hmac_response = assertion.extension_results.get("hmacGetSecret", None)
+            if hmac_response is not None:
+                return hmac_response.get("output1", None), hmac_response.get("output2", None)
+        except CtapError as e:
+            if e.code != CtapError.ERR.NO_CREDENTIALS:
+                raise e
+
+    raise FIDOException("No authenticator provided key material")

pykeepass/kdbx_parsing/common.py

@@ -116,7 +116,54 @@ def aes_kdf(key, rounds, key_composite):
     return hashlib.sha256(transformed_key).digest()
 
 
-def compute_key_composite(password=None, keyfile=None):
+def compute_keyfile_part_of_composite(keyfile):
+    """Compute just a keyfile's contribution to a database composite key."""
+    if hasattr(keyfile, "read"):
+        if hasattr(keyfile, "seekable") and keyfile.seekable():
+            keyfile.seek(0)
+        keyfile_bytes = keyfile.read()
+    else:
+        with open(keyfile, 'rb') as f:
+            keyfile_bytes = f.read()
+    # try to read XML keyfile
+    try:
+        tree = etree.fromstring(keyfile_bytes)
+        version = tree.find('Meta/Version').text
+        data_element = tree.find('Key/Data')
+        if version.startswith('1.0'):
+            return base64.b64decode(data_element.text)
+        elif version.startswith('2.0'):
+            # read keyfile data and convert to bytes
+            keyfile_composite = bytes.fromhex(data_element.text.strip())
+            # validate bytes against hash
+            hash = bytes.fromhex(data_element.attrib['Hash'])
+            hash_computed = hashlib.sha256(keyfile_composite).digest()[:4]
+            assert hash == hash_computed, "Keyfile has invalid hash"
+            return keyfile_composite
+        else:
+            raise AttributeError("Invalid version in keyfile")
+    # otherwise, try to read plain keyfile
+    except (etree.XMLSyntaxError, UnicodeDecodeError, AttributeError):
+        try:
+            try:
+                int(keyfile_bytes, 16)
+                is_hex = True
+            except ValueError:
+                is_hex = False
+            # if the length is 32 bytes we assume it is the key
+            if len(keyfile_bytes) == 32:
+                return keyfile_bytes
+            # if the length is 64 bytes we assume the key is hex encoded
+            elif len(keyfile_bytes) == 64 and is_hex:
+                return codecs.decode(keyfile_bytes, 'hex')
+            # anything else may be a file to hash for the key
+            else:
+                return hashlib.sha256(keyfile_bytes).digest()
+        except:
+            raise IOError('Could not read keyfile')
+
+
+def compute_key_composite(password=None, keyfile=None, additional_parts=None):
     """Compute composite key.
     Used in header verification and payload decryption."""
 
@@ -126,55 +173,15 @@ def compute_key_composite(password=None, keyfile=None):
     else:
         password_composite = b''
     # hash the keyfile
-    if keyfile:
-        if hasattr(keyfile, "read"):
-            if hasattr(keyfile, "seekable") and keyfile.seekable():
-                keyfile.seek(0)
-            keyfile_bytes = keyfile.read()
-        else:
-            with open(keyfile, 'rb') as f:
-                keyfile_bytes = f.read()
-        # try to read XML keyfile
-        try:
-            tree = etree.fromstring(keyfile_bytes)
-            version = tree.find('Meta/Version').text
-            data_element = tree.find('Key/Data')
-            if version.startswith('1.0'):
-                keyfile_composite = base64.b64decode(data_element.text)
-            elif version.startswith('2.0'):
-                # read keyfile data and convert to bytes
-                keyfile_composite = bytes.fromhex(data_element.text.strip())
-                # validate bytes against hash
-                hash = bytes.fromhex(data_element.attrib['Hash'])
-                hash_computed = hashlib.sha256(keyfile_composite).digest()[:4]
-                assert hash == hash_computed, "Keyfile has invalid hash"
-            else:
-                raise AttributeError("Invalid version in keyfile")
-        # otherwise, try to read plain keyfile
-        except (etree.XMLSyntaxError, UnicodeDecodeError, AttributeError):
-            try:
-                try:
-                    int(keyfile_bytes, 16)
-                    is_hex = True
-                except ValueError:
-                    is_hex = False
-                # if the length is 32 bytes we assume it is the key
-                if len(keyfile_bytes) == 32:
-                    keyfile_composite = keyfile_bytes
-                # if the length is 64 bytes we assume the key is hex encoded
-                elif len(keyfile_bytes) == 64 and is_hex:
-                    keyfile_composite = codecs.decode(keyfile_bytes, 'hex')
-                # anything else may be a file to hash for the key
-                else:
-                    keyfile_composite = hashlib.sha256(keyfile_bytes).digest()
-            except:
-                raise IOError('Could not read keyfile')
+    keyfile_composite = compute_keyfile_part_of_composite(keyfile) if keyfile else b''
 
-    else:
-        keyfile_composite = b''
+    # create composite key from password, keyfile, and other composites
+    overall_composite = password_composite + keyfile_composite
+    if additional_parts is not None:
+        for part in additional_parts:
+            overall_composite += part
 
-    # create composite key from password and keyfile composites
-    return hashlib.sha256(password_composite + keyfile_composite).digest()
+    return hashlib.sha256(overall_composite).digest()
 
 
 def compute_master(context):
@@ -188,6 +195,34 @@ def compute_master(context):
     return master_key
 
 
+def populate_custom_data(kdbx, d):
+    if len(d.keys()) > 0:
+        vd = Container(
+            version=b'\x00\x01',
+            dict=d,
+        )
+        kdbx.header.value.dynamic_header.update(
+            {
+                "public_custom_data":
+                    Container(
+                        id='public_custom_data',
+                        data=vd,
+                        next_byte=0xFF,
+                    )
+            }
+        )
+    else:
+        # Removing header entirely
+        if "public_custom_data" in kdbx.header.value.dynamic_header:
+            del kdbx.header.value.dynamic_header["public_custom_data"]
+
+    # Beyond Python 3.7, construct makes the base class of a Container be `dict` instead of `OrderedDict`
+    # So emulate move_to_end by removing and re-inserting the element
+    end_el = kdbx.header.value.dynamic_header["end"]
+    del kdbx.header.value.dynamic_header["end"]
+    kdbx.header.value.dynamic_header["end"] = end_el
+
+
 # -------------------- XML Processing --------------------