Add JWT Token authenticator

Author: stveit

src/argus/auth/authentication.py

@@ -1,10 +1,15 @@
 from datetime import timedelta
+import urllib.request, json
+
+import jwt
 
 from django.conf import settings
 from django.utils import timezone
-from rest_framework.authentication import TokenAuthentication
+from rest_framework.authentication import TokenAuthentication, BaseAuthentication
 from rest_framework.exceptions import AuthenticationFailed
 
+from .models import User
+
 
 class ExpiringTokenAuthentication(TokenAuthentication):
     EXPIRATION_DURATION = timedelta(days=settings.AUTH_TOKEN_EXPIRES_AFTER_DAYS)
@@ -17,3 +22,55 @@ def authenticate_credentials(self, key):
             raise AuthenticationFailed("Token has expired.")
 
         return user, token
+
+
+class JWTAuthentication(BaseAuthentication):
+    def authenticate(self, request):
+        try:
+            raw_token = self.get_raw_jwt_token(request)
+        except ValueError:
+            return None
+        try:
+            validated_token = jwt.decode(
+                jwt=raw_token,
+                algorithms=["RS256", "RS384", "RS512"],
+                key=self.get_public_key(),
+                options={
+                    "require": [
+                        "exp",
+                        "nbf",
+                        "aud",
+                        "iss",
+                        "sub",
+                    ]
+                },
+                audience=settings.JWT_AUDIENCE,
+                issuer=settings.JWT_ISSUER,
+            )
+        except jwt.exceptions.PyJWTError as e:
+            raise AuthenticationFailed(f"Error validating JWT token: {e}")
+        username = validated_token["sub"]
+        user = User.objects.get(username=username)
+        if not user:
+            raise AuthenticationFailed(f"No user found for username {username}")
+
+        return user, validated_token
+
+    def get_public_key(self):
+        with urllib.request.urlopen(settings.JWK_ENDPOINT) as url:
+            jwk_data = json.load(url)
+        public_key = jwk_data["keys"][0]["n"]
+        return public_key
+
+    def get_raw_jwt_token(self, request):
+        """Raises ValueError if a jwt token could not be found"""
+        auth_header = request.META.get("HTTP_AUTHORIZATION", None)
+        if not auth_header:
+            raise ValueError("No Authorization header found")
+        try:
+            scheme, token = auth_header.split()
+        except ValueError as e:
+            raise ValueError(f"Failed to parse Authorization header: {e}")
+        if scheme.lower() != "bearer":
+            raise ValueError(f"Invalid Authorization scheme: {scheme}")
+        return token

src/argus/site/settings/base.py

@@ -187,6 +187,7 @@
         "argus.auth.authentication.ExpiringTokenAuthentication",
         # For BrowsableAPIRenderer
         "rest_framework.authentication.SessionAuthentication",
+        "argus.auth.authentication.JWTAuthentication",
     ),
     "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
     "DEFAULT_RENDERER_CLASSES": (
@@ -301,3 +302,7 @@
 #
 # SOCIAL_AUTH_DATAPORTEN_FEIDE_KEY = SOCIAL_AUTH_DATAPORTEN_KEY
 # SOCIAL_AUTH_DATAPORTEN_FEIDE_SECRET = SOCIAL_AUTH_DATAPORTEN_SECRET
+
+JWK_ENDPOINT = get_str_env("JWK_ENDPOINT")
+JWT_ISSUER = get_str_env("JWT_ISSUER")
+JWT_AUDIENCE = get_str_env("JWT_AUDIENCE")