backends.py

# -*- coding: utf-8 -*-
from datetime import datetime
from django.conf import settings
from django.contrib.auth.models import User
from django.utils.timezone import utc
from custom_authlib import client  # client library for custom authentication
from django_auth_custom_example.models import CustomAuthToken


class CustomAuthBackend(object):

    user = None

    def __init__(self):
        pass

    class AuthenticationFailed(Exception):
        pass

    def authenticate(self, **creds):
        if self.user is None:
            try:
                # custom authentication client
                cli = client.CustomAuthClient(settings.AUTH_CUSTOM_SERVER_URI)

                _username = creds.get('username')
                _password = creds.get('password')
                _token = creds.get('auth_token')

                if _token:
                    # check token
                    token = self._token_authenticate(cli, _token)

                elif _username and _password:
                    # password login
                    token = self._username_authenticate(cli,
                                                        _username,
                                                        _password)
                self.user = token.user
            except self.AuthenticationFailed as exc:
                print(exc)
        return self.user

    def _token_authenticate(self, cli, auth_token):
        cli.auth_token = auth_token
        res = cli.detail_token()
        if res.ok is False:
            raise self.AuthenticationFailed('auth token is invalid.')

        _token_type = res.json().get('type')
        if _token_type is None:
            raise self.AuthenticationFailed('token type is invalid.')

        _json = res.json()
        _uid = _json.get('uid')
        _project_code = _json.get('project_code')
        if _uid:
            _user_id = _uid
        else:
            _user_id = auth_token
        try:
            user = User.objects.get(username=_user_id)
        except User.DoesNotExist:
            user = User(username=_user_id)
            user.set_unusable_password()
            user.is_active = True
            user.save()
        try:
            token = CustomAuthToken.objects.get(user=user)
            token.expire = convert_timestamp(
                _json.get('expire'))
            token.token = cli.auth_token
            if _project_code and _uid is None:
                token.project_code = _project_code
        except CustomAuthToken.DoesNotExist:
            token = CustomAuthToken(user=user,
                                    token=cli.auth_token,
                                    expire=convert_timestamp(
                                        _json.get('expire')),
                                    token_type=_token_type)
            if _uid:
                token.uid = _uid
            elif _project_code:
                token.project_code = _project_code
        token.save()

        return token

    def _username_authenticate(self, cli, username, password):
        res_login = cli.login(username, password)

        if res_login.ok is False:
            raise self.AuthenticationFailed('login failure')
        cli.auth_token = res_login.json().get('token')
        res_token = cli.detail_token()

        try:
            user = User.objects.get(username=username)

            # check local user
            if user.password.find('!') != 0:
                raise self.AuthenticationFailed('local user is skipped.')
        except User.DoesNotExist:
            user = User(username=username)
            user.set_unusable_password()
            user.is_active = True
            user.save()

        _expire = convert_timestamp(res_token.json().get('expire'))
        try:
            token = CustomAuthToken.objects.get(user=user)
            token.token = cli.auth_token
            token.expire = _expire
        except CustomAuthToken.DoesNotExist:
            token = CustomAuthToken(user=user,
                                    token=cli.auth_token,
                                    expire=_expire,
                                    token_type=res_token.json().get('type'),
                                    uid=res_token.json().get('uid'))
        token.save()
        return token

    def get_user(self, user_id):
        try:
            return User.objects.get(pk=user_id)
        except User.DoesNotExist:
            return None


def convert_timestamp(timestamp):
    if timestamp is not None:
        return utc.localize(datetime.utcfromtimestamp(timestamp))
    return timestamp