MSAL stands for Microsoft Authetication Library enables developers to acquire security tokens from the Microsoft identity platform to autheticate users and access secured web API (documentation). Among various authtication methods, I’ll summarize how to autheticate through an interactive login in the web app built with a Flask and Python.

1. Environment variables

Let’s create a .env file first for environment variables.

  • CLIENT_ID

    • It’s the application (client) identifier you receive when you regsiter your app in the Azure portal.
    • Thiks of it as the unique ID that tells Azure AD who is making the request through your code.

  • CLIENT_SECRET

    • It’s a secret key that proves the request really came from your application, so never hard-code it in your source and always keep it secure by using environment variables.

  • TENANT_ID

    • It’s the unique identifier of your Azure AD tenant (directory).
    • Think of it as the “AD space” that each company or organization owns.
    • This value tells Azure AD which directory (which company) you want to autheticate against.

  • AUTHROITY

    • This is the base URL of the authetication server. It usually looks like:

      https://login.microsoftonline.com/<TENANT_ID>

    • In other words, it tells your code where to send the token request.

  • REDIRECT_URI

    • This is the callback URL in your app where Azure AD will send the autorization code after the user signs in and gives consent.
    • For example, http://localhost:5000/getAToken.
    • It must match exactly what you’ve registerd in Azure AD, or Azure AD won’t redirect back to your app.

  • SCOPE

    • This is a list of permission strings that your app is requesting.
    • For example, ["User.Read"] means “Give me permission to read my profile from Microsoft Graph.”
    • Always request only the minimum permissions your app actually needs for better eecurity.
    • You can get a bunch of permissions you already have through https://graph.microsoft.com/.default.
#.env
CLIENT_ID=<your key>
CLIENT_SECRET=<your key>
TENANT_ID=<your key>
AUTHORITY=https://login.microsoftonline.com/<TENANT_ID>
REDIRECT_URI=http://localhost:5000/getAToken
SCOPE=["User.Read"]

We can load this via load_dotenv.

from dotenv import load_dotenv

load_dotenv()
CLIENT_ID = os.getenv("CLIENT_ID")
# ...

2. Flask app

Let’s create a Flask app. Flask(__name__) returns a new web aplication instance and sets its rooth path to the directory of the current module (i.e. the folder where ytour scripts lives). Internally, Flask uses this root_path to locate resources such as templates and static files. Next, we assign a random secret key to the app so that session data (for example, login status) is securely signed and protected.

from flask import Flask

app = Flask(__name__)
app.secret_key = os.urandom(24)

3. MSAL token cache

Once we login, we can keep using AT/RT(Access Token/Refresh Token) if we store a token cache in local as a file.

from msal import SerializableTokenCache

cache = SerializableTokenCache

# Load a cahce file, `msal_cache.bin` if you already have.
# Or save it as a file with `cache.serialize()` before terminating the program

4. MSAL ConfidentialClientApplication

Now we need a client object to get and refresh tokens.

from msal import ConfidentialClientApplication

client = ColnfidentialClientApplication(
    CLIENT_ID,
    authority=AUTHORITY,
    client_credential=CLIENT_SECRET,
    token_cahce=cache
)

5. Helper function to get an access token

This function checks whether the current Access Token (AT) is still valid. If it is, returns the existing AT. If it has expired, it uses the Refresh Token (RT) to request and return a new AT.

def get_access_token():

    # 1) Check a client info from the cache.
    accounts = client.get_accoutns()

    # 2) If we have an account, check if the token is valid and return.
    result = client.acquire_token_silent(
        scopes=SCOPE,
        account=accounts[0]
    )

    return result and result.get("access_token")

The last line return result and result.get("access_token") is same as belows:

if not result:
    return None
return result.get("access_token")

To refresh and save a token:

def refresh_and_save_token(token_file):
    token = get_access_token()
    if not token:
        app.logger.warning("Failed to refresh token.")
        return
    
    try:
        with open(token_file, "w") as f:
            f.write(token)
        app.logger.info(f"New token is stored in {token_file}.")
    except Exception as e:
        app.logger.error(f"Token file write error: {e}")

6. Main routes(endpoints)

  1. /login
  • Redirect the user to the Azure AD (Active Directory) login page.
  • Store a random state parameter for CSRF protection (Cross-Site Request Forgery).

    • When the app redirects to /login, generate a random string (the state) and store it in the session:

      session["state"] = generated_state

    • Include this same state in the Azure AD login URL (authorization request) as:

      &state=generated_state
  1. /getAToken
  • After login and consent, Azure AD redirects to this endpoint with code and state.
  • Call acquire_token_authorization_code(code, scopes, redirects_uri) to exchange the code for an Access Token (AT) and Refresh Token (RT), which MSAL then cahces.
  • Save user info (ID token claims) in session["user"], then redirect to /.
  • Make sure your Azure app’s Authentication settigns include the redirect URI (e.g. http://localhost:5000/getAToken)
    • On Azure Portal: Azure AD -> App registraion -> My App -> Authetication
  1. / (index)
  • If session[“user”] does not exist, redirect to /login.
  • Otherwise, call get_access_token() to retrieve a valid AT and return it as JSON.

With this flow, the user logs in once and can automatically refresh the AT each time they revisit http://localhost:5000.

7. Background scheduler

With the scheduler, we don’t need to manually refresh by visiting http://localhost:5000/. We st art the scheduler as soon as the Flask app receives its first request. Then, at reuglar interval, it simply calls get_access_token().

from apscheduler.schedulers.backgroun import BackgroundScheduler

scheduler = BackgroundScheduler(timezone="Asia/Seoul")

# For example, every 24 hours.
scheduler.add_job(func=get_access_token, trigger='interval', hours=24, id='refresh_access_token', replace_existing=True)

if os.environ.get("WERKZEUG_RUN_MAIN") == "true":
    scheduler.start()

8. Start app

if __name__=="__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

9. Full code

import os
import uuid
import atexit
import logging
from flask import Flask, session, redirect, request, url_for, jsonify
from dotenv import load_dotenv
from msal import ConfidentialClientApplication, SerializableTokenCache
from apscheduler.schedulers.background import BackgroundScheduler

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Environment variable validation
CLIENT_ID = os.getenv("EMAIL_CLIENT_ID")
CLIENT_SECRET = os.getenv("EMAIL_CLIENT_SECRET")
AUTHORITY = os.getenv("EMAIL_AUTHORITY")
REDIRECT_URI = os.getenv("EMAIL_REDIRECT_URI")
SCOPE = [os.getenv("EMAIL_SCOPE")]

# Validate required environment variables
missing_vars = []
for var_name, var_value in [
    ("EMAIL_CLIENT_ID", CLIENT_ID),
    ("EMAIL_CLIENT_SECRET", CLIENT_SECRET),
    ("EMAIL_AUTHORITY", AUTHORITY),
    ("EMAIL_REDIRECT_URI", REDIRECT_URI),
    ("EMAIL_SCOPE", SCOPE[0] if SCOPE else None)
]:
    if not var_value:
        missing_vars.append(var_name)

if missing_vars:
    logger.error(f"Missing required environment variables: {', '.join(missing_vars)}")
    logger.error("Please create a .env file with the required variables.")
    raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}")

# Initialize Flask app and secret key
app = Flask(__name__)
app.secret_key = os.urandom(24)

# Initialize MSAL token cache (file-based)
cache = SerializableTokenCache()
CACHE_FILE = os.path.join(os.path.dirname(__file__), "msal_cache.bin")
if os.path.exists(CACHE_FILE):
    try:
        with open(CACHE_FILE, "r") as f:
            cache.deserialize(f.read())
        logger.info("Token cache loaded successfully")
    except Exception as e:
        logger.warning(f"Failed to load token cache: {e}")

def save_cache():
    try:
        with open(CACHE_FILE, "w") as f:
            f.write(cache.serialize())
        logger.info("Token cache saved successfully")
    except Exception as e:
        logger.error(f"Failed to save token cache: {e}")

atexit.register(save_cache)

# Create MSAL client
try:
    client = ConfidentialClientApplication(
        CLIENT_ID,
        authority=AUTHORITY,
        client_credential=CLIENT_SECRET,
        token_cache=cache
    )
    logger.info("MSAL client initialized successfully")
except Exception as e:
    logger.error(f"Failed to initialize MSAL client: {e}")
    raise

# Function to get access token
def get_access_token():
    try:
        accounts = client.get_accounts()
        if not accounts:
            logger.warning("No accounts found in token cache")
            return None

        result = client.acquire_token_silent(scopes=SCOPE, account=accounts[0])
        if not result or "access_token" not in result:
            logger.warning("Failed to acquire token silently")
            return None

        logger.info("Access token acquired successfully")
        return result["access_token"]
    except Exception as e:
        logger.error(f"Error getting access token: {e}")
        return None

def refresh_and_save_token(token_file):
    token = get_access_token()
    if not token:
        logger.warning("Failed to refresh token.")
        return

    try:
        with open(token_file, "w") as f:
            f.write(token)
        logger.info(f"New token is stored in {token_file}.")
    except Exception as e:
        logger.error(f"Token file write error: {e}")

# Login endpoint and callback endpoint
@app.route("/login")
def login():
    try:
        session["state"] = str(uuid.uuid4())
        auth_url = client.get_authorization_request_url(
            scopes=SCOPE,
            state=session["state"],
            redirect_uri=REDIRECT_URI
        )
        logger.info("Redirecting to Azure AD login")
        return redirect(auth_url)
    except Exception as e:
        logger.error(f"Error in login endpoint: {e}")
        return jsonify({"error": "Login failed"}), 500

@app.route("/getAToken")
def authorized():
    try:
        if request.args.get("state") != session.get("state"):
            logger.error("State mismatch in authorization callback")
            return jsonify({"error": "State mismatch"}), 400

        code = request.args.get("code")
        if not code:
            logger.error("No authorization code received")
            return jsonify({"error": "No authorization code"}), 400

        result = client.acquire_token_by_authorization_code(
            code, scopes=SCOPE, redirect_uri=REDIRECT_URI
        )

        if "access_token" in result:
            session["user"] = result.get("id_token_claims")
            logger.info("Token acquired successfully")
            return redirect(url_for("index"))
        else:
            error_msg = result.get('error_description', 'Unknown error')
            logger.error(f"Failed to acquire token: {error_msg}")
            return jsonify({"error": f"Failed to acquire token: {error_msg}"}), 400
    except Exception as e:
        logger.error(f"Error in authorization callback: {e}")
        return jsonify({"error": "Authorization failed"}), 500

@app.route("/")
def index():
    try:
        if not session.get("user"):
            logger.info("User not authenticated, redirecting to login")
            return redirect(url_for("login"))

        token = get_access_token()
        if not token:
            logger.info("No valid token, redirecting to login")
            return redirect(url_for("login"))

        user_info = {
            "name": session["user"].get("name", "Unknown"),
            "email": session["user"].get("preferred_username", "Unknown"),
            "access_token": token[:50] + "..." if len(token) > 50 else token
        }
        logger.info(f"User {user_info['name']} accessed the application")
        return jsonify(user_info)
    except Exception as e:
        logger.error(f"Error in index endpoint: {e}")
        return jsonify({"error": "Internal server error"}), 500

@app.route("/health")
def health():
    """Health check endpoint"""
    return jsonify({"status": "healthy", "message": "Azure Auth service is running"})

@app.route("/token")
def get_token():
    """Endpoint to get current access token"""
    try:
        token = get_access_token()
        if token:
            return jsonify({"access_token": token})
        else:
            return jsonify({"error": "No valid token available"}), 401
    except Exception as e:
        logger.error(f"Error getting token: {e}")
        return jsonify({"error": "Failed to get token"}), 500

# APScheduler setup and start
scheduler = BackgroundScheduler(timezone="Asia/Seoul")
scheduler.add_job(
    func=refresh_and_save_token,
    trigger='cron',
    hour=7,
    minute=0,
    kwargs={
        "token_file": os.path.join(os.path.dirname(__file__), "access_token.txt"),
    },
    id='refresh_access_token',
    replace_existing=True
)

# Run the app
if __name__ == "__main__":
    try:
        if os.environ.get("WERKZEUG_RUN_MAIN") == "true":
            scheduler.start()
            logger.info("Scheduler started successfully")

        port = int(os.getenv("PORT", 5000))
        host = os.getenv("HOST", "0.0.0.0")

        logger.info(f"Starting Flask app on {host}:{port}")
        app.run(host=host, port=port, debug=True)
    except Exception as e:
        logger.error(f"Failed to start application: {e}")
        raise

10. Remote server installation

If you want to depoy this app on a remote server, you need a public domain name or IP address with HTTPS. Azure AD requires HTTPS for production apps; HTTP is only allowed for localhost during development.

  1. Redirect URI in Azure Portal

  • Go to Azure AD -> App registraion -> [Your App] -> Authentication, and add:

    https://<your-domain-or-ip>:<your-port-if-you-use>/getAToken

  • Make sure this matches exactly the address your app will use.

  1. Client access URL

  • Open your browser and go to:

    https://<your-domain-or-ip>[:<your-port-of-you-use>]/

  • The root(/) path checks if you’re logged in. If not, it redirects to:

    https://<your-domain-or-ip>[:<your-port-of-you-use>]/login

  • After you sign in and consent, Azure AD redirects you to:

    https://<your-domain-or-ip>[:<your-port-of-you-use>]/getAToken

  • This callback completes the token exchanges and brings you back to the app.

However, Azure AD doesn’t accept self-signed SSL certificates or the default domain of an AWS EC2 instance. As an alternative for remote deployment, you can use SSH port forwarding-just keep your terminal open while you authenticate from the client.

ssh -i key.pem -L 5000:localhost:5000 ec2-user@ec2-ip

11. Misc.

When you run the app in debug mode, Flask actually starts two processes under the hood because of its automatic reloader (it watches for code changes). This means scheduler.start() can run twice. On the second call, APScheduler will raise a SchedulerAlreadyRuningError.