基于Streamlit的Azure Web App集成Microsoft身份验证及Blob文件夹权限控制问题求助

我们团队正在开发一个部署在Azure Web App上的Streamlit应用，目前卡在了两个核心需求的实现上，想求助社区的大佬们帮忙看看：

• Microsoft身份验证集成：希望只允许特定域组的用户无缝登录应用，不需要跳转到外部的Microsoft认证链接；
• Blob存储权限控制：根据登录用户的邮箱，过滤其能访问的Blob存储文件夹，实现用户级的内容隔离。

我们已经给应用配置了获取用户邮箱和个人资料的委托权限，但试了好几种实现方式都没达到预期效果，下面是我们尝试过的代码和遇到的问题：

目前尝试的交互式认证实现 #

我们先用DefaultAzureCredential从Azure Key Vault中获取认证所需的密钥，然后用PublicClientApplication实现交互式Token获取，代码如下：

import os
import jwt
import datetime
import streamlit as st
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
from msal import PublicClientApplication

# 凭证与Key Vault连接
credential = DefaultAzureCredential()
KEY_VAULT_NAME = os.environ.get("KEY_VAULT_NAME")
KV_URI = f"https://{KEY_VAULT_NAME}.vault.azure.net"
client = SecretClient(vault_url=KV_URI, credential=credential)

# SSO配置 - 从Key Vault读取Azure AD参数
TENANT_ID = client.get_secret("tenantid").value
CLIENT_ID = client.get_secret("clientid").value
CLIENT_SECRET = client.get_secret("clientsecret").value
scope = ["https://graph.microsoft.com/User.Read"]

app = PublicClientApplication(
    client_id=CLIENT_ID, 
    authority=f"https://login.microsoftonline.com/{TENANT_ID}"
)

result = None 
if not result:
   result = app.acquire_token_interactive(scopes=["User.Read"])

if "access_token" in result:
    access_token = result["access_token"]
    if "id_token" in result:
        id_token = result["id_token"]
        user_info = jwt.decode(id_token, options={"verify_signature": False})
        st.write(f"Welcome, {user_info.get('name', 'User')}!")
        st.write(f"Your email: {user_info.get('preferred_username', 'Not available')}")
    else:
         st.write('No id_token in results')

email = user_info.get('mail', user_info.get('preferred_username', None))

if email:
    # is_valid_email、ALLOWED_EMAILS、df_access为自定义/预定义变量
    if not is_valid_email(email):
        st.error("Invalid email format. Please enter a valid email address.")
    elif email in ALLOWED_EMAILS:
        company_list = df_access['Companies'][df_access['Email']==email].unique()  # 获取用户对应权限的公司列表
        if len(company_list) >0:
            st.session_state.logged_in = True
            st.session_state.email = email
            st.session_state.login_time = datetime.now()
            st.success("Login successful!")
            st.rerun()
        else:
            st.error("No companies available for this email.")
    else:
        st.error("Invalid email. Please try again.")
        st.session_state.logged_in = False
else:
    st.error("Email not found in user information.")

这种方式的问题在于，交互式认证会弹出独立的登录窗口，既不符合我们“无缝登录”的需求，部署到Azure Web App后也无法正常运行。

之前尝试的st.query_params()验证方式 #

我们还试过通过st.query_params()获取并验证ID Token的方式，想实现无跳转的身份验证，但同样没有成功：

import requests
import jwt
from jwt import algorithms as RSAAlgorithm
import json
import streamlit as st

# 依赖TENANT_ID、CLIENT_ID等预定义变量
DISCOVERY_URL = f"https://login.microsoftonline.com/{TENANT_ID}/v2.0/.well-known/openid-configuration"

def get_public_keys():
    """从Azure AD获取公钥用于验证ID Token"""
    response = requests.get(DISCOVERY_URL)
    response.raise_for_status()
    jwks_uri = response.json()["jwks_uri"]
    jwks_response = requests.get(jwks_uri)
    jwks_response.raise_for_status()
    keys = jwks_response.json()["keys"]
    print(keys)
    return {key["kid"]: RSAAlgorithm.from_jwk(json.dumps(key)) for key in keys}

def validate_id_token(token, public_keys):
    """使用Azure AD公钥解码并验证ID Token"""
    unverified_header = jwt.get_unverified_header(token)
    kid = unverified_header["kid"]
    
    if kid not in public_keys:
        raise ValueError("Invalid token key ID (kid).")

    # 解码并验证Token合法性
    decoded_token = jwt.decode(
        token,
        public_keys[kid],
        algorithms=["RS256"],
        audience=CLIENT_ID,
        issuer=f"https://login.microsoftonline.com/{TENANT_ID}/v2.0"
    )
    return decoded_token

def get_token_from_headers():
    """尝试从Streamlit请求参数中提取ID Token"""
    headers = st.query_params()
    token = headers.get("id_token", [None])[0]
    if not token:
        st.error("No ID token found in the headers.")
        st.stop()
    return token

public_keys = get_public_keys()
token = get_token_from_headers()
user_info = validate_id_token(token, public_keys)

除此之外，我们还尝试过调用Microsoft Graph API来获取用户组信息和权限，但依然没解决核心的无缝登录和Blob权限过滤问题。

想请教大家：

1. 部署在Azure Web App上的Streamlit应用，应该怎么配置才能实现无需外部跳转的Microsoft无缝身份验证，同时限制特定域组的用户访问？
2. 如何根据用户邮箱，正确实现Blob存储文件夹的权限隔离，确保用户只能访问自己有权限的内容？

希望能得到大家的思路或代码修改建议，非常感谢！

备注：内容来源于stack exchange，提问作者G.Foutzopoulos