diff --git a/app/app.py b/app/app.py index e4c4e07..b6658d2 100755 --- a/app/app.py +++ b/app/app.py @@ -15,15 +15,17 @@ from flask_bootstrap import Bootstrap, StaticCDN from flask_debugtoolbar import DebugToolbarExtension from flask_login import LoginManager from flask_migrate import Migrate, MigrateCommand -from flask_oauthlib.client import OAuth, OAuthException from flask_script import Manager, Server -from login import init_login from markupsafe import Markup + +from admin import init_admin +from auth.login import init_login +from auth.zeus import init_oauth +from config import Configuration from models import db from models.anonymous_user import AnonymouseUser from sentry_sdk.integrations.flask import FlaskIntegration from utils import euro_string, price_range_string, ignore_none -from zeus import init_oauth def register_plugins(app: Flask) -> Manager: @@ -100,18 +102,22 @@ def add_routes(application: Flask) -> None: # import views # TODO convert to blueprint # import views.stats # TODO convert to blueprint - from login import auth_bp + from auth.login import auth_bp + from auth.microsoft import auth_microsoft_bp + from auth.zeus import auth_zeus_bp from views.debug import debug_bp from views.general import general_bp from views.order import order_bp from views.stats import stats_blueprint - from zeus import oauth_bp application.register_blueprint(general_bp, url_prefix="/") application.register_blueprint(order_bp, url_prefix="/order") application.register_blueprint(stats_blueprint, url_prefix="/stats") application.register_blueprint(auth_bp, url_prefix="/") - application.register_blueprint(oauth_bp, url_prefix="/") + if Configuration.ENABLE_MICROSOFT_AUTH: + application.register_blueprint(auth_microsoft_bp, + url_prefix="/users/auth/microsoft_graph_auth") # "/auth/microsoft") + application.register_blueprint(auth_zeus_bp, url_prefix="/auth/zeus") if application.debug: application.register_blueprint(debug_bp, url_prefix="/debug") @@ -175,6 +181,10 @@ def create_app(): add_routes(app) add_template_filters(app) + @app.context_processor + def inject_config(): + return dict(configuration=Configuration) + return app, app_manager diff --git a/app/login.py b/app/auth/login.py similarity index 67% rename from app/login.py rename to app/auth/login.py index a7382fa..0cc621f 100644 --- a/app/login.py +++ b/app/auth/login.py @@ -1,31 +1,25 @@ -"Script for everything related to logging in and out" +"""Script for everything related to logging in and out""" from flask import Blueprint, abort, redirect, session, url_for from flask_login import current_user, logout_user from models import User from werkzeug.wrappers import Response -from zeus import zeus_login auth_bp = Blueprint("auth_bp", __name__) def init_login(app) -> None: - "Initialize the login" + """Initialize the login""" + # pylint: disable=W0612 @app.login_manager.user_loader def load_user(userid) -> User: - "Load the user" + """Load the user""" return User.query.filter_by(id=userid).first() -@auth_bp.route("/login") -def login(): - "Function to handle a user trying to log in" - return zeus_login() - - @auth_bp.route("/logout") def logout() -> Response: - "Function to handle a user trying to log out" + """Function to handle a user trying to log out""" if "zeus_token" in session: session.pop("zeus_token", None) logout_user() @@ -33,6 +27,6 @@ def logout() -> Response: def before_request() -> None: - "Function for what has to be done before a request" + """Function for what has to be done before a request""" if current_user.is_anonymous() or not current_user.is_allowed(): abort(401) diff --git a/app/auth/microsoft.py b/app/auth/microsoft.py new file mode 100644 index 0000000..059f4eb --- /dev/null +++ b/app/auth/microsoft.py @@ -0,0 +1,77 @@ +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="ugentbe.onmicrosoft.com") + + +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) + client.set_token(resp.data) + + resp = client.users.get_me() + microsoft_uuid = resp.data['id'] + username = resp.data['userPrincipalName'] + + # Fail if fields are not populated + if not microsoft_uuid or not username: + flash("You're not allowed to enter, please contact a system administrator") + return redirect(url_for("general_bp.home")) + + # Find existing user by Microsoft UUID (userPrincipalName can change) + user = User.query.filter_by(microsoft_uuid=microsoft_uuid).first() + if user: + return login_and_redirect_user(user) + + # Find existing user by username (pre-existing account) + user = User.query.filter_by(username=username).first() + if user: + return login_and_redirect_user(user) + + # No user found, create a new one + user = create_user(username, microsoft_uuid=microsoft_uuid) + return login_and_redirect_user(user) + + +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=microsoft_uuid) + db.session.add(user) + db.session.commit() + return user diff --git a/app/zeus.py b/app/auth/zeus.py similarity index 78% rename from app/zeus.py rename to app/auth/zeus.py index 09a4b04..2e3b7cf 100644 --- a/app/zeus.py +++ b/app/auth/zeus.py @@ -4,24 +4,30 @@ import typing from flask import (Blueprint, current_app, flash, redirect, request, session, url_for) from flask_login import login_user -from flask_oauthlib.client import OAuth, OAuthException +from flask_oauthlib.client import OAuth, OAuthException, OAuthRemoteApp from models import User, db from werkzeug.wrappers import Response -oauth_bp = Blueprint("oauth_bp", __name__) +auth_zeus_bp = Blueprint("auth_zeus_bp", __name__) def zeus_login(): - "Log in using ZeusWPI" + """Log in using ZeusWPI""" return current_app.zeus.authorize( - callback=url_for("oauth_bp.authorized", _external=True)) + callback=url_for("auth_zeus_bp.authorized", _external=True)) -@oauth_bp.route("/login/zeus/authorized") +@auth_zeus_bp.route("/login") +def login(): + """Function to handle a user trying to log in""" + return zeus_login() + + +@auth_zeus_bp.route("/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" + """Check authorized status""" resp = current_app.zeus.authorized_response() if resp is None: # pylint: disable=C0301 @@ -45,8 +51,8 @@ def authorized() -> typing.Any: return redirect(url_for("general_bp.home")) -def init_oauth(app): - "Initialize the OAuth for ZeusWPI" +def init_oauth(app) -> OAuthRemoteApp: + """Initialize the OAuth for ZeusWPI""" oauth = OAuth(app) zeus = oauth.remote_app( @@ -69,13 +75,13 @@ def init_oauth(app): def login_and_redirect_user(user) -> Response: - "Log in the user and then redirect them" + """Log in the user and then redirect them""" login_user(user) return redirect(url_for("general_bp.home")) def create_user(username) -> User: - "Create a temporary user if it is needed" + """Create a temporary user if it is needed""" user = User() user.configure(username, False, 1, associations=["zeus"]) db.session.add(user) diff --git a/app/config.example.py b/app/config.example.py index 9606e09..1b016a8 100644 --- a/app/config.example.py +++ b/app/config.example.py @@ -1,4 +1,4 @@ -"An example for a Haldis config" +"""An example for a Haldis config""" # config @@ -15,3 +15,7 @@ class Configuration: SENTRY_DSN = None ZEUS_KEY = "tomtest" ZEUS_SECRET = "blargh" + + ENABLE_MICROSOFT_AUTH = False + MICROSOFT_AUTH_ID = "" + MICROSOFT_AUTH_SECRET = "" diff --git a/app/hlds/definitions.py b/app/hlds/definitions.py index 961d14b..7f6e0c1 100644 --- a/app/hlds/definitions.py +++ b/app/hlds/definitions.py @@ -1,7 +1,6 @@ # Import this class to load the standard HLDS definitions - import subprocess -from os import path +from pathlib import Path from typing import List from .models import Location @@ -12,10 +11,11 @@ __all__ = ["location_definitions", "location_definition_version"] # pylint: disable=invalid-name # TODO Use proper way to get resources, see https://stackoverflow.com/a/10935674 -DATA_DIR = path.join(path.dirname(__file__), "..", "..", "menus") +ROOT_DIR = Path(__file__).parent.parent.parent +DATA_DIR = ROOT_DIR / "menus" -location_definitions: List[Location] = parse_all_directory(DATA_DIR) +location_definitions: List[Location] = parse_all_directory(str(DATA_DIR)) location_definitions.sort(key=lambda l: l.name) -proc = subprocess.run(["git", "rev-parse", "HEAD"], stdout=subprocess.PIPE, check=True) +proc = subprocess.run(["git", "rev-parse", "HEAD"], stdout=subprocess.PIPE, cwd=str(ROOT_DIR), check=True) location_definition_version = proc.stdout.decode().strip() diff --git a/app/migrations/versions/89b2c980b663_.py b/app/migrations/versions/89b2c980b663_.py new file mode 100644 index 0000000..d859e69 --- /dev/null +++ b/app/migrations/versions/89b2c980b663_.py @@ -0,0 +1,26 @@ +"""empty message + +Revision ID: 89b2c980b663 +Revises: 9eac0f3d7b1e +Create Date: 2023-04-20 02:01:54.558602 + +""" + +# revision identifiers, used by Alembic. +revision = '89b2c980b663' +down_revision = '9eac0f3d7b1e' + +from alembic import op +import sqlalchemy as sa + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.add_column('user', sa.Column('microsoft_uuid', sa.VARCHAR(length=120), nullable=True)) + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('user', 'microsoft_uuid') + # ### end Alembic commands ### diff --git a/app/models/user.py b/app/models/user.py index 974fb81..5110cf7 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -5,12 +5,14 @@ from models import db class User(db.Model): - "Class used for configuring the User model in the database" + """Class used for configuring the User model in the database""" id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) admin = db.Column(db.Boolean) bias = db.Column(db.Integer) - # Assocation logic + # Microsoft OAUTH info + microsoft_uuid = db.Column(db.String(120), unique=True) + # Association logic associations = db.Column(db.String(255), nullable=False, server_default="") # Relations @@ -25,13 +27,14 @@ class User(db.Model): def association_list(self) -> List[str]: return self.associations.split(",") - def configure(self, username: str, admin: bool, bias: int, associations: Optional[List[str]] = None) -> None: + def configure(self, username: str, admin: bool, bias: int, *, microsoft_uuid: str = None, associations: Optional[List[str]] = None) -> None: """Configure the User""" if associations is None: associations = [] self.username = username self.admin = admin self.bias = bias + self.microsoft_uuid = microsoft_uuid self.associations = ",".join(associations) # pylint: disable=C0111, R0201 diff --git a/app/templates/layout.html b/app/templates/layout.html index 72b85b7..ab5d5c2 100644 --- a/app/templates/layout.html +++ b/app/templates/layout.html @@ -63,7 +63,8 @@