from django.utils.translation import gettext as _ from rest_framework import authentication from rest_framework.exceptions import AuthenticationFailed from pingcap_sso.models import ApiToken class APITokenAuthentication(authentication.BasicAuthentication): def authenticate_header(self, request): return 'xBasic realm="%s"' % self.www_authenticate_realm def authenticate(self, request): auth = authentication.get_authorization_header(request).split() if not auth or auth[0].lower() != b'bearer': return None if len(auth) == 1: msg = _('Invalid token header. No credentials provided.') raise AuthenticationFailed(msg) elif len(auth) > 2: msg = _('Invalid token header. Token string should not contain spaces.') raise AuthenticationFailed(msg) return self.authenticate_credentials(auth[1]) def authenticate_credentials(self, token): try: token = ApiToken.objects.filter( token=token.decode('utf-8')).select_related('user').get() except Exception: raise AuthenticationFailed(_('Invalid token')) if not token.user.is_active: raise AuthenticationFailed(_('User has not been activated')) return (token.user, token) class SessionAuthentication(authentication.SessionAuthentication): def authenticate_header(self, request): return 'xBasic realm="cookie"' class CsrfExemptSessionAuthentication(authentication.SessionAuthentication): def authenticate_header(self, request): return 'xBasic realm="cookies"' def enforce_csrf(self, request): return