Customized input fields

[LouisDelbosc]

Hello, I'd like to obtain the pair of keys but instead of having just username and password, I need to have a more complexe structure (because I'm building a mobile app).

To better illustrate, here is a simple unit test:

class UserAuthTest(TestCase):
    def test_user_login(self):
        user = User.objects.create_user(
            phone_number="+1234567890",
            password="12345",
        )
        device = DeviceFactory(user=user)
        response = self.client.post(
            "/api/v1/token/pair",
            {
                "user": {"phone_number": "+1234567890", "password": "12345"},
                "device": {"vendor_uuid": device.vendor_uuid},
            },
            content_type="application/json",
        )
        json = response.json()
        self.assertEqual(json, {
            "id": "some-id",
            "access": "accesstoken",
            "is_new_user": False,
            "is_new_device": False,
            "user": {
                "uuid": user.uuid,
            },
            "device": {
                "uuid": device.uuid,
            }
        })
        self.assertEqual(response.status_code, 200)

I override the schema

class UserSchema(Schema):
    uuid: UUID


class MyTokenObtainPairOutSchema(Schema):
    refresh: str
    access: str
    user: UserSchema


class TokenSessionInputSchema(TokenObtainPairInputSchema):

    def output_schema(self):
        out_dict = self.get_response_schema_init_kwargs()
        out_dict.update(user=UserSchema.from_orm(self._user))
        return MyTokenObtainPairOutSchema(**out_dict)

    @classmethod
    def validate_values(cls, values: Dict) -> Dict:
        user = values["user"]
        return {**values, "user": super().validate_values(user)}

However I got the following error:
{'detail': [{'type': 'missing', 'loc': ['body', 'user_token', 'password'], 'msg': 'Field required'}, {'type': 'missing', 'loc': ['body', 'user_token', 'phone_number'], 'msg': 'Field required'}]}.

In the return of validate_values, he doesn't find the password and phone_number field. So when fixing the validate_values function like following

@classmethod
def validate_values(cls, values: Dict) -> Dict:
        user = values["user"]
        return super().validate_values(user)

I got another error: pydantic_core._pydantic_core.ValidationError: 1 validation error for NinjaResponseSchema response.phone_number but I did not specify phone_number for django-ninja-jwt so I don't know how did he found phone_number.

Well sorry for this long issue. I tried debugging but I don't know how to configure a custom input (and not only username and password) and keep the data inside the classes to have custom behavior (for example creating a device when there is none in the database).

Thanks for your help.

[LouisDelbosc]

With some more test I succeed in two things here is latest testing version:

class DeviceSchema(Schema):
    class Config:
        model = Device
        include = ("vendor_uuid", )
    vendor_uuid: UUID

class DeviceOutSchema(Schema):
    class Config:
        model = Device
        include = ("vendor_uuid", )
    vendor_uuid: UUID
    uuid: UUID

class UserOutSchema(Schema):
    uuid: UUID


class MyTokenObtainPairOutSchema(Schema):
    refresh: str
    access: str
    user: UserOutSchema
    device: DeviceOutSchema


class TokenSessionInputSchema(TokenObtainPairInputSchema):
    device: DeviceSchema

    def output_schema(self):
        out_dict = self.get_response_schema_init_kwargs()
        out_dict.update(user=UserOutSchema.from_orm(self._user))
        return MyTokenObtainPairOutSchema(**out_dict)


@api_controller('/token', tags=['Auth'])
class MyTokenObtainPairController(TokenObtainPairController):
    @route.post(
        "/pair", response=MyTokenObtainPairOutSchema, url_name="token_obtain_pair"
    )
    def obtain_token(self, user_token: TokenSessionInputSchema):
        return user_token.output_schema()

With the controller I don't have the typing issues. I add the device field in the TokenSessionInputSchema and I see I can access those data with self.dict() it seems.
Is this the right way to do it ?
If I want to add side-effects, should I do it in the output_schema ? (creating the device if it doesn't exist)

How should I add the User schema in the TokenSessionInput to have input data in the shape of {user: {**credentials}, device: {device}}

[eadwinCode]

@LouisDelbosc Sorry I think I am confused here. Correct me if I am wrong,

• you want to obtain a pair of keys without a username and password
• you also want to know the best place to create a device if it does not exist

[eadwinCode]

In my opinion, the best way to create a device when its not existing is in validate_values method

class TokenSessionInputSchema(TokenObtainPairInputSchema):
    device: DeviceSchema

    def output_schema(self):
        out_dict = self.get_response_schema_init_kwargs()
        out_dict.update(user=UserOutSchema.from_orm(self._user))
        return MyTokenObtainPairOutSchema(**out_dict)

    @classmethod
    def validate_values(cls, values: Dict) -> Dict:
        update_values = super().validate_values(values)
        # read the device data in `update_values` and search for device in database
        device_schema = DeviceSchema(**update_values['device'])
        device_model = DeviceModel.objects.filter(vendor_uuid=device_schema.vendor_uuid).first()
        
        if not device_model:
            # if there is no device then create one here
            device = DeviceFactory(user=cls._user)
            update_values.update({"device": {"vendor_uuid": device.vendor_uuid}})
        return update_values

[LouisDelbosc]

Nice thank you @eadwinCode, you answer one of my question.

Talking with my team, we'd like to have a passwordless authentication, where we send a code with mail or phone text. Do you we could use the django-ninja-jwt for that ? I saw a blog post using restframework-simple-jwt to achieve it here however instead of putting my token to a user, I'd to link it to a session, a data structure with my user and my device (so a user can be connected with multiples device).

Do you think it's feasible ?

[eadwinCode]

I think if you follow the pattern referenced in the article you shared using NinjaJWT, you will achieve the same result. The article didnt really use restframework-simple-jwt in a deep. So, if you have your user, you can simple call the RefreshToken from ninja_jwt and generate a token for the use. An example

from ninja_jwt.token import RefreshToken

class User(AbstractBaseUser, PermissionsMixin):
    # Our basic user db model
    # This has all the attributes that are set on a user
    # as well as methods to work with them
    # including this method, which issues a user's tokens
    def get_new_tokens(self) -> Dict[str, str]:
        blacklist_tokens(self)
        refresh = RefreshToken.for_user(self)
        return {“refresh”: str(refresh), “access”: str(refresh.access_token)}

About linking it to a session, you might want to create a server session because the client is a mobile phone and not a browser. And ninja-jwt does not offer that solution. But it might be leveraged in one or two actions in the general process

[LouisDelbosc]

I made some modification for my needs, I create others tokens for partial auth and mobile authentications:

class MobileRefreshToken(RefreshToken):
    token_type = "mobile_refresh"

    @classmethod
    def for_session(cls, session) -> "Token":
        session_id = str(session.uuid)
        token = cls().for_user(session.user)
        token["session_id"] = session_id
        return token

I did not create a get_new_tokens() function on my Session model, maybe I'll do it later.

Finally for the tokens, I did not use the output_schema, I created the token manually.

# views.py
@router.post("/sessions/", response=SessionOut)
def partial_authentication(request, data: SessionIn):
    user, usr_created = User.objects.get_or_create(phone_number=data.phone_number)
    device, dev_created = Device.objects.get_or_create(user=user, **data.device.dict())
    try:
        session = get_alive_session(user, device)
    except Session.DoesNotExist:
        session = Session.objects.create(
            user=user,
            device=device,
        )
        session.get_token()  # generate partial_auth_token
    return session


class MFAParams(Schema):
    mfa_code: str
    partial_auth_token: str


class PatchSessionOut(Schema):
    id: str
    user: UserOut
    device: DeviceOut
    tokens: typing.Any

    @staticmethod
    def resolve_tokens(obj):
        refresh = MobileRefreshToken.for_session(obj)
        return {"refresh": str(refresh), "access": str(refresh.access_token)}


@router.patch("/sessions/", response=PatchSessionOut)
def validate_partial_authentication(request, params: MFAParams):
    try:
        token = PartialAuthToken(params.partial_auth_token)
        token.blacklist()
    except TokenError as e:
        raise HttpError(401, "Token not valid") from e

    try:
        session = Session.objects.get(uuid=token.payload["session_id"])
        if session.mfa_code == params.mfa_code:
            session.status == "confirmed"
            session.save()
        else:
            raise HttpError(400, "Wrong code")
    except Session.DoesNotExist:
        raise HttpError(404, "Session not found")

    return session

I put the token.blacklist() just after the return in case the mfa_code is wrong.
Do you think it the right place ? Where should I make sure to blacklist a token ?