implement sdo api

.gitignore

@@ -1,4 +1,5 @@
 
+db/
 log/
 
 

logger.py

@@ -0,0 +1,79 @@
+# from https://blog.csdn.net/u011843342/article/details/131925837
+
+import logging
+import sys
+from typing import Dict, Any
+ 
+from loguru import logger
+ 
+LOG_FORMAT = '{time:YYYY-MM-DD HH:mm:ss} [{level}] {module}:{name}:{line} - {message}'
+
+LOG_PATH = "./log"
+
+logger.add(f"{LOG_PATH}/info.log", filter=lambda record: "INFO" in record['level'].name, rotation="10 MB",
+           retention="3 days", level="INFO", format=LOG_FORMAT)
+logger.add(f"{LOG_PATH}/debug.log", filter=lambda record: "DEBUG" in record['level'].name, rotation="10 MB",
+           retention="3 days", level="DEBUG", format=LOG_FORMAT)
+logger.add(f"{LOG_PATH}/error.log", filter=lambda record: "ERROR" in record['level'].name, rotation="10 MB",
+           retention="1 days", level="ERROR", format=LOG_FORMAT)
+ 
+S_LOGGING_CONFIG_DEFAULTS: Dict[str, Any] = dict(  # no cov
+    version=1,
+    disable_existing_loggers=False,
+    loggers={
+        "sanic.root": {"level": "INFO", "handlers": ["console"], "propagate": False},
+        "sanic.error": {
+            "level": "INFO",
+            "handlers": ["error_console"],
+            "propagate": False,
+            "qualname": "sanic.error",
+        },
+        "sanic.access": {
+            "level": "INFO",
+            "handlers": ["access_console"],
+            "propagate": False,
+            "qualname": "sanic.access",
+        },
+        "sanic.server": {
+            "level": "INFO",
+            "handlers": ["console"],
+            "propagate": False,
+            "qualname": "sanic.server",
+        },
+    },
+    handlers={
+        "console": {
+            "class": "logger.InterceptHandler",
+        },
+        "error_console": {
+            "class": "logger.InterceptHandler",
+        },
+        "access_console": {
+            "class": "logger.InterceptHandler",
+        },
+    }
+)
+ 
+ 
+class InterceptHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord):
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = record.levelno
+        # Find caller from where originated the logged message
+        frame, depth = logging.currentframe(), 2
+        while frame.f_code.co_filename == logging.__file__: # type: ignore
+            frame = frame.f_back # type: ignore
+            depth += 1
+ 
+        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+ 
+ 
+def setup_log():
+    logging.root.handlers = [InterceptHandler()]
+    logging.root.setLevel("DEBUG")
+    for name in logging.root.manager.loggerDict.keys():
+        logging.getLogger(name).handlers = []
+        logging.getLogger(name).propagate = True
+    logger.configure(handlers=[{"sink": sys.stdout, "serialize": False}])

main.py

@@ -0,0 +1,31 @@
+from sanic import Sanic, Request
+from sanic.response import json
+from sanic.log import logger
+
+from logger import setup_log, S_LOGGING_CONFIG_DEFAULTS
+
+from sdo import bp_sdo
+
+
+
+setup_log()
+app = Sanic("kotori", log_config=S_LOGGING_CONFIG_DEFAULTS)
+
+app.blueprint(bp_sdo)
+
+app.config["PUBKEY_PATH"] = "./publickey.pem"
+app.config["PRIVKEY_PATH"] = "./privatekey.pem"
+
+@app.middleware("request")
+async def callback_request(request: Request):
+    logger.info(f"{request.method} - {request.path}")
+
+
+
+
+@app.route('/')
+async def test(request):
+    return json({'hello': 'world'})
+
+if __name__ == '__main__':
+    app.run(port=8080)

privatekey.pem

@@ -0,0 +1,16 @@
+-----BEGIN PRIVATE KEY-----
+MIICeAIBADANBgkqhkiG9w0BAQEFAASCAmIwggJeAgEAAoGBANa+kI5Aa9YSBq4K
+veWRBA516gmGvvNYb4Xo7BqFHHwmf5JJ4AblG6vP66kX6BVRGTpg0RvAyq4bXxaF
+kMn7asnAq0I/BC0HhgkCRbXV4PnpY29xwz/Gc84fvWFyRzSLMAmHzBWF5QRz/dEJ
+4hX2anxEVUqivVdbiVPBMsRzIrJlAgMBAAECgYEAkByxDBXliHLucqwYxk/shfvR
+xPTdfpdxuE8Y668674IyiKz1IAtu532QbBj0EiIrqQbCLjrSiccqqAWYpkQIZAO1
+r6YnhXgH7pooxPHFAopdV4oTcwvPQC1S12oUAxkEb1VsYm2Fj6NA7JFrHRVmoKSb
+NuBnh5AV910cgDWzdp0CQQDeK8JswR505cDK1aYYHyKpF2h5sPsF1ToKzD5YAA3j
++57p1+KgkPNAKyyJQ7ToHpGFg4yS5Uuuttr9briibMYLAkEA93FRaHVKrguJTxFO
+SA6jJGGxQcFvlF/mufzWlw88gnqYNJzTQar/lBULzl4Xql/Vy5uzRcUyyV0fuCGK
+x+rfTwJBAJgXuuz6s9/w3S++XQtMXU0GolYUi3QtyaNUuSVDPD8jpWGOki27rVrz
+c3Swrirtqk+Ng/GYGVyM/5PZdXp0HosCQQCZ1XFvJ7yOB84NwgyQ78itTa8N2lys
+OhMPfglLUMWluOH3k6gjI1RRk+QLIKRF397i/qGttrOkTKjzqKbHM1YxAkBijDjU
+mHW6YI5Gl/1cLlMfktKCAcS7y4MlV1pt9y91mw4j3km5+Ol7jGRKWODU+dmOGDWy
+pIyYbpYxUwUxIe13
+-----END PRIVATE KEY-----

publickey.pem

@@ -0,0 +1,6 @@
+-----BEGIN PUBLIC KEY-----
+MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDWvpCOQGvWEgauCr3lkQQOdeoJ
+hr7zWG+F6OwahRx8Jn+SSeAG5Rurz+upF+gVURk6YNEbwMquG18WhZDJ+2rJwKtC
+PwQtB4YJAkW11eD56WNvccM/xnPOH71hckc0izAJh8wVheUEc/3RCeIV9mp8RFVK
+or1XW4lTwTLEcyKyZQIDAQAB
+-----END PUBLIC KEY-----

sdo/__init__.py

@@ -0,0 +1,7 @@
+from sanic import Blueprint
+
+from .misc import bp_sdo_misc
+from .v1 import bp_sdo_v1
+
+bp_sdo = Blueprint.group(bp_sdo_misc, bp_sdo_v1)
+

sdo/misc.py

@@ -0,0 +1,16 @@
+from sanic import Blueprint, Request, json
+from sanic.log import logger
+
+bp_sdo_misc = Blueprint("sdo_misc")
+
+
+@bp_sdo_misc.route("/report/ge/app", methods=["GET", "POST"])
+async def report(request: Request):
+    # logger.debug(request.body.decode())
+    return json({ "code": 0, "msg": "", "data": { "needReport": 0 } })
+
+
+@bp_sdo_misc.route("/agreement/all", methods=["GET", "POST"])
+async def agreement(request: Request):
+    logger.debug(request.body.decode())
+    return json({ "return_code": 0, "error_type": 0, "return_message": "", "data": { } })

sdo/model.py

@@ -0,0 +1,20 @@
+from peewee import SqliteDatabase, Model, IntegerField, TextField
+
+db = SqliteDatabase('./db/sdo.db', pragmas={'journal_mode': 'wal'})
+
+class BaseModel(Model):
+    class Meta:
+        database = db
+
+class SdoUser(BaseModel):
+    id = IntegerField(primary_key=True)
+    user_id = IntegerField(null=True)
+    phone = TextField()
+    password = TextField(null=True)
+    
+    autokey = TextField(null=True)
+    ticket = TextField(null=True)
+    last_login_time = IntegerField(null=True)
+    sifkey = TextField(null=True)
+    
+db.create_tables([SdoUser])

sdo/v1.py

@@ -0,0 +1,244 @@
+from sanic import Blueprint, Request, json, Sanic
+from sanic.log import logger
+from sanic.exceptions import SanicException
+
+import urllib.parse
+import uuid
+import json as jsonlib
+import time
+
+from utils.crypto import rsa_decrypt
+from utils.crypto import des3_encrypt, des3_decrypt
+from utils.crypto import base64_decode, base64_encode, md5
+from .model import SdoUser
+
+
+
+bp_sdo_v1 = Blueprint("sdo_v1", url_prefix="v1")
+
+
+
+@bp_sdo_v1.route("/account/active", methods=["GET", "POST"])
+async def account_active(request: Request):
+    logger.debug(request.body.decode())
+    return json({ "code": 0, "msg": "ok", "data": { "message": "ok", "result": 0 } })
+
+
+
+@bp_sdo_v1.route("/basic/publickey", methods=["GET", "POST"])
+async def basic_publickey(request: Request):
+    with open(Sanic.get_app().config["PUBKEY_PATH"], "r") as f:
+        public_key =  f.read()
+    public_key = public_key.replace("\n", "").replace("/", "\\/")
+    public_key = public_key.replace("-----BEGIN PUBLIC KEY-----", "")
+    public_key = public_key.replace("-----END PUBLIC KEY-----", "")
+    return json({ "code": 0, "msg": "", "data": { "result": 0, "message": "ok", "key": public_key, "method": "rsa" } })
+
+
+
+def encrypt_resp(resp, key):
+    return base64_encode(des3_encrypt(resp.encode(), key.encode()[:24]))
+
+@bp_sdo_v1.route("/basic/handshake", methods=["GET", "POST"])
+async def basic_handshake(request: Request):
+    with open(Sanic.get_app().config["PRIVKEY_PATH"], "r") as f:
+        private_key =  f.read()
+    data = rsa_decrypt(base64_decode(request.body.decode()), private_key)
+    logger.debug(data)
+    
+    param = dict(urllib.parse.parse_qsl(data))
+    rand_key = param["randkey"]
+    device_id = request.headers["X-DEVICEID"]
+    ctx = Sanic.get_app().ctx
+    if not hasattr(ctx, "rand_keys"): ctx.rand_keys = {}
+    ctx.rand_keys[device_id] = rand_key
+    logger.debug(f"{rand_key=} {device_id=}")
+    
+    token = uuid.uuid4().hex
+    resp = '{"message":"ok","result":0,"token":"' + token +'"}'
+    encrypted_resp = encrypt_resp(resp, rand_key)
+    logger.debug(encrypted_resp)
+    return json({ "code": 0, "msg": "ok", "data": encrypted_resp })
+
+
+
+@bp_sdo_v1.route("/account/initialize", methods=["GET", "POST"])
+async def account_initialize(request: Request):
+    device_id = request.headers["X-DEVICEID"]
+    rand_key = Sanic.get_app().ctx.rand_keys[device_id]
+    logger.debug(f"{rand_key=} {device_id=}")
+    data = des3_decrypt(base64_decode(request.body.decode()), rand_key.encode()[:24]).decode()
+    logger.debug(data)
+    initialize_resp = {
+        "brand_logo": "",
+        "brand_name": "",
+        "daoyu_clientid": "",
+        "daoyu_download_url": "",
+        "device_feature": "",
+        "display_thirdaccout": 0,
+        "force_show_agreement": 0,
+        "greport_log_level": "",
+        "guest_enable": 0,
+        "is_match": 0,
+        "log_level": "",
+        "login_button": [],
+        "login_icon": [],
+        "login_limit_enable": 0,
+        "need_float_window_permission": 0,
+        "new_device_id_server": "",
+        "qq_appId": "",
+        "qq_key": "",
+        "show_guest_confirm": 0,
+        "voicetip_button": 0,
+        "voicetip_one": "",
+        "voicetip_two": "",
+        "wegame_appid": "",
+        "wegame_appkey": "",
+        "wegame_clientid": "",
+        "wegame_companyId": "",
+        "wegame_loginUrl": "",
+        "weibo_appKey": "",
+        "weibo_redirectUrl": "",
+        "weixin_appId": "",
+        "weixin_key": "",
+    }
+    initialize_resp.update({
+		"brand_logo": "http://gskd.sdo.com/ghome/ztc/logo/og/logo_xhdpi.png",
+		"brand_name": "盛趣游戏",
+		"force_show_agreement": 1,
+		"greport_log_level": "off",
+		"log_level": "off",
+		"login_button": ["official"],
+		"login_icon": [],
+		"need_float_window_permission": 0, # 1,
+		"new_device_id_server": md5(device_id.encode()).hex(),
+		"show_guest_confirm": 1,
+		"voicetip_button": 1,
+    })
+    logger.debug(initialize_resp)
+    resp = jsonlib.dumps(initialize_resp)
+    encrypted_resp = encrypt_resp(resp, rand_key)
+    logger.debug(encrypted_resp)
+    return json({ "code": 0, "msg": "ok", "data": encrypted_resp })
+
+
+
+@bp_sdo_v1.route("/account/login", methods=["GET", "POST"])
+async def account_login(request: Request):
+    device_id = request.headers["X-DEVICEID"]
+    rand_key = Sanic.get_app().ctx.rand_keys[device_id]
+    logger.debug(f"{rand_key=} {device_id=}")
+    data = des3_decrypt(base64_decode(request.body.decode()), rand_key.encode()[:24]).decode()
+    logger.debug(data)
+    
+    param = dict(urllib.parse.parse_qsl(data))
+    phone, password = param.get("phone"), param.get("password")
+    if not phone or not password:
+        raise SanicException("Empty phone or password", status_code=403)
+    hashed_password = md5((password + "shiosalt").encode()).hex()
+    logger.debug(f"{phone=} {hashed_password=}")
+    
+    login_resp = {
+        "activation": 0,
+        "autokey": "",
+        "captchaParams": "",
+        "checkCodeGuid": "",
+        "checkCodeUrl": "",
+        "hasExtendAccs": 0,
+        "has_realInfo": 1,
+        "imagecodeType": 0,
+        "isNewUser": 0,
+        "message": "ok",
+        "nextAction": 0,
+        "prompt_msg": "",
+        "realInfoNotification": "",
+        "realInfo_force": 1,
+        "realInfo_force_pay": 0,
+        "realInfo_status": 0,
+        "realInfo_status_pay": 0,
+        "result": 0,
+        "sdg_height": 0,
+        "sdg_width": 0,
+        "ticket": "",
+        "userAttribute": "0",
+        "userid": "",
+    }
+    
+    user, created = SdoUser.get_or_create(phone=phone)
+    if not created:
+        if user.password != hashed_password:
+            return json({ "code": 31, "msg": "密码有误，请联系维护者！", "data": {} })
+        user_id = user.user_id
+    else:
+        user_id = str(int(time.time()))
+        query = SdoUser.update(user_id=user_id, password=hashed_password) \
+                       .where(SdoUser.phone == phone)
+        query.execute()
+
+    sifkey = "SIF_" + uuid.uuid4().hex
+    autokey = "AUTO_" + uuid.uuid4().hex
+    ticket = "TICKET_" + uuid.uuid4().hex
+    login_time = int(time.time())
+    query = SdoUser.update(sifkey=sifkey,
+                           autokey=autokey, 
+                           ticket=ticket, 
+                           last_login_time=login_time) \
+                   .where(SdoUser.phone == phone)
+    query.execute()
+    
+    login_resp.update({
+        "autokey": autokey,
+        "ticket": ticket,
+        "userid": user_id
+    })
+    logger.debug(login_resp)
+    resp = jsonlib.dumps(login_resp)
+    encrypted_resp = encrypt_resp(resp, rand_key)
+    logger.debug(encrypted_resp)
+    return json({ "code": 0, "msg": "ok", "data": encrypted_resp })
+
+
+
+
+@bp_sdo_v1.route("/account/loginauto", methods=["GET", "POST"])
+async def account_loginauto(request: Request):
+    device_id = request.headers["X-DEVICEID"]
+    rand_key = Sanic.get_app().ctx.rand_keys[device_id]
+    logger.debug(f"{rand_key=} {device_id=}")
+    data = des3_decrypt(base64_decode(request.body.decode()), rand_key.encode()[:24]).decode()
+    logger.debug(data)
+    
+    param = dict(urllib.parse.parse_qsl(data))
+    autokey = param.get("autokey")
+    if not autokey:
+        raise SanicException("Empty autokey", status_code=403)
+    
+    user = SdoUser.get_or_none(autokey=autokey)
+    if not user:
+        return json({ "code": 31, "msg": "账号不存在或者登陆状态已过期！", "data": {} })
+
+    user_id, ticket = user.user_id, user.ticket
+    login_auto_resp = {
+        "result": 0,
+        "message": "ok",
+        "autokey": autokey,
+        "userid": user_id,
+        "ticket": ticket
+    }
+    logger.debug(login_auto_resp)
+    resp = jsonlib.dumps(login_auto_resp)
+    encrypted_resp = encrypt_resp(resp, rand_key)
+    logger.debug(encrypted_resp)
+    return json({ "code": 0, "msg": "ok", "data": encrypted_resp })
+
+
+
+
+
+
+
+@bp_sdo_v1.route("/basic/loginarea", methods=["GET", "POST"])
+async def basic_loginarea(request: Request):
+    user_id = ""
+    if request.form: user_id = request.form["userid"][0]
+    return json({ "code": 0, "msg": "ok", "data": { "userid": user_id } })

utils/crypto.py

@@ -0,0 +1,32 @@
+import base64
+import hashlib
+from Crypto.PublicKey import RSA
+from Crypto.Cipher import PKCS1_v1_5
+from Crypto.Cipher import DES3
+from Crypto.Util.Padding import pad, unpad
+
+def md5(orig: bytes) -> bytes:
+    m = hashlib.md5()
+    m.update(orig)
+    return bytes.fromhex(m.hexdigest())
+
+
+def base64_encode(orig: bytes) -> str:
+    return base64.b64encode(orig).decode()
+
+def base64_decode(b64: str) -> bytes:
+    return base64.b64decode(b64)
+
+def rsa_decrypt(cipher_text: bytes, private_key: str) -> str:
+    cipher = PKCS1_v1_5.new(RSA.importKey(private_key)) 
+    decrypt_text = cipher.decrypt(cipher_text, b"rsa")
+    return decrypt_text.decode("utf-8")
+
+
+def des3_decrypt(cipher_text: bytes, key: bytes) -> bytes:
+    cipher = DES3.new(key, DES3.MODE_ECB)
+    return unpad(cipher.decrypt(cipher_text), 16)
+
+def des3_encrypt(origin_text: bytes, key: bytes) -> bytes:
+    cipher = DES3.new(key, DES3.MODE_ECB)
+    return cipher.encrypt(pad(origin_text, 16))