如何在重定向后在Python脚本中获得OAuth授权?

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

我需要编写一个Python脚本来运行OAuth 2.0客户端流程。在流程中,我需要让用户运行Python脚本,打开浏览器以登录并授予访问权限,然后获得授权授予以换取访问令牌并继续流程。我一直在寻找,并没有找到一个很好的答案,如何做到这一点,因为大多数例子不涉及实际的OAuth登录或需要复制/粘贴。有没有一种干净的方法可以无缝地将流集成到我的Python代码中以供本地使用?
我简单地尝试了线程和http来创建一个本地主机服务器来获取请求,尝试了flask,并试图看看创建一个协议处理程序是否有帮助。我在所有这些方面都取得了一些成功,并且能够通过流程,但似乎仍然无法从重定向URI中提取授权代码。脚本将在内部运行,不会分发,但至少需要足够干净,以免发生复制/粘贴。

b1payxdu

b1payxdu1#

下面的代码将创建一个临时本地主机,并允许从URL中抓取授权许可,沿着检索和返回访问令牌。

import requests
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import webbrowser

access_token = None

# Simple HTTP request handler to capture the authorization code
class AuthorizationHandler(BaseHTTPRequestHandler):
    authorization_code = None

    def do_GET(self):
        global access_token
        if self.path.startswith("/oauth-callback?"):
            # Extract the authorization code from the query parameters
            authorization_code = self.path.split("&code=")[1]

            # Display the authorization code
            print("Authorization Code:", self.authorization_code)

            # Send a response to the browser
            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"<h1>Authorization Code Received</h1>")

            # Obtain OAuth Client Access Token
            headers = {'accept':'*/*'}
            parameters = {'client_id':'{{Client_ID}}','client_secret':'{{Client_Secret}}','access_code':authorization_code}
            response = requests.get("{{tokenURL}}",headers=headers,params=parameters)
            print(response.text)
            token = response.text

            # Log in using Access Token from OAuth
            headers = {'accept':'*/*','Content-Type':'application/json'}
            body = {'token':token}
            response = requests.post("{{loginURL}}",headers=headers,json=body)
            print(response.text)
            json_response = response.json()
            access_token= json_response["token"]

            # Exit the function after obtaining the token
            raise SystemExit

# Start a temporary HTTP server in a separate thread
def start_temp_server():
    server = HTTPServer(("localhost", 8000), AuthorizationHandler)
    server.serve_forever()

# Example usage
authorization_url = "{{AuthorizationURL}}"

# Start the temporary server in a separate thread
server_thread = threading.Thread(target=start_temp_server)
server_thread.start()

# Open the authorization URL in the default web browser
webbrowser.open(authorization_url)

server_thread.join()

字符串

相关问题