kotori/sif/common.py

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)))