90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
import base64
|
|
import datetime
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
import jwt
|
|
|
|
SECRET_KEY = 'IrOYFjXtQPuofoXEpsR+2VsvjnaCWkPzvfmym1qNmcI='
|
|
"""
|
|
自定义密钥,生成方法见 generate_secret_key() 函数。该密钥应当在应用程序服务器启动时更新,并使用动态生成的密钥。
|
|
"""
|
|
|
|
PRIVATE_ISS = 'hǎi_tén_education_technology_co_ltd'
|
|
"""
|
|
签发者签名。
|
|
"""
|
|
|
|
|
|
def generate_secret_key():
|
|
"""
|
|
生成如 IrOYFjXtQPuofoXEpsR+2VsvjnaCWkPzvfmym1qNmcI= 的随机串。
|
|
"""
|
|
return base64.b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes).decode()
|
|
|
|
|
|
def reset_secret_key():
|
|
"""
|
|
重置加密密钥。
|
|
|
|
:return: 加密密钥
|
|
"""
|
|
global SECRET_KEY
|
|
SECRET_KEY = generate_secret_key()
|
|
return SECRET_KEY
|
|
|
|
|
|
def get_secret_key():
|
|
"""
|
|
取得加密密钥。
|
|
|
|
:return: 加密密钥
|
|
"""
|
|
global SECRET_KEY
|
|
return SECRET_KEY
|
|
|
|
|
|
def encode_token(exp: Optional[datetime.datetime] = None, **kwargs):
|
|
"""
|
|
对用户信息加密生成令牌。
|
|
|
|
:param exp: 过期时间,不传默认 7 天过期
|
|
:param kwargs: 要加入到数据部分的参数
|
|
:return: 加密后的 token 内容
|
|
"""
|
|
try:
|
|
iat = datetime.datetime.now(datetime.timezone.utc)
|
|
exp = exp if exp else iat + datetime.timedelta(days=7)
|
|
# 即 JWT 三部分中的载荷部分
|
|
# 过期时间最后是和 UTC 作比较,所以设置的时候使用 datetime.datetime.now()
|
|
payload = {
|
|
'iss': PRIVATE_ISS, # 签发者
|
|
'iat': iat, # 签发时间
|
|
'exp': exp, # 过期时间,这里设置7天
|
|
'params': {} # 参数,存放用户自定义数据
|
|
}
|
|
|
|
payload['params'].update(kwargs)
|
|
|
|
# 开始进行加密,返回字符串,传入例如密钥,指定加密算法
|
|
# encode 返回的是 bytes,需要 decode() 得到 str,不转换的话,在封装json的时候报错
|
|
auth_token = jwt.encode(payload, SECRET_KEY, algorithm='HS256')
|
|
if isinstance(auth_token, bytes):
|
|
auth_token = auth_token.decode()
|
|
return auth_token
|
|
except Exception as e:
|
|
raise e
|
|
|
|
|
|
def decode_token(auth_token: str):
|
|
"""
|
|
解码 token 并从中提取用户信息,进行验证。
|
|
|
|
:param auth_token: 令牌
|
|
"""
|
|
# 如果需要关闭过期时间的验证,可以在 options 中使用 verify_exp
|
|
# jwt.decode(auth_token, secret_key, issuer=private_iss, algorithms=['HS256'], options={'verify_exp': False})
|
|
# 传入了密钥和算法,和加密是对应的,因此密钥一定不要泄露
|
|
token_payload = jwt.decode(auth_token, SECRET_KEY, issuer=PRIVATE_ISS, algorithms=['HS256'])
|
|
return token_payload
|