Alas/module/luahook/crack.py

import hashlib
import json
import os
import time
from pathlib import Path
from typing import Union, List, Type

import requests
from pydantic import BaseModel

from module.config.config import AzurLaneConfig
from module.config.deep import deep_get
from module.device.device import Device
from module.luahook.exception import CrackerError
from module.logger import logger

from module.luahook.api import CrackApi
from module.luahook.op import CrackOp

REMOTE_PORT = 23897


class Cracker(CrackApi):
    def __init__(self, config: AzurLaneConfig, device: Device):
        self.config = config
        self.device = device
        super().__init__(f"http://127.0.0.1:{device.adb.forward_port(REMOTE_PORT)}")


def do_crack_op_on_func(
        before_call: Union[Type[CrackOp.Op], List[Type[CrackOp.Op]]] = None,
        after_call: Union[Type[CrackOp.Op], List[Type[CrackOp.Op]]] = None
):
    def wrapper(func):
        def inner(*args, **kwargs):
            obj = args[0]
            if before_call is not None:
                do_crack_op(obj.config, obj.device, before_call)
            try:
                ret = func(*args, **kwargs)
            except Exception as e:
                if after_call is not None:
                    do_crack_op(obj.config, obj.device, after_call)
                raise e
            if after_call is not None:
                do_crack_op(obj.config, obj.device, after_call)
            return ret

        return inner

    return wrapper


g_cracker_has_inited = False


def do_crack_op(config: AzurLaneConfig, device: Device, ops: Union[Type[CrackOp.Op], List[Type[CrackOp.Op]]]):
    if not config.full_config.Hook_HookGeneral_Enable:
        return
    if not device.app_is_running():
        logger.info("Game not running, do not crack")
        return
    if isinstance(ops, list):
        l = ops
    else:
        if ops == CrackOp.EnableAll:
            l = CrackOp.ALL_ENABLE_OPS
        else:
            l = [ops]

    full_config = config.full_config

    base_url = f"http://127.0.0.1:{device.adb.forward_port(REMOTE_PORT)}"
    timeout = full_config.Hook_HookGeneral_RequestTimeLimit
    api = CrackApi(base_url, timeout=timeout)

    without_pause = full_config.Hook_HookGeneral_OperateWithoutPause

    global g_cracker_has_inited
    if not g_cracker_has_inited:
        if without_pause:
            api.init_without_pause()
        else:
            api.init()
        g_cracker_has_inited = True

    for op in l:
        if issubclass(op, CrackOp.Op):
            obj = op(full_config, api)
            if without_pause:
                obj.execute_without_pause()
            else:
                obj.execute_with_pause()
        else:
            logger.error(f"Unsupported op: {op}")


crack_op = do_crack_op_on_func


def disable_all_crack(f):
    def wrapper(*args, **kwargs):
        obj = args[0]
        logger.info("Disabe all luahook cracks")
        do_crack_op(obj.config, obj.device, CrackOp.DisableAll)
        return f(*args, **kwargs)

    return wrapper


def enable_all_crack(f):
    def wrapper(*args, **kwargs):
        obj = args[0]
        do_crack_op(obj.config, obj.device, CrackOp.EnableAll)
        return f(*args, **kwargs)

    return wrapper


def luahook_crack_all(config: AzurLaneConfig, device: Device):
    logger.info("Crack all with luahook")
    do_crack_op(config, device, CrackOp.EnableAll)


def luahook_disable_all(config: AzurLaneConfig, device: Device):
    logger.info("Disable all luahook")
    do_crack_op(config, device, CrackOp.DisableAll)


def chapter_task_crack(f):
    def wrapper(*args, **kwargs):
        obj = args[0]
        do_crack_op(obj.config, obj.device, CrackOp.CHAPTER_CRACK_OPS)
        return f(*args, **kwargs)

    return wrapper


def opsi_task_crack(f):
    def wrapper(*args, **kwargs):
        obj = args[0]
        do_crack_op(obj.config, obj.device, CrackOp.OPSI_CRACK_OPS)
        return f(*args, **kwargs)

    return wrapper


class UpdateServerApi:
    class ResourceFile(BaseModel):
        arch: str
        file: str

    def __init__(self, api_url: str = ""):
        self.api_url = api_url

    def post(self, path: str, data=None):
        logger.info(f"UpdateServerApi post: {path}")
        url = f'{self.api_url}/{path}'
        try:
            response = requests.post(url, data=data)  # TODO: add timeout
        except requests.exceptions.Timeout:
            raise CrackerError('UpdateServerApi request timeout')
        except Exception as e:
            raise CrackerError(f'UpdateServerApi request error: {e}')
        if response.status_code != 200:
            raise CrackerError(f'UpdateServerApi response error: {response.status_code}')
        return response

    def download_file(self, arch: str, file_name: str) -> bytes:
        res = self.post("files", UpdateServerApi.ResourceFile(arch=arch, file=file_name).json())
        return res.content

    def get_hash(self, arch: str, file_name: str) -> str:
        res = self.post("get_hash", UpdateServerApi.ResourceFile(arch=arch, file=file_name).json())
        return res.text


class CrackResource:
    ARCH_MAP = {
        "x86": "x86",
        "x86_64": "x86_64",
        "armv7": "armeabi-v7a",
        "arm": "armeabi-v7a",
        "arm64": "arm64-v8a",
        "aarch64": "arm64-v8a",
        "aarch": "armeabi-v7a",
    }

    def __init__(self, config: AzurLaneConfig, device: Device):
        self.config = config
        self.device = device

        arch_conf = config.full_config.Hook_HookGeneral_Architecture

        if arch_conf == "auto":
            logger.warning("Please Tools-HookOp to auto detect, use default x86_64")
            config.full_config.Hook_HookGeneral_Architecture = "x86_64"
            self.arch = "x86_64"
        else:
            self.arch = arch_conf
            logger.info(f"Use arch: {self.arch}")

        self.resource_root = f"./bin/hook"
        self.resource_dir = f"{self.resource_root}/{self.arch}"
        self.version_file = f"{self.resource_dir}/version.json"

        self.first_init = False
        self.has_new_version = False

        self.update_server = self.__get_update_server()

        self.upload_dir = f"/data/hook/{self.arch}"

        lib_dir = config.full_config.Hook_HookGeneral_GameLibDir
        if lib_dir:
            self.game_lib_dir = Path(self.__format_str(lib_dir)).as_posix()
        else:
            self.game_lib_dir = None

    def __file_hash(self, file_name):
        with open(file_name, "rb") as f:
            return hashlib.md5(f.read()).hexdigest()

    def __gen_version_file_from_local(self):
        d = {}
        for file_name in os.listdir(self.resource_dir):
            f = f"{self.resource_dir}/{file_name}"
            if not os.path.isfile(f):
                continue
            if file_name.endswith(".json"):
                continue

            hash_value = self.__file_hash(f).upper()
            d[file_name] = hash_value
        with open(self.version_file, "w") as f:
            json.dump(d, f, indent=4)

    def __ensure_local_resource(self):
        if self.update_server is None:
            logger.info("No update server, skip ensure local resource")
            return

        if not os.path.exists(f"{self.resource_dir}/libcracker.so"):
            self.__download_resource("libcracker.so")
            self.first_init = True

        if not os.path.exists(f"{self.resource_dir}/patchelf"):
            self.__download_resource("patchelf")
            self.first_init = True

    __update_local_version = __gen_version_file_from_local

    def __ensure_crack_resource(self, ):
        if not os.path.exists(self.resource_root):
            os.makedirs(self.resource_root)
        if not os.path.exists(self.resource_dir):
            os.makedirs(self.resource_dir)
        self.__ensure_local_resource()
        if not os.path.exists(self.version_file):
            self.__gen_version_file_from_local()

    def __download_resource(self, file_name: str):
        if self.update_server is None:
            logger.info("No update server, skip download resource")
            return
        logger.info(f"Download resource: {self.arch}/{file_name}")
        res = self.update_server.download_file(self.arch, file_name)
        with open(f"{self.resource_dir}/{file_name}", "wb") as f:
            f.write(res)

    def __is_same_version(self, version_json: dict, file_name: str):
        h = self.update_server.get_hash(self.arch, file_name)
        return h == version_json.get(file_name, "")

    def __check_update(self):
        if self.update_server is None:
            logger.info("No update server, skip update check, use local resource")
            return
        if not self.config.full_config.Hook_HookGeneral_UpdateEveryTime:
            logger.info("Update skip, use local resource")
            return
        with open(self.version_file, "r") as f:
            local_version = json.load(f)

        if not self.__is_same_version(local_version, "libcracker.so"):
            self.__download_resource("libcracker.so")
            self.has_new_version = True
        if not self.__is_same_version(local_version, "patchelf"):
            self.__download_resource("patchelf")
            self.has_new_version = True

        if self.has_new_version:
            self.__update_local_version()

    def __format_str(self, s: str) -> str:
        return ("".join([i for i in s if i.isprintable()])).replace(" ", "").replace("\n", "")

    def __get_update_server(self) -> Union[UpdateServerApi, None]:
        update_server_url: str = self.config.full_config.Hook_HookGeneral_UpdateServer
        update_server_url = self.__format_str(update_server_url)
        if not update_server_url:
            return None
        return UpdateServerApi(update_server_url)

    def __push(self, file_name: str):
        self.device.adb_shell(f"mkdir -p {self.upload_dir}")
        self.device.adb_shell(f"rm {self.upload_dir}/{file_name}")
        self.device.adb_push(f"{self.resource_dir}/{file_name}", f"{self.upload_dir}/{file_name}")
        self.device.adb_shell(f"chmod 777 {self.upload_dir}/{file_name}")

    def __push_resource(self):
        logger.info("Push resource to device")
        self.__push("libcracker.so")
        self.__push("patchelf")

    def __do_inject(self):
        if not self.game_lib_dir:
            raise CrackerError("GameLibDir not set")

        logger.info(f"GameLibDir: {self.game_lib_dir}, do inject")
        cmd = f"cd {self.upload_dir} && ./patchelf --local-patch --game-lib-dir '{self.game_lib_dir}'"
        self.device.adb_shell(cmd)

    def __is_exist(self, path):
        cmd = f"if [ -f {path} ]; then echo 'exist'; else if [ -d {path} ]; then echo 'exist'; else echo 'not exist'; fi; fi"
        return self.device.adb_shell(cmd) == "exist"

    def __is_remote_file_exist(self) -> bool:
        return self.__is_exist(self.upload_dir) and self.__is_exist(
            f"{self.upload_dir}/libcracker.so") and self.__is_exist(f"{self.upload_dir}/patchelf")

    def __adb_su_do(self, cmd: str):
        return self.device.adb_shell(f"su -c '{cmd}'")

    def __adb_root(self):
        self.device.adb_command(["root"])

    def __check_game_lib_dir(self):
        return self.__is_exist(f"{self.game_lib_dir}/libil2cpp.so") and self.__is_exist(
            f"{self.game_lib_dir}/libtolua.so")

    def auto_detect(self):
        if not self.device.app_is_running():
            self.device.app_start()
            time.sleep(5)
        self.__adb_root()
        pid = self.device.adb_shell(f"pidof {self.device.package}")
        maps = self.device.adb_shell(f"cat /proc/{pid}/maps")

        image_list = maps.splitlines()

        libtolua_path = None
        for image in image_list:
            if image.find("libtolua.so") != -1:
                libtolua_path = image.split(" ")[-1]
                break
        if libtolua_path is None:
            logger.error("Cannot find libtolua.so path")
            return

        game_lib_path = libtolua_path.replace("/libtolua.so", "")
        arch = CrackResource.ARCH_MAP[game_lib_path.rsplit("/", maxsplit=1)[-1]]

        logger.info(f"Game lib path: {game_lib_path}")
        logger.info(f"Game arch: {arch}")
        full_config = self.config.full_config
        full_config.Hook_HookGeneral_Architecture = arch
        full_config.Hook_HookGeneral_GameLibDir = game_lib_path

    def ensure(self):
        if not self.config.full_config.Hook_HookGeneral_Enable:
            return

        if not self.__check_game_lib_dir():
            raise CrackerError(f"GameLibDir '{self.game_lib_dir}' is invalid")

        self.__adb_root()
        self.__ensure_crack_resource()
        if not self.first_init:
            self.__check_update()
        if self.has_new_version or not self.__is_remote_file_exist() or self.config.full_config.Hook_HookGeneral_PushEveryTime:
            self.__push_resource()
            self.device.app_stop()
            self.config.task_call("Restart")
            self.__do_inject()