149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
# Copyright (C) 2014-2015 Andrey Antukh <niwi@niwi.be>
|
|
# Copyright (C) 2014-2015 Jesús Espino <jespinog@gmail.com>
|
|
# Copyright (C) 2014-2015 David Barragán <bameda@dbarragan.com>
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
# This code is partially taken from django-rest-framework:
|
|
# Copyright (c) 2011-2014, Tom Christie
|
|
|
|
"""
|
|
Provides various authentication policies.
|
|
"""
|
|
import base64
|
|
|
|
from django.contrib.auth import authenticate
|
|
from django.middleware.csrf import CsrfViewMiddleware
|
|
|
|
from taiga.base import exceptions
|
|
|
|
from . import HTTP_HEADER_ENCODING
|
|
|
|
|
|
def get_authorization_header(request):
|
|
"""
|
|
Return request's 'Authorization:' header, as a bytestring.
|
|
|
|
Hide some test client ickyness where the header can be unicode.
|
|
"""
|
|
auth = request.META.get('HTTP_AUTHORIZATION', b'')
|
|
if type(auth) == type(''):
|
|
# Work around django test client oddness
|
|
auth = auth.encode(HTTP_HEADER_ENCODING)
|
|
return auth
|
|
|
|
|
|
class CSRFCheck(CsrfViewMiddleware):
|
|
def _reject(self, request, reason):
|
|
# Return the failure reason instead of an HttpResponse
|
|
return reason
|
|
|
|
|
|
class BaseAuthentication(object):
|
|
"""
|
|
All authentication classes should extend BaseAuthentication.
|
|
"""
|
|
|
|
def authenticate(self, request):
|
|
"""
|
|
Authenticate the request and return a two-tuple of (user, token).
|
|
"""
|
|
raise NotImplementedError(".authenticate() must be overridden.")
|
|
|
|
def authenticate_header(self, request):
|
|
"""
|
|
Return a string to be used as the value of the `WWW-Authenticate`
|
|
header in a `401 Unauthenticated` response, or `None` if the
|
|
authentication scheme should return `403 Permission Denied` responses.
|
|
"""
|
|
pass
|
|
|
|
|
|
class BasicAuthentication(BaseAuthentication):
|
|
"""
|
|
HTTP Basic authentication against username/password.
|
|
"""
|
|
www_authenticate_realm = 'api'
|
|
|
|
def authenticate(self, request):
|
|
"""
|
|
Returns a `User` if a correct username and password have been supplied
|
|
using HTTP Basic authentication. Otherwise returns `None`.
|
|
"""
|
|
auth = get_authorization_header(request).split()
|
|
|
|
if not auth or auth[0].lower() != b'basic':
|
|
return None
|
|
|
|
if len(auth) == 1:
|
|
msg = 'Invalid basic header. No credentials provided.'
|
|
raise exceptions.AuthenticationFailed(msg)
|
|
elif len(auth) > 2:
|
|
msg = 'Invalid basic header. Credentials string should not contain spaces.'
|
|
raise exceptions.AuthenticationFailed(msg)
|
|
|
|
try:
|
|
auth_parts = base64.b64decode(auth[1]).decode(HTTP_HEADER_ENCODING).partition(':')
|
|
except (TypeError, UnicodeDecodeError):
|
|
msg = 'Invalid basic header. Credentials not correctly base64 encoded'
|
|
raise exceptions.AuthenticationFailed(msg)
|
|
|
|
userid, password = auth_parts[0], auth_parts[2]
|
|
return self.authenticate_credentials(userid, password)
|
|
|
|
def authenticate_credentials(self, userid, password):
|
|
"""
|
|
Authenticate the userid and password against username and password.
|
|
"""
|
|
user = authenticate(username=userid, password=password)
|
|
if user is None or not user.is_active:
|
|
raise exceptions.AuthenticationFailed('Invalid username/password')
|
|
return (user, None)
|
|
|
|
def authenticate_header(self, request):
|
|
return 'Basic realm="%s"' % self.www_authenticate_realm
|
|
|
|
|
|
class SessionAuthentication(BaseAuthentication):
|
|
"""
|
|
Use Django's session framework for authentication.
|
|
"""
|
|
|
|
def authenticate(self, request):
|
|
"""
|
|
Returns a `User` if the request session currently has a logged in user.
|
|
Otherwise returns `None`.
|
|
"""
|
|
|
|
# Get the underlying HttpRequest object
|
|
request = request._request
|
|
user = getattr(request, 'user', None)
|
|
|
|
# Unauthenticated, CSRF validation not required
|
|
if not user or not user.is_active:
|
|
return None
|
|
|
|
self.enforce_csrf(request)
|
|
|
|
# CSRF passed with authenticated user
|
|
return (user, None)
|
|
|
|
def enforce_csrf(self, request):
|
|
"""
|
|
Enforce CSRF validation for session based authentication.
|
|
"""
|
|
reason = CSRFCheck().process_view(request, None, (), {})
|
|
if reason:
|
|
# CSRF failed, bail with explicit error message
|
|
raise exceptions.AuthenticationFailed('CSRF Failed: %s' % reason)
|