mirror of https://github.com/jumpserver/jumpserver
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
#
|
||
|
|
|
||
|
|
from rest_framework import authentication, exceptions
|
||
|
|
from django.utils.translation import ugettext as _
|
||
|
|
|
||
|
|
from common.utils import unsign
|
||
|
|
from .models import User
|
||
|
|
|
||
|
|
|
||
|
|
class APPSignAuthentication(authentication.BaseAuthentication):
|
||
|
|
keyword = 'Sign'
|
||
|
|
model = User
|
||
|
|
|
||
|
|
def authenticate(self, request):
|
||
|
|
auth = authentication.get_authorization_header(request).split()
|
||
|
|
|
||
|
|
if not auth or auth[0].lower() != self.keyword.lower().encode():
|
||
|
|
return None
|
||
|
|
|
||
|
|
if len(auth) == 1:
|
||
|
|
msg = _('Invalid sign header. No credentials provided.')
|
||
|
|
raise exceptions.AuthenticationFailed(msg)
|
||
|
|
elif len(auth) > 2:
|
||
|
|
msg = _('Invalid sign header. Sign string should not contain spaces.')
|
||
|
|
raise exceptions.AuthenticationFailed(msg)
|
||
|
|
|
||
|
|
try:
|
||
|
|
sign = auth[1].decode()
|
||
|
|
except UnicodeError:
|
||
|
|
msg = _('Invalid token header. Sign string should not contain invalid characters.')
|
||
|
|
raise exceptions.AuthenticationFailed(msg)
|
||
|
|
|
||
|
|
return self.authenticate_credentials(sign)
|
||
|
|
|
||
|
|
def authenticate_credentials(self, key):
|
||
|
|
try:
|
||
|
|
token = self.model.objects.select_related('user').get(key=key)
|
||
|
|
except self.model.DoesNotExist:
|
||
|
|
raise exceptions.AuthenticationFailed(_('Invalid token.'))
|
||
|
|
|
||
|
|
if not token.user.is_active:
|
||
|
|
raise exceptions.AuthenticationFailed(_('User inactive or deleted.'))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
pass
|