diff --git a/core/classes/protocols/interface.py b/core/classes/protocols/interface.py index dc3546d..e5ab6b6 100644 --- a/core/classes/protocols/interface.py +++ b/core/classes/protocols/interface.py @@ -4,12 +4,38 @@ from core.classes.protocols.command_handler import CommandHandler if TYPE_CHECKING: from core.definition import MClient, MSasl - + from core.irc import Irc class IProtocol(ABC): Handler: Optional[CommandHandler] = None + def __init__(self, uplink: 'Irc'): + self.name: Optional[str] = None + self.protocol_version: int = -1 + self.known_protocol: set[str] = {} + + self._Irc = uplink + self._Config = uplink.Config + self._Base = uplink.Base + self._Settings = uplink.Base.Settings + self._Utils = uplink.Loader.Utils + self._Logs = uplink.Loader.Logs + + self.Handler = CommandHandler(uplink.Loader) + + self.init_protocol() + + self._Logs.info(f"[PROTOCOL] Protocol [{self.__class__.__name__}] loaded!") + + @abstractmethod + def init_protocol(self): + """_summary_ + + Returns: + _type_: _description_ + """ + @abstractmethod def get_ircd_protocol_poisition(self, cmd: list[str], log: bool = False) -> tuple[int, Optional[str]]: """Get the position of known commands diff --git a/core/classes/protocols/unreal6.py b/core/classes/protocols/unreal6.py index 66933de..f68a817 100644 --- a/core/classes/protocols/unreal6.py +++ b/core/classes/protocols/unreal6.py @@ -3,29 +3,18 @@ from re import match, findall, search from datetime import datetime from typing import TYPE_CHECKING, Any, Optional from ssl import SSLEOFError, SSLError - -from core.classes.protocols.command_handler import CommandHandler from core.classes.protocols.interface import IProtocol from core.utils import tr if TYPE_CHECKING: - from core.irc import Irc from core.classes.sasl import Sasl from core.definition import MClient, MSasl class Unrealircd6(IProtocol): - def __init__(self, ircInstance: 'Irc'): + def init_protocol(self) -> None: self.name = 'UnrealIRCD-6' self.protocol_version = 6100 - - self.__Irc = ircInstance - self.__Config = ircInstance.Config - self.__Base = ircInstance.Base - self.__Settings = ircInstance.Base.Settings - self.__Utils = ircInstance.Loader.Utils - self.__Logs = ircInstance.Loader.Logs - self.known_protocol: set[str] = {'SJOIN', 'UID', 'MD', 'QUIT', 'SQUIT', 'EOS', 'PRIVMSG', 'MODE', 'UMODE2', 'VERSION', 'REPUTATION', 'SVS2MODE', @@ -33,9 +22,6 @@ class Unrealircd6(IProtocol): 'PROTOCTL', 'SERVER', 'SMOD', 'TKL', 'NETINFO', '006', '007', '018'} - self.Handler = CommandHandler(ircInstance.Loader) - - self.__Logs.info(f"[PROTOCOL] Protocol [{__name__}] loaded!") def get_ircd_protocol_poisition(self, cmd: list[str], log: bool = False) -> tuple[int, Optional[str]]: """Get the position of known commands @@ -52,12 +38,12 @@ class Unrealircd6(IProtocol): return index, token.upper() if log: - self.__Logs.debug(f"[IRCD LOGS] You need to handle this response: {cmd}") + self._Logs.debug(f"[IRCD LOGS] You need to handle this response: {cmd}") return (-1, None) def register_command(self) -> None: - m = self.__Irc.Loader.Definition.MIrcdCommand + m = self._Irc.Loader.Definition.MIrcdCommand self.Handler.register(m(command_name="PING", func=self.on_server_ping)) self.Handler.register(m(command_name="UID", func=self.on_uid)) self.Handler.register(m(command_name="QUIT", func=self.on_quit)) @@ -121,27 +107,27 @@ class Unrealircd6(IProtocol): string (Str): contient la commande à envoyer au serveur. """ try: - with self.__Base.lock: - self.__Irc.IrcSocket.send(f"{message}\r\n".encode(self.__Config.SERVEUR_CHARSET[0])) + with self._Base.lock: + self._Irc.IrcSocket.send(f"{message}\r\n".encode(self._Config.SERVEUR_CHARSET[0])) if print_log: - self.__Logs.debug(f'<< {message}') + self._Logs.debug(f'<< {message}') except UnicodeDecodeError as ude: - self.__Logs.error(f'Decode Error try iso-8859-1 - {ude} - {message}') - self.__Irc.IrcSocket.send(f"{message}\r\n".encode(self.__Config.SERVEUR_CHARSET[1],'replace')) + self._Logs.error(f'Decode Error try iso-8859-1 - {ude} - {message}') + self._Irc.IrcSocket.send(f"{message}\r\n".encode(self._Config.SERVEUR_CHARSET[1],'replace')) except UnicodeEncodeError as uee: - self.__Logs.error(f'Encode Error try iso-8859-1 - {uee} - {message}') - self.__Irc.IrcSocket.send(f"{message}\r\n".encode(self.__Config.SERVEUR_CHARSET[1],'replace')) + self._Logs.error(f'Encode Error try iso-8859-1 - {uee} - {message}') + self._Irc.IrcSocket.send(f"{message}\r\n".encode(self._Config.SERVEUR_CHARSET[1],'replace')) except AssertionError as ae: - self.__Logs.warning(f'Assertion Error {ae} - message: {message}') + self._Logs.warning(f'Assertion Error {ae} - message: {message}') except SSLEOFError as soe: - self.__Logs.error(f"SSLEOFError: {soe} - {message}") + self._Logs.error(f"SSLEOFError: {soe} - {message}") except SSLError as se: - self.__Logs.error(f"SSLError: {se} - {message}") + self._Logs.error(f"SSLError: {se} - {message}") except OSError as oe: - self.__Logs.error(f"OSError: {oe} - {message}") + self._Logs.error(f"OSError: {oe} - {message}") except AttributeError as ae: - self.__Logs.critical(f"Attribute Error: {ae}") + self._Logs.critical(f"Attribute Error: {ae}") def send_priv_msg(self, nick_from: str, msg: str, channel: str = None, nick_to: str = None): """Sending PRIVMSG to a channel or to a nickname by batches @@ -153,12 +139,12 @@ class Unrealircd6(IProtocol): nick_to (str, optional): The reciever nickname. Defaults to None. """ try: - batch_size = self.__Config.BATCH_SIZE - User_from = self.__Irc.User.get_user(nick_from) - User_to = self.__Irc.User.get_user(nick_to) if not nick_to is None else None + batch_size = self._Config.BATCH_SIZE + User_from = self._Irc.User.get_user(nick_from) + User_to = self._Irc.User.get_user(nick_to) if not nick_to is None else None if User_from is None: - self.__Logs.error(f"The sender nickname [{nick_from}] do not exist") + self._Logs.error(f"The sender nickname [{nick_from}] do not exist") return None if not channel is None: @@ -172,8 +158,8 @@ class Unrealircd6(IProtocol): self.send2socket(f":{nick_from} PRIVMSG {User_to.uid} :{batch}") except Exception as err: - self.__Logs.error(f"General Error: {err}") - self.__Logs.error(f"General Error: {nick_from} - {channel} - {nick_to}") + self._Logs.error(f"General Error: {err}") + self._Logs.error(f"General Error: {nick_from} - {channel} - {nick_to}") def send_notice(self, nick_from: str, nick_to: str, msg: str) -> None: """Sending NOTICE by batches @@ -184,12 +170,12 @@ class Unrealircd6(IProtocol): nick_to (str): The reciever nickname """ try: - batch_size = self.__Config.BATCH_SIZE - User_from = self.__Irc.User.get_user(nick_from) - User_to = self.__Irc.User.get_user(nick_to) + batch_size = self._Config.BATCH_SIZE + User_from = self._Irc.User.get_user(nick_from) + User_to = self._Irc.User.get_user(nick_to) if User_from is None or User_to is None: - self.__Logs.error(f"The sender [{nick_from}] or the Reciever [{nick_to}] do not exist") + self._Logs.error(f"The sender [{nick_from}] or the Reciever [{nick_to}] do not exist") return None for i in range(0, len(str(msg)), batch_size): @@ -197,31 +183,31 @@ class Unrealircd6(IProtocol): self.send2socket(f":{User_from.uid} NOTICE {User_to.uid} :{batch}") except Exception as err: - self.__Logs.error(f"General Error: {err}") + self._Logs.error(f"General Error: {err}") def send_link(self): """Créer le link et envoyer les informations nécessaires pour la connexion au serveur. """ - service_id = self.__Config.SERVICE_ID - service_nickname = self.__Config.SERVICE_NICKNAME - service_username = self.__Config.SERVICE_USERNAME - service_realname = self.__Config.SERVICE_REALNAME - service_channel_log = self.__Config.SERVICE_CHANLOG - service_info = self.__Config.SERVICE_INFO - service_smodes = self.__Config.SERVICE_SMODES - service_cmodes = self.__Config.SERVICE_CMODES - service_umodes = self.__Config.SERVICE_UMODES - service_hostname = self.__Config.SERVICE_HOST - service_name = self.__Config.SERVICE_NAME + service_id = self._Config.SERVICE_ID + service_nickname = self._Config.SERVICE_NICKNAME + service_username = self._Config.SERVICE_USERNAME + service_realname = self._Config.SERVICE_REALNAME + service_channel_log = self._Config.SERVICE_CHANLOG + service_info = self._Config.SERVICE_INFO + service_smodes = self._Config.SERVICE_SMODES + service_cmodes = self._Config.SERVICE_CMODES + service_umodes = self._Config.SERVICE_UMODES + service_hostname = self._Config.SERVICE_HOST + service_name = self._Config.SERVICE_NAME protocolversion = self.protocol_version - server_password = self.__Config.SERVEUR_PASSWORD - server_link = self.__Config.SERVEUR_LINK - server_id = self.__Config.SERVEUR_ID + server_password = self._Config.SERVEUR_PASSWORD + server_link = self._Config.SERVEUR_LINK + server_id = self._Config.SERVEUR_ID - version = self.__Config.CURRENT_VERSION - unixtime = self.__Utils.get_unixtime() + version = self._Config.CURRENT_VERSION + unixtime = self._Utils.get_unixtime() self.send2socket(f":{server_id} PASS :{server_password}", print_log=False) self.send2socket(f":{server_id} PROTOCTL SID NOQUIT NICKv2 SJOIN SJ3 NICKIP TKLEXT2 NEXTBANS CLK EXTSWHOIS MLOCK MTAGS") @@ -236,7 +222,7 @@ class Unrealircd6(IProtocol): self.send2socket(f":{server_id} TKL + Q * {service_nickname} {service_hostname} 0 {unixtime} :Reserved for services") self.send2socket(f":{service_id} MODE {service_channel_log} {service_cmodes}") - self.__Logs.debug(f'>> {__name__} Link information sent to the server') + self._Logs.debug(f'>> {__name__} Link information sent to the server') def send_gline(self, nickname: str, hostname: str, set_by: str, expire_timestamp: int, set_at_timestamp: int, reason: str) -> None: """Send a gline command to the server @@ -251,7 +237,7 @@ class Unrealircd6(IProtocol): """ # TKL + G user host set_by expire_timestamp set_at_timestamp :reason - self.send2socket(f":{self.__Config.SERVEUR_ID} TKL + G {nickname} {hostname} {set_by} {expire_timestamp} {set_at_timestamp} :{reason}") + self.send2socket(f":{self._Config.SERVEUR_ID} TKL + G {nickname} {hostname} {set_by} {expire_timestamp} {set_at_timestamp} :{reason}") return None @@ -261,10 +247,10 @@ class Unrealircd6(IProtocol): Args: newnickname (str): New nickname of the server """ - self.send2socket(f":{self.__Config.SERVICE_NICKNAME} NICK {newnickname}") + self.send2socket(f":{self._Config.SERVICE_NICKNAME} NICK {newnickname}") - userObj = self.__Irc.User.get_user(self.__Config.SERVICE_NICKNAME) - self.__Irc.User.update_nickname(userObj.uid, newnickname) + userObj = self._Irc.User.get_user(self._Config.SERVICE_NICKNAME) + self._Irc.User.update_nickname(userObj.uid, newnickname) return None def send_set_mode(self, modes: str, *, nickname: Optional[str] = None, channel_name: Optional[str] = None, params: Optional[str] = None) -> None: @@ -276,16 +262,16 @@ class Unrealircd6(IProtocol): channel_name (Optional[str]): The channel name params (Optional[str]): Parameters like password. """ - service_id = self.__Config.SERVICE_ID + service_id = self._Config.SERVICE_ID if modes[0] not in ['+', '-']: - self.__Logs.error(f"[MODE ERROR] The mode you have provided is missing the sign: {modes}") + self._Logs.error(f"[MODE ERROR] The mode you have provided is missing the sign: {modes}") return None if nickname and channel_name: # :98KAAAAAB MODE #services +o defenderdev - if not self.__Irc.Channel.is_valid_channel(channel_name): - self.__Logs.error(f"[MODE ERROR] The channel is not valid: {channel_name}") + if not self._Irc.Channel.is_valid_channel(channel_name): + self._Logs.error(f"[MODE ERROR] The channel is not valid: {channel_name}") return None self.send2socket(f":{service_id} MODE {channel_name} {modes} {nickname}") return None @@ -295,8 +281,8 @@ class Unrealircd6(IProtocol): return None if nickname is None and channel_name: - if not self.__Irc.Channel.is_valid_channel(channel_name): - self.__Logs.error(f"[MODE ERROR] The channel is not valid: {channel_name}") + if not self._Irc.Channel.is_valid_channel(channel_name): + self._Logs.error(f"[MODE ERROR] The channel is not valid: {channel_name}") return None self.send2socket(f":{service_id} MODE {channel_name} {modes} {params}") return None @@ -313,20 +299,20 @@ class Unrealircd6(IProtocol): def send_ungline(self, nickname:str, hostname: str) -> None: - self.send2socket(f":{self.__Config.SERVEUR_ID} TKL - G {nickname} {hostname} {self.__Config.SERVICE_NICKNAME}") + self.send2socket(f":{self._Config.SERVEUR_ID} TKL - G {nickname} {hostname} {self._Config.SERVICE_NICKNAME}") return None def send_kline(self, nickname: str, hostname: str, set_by: str, expire_timestamp: int, set_at_timestamp: int, reason: str) -> None: # TKL + k user host set_by expire_timestamp set_at_timestamp :reason - self.send2socket(f":{self.__Config.SERVEUR_ID} TKL + k {nickname} {hostname} {set_by} {expire_timestamp} {set_at_timestamp} :{reason}") + self.send2socket(f":{self._Config.SERVEUR_ID} TKL + k {nickname} {hostname} {set_by} {expire_timestamp} {set_at_timestamp} :{reason}") return None def send_unkline(self, nickname:str, hostname: str) -> None: - self.send2socket(f":{self.__Config.SERVEUR_ID} TKL - K {nickname} {hostname} {self.__Config.SERVICE_NICKNAME}") + self.send2socket(f":{self._Config.SERVEUR_ID} TKL - K {nickname} {hostname} {self._Config.SERVICE_NICKNAME}") return None @@ -336,15 +322,15 @@ class Unrealircd6(IProtocol): Args: channel (str): Channel to join """ - if not self.__Irc.Channel.is_valid_channel(channel): - self.__Logs.error(f"The channel [{channel}] is not valid") + if not self._Irc.Channel.is_valid_channel(channel): + self._Logs.error(f"The channel [{channel}] is not valid") return None - self.send2socket(f":{self.__Config.SERVEUR_ID} SJOIN {self.__Utils.get_unixtime()} {channel} {self.__Config.SERVICE_UMODES} :{self.__Config.SERVICE_ID}") - self.send2socket(f":{self.__Config.SERVICE_ID} MODE {channel} {self.__Config.SERVICE_UMODES} {self.__Config.SERVICE_ID}") + self.send2socket(f":{self._Config.SERVEUR_ID} SJOIN {self._Utils.get_unixtime()} {channel} {self._Config.SERVICE_UMODES} :{self._Config.SERVICE_ID}") + self.send2socket(f":{self._Config.SERVICE_ID} MODE {channel} {self._Config.SERVICE_UMODES} {self._Config.SERVICE_ID}") # Add defender to the channel uids list - self.__Irc.Channel.insert(self.__Irc.Loader.Definition.MChannel(name=channel, uids=[self.__Config.SERVICE_ID])) + self._Irc.Channel.insert(self._Irc.Loader.Definition.MChannel(name=channel, uids=[self._Config.SERVICE_ID])) return None def send_sapart(self, nick_to_sapart: str, channel_name: str) -> None: @@ -357,20 +343,20 @@ class Unrealircd6(IProtocol): """ try: - userObj = self.__Irc.User.get_user(uidornickname=nick_to_sapart) - chanObj = self.__Irc.Channel.get_channel(channel_name) - service_uid = self.__Config.SERVICE_ID + userObj = self._Irc.User.get_user(uidornickname=nick_to_sapart) + chanObj = self._Irc.Channel.get_channel(channel_name) + service_uid = self._Config.SERVICE_ID if userObj is None or chanObj is None: return None self.send2socket(f":{service_uid} SAPART {userObj.nickname} {chanObj.name}") - self.__Irc.Channel.delete_user_from_channel(chanObj.name, userObj.uid) + self._Irc.Channel.delete_user_from_channel(chanObj.name, userObj.uid) return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def send_sajoin(self, nick_to_sajoin: str, channel_name: str) -> None: """_summary_ @@ -381,9 +367,9 @@ class Unrealircd6(IProtocol): """ try: - userObj = self.__Irc.User.get_user(uidornickname=nick_to_sajoin) - chanObj = self.__Irc.Channel.get_channel(channel_name) - service_uid = self.__Config.SERVICE_ID + userObj = self._Irc.User.get_user(uidornickname=nick_to_sajoin) + chanObj = self._Irc.Channel.get_channel(channel_name) + service_uid = self._Config.SERVICE_ID if userObj is None: # User not exist: leave @@ -391,53 +377,53 @@ class Unrealircd6(IProtocol): if chanObj is None: # Channel not exist - if not self.__Irc.Channel.is_valid_channel(channel_name): + if not self._Irc.Channel.is_valid_channel(channel_name): # Incorrect channel: leave return None # Create the new channel with the uid - newChanObj = self.__Irc.Loader.Definition.MChannel(name=channel_name, uids=[userObj.uid]) - self.__Irc.Channel.insert(newChanObj) + newChanObj = self._Irc.Loader.Definition.MChannel(name=channel_name, uids=[userObj.uid]) + self._Irc.Channel.insert(newChanObj) self.send2socket(f":{service_uid} SAJOIN {userObj.nickname} {newChanObj.name}") else: - self.__Irc.Channel.add_user_to_a_channel(channel_name=channel_name, uid=userObj.uid) + self._Irc.Channel.add_user_to_a_channel(channel_name=channel_name, uid=userObj.uid) self.send2socket(f":{service_uid} SAJOIN {userObj.nickname} {chanObj.name}") return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def send_svspart(self, nick_to_part: str, channels: list[str], reason: str) -> None: - user_obj = self.__Irc.User.get_user(nick_to_part) + user_obj = self._Irc.User.get_user(nick_to_part) if user_obj is None: - self.__Logs.debug(f"[SVSPART] The nickname {nick_to_part} do not exist!") + self._Logs.debug(f"[SVSPART] The nickname {nick_to_part} do not exist!") return None - channels_list = ','.join([channel for channel in channels if self.__Irc.Channel.is_valid_channel(channel)]) - service_id = self.__Config.SERVICE_ID + channels_list = ','.join([channel for channel in channels if self._Irc.Channel.is_valid_channel(channel)]) + service_id = self._Config.SERVICE_ID self.send2socket(f':{service_id} SVSPART {user_obj.nickname} {channels_list} {reason}') return None def send_svsjoin(self, nick_to_part: str, channels: list[str], keys: list[str]) -> None: - user_obj = self.__Irc.User.get_user(nick_to_part) + user_obj = self._Irc.User.get_user(nick_to_part) if user_obj is None: - self.__Logs.debug(f"[SVSJOIN] The nickname {nick_to_part} do not exist!") + self._Logs.debug(f"[SVSJOIN] The nickname {nick_to_part} do not exist!") return None - channels_list = ','.join([channel for channel in channels if self.__Irc.Channel.is_valid_channel(channel)]) + channels_list = ','.join([channel for channel in channels if self._Irc.Channel.is_valid_channel(channel)]) keys_list = ','.join([key for key in keys]) - service_id = self.__Config.SERVICE_ID + service_id = self._Config.SERVICE_ID self.send2socket(f':{service_id} SVSJOIN {user_obj.nickname} {channels_list} {keys_list}') return None def send_svsmode(self, nickname: str, user_mode: str) -> None: try: - user_obj = self.__Irc.User.get_user(uidornickname=nickname) - service_uid = self.__Config.SERVICE_ID + user_obj = self._Irc.User.get_user(uidornickname=nickname) + service_uid = self._Config.SERVICE_ID if user_obj is None: return None @@ -445,16 +431,16 @@ class Unrealircd6(IProtocol): self.send2socket(f':{service_uid} SVSMODE {nickname} {user_mode}') # Update new mode - self.__Irc.User.update_mode(user_obj.uid, user_mode) + self._Irc.User.update_mode(user_obj.uid, user_mode) return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def send_svs2mode(self, nickname: str, user_mode: str) -> None: try: - user_obj = self.__Irc.User.get_user(uidornickname=nickname) - service_uid = self.__Config.SERVICE_ID + user_obj = self._Irc.User.get_user(uidornickname=nickname) + service_uid = self._Config.SERVICE_ID if user_obj is None: return None @@ -462,11 +448,11 @@ class Unrealircd6(IProtocol): self.send2socket(f':{service_uid} SVS2MODE {nickname} {user_mode}') # Update new mode - self.__Irc.User.update_mode(user_obj.uid, user_mode) + self._Irc.User.update_mode(user_obj.uid, user_mode) return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def send_svslogin(self, client_uid: str, user_account: str) -> None: """Log a client into his account. @@ -476,9 +462,9 @@ class Unrealircd6(IProtocol): user_account (str): The account of the user """ try: - self.send2socket(f":{self.__Irc.Config.SERVEUR_LINK} SVSLOGIN {self.__Settings.MAIN_SERVER_HOSTNAME} {client_uid} {user_account}") + self.send2socket(f":{self._Irc.Config.SERVEUR_LINK} SVSLOGIN {self._Settings.MAIN_SERVER_HOSTNAME} {client_uid} {user_account}") except Exception as err: - self.__Logs.error(f'General Error: {err}') + self._Logs.error(f'General Error: {err}') def send_svslogout(self, client_obj: 'MClient') -> None: """Logout a client from his account @@ -489,11 +475,11 @@ class Unrealircd6(IProtocol): try: c_uid = client_obj.uid c_nickname = client_obj.nickname - self.send2socket(f":{self.__Irc.Config.SERVEUR_LINK} SVSLOGIN {self.__Settings.MAIN_SERVER_HOSTNAME} {c_uid} 0") + self.send2socket(f":{self._Irc.Config.SERVEUR_LINK} SVSLOGIN {self._Settings.MAIN_SERVER_HOSTNAME} {c_uid} 0") self.send_svs2mode(c_nickname, '-r') except Exception as err: - self.__Logs.error(f'General Error: {err}') + self._Logs.error(f'General Error: {err}') def send_quit(self, uid: str, reason: str, print_log: True) -> None: """Send quit message @@ -504,18 +490,18 @@ class Unrealircd6(IProtocol): uidornickname (str): The UID or the Nickname reason (str): The reason for the quit """ - user_obj = self.__Irc.User.get_user(uidornickname=uid) - reputationObj = self.__Irc.Reputation.get_reputation(uidornickname=uid) + user_obj = self._Irc.User.get_user(uidornickname=uid) + reputationObj = self._Irc.Reputation.get_reputation(uidornickname=uid) if not user_obj is None: self.send2socket(f":{user_obj.uid} QUIT :{reason}", print_log=print_log) - self.__Irc.User.delete(user_obj.uid) + self._Irc.User.delete(user_obj.uid) if not reputationObj is None: - self.__Irc.Reputation.delete(reputationObj.uid) + self._Irc.Reputation.delete(reputationObj.uid) - if not self.__Irc.Channel.delete_user_from_all_channel(uid): - self.__Logs.error(f"The UID [{uid}] has not been deleted from all channels") + if not self._Irc.Channel.delete_user_from_all_channel(uid): + self._Logs.error(f"The UID [{uid}] has not been deleted from all channels") return None @@ -534,28 +520,28 @@ class Unrealircd6(IProtocol): print_log (bool, optional): print logs if true. Defaults to True. """ # {self.Config.SERVEUR_ID} UID - # {clone.nickname} 1 {self.__Utils.get_unixtime()} {clone.username} {clone.hostname} {clone.uid} * {clone.umodes} {clone.vhost} * {self.Base.encode_ip(clone.remote_ip)} :{clone.realname} + # {clone.nickname} 1 {self._Utils.get_unixtime()} {clone.username} {clone.hostname} {clone.uid} * {clone.umodes} {clone.vhost} * {self.Base.encode_ip(clone.remote_ip)} :{clone.realname} try: - unixtime = self.__Utils.get_unixtime() - encoded_ip = self.__Base.encode_ip(remote_ip) + unixtime = self._Utils.get_unixtime() + encoded_ip = self._Base.encode_ip(remote_ip) # Create the user - self.__Irc.User.insert( - self.__Irc.Loader.Definition.MUser( + self._Irc.User.insert( + self._Irc.Loader.Definition.MUser( uid=uid, nickname=nickname, username=username, realname=realname,hostname=hostname, umodes=umodes, vhost=vhost, remote_ip=remote_ip ) ) - uid_msg = f":{self.__Config.SERVEUR_ID} UID {nickname} 1 {unixtime} {username} {hostname} {uid} * {umodes} {vhost} * {encoded_ip} :{realname}" + uid_msg = f":{self._Config.SERVEUR_ID} UID {nickname} 1 {unixtime} {username} {hostname} {uid} * {umodes} {vhost} * {encoded_ip} :{realname}" self.send2socket(uid_msg, print_log=print_log) return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def send_join_chan(self, uidornickname: str, channel: str, password: str = None, print_log: bool = True) -> None: """Joining a channel @@ -567,34 +553,34 @@ class Unrealircd6(IProtocol): print_log (bool, optional): Write logs. Defaults to True. """ - userObj = self.__Irc.User.get_user(uidornickname) + userObj = self._Irc.User.get_user(uidornickname) passwordChannel = password if not password is None else '' if userObj is None: return None - if not self.__Irc.Channel.is_valid_channel(channel): - self.__Logs.error(f"The channel [{channel}] is not valid") + if not self._Irc.Channel.is_valid_channel(channel): + self._Logs.error(f"The channel [{channel}] is not valid") return None self.send2socket(f":{userObj.uid} JOIN {channel} {passwordChannel}", print_log=print_log) - if uidornickname == self.__Config.SERVICE_NICKNAME or uidornickname == self.__Config.SERVICE_ID: - self.send2socket(f":{self.__Config.SERVICE_ID} MODE {channel} {self.__Config.SERVICE_UMODES} {self.__Config.SERVICE_ID}") + if uidornickname == self._Config.SERVICE_NICKNAME or uidornickname == self._Config.SERVICE_ID: + self.send2socket(f":{self._Config.SERVICE_ID} MODE {channel} {self._Config.SERVICE_UMODES} {self._Config.SERVICE_ID}") # Add defender to the channel uids list - self.__Irc.Channel.insert(self.__Irc.Loader.Definition.MChannel(name=channel, uids=[userObj.uid])) + self._Irc.Channel.insert(self._Irc.Loader.Definition.MChannel(name=channel, uids=[userObj.uid])) # Set the automode to the user if 'r' not in userObj.umodes and 'o' not in userObj.umodes: return None db_data: dict[str, str] = {"nickname": userObj.nickname, "channel": channel} - db_query = self.__Base.db_execute_query("SELECT id, mode FROM command_automode WHERE nickname = :nickname AND channel = :channel", db_data) + db_query = self._Base.db_execute_query("SELECT id, mode FROM command_automode WHERE nickname = :nickname AND channel = :channel", db_data) db_result = db_query.fetchone() if db_result is not None: id, mode = db_result - self.send2socket(f":{self.__Config.SERVICE_ID} MODE {channel} {mode} {userObj.nickname}") + self.send2socket(f":{self._Config.SERVICE_ID} MODE {channel} {mode} {userObj.nickname}") return None @@ -607,35 +593,35 @@ class Unrealircd6(IProtocol): print_log (bool, optional): Write logs. Defaults to True. """ - userObj = self.__Irc.User.get_user(uidornickname) + userObj = self._Irc.User.get_user(uidornickname) if userObj is None: - self.__Logs.error(f"The user [{uidornickname}] is not valid") + self._Logs.error(f"The user [{uidornickname}] is not valid") return None - if not self.__Irc.Channel.is_valid_channel(channel): - self.__Logs.error(f"The channel [{channel}] is not valid") + if not self._Irc.Channel.is_valid_channel(channel): + self._Logs.error(f"The channel [{channel}] is not valid") return None self.send2socket(f":{userObj.uid} PART {channel}", print_log=print_log) # Add defender to the channel uids list - self.__Irc.Channel.delete_user_from_channel(channel, userObj.uid) + self._Irc.Channel.delete_user_from_channel(channel, userObj.uid) return None def send_mode_chan(self, channel_name: str, channel_mode: str) -> None: - channel = self.__Irc.Channel.is_valid_channel(channel_name) + channel = self._Irc.Channel.is_valid_channel(channel_name) if not channel: - self.__Logs.error(f'The channel [{channel_name}] is not correct') + self._Logs.error(f'The channel [{channel_name}] is not correct') return None - self.send2socket(f":{self.__Config.SERVICE_NICKNAME} MODE {channel_name} {channel_mode}") + self.send2socket(f":{self._Config.SERVICE_NICKNAME} MODE {channel_name} {channel_mode}") return None def send_raw(self, raw_command: str) -> None: - self.send2socket(f":{self.__Config.SERVICE_NICKNAME} {raw_command}") + self.send2socket(f":{self._Config.SERVICE_NICKNAME} {raw_command}") return None @@ -650,7 +636,7 @@ class Unrealircd6(IProtocol): serverMsg (list[str]): The UID ircd response """ umodes = str(serverMsg[10]) - remote_ip = self.__Base.decode_ip(str(serverMsg[13])) if 'S' not in umodes else '127.0.0.1' + remote_ip = self._Base.decode_ip(str(serverMsg[13])) if 'S' not in umodes else '127.0.0.1' # Extract Geoip information pattern = r'^.*geoip=cc=(\S{2}).*$' @@ -732,8 +718,8 @@ class Unrealircd6(IProtocol): response = { "uid_sender": scopy[0].replace(':', ''), - "uid_reciever": self.__Irc.User.get_uid(scopy[2]), - "channel": scopy[2] if self.__Irc.Channel.is_valid_channel(scopy[2]) else None, + "uid_reciever": self._Irc.User.get_uid(scopy[2]), + "channel": scopy[2] if self._Irc.Channel.is_valid_channel(scopy[2]) else None, "message": " ".join(scopy[3:]) } return response @@ -755,19 +741,19 @@ class Unrealircd6(IProtocol): uid_user_to_edit = serverMsg[2] umode = serverMsg[3] - userObj = self.__Irc.User.get_user(uid_user_to_edit) + userObj = self._Irc.User.get_user(uid_user_to_edit) if userObj is None: return None - if self.__Irc.User.update_mode(userObj.uid, umode): + if self._Irc.User.update_mode(userObj.uid, umode): return None return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_mode(self, serverMsg: list[str]) -> None: """Handle mode coming from a server @@ -790,7 +776,7 @@ class Unrealircd6(IProtocol): try: # [':adator_', 'UMODE2', '-iwx'] - userObj = self.__Irc.User.get_user(str(serverMsg[0]).lstrip(':')) + userObj = self._Irc.User.get_user(str(serverMsg[0]).lstrip(':')) userMode = serverMsg[2] if userObj is None: # If user is not created @@ -800,16 +786,16 @@ class Unrealircd6(IProtocol): old_umodes = userObj.umodes # TODO : User object should be able to update user modes - if self.__Irc.User.update_mode(userObj.uid, userMode): + if self._Irc.User.update_mode(userObj.uid, userMode): return None - # self.__Logs.debug(f"Updating user mode for [{userObj.nickname}] [{old_umodes}] => [{userObj.umodes}]") + # self._Logs.debug(f"Updating user mode for [{userObj.nickname}] [{old_umodes}] => [{userObj.umodes}]") return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_quit(self, serverMsg: list[str]) -> None: """Handle quit coming from a server @@ -822,18 +808,18 @@ class Unrealircd6(IProtocol): uid_who_quit = str(serverMsg[1]).lstrip(':') - self.__Irc.Channel.delete_user_from_all_channel(uid_who_quit) - self.__Irc.User.delete(uid_who_quit) - self.__Irc.Client.delete(uid_who_quit) - self.__Irc.Reputation.delete(uid_who_quit) - self.__Irc.Admin.delete(uid_who_quit) + self._Irc.Channel.delete_user_from_all_channel(uid_who_quit) + self._Irc.User.delete(uid_who_quit) + self._Irc.Client.delete(uid_who_quit) + self._Irc.Reputation.delete(uid_who_quit) + self._Irc.Admin.delete(uid_who_quit) return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_squit(self, serverMsg: list[str]) -> None: """Handle squit coming from a server @@ -845,15 +831,15 @@ class Unrealircd6(IProtocol): server_hostname = serverMsg[2] uid_to_delete = None - for s_user in self.__Irc.User.UID_DB: + for s_user in self._Irc.User.UID_DB: if s_user.hostname == server_hostname and 'S' in s_user.umodes: uid_to_delete = s_user.uid if uid_to_delete is None: return None - self.__Irc.User.delete(uid_to_delete) - self.__Irc.Channel.delete_user_from_all_channel(uid_to_delete) + self._Irc.User.delete(uid_to_delete) + self._Irc.Channel.delete_user_from_all_channel(uid_to_delete) return None @@ -887,9 +873,9 @@ class Unrealircd6(IProtocol): if user_modes is None or prefix is None or host_server_id is None: return None - self.__Config.HSID = host_server_id - self.__Settings.PROTOCTL_USER_MODES = list(user_modes) - self.__Settings.PROTOCTL_PREFIX = list(prefix) + self._Config.HSID = host_server_id + self._Settings.PROTOCTL_USER_MODES = list(user_modes) + self._Settings.PROTOCTL_PREFIX = list(prefix) return None @@ -906,17 +892,17 @@ class Unrealircd6(IProtocol): uid = str(serverMsg[1]).lstrip(':') newnickname = serverMsg[3] - self.__Irc.User.update_nickname(uid, newnickname) - self.__Irc.Client.update_nickname(uid, newnickname) - self.__Irc.Admin.update_nickname(uid, newnickname) - self.__Irc.Reputation.update(uid, newnickname) + self._Irc.User.update_nickname(uid, newnickname) + self._Irc.Client.update_nickname(uid, newnickname) + self._Irc.Admin.update_nickname(uid, newnickname) + self._Irc.Reputation.update(uid, newnickname) return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_sjoin(self, serverMsg: list[str]) -> None: """Handle sjoin coming from a server @@ -953,13 +939,13 @@ class Unrealircd6(IProtocol): # Boucle qui va ajouter l'ensemble des users (UID) for i in range(start_boucle, len(serverMsg_copy)): parsed_UID = str(serverMsg_copy[i]) - clean_uid = self.__Utils.clean_uid(parsed_UID) + clean_uid = self._Utils.clean_uid(parsed_UID) if not clean_uid is None and len(clean_uid) == 9: list_users.append(clean_uid) if list_users: - self.__Irc.Channel.insert( - self.__Irc.Loader.Definition.MChannel( + self._Irc.Channel.insert( + self._Irc.Loader.Definition.MChannel( name=channel, uids=list_users ) @@ -967,9 +953,9 @@ class Unrealircd6(IProtocol): return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_part(self, serverMsg: list[str]) -> None: """Handle part coming from a server @@ -981,13 +967,13 @@ class Unrealircd6(IProtocol): # ['@unrealircd.org', ':001EPFBRD', 'PART', '#welcome', ':WEB', 'IRC', 'Paris'] uid = str(serverMsg[1]).lstrip(':') channel = str(serverMsg[3]).lower() - self.__Irc.Channel.delete_user_from_channel(channel, uid) + self._Irc.Channel.delete_user_from_channel(channel, uid) return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_eos(self, serverMsg: list[str]) -> None: """Handle EOS coming from a server @@ -999,69 +985,69 @@ class Unrealircd6(IProtocol): # [':001', 'EOS'] server_msg_copy = serverMsg.copy() hsid = str(server_msg_copy[0]).replace(':','') - if hsid == self.__Config.HSID: - if self.__Config.DEFENDER_INIT == 1: - current_version = self.__Config.CURRENT_VERSION - latest_version = self.__Config.LATEST_VERSION - if self.__Base.check_for_new_version(False): + if hsid == self._Config.HSID: + if self._Config.DEFENDER_INIT == 1: + current_version = self._Config.CURRENT_VERSION + latest_version = self._Config.LATEST_VERSION + if self._Base.check_for_new_version(False): version = f'{current_version} >>> {latest_version}' else: version = f'{current_version}' print(f"################### DEFENDER ###################") print(f"# SERVICE CONNECTE ") - print(f"# SERVEUR : {self.__Config.SERVEUR_IP} ") - print(f"# PORT : {self.__Config.SERVEUR_PORT} ") - print(f"# SSL : {self.__Config.SERVEUR_SSL} ") - print(f"# SSL VER : {self.__Config.SSL_VERSION} ") - print(f"# NICKNAME : {self.__Config.SERVICE_NICKNAME} ") - print(f"# CHANNEL : {self.__Config.SERVICE_CHANLOG} ") + print(f"# SERVEUR : {self._Config.SERVEUR_IP} ") + print(f"# PORT : {self._Config.SERVEUR_PORT} ") + print(f"# SSL : {self._Config.SERVEUR_SSL} ") + print(f"# SSL VER : {self._Config.SSL_VERSION} ") + print(f"# NICKNAME : {self._Config.SERVICE_NICKNAME} ") + print(f"# CHANNEL : {self._Config.SERVICE_CHANLOG} ") print(f"# VERSION : {version} ") print(f"################################################") - self.__Logs.info(f"################### DEFENDER ###################") - self.__Logs.info(f"# SERVICE CONNECTE ") - self.__Logs.info(f"# SERVEUR : {self.__Config.SERVEUR_IP} ") - self.__Logs.info(f"# PORT : {self.__Config.SERVEUR_PORT} ") - self.__Logs.info(f"# SSL : {self.__Config.SERVEUR_SSL} ") - self.__Logs.info(f"# SSL VER : {self.__Config.SSL_VERSION} ") - self.__Logs.info(f"# NICKNAME : {self.__Config.SERVICE_NICKNAME} ") - self.__Logs.info(f"# CHANNEL : {self.__Config.SERVICE_CHANLOG} ") - self.__Logs.info(f"# VERSION : {version} ") - self.__Logs.info(f"################################################") + self._Logs.info(f"################### DEFENDER ###################") + self._Logs.info(f"# SERVICE CONNECTE ") + self._Logs.info(f"# SERVEUR : {self._Config.SERVEUR_IP} ") + self._Logs.info(f"# PORT : {self._Config.SERVEUR_PORT} ") + self._Logs.info(f"# SSL : {self._Config.SERVEUR_SSL} ") + self._Logs.info(f"# SSL VER : {self._Config.SSL_VERSION} ") + self._Logs.info(f"# NICKNAME : {self._Config.SERVICE_NICKNAME} ") + self._Logs.info(f"# CHANNEL : {self._Config.SERVICE_CHANLOG} ") + self._Logs.info(f"# VERSION : {version} ") + self._Logs.info(f"################################################") - self.send_sjoin(self.__Config.SERVICE_CHANLOG) + self.send_sjoin(self._Config.SERVICE_CHANLOG) - if self.__Base.check_for_new_version(False): + if self._Base.check_for_new_version(False): self.send_priv_msg( - nick_from=self.__Config.SERVICE_NICKNAME, + nick_from=self._Config.SERVICE_NICKNAME, msg=f" New Version available {version}", - channel=self.__Config.SERVICE_CHANLOG + channel=self._Config.SERVICE_CHANLOG ) # Initialisation terminé aprés le premier PING self.send_priv_msg( - nick_from=self.__Config.SERVICE_NICKNAME, - msg=tr("[ %sINFORMATION%s ] >> %s is ready!", self.__Config.COLORS.green, self.__Config.COLORS.nogc, self.__Config.SERVICE_NICKNAME), - channel=self.__Config.SERVICE_CHANLOG + nick_from=self._Config.SERVICE_NICKNAME, + msg=tr("[ %sINFORMATION%s ] >> %s is ready!", self._Config.COLORS.green, self._Config.COLORS.nogc, self._Config.SERVICE_NICKNAME), + channel=self._Config.SERVICE_CHANLOG ) - self.__Config.DEFENDER_INIT = 0 + self._Config.DEFENDER_INIT = 0 # Send EOF to other modules - for module in self.__Irc.ModuleUtils.model_get_loaded_modules().copy(): + for module in self._Irc.ModuleUtils.model_get_loaded_modules().copy(): module.class_instance.cmd(server_msg_copy) # Join saved channels & load existing modules - self.__Irc.join_saved_channels() - self.__Irc.ModuleUtils.db_load_all_existing_modules(self.__Irc) + self._Irc.join_saved_channels() + self._Irc.ModuleUtils.db_load_all_existing_modules(self._Irc) return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Key Error: {ie}") + self._Logs.error(f"{__name__} - Key Error: {ie}") except KeyError as ke: - self.__Logs.error(f"{__name__} - Key Error: {ke}") + self._Logs.error(f"{__name__} - Key Error: {ke}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_reputation(self, serverMsg: list[str]) -> None: """Handle REPUTATION coming from a server @@ -1072,27 +1058,27 @@ class Unrealircd6(IProtocol): try: # :001 REPUTATION 127.0.0.1 118 server_msg_copy = serverMsg.copy() - self.__Irc.first_connexion_ip = server_msg_copy[2] - self.__Irc.first_score = 0 + self._Irc.first_connexion_ip = server_msg_copy[2] + self._Irc.first_score = 0 if str(server_msg_copy[3]).find('*') != -1: # If * available, it means that an ircop changed the repurtation score # means also that the user exist will try to update all users with same IP - self.__Irc.first_score = int(str(server_msg_copy[3]).replace('*','')) - for user in self.__Irc.User.UID_DB: - if user.remote_ip == self.__Irc.first_connexion_ip: - user.score_connexion = self.__Irc.first_score + self._Irc.first_score = int(str(server_msg_copy[3]).replace('*','')) + for user in self._Irc.User.UID_DB: + if user.remote_ip == self._Irc.first_connexion_ip: + user.score_connexion = self._Irc.first_score else: - self.__Irc.first_score = int(server_msg_copy[3]) + self._Irc.first_score = int(server_msg_copy[3]) # Possibilité de déclancher les bans a ce niveau. except IndexError as ie: - self.__Logs.error(f'Index Error {__name__}: {ie}') + self._Logs.error(f'Index Error {__name__}: {ie}') except ValueError as ve: - self.__Irc.first_score = 0 - self.__Logs.error(f'Value Error {__name__}: {ve}') + self._Irc.first_score = 0 + self._Logs.error(f'Value Error {__name__}: {ve}') except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_uid(self, serverMsg: list[str]) -> None: """Handle uid message coming from the server @@ -1116,7 +1102,7 @@ class Unrealircd6(IProtocol): vhost = str(serverMsg[11]) if not 'S' in umodes: - remote_ip = self.__Base.decode_ip(str(serverMsg[13])) + remote_ip = self._Base.decode_ip(str(serverMsg[13])) else: remote_ip = '127.0.0.1' @@ -1142,10 +1128,10 @@ class Unrealircd6(IProtocol): else: geoip = None - score_connexion = self.__Irc.first_score + score_connexion = self._Irc.first_score - self.__Irc.User.insert( - self.__Irc.Loader.Definition.MUser( + self._Irc.User.insert( + self._Irc.Loader.Definition.MUser( uid=uid, nickname=nickname, username=username, @@ -1165,13 +1151,13 @@ class Unrealircd6(IProtocol): ) # Auto Auth admin via fingerprint - dnickname = self.__Config.SERVICE_NICKNAME - dchanlog = self.__Config.SERVICE_CHANLOG - GREEN = self.__Config.COLORS.green - RED = self.__Config.COLORS.red - NOGC = self.__Config.COLORS.nogc + dnickname = self._Config.SERVICE_NICKNAME + dchanlog = self._Config.SERVICE_CHANLOG + GREEN = self._Config.COLORS.green + RED = self._Config.COLORS.red + NOGC = self._Config.COLORS.nogc - for module in self.__Irc.ModuleUtils.model_get_loaded_modules().copy(): + for module in self._Irc.ModuleUtils.model_get_loaded_modules().copy(): module.class_instance.cmd(serverMsg) # SASL authentication @@ -1179,10 +1165,10 @@ class Unrealircd6(IProtocol): uid = serverMsg[8] nickname = serverMsg[3] - sasl_obj = self.__Irc.Sasl.get_sasl_obj(uid) + sasl_obj = self._Irc.Sasl.get_sasl_obj(uid) if sasl_obj: if sasl_obj.auth_success: - self.__Irc.insert_db_admin(sasl_obj.client_uid, sasl_obj.username, sasl_obj.level, sasl_obj.language) + self._Irc.insert_db_admin(sasl_obj.client_uid, sasl_obj.username, sasl_obj.level, sasl_obj.language) self.send_priv_msg(nick_from=dnickname, msg=tr("[ %sSASL AUTH%s ] - %s (%s) is now connected successfuly to %s", GREEN, NOGC, nickname, sasl_obj.username, dnickname), channel=dchanlog) @@ -1194,12 +1180,12 @@ class Unrealircd6(IProtocol): self.send_notice(nick_from=dnickname, nick_to=nickname, msg=tr("Wrong password!")) # Delete sasl object! - self.__Irc.Sasl.delete_sasl_client(uid) + self._Irc.Sasl.delete_sasl_client(uid) return None # If no sasl authentication then auto connect via fingerprint - if self.__Irc.Admin.db_auth_admin_via_fingerprint(fingerprint, uid): - admin = self.__Irc.Admin.get_admin(uid) + if self._Irc.Admin.db_auth_admin_via_fingerprint(fingerprint, uid): + admin = self._Irc.Admin.get_admin(uid) account = admin.account if admin else '' self.send_priv_msg(nick_from=dnickname, msg=tr("[ %sFINGERPRINT AUTH%s ] - %s (%s) is now connected successfuly to %s", GREEN, NOGC, nickname, account, dnickname), @@ -1208,9 +1194,9 @@ class Unrealircd6(IProtocol): return None except IndexError as ie: - self.__Logs.error(f"{__name__} - Index Error: {ie}") + self._Logs.error(f"{__name__} - Index Error: {ie}") except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_privmsg(self, serverMsg: list[str]) -> None: """Handle PRIVMSG message coming from the server @@ -1226,32 +1212,32 @@ class Unrealircd6(IProtocol): cmd.pop(0) get_uid_or_nickname = str(cmd[0].replace(':','')) - user_trigger = self.__Irc.User.get_nickname(get_uid_or_nickname) - dnickname = self.__Config.SERVICE_NICKNAME - pattern = fr'(:\{self.__Config.SERVICE_PREFIX})(.*)$' + user_trigger = self._Irc.User.get_nickname(get_uid_or_nickname) + dnickname = self._Config.SERVICE_NICKNAME + pattern = fr'(:\{self._Config.SERVICE_PREFIX})(.*)$' hcmds = search(pattern, ' '.join(cmd)) # va matcher avec tout les caractéres aprés le . if hcmds: # Commande qui commencent par le point liste_des_commandes = list(hcmds.groups()) convert_to_string = ' '.join(liste_des_commandes) arg = convert_to_string.split() - arg.remove(f':{self.__Config.SERVICE_PREFIX}') - if not self.__Irc.Commands.is_command_exist(arg[0]): - self.__Logs.debug(f"This command {arg[0]} is not available") + arg.remove(f':{self._Config.SERVICE_PREFIX}') + if not self._Irc.Commands.is_command_exist(arg[0]): + self._Logs.debug(f"This command {arg[0]} is not available") self.send_notice( - nick_from=self.__Config.SERVICE_NICKNAME, + nick_from=self._Config.SERVICE_NICKNAME, nick_to=user_trigger, - msg=f"This command [{self.__Config.COLORS.bold}{arg[0]}{self.__Config.COLORS.bold}] is not available" + msg=f"This command [{self._Config.COLORS.bold}{arg[0]}{self._Config.COLORS.bold}] is not available" ) return None cmd_to_send = convert_to_string.replace(':','') - self.__Base.log_cmd(user_trigger, cmd_to_send) + self._Base.log_cmd(user_trigger, cmd_to_send) - fromchannel = str(cmd[2]).lower() if self.__Irc.Channel.is_valid_channel(cmd[2]) else None - self.__Irc.hcmds(user_trigger, fromchannel, arg, cmd) + fromchannel = str(cmd[2]).lower() if self._Irc.Channel.is_valid_channel(cmd[2]) else None + self._Irc.hcmds(user_trigger, fromchannel, arg, cmd) - if cmd[2] == self.__Config.SERVICE_ID: + if cmd[2] == self._Config.SERVICE_ID: pattern = fr'^:.*?:(.*)$' hcmds = search(pattern, ' '.join(cmd)) @@ -1275,30 +1261,30 @@ class Unrealircd6(IProtocol): self.on_ping(srv_msg) return None - if not self.__Irc.Commands.is_command_exist(arg[0]): - self.__Logs.debug(f"This command {arg[0]} sent by {user_trigger} is not available") + if not self._Irc.Commands.is_command_exist(arg[0]): + self._Logs.debug(f"This command {arg[0]} sent by {user_trigger} is not available") return None - # if not arg[0].lower() in self.__Irc.module_commands_list: - # self.__Logs.debug(f"This command {arg[0]} sent by {user_trigger} is not available") + # if not arg[0].lower() in self._Irc.module_commands_list: + # self._Logs.debug(f"This command {arg[0]} sent by {user_trigger} is not available") # return False cmd_to_send = convert_to_string.replace(':','') - self.__Base.log_cmd(user_trigger, cmd_to_send) + self._Base.log_cmd(user_trigger, cmd_to_send) fromchannel = None if len(arg) >= 2: - fromchannel = str(arg[1]).lower() if self.__Irc.Channel.is_valid_channel(arg[1]) else None + fromchannel = str(arg[1]).lower() if self._Irc.Channel.is_valid_channel(arg[1]) else None - self.__Irc.hcmds(user_trigger, fromchannel, arg, cmd) + self._Irc.hcmds(user_trigger, fromchannel, arg, cmd) return None except KeyError as ke: - self.__Logs.error(f"Key Error: {ke}") + self._Logs.error(f"Key Error: {ke}") except AttributeError as ae: - self.__Logs.error(f"Attribute Error: {ae}") + self._Logs.error(f"Attribute Error: {ae}") except Exception as err: - self.__Logs.error(f"General Error: {err} - {srv_msg}" , exc_info=True) + self._Logs.error(f"General Error: {err} - {srv_msg}" , exc_info=True) def on_server_ping(self, serverMsg: list[str]) -> None: """Send a PONG message to the server @@ -1312,7 +1298,7 @@ class Unrealircd6(IProtocol): return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_server(self, serverMsg: list[str]) -> None: """_summary_ @@ -1323,9 +1309,9 @@ class Unrealircd6(IProtocol): try: # ['SERVER', 'irc.local.org', '1', ':U6100-Fhn6OoE-001', 'Local', 'Server'] sCopy = serverMsg.copy() - self.__Irc.Settings.MAIN_SERVER_HOSTNAME = sCopy[1] + self._Irc.Settings.MAIN_SERVER_HOSTNAME = sCopy[1] except Exception as err: - self.__Logs.error(f'General Error: {err}') + self._Logs.error(f'General Error: {err}') def on_version(self, serverMsg: list[str]) -> None: """Sending Server Version to the server @@ -1337,19 +1323,19 @@ class Unrealircd6(IProtocol): # Réponse a un CTCP VERSION try: - nickname = self.__Irc.User.get_nickname(self.__Utils.clean_uid(serverMsg[1])) - dnickname = self.__Config.SERVICE_NICKNAME + nickname = self._Irc.User.get_nickname(self._Utils.clean_uid(serverMsg[1])) + dnickname = self._Config.SERVICE_NICKNAME arg = serverMsg[4].replace(':', '') if nickname is None: return None if arg == '\x01VERSION\x01': - self.send2socket(f':{dnickname} NOTICE {nickname} :\x01VERSION Service {self.__Config.SERVICE_NICKNAME} V{self.__Config.CURRENT_VERSION}\x01') + self.send2socket(f':{dnickname} NOTICE {nickname} :\x01VERSION Service {self._Config.SERVICE_NICKNAME} V{self._Config.CURRENT_VERSION}\x01') return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_time(self, serverMsg: list[str]) -> None: """Sending TIME answer to a requestor @@ -1361,10 +1347,10 @@ class Unrealircd6(IProtocol): # Réponse a un CTCP VERSION try: - nickname = self.__Irc.User.get_nickname(self.__Utils.clean_uid(serverMsg[1])) - dnickname = self.__Config.SERVICE_NICKNAME + nickname = self._Irc.User.get_nickname(self._Utils.clean_uid(serverMsg[1])) + dnickname = self._Config.SERVICE_NICKNAME arg = serverMsg[4].replace(':', '') - current_datetime = self.__Utils.get_sdatetime() + current_datetime = self._Utils.get_sdatetime() if nickname is None: return None @@ -1374,7 +1360,7 @@ class Unrealircd6(IProtocol): return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_ping(self, serverMsg: list[str]) -> None: """Sending a PING answer to requestor @@ -1386,30 +1372,30 @@ class Unrealircd6(IProtocol): # Réponse a un CTCP VERSION try: - nickname = self.__Irc.User.get_nickname(self.__Utils.clean_uid(serverMsg[1])) - dnickname = self.__Config.SERVICE_NICKNAME + nickname = self._Irc.User.get_nickname(self._Utils.clean_uid(serverMsg[1])) + dnickname = self._Config.SERVICE_NICKNAME arg = serverMsg[4].replace(':', '') if nickname is None: - self.__Logs.debug(serverMsg) + self._Logs.debug(serverMsg) return None if arg == '\x01PING': recieved_unixtime = int(serverMsg[5].replace('\x01','')) - current_unixtime = self.__Utils.get_unixtime() + current_unixtime = self._Utils.get_unixtime() ping_response = current_unixtime - recieved_unixtime - # self.__Irc.send2socket(f':{dnickname} NOTICE {nickname} :\x01PING {ping_response} secs\x01') + # self._Irc.send2socket(f':{dnickname} NOTICE {nickname} :\x01PING {ping_response} secs\x01') self.send_notice( nick_from=dnickname, nick_to=nickname, msg=f"\x01PING {ping_response} secs\x01" ) - self.__Logs.debug(serverMsg) + self._Logs.debug(serverMsg) return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_version_msg(self, serverMsg: list[str]) -> None: """Handle version coming from the server @@ -1423,25 +1409,25 @@ class Unrealircd6(IProtocol): if '@' in list(serverMsg_copy[0])[0]: serverMsg_copy.pop(0) - getUser = self.__Irc.User.get_user(self.__Utils.clean_uid(serverMsg_copy[0])) + getUser = self._Irc.User.get_user(self._Utils.clean_uid(serverMsg_copy[0])) if getUser is None: return None - response_351 = f"{self.__Config.SERVICE_NAME.capitalize()}-{self.__Config.CURRENT_VERSION} {self.__Config.SERVICE_HOST} {self.name}" - self.send2socket(f':{self.__Config.SERVICE_HOST} 351 {getUser.nickname} {response_351}') + response_351 = f"{self._Config.SERVICE_NAME.capitalize()}-{self._Config.CURRENT_VERSION} {self._Config.SERVICE_HOST} {self.name}" + self.send2socket(f':{self._Config.SERVICE_HOST} 351 {getUser.nickname} {response_351}') - modules = self.__Irc.ModuleUtils.get_all_available_modules() + modules = self._Irc.ModuleUtils.get_all_available_modules() response_005 = ' | '.join(modules) - self.send2socket(f':{self.__Config.SERVICE_HOST} 005 {getUser.nickname} {response_005} are supported by this server') + self.send2socket(f':{self._Config.SERVICE_HOST} 005 {getUser.nickname} {response_005} are supported by this server') - response_005 = ''.join(self.__Settings.PROTOCTL_USER_MODES) - self.send2socket(f":{self.__Config.SERVICE_HOST} 005 {getUser.nickname} {response_005} are supported by this server") + response_005 = ''.join(self._Settings.PROTOCTL_USER_MODES) + self.send2socket(f":{self._Config.SERVICE_HOST} 005 {getUser.nickname} {response_005} are supported by this server") return None except Exception as err: - self.__Logs.error(f"{__name__} - General Error: {err}") + self._Logs.error(f"{__name__} - General Error: {err}") def on_smod(self, serverMsg: list[str]) -> None: """Handle SMOD message coming from the server @@ -1456,11 +1442,11 @@ class Unrealircd6(IProtocol): for smod in modules: smod_split = smod.split(':') - sModObj = self.__Irc.Loader.Definition.MSModule(type=smod_split[0], name=smod_split[1], version=smod_split[2]) - self.__Settings.SMOD_MODULES.append(sModObj) + sModObj = self._Irc.Loader.Definition.MSModule(type=smod_split[0], name=smod_split[1], version=smod_split[2]) + self._Settings.SMOD_MODULES.append(sModObj) except Exception as err: - self.__Logs.error(f'General Error: {err}') + self._Logs.error(f'General Error: {err}') def on_sasl(self, serverMsg: list[str]) -> Optional['MSasl']: """Handle SASL coming from a server @@ -1475,9 +1461,9 @@ class Unrealircd6(IProtocol): # [':irc.local.org', 'SASL', 'defender-dev.deb.biz.st', '0014ZZH1F', 'S', 'EXTERNAL', 'zzzzzzzkey'] # [':irc.local.org', 'SASL', 'defender-dev.deb.biz.st', '00157Z26U', 'C', 'sasakey=='] # [':irc.local.org', 'SASL', 'defender-dev.deb.biz.st', '00157Z26U', 'D', 'A'] - psasl = self.__Irc.Sasl + psasl = self._Irc.Sasl sasl_enabled = False - for smod in self.__Settings.SMOD_MODULES: + for smod in self._Settings.SMOD_MODULES: if smod.name == 'sasl': sasl_enabled = True break @@ -1489,7 +1475,7 @@ class Unrealircd6(IProtocol): client_uid = sCopy[3] if len(sCopy) >= 6 else None sasl_obj = None sasl_message_type = sCopy[4] if len(sCopy) >= 6 else None - psasl.insert_sasl_client(self.__Irc.Loader.Definition.MSasl(client_uid=client_uid)) + psasl.insert_sasl_client(self._Irc.Loader.Definition.MSasl(client_uid=client_uid)) sasl_obj = psasl.get_sasl_obj(client_uid) if sasl_obj is None: @@ -1507,13 +1493,13 @@ class Unrealircd6(IProtocol): sasl_obj.mechanisme = str(sCopy[5]) if sasl_obj.mechanisme == "PLAIN": - self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {sasl_obj.client_uid} C +") + self.send2socket(f":{self._Config.SERVEUR_LINK} SASL {self._Settings.MAIN_SERVER_HOSTNAME} {sasl_obj.client_uid} C +") elif sasl_obj.mechanisme == "EXTERNAL": if str(sCopy[5]) == "+": return None sasl_obj.fingerprint = str(sCopy[6]) - self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {sasl_obj.client_uid} C +") + self.send2socket(f":{self._Config.SERVEUR_LINK} SASL {self._Settings.MAIN_SERVER_HOSTNAME} {sasl_obj.client_uid} C +") self.on_sasl_authentication_process(sasl_obj) return sasl_obj @@ -1537,7 +1523,7 @@ class Unrealircd6(IProtocol): return sasl_obj except Exception as err: - self.__Logs.error(f'General Error: {err}', exc_info=True) + self._Logs.error(f'General Error: {err}', exc_info=True) def on_sasl_authentication_process(self, sasl_model: 'MSasl') -> bool: s = sasl_model @@ -1545,12 +1531,12 @@ class Unrealircd6(IProtocol): def db_get_admin_info(*, username: Optional[str] = None, password: Optional[str] = None, fingerprint: Optional[str] = None) -> Optional[dict[str, Any]]: if fingerprint: mes_donnees = {'fingerprint': fingerprint} - query = f"SELECT user, level, language FROM {self.__Config.TABLE_ADMIN} WHERE fingerprint = :fingerprint" + query = f"SELECT user, level, language FROM {self._Config.TABLE_ADMIN} WHERE fingerprint = :fingerprint" else: - mes_donnees = {'user': username, 'password': self.__Utils.hash_password(password)} - query = f"SELECT user, level, language FROM {self.__Config.TABLE_ADMIN} WHERE user = :user AND password = :password" + mes_donnees = {'user': username, 'password': self._Utils.hash_password(password)} + query = f"SELECT user, level, language FROM {self._Config.TABLE_ADMIN} WHERE user = :user AND password = :password" - result = self.__Base.db_execute_query(query, mes_donnees) + result = self._Base.db_execute_query(query, mes_donnees) user_from_db = result.fetchone() if user_from_db: return {'user': user_from_db[0], 'level': user_from_db[1], 'language': user_from_db[2]} @@ -1564,11 +1550,11 @@ class Unrealircd6(IProtocol): s.auth_success = True s.level = admin_info.get('level', 0) s.language = admin_info.get('language', 'EN') - self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S") - self.send2socket(f":{self.__Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful") + self.send2socket(f":{self._Config.SERVEUR_LINK} SASL {self._Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S") + self.send2socket(f":{self._Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful") else: - self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F") - self.send2socket(f":{self.__Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed") + self.send2socket(f":{self._Config.SERVEUR_LINK} SASL {self._Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F") + self.send2socket(f":{self._Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed") elif s.message_type == 'S' and s.mechanisme == 'EXTERNAL': # Connection using fingerprints @@ -1579,12 +1565,12 @@ class Unrealircd6(IProtocol): s.level = admin_info.get('level', 0) s.username = admin_info.get('user', None) s.language = admin_info.get('language', 'EN') - self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S") - self.send2socket(f":{self.__Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful") + self.send2socket(f":{self._Config.SERVEUR_LINK} SASL {self._Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S") + self.send2socket(f":{self._Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful") else: # "904 :SASL authentication failed" - self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F") - self.send2socket(f":{self.__Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed") + self.send2socket(f":{self._Config.SERVEUR_LINK} SASL {self._Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F") + self.send2socket(f":{self._Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed") def on_md(self, serverMsg: list[str]) -> None: """Handle MD responses @@ -1600,7 +1586,7 @@ class Unrealircd6(IProtocol): var = str(scopy[4]).lower() value = str(scopy[5]).replace(':', '') - user_obj = self.__Irc.User.get_user(uid) + user_obj = self._Irc.User.get_user(uid) if user_obj is None: return None @@ -1614,7 +1600,7 @@ class Unrealircd6(IProtocol): ... except Exception as e: - self.__Logs.error(f"General Error: {e}") + self._Logs.error(f"General Error: {e}") def on_kick(self, serverMsg: list[str]) -> None: """When a user is kicked out from a channel @@ -1628,5 +1614,5 @@ class Unrealircd6(IProtocol): channel = scopy[3] # Delete the user from the channel. - self.__Irc.Channel.delete_user_from_channel(channel, uid) + self._Irc.Channel.delete_user_from_channel(channel, uid) return None \ No newline at end of file