问题描述
阅读文档后，我很难理解如何使用 Authlib 为 OpenID Connect 提供程序实现授权代码流。阅读文档后，我尝试实现下面列出的以下代码。
/login 端点使用 authlib 重定向到身份提供者的授权，在本例中为 Cognito。这将重定向到 /aws_cognito_redirect，我目前已经实现了它来解决检索令牌的代码挑战。
我的问题是：
- 如何使用
authlib来解决代码挑战而不是自己实现这一部分？ - Authlib 是否提供在 HTTP Only cookie 中返回令牌并在包含 cookie 的后续请求中验证令牌的功能？例如，Authlib 是否允许将端点装饰/标记为受保护，在这种情况下，它将验证 HTTP Only cookie 中的令牌？
更新
在检查 source code 之后，我最终想出了如何将 Authlib 与 FastAPI 结合使用来解决代码挑战。源代码包含在此问题的末尾。
我将问题悬而未决，因为第二部分仍未得到解答。
目前，这个 question 表明可以使用 ResourceProtector 类来满足我的需要。但是，它有一个 parse_request_authorization 方法，用于检查不记名令牌请求的 Authorization 标头。所以...我假设该方法是对 ResourceProtector 类进行子类化并覆盖此方法以检查对仅 HTTP cookie 的请求并提取其中包含的 JWT 以进行验证？？此功能是否由 Authlib 实施和提供？
或者，也调查一下我是否可以集成 fastapi-login 来实现此功能。
附录：源代码
import base64
from functools import lru_cache
import httpx
from authlib.integrations.starlette_client import OAuth
from fastapi import Depends,FastAPI,Request,Response
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware
from . import config
@lru_cache()
def get_settings() -> config.Settings:
"""Create config settings instance encapsulating app config."""
return config.Settings()
def get_auth_base_url(region: str,userpool_id: str) -> str:
# base_url = "https://cognito-idp.us-east-2.amazonaws.com/us-east-2_QqNgzdtT5"
base_url = "https://cognito-idp." + region + ".amazonaws.com/" + userpool_id
return base_url
app = FastAPI()
app.add_middleware(SessionMiddleware,secret_key="secretly")
oauth = OAuth()
oauth.register(
"cognito",client_id=get_settings().client_id,client_secret=get_settings().client_secret,server_Metadata_url=get_auth_base_url(
get_settings().region,get_settings().userpool_id
)
+ "/.well-kNown/openid-configuration",client_kwargs={"scope": "openid email"},)
def encode_auth_header(client_id: str,client_secret: str) -> str:
"""Encode client id and secret as base64 client_id:client_secret."""
secret = base64.b64encode(
bytes(client_id,"utf-8") + b":" + bytes(client_secret,"utf-8")
)
return "Basic " + secret.decode()
@app.get("/login")
async def login(request: Request):
"""Redirect to /aws_cognito_redirect endpoint."""
cognito = oauth.create_client("cognito")
redirect_uri = request.url_for("read_code_challenge")
return await cognito.authorize_redirect(request,redirect_uri)
@app.get("/aws_cognito_redirect")
async def read_code_challenge(
request: Request,response: Response,settings: config.Settings = Depends(get_settings),):
"""Retrieve tokens from oauth2/token endpoint and return session cookie."""
code = request.query_params["code"]
print("/aws_cognito_redirect received code := ",code)
auth_secret = encode_auth_header(settings.client_id,settings.client_secret)
headers = {"Authorization": auth_secret}
print("Authorization:" + str(headers["Authorization"]))
payload = {
"client_id": settings.client_id,"code": code,"grant_type": "authorization_code","redirect_uri": settings.redirect_uri,}
token_url = (
"https://"
+ settings.domain
+ ".auth."
+ settings.region
+ ".amazoncognito.com/oauth2/token"
)
async with httpx.Asyncclient() as client:
tokens = await client.post(
token_url,data=payload,headers=headers,)
tokens.raise_for_status()
print("Tokens\n" + str(tokens.json()))
response.set_cookie(key="jwt",value=tokens.content,httponly=True)
import base64
from functools import lru_cache
from authlib.integrations.starlette_client import OAuth
from fastapi import FastAPI,Request
from starlette.middleware.sessions import SessionMiddleware
from . import config
@lru_cache()
def get_settings() -> config.Settings:
"""Create config settings instance encapsulating app config."""
return config.Settings()
@lru_cache
def get_auth_base_url(region: str,userpool_id: str) -> str:
"""Return cognito discover points base url from region and userpool ID."""
return ("https://cognito-idp." + region + ".amazonaws.com/" + userpool_id)
app = FastAPI()
app.add_middleware(SessionMiddleware,secret_key="some-random-string")
oauth = OAuth()
cognito = oauth.register(
"cognito","utf-8")
)
return "Basic " + secret.decode()
@app.get("/")
async def login(request: Request):
"""Redirect to /aws_cognito_redirect endpoint after sign-in."""
redirect_uri = request.url_for("read_code_challenge")
return await cognito.authorize_redirect(request,redirect_uri)
@app.get("/aws_cognito_redirect")
async def read_code_challenge(request: Request):
"""Request a token from cognito using code challenge response."""
return await cognito.authorize_access_token(request)
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）