Python 和带有 Flask 的 AzureAD - ms-identity-python-webapp 示例仅间歇性登录
Python and AzureAD with Flask - ms-identity-python-webapp example only logging in intermittently
我正在尝试从此处设置 Azure AD 身份验证示例:
https://github.com/Azure-Samples/ms-identity-python-webapp
我在 AzureAD 上设置了所有配置和权限,但我在登录时遇到间歇性问题。
我已将问题追查到会话密钥,要么没有在 session.get('user'):
内正确写入或读取
def index():
if not session.get("user"):
return redirect(url_for("login"))
return render_template('index.html', user=session["user"], version=msal.__version__)
此外,
if request.args.get('state') != session.get("state"):
return redirect(url_for("index")) # No-OP. Goes back to Index page
永远不会 returns 正确,但如果我将其注释掉,有时一切都会正常,它将登录,并允许访问图形应用程序。
我已经在本地设置了 flask_session 目录,它会为每次登录创建缓存(大约 8kb)。
清除 flask_session 文件夹并重新启动应用程序似乎有所帮助,但并不可靠。
如有任何帮助,我们将不胜感激!
为方便起见复制在这里的完整示例:
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
import app_config
app = Flask(__name__)
app.config.from_object(app_config)
Session(app)
# This section is needed for url_for("foo", _external=True) to automatically
# generate http scheme when this sample is running on localhost,
# and to generate https scheme when it is deployed behind reversed proxy.
# See also https://flask.palletsprojects.com/en/1.0.x/deploying/wsgi-standalone/#proxy-setups
from werkzeug.middleware.proxy_fix import ProxyFix
app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)
@app.route("/")
def index():
if not session.get("user"):
return redirect(url_for("login"))
return render_template('index.html', user=session["user"], version=msal.__version__)
@app.route("/login")
def login():
session["state"] = str(uuid.uuid4())
# Technically we could use empty list [] as scopes to do just sign in,
# here we choose to also collect end user consent upfront
auth_url = _build_auth_url(scopes=app_config.SCOPE, state=session["state"])
return render_template("login.html", auth_url=auth_url, version=msal.__version__)
@app.route(app_config.REDIRECT_PATH) # Its absolute URL must match your app's redirect_uri set in AAD
def authorized():
if request.args.get('state') != session.get("state"):
return redirect(url_for("index")) # No-OP. Goes back to Index page
if "error" in request.args: # Authentication/Authorization failure
return render_template("auth_error.html", result=request.args)
if request.args.get('code'):
cache = _load_cache()
result = _build_msal_app(cache=cache).acquire_token_by_authorization_code(
request.args['code'],
scopes=app_config.SCOPE, # Misspelled scope would cause an HTTP 400 error here
redirect_uri=url_for("authorized", _external=True))
if "error" in result:
return render_template("auth_error.html", result=result)
session["user"] = result.get("id_token_claims")
_save_cache(cache)
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))
@app.route("/graphcall")
def graphcall():
token = _get_token_from_cache(app_config.SCOPE)
if not token:
return redirect(url_for("login"))
graph_data = requests.get( # Use token to call downstream service
app_config.ENDPOINT,
headers={'Authorization': 'Bearer ' + token['access_token']},
).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_url(authority=None, scopes=None, state=None):
return _build_msal_app(authority=authority).get_authorization_request_url(
scopes or [],
state=state or str(uuid.uuid4()),
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_url=_build_auth_url) # Used in template
if __name__ == "__main__":
app.run()
我找到了解决方案,或者更准确地说是我的失败:
这个回购:
https://github.com/rayluo/flask-session
与此不同:
https://github.com/fengsp/flask-session
包含的 requirements.txt 从 rayluo(谢谢!)加载 newer/maintained flask-session,而不是从 pip 加载。
所以对于其他发现此问题的人,请确保您使用正确的版本!
我是 Azure AD 身份验证示例的维护者之一:https://github.com/Azure-Samples/ms-identity-python-webapp
我是我的变通办法回购协议的作者 https://github.com/rayluo/flask-session
在撰写本文时,间接上游模块 cachelib 已在 2 天前发布了一个错误修复,大约在同一时间您遇到了这个问题。因此,对于这里的任何未来读者:您不必显式切换即可使用上述解决方法。该 Azure AD Web 应用程序示例将正确声明它需要的正确依赖项。
我正在尝试从此处设置 Azure AD 身份验证示例: https://github.com/Azure-Samples/ms-identity-python-webapp
我在 AzureAD 上设置了所有配置和权限,但我在登录时遇到间歇性问题。
我已将问题追查到会话密钥,要么没有在 session.get('user'):
内正确写入或读取def index():
if not session.get("user"):
return redirect(url_for("login"))
return render_template('index.html', user=session["user"], version=msal.__version__)
此外,
if request.args.get('state') != session.get("state"):
return redirect(url_for("index")) # No-OP. Goes back to Index page
永远不会 returns 正确,但如果我将其注释掉,有时一切都会正常,它将登录,并允许访问图形应用程序。
我已经在本地设置了 flask_session 目录,它会为每次登录创建缓存(大约 8kb)。
清除 flask_session 文件夹并重新启动应用程序似乎有所帮助,但并不可靠。
如有任何帮助,我们将不胜感激!
为方便起见复制在这里的完整示例:
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
import app_config
app = Flask(__name__)
app.config.from_object(app_config)
Session(app)
# This section is needed for url_for("foo", _external=True) to automatically
# generate http scheme when this sample is running on localhost,
# and to generate https scheme when it is deployed behind reversed proxy.
# See also https://flask.palletsprojects.com/en/1.0.x/deploying/wsgi-standalone/#proxy-setups
from werkzeug.middleware.proxy_fix import ProxyFix
app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)
@app.route("/")
def index():
if not session.get("user"):
return redirect(url_for("login"))
return render_template('index.html', user=session["user"], version=msal.__version__)
@app.route("/login")
def login():
session["state"] = str(uuid.uuid4())
# Technically we could use empty list [] as scopes to do just sign in,
# here we choose to also collect end user consent upfront
auth_url = _build_auth_url(scopes=app_config.SCOPE, state=session["state"])
return render_template("login.html", auth_url=auth_url, version=msal.__version__)
@app.route(app_config.REDIRECT_PATH) # Its absolute URL must match your app's redirect_uri set in AAD
def authorized():
if request.args.get('state') != session.get("state"):
return redirect(url_for("index")) # No-OP. Goes back to Index page
if "error" in request.args: # Authentication/Authorization failure
return render_template("auth_error.html", result=request.args)
if request.args.get('code'):
cache = _load_cache()
result = _build_msal_app(cache=cache).acquire_token_by_authorization_code(
request.args['code'],
scopes=app_config.SCOPE, # Misspelled scope would cause an HTTP 400 error here
redirect_uri=url_for("authorized", _external=True))
if "error" in result:
return render_template("auth_error.html", result=result)
session["user"] = result.get("id_token_claims")
_save_cache(cache)
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))
@app.route("/graphcall")
def graphcall():
token = _get_token_from_cache(app_config.SCOPE)
if not token:
return redirect(url_for("login"))
graph_data = requests.get( # Use token to call downstream service
app_config.ENDPOINT,
headers={'Authorization': 'Bearer ' + token['access_token']},
).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_url(authority=None, scopes=None, state=None):
return _build_msal_app(authority=authority).get_authorization_request_url(
scopes or [],
state=state or str(uuid.uuid4()),
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_url=_build_auth_url) # Used in template
if __name__ == "__main__":
app.run()
我找到了解决方案,或者更准确地说是我的失败:
这个回购: https://github.com/rayluo/flask-session
与此不同: https://github.com/fengsp/flask-session
包含的 requirements.txt 从 rayluo(谢谢!)加载 newer/maintained flask-session,而不是从 pip 加载。
所以对于其他发现此问题的人,请确保您使用正确的版本!
我是 Azure AD 身份验证示例的维护者之一:https://github.com/Azure-Samples/ms-identity-python-webapp 我是我的变通办法回购协议的作者 https://github.com/rayluo/flask-session
在撰写本文时,间接上游模块 cachelib 已在 2 天前发布了一个错误修复,大约在同一时间您遇到了这个问题。因此,对于这里的任何未来读者:您不必显式切换即可使用上述解决方法。该 Azure AD Web 应用程序示例将正确声明它需要的正确依赖项。