enviPy-bayer/epauth/views.py

import logging

import msal
from django.conf import settings as s
from django.contrib.auth import get_user_model
from django.contrib.auth import login
from django.http import HttpResponse
from django.shortcuts import redirect

from epdb.logic import UserManager, GroupManager
from epdb.models import Group
from epdb.views import get_remote_address

auth_log = logging.getLogger("auth")


def get_msal_app_with_cache(request):
    """
    Create MSAL app with session-based token cache.
    """
    cache = msal.SerializableTokenCache()

    # Load cache from session if it exists
    if request.session.get("msal_token_cache"):
        cache.deserialize(request.session["msal_token_cache"])

    msal_app = msal.ConfidentialClientApplication(
        client_id=s.MS_ENTRA_CLIENT_ID,
        client_credential=s.MS_ENTRA_CLIENT_SECRET,
        authority=s.MS_ENTRA_AUTHORITY,
        token_cache=cache,
    )

    return msal_app, cache


def entra_login(request):

    auth_log.info(f"Login request from {get_remote_address(request)}")

    msal_app = msal.ConfidentialClientApplication(
        client_id=s.MS_ENTRA_CLIENT_ID,
        client_credential=s.MS_ENTRA_CLIENT_SECRET,
        authority=s.MS_ENTRA_AUTHORITY,
    )

    flow = msal_app.initiate_auth_code_flow(
        scopes=s.MS_ENTRA_SCOPES, redirect_uri=s.MS_ENTRA_REDIRECT_URI
    )

    request.session["msal_auth_flow"] = flow
    return redirect(flow["auth_uri"])


def entra_callback(request):
    msal_app, cache = get_msal_app_with_cache(request)

    flow = request.session.pop("msal_auth_flow", None)
    if not flow:
        return redirect("/")

    # Acquire token using the flow and callback request
    result = msal_app.acquire_token_by_auth_code_flow(flow, request.GET)

    if "error" in result:
        auth_log.error(f"Login attempt by {get_remote_address(request)} failed due to {result['error']}")
        return redirect("/")

    # Save the token cache to session
    if cache.has_state_changed:
        request.session["msal_token_cache"] = cache.serialize()

    claims = result["id_token_claims"]

    user_name = claims.get("name")
    # preferred_username is a fallback for 2nd CWID
    user_email = claims.get("emailaddress", claims.get("email", claims.get("preferred_username")))
    user_oid = claims.get("oid")

    if not all([user_name, user_email, user_oid]):
        raise ValueError("Missing required claims in ID token")

    # Get implementing class
    User = get_user_model()

    registered = False
    if User.objects.filter(uuid=user_oid).exists():
        u = User.objects.get(uuid=user_oid)

        if u.username != user_name:
            u.username = user_name
            u.save()

    else:
        auth_log.info(f"Registering {user_name} with OID {user_oid}")
        u = UserManager.create_user(user_name, user_email, None, uuid=user_oid, is_active=True)
        registered = True

    auth_log.info(f"User {user_name} {'(admin) ' if u.is_superuser else ''}with OID {user_oid} successfully logged in as {u.username} from {get_remote_address(request)}")
    login(request, u)

    # EDIT START

    # Ensure groups exists in eP
    for id, name in s.ENTRA_SECRET_GROUPS.items():
        if not Group.objects.filter(uuid=id).exists():
            g = GroupManager.create_group(User.objects.get(username="admin"), name, f"Synced Entra Group {name} ",
                                          uuid=id)
        else:
            g = Group.objects.get(uuid=id)
        # Ensure its secret
        g.secret = True
        g.save()

    for id, name in s.ENTRA_GROUPS.items():
        if not Group.objects.filter(uuid=id).exists():
            g = GroupManager.create_group(User.objects.get(username="admin"), name, f"Synced Entra Group {name} ",
                                          uuid=id)
        else:
            g = Group.objects.get(uuid=id)

    sync_groups = list(s.ENTRA_GROUPS.keys()) + list(s.ENTRA_SECRET_GROUPS.keys())
    user_groups = claims.get("groups", [])

    for uuid in sync_groups:
        if uuid in user_groups:
            g = Group.objects.get(uuid=uuid)
            if not g.user_member.contains(u):
                g.user_member.add(u)
                auth_log.info(f"Login Group Sync: Adding {u.username} to Group {g.name} ({ uuid })")
        else:
            g = Group.objects.get(uuid=uuid)
            if g.user_member.contains(u):
                g.user_member.remove(u)
                auth_log.info(f"Login Group Sync: Removing {u.username} from Group {g.name} ({ uuid })")

    if registered:
        # #72 make package secret if user is part of a secret group
        for id, name in s.ENTRA_SECRET_GROUPS.items():
            group = Group.objects.get(uuid=id)
            if group.user_member.contains(u):
                # User is eligible for secrete
                pack = u.default_package
                pack.data_pool = group
                pack.classification_level = pack.Classification.SECRET
                pack.save()

    # EDIT END

    return redirect(s.SERVER_URL)  # Handle errors


def get_access_token_from_request(request, scopes=None):
    """
    Get an access token from the request using MSAL token cache.
    """

    # Check if auth via Access Token
    if request.headers.get("Authorization"):
        return {"access_token": request.headers.get("Authorization").split(" ")[1]}

    if scopes is None:
        scopes = s.MS_ENTRA_SCOPES

    # Get user from request (must be authenticated)
    if not request.user.is_authenticated:
        return None

    # Create MSAL app with persistent cache
    msal_app, cache = get_msal_app_with_cache(request)

    # Try to get accounts from cache
    accounts = msal_app.get_accounts()

    if not accounts:
        return None

    # Find the account that matches the current user
    user_account = None
    for account in accounts:
        if account.get("local_account_id") == str(request.user.uuid):
            user_account = account
            break

    # If no matching account found, use the first available account
    if not user_account and accounts:
        user_account = accounts[0]

    if not user_account:
        return None

    # Try to acquire token silently from cache
    result = msal_app.acquire_token_silent(scopes=scopes, account=user_account)

    # Save cache changes back to session
    if cache.has_state_changed:
        request.session["msal_token_cache"] = cache.serialize()

    if result and "access_token" in result:
        return result

    return None


def get_token(request):
    token = get_access_token_from_request(request)
    msg = f"{token}"
    return HttpResponse(msg, content_type='text/plain')