python實現(xiàn)微博oauth2認證

主要流程：

實現(xiàn)OAuth 2.0流程：編寫B(tài)ottle路由來處理授權(quán)請求玄坦、回調(diào)和令牌交換敬尺。

JWT處理：在用戶通過OAuth驗證后肢执，生成JWT并將其發(fā)送給客戶端。后續(xù)請求可以使用JWT進行身份驗證。

# !/usr/bin/env python3

# -*- coding: utf-8 -*-

import requests

import json

import jwt

import os

from datetime import datetime, timedelta

from http import HTTPStatus

from bottle import Bottle, request, response, redirect, template, static_file, run

from requests.packages.urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

app = Bottle()

# 生成一個隨機的32長度的密鑰

# JWT_SECRET_KEY = os.urandom(32)

JWT_SECRET_KEY = '9q8w7e60asjhywe4f1qaz2wsx9jkseduc4rfv'.encode()

COOKIE_SECRET = '9q8w7e6r1a2s3d4f'

COOKIE_KEY = "X-Auth-Cookie"

CLIENT_ID = '384983687214'

CLIENT_SECRET = 'b1fboo33bgb827427bf119apad882412'

AUTHORIZATION_URL = 'https://api.weibo.com/oauth2/authorize'

ACCESS_TOKEN_URL = 'https://api.weibo.com/oauth2/access_token'

USER_INFO_URL = 'https://api.weibo.com/oauth2/get_token_info'

# REDIRECT_URL = 'http://localhost:5000/callback'

# REDIRECT_URL = 'http://localhost:8001/user/callback'

# REDIRECT_URL = 'http://localhost:8001/user/localLogin'

# REDIRECT_URL = 'https://b97b-223-112-131-202.ngrok-free.app/api/vdi-server/user/localLogin'

REDIRECT_URL = 'https://5111-112-81-204-118.ngrok-free.app/callback'

@app.route('/')

def index():

? ? # 已登錄瀏覽器使用cookie校驗或瀏覽器客戶端均使用token

? ? cookie_value = request.get_cookie(COOKIE_KEY, secret=COOKIE_SECRET)

? ? print('cookie_value=', cookie_value)

? ? if cookie_value:

? ? ? ? # 校驗cookie

? ? ? ? return template('''

? ? ? ? <html>

? ? ? ? <body>

? ? ? ? ? ? <h1>Welcome, {{cookie_value}}!</h1>

? ? ? ? </body>

? ? ? ? </html>

? ? ? ? ''', cookie_value=cookie_value)

? ? # 未登錄構(gòu)造授權(quán)鏈接

? ? auth_url = f"{AUTHORIZATION_URL}?client_id={CLIENT_ID}&response_type=code&redirect_uri={REDIRECT_URL}&state=terminal_or_admin_id"

? ? return template('''

? ? <html>

? ? <body>

? ? ? ? <h4>auth_url, {{auth_url}}!</h4>

? ? ? ? <a href="{{auth_url}}">Login with WeiBo</a>

? ? </body>

? ? </html>

? ? ''', auth_url=auth_url)

? ? # print(auth_url)

? ? # redirect(auth_url)

@app.route('/callback')

def app_redirect():

? ? # 從WeiBo獲取code

? ? query = request.query

? ? print('code_request_url=', request.url)

? ? print(query, type(query))

? ? code = query.get('code')

? ? if not code:

? ? ? ? print('params code not found!')

? ? ? ? return template('''?

? ? ? ? <html>?

? ? ? ? <body>?

? ? ? ? ? ? <h1>Welcome, {{login}}!</h1>

? ? ? ? </body>?

? ? ? ? </html>?

? ? ? ? ''', login='NOT FOUND')

? ? print('code=', code)

? ? print('state=', query.get('state'))

? ? # 使用code獲取access_token

? ? data = {

? ? ? ? 'client_id': CLIENT_ID,

? ? ? ? 'client_secret': CLIENT_SECRET,

? ? ? ? 'grant_type': 'authorization_code',

? ? ? ? 'redirect_uri': REDIRECT_URL,

? ? ? ? 'code': code,

? ? }

? ? headers = {'Accept': 'application/json'}

? ? res = requests.post(ACCESS_TOKEN_URL, params=data, headers=headers, verify=False, timeout=5)

? ? print('request url', request.url)

? ? if res.status_code != HTTPStatus.OK:

? ? ? ? print('request token error', res.status_code)

? ? ? ? return

? ? print('res.json() = ', res.json())

? ? access_token = res.json().get('access_token')

? ? print('access_token=', access_token)

? ? # 使用access_token獲取用戶信息

? ? # headers = {'Authorization': f'token {access_token}'}

? ? user_url = f"{USER_INFO_URL}?access_token={access_token}"

? ? user_res = requests.post(user_url, verify=False, timeout=5)

? ? if user_res.status_code != HTTPStatus.OK:

? ? ? ? print('request user error', user_res.status_code)

? ? ? ? return

? ? user_data = user_res.json()

? ? # 方式一瀏覽器可設(shè)置cookie? 或客戶端和客戶端均使用token jwt等

? ? response.set_cookie(COOKIE_KEY, user_data['uid'], secret=COOKIE_SECRET, max_age=6000)

? ? # 方式二生成JWT

? ? jwt_token = generate_jwt(user_data['uid'], user_data['uid'])

? ? user_data['jwt'] = jwt_token

? ? print(user_data)

? ? # return user_data

? ? # demo直接顯示用戶信息實際應(yīng)返回token或jwt 客戶端或瀏覽器存儲每次接口訪問攜帶過來

? ? return template('''

? ? <html>

? ? <body>

? ? ? ? <h1>Welcome, {{uid}}!</h1>

? ? ? ? <p>Your WeiBo id is {{uid}}.</p>

? ? ? ? <p>Your WeiBo login is {{uid}}.</p>

? ? ? ? <p>Your Jwt token is {{jwt}}.</p>

? ? </body>

? ? </html>

? ? ''', **user_data)

# 生成JWT的函數(shù)

def generate_jwt(user_id, user_name, expiration_time=3600):

? ? payload = {

? ? ? ? 'user_id': user_id,

? ? ? ? 'user_name': user_name,

? ? ? ? 'exp': datetime.utcnow() + timedelta(seconds=expiration_time),

? ? ? ? 'iat': datetime.utcnow()

? ? }

? ? return jwt.encode(payload, JWT_SECRET_KEY, algorithm='HS256')

# 簡單的JWT驗證示例（需要額外路由處理JWT驗證）

@app.route('/protected')

def protected():

? ? token = request.headers.get('Authorization')

? ? if not token:

? ? ? ? return "JWT token is missing"

? ? try:

? ? ? ? payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=['HS256'])

? ? ? ? return f"Welcome, {payload['user_name']}!"

? ? except jwt.ExpiredSignatureError:

? ? ? ? return "The token has expired"

? ? except jwt.InvalidTokenError:

? ? ? ? return "Invalid token"

if __name__ == '__main__':

? ? run(app, host='localhost', port=5000)