diff --git a/core/classes/interfaces/iprotocol.py b/core/classes/interfaces/iprotocol.py index d730097..87b921f 100644 --- a/core/classes/interfaces/iprotocol.py +++ b/core/classes/interfaces/iprotocol.py @@ -13,7 +13,7 @@ class IProtocol(ABC): def __init__(self, uplink: 'Irc'): self.name: Optional[str] = None self.protocol_version: int = -1 - self.known_protocol: set[str] = {} + self.known_protocol: set[str] = set() self._Irc = uplink self._Config = uplink.Config @@ -32,10 +32,7 @@ class IProtocol(ABC): @abstractmethod def init_protocol(self): - """_summary_ - - Returns: - _type_: _description_ + """Init protocol """ @abstractmethod @@ -316,46 +313,46 @@ class IProtocol(ABC): # ------------------------------------------------------------------------ @abstractmethod - def parse_uid(self, serverMsg: list[str]) -> Optional['MUser']: + def parse_uid(self, server_msg: list[str]) -> Optional['MUser']: """Parse UID and return dictionary. Args: - serverMsg (list[str]): The UID IRCD message + server_msg (list[str]): The UID IRCD message Returns: Optional[MUser]: The MUser object or None """ @abstractmethod - def parse_quit(self, serverMsg: list[str]) -> tuple[Optional['MUser'], str]: + def parse_quit(self, server_msg: list[str]) -> tuple[Optional['MUser'], str]: """Parse quit and return dictionary. >>> [':97KAAAAAB', 'QUIT', ':Quit:', 'this', 'is', 'my', 'reason', 'to', 'quit'] Args: - serverMsg (list[str]): The server message to parse + server_msg (list[str]): The server message to parse Returns: tuple[MUser, str]: The User Who Quit Object and the reason. """ @abstractmethod - def parse_nick(self, serverMsg: list[str]) -> dict[str, str]: + def parse_nick(self, server_msg: list[str]) -> dict[str, str]: """Parse nick changes and return dictionary. >>> [':97KAAAAAC', 'NICK', 'testinspir', '1757360740'] Args: - serverMsg (list[str]): The server message to parse + server_msg (list[str]): The server message to parse Returns: dict[str, str]: The response as dictionary. """ @abstractmethod - def parse_privmsg(self, serverMsg: list[str]) -> tuple[Optional['MUser'], Optional['MUser'], Optional['MChannel'], str]: + def parse_privmsg(self, server_msg: list[str]) -> tuple[Optional['MUser'], Optional['MUser'], Optional['MChannel'], str]: """Parse PRIVMSG message. >>> [':97KAAAAAE', 'PRIVMSG', '#welcome', ':This', 'is', 'my', 'public', 'message'] Args: - serverMsg (list[str]): The server message to parse + server_msg (list[str]): The server message to parse Returns: tuple[MUser(Sender), MUser(Reciever), MChannel, str]: Sender user model, reciever user model, Channel model, messgae. @@ -366,174 +363,174 @@ class IProtocol(ABC): # ------------------------------------------------------------------------ @abstractmethod - def on_svs2mode(self, serverMsg: list[str]) -> None: + def on_svs2mode(self, server_msg: list[str]) -> None: """Handle svs2mode coming from a server >>> [':00BAAAAAG', 'SVS2MODE', '001U01R03', '-r'] Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_mode(self, serverMsg: list[str]) -> None: + def on_mode(self, server_msg: list[str]) -> None: """Handle mode coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_umode2(self, serverMsg: list[str]) -> None: + def on_umode2(self, server_msg: list[str]) -> None: """Handle umode2 coming from a server >>> [':adator_', 'UMODE2', '-i'] Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_quit(self, serverMsg: list[str]) -> None: + def on_quit(self, server_msg: list[str]) -> None: """Handle quit coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_squit(self, serverMsg: list[str]) -> None: + def on_squit(self, server_msg: list[str]) -> None: """Handle squit coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_protoctl(self, serverMsg: list[str]) -> None: + def on_protoctl(self, server_msg: list[str]) -> None: """Handle protoctl coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_nick(self, serverMsg: list[str]) -> None: + def on_nick(self, server_msg: list[str]) -> None: """Handle nick coming from a server new nickname Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_sjoin(self, serverMsg: list[str]) -> None: + def on_sjoin(self, server_msg: list[str]) -> None: """Handle sjoin coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_part(self, serverMsg: list[str]) -> None: + def on_part(self, server_msg: list[str]) -> None: """Handle part coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_eos(self, serverMsg: list[str]) -> None: + def on_eos(self, server_msg: list[str]) -> None: """Handle EOS coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_reputation(self, serverMsg: list[str]) -> None: + def on_reputation(self, server_msg: list[str]) -> None: """Handle REPUTATION coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_uid(self, serverMsg: list[str]) -> None: + def on_uid(self, server_msg: list[str]) -> None: """Handle uid message coming from the server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_privmsg(self, serverMsg: list[str]) -> None: + def on_privmsg(self, server_msg: list[str]) -> None: """Handle PRIVMSG message coming from the server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_server_ping(self, serverMsg: list[str]) -> None: + def on_server_ping(self, server_msg: list[str]) -> None: """Send a PONG message to the server Args: - serverMsg (list[str]): List of str coming from the server + server_msg (list[str]): List of str coming from the server """ @abstractmethod - def on_server(self, serverMsg: list[str]) -> None: + def on_server(self, server_msg: list[str]) -> None: """_summary_ Args: - serverMsg (list[str]): _description_ + server_msg (list[str]): _description_ """ @abstractmethod - def on_version(self, serverMsg: list[str]) -> None: + def on_version(self, server_msg: list[str]) -> None: """Sending Server Version to the server Args: - serverMsg (list[str]): List of str coming from the server + server_msg (list[str]): List of str coming from the server """ @abstractmethod - def on_time(self, serverMsg: list[str]) -> None: + def on_time(self, server_msg: list[str]) -> None: """Sending TIME answer to a requestor Args: - serverMsg (list[str]): List of str coming from the server + server_msg (list[str]): List of str coming from the server """ @abstractmethod - def on_ping(self, serverMsg: list[str]) -> None: + def on_ping(self, server_msg: list[str]) -> None: """Sending a PING answer to requestor Args: - serverMsg (list[str]): List of str coming from the server + server_msg (list[str]): List of str coming from the server """ @abstractmethod - def on_version_msg(self, serverMsg: list[str]) -> None: + def on_version_msg(self, server_msg: list[str]) -> None: """Handle version coming from the server \n ex. /version Defender Args: - serverMsg (list[str]): Original message from the server + server_msg (list[str]): Original message from the server """ @abstractmethod - def on_smod(self, serverMsg: list[str]) -> None: + def on_smod(self, server_msg: list[str]) -> None: """Handle SMOD message coming from the server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message """ @abstractmethod - def on_sasl(self, serverMsg: list[str]) -> Optional['MSasl']: + def on_sasl(self, server_msg: list[str]) -> Optional['MSasl']: """Handle SASL coming from a server Args: - serverMsg (list[str]): Original server message + server_msg (list[str]): Original server message Returns: @@ -551,18 +548,18 @@ class IProtocol(ABC): """ @abstractmethod - def on_md(self, serverMsg: list[str]) -> None: + def on_md(self, server_msg: list[str]) -> None: """Handle MD responses [':001', 'MD', 'client', '001MYIZ03', 'certfp', ':d1235648...'] Args: - serverMsg (list[str]): The server reply + server_msg (list[str]): The server reply """ @abstractmethod - def on_kick(self, serverMsg: list[str]) -> None: + def on_kick(self, server_msg: list[str]) -> None: """When a user is kicked out from a channel Eg. ['@unrealircd.org...', ':001', 'KICK', '#jsonrpc', '001ELW13T', ':Kicked', 'from', 'JSONRPC', 'User'] Args: - serverMsg (list[str]): The server message + server_msg (list[str]): The server message """ diff --git a/core/classes/protocols/inspircd.py b/core/classes/protocols/inspircd.py index a921b50..7ff0e54 100644 --- a/core/classes/protocols/inspircd.py +++ b/core/classes/protocols/inspircd.py @@ -254,7 +254,7 @@ class Inspircd(IProtocol): self.send2socket(f":{server_id} SQUIT {server_link} :{reason}") return None - def send_ungline(self, nickname:str, hostname: str) -> None: + def send_ungline(self, nickname: str, hostname: str) -> None: self.send2socket(f":{self._Config.SERVEUR_ID} TKL - G {nickname} {hostname} {self._Config.SERVICE_NICKNAME}") @@ -312,7 +312,9 @@ class Inspircd(IProtocol): return None - def send_uid(self, nickname:str, username: str, hostname: str, uid:str, umodes: str, vhost: str, remote_ip: str, realname: str, print_log: bool = True) -> None: + def send_uid(self, nickname: str, username: str, hostname: str, + uid:str, umodes: str, vhost: str, remote_ip: str, + realname: str, print_log: bool = True) -> None: """Send UID to the server [:] UID []+ : Args: @@ -384,7 +386,7 @@ class Inspircd(IProtocol): self._Irc.Channel.insert(self._Irc.Loader.Definition.MChannel(name=channel, uids=[user_obj.uid])) return None - def send_part_chan(self, uidornickname:str, channel: str, print_log: bool = True) -> None: + def send_part_chan(self, uidornickname: str, channel: str, print_log: bool = True) -> None: """Part from a channel Args: @@ -409,7 +411,7 @@ class Inspircd(IProtocol): self._Irc.Channel.delete_user_from_channel(channel, user_obj.uid) return None - def send_unkline(self, nickname:str, hostname: str) -> 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}") @@ -1263,7 +1265,6 @@ class Inspircd(IProtocol): return sender, reciever, channel, message - # ------------------------------------------------------------------------ # IRC SENDER METHODS # ------------------------------------------------------------------------ diff --git a/core/classes/protocols/unreal6.py b/core/classes/protocols/unreal6.py index e239a7b..c80552e 100644 --- a/core/classes/protocols/unreal6.py +++ b/core/classes/protocols/unreal6.py @@ -1,5 +1,4 @@ from base64 import b64decode -from optparse import Option from re import match, findall, search from datetime import datetime from typing import TYPE_CHECKING, Any, Optional @@ -10,7 +9,6 @@ from core.utils import tr if TYPE_CHECKING: from core.classes.modules.sasl import Sasl from core.definition import MClient, MSasl, MUser, MChannel - from core.loader import Loader class Unrealircd6(IProtocol): @@ -106,7 +104,8 @@ class Unrealircd6(IProtocol): """Envoit les commandes à envoyer au serveur. Args: - string (Str): contient la commande à envoyer au serveur. + message (str): contient la commande à envoyer au serveur. + print_log (bool): True print log message in the console """ try: with self._Base.lock: @@ -142,22 +141,22 @@ class Unrealircd6(IProtocol): """ 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 + 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: + if user_from is None: self._Logs.error(f"The sender nickname [{nick_from}] do not exist") return None if not channel is None: for i in range(0, len(str(msg)), batch_size): batch = str(msg)[i:i+batch_size] - self.send2socket(f":{User_from.uid} PRIVMSG {channel} :{batch}") + self.send2socket(f":{user_from.uid} PRIVMSG {channel} :{batch}") if not nick_to is None: for i in range(0, len(str(msg)), batch_size): batch = str(msg)[i:i+batch_size] - self.send2socket(f":{nick_from} PRIVMSG {User_to.uid} :{batch}") + self.send2socket(f":{nick_from} PRIVMSG {user_to.uid} :{batch}") except Exception as err: self._Logs.error(f"General Error: {err}") @@ -173,16 +172,16 @@ class Unrealircd6(IProtocol): """ 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) + 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: + 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") return None for i in range(0, len(str(msg)), batch_size): batch = str(msg)[i:i+batch_size] - self.send2socket(f":{User_from.uid} NOTICE {User_to.uid} :{batch}") + self.send2socket(f":{user_from.uid} NOTICE {user_to.uid} :{batch}") except Exception as err: self._Logs.error(f"General Error: {err}") @@ -199,7 +198,7 @@ class Unrealircd6(IProtocol): 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_umodes = self._Config.SERVICE_UMODES service_hostname = self._Config.SERVICE_HOST service_name = self._Config.SERVICE_NAME protocolversion = self.protocol_version @@ -251,8 +250,8 @@ class Unrealircd6(IProtocol): """ 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) + user_obj = self._Irc.User.get_user(self._Config.SERVICE_NICKNAME) + self._Irc.User.update_nickname(user_obj.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: @@ -339,21 +338,20 @@ class Unrealircd6(IProtocol): """_summary_ Args: - from_nick (str): _description_ - nick_to (str): _description_ + nick_to_sapart (str): _description_ channel_name (str): _description_ """ try: - userObj = self._Irc.User.get_user(uidornickname=nick_to_sapart) - chanObj = self._Irc.Channel.get_channel(channel_name) + user_obj = self._Irc.User.get_user(uidornickname=nick_to_sapart) + chan_obj = self._Irc.Channel.get_channel(channel_name) service_uid = self._Config.SERVICE_ID - if userObj is None or chanObj is None: + if user_obj is None or chan_obj 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.send2socket(f":{service_uid} SAPART {user_obj.nickname} {chan_obj.name}") + self._Irc.Channel.delete_user_from_channel(chan_obj.name, user_obj.uid) return None @@ -369,28 +367,28 @@ class Unrealircd6(IProtocol): """ try: - userObj = self._Irc.User.get_user(uidornickname=nick_to_sajoin) - chanObj = self._Irc.Channel.get_channel(channel_name) + user_obj = self._Irc.User.get_user(uidornickname=nick_to_sajoin) + chan_obj = self._Irc.Channel.get_channel(channel_name) service_uid = self._Config.SERVICE_ID - if userObj is None: + if user_obj is None: # User not exist: leave return None - if chanObj is None: + if chan_obj is None: # Channel not exist 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) - self.send2socket(f":{service_uid} SAJOIN {userObj.nickname} {newChanObj.name}") + new_chan_obj = self._Irc.Loader.Definition.MChannel(name=channel_name, uids=[user_obj.uid]) + self._Irc.Channel.insert(new_chan_obj) + self.send2socket(f":{service_uid} SAJOIN {user_obj.nickname} {new_chan_obj.name}") else: - 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}") + self._Irc.Channel.add_user_to_a_channel(channel_name=channel_name, uid=user_obj.uid) + self.send2socket(f":{service_uid} SAJOIN {user_obj.nickname} {chan_obj.name}") return None @@ -472,7 +470,7 @@ class Unrealircd6(IProtocol): """Logout a client from his account Args: - client_uid (str): The Client UID + client_obj (MClient): The Client object """ try: c_uid = client_obj.uid @@ -483,31 +481,33 @@ class Unrealircd6(IProtocol): except Exception as err: self._Logs.error(f'General Error: {err}') - def send_quit(self, uid: str, reason: str, print_log: True) -> None: + def send_quit(self, uid: str, reason: str, print_log: bool = True) -> None: """Send quit message - Delete uid from User object - Delete uid from Reputation object Args: - uidornickname (str): The UID or the Nickname + uid (str): The UID or the Nickname reason (str): The reason for the quit + print_log (bool): Print the log """ user_obj = self._Irc.User.get_user(uidornickname=uid) - reputationObj = self._Irc.Reputation.get_reputation(uidornickname=uid) + reputation_obj = 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) - if not reputationObj is None: - self._Irc.Reputation.delete(reputationObj.uid) + if not reputation_obj is None: + self._Irc.Reputation.delete(reputation_obj.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") return None - def send_uid(self, nickname:str, username: str, hostname: str, uid:str, umodes: str, vhost: str, remote_ip: str, realname: str, print_log: bool = True) -> None: + def send_uid(self, nickname:str, username: str, hostname: str, uid:str, umodes: str, + vhost: str, remote_ip: str, realname: str, print_log: bool = True) -> None: """Send UID to the server - Insert User to User Object Args: @@ -555,34 +555,34 @@ class Unrealircd6(IProtocol): print_log (bool, optional): Write logs. Defaults to True. """ - userObj = self._Irc.User.get_user(uidornickname) - passwordChannel = password if not password is None else '' + user_obj = self._Irc.User.get_user(uidornickname) + pwd_channel = password if not password is None else '' - if userObj is None: + if user_obj is None: return None 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) + self.send2socket(f":{user_obj.uid} JOIN {channel} {pwd_channel}", 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}") # 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=[user_obj.uid])) # Set the automode to the user - if 'r' not in userObj.umodes and 'o' not in userObj.umodes: + if 'r' not in user_obj.umodes and 'o' not in user_obj.umodes: return None - db_data: dict[str, str] = {"nickname": userObj.nickname, "channel": channel} + db_data: dict[str, str] = {"nickname": user_obj.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_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}") + id_cmd_automode, mode = db_result + self.send2socket(f":{self._Config.SERVICE_ID} MODE {channel} {mode} {user_obj.nickname}") return None