|
|
|
|
from sanic import Sanic, json, SanicException
|
|
|
|
|
from sanic.log import logger
|
|
|
|
|
|
|
|
|
|
from decorator import decorator
|
|
|
|
|
import json as jsonlib
|
|
|
|
|
import urllib.parse
|
|
|
|
|
|
|
|
|
|
from utils.crypto import base64_encode
|
|
|
|
|
from utils.crypto import 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)))
|
