python实现微博oauth2认证

主要流程：

实现OAuth 2.0流程：编写Bottle路由来处理授权请求、回调和令牌交换。

JWT处理：在用户通过OAuth验证后，生成JWT并将其发送给客户端。后续请求可以使用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)

# 未登录构造授权链接

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()

# 方式一浏览器可设置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直接显示用户信息实际应返回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的函数

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)