-
-
Notifications
You must be signed in to change notification settings - Fork 567
/
authentication.py
93 lines (78 loc) · 3.44 KB
/
authentication.py
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
"""API Authentication"""
from typing import Any, Optional
from django.conf import settings
from rest_framework.authentication import BaseAuthentication, get_authorization_header
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.request import Request
from structlog.stdlib import get_logger
from authentik.core.middleware import KEY_AUTH_VIA, LOCAL
from authentik.core.models import Token, TokenIntents, User
from authentik.outposts.models import Outpost
from authentik.providers.oauth2.constants import SCOPE_AUTHENTIK_API
from authentik.providers.oauth2.models import RefreshToken
LOGGER = get_logger()
def validate_auth(header: bytes) -> str:
"""Validate that the header is in a correct format,
returns type and credentials"""
auth_credentials = header.decode().strip()
if auth_credentials == "" or " " not in auth_credentials:
return None
auth_type, _, auth_credentials = auth_credentials.partition(" ")
if auth_type.lower() != "bearer":
LOGGER.debug("Unsupported authentication type, denying", type=auth_type.lower())
raise AuthenticationFailed("Unsupported authentication type")
if auth_credentials == "": # nosec # noqa
raise AuthenticationFailed("Malformed header")
return auth_credentials
def bearer_auth(raw_header: bytes) -> Optional[User]:
"""raw_header in the Format of `Bearer ....`"""
auth_credentials = validate_auth(raw_header)
if not auth_credentials:
return None
if not hasattr(LOCAL, "authentik"):
LOCAL.authentik = {}
# first, check traditional tokens
key_token = Token.filter_not_expired(
key=auth_credentials, intent=TokenIntents.INTENT_API
).first()
if key_token:
LOCAL.authentik[KEY_AUTH_VIA] = "api_token"
return key_token.user
# then try to auth via JWT
jwt_token = RefreshToken.filter_not_expired(
refresh_token=auth_credentials, _scope__icontains=SCOPE_AUTHENTIK_API
).first()
if jwt_token:
# Double-check scopes, since they are saved in a single string
# we want to check the parsed version too
if SCOPE_AUTHENTIK_API not in jwt_token.scope:
raise AuthenticationFailed("Token invalid/expired")
LOCAL.authentik[KEY_AUTH_VIA] = "jwt"
return jwt_token.user
# then try to auth via secret key (for embedded outpost/etc)
user = token_secret_key(auth_credentials)
if user:
LOCAL.authentik[KEY_AUTH_VIA] = "secret_key"
return user
raise AuthenticationFailed("Token invalid/expired")
def token_secret_key(value: str) -> Optional[User]:
"""Check if the token is the secret key
and return the service account for the managed outpost"""
from authentik.outposts.managed import MANAGED_OUTPOST
if value != settings.SECRET_KEY:
return None
outposts = Outpost.objects.filter(managed=MANAGED_OUTPOST)
if not outposts:
return None
outpost = outposts.first()
return outpost.user
class TokenAuthentication(BaseAuthentication):
"""Token-based authentication using HTTP Bearer authentication"""
def authenticate(self, request: Request) -> tuple[User, Any] | None:
"""Token-based authentication using HTTP Bearer authentication"""
auth = get_authorization_header(request)
user = bearer_auth(auth)
# None is only returned when the header isn't set.
if not user:
return None
return (user, None) # pragma: no cover