Microsoft 身份验证 - 移植到 FastAPI 的 Python Flask msal 示例应用程序

Question

我没有做太多的网络工作，但我最近开始使用 FastAPI 并正在构建一个带有 jinja2 模板的 MVC 应用程序，该应用程序使用 PowerBI 嵌入式容量来服务于应用程序拥有的数据排列中的多个嵌入式分析。所有这些都非常有效。但是，我想添加更多模块，我想使用 msal 包通过将用户路由到 Microsoft 登录页面来进行用户身份验证，让他们登录我在 Azure 中设置的多租户应用程序服务，然后通过重定向 URI 重定向回我的页面，获取令牌并继续进行授权。 Microsoft 保存了一个很好的例子，我们的 here 用于在 Flask 中执行此操作。但是，我很适合将示例移植到 FastAPI。

我可以让用户进入登录屏幕并登录，但我无法在我的回调 URI 中捕获令牌 - 它是适当的路由，但我无法从响应中捕获令牌。

有没有人（或者任何人都可以）使用这个超级简单的 Flask 示例并将其移植到 FastAPI？我在网上找到的 FAPI 的所有内容都是 API 的后端令牌承载标头 - 不适用于 MVC 应用程序。

这是我当前的代码。凌乱，因为我内置了“测试”。

import msal
import requests
from fastapi import APIRouter, Request, Response
from fastapi.responses import RedirectResponse
from starlette.templating import Jinja2Templates

from config import get_settings

settings = get_settings()
router = APIRouter()
templates = Jinja2Templates('templates')


# Works
@router.get('/login', include_in_schema=False)
async def login(request: Request):
    request.session['flow'] = _build_auth_code_flow(scopes=settings.AUTH_SCOPE)
    login_url = request.session['flow']['auth_uri']
    return templates.TemplateResponse('error.html', {'request': request, 'message': login_url})


# DOES NOT WORK - Pretty sure error is in here --------------------
@router.get('/getAToken', response_class=Response, include_in_schema=False)
async def authorize(request: Request):
    try:
        cache = _load_cache(request)
        result = _build_msal_app(cache=cache).acquire_token_by_auth_code_flow(
            request.session.get('flow'), request.session
        )
        if 'error' in result:
            return templates.TemplateResponse('error.html', {'request': request, 'message': result})
        request.session['user'] = result.get('id_token_claims')
        _save_cache(cache)
    except Exception as error:
        return templates.TemplateResponse('error.html', {'request': request, 'message': f'{error}: {str(request.query_params)}'})
    return templates.TemplateResponse('error.html', {'request': request, 'message': result})
# -----------------------------------------------------

    
def _load_cache(request: Request):
    cache = msal.SerializableTokenCache()
    if request.session.get("token_cache"):
        cache.deserialize(request.session["token_cache"])
    return cache


def _save_cache(request: Request, cache):
    if cache.has_state_changed:
        request.session["token_cache"] = cache.serialize()


def _build_msal_app(cache=None, authority=None):
    return msal.ConfidentialClientApplication(
        settings.CLIENT_ID,
        authority=authority or settings.AUTH_AUTHORITY,
        client_credential=settings.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=settings.AUTH_REDIRECT)


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

非常感谢任何帮助。很高兴回答任何问题。谢谢。

Answer 1

这是因为 FastAPI 会话变量在客户端存储为 cookie，其数据限制为 4096 字节。从重定向 url 存储的数据会推动 cookie 大小超过此限制，并导致数据不被存储。 Starlette-session 是另一种 SessionMiddleware，它在服务器端存储变量，消除了 cookie 限制。下面是一个基本的（但比较杂乱的）实现：

from fastapi import FastAPI
from fastapi.templating import Jinja2Templates

from starlette.requests import Request
from starlette.responses import RedirectResponse

from starlette_session import SessionMiddleware
from starlette_session.backends import BackendType

from redis import Redis

import uvicorn
import functools
import msal


app_client_id = "sample_msal_client_id"
app_client_secret = "sample_msal_client_secret"
tenant_id = "sample_msal_tenant_id"

app = FastAPI()


redis_client = Redis(host="localhost", port=6379)
app.add_middleware(
    SessionMiddleware,
    secret_key="SECURE_SECRET_KEY",
    cookie_name="auth_cookie",
    backend_type=BackendType.redis,
    backend_client=redis_client,
)

templates = Jinja2Templates(directory="templates")

default_scope = ["https://graph.microsoft.com/.default"]
token_cache_key = "token_cache"

# Private Functions - Start
def _load_cache(session):
    cache = msal.SerializableTokenCache()
    if session.get(token_cache_key):
        cache.deserialize(session[token_cache_key])
    return cache

def _save_cache(cache,session):
    if cache.has_state_changed:
        session[token_cache_key] = cache.serialize()

def _build_msal_app(cache=None):
    return msal.ConfidentialClientApplication(
        app_client_id, 
        client_credential=app_client_secret,
        authority=f"https://login.microsoftonline.com/{tenant_id}",
        token_cache=cache
    )

def _build_auth_code_flow(request):
    return _build_msal_app().initiate_auth_code_flow(
        default_scope, #Scopes
        redirect_uri=request.url_for("callback") #Redirect URI
    )

def _get_token_from_cache(session):
    cache = _load_cache(session)  # 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(default_scope, account=accounts[0])
        _save_cache(cache,session)
        return result
# Private Functions - End


# Custom Decorators - Start
def authenticated_endpoint(func):
    @functools.wraps(func)
    def is_authenticated(*args,**kwargs):
        try:
            request = kwargs["request"]
            token = _get_token_from_cache(request.session)
            if not token:
                return RedirectResponse(request.url_for("login"))
            return func(*args,**kwargs)
        except:
            return RedirectResponse(request.url_for("login"))

    return is_authenticated
# Custom Decorators - End


# Endpoints - Start
@app.get("/")
@authenticated_endpoint
def index(request:Request):
    return {
        "result": "good"
    }

@app.get("/login")
def login(request:Request):
    return templates.TemplateResponse("login.html",{
        "version": msal.__version__,
        'request': request,
        "config": {
            "B2C_RESET_PASSWORD_AUTHORITY": False
        }
    })

@app.get("/oauth/redirect")
def get_redirect_url(request:Request):
    request.session["flow"] = _build_auth_code_flow(request)
    return RedirectResponse(request.session["flow"]["auth_uri"])

@app.get("/callback")
async def callback(request:Request):
    cache = _load_cache(request.session)
    result = _build_msal_app(cache=cache).acquire_token_by_auth_code_flow(request.session.get("flow", {}), dict(request.query_params))
    if "error" in result:
        return templates.TemplateResponse("auth_error.html",{
            "result": result,
            'request': request
        })
    request.session["user"] = result.get("id_token_claims")
    request.session[token_cache_key] = cache.serialize()
    return RedirectResponse(request.url_for("index"))
# Endpoints - End

if __name__ == "__main__":
    uvicorn.run("main:app",host='0.0.0.0', port=4557,reload=True)`