使用OAuth2 client_credentials的自动令牌获取与Python请求流

3npbholx  于 2023-08-02  发布在  Python
关注(0)|答案(1)|浏览(118)

我想使用一个通过 PythonOAuth2client_credentials流进行身份验证的API。
在pyhton中,使用最广泛的HTTP客户端是Requests,而Requests具有许多高级特性和扩展,其中一些特性和扩展是围绕着与OAuth2一起使用的。
然而,Oauth2是一个复杂的野兽,支持4个不同的流,其中client_credentials是最简单的。尽管我已经尝试过了,但我还没有找到一个简单的例子,可以将Requestsclient_credentials流一起使用。它总是在文章的最后提到,作为一个旁注或事后的想法后,解释了所有其他流程在intricated细节。
client_credentials流程中,会发生以下情况:

1.将以下编码为x-www-form-urlencoded的内容POST到“token endpoint”:

{
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "all", # Or whatever scope you care about
}

字符串
其中client_idclient_secret由您正在访问的服务提供。

2.以下内容将以json形式返回(假设验证成功):

{
    "token_type":"Bearer",
    "scope":"openid email profile offline_access roles read:tickets read:customers ...",
    "access_token":"XXX_THIS_TOKEN_IS_VERY_SECRET_XXX"
}

3.使用返回数据中的access_token访问受保护资源:

您可以通过以下方式在所有对受保护资源的请求中的Authorization: Bearer头中设置它来使用它:

Authorization: Bearer XXX_THIS_TOKEN_IS_VERY_SECRET_XXX


令牌将从生成开始持续1小时,并且当资源请求返回401 auth错误时,只需重复步骤#1和#2以生成要使用的新令牌。
我当前的代码手动地从令牌端点请求令牌,从生成的json中解析令牌,并将其填充到后续请求的HTTP头中。然后我继续检测401错误并重新验证。这是可行的,但它是非常tediouos。我现在不得不用这个丑陋的烂摊子 Package 每个请求用法。
我希望有一种方法可以做到这一点:

client = MagicRequestsWrapper(client_id=client_id, client_secret=client_secret, token_endpoint=token_endpoint, scope=scope)

res = client.get(protected_resource_endpoint)


其中所有的oauth内容现在都被隐藏了,客户端的行为与普通的Requests客户端完全相同。
我怎么能做到呢?

5sxhfpxr

5sxhfpxr1#

我在写下这个问题后不久就找到了答案。

import requests
from requests_oauth2client import OAuth2Client, OAuth2ClientCredentialsAuth

def oauth2_session(client_id, client_secret, base_url, scope="all"):
    token_url = f"{base_url}/auth/token" # This may be different for your endpoint

    oauth2client = OAuth2Client( token_endpoint=token_url, client_id=client_id, client_secret=client_secret)
    auth = OAuth2ClientCredentialsAuth( oauth2client, scope=scope, resource=base_url)

    session = requests.Session()
    session.auth = auth
    return session

字符串
示例用法如下:

# Create a session
base_url = "https://myservice.com"
my_session = oauth2_session(client_id=os.environ.get("CLIENT_ID"), client_secret=os.environ.get("CLIENT_SECRET"), base_url=base_url)

# Use it as normal requests object to post and get
result = my_session.get(f"{base_url}/api/some_resource")

my_session.post(f"{base_url}/api/some_resource", json={"test":123})

相关问题