__init__.py 3.75 KB
Newer Older
Lukas Burgey's avatar
Lukas Burgey committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

import logging
import json

from urllib.error import HTTPError
from urllib.request import Request, urlopen
from rest_framework.authentication import BaseAuthentication, SessionAuthentication
from rest_framework.exceptions import AuthenticationFailed

from . import utils
from .models import OIDCConfig


LOGGER = logging.getLogger(__name__)


class OIDCTokenAuthBackend:

    def get_userinfo(self, oidc_client, access_token):
        req = Request(
            oidc_client.provider_info['userinfo_endpoint'],
        )
        req.add_header('Authorization', access_token)

        # execute request
        userinfo_bytes = urlopen(req).read()

        user_info = json.loads(userinfo_bytes.decode('UTF-8'))
        user_info['iss'] = oidc_client.provider_info['issuer']

        return user_info

    # raises OIDCConfig.DoesNotExist if no idp can be determined
    def get_idp(self, request):
        # OPTION 1: issuer set in the 'X-Issuer' header
        if 'HTTP_X_ISSUER' in request.META:
            return OIDCConfig.objects.get(issuer_uri=request.META['HTTP_X_ISSUER'])

        # OPTION 2: issuer set in users session (before redirecting to IdP)
        idp_id = utils.get_session(request, 'idp_id', None)
        if idp_id is not None:
            try:
                return OIDCConfig.objects.get(id=idp_id)
            except OIDCConfig.DoesNotExist:
                pass

        raise OIDCConfig.DoesNotExist('Unable to determine IdP')

    def get_user(self, user_id):
        from feudal.backend.models.users import User
        query = User.objects.filter(
            user_type='oidcuser',
            pk=user_id,
        )
        if query.exists():
            if len(query) == 1:
                return query.first()

            LOGGER.error(
                'OIDCTokenAuthBackend: query for user id %s: %s results',
                user_id,
                len(query),
            )
            return None

        LOGGER.error(
            'OIDCTokenAuthBackend: query for user id %s: no results',
            user_id,
        )
        return None

    def authenticate(self, request):
        if 'HTTP_AUTHORIZATION' not in request.META:
            raise AuthenticationFailed(
                detail='No token. Set the HTTP Authorization header to the value of your access token.',
            )

        access_token = request.META['HTTP_AUTHORIZATION']

        from feudal.backend.models.users import User

        try:
            idp = self.get_idp(request)
        except OIDCConfig.DoesNotExist:
            raise AuthenticationFailed(
                detail='Unable to determine IdP. Set \'X-Issuer\' header to the issuer URI of your access token.',
            )

        # get the user info from the idp
        try:
            userinfo = self.get_userinfo(
                idp,
                access_token,
            )
        except HTTPError as http_error:
            raise AuthenticationFailed(detail='Error retrieving user info: {}'.format(http_error))

        try:
            user = User.get_user(
                userinfo,
                idp,
            )
            LOGGER.info('Authenticated: %s', user)
            return user

        except User.DoesNotExist:
            raise AuthenticationFailed(detail='No such user')

        except Exception as exception:
            LOGGER.error(exception)
            raise AuthenticationFailed(detail='Error during authentication')


class CsrfExemptSessionAuthentication(SessionAuthentication):
    def enforce_csrf(self, request):
        # No CSRF enforcing
        return


class OIDCTokenAuthHTTPBackend(BaseAuthentication):

    def authenticate_header(self, request):
        return 'Bearer realm="openidconnect"'

    def authenticate(self, request):
        return (OIDCTokenAuthBackend().authenticate(request), None)