haldis/app/auth/microsoft.py

import typing

from flask import Blueprint, url_for, request, redirect, flash, Response
from flask_login import login_user
from microsoftgraph.client import Client

from config import Configuration
from models import User, db

auth_microsoft_bp = Blueprint("auth_microsoft_bp", __name__)

client = Client(Configuration.MICROSOFT_AUTH_ID,
                Configuration.MICROSOFT_AUTH_SECRET,
                account_type='common')  # by default common, thus account_type is optional parameter.


def microsoft_login():
    """Log in using Microsoft"""
    scope = ["openid", "profile", "User.Read", "User.Read.All"]
    url = client.authorization_url(url_for("auth_microsoft_bp.authorized", _external=True), scope, state=None)
    return redirect(url)


@auth_microsoft_bp.route("/login")
def login():
    """Function to handle a user trying to log in"""
    return microsoft_login()


@auth_microsoft_bp.route("callback")  # "/authorized")
def authorized() -> typing.Any:
    # type is 'typing.Union[str, Response]', but this errors due to
    #   https://github.com/python/mypy/issues/7187
    """Check authorized status"""

    oauth_code = request.args['code']

    resp = client.exchange_code(url_for("auth_microsoft_bp.authorized", _external=True), oauth_code)

    # access_token = resp.data['access_token']
    # id_token = resp.data['id_token']
    # expires_in = resp.data['expires_in']

    client.set_token(resp.data)

    resp = client.users.get_me()
    # print(resp.data)

    username = resp.data['userPrincipalName']
    microsoft_uuid = resp.data['id']

    user = User.query.filter_by(username=username).first()

    if username and user:
        return login_and_redirect_user(user)
    elif username:
        # TODO Save 'ugent_username' or something similar
        user = create_user(username, microsoft_uuid)
        return login_and_redirect_user(user)

    flash("You're not allowed to enter, please contact a system administrator")
    return redirect(url_for("general_bp.home"))


def login_and_redirect_user(user) -> Response:
    """Log in the user and then redirect them"""
    login_user(user)
    return redirect(url_for("general_bp.home"))


def create_user(username, microsoft_uuid) -> User:
    """Create a temporary user if it is needed"""
    user = User()
    user.configure(username, False, 1, microsoft_uuid)
    db.session.add(user)
    db.session.commit()
    return user