您好,登錄后才能下訂單哦!
一、準備工作
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/oauth3/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/oauth3/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()
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
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持億速云。
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。