主要流程：
实现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)
网友评论