如何使用 Azure B2G 和 FastAPI 拥有多个范围？

Question

我正在这家公司实习，他们要求我研究 Azure B2C 以及如何将其与他们的 FastAPI 集成。我有这个小测试 API 来制作演示。我一直在使用 fastapi-azure-auth 并遵循他们的文档here。这相当容易。我有一个类可以从 .env

获取所有必需的值

from pydantic import AnyHttpUrl, ConfigDict, computed_field
from pydantic_settings import BaseSettings, SettingsConfigDict


class AzureSettings(BaseSettings):
    BACKEND_CORS_ORIGINS: list[str | AnyHttpUrl] = ['http://localhost:8000']
    TENANT_NAME: str = ""
    APP_CLIENT_ID: str = ""
    OPENAPI_CLIENT_ID: str = ""
    AUTH_POLICY_NAME: str = ""
    SCOPE: str = ""

    @computed_field
    @property
    def SCOPE_NAME(self) -> str:
        return f'https://{self.TENANT_NAME}.onmicrosoft.com/{self.APP_CLIENT_ID}/{self.SCOPE}'

    @computed_field
    @property
    def SCOPES(self) -> dict:
        return {
            self.SCOPE_NAME: self.SCOPE
        }

    @computed_field
    @property
    def OPENID_CONFIG_URL(self) -> str:
        return f'YOUR_URL_HERE"'

    @computed_field
    @property
    def OPENAPI_AUTHORIZATION_URL(self) -> str:
        return f'YOUR_URL_HERE"'

    @computed_field
    @property
    def OPENAPI_TOKEN_URL(self) -> str:
        return f'YOUR_URL_HERE'


    model_config = SettingsConfigDict(
        ConfigDict(extra='allow'),
        env_file='.env',
        env_file_encoding='utf-8',
        case_sensitive=True, 
    )
    
azure_premium = AzureSettings(SCOPE="premium")
azure_admin = AzureSettings(SCOPE="admin")
azure_basica = AzureSettings(SCOPE="basica")

这几乎是他们提供的普通代码，我只是修改了它，以便您可以创建具有不同范围的对象。

但是，当我在路线上使用它时，只有最后创建的作用域显示在 FastAPI 上，而我一生都无法弄清楚。

这是我的主要：

description = """
![API Logo](https://mexicocity.cdmx.gob.mx/wp-content/themes/travel-cdmx/src/images/axolotl.svg)
<br></br>
FastApi for learning and testing. API.
"""

limiter = Limiter(key_func=get_remote_address)

app = FastAPI(
    title="Learning and testing. API",
    description=description,
    contact={'email': '[email protected]'},
    swagger_ui_oauth2_redirect_url='/oauth2-redirect',
    swagger_ui_init_oauth={
        'usePkceWithAuthorizationCodeGrant': True,
        'clientId': azure_basica.OPENAPI_CLIENT_ID,
    }
)


app.mount("/static", StaticFiles(directory="static"), name="static")

app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Protege /docs y /redoc con api_key
expected_api_key = os.getenv("API_KEY")

# Middleware para verificar la API key en las rutas /docs y /redoc
@app.middleware("http")
async def check_api_key(request: Request, call_next):
    if request.url.path in ["/docs", "/redoc"]:
        api_key = request.query_params.get("api_key")

        if api_key != expected_api_key:
            return JSONResponse(status_code=401, content={"detail": "API Key inválida"})

    response = await call_next(request)
    return response

# Database creation
'''
If Mockup is true, uses a SQLite file, otherwise local PostgreSQL instance.
'''
create_db_and_tables()
db = next(get_session_context())
#create_defaultAdmin_user(db)
#create_defaultClient_user(db)

mockup = True

if mockup:
    from database.db_initdata_mockup import insert_libros, insert_pelicula_serie
    insert_libros(db)
    insert_pelicula_serie(db)    
    
# Basic exception handling
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

# General exception handling
@app.exception_handler(Exception)
async def exception_handler(request: Request, exc: Exception):
    logger.exception(f"Unexpected error: {exc}")
    return JSONResponse(
        status_code=500,
        content={
            "message": f"Unexpected Error: {exc.__class__.__name__}.",
            "description": str(exc)
        }
    )

# HTTP exception handling
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail}
    )

azure_scheme_scope_basic = B2CMultiTenantAuthorizationCodeBearer(
    app_client_id=azure_basica.APP_CLIENT_ID,
    openid_config_url=azure_basica.OPENID_CONFIG_URL,
    openapi_authorization_url=azure_basica.OPENAPI_AUTHORIZATION_URL,
    openapi_token_url=azure_basica.OPENAPI_TOKEN_URL,
    scopes=azure_basica.SCOPES,
    validate_iss=False,
)

azure_scheme_scope_premium = B2CMultiTenantAuthorizationCodeBearer(
    app_client_id=azure_premium.APP_CLIENT_ID,
    openid_config_url=azure_premium.OPENID_CONFIG_URL,
    openapi_authorization_url=azure_premium.OPENAPI_AUTHORIZATION_URL,
    openapi_token_url=azure_premium.OPENAPI_TOKEN_URL,
    scopes=azure_premium.SCOPES,
    validate_iss=False,
)


@app.on_event('startup')
async def load_config() -> None:
    """
    Load OpenID config on startup.
    """
    await azure_scheme_scope_basic.openid_config.load_config()
    await azure_scheme_scope_premium.openid_config.load_config()


app.include_router(libros_router, dependencies=[Security(azure_scheme_scope_basic)])
app.include_router(peliculasSeries_router, dependencies=[Security(azure_scheme_scope_premium)])
app.include_router(librosPeliculasCombinados_router)
app.include_router(usuarios_router)

这就是单击右上角全局身份验证时在 /docs 中显示的方式。

当我进行单端点身份验证时，范围甚至没有出现。

Azure B2C 配置工作正常，因为我可以通过全局身份验证进行身份验证。只是只出现了最后一个范围。

将范围要求移至路由文件本身（相同的结果），将范围创建移至对象初始化（相同的结果）

Answer 1

我或多或少已经解决了。 B2CMultiTenantAuthorizationCodeBearer 是一个静态类，因此它当然每次都会重写。范围还需要作为字典传递。看起来像这样。

@computed_field
@property
def SCOPE_NAME(self) -> str:
    return f'https://{self.TENANT_NAME}.onmicrosoft.com/{self.APP_CLIENT_ID}'

@computed_field
@property
def SCOPES(self) -> dict:
    return {
         f'{self.SCOPE_NAME}/admin': "Acceso a todos los endpoint", f'{self.SCOPE_NAME}/premium': "Acceso a los endpoint premium",     f'{self.SCOPE_NAME}/basica': "Acceso básico", 
    }

我总体上仍然有一些小问题，但我有信心我会解决它们。