Untitled

import logging
from typing import NewType

from django.conf import settings
from django.utils.translation import gettext_lazy as _
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication, get_authorization_header

from .models import User

SecretKey = NewType("SecretKey", bytes)
TelegramID = NewType("TelegramID", bytes)


class TelegramClientAuthentication(BaseAuthentication):
    """
    Custom authentication backend for Telegram Microservice.
    Allows TelegramBOT to authenticate users using their ID and SECRET_KEY from BACKEND.
    HTTP header, for example:
        Authorization: SECRET_KEY TELEGRAM_USER_ID
    """

    keyword = settings.SECRET_KEY
    model = None

    def get_model(self):
        if self.model is not None:
            return self.model
        from rest_framework.authtoken.models import Token

        return Token

    def authenticate(self, request):
        # get token and telegram_id from request header
        # [b'xvnPFGWjpCo9n7DLg8Zf2SIAs9bTYj9jVmb3AW9XMyXozDwbC5efYf0Hc6RW90mE', b'123456789']
        auth: list[SecretKey, TelegramID] = get_authorization_header(  # noqa
            request
        ).split()

        if not auth or auth[0].lower() != self.keyword.lower().encode():
            return None
        if len(auth) == 1:
            msg = _("Invalid token header. No credentials provided.")
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = _("Invalid token header. Token string should not contain spaces.")
            raise exceptions.AuthenticationFailed(msg)
        try:
            user_id = auth[1].decode()
        except UnicodeError:
            msg = _(
                "Invalid token header. Token string should not contain invalid characters."
            )
            raise exceptions.AuthenticationFailed(msg)
        res = self.get_user_instance(user_id), None
        return res

    @staticmethod
    def get_user_instance(user_id: int or str) -> User:
        try:
            user = User.objects.get(id=user_id)
        except User.DoesNotExist:
            logging.info(f"User with id {user_id} does not exist.")
            raise exceptions.AuthenticationFailed(_("User does not exist."))
        return user