leguadmin/pointapi
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

列表加签问题

Browse Source
This commit is contained in:
wuaho 2021-04-17 15:56:14 +08:00
parent 893a129c59
commit 42f1d7181e
2 changed files with 20 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
common/verification.py
View File

@ -1,18 +1,25 @@
import hashlib import hashlib
from copy import deepcopy
from settings import settings from settings import settings
__all__ = 'restore_field', 'sort_kv', 'check_sign', 'check_preset' __all__ = 'restore_field', 'sort_kv', 'check_sign', 'check_preset'
def restore_field(data: dict) -> dict: def restore_field(data: dict) -> tuple:
res = dict() res = dict()
sign_kv = dict()
for k, v in data.items(): for k, v in data.items():
key = settings.FIELD_MAP.get(k) or k
if isinstance(v, dict): if isinstance(v, dict):
res[settings.FIELD_MAP.get(k) or k] = restore_field(v) res[key], sign_kv[key] = restore_field(v)
else: else:
res[settings.FIELD_MAP.get(k) or k] = v res[key] = v
return res if isinstance(v, list):
sign_kv[key] = ','.join(map(lambda x: str(x), v))
else:
sign_kv[key] = v
return res, sign_kv
def check_preset(data: dict): def check_preset(data: dict):

13
routers/point.py
View File

@ -13,7 +13,7 @@ router = APIRouter()
@router.post("/") @router.post("/")
async def point(request: Request, items: DataModel): async def point(request: Request, items: DataModel):
try: try:
public_data: dict = restore_field(items.public) public_data, _ = restore_field(items.public)
rdb = request.app.state.redis rdb = request.app.state.redis
data_factory = request.app.state.data_factory data_factory = request.app.state.data_factory
output_factory = request.app.state.output_factory output_factory = request.app.state.output_factory
@ -24,10 +24,15 @@ async def point(request: Request, items: DataModel):
if not output_factory.name: if not output_factory.name:
return {'code': -1, 'msg': '没有匹配到appid'} return {'code': -1, 'msg': '没有匹配到appid'}
for item in items.data: for item in items.data:
data: dict = restore_field(item) data, sign_data = restore_field(item)
sign = data.pop('sign') _ = data.pop('sign')
properties = data.pop('properties') properties = data.pop('properties')
if not check_sign(sign, settings.SALT.get(appid, ''), data, properties):
#
sign = sign_data.pop('sign')
sign_properties = sign_data.pop('properties')
if not check_sign(sign, settings.SALT.get(appid, ''), sign_data, sign_properties):
# continue # continue
return {"code": -1, 'msg': '签名错误'} return {"code": -1, 'msg': '签名错误'}
single_data = dict(**public_data, **data, properties=properties) single_data = dict(**public_data, **data, properties=properties)