import hashlib import json import os 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.exception import CrackerError from module.logger import logger from module.luahook.api import CrackApi from module.luahook.op import CrackOp ALL_ENABLE_OPS = [ CrackOp.EnableGlobalShipProperties, CrackOp.EnableChapterFastMove, CrackOp.EnableRemoveHardModeLimit, CrackOp.EnableFakePlayer, CrackOp.EnableNoBBAnimation, CrackOp.EnableNoEmotionWarning, CrackOp.EnableOpsiFastMove, CrackOp.EnableRemoveHardModeShipTypeLimit, CrackOp.EnableRemoveHardModeShipPropertiesLimit, CrackOp.EnableGGFactor, CrackOp.EnableGlobalSpeedup, CrackOp.EnableExerciseGodMod, CrackOp.EnableExerciseMorePower, ] 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 fix_Hook_ShipProperty_config(config: AzurLaneConfig): enable = deep_get(config.data, "Hook.ShipProperty.Enable", False) if not enable: return else: config.modified["Hook.ShipProperty.Method"] = "final_properties" 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 def do_crack_op(config: AzurLaneConfig, device: Device, ops: Union[Type[CrackOp.Op], List[Type[CrackOp.Op]]]): if not deep_get(config.data, "Hook.HookGeneral.Enable", False): return if isinstance(ops, list): l = ops else: if ops == CrackOp.EnableAll: l = ALL_ENABLE_OPS else: l = [ops] base_url = f"http://127.0.0.1:{device.adb.forward_port(REMOTE_PORT)}" timeout = deep_get(config.data, "Hook.HookGeneral.RequestTimeLimit", 10) api = CrackApi(base_url, timeout=timeout) for op in l: if op == CrackOp.DisableAll: api.disable_all() elif op == CrackOp.EnableGlobalShipProperties: fix_Hook_ShipProperty_config(config) if deep_get(config.data, "Hook.ShipProperty.Method", "disable") != "final_properties": continue api.update_global_ship_properties( CrackApi.ShipProperties( armor=float(deep_get(config.data, "Hook.ShipProperty.Armor", -1)), speed=float(deep_get(config.data, "Hook.ShipProperty.Speed", -1)), antiaircraft=float(deep_get(config.data, "Hook.ShipProperty.AntiAircraft", -1)), oxy_recovery_bench=float(deep_get(config.data, "Hook.ShipProperty.OxyRecoveryBench", -1)), torpedo=float(deep_get(config.data, "Hook.ShipProperty.Torpedo", -1)), hit=float(deep_get(config.data, "Hook.ShipProperty.Hit", -1)), sonarRange=float(deep_get(config.data, "Hook.ShipProperty.SonarRange", -1)), attack_duration=float(deep_get(config.data, "Hook.ShipProperty.AttackDuration", -1)), raid_distance=float(deep_get(config.data, "Hook.ShipProperty.RaidDistance", -1)), oxy_recovery_surface=float(deep_get(config.data, "Hook.ShipProperty.OxyRecoverySurface", -1)), oxy_recovery=float(deep_get(config.data, "Hook.ShipProperty.OxyRecovery", -1)), dodge=float(deep_get(config.data, "Hook.ShipProperty.Dodge", -1)), luck=float(deep_get(config.data, "Hook.ShipProperty.Luck", -1)), reload=float(deep_get(config.data, "Hook.ShipProperty.Reload", -1)), oxy_cost=float(deep_get(config.data, "Hook.ShipProperty.OxyCost", -1)), durability=float(deep_get(config.data, "Hook.ShipProperty.Durability", -1)), air=float(deep_get(config.data, "Hook.ShipProperty.Air", -1)), oxy_max=float(deep_get(config.data, "Hook.ShipProperty.OxyMax", -1)), cannon=float(deep_get(config.data, "Hook.ShipProperty.Cannon", -1)), antisub=float(deep_get(config.data, "Hook.ShipProperty.AntiSub", -1)), ) ) api.enable_global_ship_properties_crack() elif op == CrackOp.DisableGlobalShipProperties: api.disable_global_ship_properties_crack() elif op == CrackOp.EnableChapterFastMove: if not deep_get(config.data, "Hook.Misc.ChapterMove", False): continue api.enable_fast_stage_move() elif op == CrackOp.DisableChapterFastMove: api.disable_fast_stage_move() elif op == CrackOp.EnableRemoveHardModeShipPropertiesLimit: if deep_get(config.data, "Hook.Misc.RemoveHardMapLimit", "") != "remove_ship_properties_limit": continue api.enable_remove_hard_mode_ship_properties_limit() elif op == CrackOp.DisableRemoveHardModeShipPropertiesLimit: api.disable_remove_hard_mode_ship_properties_limit() elif op == CrackOp.EnableRemoveHardModeShipTypeLimit: if deep_get(config.data, "Hook.Misc.RemoveHardMapLimit", "") != "remove_ship_type_limit": continue api.enable_remove_hard_mode_ship_type_limit() elif op == CrackOp.DisableRemoveHardModeShipTypeLimit: api.disable_remove_hard_mode_ship_type_limit() elif op == CrackOp.EnableRemoveHardModeLimit: if deep_get(config.data, "Hook.Misc.RemoveHardMapLimit", "") != "remove_both": continue api.enable_remove_hard_mode_limit() elif op == CrackOp.DisableRemoveHardModeLimit: api.disable_remove_hard_mode_limit() elif op == CrackOp.EnableFakePlayer: if not deep_get(config.data, "Hook.FakePlayer.Enable", False): continue api.update_fake_player_info( CrackApi.FakePlayerInfo( name=str(deep_get(config.data, "Hook.FakePlayer.Name", "")), level=str(deep_get(config.data, "Hook.FakePlayer.Level", "")), id=str(deep_get(config.data, "Hook.FakePlayer.Id", "")), ) ) api.enable_fake_player() elif op == CrackOp.DisableFakePlayer: api.disable_fake_player() elif op == CrackOp.EnableNoBBAnimation: if not deep_get(config.data, "Hook.Misc.NoBBAnimation", False): continue api.enable_no_bb_animation() elif op == CrackOp.DisableNoBBAnimation: api.disable_no_bb_animation() elif op == CrackOp.EnableNoEmotionWarning: if not deep_get(config.data, "Hook.Misc.NoEmotionWarning", False): continue api.enable_no_emotion_warning() elif op == CrackOp.DisableNoEmotionWarning: api.disable_no_emotion_warning() elif op == CrackOp.IsAlive: api.is_alive() elif op == CrackOp.EnableOpsiFastMove: api.enable_opsi_fast_move() elif op == CrackOp.DisableOpsiFastMove: api.disable_opsi_fast_move() elif op == CrackOp.EnableGGFactor: if deep_get(config.data, "Hook.ShipProperty.Method", "disable") != "gg_factor": continue api.update_gg_factor(float(deep_get(config.data, "Hook.ShipProperty.Factor", 1.0))) api.enable_gg_factor() elif op == CrackOp.DisableGGFactor: api.disable_gg_factor() elif op == CrackOp.EnableGlobalSpeedup: rate = float(deep_get(config.data, "Hook.Misc.GlobalSpeedup", 1.0)) if rate == 1.0: continue api.update_global_speedup_rate(CrackApi.GlobalSpeedupRate(rate=rate)) api.enable_global_speedup() elif op == CrackOp.DisableGlobalSpeedup: api.disable_global_speedup() elif op == CrackOp.EnableExerciseGodMod: if deep_get(config.data, "Hook.Misc.ExerciseGodMod", False): api.enable_exercise_god_mode() elif op == CrackOp.DisableExerciseGodMod: api.disable_exercise_god_mode() elif op == CrackOp.EnableExerciseMorePower: rate = float(deep_get(config.data, "Hook.Misc.ExerciseMorePower", -1.0)) if rate == -1.0: continue api.update_exercise_more_power_rate(CrackApi.ExerciseMorePowerRate(rate=rate)) api.enable_exercise_more_power() elif op == CrackOp.DisableExerciseMorePower: api.disable_exercise_more_power() 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) CHAPTER_CRACK_OPS = [ CrackOp.EnableChapterFastMove, CrackOp.EnableNoBBAnimation, CrackOp.EnableNoEmotionWarning, CrackOp.EnableGlobalShipProperties, CrackOp.EnableRemoveHardModeLimit, CrackOp.EnableFakePlayer, CrackOp.EnableGGFactor, CrackOp.EnableGlobalSpeedup, ] def chapter_task_crack(f): def wrapper(*args, **kwargs): obj = args[0] do_crack_op(obj.config, obj.device, CHAPTER_CRACK_OPS) return f(*args, **kwargs) return wrapper OPSI_CRACK_OPS = [ CrackOp.EnableNoBBAnimation, CrackOp.EnableOpsiFastMove, CrackOp.EnableGlobalShipProperties, CrackOp.EnableFakePlayer, CrackOp.EnableGGFactor, CrackOp.EnableGlobalSpeedup, ] def opsi_task_crack(f): def wrapper(*args, **kwargs): obj = args[0] do_crack_op(obj.config, obj.device, 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", "armv7": "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 = deep_get(config.data, "Hook.HookGeneral.Architecture") if arch_conf == "auto": arch_ret = device.adb_shell("uname -m").lower() self.arch = self.ARCH_MAP.get(arch_ret, "") if not self.arch: raise CrackerError(f"Unsupported arch: {arch_ret}") logger.info(f"Arch: {arch_ret} -> {self.arch}") 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}" self.game_lib_dir = Path(self.__format_str(deep_get(self.config.data, "Hook.HookGeneral.GameLibDir", ""))).as_posix() 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 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 = deep_get(self.config.data, "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}", timeout=300) 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 ensure(self): if not deep_get(self.config.data, "Hook.HookGeneral.Enable", False): 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 deep_get(self.config.data, "Hook.HookGeneral.PushEveryTime", True): self.__push_resource() self.device.app_stop() self.config.task_call("Restart") self.__do_inject()