Merge branch 'develop' into 'main'

chore: Add transports to backup cred

See merge request locker/api-core!446

locker_server/api/v1_0/backup_credentials/serializers.py

@@ -1,7 +1,8 @@
 from rest_framework import serializers
 
 from locker_server.core.entities.user.backup_credential import BackupCredential
-from locker_server.shared.constants.backup_credential import LIST_CREDENTIAL_TYPE, CREDENTIAL_TYPE_HMAC
+from locker_server.shared.constants.backup_credential import LIST_CREDENTIAL_TYPE, CREDENTIAL_TYPE_HMAC, \
+    WEBAUTHN_VALID_TRANSPORTS
 
 
 class ListBackupCredentialSerializer(serializers.Serializer):
@@ -13,6 +14,7 @@ def to_representation(self, instance: BackupCredential):
             "key": instance.key,
             "fd_credential_id": instance.fd_credential_id,
             "fd_random": instance.fd_random,
+            "fd_transports": instance.fd_transports if instance.fd_transports else WEBAUTHN_VALID_TRANSPORTS,
             "name": instance.name,
             "type": instance.type,
             "last_use_date": instance.last_use_date
@@ -31,5 +33,15 @@ class CreateBackupCredentialSerializer(serializers.Serializer):
     key = serializers.CharField(required=False, allow_null=True, allow_blank=True)
     fd_credential_id = serializers.CharField(max_length=255, required=False, allow_blank=True, allow_null=True)
     fd_random = serializers.CharField(max_length=128, required=False, allow_blank=True, allow_null=True)
+    fd_transports = serializers.ListSerializer(
+        child=serializers.CharField(max_length=64, required=True), required=False, allow_empty=True, allow_null=True
+    )
     name = serializers.CharField(max_length=255, required=False, allow_blank=False)
     type = serializers.ChoiceField(choices=LIST_CREDENTIAL_TYPE, required=False, default=CREDENTIAL_TYPE_HMAC)
+
+    def validate(self, data):
+        transports = data.get("fd_transports")
+        if transports and not any(valid_transport in transports for valid_transport in WEBAUTHN_VALID_TRANSPORTS):
+            raise serializers.ValidationError(detail={"transports": ["The transports is not valid"]})
+        data["fd_transports"] = ",".join(transports) if transports else None
+        return data

locker_server/api/v1_0/passwordless/serializers.py

@@ -1,10 +1,21 @@
 from rest_framework import serializers
 
-from locker_server.shared.constants.backup_credential import LIST_CREDENTIAL_TYPE, CREDENTIAL_TYPE_HMAC
+from locker_server.shared.constants.backup_credential import LIST_CREDENTIAL_TYPE, CREDENTIAL_TYPE_HMAC, \
+    WEBAUTHN_VALID_TRANSPORTS
 
 
 class PasswordlessCredentialSerializer(serializers.Serializer):
     credential_id = serializers.CharField(max_length=255)
     random = serializers.CharField(max_length=64, required=False, allow_null=True, allow_blank=True)
+    transports = serializers.ListSerializer(
+        child=serializers.CharField(max_length=64, required=True), required=False, allow_empty=True, allow_null=True
+    )
     name = serializers.CharField(max_length=255, allow_blank=False)
     type = serializers.ChoiceField(choices=LIST_CREDENTIAL_TYPE, required=False, default=CREDENTIAL_TYPE_HMAC)
+
+    def validate(self, data):
+        transports = data.get("transports")
+        if transports and not any(valid_transport in transports for valid_transport in WEBAUTHN_VALID_TRANSPORTS):
+            raise serializers.ValidationError(detail={"transports": ["The transports is not valid"]})
+        data["transports"] = ",".join(transports) if transports else None
+        return data

locker_server/api/v1_0/passwordless/views.py

@@ -9,6 +9,7 @@
 from locker_server.api.api_base_view import APIBaseViewSet
 from locker_server.api.permissions.locker_permissions.passwordless_pwd_permission import PasswordlessPwdPermission
 from locker_server.core.exceptions.user_exception import UserDoesNotExistException
+from locker_server.shared.constants.backup_credential import WEBAUTHN_VALID_TRANSPORTS
 from .serializers import PasswordlessCredentialSerializer
 
 
@@ -41,6 +42,7 @@ def credential(self, request, *args, **kwargs):
                 user_backup_credentials_data.append({
                     "credential_id": backup_credential.fd_credential_id,
                     "random": backup_credential.fd_random,
+                    "transports": backup_credential.fd_transports or WEBAUTHN_VALID_TRANSPORTS,
                     "name": backup_credential.name,
                     "type": backup_credential.type,
                     "creation_date": backup_credential.creation_date,
@@ -49,6 +51,7 @@ def credential(self, request, *args, **kwargs):
             return Response(status=status.HTTP_200_OK, data={
                 "credential_id": user.fd_credential_id,
                 "random": user.fd_random,
+                "transports": user.fd_transports or WEBAUTHN_VALID_TRANSPORTS,
                 "name": user.fd_name,
                 "type": user.fd_type,
                 "creation_date": user.fd_creation_date,
@@ -65,8 +68,10 @@ def credential(self, request, *args, **kwargs):
             name = validated_data.get("name")
             fd_type = validated_data.get("type")
             credential_random = validated_data.get("random") or random.randbytes(16).hex()
+            transports = validated_data.get("transports")
             user = self.user_service.update_passwordless_cred(
                 user=user, fd_credential_id=credential_id, fd_random=credential_random,
+                fd_transports=transports,
                 fd_name=name,
                 fd_type=fd_type
             )

locker_server/api_orm/abstracts/users/backup_credential.py

@@ -22,6 +22,7 @@ class AbstractBackupCredentialORM(models.Model):
     # Passwordless config
     fd_credential_id = models.CharField(max_length=255, null=True)
     fd_random = models.CharField(max_length=128, null=True)
+    fd_transports = models.CharField(max_length=255, blank=True, null=True, default=None)
 
     # Security keys info
     name = models.CharField(max_length=128, null=True, default=None)
@@ -41,3 +42,8 @@ def check_master_password(self, raw_password):
         if not self.master_password:
             return False
         return check_password(raw_password, self.master_password)
+
+    def get_fd_transports(self):
+        if not self.fd_transports:
+            return []
+        return self.fd_transports.split(",")

locker_server/api_orm/abstracts/users/users.py

@@ -42,6 +42,7 @@ class AbstractUserORM(models.Model):
     login_method = models.CharField(max_length=32, default=LOGIN_METHOD_PASSWORD)
     fd_credential_id = models.CharField(max_length=255, null=True)
     fd_random = models.CharField(max_length=128, null=True)
+    fd_transports = models.CharField(max_length=255, blank=True, null=True, default=None)
     fd_name = models.CharField(null=True, max_length=255, default=None)
     fd_type = models.CharField(null=True, max_length=128, default=CREDENTIAL_TYPE_HMAC)
     fd_creation_date = models.FloatField(null=True)
@@ -98,3 +99,8 @@ def get_onboarding_process(self):
         if not self.onboarding_process:
             return DEFAULT_ONBOARDING_PROCESS
         return ast.literal_eval(str(self.onboarding_process))
+
+    def get_fd_transports(self):
+        if not self.fd_transports:
+            return []
+        return self.fd_transports.split(",")

locker_server/api_orm/migrations/0024_auto_20240528_1045.py

@@ -0,0 +1,23 @@
+# Generated by Django 3.2.23 on 2024-05-28 03:45
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api_orm', '0023_auto_20240403_1446'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='backupcredentialorm',
+            name='fd_transports',
+            field=models.CharField(blank=True, default=None, max_length=255, null=True),
+        ),
+        migrations.AddField(
+            model_name='userorm',
+            name='fd_transports',
+            field=models.CharField(blank=True, default=None, max_length=255, null=True),
+        ),
+    ]

locker_server/api_orm/model_parsers/user_parsers.py

@@ -42,6 +42,7 @@ def parse_user(cls, user_orm: UserORM) -> User:
             login_method=user_orm.login_method,
             fd_credential_id=user_orm.fd_credential_id,
             fd_random=user_orm.fd_random,
+            fd_transports=user_orm.get_fd_transports(),
             fd_name=user_orm.fd_name,
             fd_type=user_orm.fd_type,
             fd_creation_date=user_orm.fd_creation_date,
@@ -137,6 +138,7 @@ def parse_backup_credential(cls, backup_credential_orm: BackupCredentialORM) ->
             private_key=backup_credential_orm.private_key,
             fd_credential_id=backup_credential_orm.fd_credential_id,
             fd_random=backup_credential_orm.fd_random,
+            fd_transports=backup_credential_orm.get_fd_transports(),
             kdf_iterations=backup_credential_orm.kdf_iterations,
             kdf=backup_credential_orm.kdf,
             user=cls.parse_user(user_orm=backup_credential_orm.user),

locker_server/api_orm/models/users/backup_credentials.py

@@ -19,6 +19,7 @@ def create(cls, **data):
             creation_date=data.get("creation_date", now()),
             fd_credential_id=data.get("fd_credential_id"),
             fd_random=data.get("fd_random"),
+            fd_transports=data.get("fd_transports"),
             user_id=data.get("user_id"),
             name=data.get("name"),
             type=data.get("type") or data.get("fd_type")

locker_server/api_orm/repositories/user_repository.py

@@ -807,14 +807,16 @@ def update_login_time_user(self, user_id: int, update_data) -> Optional[User]:
         user_orm.save()
         return ModelParser.user_parser().parse_user(user_orm=user_orm)
 
-    def update_passwordless_cred(self, user_id: int, fd_credential_id: str, fd_random: str, fd_name: str,
-                                 fd_type: str = None) -> User:
+    def update_passwordless_cred(self,
+                                 user_id: int, fd_credential_id: str, fd_random: str, fd_transports,
+                                 fd_name: str, fd_type: str = None) -> User:
         try:
             user_orm = UserORM.objects.get(user_id=user_id)
         except UserORM.DoesNotExist:
             return None
         user_orm.fd_credential_id = fd_credential_id
         user_orm.fd_random = fd_random
+        user_orm.fd_transports = fd_transports
         user_orm.fd_name = fd_name
         user_orm.fd_creation_date = now()
         if fd_type is not None:

locker_server/core/entities/user/backup_credential.py

@@ -1,4 +1,5 @@
 import ast
+from typing import List
 
 from locker_server.core.entities.user.user import User
 
@@ -7,7 +8,8 @@ class BackupCredential(object):
     def __init__(self, backup_credential_id: str, user: User,
                  master_password: str = None, master_password_hint: str = "", key: str = None,
                  public_key: str = None, private_key: str = None, creation_date: float = 0,
-                 fd_credential_id: str = None, fd_random: str = None, kdf: int = 0, kdf_iterations: int = 0,
+                 fd_credential_id: str = None, fd_random: str = None, fd_transports: List[str] = None,
+                 kdf: int = 0, kdf_iterations: int = 0,
                  name: str = None, last_use_date: float = None, type: str = None
                  ):
         self._backup_credential_id = backup_credential_id
@@ -19,6 +21,7 @@ def __init__(self, backup_credential_id: str, user: User,
         self._private_key = private_key
         self._fd_credential_id = fd_credential_id
         self._fd_random = fd_random
+        self._fd_transports = fd_transports
         self._kdf_iterations = kdf_iterations
         self._kdf = kdf
         self._user = user
@@ -62,6 +65,10 @@ def fd_credential_id(self):
     def fd_random(self):
         return self._fd_random
 
+    @property
+    def fd_transports(self):
+        return self._fd_transports
+
     @property
     def user(self):
         return self._user

locker_server/core/entities/user/user.py

@@ -1,3 +1,5 @@
+from typing import List
+
 from locker_server.shared.constants.account import DEFAULT_KDF_ITERATIONS, LOGIN_METHOD_PASSWORD, \
     DEFAULT_ONBOARDING_PROCESS
 from locker_server.shared.constants.lang import LANG_ENGLISH
@@ -14,7 +16,7 @@ def __init__(self, user_id: int, internal_id: str = None, creation_date: float =
                  api_key: str = None, timeout: int = 20160, timeout_action: str = "lock", is_leaked: bool = False,
                  use_relay_subdomain: bool = False, last_request_login: float = None, login_failed_attempts: int = 0,
                  login_block_until: float = None, login_method: str = LOGIN_METHOD_PASSWORD,
-                 fd_credential_id: str = None, fd_random: str = None,
+                 fd_credential_id: str = None, fd_random: str = None, fd_transports: List[str] = None,
                  onboarding_process: str = DEFAULT_ONBOARDING_PROCESS, saas_source: str = None,
                  email: str = None, full_name: str = None, language: str = LANG_ENGLISH,
                  is_factor2: bool = False, base32_secret_factor2: str = "", is_super_admin: bool = False,
@@ -49,6 +51,7 @@ def __init__(self, user_id: int, internal_id: str = None, creation_date: float =
         self._login_method = login_method
         self._fd_credential_id = fd_credential_id
         self._fd_random = fd_random
+        self._fd_transports = fd_transports
         self._fd_name = fd_name
         self._fd_type = fd_type
         self._fd_creation_date = fd_creation_date
@@ -183,6 +186,10 @@ def fd_credential_id(self):
     def fd_random(self):
         return self._fd_random
 
+    @property
+    def fd_transports(self):
+        return self._fd_transports
+
     @property
     def onboarding_process(self):
         return self._onboarding_process

locker_server/core/repositories/user_repository.py

@@ -127,8 +127,8 @@ def update_login_time_user(self, user_id: int, update_data) -> Optional[User]:
 
     @abstractmethod
     def update_passwordless_cred(self,
-                                 user_id: int, fd_credential_id: str, fd_random: str, fd_name: str,
-                                 fd_type: str = None) -> User:
+                                 user_id: int, fd_credential_id: str, fd_random: str, fd_transports,
+                                 fd_name: str, fd_type: str = None) -> User:
         pass
 
     @abstractmethod

locker_server/core/services/user_service.py

@@ -115,10 +115,10 @@ def update_user(self, user_id: int, user_update_data) -> Optional[User]:
             raise UserDoesNotExistException
         return user
 
-    def update_passwordless_cred(self, user: User, fd_credential_id: str, fd_random: str, fd_name: str,
+    def update_passwordless_cred(self, user: User, fd_credential_id: str, fd_transports, fd_random: str, fd_name: str,
                                  fd_type: str) -> User:
         return self.user_repository.update_passwordless_cred(
-            user_id=user.user_id, fd_credential_id=fd_credential_id, fd_random=fd_random,
+            user_id=user.user_id, fd_credential_id=fd_credential_id, fd_random=fd_random, fd_transports=fd_transports,
             fd_name=fd_name,
             fd_type=fd_type
         )

locker_server/shared/constants/backup_credential.py

@@ -1,4 +1,16 @@
+from webauthn.helpers.structs import AuthenticatorTransport
+
+
 BACKUP_CREDENTIAL_MAX = 6
 CREDENTIAL_TYPE_HMAC = "hmac"
 CREDENTIAL_TYPE_PRF = "prf"
 LIST_CREDENTIAL_TYPE = [CREDENTIAL_TYPE_PRF, CREDENTIAL_TYPE_HMAC]
+
+
+WEBAUTHN_VALID_TRANSPORTS = [
+    AuthenticatorTransport.USB,
+    AuthenticatorTransport.NFC,
+    AuthenticatorTransport.BLE,
+    AuthenticatorTransport.HYBRID,
+    AuthenticatorTransport.INTERNAL,
+]

requirements.txt

@@ -68,4 +68,6 @@ geoip2==4.7.0
 pandas==2.2.1
 
 notion-client==2.2.1
-httpx==0.27.0
+httpx==0.27.0
+
+webauthn==1.11.1