9 Commits

Author SHA1 Message Date
adator
be36c56ceb Merge pull request #73 from adator85/update/protocol
Update/protocol
2025-07-20 19:44:49 +02:00
adator
ac332e6802 Update version number 2025-07-20 19:42:59 +02:00
adator
6063ceba35 Update Protocol Connection 2025-07-20 19:41:10 +02:00
adator
2204c4fdf8 Merge pull request #72 from adator85/V6.X.X
V6.1.1 Updates
2024-12-08 23:03:01 +01:00
adator
befe452df8 V6.1.1 Updates 2024-12-08 23:02:36 +01:00
adator
b98a20ad45 Merge pull request #71 from adator85/V6.X.X
V6.1.0 update the help command
2024-12-08 12:56:18 +01:00
adator
cb042a5411 V6.1.0 update the help command 2024-12-08 12:52:36 +01:00
adator
dc63df08cf Merge pull request #70 from adator85/V6.X.X
Update on_version method
2024-11-22 23:27:15 +01:00
adator
a3edf48120 Update on_version method 2024-11-22 23:26:39 +01:00
17 changed files with 774 additions and 279 deletions

View File

@@ -661,10 +661,25 @@ class Base:
)
'''
table_core_client = f'''CREATE TABLE IF NOT EXISTS {self.Config.TABLE_CLIENT} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
createdOn TEXT,
account TEXT,
nickname TEXT,
hostname TEXT,
vhost TEXT,
realname TEXT,
email TEXT,
password TEXT,
level INTEGER
)
'''
self.db_execute_query(table_core_log)
self.db_execute_query(table_core_log_command)
self.db_execute_query(table_core_module)
self.db_execute_query(table_core_admin)
self.db_execute_query(table_core_client)
self.db_execute_query(table_core_channel)
self.db_execute_query(table_core_config)
@@ -743,7 +758,7 @@ class Base:
except TypeError:
return None
def is_valid_ip(self, ip_to_control:str) -> bool:
def is_valid_ip(self, ip_to_control: str) -> bool:
try:
if ip_to_control in self.Config.WHITELISTED_IP:
@@ -754,6 +769,26 @@ class Base:
except ValueError:
return False
def is_valid_email(self, email_to_control: str) -> bool:
"""Check if the email is valid
Args:
email_to_control (str): email to control
Returns:
bool: True is the email is correct
"""
try:
pattern = '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
if re.match(pattern, email_to_control):
return True
else:
return False
except Exception as err:
self.logs.error(f'General Error: {err}')
return False
def decode_ip(self, ip_b64encoded: str) -> Union[str, None]:
binary_ip = b64decode(ip_b64encoded)
@@ -807,6 +842,27 @@ class Base:
# Vider le dictionnaire de fonction
self.periodic_func.clear()
def execute_dynamic_method(self, obj: object, method_name: str, params: list) -> None:
"""#### Ajouter les méthodes a éxecuter dans un dictionnaire
Les methodes seront exécuter par heartbeat.
Args:
obj (object): Une instance de la classe qui va etre executer
method_name (str): Le nom de la méthode a executer
params (list): les parametres a faire passer
Returns:
None: aucun retour attendu
"""
self.periodic_func[len(self.periodic_func) + 1] = {
'object': obj,
'method_name': method_name,
'param': params
}
self.logs.debug(f'Method to execute : {str(self.periodic_func)}')
return None
def clean_uid(self, uid:str) -> Union[str, None]:
"""Clean UID by removing @ / % / + / ~ / * / :

243
core/classes/client.py Normal file
View File

@@ -0,0 +1,243 @@
from re import sub
from typing import Union, TYPE_CHECKING
from dataclasses import asdict
if TYPE_CHECKING:
from core.base import Base
from core.definition import MClient
class Client:
CLIENT_DB: list['MClient'] = []
def __init__(self, baseObj: 'Base') -> None:
self.Logs = baseObj.logs
self.Base = baseObj
return None
def insert(self, newUser: 'MClient') -> bool:
"""Insert a new User object
Args:
newUser (UserModel): New userModel object
Returns:
bool: True if inserted
"""
userObj = self.get_Client(newUser.uid)
if not userObj is None:
# User already created return False
return False
self.CLIENT_DB.append(newUser)
return True
def update_nickname(self, uid: str, newNickname: str) -> bool:
"""Update the nickname starting from the UID
Args:
uid (str): UID of the user
newNickname (str): New nickname
Returns:
bool: True if updated
"""
userObj = self.get_Client(uidornickname=uid)
if userObj is None:
return False
userObj.nickname = newNickname
return True
def update_mode(self, uidornickname: str, modes: str) -> bool:
"""Updating user mode
Args:
uidornickname (str): The UID or Nickname of the user
modes (str): new modes to update
Returns:
bool: True if user mode has been updaed
"""
response = True
userObj = self.get_Client(uidornickname=uidornickname)
if userObj is None:
return False
action = modes[0]
new_modes = modes[1:]
existing_umodes = userObj.umodes
umodes = userObj.umodes
if action == '+':
for nm in new_modes:
if nm not in existing_umodes:
umodes += nm
elif action == '-':
for nm in new_modes:
if nm in existing_umodes:
umodes = umodes.replace(nm, '')
else:
return False
liste_umodes = list(umodes)
final_umodes_liste = [x for x in self.Base.Settings.PROTOCTL_USER_MODES if x in liste_umodes]
final_umodes = ''.join(final_umodes_liste)
userObj.umodes = f"+{final_umodes}"
return response
def delete(self, uid: str) -> bool:
"""Delete the User starting from the UID
Args:
uid (str): UID of the user
Returns:
bool: True if deleted
"""
userObj = self.get_Client(uidornickname=uid)
if userObj is None:
return False
self.CLIENT_DB.remove(userObj)
return True
def get_Client(self, uidornickname: str) -> Union['MClient', None]:
"""Get The Client Object model
Args:
uidornickname (str): UID or Nickname
Returns:
UserModel|None: The UserModel Object | None
"""
User = None
for record in self.CLIENT_DB:
if record.uid == uidornickname:
User = record
elif record.nickname == uidornickname:
User = record
return User
def get_uid(self, uidornickname:str) -> Union[str, None]:
"""Get the UID of the user starting from the UID or the Nickname
Args:
uidornickname (str): UID or Nickname
Returns:
str|None: Return the UID
"""
userObj = self.get_Client(uidornickname=uidornickname)
if userObj is None:
return None
return userObj.uid
def get_nickname(self, uidornickname:str) -> Union[str, None]:
"""Get the Nickname starting from UID or the nickname
Args:
uidornickname (str): UID or Nickname of the user
Returns:
str|None: the nickname
"""
userObj = self.get_Client(uidornickname=uidornickname)
if userObj is None:
return None
return userObj.nickname
def get_Client_AsDict(self, uidornickname: str) -> Union[dict[str, any], None]:
"""Transform User Object to a dictionary
Args:
uidornickname (str): The UID or The nickname
Returns:
Union[dict[str, any], None]: User Object as a dictionary or None
"""
userObj = self.get_Client(uidornickname=uidornickname)
if userObj is None:
return None
return asdict(userObj)
def is_exist(self, uidornikname: str) -> bool:
"""Check if the UID or the nickname exist in the USER DB
Args:
uidornickname (str): The UID or the NICKNAME
Returns:
bool: True if exist
"""
userObj = self.get_Client(uidornickname=uidornikname)
if userObj is None:
return False
return True
def db_is_account_exist(self, account: str) -> bool:
"""Check if the account exist in the database
Args:
account (str): The account to check
Returns:
bool: True if exist
"""
table_client = self.Base.Config.TABLE_CLIENT
account_to_check = {'account': account.lower()}
account_to_check_query = self.Base.db_execute_query(f"""
SELECT id FROM {table_client} WHERE LOWER(account) = :account
""", account_to_check)
account_to_check_result = account_to_check_query.fetchone()
if account_to_check_result:
self.Logs.error(f"Account ({account}) already exist")
return True
return False
def clean_uid(self, uid: str) -> Union[str, None]:
"""Clean UID by removing @ / % / + / ~ / * / :
Args:
uid (str): The UID to clean
Returns:
str: Clean UID without any sign
"""
pattern = fr'[:|@|%|\+|~|\*]*'
parsed_UID = sub(pattern, '', uid)
if not parsed_UID:
return None
return parsed_UID

View File

@@ -401,7 +401,9 @@ class Inspircd:
uid = str(serverMsg[1]).lstrip(':')
newnickname = serverMsg[3]
self.__Irc.User.update(uid, newnickname)
self.__Irc.User.update_nickname(uid, newnickname)
self.__Irc.Client.update_nickname(uid, newnickname)
self.__Irc.Admin.update_nickname(uid, newnickname)
return None

View File

@@ -2,9 +2,6 @@ from re import match, findall, search
from datetime import datetime
from typing import TYPE_CHECKING, Union
from ssl import SSLEOFError, SSLError
from dataclasses import dataclass
from websockets import serve
if TYPE_CHECKING:
from core.irc import Irc
@@ -13,6 +10,7 @@ class Unrealircd6:
def __init__(self, ircInstance: 'Irc'):
self.name = 'UnrealIRCD-6'
self.protocol_version = 6100
self.__Irc = ircInstance
self.__Config = ircInstance.Config
@@ -161,6 +159,7 @@ class Unrealircd6:
umodes = self.__Config.SERVICE_UMODES
host = self.__Config.SERVICE_HOST
service_name = self.__Config.SERVICE_NAME
protocolversion = self.protocol_version
password = self.__Config.SERVEUR_PASSWORD
link = self.__Config.SERVEUR_LINK
@@ -172,16 +171,15 @@ class Unrealircd6:
self.send2socket(f":{server_id} PASS :{password}", print_log=False)
self.send2socket(f":{server_id} PROTOCTL SID NOQUIT NICKv2 SJOIN SJ3 NICKIP TKLEXT2 NEXTBANS CLK EXTSWHOIS MLOCK MTAGS")
# self.__Irc.send2socket(f":{sid} PROTOCTL NICKv2 VHP UMODE2 NICKIP SJOIN SJOIN2 SJ3 NOQUIT TKLEXT MLOCK SID MTAGS")
self.send2socket(f":{server_id} PROTOCTL EAUTH={link},,,{service_name}-v{version}")
self.send2socket(f":{server_id} PROTOCTL EAUTH={link},{protocolversion},,{service_name}-v{version}")
self.send2socket(f":{server_id} PROTOCTL SID={server_id}")
self.send2socket(f":{server_id} PROTOCTL BOOTED={unixtime}")
self.send2socket(f":{server_id} SERVER {link} 1 :{info}")
self.send2socket(f":{server_id} {nickname} :Reserved for services")
self.send2socket(f":{server_id} UID {nickname} 1 {unixtime} {username} {host} {service_id} * {smodes} * * fwAAAQ== :{realname}")
self.sjoin(chan)
self.send2socket(f":{server_id} TKL + Q * {nickname} {host} 0 {unixtime} :Reserved for services")
self.send2socket(f":{service_id} MODE {chan} {cmodes}")
# self.send2socket(f":{service_id} MODE {chan} {umodes} {service_id}")
self.__Base.logs.debug(f'>> {__name__} Link information sent to the server')
@@ -201,7 +199,7 @@ class Unrealircd6:
self.send2socket(f":{self.__Config.SERVICE_NICKNAME} NICK {newnickname}")
userObj = self.__Irc.User.get_User(self.__Config.SERVICE_NICKNAME)
self.__Irc.User.update(userObj.uid, newnickname)
self.__Irc.User.update_nickname(userObj.uid, newnickname)
return None
def squit(self, server_id: str, server_link: str, reason: str) -> None:
@@ -423,6 +421,18 @@ class Unrealircd6:
# Add defender to the channel uids list
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_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}")
return None
def send_part_chan(self, uidornickname:str, channel: str, print_log: bool = True) -> None:
@@ -450,6 +460,26 @@ class Unrealircd6:
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_Channel(channelToCheck=channel_name)
if not channel:
self.__Base.logs.error(f'The channel [{channel_name}] is not correct')
return None
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}")
return None
#####################
# HANDLE EVENTS #
#####################
def on_svs2mode(self, serverMsg: list[str]) -> None:
"""Handle svs2mode coming from a server
@@ -530,6 +560,7 @@ class Unrealircd6:
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.Clone.delete(uid_who_quit)
@@ -611,7 +642,9 @@ class Unrealircd6:
uid = str(serverMsg[1]).lstrip(':')
newnickname = serverMsg[3]
self.__Irc.User.update(uid, newnickname)
self.__Irc.User.update_nickname(uid, newnickname)
self.__Irc.Client.update_nickname(uid, newnickname)
self.__Irc.Admin.update_nickname(uid, newnickname)
return None
@@ -780,7 +813,7 @@ class Unrealircd6:
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.first_score
user.score_connexion = self.__Irc.first_score
else:
self.__Irc.first_score = int(server_msg_copy[3])
@@ -865,11 +898,10 @@ class Unrealircd6:
try:
srv_msg = serverMsg.copy()
# Supprimer la premiere valeur
if srv_msg[0].startswith('@'):
srv_msg.pop(0)
cmd = srv_msg
cmd = serverMsg.copy()
# Supprimer la premiere valeur si MTAGS activé
if cmd[0].startswith('@'):
cmd.pop(0)
# Hide auth logs
if len(cmd) == 7:
@@ -894,7 +926,7 @@ class Unrealircd6:
convert_to_string = ' '.join(liste_des_commandes)
arg = convert_to_string.split()
arg.remove(f':{self.__Config.SERVICE_PREFIX}')
if not arg[0].lower() in self.__Irc.commands:
if not arg[0].lower() in self.__Irc.module_commands_list:
self.__Base.logs.debug(f"This command {arg[0]} is not available")
self.send_notice(
nick_from=self.__Config.SERVICE_NICKNAME,
@@ -933,7 +965,7 @@ class Unrealircd6:
self.on_ping(srv_msg)
return False
if not arg[0].lower() in self.__Irc.commands:
if not arg[0].lower() in self.__Irc.module_commands_list:
self.__Base.logs.debug(f"This command {arg[0]} sent by {user_trigger} is not available")
return False
@@ -949,8 +981,10 @@ class Unrealircd6:
except KeyError as ke:
self.__Base.logs.error(f"Key Error: {ke}")
except AttributeError as ae:
self.__Base.logs.error(f"Attribute Error: {ae}")
except Exception as err:
self.__Base.logs.error(f"General Error: {err}")
self.__Base.logs.error(f"General Error: {err} - {srv_msg}")
def on_server_ping(self, serverMsg: list[str]) -> None:
"""Send a PONG message to the server

View File

@@ -3,13 +3,13 @@ from socket import socket
class Settings:
RUNNING_TIMERS: list[Timer] = []
RUNNING_THREADS: list[Thread] = []
RUNNING_SOCKETS: list[socket] = []
PERIODIC_FUNC: dict[object] = {}
LOCK: RLock = RLock()
RUNNING_TIMERS: list[Timer] = []
RUNNING_THREADS: list[Thread] = []
RUNNING_SOCKETS: list[socket] = []
PERIODIC_FUNC: dict[object] = {}
LOCK: RLock = RLock()
CONSOLE: bool = False
CONSOLE: bool = False
PROTOCTL_USER_MODES: list[str] = []
PROTOCTL_PREFIX: list[str] = []
PROTOCTL_USER_MODES: list[str] = []
PROTOCTL_PREFIX: list[str] = []

View File

@@ -37,7 +37,7 @@ class User:
return True
def update(self, uid: str, newNickname: str) -> bool:
def update_nickname(self, uid: str, newNickname: str) -> bool:
"""Update the nickname starting from the UID
Args:

View File

@@ -3,6 +3,24 @@ from dataclasses import dataclass, field
from typing import Literal
from os import sep
@dataclass
class MClient:
"""Model Client for registred nickname"""
uid: str = None
account: str = None
nickname: str = None
username: str = None
realname: str = None
hostname: str = None
umodes: str = None
vhost: str = None
isWebirc: bool = False
isWebsocket: bool = False
remote_ip: str = None
score_connexion: int = 0
geoip: str = None
connexion_datetime: datetime = field(default=datetime.now())
@dataclass
class MUser:
"""Model User"""
@@ -83,6 +101,7 @@ class ColorModel:
yellow: str = "\x0306"
bold: str = "\x02"
nogc: str = "\x03"
underline: str = "\x1F"
@dataclass
class MConfig:
@@ -214,8 +233,11 @@ class MConfig:
LOGGING_NAME: str = "defender"
"""The name of the Logging instance"""
TABLE_CLIENT: str = "core_client"
"""Core Client table"""
TABLE_ADMIN: str = "core_admin"
"""Admin table"""
"""Core Admin table"""
TABLE_COMMAND: str = "core_command"
"""Core command table"""

View File

@@ -1,5 +1,3 @@
from ast import parse
from http import server
import sys
import socket
import threading
@@ -11,8 +9,6 @@ import traceback
from ssl import SSLSocket
from datetime import datetime, timedelta
from typing import Union
from websockets import serve
from core.loader import Loader
from core.classes.protocol import Protocol
@@ -65,6 +61,9 @@ class Irc:
# Use Admin Instance
self.Admin = self.Loader.Admin
# Use Client Instance
self.Client = self.Loader.Client
# Use Channel Instance
self.Channel = self.Loader.Channel
@@ -80,27 +79,51 @@ class Irc:
# define first reputation score to 0
self.first_score: int = 0
# Define first IP connexion
self.first_connexion_ip: str = None
# Define the dict that will contain all loaded modules
self.loaded_classes:dict[str, 'Irc'] = {} # Definir la variable qui contiendra la liste modules chargés
# Global full module commands that contains level, module name, commands and description
self.module_commands: dict[int, dict[str, dict[str, str]]] = {}
# Global command list contains only the commands
self.module_commands_list: list[str] = []
self.build_command(0, 'core', 'help', 'This provide the help')
self.build_command(0, 'core', 'auth', 'Login to the IRC Service')
self.build_command(0, 'core', 'copyright', 'Give some information about the IRC Service')
self.build_command(0, 'core', 'uptime', 'Give you since when the service is connected')
self.build_command(0, 'core', 'firstauth', 'First authentication of the Service')
self.build_command(0, 'core', 'register', f'Register your nickname /msg {self.Config.SERVICE_NICKNAME} REGISTER <password> <email>')
self.build_command(0, 'core', 'identify', f'Identify yourself with your password /msg {self.Config.SERVICE_NICKNAME} IDENTIFY <account> <password>')
self.build_command(0, 'core', 'logout', 'Reverse the effect of the identify command')
self.build_command(1, 'core', 'load', 'Load an existing module')
self.build_command(1, 'core', 'unload', 'Unload a module')
self.build_command(1, 'core', 'reload', 'Reload a module')
self.build_command(1, 'core', 'deauth', 'Deauth from the irc service')
self.build_command(1, 'core', 'checkversion', 'Check the version of the irc service')
self.build_command(2, 'core', 'show_modules', 'Display a list of loaded modules')
self.build_command(2, 'core', 'show_timers', 'Display active timers')
self.build_command(2, 'core', 'show_threads', 'Display active threads in the system')
self.build_command(2, 'core', 'show_channels', 'Display a list of active channels')
self.build_command(2, 'core', 'show_users', 'Display a list of connected users')
self.build_command(2, 'core', 'show_clients', 'Display a list of connected clients')
self.build_command(2, 'core', 'show_admins', 'Display a list of administrators')
self.build_command(2, 'core', 'show_configuration', 'Display the current configuration settings')
self.build_command(3, 'core', 'quit', 'Disconnect the bot or user from the server.')
self.build_command(3, 'core', 'restart', 'Restart the bot or service.')
self.build_command(3, 'core', 'addaccess', 'Add a user or entity to an access list with specific permissions.')
self.build_command(3, 'core', 'editaccess', 'Modify permissions for an existing user or entity in the access list.')
self.build_command(3, 'core', 'delaccess', 'Remove a user or entity from the access list.')
self.build_command(4, 'core', 'rehash', 'Reload the configuration file without restarting')
self.build_command(4, 'core', 'raw', 'Send a raw command directly to the IRC server')
# Define the IrcSocket object
self.IrcSocket:Union[socket.socket, SSLSocket] = None
# Liste des commandes internes du bot
self.commands_level = {
0: ['help', 'auth', 'copyright', 'uptime', 'firstauth'],
1: ['load','reload','unload', 'deauth', 'checkversion'],
2: ['show_modules', 'show_timers', 'show_threads', 'show_channels', 'show_users', 'show_admins', 'show_configuration'],
3: ['quit', 'restart','addaccess','editaccess', 'delaccess'],
4: ['rehash']
}
# l'ensemble des commandes.
self.commands = []
for level, commands in self.commands_level.items():
for command in self.commands_level[level]:
self.commands.append(command)
self.__create_table()
self.Base.create_thread(func=self.heartbeat, func_args=(self.beat, ))
@@ -114,6 +137,7 @@ class Irc:
ircInstance (Irc): Instance of Irc object.
"""
try:
self.init_service_user()
self.__create_socket()
self.__connect_to_irc(ircInstance)
except AssertionError as ae:
@@ -305,6 +329,71 @@ class Irc:
# FIN CONNEXION IRC #
##############################################
def build_command(self, level: int, module_name: str, command_name: str, command_description: str) -> None:
"""This method build the commands variable
Args:
level (int): The Level of the command
module_name (str): The module name
command_name (str): The command name
command_description (str): The description of the command
"""
self.module_commands.setdefault(level, {}).setdefault(module_name, {}).update({command_name: command_description})
self.module_commands_list.append(command_name)
return None
def generate_help_menu(self, nickname: str) -> None:
# Check if the nickname is an admin
admin_obj = self.Admin.get_Admin(nickname)
dnickname = self.Config.SERVICE_NICKNAME
color_bold = self.Config.COLORS.bold
color_nogc = self.Config.COLORS.nogc
color_blue = self.Config.COLORS.blue
color_black = self.Config.COLORS.black
color_underline = self.Config.COLORS.underline
current_level = 0
count = 0
if admin_obj is not None:
current_level = admin_obj.level
self.Protocol.send_notice(nick_from=dnickname,nick_to=nickname, msg=f" ***************** LISTE DES COMMANDES *****************")
for level, modules in self.module_commands.items():
if level > current_level:
break
if count > 0:
self.Protocol.send_notice(nick_from=dnickname, nick_to=nickname, msg=" ")
self.Protocol.send_notice(nick_from=dnickname, nick_to=nickname, msg=f"{color_blue}{color_bold}Level {level}:{color_nogc}")
for module_name, commands in modules.items():
self.Protocol.send_notice(nick_from=dnickname, nick_to=nickname, msg=f"{color_black} {color_underline}Module: {module_name}{color_nogc}")
for command, description in commands.items():
self.Protocol.send_notice(nick_from=dnickname, nick_to=nickname, msg=f" {command:<20}: {description}")
count += 1
self.Protocol.send_notice(nick_from=dnickname,nick_to=nickname,msg=f" ***************** FIN DES COMMANDES *****************")
return None
def is_cmd_allowed(self, nickname: str, command_name: str) -> bool:
admin_obj = self.Admin.get_Admin(nickname)
current_level = 0
if admin_obj is not None:
current_level = admin_obj.level
for level, modules in self.module_commands.items():
for module_name, commands in modules.items():
for command, description in commands.items():
if command.lower() == command_name.lower() and level <= current_level:
return True
return False
def __create_table(self):
"""## Create core tables
"""
@@ -504,12 +593,6 @@ class Irc:
if class_name in self.loaded_classes:
self.loaded_classes[class_name].unload()
for level, command in self.loaded_classes[class_name].commands_level.items():
# Supprimer la commande de la variable commands
for c in self.loaded_classes[class_name].commands_level[level]:
self.commands.remove(c)
self.commands_level[level].remove(c)
del self.loaded_classes[class_name]
# Supprimer le module de la base de données
@@ -540,13 +623,6 @@ class Irc:
# Supprimer la class déja instancier
if class_name in self.loaded_classes:
# Supprimer les commandes déclarer dans la classe
for level, command in self.loaded_classes[class_name].commands_level.items():
# Supprimer la commande de la variable commands
for c in self.loaded_classes[class_name].commands_level[level]:
self.commands.remove(c)
self.commands_level[level].remove(c)
del self.loaded_classes[class_name]
my_class = getattr(the_module, class_name, None)
@@ -680,31 +756,6 @@ class Irc:
self.Logs.info(response)
return response
def is_cmd_allowed(self, nickname:str, cmd:str) -> bool:
# Vérifier si le user est identifié et si il a les droits
is_command_allowed = False
uid = self.User.get_uid(nickname)
get_admin = self.Admin.get_Admin(uid)
if not get_admin is None:
admin_level = get_admin.level
for ref_level, ref_commands in self.commands_level.items():
# print(f"LevelNo: {ref_level} - {ref_commands} - {admin_level}")
if ref_level <= int(admin_level):
# print(f"LevelNo: {ref_level} - {ref_commands}")
if cmd in ref_commands:
is_command_allowed = True
else:
for ref_level, ref_commands in self.commands_level.items():
if ref_level == 0:
# print(f"LevelNo: {ref_level} - {ref_commands}")
if cmd in ref_commands:
is_command_allowed = True
return is_command_allowed
def logs(self, log_msg:str) -> None:
"""Log to database if you want
@@ -1182,36 +1233,157 @@ class Irc:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Impossible de supprimer l'utilisateur.")
self.Logs.warning(f":{dnickname} NOTICE {fromuser} : Impossible de supprimer l'utilisateur.")
case 'register':
# Register PASSWORD EMAIL
try:
if len(cmd) < 3:
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f'/msg {self.Config.SERVICE_NICKNAME} {command.upper()} <PASSWORD> <EMAIL>'
)
return None
password = cmd[1]
email = cmd[2]
if not self.Base.is_valid_email(email_to_control=email):
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg='The email is not valid. You must provide a valid email address (first.name@email.extension)'
)
return None
user_obj = self.User.get_User(fromuser)
if user_obj is None:
self.Logs.error(f"Nickname ({fromuser}) doesn't exist, it is impossible to register this nickname")
return None
# If the account already exist.
if self.Client.db_is_account_exist(fromuser):
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f"Your account already exist, please try to login instead /msg {self.Config.SERVICE_NICKNAME} IDENTIFY <account> <password>"
)
return None
# If the account doesn't exist then insert into database
data_to_record = {
'createdOn': self.Base.get_datetime(), 'account': fromuser,
'nickname': user_obj.nickname, 'hostname': user_obj.hostname, 'vhost': user_obj.vhost, 'realname': user_obj.realname, 'email': email,
'password': self.Base.crypt_password(password=password), 'level': 0
}
insert_to_db = self.Base.db_execute_query(f"""
INSERT INTO {self.Config.TABLE_CLIENT}
(createdOn, account, nickname, hostname, vhost, realname, email, password, level)
VALUES
(:createdOn, :account, :nickname, :hostname, :vhost, :realname, :email, :password, :level)
""", data_to_record)
if insert_to_db.rowcount > 0:
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f"You have register your nickname successfully"
)
return None
except ValueError as ve:
self.Logs.error(f"Value Error : {ve}")
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f" {self.Config.SERVICE_PREFIX}{command.upper()} <PASSWORD> <EMAIL>")
case 'identify':
# Identify ACCOUNT PASSWORD
try:
if len(cmd) < 3:
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f'/msg {self.Config.SERVICE_NICKNAME} {command.upper()} <ACCOUNT> <PASSWORD>'
)
return None
account = str(cmd[1]) # account
encrypted_password = self.Base.crypt_password(cmd[2])
user_obj = self.User.get_User(fromuser)
client_obj = self.Client.get_Client(user_obj.uid)
if client_obj is not None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"You are already logged in")
return None
db_query = f"SELECT account FROM {self.Config.TABLE_CLIENT} WHERE account = :account AND password = :password"
db_param = {'account': account, 'password': encrypted_password}
exec_query = self.Base.db_execute_query(
db_query,
db_param
)
result_query = exec_query.fetchone()
if result_query:
account = result_query[0]
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"You are now logged in")
client = self.Loader.Definition.MClient(
uid=user_obj.uid, account=account, nickname=fromuser,
username=user_obj.username, realname=user_obj.realname, hostname=user_obj.hostname, umodes=user_obj.umodes, vhost=user_obj.vhost,
isWebirc=user_obj.isWebirc, isWebsocket=user_obj.isWebsocket, remote_ip=user_obj.remote_ip, score_connexion=user_obj.score_connexion,
geoip=user_obj.geoip, connexion_datetime=user_obj.connexion_datetime
)
self.Client.insert(client)
self.Protocol.send_svs_mode(nickname=fromuser, user_mode='+r')
else:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Wrong password or account")
return None
except ValueError as ve:
self.Logs.error(f"Value Error: {ve}")
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f'/msg {self.Config.SERVICE_NICKNAME} {command.upper()} <ACCOUNT> <PASSWORD>'
)
except Exception as err:
self.Logs.error(f"General Error: {err}")
case 'logout':
try:
# LOGOUT <account>
if len(cmd) < 2:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"/msg {dnickname} {command.upper()} <account>")
return None
user_obj = self.User.get_User(fromuser)
if user_obj is None:
self.Logs.error(f"The User [{fromuser}] is not available in the database")
return None
client_obj = self.Client.get_Client(user_obj.uid)
if client_obj is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="Nothing to logout. please login first")
return None
self.Protocol.send_svs_mode(nickname=fromuser, user_mode='-r')
self.Client.delete(user_obj.uid)
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"You have been logged out successfully")
except ValueError as ve:
self.Logs.error(f"Value Error: {ve}")
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"/msg {dnickname} {command.upper()} <account>")
except Exception as err:
self.Logs.error(f"General Error: {err}")
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"/msg {dnickname} {command.upper()} <account>")
case 'help':
count_level_definition = 0
get_admin = self.Admin.get_Admin(uid)
if not get_admin is None:
user_level = get_admin.level
else:
user_level = 0
self.Protocol.send_notice(nick_from=dnickname,nick_to=fromuser,msg=f" ***************** LISTE DES COMMANDES *****************")
self.Protocol.send_notice(nick_from=dnickname,nick_to=fromuser,msg=f" ")
for levDef in self.commands_level:
if int(user_level) >= int(count_level_definition):
self.Protocol.send_notice(nick_from=dnickname,nick_to=fromuser,
msg=f" ***************** {self.Config.COLORS.nogc}[ {self.Config.COLORS.green}LEVEL {str(levDef)} {self.Config.COLORS.nogc}] *****************"
)
batch = 7
for i in range(0, len(self.commands_level[count_level_definition]), batch):
groupe = self.commands_level[count_level_definition][i:i + batch] # Extraire le groupe
batch_commands = ' | '.join(groupe)
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f" {batch_commands}")
self.Protocol.send_notice(nick_from=dnickname,nick_to=fromuser,msg=f" ")
count_level_definition += 1
self.Protocol.send_notice(nick_from=dnickname,nick_to=fromuser,msg=f" ***************** FIN DES COMMANDES *****************")
self.generate_help_menu(nickname=fromuser)
case 'load':
try:
@@ -1254,6 +1426,9 @@ class Irc:
self.Base.shutdown()
self.Base.execute_periodic_action()
for chan_name in self.Channel.UID_CHANNEL_DB:
self.Protocol.send_mode_chan(chan_name.name, '-l')
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
@@ -1305,13 +1480,15 @@ class Irc:
current_version = self.Config.CURRENT_VERSION
latest_version = self.Config.LATEST_VERSION
mods = ["core.config", "core.base", "core.classes.protocols.unreal6", "core.classes.protocol"]
mods = ["core.definition", "core.config", "core.base", "core.classes.protocols.unreal6", "core.classes.protocol"]
mod_unreal6 = sys.modules['core.classes.protocols.unreal6']
mod_protocol = sys.modules['core.classes.protocol']
mod_base = sys.modules['core.base']
mod_config = sys.modules['core.classes.config']
mod_definition = sys.modules['core.definition']
importlib.reload(mod_definition)
importlib.reload(mod_config)
self.Config = self.Loader.ConfModule.Configuration().ConfigObject
self.Config.HSID = hsid
@@ -1326,7 +1503,7 @@ class Irc:
config_dict: dict = self.Config.__dict__
for key, value in conf_bkp_dict.items():
if config_dict[key] != value:
if config_dict[key] != value and key != 'COLORS':
self.Protocol.send_priv_msg(
nick_from=self.Config.SERVICE_NICKNAME,
msg=f'[{key}]: {value} ==> {config_dict[key]}',
@@ -1439,6 +1616,16 @@ class Irc:
msg=f"UID : {db_user.uid} - isWebirc: {db_user.isWebirc} - isWebSocket: {db_user.isWebsocket} - Nickname: {db_user.nickname} - Connection: {db_user.connexion_datetime}"
)
case 'show_clients':
count_users = len(self.Client.CLIENT_DB)
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Total Connected Clients: {count_users}")
for db_client in self.Client.CLIENT_DB:
self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f"UID : {db_client.uid} - isWebirc: {db_client.isWebirc} - isWebSocket: {db_client.isWebsocket} - Nickname: {db_client.nickname} - Account: {db_client.account} - Connection: {db_client.connexion_datetime}"
)
case 'show_admins':
for db_admin in self.Admin.UID_ADMIN_DB:
@@ -1481,5 +1668,9 @@ class Irc:
(fromuser, )
)
case 'raw':
raw_command = ' '.join(cmd[1:])
self.Protocol.send_raw(raw_command)
case _:
pass

View File

@@ -1,4 +1,4 @@
from core.classes import user, admin, channel, clone, reputation, settings
from core.classes import user, admin, client, channel, clone, reputation, settings
import core.definition as df
import core.base as baseModule
import core.classes.config as confModule
@@ -23,6 +23,8 @@ class Loader:
self.User: user.User = user.User(self.Base)
self.Client: client.Client = client.Client(self.Base)
self.Admin: admin.Admin = admin.Admin(self.Base)
self.Channel: channel.Channel = channel.Channel(self.Base)

View File

@@ -1,30 +1,21 @@
from core import installation
#############################################
# @Version : 1 #
# @Version : 6 #
# Requierements : #
# Python3.10 or higher #
# SQLAlchemy, requests, psutil #
# UnrealIRCD 6.2.2 or higher #
#############################################
#########################
# LANCEMENT DE DEFENDER #
#########################
# 1. Chargement de la configuration
# 2. Chargement de l'ensemble des classes
# 3.
#
try:
installation.Install()
from core.loader import Loader
from core.irc import Irc
loader = Loader()
ircInstance = Irc(loader)
# loader = Loader()
ircInstance = Irc(Loader())
ircInstance.init_irc(ircInstance)
except AssertionError as ae:

View File

@@ -1,7 +1,5 @@
from dataclasses import dataclass, fields, field
import copy
from dataclasses import dataclass
import random, faker, time, logging
from datetime import datetime
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -45,9 +43,7 @@ class Clone():
self.Definition = ircInstance.Loader.Definition
# Créer les nouvelles commandes du module
self.commands_level = {
1: ['clone']
}
self.Irc.build_command(1, self.module_name, 'clone', 'Connect, join, part, kill and say clones')
# Init the module (Mandatory)
self.__init_module()
@@ -57,9 +53,6 @@ class Clone():
def __init_module(self) -> None:
# Enrigstrer les nouvelles commandes dans le code
self.__set_commands(self.commands_level)
# Créer les tables necessaire a votre module (ce n'es pas obligatoire)
self.__create_tables()
@@ -79,20 +72,6 @@ class Clone():
self.Protocol.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} +nts")
self.Protocol.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} +k {self.Config.CLONE_CHANNEL_PASSWORD}")
def __set_commands(self, commands:dict[int, list[str]]) -> None:
"""### Rajoute les commandes du module au programme principal
Args:
commands (list): Liste des commandes du module
"""
for level, com in commands.items():
for c in commands[level]:
if not c in self.Irc.commands:
self.Irc.commands_level[level].append(c)
self.Irc.commands.append(c)
return None
def __create_tables(self) -> None:
"""Methode qui va créer la base de donnée si elle n'existe pas.
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module
@@ -155,7 +134,7 @@ class Clone():
vhost = ''.join(rand_1) + '.' + ''.join(rand_2) + '.' + ''.join(rand_3) + '.IP'
return vhost
def generate_clones(self, group: str = 'Default') -> None:
def generate_clones(self, group: str = 'Default', auto_remote_ip: bool = False) -> None:
try:
fakeEN = self.fakeEN
@@ -188,7 +167,7 @@ class Clone():
department = fakeFR.department_name()
realname = f'{age} {gender} {department}'
decoded_ip = fakeEN.ipv4_private()
decoded_ip = fakeEN.ipv4_private() if auto_remote_ip else '127.0.0.1'
hostname = fakeEN.hostname()
vhost = self.generate_vhost()
@@ -231,10 +210,10 @@ class Clone():
except Exception as err:
self.Logs.error(f"General Error: {err}")
def thread_connect_clones(self, number_of_clones:int , group: str, interval: float = 0.2) -> None:
def thread_connect_clones(self, number_of_clones:int , group: str = 'Default', auto_remote_ip: bool = False, interval: float = 0.2) -> None:
for i in range(0, number_of_clones):
self.generate_clones(group=group)
self.generate_clones(group=group, auto_remote_ip=auto_remote_ip)
for clone in self.Clone.UID_CLONE_DB:
@@ -331,15 +310,15 @@ class Clone():
case 'connect':
try:
# clone connect 5 Group 3
# clone connect 5 GroupName 3
self.stop = False
number_of_clones = int(cmd[2])
group = str(cmd[3]).lower()
connection_interval = int(cmd[4]) if len(cmd) == 5 else 0.5
connection_interval = int(cmd[4]) if len(cmd) == 5 else 0.2
self.Base.create_thread(
func=self.thread_connect_clones,
func_args=(number_of_clones, group, connection_interval)
func_args=(number_of_clones, group, False, connection_interval)
)
except Exception as err:

View File

@@ -40,21 +40,62 @@ class Command:
# Add User object to the module (Mandatory)
self.User = ircInstance.User
# Add Client object to the module (Mandatory)
self.Client = ircInstance.Client
# Add Channel object to the module (Mandatory)
self.Channel = ircInstance.Channel
# Create module commands (Mandatory)
self.commands_level = {
1: ['join', 'part','owner', 'deowner', 'protect', 'deprotect', 'op',
'deop', 'halfop', 'dehalfop', 'voice','devoice', 'topic'],
2: ['opall', 'deopall', 'devoiceall', 'voiceall', 'ban', 'automode',
'unban', 'kick', 'kickban', 'umode',
'mode', 'get_mode', 'svsjoin', 'svspart', 'svsnick',
'wallops', 'globops','gnotice','whois', 'names', 'invite', 'inviteme',
'sajoin', 'sapart', 'kill', 'gline', 'ungline', 'kline',
'unkline', 'shun', 'unshun', 'glinelist', 'shunlist', 'klinelist'],
3: ['map']
}
self.Irc.build_command(1, self.module_name, 'join', 'Join a channel')
self.Irc.build_command(1, self.module_name, 'assign', 'Assign a user to a role or task')
self.Irc.build_command(1, self.module_name, 'part', 'Leave a channel')
self.Irc.build_command(1, self.module_name, 'unassign', 'Remove a user from a role or task')
self.Irc.build_command(1, self.module_name, 'owner', 'Give channel ownership to a user')
self.Irc.build_command(1, self.module_name, 'deowner', 'Remove channel ownership from a user')
self.Irc.build_command(1, self.module_name, 'protect', 'Protect a user from being kicked')
self.Irc.build_command(1, self.module_name, 'deprotect', 'Remove protection from a user')
self.Irc.build_command(1, self.module_name, 'op', 'Grant operator privileges to a user')
self.Irc.build_command(1, self.module_name, 'deop', 'Remove operator privileges from a user')
self.Irc.build_command(1, self.module_name, 'halfop', 'Grant half-operator privileges to a user')
self.Irc.build_command(1, self.module_name, 'dehalfop', 'Remove half-operator privileges from a user')
self.Irc.build_command(1, self.module_name, 'voice', 'Grant voice privileges to a user')
self.Irc.build_command(1, self.module_name, 'devoice', 'Remove voice privileges from a user')
self.Irc.build_command(1, self.module_name, 'topic', 'Change the topic of a channel')
self.Irc.build_command(2, self.module_name, 'opall', 'Grant operator privileges to all users')
self.Irc.build_command(2, self.module_name, 'deopall', 'Remove operator privileges from all users')
self.Irc.build_command(2, self.module_name, 'devoiceall', 'Remove voice privileges from all users')
self.Irc.build_command(2, self.module_name, 'voiceall', 'Grant voice privileges to all users')
self.Irc.build_command(2, self.module_name, 'ban', 'Ban a user from a channel')
self.Irc.build_command(2, self.module_name, 'automode', 'Automatically set user modes upon join')
self.Irc.build_command(2, self.module_name, 'unban', 'Remove a ban from a user')
self.Irc.build_command(2, self.module_name, 'kick', 'Kick a user from a channel')
self.Irc.build_command(2, self.module_name, 'kickban', 'Kick and ban a user from a channel')
self.Irc.build_command(2, self.module_name, 'umode', 'Set user mode')
self.Irc.build_command(2, self.module_name, 'mode', 'Set channel mode')
self.Irc.build_command(2, self.module_name, 'get_mode', 'Retrieve current channel mode')
self.Irc.build_command(2, self.module_name, 'svsjoin', 'Force a user to join a channel')
self.Irc.build_command(2, self.module_name, 'svspart', 'Force a user to leave a channel')
self.Irc.build_command(2, self.module_name, 'svsnick', 'Force a user to change their nickname')
self.Irc.build_command(2, self.module_name, 'wallops', 'Send a message to all operators')
self.Irc.build_command(2, self.module_name, 'globops', 'Send a global operator message')
self.Irc.build_command(2, self.module_name, 'gnotice', 'Send a global notice')
self.Irc.build_command(2, self.module_name, 'whois', 'Get information about a user')
self.Irc.build_command(2, self.module_name, 'names', 'List users in a channel')
self.Irc.build_command(2, self.module_name, 'invite', 'Invite a user to a channel')
self.Irc.build_command(2, self.module_name, 'inviteme', 'Invite yourself to a channel')
self.Irc.build_command(2, self.module_name, 'sajoin', 'Force yourself into a channel')
self.Irc.build_command(2, self.module_name, 'sapart', 'Force yourself to leave a channel')
self.Irc.build_command(2, self.module_name, 'kill', 'Disconnect a user from the server')
self.Irc.build_command(2, self.module_name, 'gline', 'Ban a user from the entire server')
self.Irc.build_command(2, self.module_name, 'ungline', 'Remove a global server ban')
self.Irc.build_command(2, self.module_name, 'kline', 'Ban a user based on their hostname')
self.Irc.build_command(2, self.module_name, 'unkline', 'Remove a K-line ban')
self.Irc.build_command(2, self.module_name, 'shun', 'Prevent a user from sending messages')
self.Irc.build_command(2, self.module_name, 'unshun', 'Remove a shun from a user')
self.Irc.build_command(2, self.module_name, 'glinelist', 'List all global bans')
self.Irc.build_command(2, self.module_name, 'shunlist', 'List all shunned users')
self.Irc.build_command(2, self.module_name, 'klinelist', 'List all K-line bans')
self.Irc.build_command(3, self.module_name, 'map', 'Show the server network map')
# Init the module
self.__init_module()
@@ -64,9 +105,6 @@ class Command:
def __init_module(self) -> None:
# Insert module commands into the core one (Mandatory)
self.__set_commands(self.commands_level)
# Create you own tables (Mandatory)
self.__create_tables()
@@ -79,20 +117,6 @@ class Command:
return None
def __set_commands(self, commands: dict[int, list[str]]) -> None:
"""### Rajoute les commandes du module au programme principal
Args:
commands (list): Liste des commandes du module
"""
for level, com in commands.items():
for c in commands[level]:
if c not in self.Irc.commands:
self.Irc.commands_level[level].append(c)
self.Irc.commands.append(c)
return None
def __create_tables(self) -> None:
"""Methode qui va créer la base de donnée si elle n'existe pas.
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module
@@ -248,14 +272,19 @@ class Command:
user_uid = self.User.clean_uid(cmd[5])
userObj: MUser = self.User.get_User(user_uid)
channel_name = cmd[4] if self.Channel.Is_Channel(cmd[4]) else None
client_obj = self.Client.get_Client(user_uid)
nickname = userObj.nickname if userObj is not None else None
if client_obj is not None:
nickname = client_obj.account
if userObj is None:
return None
if 'r' not in userObj.umodes and 'o' not in userObj.umodes:
if 'r' not in userObj.umodes and 'o' not in userObj.umodes and not self.Client.is_exist(userObj.uid):
return None
db_data: dict[str, str] = {"nickname": userObj.nickname, "channel": channel_name}
db_data: dict[str, str] = {"nickname": nickname, "channel": channel_name}
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:
@@ -782,7 +811,7 @@ class Command:
except Exception as err:
self.Logs.warning(f'Unknown Error: {str(err)}')
case 'join':
case 'join' | 'assign':
try:
sent_channel = str(cmd[1]) if self.Channel.Is_Channel(cmd[1]) else None
@@ -800,7 +829,7 @@ class Command:
except Exception as err:
self.Logs.warning(f'Unknown Error: {str(err)}')
case 'part':
case 'part' | 'unassign':
try:
sent_channel = str(cmd[1]) if self.Channel.Is_Channel(cmd[1]) else None
@@ -1098,18 +1127,19 @@ class Command:
case 'sapart':
try:
# .sapart nickname #channel
nickname = str(cmd[1])
channel = str(cmd[2])
if len(cmd) < 3:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f" /msg {dnickname} {command.upper()} nickname #channel")
return None
nickname = str(cmd[1])
channel = str(cmd[2])
self.Protocol.send_sapart(nick_to_sapart=nickname, channel_name=channel)
except KeyError as ke:
self.Logs.error(ke)
except Exception as err:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f" /msg {dnickname} {command.upper()} nickname #channel")
self.Logs.warning(f'Unknown Error: {str(err)}')
self.Logs.error(f'Unknown Error: {str(err)}')
case 'svsnick':
try:

View File

@@ -81,11 +81,16 @@ class Defender():
self.Reputation = ircInstance.Reputation
# Create module commands (Mandatory)
self.commands_level = {
0: ['code'],
1: ['info', 'autolimit'],
3: ['reputation','proxy_scan', 'flood', 'status', 'timer','show_reputation', 'sentinel']
}
self.Irc.build_command(0, self.module_name, 'code', 'Display the code or key for access')
self.Irc.build_command(1, self.module_name, 'info', 'Provide information about the channel or server')
self.Irc.build_command(1, self.module_name, 'autolimit', 'Automatically set channel user limits')
self.Irc.build_command(3, self.module_name, 'reputation', 'Check or manage user reputation')
self.Irc.build_command(3, self.module_name, 'proxy_scan', 'Scan users for proxy connections')
self.Irc.build_command(3, self.module_name, 'flood', 'Handle flood detection and mitigation')
self.Irc.build_command(3, self.module_name, 'status', 'Check the status of the server or bot')
self.Irc.build_command(3, self.module_name, 'timer', 'Set or manage timers')
self.Irc.build_command(3, self.module_name, 'show_reputation', 'Display reputation information')
self.Irc.build_command(3, self.module_name, 'sentinel', 'Monitor and guard the channel or server')
# Init the module (Mandatory)
self.__init_module()
@@ -95,9 +100,6 @@ class Defender():
def __init_module(self) -> None:
# Insert module commands into the core one (Mandatory)
self.__set_commands(self.commands_level)
# Create you own tables if needed (Mandatory)
self.__create_tables()
@@ -150,20 +152,6 @@ class Defender():
return None
def __set_commands(self, commands: dict[int, list[str]]) -> None:
"""### Rajoute les commandes du module au programme principal
Args:
commands (list): Liste des commandes du module
"""
for level, com in commands.items():
for c in commands[level]:
if c not in self.Irc.commands:
self.Irc.commands_level[level].append(c)
self.Irc.commands.append(c)
return None
def __create_tables(self) -> None:
"""Methode qui va créer la base de donnée si elle n'existe pas.
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module
@@ -1012,11 +1000,18 @@ class Defender():
case 'MODE':
# ['...', ':001XSCU0Q', 'MODE', '#jail', '+b', '~security-group:unknown-users']
# ['@unrealircd.org/...', ':001C0MF01', 'MODE', '#services', '+l', '1']
channel = str(cmd[3])
mode = str(cmd[4])
group_to_check = str(cmd[5:])
group_to_unban = '~security-group:unknown-users'
if self.ModConfig.autolimit == 1:
if mode == '+l' or mode == '-l':
chan = self.Channel.get_Channel(channel)
self.Protocol.send2socket(f":{self.Config.SERVICE_ID} MODE {chan.name} +l {len(chan.uids) + self.ModConfig.autolimit_amount}")
if self.Config.SALON_JAIL == channel:
if mode == '+b' and group_to_unban in group_to_check:
self.Protocol.send2socket(f":{service_id} MODE {self.Config.SALON_JAIL} -b ~security-group:unknown-users")
@@ -1721,6 +1716,9 @@ class Defender():
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' {color_green if self.ModConfig.abuseipdb_scan == 1 else color_red}abuseipdb_scan{nogc} ==> {self.ModConfig.abuseipdb_scan}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' {color_green if self.ModConfig.freeipapi_scan == 1 else color_red}freeipapi_scan{nogc} ==> {self.ModConfig.freeipapi_scan}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' {color_green if self.ModConfig.cloudfilt_scan == 1 else color_red}cloudfilt_scan{nogc} ==> {self.ModConfig.cloudfilt_scan}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' [{color_green if self.ModConfig.autolimit == 1 else color_red}Autolimit{nogc}] ==> {self.ModConfig.autolimit}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' {color_green if self.ModConfig.autolimit == 1 else color_red}Autolimit Amount{nogc} ==> {self.ModConfig.autolimit_amount}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' {color_green if self.ModConfig.autolimit == 1 else color_red}Autolimit Interval{nogc} ==> {self.ModConfig.autolimit_interval}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' [{color_green if self.ModConfig.flood == 1 else color_red}Flood{nogc}] ==> {self.ModConfig.flood}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=' flood_action ==> Coming soon')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f' flood_message ==> {self.ModConfig.flood_message}')

View File

@@ -42,9 +42,8 @@ class Jsonrpc():
self.Channel = ircInstance.Channel
# Create module commands (Mandatory)
self.commands_level = {
1: ['jsonrpc', 'jruser']
}
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')
# Init the module
self.__init_module()
@@ -54,8 +53,6 @@ class Jsonrpc():
def __init_module(self) -> None:
# Insert module commands into the core one (Mandatory)
self.__set_commands(self.commands_level)
logging.getLogger('websockets').setLevel(logging.WARNING)
logging.getLogger('unrealircd-rpc-py').setLevel(logging.CRITICAL)
@@ -103,20 +100,6 @@ class Jsonrpc():
return None
def __set_commands(self, commands:dict[int, list[str]]) -> None:
"""### Rajoute les commandes du module au programme principal
Args:
commands (list): Liste des commandes du module
"""
for level, com in commands.items():
for c in commands[level]:
if not c in self.Irc.commands:
self.Irc.commands_level[level].append(c)
self.Irc.commands.append(c)
return None
def __create_tables(self) -> None:
"""Methode qui va créer la base de donnée si elle n'existe pas.
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module

View File

@@ -46,12 +46,11 @@ class Test():
self.Reputation = ircInstance.Reputation
# Create module commands (Mandatory)
self.commands_level = {
0: ['test-command'],
1: ['test_level_1'],
2: ['test_level_2'],
3: ['test_level_3']
}
self.Irc.build_command(0, self.module_name, 'test-command', 'Execute a test command')
self.Irc.build_command(1, self.module_name, 'test_level_1', 'Execute a level 1 test command')
self.Irc.build_command(2, self.module_name, 'test_level_2', 'Execute a level 2 test command')
self.Irc.build_command(3, self.module_name, 'test_level_3', 'Execute a level 3 test command')
# Init the module
self.__init_module()
@@ -61,9 +60,6 @@ class Test():
def __init_module(self) -> None:
# Insert module commands into the core one (Mandatory)
self.__set_commands(self.commands_level)
# Create you own tables (Mandatory)
self.__create_tables()
@@ -73,20 +69,6 @@ class Test():
return None
def __set_commands(self, commands:dict[int, list[str]]) -> None:
"""### Rajoute les commandes du module au programme principal
Args:
commands (list): Liste des commandes du module
"""
for level, com in commands.items():
for c in commands[level]:
if not c in self.Irc.commands:
self.Irc.commands_level[level].append(c)
self.Irc.commands.append(c)
return None
def __create_tables(self) -> None:
"""Methode qui va créer la base de donnée si elle n'existe pas.
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module

View File

@@ -56,9 +56,7 @@ class Votekick():
self.Channel = ircInstance.Channel
# Créer les nouvelles commandes du module
self.commands_level = {
0: ['vote']
}
self.Irc.build_command(1, self.module_name, 'vote', 'The kick vote module')
# Init the module
self.__init_module()
@@ -70,27 +68,11 @@ class Votekick():
# Add admin object to retrieve admin users
self.Admin = self.Irc.Admin
self.__set_commands(self.commands_level)
self.__create_tables()
self.join_saved_channels()
return None
def __set_commands(self, commands:dict[int, list[str]]) -> None:
"""### Rajoute les commandes du module au programme principal
Args:
commands (list): Liste des commandes du module
"""
for level, com in commands.items():
for c in commands[level]:
if not c in self.Irc.commands:
self.Irc.commands_level[level].append(c)
self.Irc.commands.append(c)
return None
def __create_tables(self) -> None:
"""Methode qui va créer la base de donnée si elle n'existe pas.
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module

View File

@@ -1,5 +1,5 @@
{
"version": "6.0.4",
"version": "6.1.2",
"requests": "2.32.3",
"psutil": "6.0.0",