import logging from Crypto.Cipher import AES import struct from typing import Dict, Optional, List, Type import random import asyncio from core.config import CoreConfig from .database import IDZData from .config import IDZConfig from .const import IDZConstants from .handlers import IDZHandlerBase HANDLER_MAP: List[Dict] class IDZKey: def __init__(self, n, d, e, hashN: int) -> None: self.N = n self.d = d self.e = e self.hashN = hashN class IDZUserDB: def __init__( self, core_cfg: CoreConfig, game_cfg: IDZConfig, keys: List[IDZKey], handlers: List[Dict], ) -> None: self.logger = logging.getLogger("idz") self.core_config = core_cfg self.game_config = game_cfg self.rsa_keys = keys self.handlers = handlers self.static_key = bytes.fromhex(self.game_config.server.aes_key) self.version = None self.version_internal = None self.skip_next = False def start(self) -> None: self.logger.info(f"Start on port {self.game_config.ports.userdb}") asyncio.create_task(asyncio.start_server(self.connection_cb, self.core_config.server.listen_address, self.game_config.ports.userdb)) def append_padding(self, data: bytes): """Appends 0s to the end of the data until it's at the correct size""" length = struct.unpack_from(" None: self.logger.debug(f"Receive data {data.hex()}") client_ip = writer.get_extra_info('peername')[0] crypt = AES.new(self.static_key, AES.MODE_ECB) try: data_dec = crypt.decrypt(data) except Exception as e: self.logger.error(f"Failed to decrypt UserDB request from {client_ip} because {e} - {data.hex()}") self.logger.debug(f"Decrypt data {data_dec.hex()}") magic = struct.unpack_from("