Proof of Concept

Identity supports B2C

Pivot to standalone Identity Web

Author: rayluo

app.py

@@ -1,8 +1,9 @@
-import uuid
 import requests
 from flask import Flask, render_template, session, request, redirect, url_for
 from flask_session import Session  # https://pythonhosted.org/Flask-Session
-import msal
+from werkzeug.exceptions import Unauthorized, Forbidden
+from identity import __version__
+from identity.web import Web, LifespanValidator
 import app_config
 
 
@@ -17,43 +18,43 @@
 from werkzeug.middleware.proxy_fix import ProxyFix
 app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)
 
+web = Web(
+    session=session,
+    authority=app.config.get("AUTHORITY"),
+    client_id=app.config["CLIENT_ID"],
+    client_credential=app.config["CLIENT_SECRET"],
+    redirect_uri="http://localhost:5000" + app.config["REDIRECT_PATH"],  # It must match your redirect_uri
+    validators=[LifespanValidator(seconds=3600, on_error=Unauthorized("Login expired"))],
+    )
+
 @app.route("/")
 def index():
-    if not session.get("user"):
+    if not web.get_user():
         return redirect(url_for("login"))
-    return render_template('index.html', user=session["user"], version=msal.__version__)
+    return render_template('index.html', user=web.get_user(), version=__version__)
 
 @app.route("/login")
 def login():
-    # Technically we could use empty list [] as scopes to do just sign in,
-    # here we choose to also collect end user consent upfront
-    session["flow"] = _build_auth_code_flow(scopes=app_config.SCOPE)
-    return render_template("login.html", auth_url=session["flow"]["auth_uri"], version=msal.__version__)
+    return render_template("login.html", version=__version__, **web.start_auth(scopes=app_config.SCOPE))
+
+@app.errorhandler(Unauthorized)
+def handler(error):
+    return redirect(url_for("login"))
 
 @app.route(app_config.REDIRECT_PATH)  # Its absolute URL must match your app's redirect_uri set in AAD
-def authorized():
-    try:
-        cache = _load_cache()
-        result = _build_msal_app(cache=cache).acquire_token_by_auth_code_flow(
-            session.get("flow", {}), request.args)
-        if "error" in result:
-            return render_template("auth_error.html", result=result)
-        session["user"] = result.get("id_token_claims")
-        _save_cache(cache)
-    except ValueError:  # Usually caused by CSRF
-        pass  # Simply ignore them
+def auth_response():
+    result = web.complete_auth(request.args)
+    if "error" in result:
+        return render_template("auth_error.html", result=result)
     return redirect(url_for("index"))
 
 @app.route("/logout")
 def logout():
-    session.clear()  # Wipe out user and its token cache from session
-    return redirect(  # Also logout from your tenant's web session
-        app_config.AUTHORITY + "/oauth2/v2.0/logout" +
-        "?post_logout_redirect_uri=" + url_for("index", _external=True))
+    return redirect(web.sign_out(url_for("index", _external=True)))
 
 @app.route("/graphcall")
 def graphcall():
-    token = _get_token_from_cache(app_config.SCOPE)
+    token = web.get_token(app_config.SCOPE)
     if not token:
         return redirect(url_for("login"))
     graph_data = requests.get(  # Use token to call downstream service
@@ -62,38 +63,6 @@ def graphcall():
         ).json()
     return render_template('display.html', result=graph_data)
 
-
-def _load_cache():
-    cache = msal.SerializableTokenCache()
-    if session.get("token_cache"):
-        cache.deserialize(session["token_cache"])
-    return cache
-
-def _save_cache(cache):
-    if cache.has_state_changed:
-        session["token_cache"] = cache.serialize()
-
-def _build_msal_app(cache=None, authority=None):
-    return msal.ConfidentialClientApplication(
-        app_config.CLIENT_ID, authority=authority or app_config.AUTHORITY,
-        client_credential=app_config.CLIENT_SECRET, token_cache=cache)
-
-def _build_auth_code_flow(authority=None, scopes=None):
-    return _build_msal_app(authority=authority).initiate_auth_code_flow(
-        scopes or [],
-        redirect_uri=url_for("authorized", _external=True))
-
-def _get_token_from_cache(scope=None):
-    cache = _load_cache()  # This web app maintains one cache per session
-    cca = _build_msal_app(cache=cache)
-    accounts = cca.get_accounts()
-    if accounts:  # So all account(s) belong to the current signed-in user
-        result = cca.acquire_token_silent(scope, account=accounts[0])
-        _save_cache(cache)
-        return result
-
-app.jinja_env.globals.update(_build_auth_code_flow=_build_auth_code_flow)  # Used in template
-
 if __name__ == "__main__":
     app.run()
 

requirements.txt

@@ -12,7 +12,7 @@ werkzeug>=2
 
 flask-session>=0.3.2,<0.5
 requests>=2,<3
-msal>=1.7,<2
+identity>=0.1,<0.2
 
 # cachelib==0.1  # Only need this if you are running Python 2
 # Note: This sample does NOT directly depend on cachelib.

templates/auth_error.html

@@ -3,9 +3,9 @@
 <head>
     <meta charset="UTF-8">
 
-    {% if config.get("B2C_RESET_PASSWORD_AUTHORITY") and "AADB2C90118" in result.get("error_description") %}
+    {% if config.get("B2C_RESET_PASSWORD_AUTHORITY") and "AADB2C90118" in result.get("error_description") %} <!-- This will be reached when user forgot their password -->
       <!-- See also https://docs.microsoft.com/en-us/azure/active-directory-b2c/active-directory-b2c-reference-policies#linking-user-flows -->
-      <meta http-equiv="refresh" content='0;{{_build_auth_code_flow(authority=config["B2C_RESET_PASSWORD_AUTHORITY"])["auth_uri"]}}'>
+      <meta http-equiv="refresh" content='0;{{config.get("B2C_RESET_PASSWORD_AUTHORITY")}}?client_id={{config.get("CLIENT_ID")}}'>
     {% endif %}
 </head>
 <body>

templates/index.html

@@ -12,12 +12,12 @@ <h2>Welcome {{ user.get("name") }}!</h2>
     {% endif %}
 
     {% if config.get("B2C_PROFILE_AUTHORITY") %}
-      <li><a href='{{_build_auth_code_flow(authority=config["B2C_PROFILE_AUTHORITY"])["auth_uri"]}}'>Edit Profile</a></li>
+      <li><a href='{{config.get("B2C_PROFILE_AUTHORITY")}}?client_id={{config.get("CLIENT_ID")}}'>Edit Profile</a></li>
     {% endif %}
 
     <li><a href="/logout">Logout</a></li>
     <hr>
-    <footer style="text-align: right">Powered by MSAL Python {{ version }}</footer>
+    <footer style="text-align: right">Powered by Identity Web {{ version }}</footer>
 </body>
 </html>
 

templates/login.html

@@ -6,14 +6,24 @@
 <body>
     <h1>Microsoft Identity Python Web App</h1>
 
-    <li><a href='{{ auth_url }}'>Sign In</a></li>
+    {% if user_code %}
+    <ol>
+    <li>To sign in, type <b>{{ user_code }}</b> into
+      <a href='{{ auth_uri }}' target=_blank>{{ auth_uri }}</a>
+      to authenticate.
+    </li>
+    <li>And then <a href="{{ url_for('auth_response') }}">proceed</a>.</li>
+    </ol>
+    {% else %}
+    <li><a href='{{ auth_uri }}'>Sign In</a></li>
+    {% endif %}
 
     {% if config.get("B2C_RESET_PASSWORD_AUTHORITY") %}
-    <li><a href='{{_build_auth_code_flow(authority=config["B2C_RESET_PASSWORD_AUTHORITY"])["auth_uri"]}}'>Reset Password</a></li>
+    <li><a href="{{config.get('B2C_RESET_PASSWORD_AUTHORITY')}}?client_id={{config.get('CLIENT_ID')}}">Reset Password</a></li>
     {% endif %}
 
     <hr>
-    <footer style="text-align: right">Powered by MSAL Python {{ version }}</footer>
+    <footer style="text-align: right">Powered by Identity Web {{ version }}</footer>
 </body>
 </html>