import logging from unrealircd_rpc_py.objects.Definition import LiveRPCResult from core.classes.interfaces.imodule import IModule import mods.jsonrpc.utils as utils import mods.jsonrpc.threads as thds from dataclasses import dataclass from unrealircd_rpc_py.ConnectionFactory import ConnectionFactory from unrealircd_rpc_py.LiveConnectionFactory import LiveConnectionFactory class Jsonrpc(IModule): @dataclass class ModConfModel: """The Model containing the module parameters """ jsonrpc: int = 0 MOD_HEADER: dict[str, str] = { 'name':'JsonRPC', 'version':'1.0.0', 'description':'Module using the unrealircd-rpc-py library', 'author':'Defender Team', 'core_version':'Defender-6' } def callback_sent_to_irc(self, response: LiveRPCResult) -> None: dnickname = self.Config.SERVICE_NICKNAME dchanlog = self.Config.SERVICE_CHANLOG green = self.Config.COLORS.green nogc = self.Config.COLORS.nogc bold = self.Config.COLORS.bold red = self.Config.COLORS.red if response.error.code != 0: self.Protocol.send_priv_msg(nick_from=dnickname, msg=f"[{bold}{red}JSONRPC ERROR{nogc}{bold}] {response.error.message} ({response.error.code})", channel=dchanlog) return None if isinstance(response.result, bool): if response.result: self.Protocol.send_priv_msg( nick_from=self.Config.SERVICE_NICKNAME, msg=f"[{bold}{green}JSONRPC{nogc}{bold}] JSONRPC Event activated on {self.Config.JSONRPC_URL}", channel=dchanlog) return None level = response.result.level if hasattr(response.result, 'level') else '' subsystem = response.result.subsystem if hasattr(response.result, 'subsystem') else '' event_id = response.result.event_id if hasattr(response.result, 'event_id') else '' log_source = response.result.log_source if hasattr(response.result, 'log_source') else '' msg = response.result.msg if hasattr(response.result, 'msg') else '' build_msg = f"{green}{log_source}{nogc}: [{bold}{level}{bold}] {subsystem}.{event_id} - {msg}" self.Protocol.send_priv_msg(nick_from=dnickname, msg=build_msg, channel=dchanlog) return None def create_tables(self) -> None: return None def load(self) -> None: logging.getLogger('websockets').setLevel(logging.WARNING) logging.getLogger('unrealircd-rpc-py').setLevel(logging.CRITICAL) logging.getLogger('unrealircd-liverpc-py').setLevel(logging.CRITICAL) self.ModConfig = self.ModConfModel(jsonrpc=0) if self.Config.SERVEUR_PROTOCOL != 'unreal6': self.Loader.ModuleUtils.unload_one_module(self.Irc, self.module_name, False) return None # Is RPC Active? self.is_streaming = False # Module Utils self.Utils = utils # Module threads self.Threads = thds # Run Garbage collector. self.Base.create_timer(10, self.MainUtils.run_python_garbage_collector) # Create module commands (Mandatory) self.Irc.build_command(1, self.module_name, 'jsonrpc', 'Activate the JSON RPC Live connection [ON|OFF]') self.Irc.build_command(1, self.module_name, 'jruser', 'Get Information about a user using JSON RPC') self.Irc.build_command(1, self.module_name, 'jrinstances', 'Get number of instances') try: self.Rpc = ConnectionFactory(self.Config.DEBUG_LEVEL).get(self.Config.JSONRPC_METHOD) self.LiveRpc = LiveConnectionFactory(self.Config.DEBUG_LEVEL).get(self.Config.JSONRPC_METHOD) sync_unixsocket = {'path_to_socket_file': self.Config.JSONRPC_PATH_TO_SOCKET_FILE} sync_http = {'url': self.Config.JSONRPC_URL, 'username': self.Config.JSONRPC_USER, 'password': self.Config.JSONRPC_PASSWORD} live_unixsocket = {'path_to_socket_file': self.Config.JSONRPC_PATH_TO_SOCKET_FILE, 'callback_object_instance' : self, 'callback_method_or_function_name': 'callback_sent_to_irc'} live_http = {'url': self.Config.JSONRPC_URL, 'username': self.Config.JSONRPC_USER, 'password': self.Config.JSONRPC_PASSWORD, 'callback_object_instance' : self, 'callback_method_or_function_name': 'callback_sent_to_irc'} sync_param = sync_unixsocket if self.Config.JSONRPC_METHOD == 'unixsocket' else sync_http live_param = live_unixsocket if self.Config.JSONRPC_METHOD == 'unixsocket' else live_http self.Rpc.setup(sync_param) self.LiveRpc.setup(live_param) if self.ModConfig.jsonrpc == 1: self.Base.create_thread(func=self.Threads.thread_subscribe, func_args=(self, ), run_once=True) return None except Exception as err: self.Protocol.send_priv_msg( nick_from=self.Config.SERVICE_NICKNAME, msg=f"[{self.Config.COLORS.red}JSONRPC ERROR{self.Config.COLORS.nogc}] {err.__str__()}", channel=self.Config.SERVICE_CHANLOG ) self.Logs.error(f"JSONRPC ERROR: {err.__str__()}") def unload(self) -> None: if self.Config.SERVEUR_PROTOCOL != 'unreal6': self.Loader.ModuleUtils.unload_one_module(self.Irc, self.module_name, False) return None if self.is_streaming: self.Protocol.send_priv_msg( nick_from=self.Config.SERVICE_NICKNAME, msg=f"[{self.Config.COLORS.green}JSONRPC INFO{self.Config.COLORS.nogc}] Shutting down RPC system!", channel=self.Config.SERVICE_CHANLOG ) self.Base.create_thread(func=self.Threads.thread_unsubscribe, func_args=(self, ), run_once=True) self.update_configuration('jsonrpc', 0) self.Irc.Commands.drop_command_by_module(self.module_name) self.Logs.debug(f"Unloading {self.module_name}") return None def cmd(self, data: list) -> None: return None def hcmds(self, user:str, channel: any, cmd: list, fullcmd: list = []) -> None: command = str(cmd[0]).lower() dnickname = self.Config.SERVICE_NICKNAME dchannel = self.Config.SERVICE_CHANLOG fromuser = user fromchannel = str(channel) if not channel is None else None match command: case 'jsonrpc': try: if len(cmd) < 2: self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'/msg {dnickname} jsonrpc on') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'/msg {dnickname} jsonrpc off') return None option = str(cmd[1]).lower() match option: case 'on': thread_name = 'thread_subscribe' if self.Base.is_thread_alive(thread_name): self.Protocol.send_priv_msg(nick_from=dnickname, channel=dchannel, msg=f"The Subscription is running") return None elif self.Base.is_thread_exist(thread_name): self.Protocol.send_priv_msg( nick_from=dnickname, channel=dchannel, msg=f"The subscription is not running, wait untill the process will be cleaned up" ) return None self.Base.create_thread(func=self.Threads.thread_subscribe, func_args=(self, ), run_once=True) self.update_configuration('jsonrpc', 1) case 'off': self.Base.create_thread(func=self.Threads.thread_unsubscribe, func_args=(self, ), run_once=True) self.update_configuration('jsonrpc', 0) except IndexError as ie: self.Logs.error(ie) case 'jruser': try: if len(cmd) < 2: self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'/msg {dnickname} jruser get nickname') option = str(cmd[1]).lower() match option: case 'get': nickname = str(cmd[2]) rpc = self.Rpc UserInfo = rpc.User.get(nickname) if UserInfo.error.code != 0: self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'{UserInfo.error.message}') return None self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'UID : {UserInfo.id}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'NICKNAME : {UserInfo.name}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'USERNAME : {UserInfo.user.username}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'REALNAME : {UserInfo.user.realname}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'MODES : {UserInfo.user.modes}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'CHANNELS : {[chan.name for chan in UserInfo.user.channels]}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'SECURITY GROUP : {UserInfo.user.security_groups}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'REPUTATION : {UserInfo.user.reputation}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'IP : {UserInfo.ip}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'COUNTRY CODE : {UserInfo.geoip.country_code}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'ASN : {UserInfo.geoip.asn}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'ASNAME : {UserInfo.geoip.asname}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'CLOAKED HOST : {UserInfo.user.cloakedhost}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'HOSTNAME : {UserInfo.hostname}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'VHOST : {UserInfo.user.vhost}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'CLIENT PORT : {UserInfo.client_port}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'SERVER PORT : {UserInfo.server_port}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'CERTFP : {UserInfo.tls.certfp}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'CIPHER : {UserInfo.tls.cipher}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'IDLE SINCE : {UserInfo.idle_since}') self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f'CONNECTED SINCE : {UserInfo.connected_since}') except IndexError as ie: self.Logs.error(ie) case 'jrinstances': try: self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"GC Collect: {self.MainUtils.run_python_garbage_collector()}") self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Nombre d'instance LiveWebsock: {self.MainUtils.get_number_gc_objects(LiveConnectionFactory)}") self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Nombre d'instance ConnectionFactory: {self.MainUtils.get_number_gc_objects(ConnectionFactory)}") self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Nombre de toute les instances: {self.MainUtils.get_number_gc_objects()}") except Exception as err: self.Logs.error(f"Unknown Error: {err}")