Feat/auth0

app/authentication/apps.py

@@ -0,0 +1,8 @@
+from django.apps import AppConfig
+
+class AuthenticationConfig(AppConfig):
+    name = 'app.authentication'
+
+    def ready(self):
+        # Add the set_user_groups signal receiver to user creation.
+        import app.authentication.signals

app/authentication/auth0.py

@@ -0,0 +1,70 @@
+import time
+
+import requests
+
+from app.constants import AUTH0_CLIENT_ID, AUTH0_CLIENT_SECRET, AUTH0_DOMAIN
+
+
+def get_user_information(user_id):
+    """
+    Gets zipped study programs + start years, name and Feide username from Auth0 given the user_id.
+
+    Example: "olanord", "Ola Nordmann", "[email protected]", [('BIDATA', '2020')].
+    """
+    token_manager = ManagementTokenManager()
+
+    response = requests.get(
+        f"https://{AUTH0_DOMAIN}/api/v2/users/{user_id}",
+        headers={"Authorization": f"Bearer {token_manager.get_token()}"},
+    ).json()
+
+    feide_username = response["nickname"]
+    name = response["name"]
+    email = response["email"]
+
+    # Example format: ['fc:fs:fs:kull:ntnu.no:BIDATA:2020H']
+    metadata = response["app_metadata"]["programs"]
+
+    programs = [p.split(":")[5] for p in metadata]
+    years = [p.split(":")[6][:4] for p in metadata]
+
+    return feide_username, name, email, zip(programs, years)
+
+
+class Singleton(type):
+    _instances = {}
+
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+
+class ManagementTokenManager(metaclass=Singleton):
+    """
+    Singleton class for getting and refreshing Auth0 Management API tokens.
+    """
+
+    def __init__(self):
+        self.token, self.duration = self._get_management_token()
+        self.timestamp = time.time()
+
+    def _get_management_token(self):
+        response = requests.post(
+            f"https://{AUTH0_DOMAIN}/oauth/token",
+            data={
+                "grant_type": "client_credentials",
+                "client_id": AUTH0_CLIENT_ID,
+                "client_secret": AUTH0_CLIENT_SECRET,
+                "audience": f"https://{AUTH0_DOMAIN}/api/v2/",
+            },
+        ).json()
+
+        return response["access_token"], response["expires_in"]
+
+    def get_token(self):
+        # Refresh token if expired, then return token.
+        if time.time() > self.timestamp + self.duration:
+            self.__init__()
+
+        return self.token

app/authentication/signals.py

@@ -0,0 +1,63 @@
+from app.content.models.user import User
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+from app.authentication.auth0 import get_user_information, get_user_information
+from app.common.enums import Groups
+from app.group.models.group import Group
+from app.group.models.membership import Membership
+
+@receiver(post_save, sender=User)
+def set_user_info_and_groups(sender, instance: User, created, **kwargs):
+    # Only run this when the user if first created and if not loading data
+    if kwargs.get("raw") or not created:
+        return
+
+    feide_username, name, email, study_programs = get_user_information(instance.user_id) # Return example: [('BIDATA', 2020), ...]
+
+    if feide_username:
+        instance.feide_id = feide_username
+
+    if name:
+        instance.first_name = name.split()[0]
+        instance.last_name = " ".join(name.split()[1:])
+
+    if email:
+        instance.email = email
+
+    if feide_username or name or email:
+        instance.save()
+
+    # For every program reported by Feide, try to add user groups.
+    for program in study_programs:
+        program_slug = _get_program_group_slug(program)
+        program_year = _get_program_year(program)
+
+        # Do not add any groups if program is not part of TIHLDE.
+        if not program_slug:
+            continue
+
+        # Automatically activate account when program is verified.
+        TIHLDE = Group.objects.get(slug=Groups.TIHLDE)
+        Membership.objects.get_or_create(user=instance, group=TIHLDE)
+
+        program_group = Group.objects.get(slug=program_slug)
+        Membership.objects.get_or_create(user=instance, group = program_group)
+
+        year_group = Group.objects.get(slug=program_year)
+        Membership.objects.get_or_create(user=instance, group = year_group)
+
+
+def _get_program_group_slug(program):
+    program_codes = ["BIDATA", "ITBAITBEDR", "BDIGSEC", "ITMAIKTSA", "ITBAINFODR", "ITBAINFO"]
+    program_slugs = ["dataingenir", "digital-forretningsutvikling", "digital-infrastruktur-og-cybersikkerhet", "digital-samhandling", "drift-studie", "informasjonsbehandling"]
+
+    try:
+        index = program_codes.index(program[0])
+    except:
+        return None
+
+    return program_slugs[index]
+
+def _get_program_year(program):
+    return program[1]

app/authentication/utils.py

@@ -0,0 +1,51 @@
+import json
+
+from django.contrib.auth import authenticate
+
+import jwt
+import requests
+
+from app.constants import AUTH0_AUDIENCE, AUTH0_DOMAIN
+from app.content.models.user import User
+
+
+def get_jwt_from_request(request):
+    auth = request.META.get("HTTP_AUTHORIZATION", None)
+    parts = auth.split()
+    token = parts[1]
+
+    return token
+
+
+def decode_jwt(token):
+    header = jwt.get_unverified_header(token)
+    jwks = requests.get(f"https://{AUTH0_DOMAIN}/.well-known/jwks.json").json()
+
+    public_key = None
+    for jwk in jwks["keys"]:
+        if jwk["kid"] == header["kid"]:
+            public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk))
+
+    if public_key is None:
+        raise Exception("Public key not found while decoding token.")
+
+    return jwt.decode(
+        token,
+        public_key,
+        audience=AUTH0_AUDIENCE,
+        issuer=f"https://{AUTH0_DOMAIN}/",
+        algorithms=["RS256"],
+    )
+
+
+def authenticate_user_with_decoded_jwt(payload):
+    user_id = payload.get("sub")
+    authenticate(remote_user=user_id)
+
+    return user_id
+
+
+def get_user_from_request(request):
+    user_id = authenticate_user_with_decoded_jwt(decode_jwt(get_jwt_from_request(request)))
+
+    return User.objects.get(user_id=user_id)