import json as jsonlib import urllib.parse from decorator import decorator from sanic import Sanic, SanicException, json from sanic.log import logger from utils.crypto import base64_encode, get_rsa_privkey, rsa_sha1_sign @decorator def check_token(f, *args, **kw): request = args[0] user_id = request.headers["user-id"] request.ctx.user_id = user_id authorize = dict(urllib.parse.parse_qsl(request.headers["authorize"])) token = authorize["token"] logger.debug(f"{request.path=} {user_id=} {token=}") return f(*args, **kw) if token != Sanic.get_app().ctx.auth_token[user_id]: raise SanicException("Auth failed", status_code=403) return f(*args, **kw) @decorator def parse_data(f, *args, **kw): request = args[0] request_data = jsonlib.loads(request.form.get("request_data", "")) # type: ignore logger.debug(f"{request.path=} {request_data=}") request.ctx.request_data = request_data return f(*args, **kw) @decorator async def pack_resp(f, *args, **kw): request = args[0] ret = await f(*args, **kw) data = { "response_data": ret, "release_info": {}, "status_code": 200 } x_message_sign = base64_encode( rsa_sha1_sign(jsonlib.dumps(data).encode(), get_rsa_privkey())) logger.debug(f"{request.path=} {data=} {x_message_sign=}") return json(data, headers={ "X-Message-Sign": x_message_sign, "Server-Version": Sanic.get_app().config["PACKAGE_VERSION"] }) def common_api(f): return check_token(parse_data(pack_resp(f)))