Django实现微信网页授权登录：从配置到中间件鉴权
本文详解在Django框架中实现微信网页授权登录的完整流程。内容涵盖服务号认证、业务域名配置及SSH端口转发等前置准备，深入解析OAuth2授权码获取、Access Token交换及用户信息接口调用。同时提供基于JWT的自定义中间件实现方案，通过解析请求头或Cookie中的Token提取OpenID，并在视图层直接获取用户身份，实现无缝的用户状态保持与安全鉴权。
开发前环境准备
一、准备工作
0x00 开发前准备
- 需拥有经过微信认证的服务号。
- 完成微信认证。
- 准备已备案的域名。
- 准备可用的服务器环境。
0x01 手动触发dns更新
0x02 配置业务域名
0x03 将服务器请求转发到本地
修改服务器的 /etc/ssh/sshd_config 文件，加入配置 GatewayPorts yes。
ssh -R 0.0.0.0:80:localhost:8080 user@server_host
微信网页授权流程实现
二、微信网页授权
0x01 授权流程
用户同意授权后，获取 code。
通过技术手段引导用户页面跳转至微信授权链接（例如在装饰器中实现跳转逻辑）：
def get_wx_authorize_url(appid : str, state: str = None):
if state is None:
state = "".join([random.choice(string.ascii_letters + string.digits) for _ in range(20)])
redirect_url = 'your callback url' # 回调链接，在这里面进行用户信息入库的操作
response_type = 'code'
scope = 'snsapi_userinfo'
wx_url = f"https://open.weixin.qq.com/connect/oauth2/authorize?appid={appid}&redirect_uri={redirect_url}&response_type={response_type}&scope={scope}&state={state}#wechat_redirect"
return wx_url利用 code 换取 access_token 和 openid。
def request_access_token(appid : str, secret : str, code: str):
secret = settings.WX_SECRET
api = f"https://api.weixin.qq.com/sns/oauth2/access_token?appid={appid}&secret={secret}&code=[code]&grant_type=authorization_code"
r = requests.get(api)
return r.json()通过 access_token 换取 用户信息。
def request_userinfo(access_token: str, openid: str):
api = f"https://api.weixin.qq.com/sns/userinfo?access_token={access_token}&openid={openid}&lang=zh_CN"
r = requests.get(api)
return r.json()将获取到的用户信息存入数据库。
需要注意：微信返回的数据编码格式为 ISO-8859-1，需转换为 utf-8 格式。
def convert_string_encoding(s: str, from_encoding: str, to_encoding: str) -> str:
"""先根据 from_encoding 转换成bytes，然后在 decode 为 to_encoding 的字符串
"""
return bytes(s, encoding=from_encoding).decode(to_encoding)
nickname = convert_string_encoding(resp['nickname'], 'ISO-8859-1', 'utf-8')授权完成后，跳转回用户原本访问的链接。
我的实现方式是在数据库中保存一条记录，以 state 作为 key。
from app.models import WXUser, RedirectUrl
from utils import get_wx_authorize_url, get_random_string
from django.shortcuts import redirect
def login_required(func):
def wrapper(request, *args, **kwargs):
openid = request.openid
try:
user = WXUser.objects.get(openid=openid)
request.wxuser = user
except WXUser.DoesNotExist:
state = get_random_string()
redirect_url = get_wx_authorize_url(state=state)
# 存储跳转链接
try:
r = RedirectUrl.objects.get(state=state)
except RedirectUrl.DoesNotExist:
r = RedirectUrl()
r.state = state
origin_url = request.get_raw_uri()
r.url = origin_url
r.save()
return redirect(redirect_url)
return func(request, *args, **kwargs)
return wrapper随后在设置的回调接口中（该接口会携带 code 和 state 参数），即可通过 state 从数据库中获取原始访问链接。
class RedirectUrl(BaseModel):
state = models.TextField(unique=True)
url = models.TextField()基于JWT的认证中间件
0x02 中间件
该中间件采用 jwt 作为认证机制。关于为何不使用 session 的原因，属于另一个话题，此处不再赘述。
从 HTTP_AUTHORIZATION（即请求头中的 Authorization 字段）或者 key 为 jwttoken 的 cookie 中提取 jwt token，解析出其中的 openid，并将其添加到 request 变量中。此后，在后续的 views 中即可通过 request.openid 直接获取 openid。
def jwt_decode(token: Union[str, bytes]) -> tuple:
"""
:param token : 可以是 bytes 也可以是 str，如果是 str，会先 encode 转成 bytes
:return: 第一个参数为 payload，第二个参数为异常类型
"""
if isinstance(token, str):
token = token.encode()
secret = settings.JWT_SECRET
try:
return jwt.decode(token, secret, algorithms=["HS256"]), None
except Exception as e:
# 统一捕捉异常：
# jwt.exceptions.DecodeError
# jwt.exceptions.InvalidSignatureError
# jwt.exceptions.ExpiredSignatureError
return None, e
class JWTAuthMiddleware(object):
"""
小程序认证中间件
"""
def __init__(self, get_response=None):
self.get_response = get_response
def __call__(self, request, *args, **kws):
token = self.get_authorization_header(request)
payload, error = jwt_decode(token)
if not error:
openid = payload['openid']
request.openid = openid
else:
request.openid = None
response = self.get_response(request, *args, **kws)
return response
def get_authorization_header(self, request):
"""
从 AUTHORIZATION 请求头或者cookie 中获取 jwt code
cookie 的 jwt code 的 key 为 jwtcode
:param request:
:return: rawtoken
"""
auth_header = request.META.get('HTTP_AUTHORIZATION', '')
cookie = request.COOKIES
rawtoken = None
if auth_header != "":
try:
rawtoken = auth_header.split(" ")[1]
except IndexError as e:
pass
if 'jwttoken' in cookie:
rawtoken = cookie['jwttoken']
return rawtoken以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
