iio/DEFENDER
1
0
Fork 0
mirror of https://github.com/iio612/DEFENDER.git synced 2026-02-13 19:24:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix issues coming from the feedback of dktmb

Browse Source
This commit is contained in:
adator
2025-12-16 00:54:30 +01:00
parent 6b8152f64f
commit f809b724a4
6 changed files with 58 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
core/base.py
View File

@@ -431,23 +431,27 @@ class Base:
self.running_iothreads.remove(id_obj) self.running_iothreads.remove(id_obj)
return result return result
def asynctask_done(self, task: Union[asyncio.Task, asyncio.Future]): def asynctask_done(self, task: Union[asyncio.Task, asyncio.Future], context: Optional[dict[str, Any]] = None):
"""Log task when done """Log task when done
Args: Args:
task (asyncio.Task): The Asyncio Task callback task (asyncio.Task): The Asyncio Task callback
""" """
name = task.get_name() if isinstance(task, asyncio.Task) else "Thread"
task_or_future = "Task" if isinstance(task, asyncio.Task) else "Future" if context:
print(context)
task_name = "Future" if isinstance(task, asyncio.Future) else task.get_name()
task_or_future = "Task"
try: try:
if task.exception(): if task.exception():
self.logs.error(f"[ASYNCIO] {task_or_future} {name} failed with exception: {task.exception()}") self.logs.error(f"[ASYNCIO] {task_or_future} {task_name} failed with exception: {task.exception()}")
else: else:
self.logs.debug(f"[ASYNCIO] {task_or_future} {name} completed successfully.") self.logs.debug(f"[ASYNCIO] {task_or_future} {task_name} completed successfully.")
except asyncio.CancelledError as ce: except asyncio.CancelledError as ce:
self.logs.debug(f"[ASYNCIO] {task_or_future} {name} terminated with cancelled error. {ce}") self.logs.debug(f"[ASYNCIO] {task_or_future} {task_name} terminated with cancelled error. {ce}")
except asyncio.InvalidStateError as ie: except asyncio.InvalidStateError as ie:
self.logs.debug(f"[ASYNCIO] {task_or_future} {name} terminated with invalid state error. {ie}") self.logs.debug(f"[ASYNCIO] {task_or_future} {task_name} terminated with invalid state error. {ie}")
def is_thread_alive(self, thread_name: str) -> bool: def is_thread_alive(self, thread_name: str) -> bool:
"""Check if the thread is still running! using the is_alive method of Threads. """Check if the thread is still running! using the is_alive method of Threads.
@@ -611,25 +615,10 @@ 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
)
'''
await self.db_execute_query(table_core_log) await self.db_execute_query(table_core_log)
await self.db_execute_query(table_core_log_command) await self.db_execute_query(table_core_log_command)
await self.db_execute_query(table_core_module) await self.db_execute_query(table_core_module)
await self.db_execute_query(table_core_admin) await self.db_execute_query(table_core_admin)
await self.db_execute_query(table_core_client)
await self.db_execute_query(table_core_channel) await self.db_execute_query(table_core_channel)
await self.db_execute_query(table_core_config) await self.db_execute_query(table_core_config)

13
core/classes/modules/admin.py
View File

@@ -67,15 +67,14 @@ class Admin:
Returns: Returns:
bool: True if the admin level has been updated bool: True if the admin level has been updated
""" """
admin_obj = self.get_admin(nickname)
for record in self.UID_ADMIN_DB: if admin_obj:
if record.nickname == nickname:
# If the admin exist, update and do not go further # If the admin exist, update and do not go further
record.level = new_admin_level admin_obj.level = new_admin_level
self._ctx.Logs.debug(f'Admin ({record.nickname}) has been updated with new level {new_admin_level}') self._ctx.Logs.debug(f'Admin ({admin_obj.nickname}) has been updated with new level {new_admin_level}')
return True return True
self._ctx.Logs.debug(f'The new level {new_admin_level} was not updated, nickname = {nickname} - The Client is not an admin') self._ctx.Logs.debug(f'The new level {new_admin_level} was not updated in local variable, nickname = {nickname} is not logged in')
return False return False
@@ -94,7 +93,7 @@ class Admin:
self._ctx.Logs.debug(f'UID ({admin_obj.uid}) has been deleted') self._ctx.Logs.debug(f'UID ({admin_obj.uid}) has been deleted')
return True return True
self._ctx.Logs.debug(f'The UID {uidornickname} was not deleted') self._ctx.Logs.debug(f'The UID {uidornickname} was not deleted from the local variable (admin not connected)')
return False return False

259
core/irc.py
View File

@@ -2,14 +2,12 @@ import asyncio
import re import re
import ssl import ssl
import threading import threading
from datetime import datetime, timedelta from typing import TYPE_CHECKING, Optional, Union
from typing import TYPE_CHECKING, Any, Optional, Union
from core.classes.modules import rehash from core.classes.modules import rehash
from core.classes.interfaces.iprotocol import IProtocol from core.classes.interfaces.iprotocol import IProtocol
from core.utils import tr from core.utils import tr
if TYPE_CHECKING: if TYPE_CHECKING:
from core.definition import MSasl
from core.loader import Loader from core.loader import Loader
class Irc: class Irc:
@@ -66,18 +64,11 @@ class Irc:
self.ctx.Commands.build_command(0, 'core', 'copyright', 'Give some information about the IRC Service') self.ctx.Commands.build_command(0, 'core', 'copyright', 'Give some information about the IRC Service')
self.ctx.Commands.build_command(0, 'core', 'uptime', 'Give you since when the service is connected') self.ctx.Commands.build_command(0, 'core', 'uptime', 'Give you since when the service is connected')
self.ctx.Commands.build_command(0, 'core', 'firstauth', 'First authentication of the Service') self.ctx.Commands.build_command(0, 'core', 'firstauth', 'First authentication of the Service')
self.ctx.Commands.build_command(0, 'core', 'register', f'Register your nickname /msg {self.ctx.Config.SERVICE_NICKNAME} REGISTER <password> <email>')
self.ctx.Commands.build_command(0, 'core', 'identify', f'Identify yourself with your password /msg {self.ctx.Config.SERVICE_NICKNAME} IDENTIFY <account> <password>')
self.ctx.Commands.build_command(0, 'core', 'logout', 'Reverse the effect of the identify command')
self.ctx.Commands.build_command(1, 'core', 'load', 'Load an existing module')
self.ctx.Commands.build_command(1, 'core', 'unload', 'Unload a module')
self.ctx.Commands.build_command(1, 'core', 'reload', 'Reload a module')
self.ctx.Commands.build_command(1, 'core', 'deauth', 'Deauth from the irc service') self.ctx.Commands.build_command(1, 'core', 'deauth', 'Deauth from the irc service')
self.ctx.Commands.build_command(1, 'core', 'checkversion', 'Check the version of the irc service') self.ctx.Commands.build_command(1, 'core', 'checkversion', 'Check the version of the irc service')
self.ctx.Commands.build_command(2, 'core', 'show_modules', 'Display a list of loaded modules') self.ctx.Commands.build_command(2, 'core', 'show_modules', 'Display a list of loaded modules')
self.ctx.Commands.build_command(2, 'core', 'show_channels', 'Display a list of active channels') self.ctx.Commands.build_command(2, 'core', 'show_channels', 'Display a list of active channels')
self.ctx.Commands.build_command(2, 'core', 'show_users', 'Display a list of connected users') self.ctx.Commands.build_command(2, 'core', 'show_users', 'Display a list of connected users')
self.ctx.Commands.build_command(2, 'core', 'show_clients', 'Display a list of connected clients')
self.ctx.Commands.build_command(2, 'core', 'show_admins', 'Display a list of administrators') self.ctx.Commands.build_command(2, 'core', 'show_admins', 'Display a list of administrators')
self.ctx.Commands.build_command(2, 'core', 'show_configuration', 'Display the current configuration settings') self.ctx.Commands.build_command(2, 'core', 'show_configuration', 'Display the current configuration settings')
self.ctx.Commands.build_command(2, 'core', 'show_cache', 'Display the current cache') self.ctx.Commands.build_command(2, 'core', 'show_cache', 'Display the current cache')
@@ -96,6 +87,9 @@ class Irc:
self.ctx.Commands.build_command(4, 'core', 'show_asyncio', 'Display active asyncio') self.ctx.Commands.build_command(4, 'core', 'show_asyncio', 'Display active asyncio')
self.ctx.Commands.build_command(4, 'core', 'start_rpc', 'Start defender jsonrpc server') self.ctx.Commands.build_command(4, 'core', 'start_rpc', 'Start defender jsonrpc server')
self.ctx.Commands.build_command(4, 'core', 'stop_rpc', 'Stop defender jsonrpc server') self.ctx.Commands.build_command(4, 'core', 'stop_rpc', 'Stop defender jsonrpc server')
self.ctx.Commands.build_command(4, 'core', 'load', 'Load an existing module')
self.ctx.Commands.build_command(4, 'core', 'unload', 'Unload a module')
self.ctx.Commands.build_command(4, 'core', 'reload', 'Reload a module')
############################################## ##############################################
# CONNEXION IRC # # CONNEXION IRC #
@@ -133,14 +127,6 @@ class Irc:
run_once=True, thread_flag=True run_once=True, thread_flag=True
) )
) )
# print("start_heartbeat.................")
# await self.ctx.Base.create_thread_io(
# self.ctx.Utils.heartbeat,
# self.ctx, self.beat,
# run_once=True, thread_flag=True
# )
while self.signal: while self.signal:
data = await self.reader.readuntil(b'\r\n') data = await self.reader.readuntil(b'\r\n')
@@ -216,53 +202,6 @@ class Irc:
return None return None
# async def on_sasl_authentication_process(self, sasl_model: 'MSasl') -> bool:
# s = sasl_model
# if sasl_model:
# async 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.ctx.Config.TABLE_ADMIN} WHERE fingerprint = :fingerprint"
# else:
# mes_donnees = {'user': username, 'password': self.ctx.Utils.hash_password(password)}
# query = f"SELECT user, level, language FROM {self.ctx.Config.TABLE_ADMIN} WHERE user = :user AND password = :password"
# result = await self.ctx.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]}
# else:
# return None
# if s.message_type == 'C' and s.mechanisme == 'PLAIN':
# # Connection via PLAIN
# admin_info = await db_get_admin_info(username=s.username, password=s.password)
# if admin_info is not None:
# s.auth_success = True
# s.level = admin_info.get('level', 0)
# s.language = admin_info.get('language', 'EN')
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} SASL {self.ctx.Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S")
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful")
# else:
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} SASL {self.ctx.Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F")
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed")
# elif s.message_type == 'S' and s.mechanisme == 'EXTERNAL':
# # Connection using fingerprints
# admin_info = await db_get_admin_info(fingerprint=s.fingerprint)
# if admin_info is not None:
# s.auth_success = True
# s.level = admin_info.get('level', 0)
# s.username = admin_info.get('user', None)
# s.language = admin_info.get('language', 'EN')
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} SASL {self.ctx.Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S")
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful")
# else:
# # "904 <nick> :SASL authentication failed"
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} SASL {self.ctx.Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F")
# await self.Protocol.send2socket(f":{self.ctx.Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed")
def insert_db_admin(self, uid: str, account: str, level: int, language: str) -> None: def insert_db_admin(self, uid: str, account: str, level: int, language: str) -> None:
user_obj = self.ctx.User.get_user(uid) user_obj = self.ctx.User.get_user(uid)
@@ -328,7 +267,7 @@ class Irc:
spassword = self.ctx.Utils.hash_password(password) spassword = self.ctx.Utils.hash_password(password)
# Check if the user already exist # Check if the user already exist
if not self.ctx.Admin.db_is_admin_exist(nickname): if not await self.ctx.Admin.db_is_admin_exist(nickname):
mes_donnees = {'datetime': self.ctx.Utils.get_sdatetime(), 'user': nickname, 'password': spassword, 'hostname': hostname, 'vhost': vhost, 'level': level, 'language': self.ctx.Config.LANG} mes_donnees = {'datetime': self.ctx.Utils.get_sdatetime(), 'user': nickname, 'password': spassword, 'hostname': hostname, 'vhost': vhost, 'level': level, 'language': self.ctx.Config.LANG}
await self.ctx.Base.db_execute_query(f'''INSERT INTO {self.ctx.Config.TABLE_ADMIN} await self.ctx.Base.db_execute_query(f'''INSERT INTO {self.ctx.Config.TABLE_ADMIN}
(createdOn, user, password, hostname, vhost, level, language) VALUES (createdOn, user, password, hostname, vhost, level, language) VALUES
@@ -345,7 +284,7 @@ class Irc:
async def thread_check_for_new_version(self, fromuser: str) -> None: async def thread_check_for_new_version(self, fromuser: str) -> None:
dnickname = self.ctx.Config.SERVICE_NICKNAME dnickname = self.ctx.Config.SERVICE_NICKNAME
response = self.ctx.Base.create_asynctask( response = await self.ctx.Base.create_asynctask(
self.ctx.Base.create_thread_io( self.ctx.Base.create_thread_io(
self.ctx.Base.check_for_new_version, True self.ctx.Base.check_for_new_version, True
) )
@@ -403,9 +342,6 @@ class Irc:
if u is None: if u is None:
return None return None
c = self.ctx.Client.get_client(u.uid)
"""The Client Object"""
fromuser = u.nickname fromuser = u.nickname
uid = u.uid uid = u.uid
self.ctx.Settings.current_admin = self.ctx.Admin.get_admin(user) # set Current admin if any. self.ctx.Settings.current_admin = self.ctx.Admin.get_admin(user) # set Current admin if any.
@@ -607,10 +543,19 @@ class Irc:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"level: from 1 to 4") await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"level: from 1 to 4")
return None return None
if len(cmd) == 4:
user_new_level = int(cmd[3])
if user_new_level > 5:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="Maximum authorized level is 5")
return None
if user_new_level == 0:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="You must provide a valid level [1 to 4]")
return None
user_to_edit = cmd[1] user_to_edit = cmd[1]
user_password = self.ctx.Utils.hash_password(cmd[2]) user_password = self.ctx.Utils.hash_password(cmd[2])
get_admin = self.ctx.Admin.get_admin(fromuser) get_admin = self.ctx.Admin.get_admin(fromuser)
if get_admin is None: if get_admin is None:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"This user {fromuser} has no Admin access") await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"This user {fromuser} has no Admin access")
return None return None
@@ -619,15 +564,9 @@ class Irc:
current_uid = uid current_uid = uid
current_user_level = get_admin.level current_user_level = get_admin.level
user_new_level = int(cmd[3]) if len(cmd) == 4 else get_admin.level if current_user == user_to_edit:
if current_user == fromuser:
user_new_level = get_admin.level user_new_level = get_admin.level
if user_new_level > 5:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="Maximum authorized level is 5")
return None
# Rechercher le user dans la base de données. # Rechercher le user dans la base de données.
mes_donnees = {'user': user_to_edit} mes_donnees = {'user': user_to_edit}
query = f"SELECT user, level FROM {self.ctx.Config.TABLE_ADMIN} WHERE user = :user" query = f"SELECT user, level FROM {self.ctx.Config.TABLE_ADMIN} WHERE user = :user"
@@ -662,6 +601,10 @@ class Irc:
case 'delaccess': case 'delaccess':
# .delaccess [USER] [CONFIRMUSER] # .delaccess [USER] [CONFIRMUSER]
if len(cmd) < 3:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Right command : /msg {dnickname} delaccess [nickname] [CONFIRMNICKNAME]")
return None
user_to_del = cmd[1] user_to_del = cmd[1]
user_confirmation = cmd[2] user_confirmation = cmd[2]
@@ -670,10 +613,6 @@ class Irc:
self.ctx.Logs.warning(f':{dnickname} NOTICE {fromuser} : Les user ne sont pas les mêmes, tu dois confirmer le user que tu veux supprimer') self.ctx.Logs.warning(f':{dnickname} NOTICE {fromuser} : Les user ne sont pas les mêmes, tu dois confirmer le user que tu veux supprimer')
return None return None
if len(cmd) < 3:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"{self.ctx.Config.SERVICE_PREFIX}delaccess [USER] [CONFIRMUSER]")
return None
get_admin = self.ctx.Admin.get_admin(fromuser) get_admin = self.ctx.Admin.get_admin(fromuser)
if get_admin is None: if get_admin is None:
@@ -745,147 +684,6 @@ class Irc:
except Exception as e: except Exception as e:
self.ctx.Logs.error(e) self.ctx.Logs.error(e)
case 'register':
# Syntax. Register PASSWORD EMAIL
try:
if len(cmd) < 3:
await self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f'/msg {self.ctx.Config.SERVICE_NICKNAME} {command.upper()} <PASSWORD> <EMAIL>'
)
return None
password = cmd[1]
email = cmd[2]
if not self.ctx.Base.is_valid_email(email_to_control=email):
await 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 = u
if user_obj is None:
self.ctx.Logs.error(f"Nickname ({fromuser}) doesn't exist, it is impossible to register this nickname")
return None
# If the account already exist.
if self.ctx.Client.db_is_account_exist(fromuser):
await self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f"Your account already exist, please try to login instead /msg {self.ctx.Config.SERVICE_NICKNAME} IDENTIFY <account> <password>"
)
return None
# If the account doesn't exist then insert into database
data_to_record = {
'createdOn': self.ctx.Utils.get_sdatetime(), 'account': fromuser,
'nickname': user_obj.nickname, 'hostname': user_obj.hostname, 'vhost': user_obj.vhost, 'realname': user_obj.realname, 'email': email,
'password': self.ctx.Utils.hash_password(password=password), 'level': 0
}
insert_to_db = await self.ctx.Base.db_execute_query(f"""
INSERT INTO {self.ctx.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:
await 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.ctx.Logs.error(f"Value Error : {ve}")
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f" {self.ctx.Config.SERVICE_PREFIX}{command.upper()} <PASSWORD> <EMAIL>")
case 'identify':
# Identify ACCOUNT PASSWORD
try:
if len(cmd) < 3:
await self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f'/msg {self.ctx.Config.SERVICE_NICKNAME} {command.upper()} <ACCOUNT> <PASSWORD>'
)
return None
account = str(cmd[1]) # account
encrypted_password = self.ctx.Utils.hash_password(cmd[2])
user_obj = u
client_obj = c
if client_obj is not None:
await 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.ctx.Config.TABLE_CLIENT} WHERE account = :account AND password = :password"
db_param = {'account': account, 'password': encrypted_password}
exec_query = await self.ctx.Base.db_execute_query(db_query, db_param)
result_query = exec_query.fetchone()
if result_query:
account = result_query[0]
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"You are now logged in")
client = self.ctx.Definition.MClient(**user_obj.to_dict(), account=account)
self.ctx.Client.insert(client)
await self.Protocol.send_svslogin(user_obj.uid, account)
await self.Protocol.send_svs2mode(nickname=fromuser, user_mode='+r')
else:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Wrong password or account")
return None
except ValueError as ve:
self.ctx.Logs.error(f"Value Error: {ve}")
await self.Protocol.send_notice(
nick_from=dnickname,
nick_to=fromuser,
msg=f'/msg {self.ctx.Config.SERVICE_NICKNAME} {command.upper()} <ACCOUNT> <PASSWORD>'
)
except Exception as err:
self.ctx.Logs.error(f"General Error: {err}")
case 'logout':
try:
# LOGOUT <account>
if len(cmd) < 2:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"/msg {dnickname} {command.upper()} <account>")
return None
user_obj = u
if user_obj is None:
self.ctx.Logs.error(f"The User [{fromuser}] is not available in the database")
return None
client_obj = c
if client_obj is None:
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="Nothing to logout. please login first")
return None
await self.Protocol.send_svslogout(client_obj)
self.ctx.Client.delete(user_obj.uid)
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"You have been logged out successfully")
except ValueError as ve:
self.ctx.Logs.error(f"Value Error: {ve}")
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"/msg {dnickname} {command.upper()} <account>")
except Exception as err:
self.ctx.Logs.error(f"General Error: {err}")
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"/msg {dnickname} {command.upper()} <account>")
case 'help': case 'help':
# Syntax. !help [module_name] # Syntax. !help [module_name]
module_name = str(cmd[1]) if len(cmd) == 2 else None module_name = str(cmd[1]) if len(cmd) == 2 else None
@@ -951,9 +749,6 @@ class Irc:
for chan_name in self.ctx.Channel.UID_CHANNEL_DB: for chan_name in self.ctx.Channel.UID_CHANNEL_DB:
await self.Protocol.send_set_mode('-l', channel_name=chan_name.name) await self.Protocol.send_set_mode('-l', channel_name=chan_name.name)
for client in self.ctx.Client.CLIENT_DB:
await self.Protocol.send_svslogout(client)
await self.Protocol.send_notice( await self.Protocol.send_notice(
nick_from=dnickname, nick_from=dnickname,
nick_to=fromuser, nick_to=fromuser,
@@ -1087,17 +882,6 @@ class Irc:
) )
return None return None
case 'show_clients':
count_users = len(self.ctx.Client.CLIENT_DB)
await self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg=f"Total Connected Clients: {count_users}")
for db_client in self.ctx.Client.CLIENT_DB:
await 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}"
)
return None
case 'show_admins': case 'show_admins':
for db_admin in self.ctx.Admin.UID_ADMIN_DB: for db_admin in self.ctx.Admin.UID_ADMIN_DB:
await self.Protocol.send_notice( await self.Protocol.send_notice(
@@ -1186,4 +970,3 @@ class Irc:
case _: case _:
pass pass

12
mods/command/mod_command.py
View File

@@ -70,9 +70,7 @@ class Command(IModule):
self.ctx.Irc.Protocol.known_protocol.add(c) self.ctx.Irc.Protocol.known_protocol.add(c)
self.ctx.Commands.build_command(2, self.module_name, 'join', 'Join a channel') self.ctx.Commands.build_command(2, self.module_name, 'join', 'Join a channel')
self.ctx.Commands.build_command(2, self.module_name, 'assign', 'Assign a user to a role or task')
self.ctx.Commands.build_command(2, self.module_name, 'part', 'Leave a channel') self.ctx.Commands.build_command(2, self.module_name, 'part', 'Leave a channel')
self.ctx.Commands.build_command(2, self.module_name, 'unassign', 'Remove a user from a role or task')
self.ctx.Commands.build_command(2, self.module_name, 'owner', 'Give channel ownership to a user') self.ctx.Commands.build_command(2, self.module_name, 'owner', 'Give channel ownership to a user')
self.ctx.Commands.build_command(2, self.module_name, 'deowner', 'Remove channel ownership from a user') self.ctx.Commands.build_command(2, self.module_name, 'deowner', 'Remove channel ownership from a user')
self.ctx.Commands.build_command(2, self.module_name, 'protect', 'Protect a user from being kicked') self.ctx.Commands.build_command(2, self.module_name, 'protect', 'Protect a user from being kicked')
@@ -218,16 +216,12 @@ class Command(IModule):
user_uid = self.ctx.User.clean_uid(cmd[5]) user_uid = self.ctx.User.clean_uid(cmd[5])
userObj: MUser = self.ctx.User.get_user(user_uid) userObj: MUser = self.ctx.User.get_user(user_uid)
channel_name = cmd[4] if self.ctx.Channel.is_valid_channel(cmd[4]) else None channel_name = cmd[4] if self.ctx.Channel.is_valid_channel(cmd[4]) else None
client_obj = self.ctx.Client.get_client(user_uid)
nickname = userObj.nickname if userObj is not None else None nickname = userObj.nickname if userObj is not None else None
if client_obj is not None:
nickname = client_obj.account
if userObj is None: if userObj is None:
return None return None
if 'r' not in userObj.umodes and 'o' not in userObj.umodes and not self.ctx.Client.is_exist(userObj.uid): if 'r' not in userObj.umodes and 'o' not in userObj.umodes:
return None return None
db_data: dict[str, str] = {"nickname": nickname.lower(), "channel": channel_name.lower()} db_data: dict[str, str] = {"nickname": nickname.lower(), "channel": channel_name.lower()}
@@ -416,7 +410,7 @@ class Command(IModule):
except Exception as err: except Exception as err:
self.ctx.Logs.warning(f'Unknown Error: {str(err)}') self.ctx.Logs.warning(f'Unknown Error: {str(err)}')
case 'join' | 'assign': case 'join':
try: try:
await self.mod_utils.set_assign_channel_to_service(self, cmd, fromuser) await self.mod_utils.set_assign_channel_to_service(self, cmd, fromuser)
except IndexError as ie: except IndexError as ie:
@@ -425,7 +419,7 @@ class Command(IModule):
except Exception as err: except Exception as err:
self.ctx.Logs.warning(f'Unknown Error: {str(err)}') self.ctx.Logs.warning(f'Unknown Error: {str(err)}')
case 'part' | 'unassign': case 'part':
try: try:
# Syntax. !part #channel # Syntax. !part #channel
await self.mod_utils.set_unassign_channel_to_service(self, cmd, fromuser) await self.mod_utils.set_unassign_channel_to_service(self, cmd, fromuser)

4
mods/command/utils.py
View File

@@ -115,9 +115,11 @@ async def set_mode_to_all(uplink: 'Command', channel_name: str, action: Literal[
users:str = '' users:str = ''
uids_split = [chan_info.uids[i:i + 6] for i in range(0, len(chan_info.uids), 6)] uids_split = [chan_info.uids[i:i + 6] for i in range(0, len(chan_info.uids), 6)]
await uplink.ctx.Irc.Protocol.send2socket(f":{service_id} MODE {channel_name} {action}{set_mode} {dnickname}") # await uplink.ctx.Irc.Protocol.send2socket(f":{service_id} MODE {channel_name} {action}{set_mode} {dnickname}")
for uid in uids_split: for uid in uids_split:
for i in range(0, len(uid)): for i in range(0, len(uid)):
if uplink.ctx.Utils.clean_uid(uid[i]) == uplink.ctx.Config.SERVICE_ID:
continue
mode += set_mode mode += set_mode
users += f'{uplink.ctx.User.get_nickname(uplink.ctx.Utils.clean_uid(uid[i]))} ' users += f'{uplink.ctx.User.get_nickname(uplink.ctx.Utils.clean_uid(uid[i]))} '
if i == len(uid) - 1: if i == len(uid) - 1:

15
mods/defender/mod_defender.py
View File

@@ -288,10 +288,19 @@ class Defender(IModule):
jail_chan = self.ctx.Config.SALON_JAIL # Salon pot de miel jail_chan = self.ctx.Config.SALON_JAIL # Salon pot de miel
jail_chan_mode = self.ctx.Config.SALON_JAIL_MODES # Mode du salon "pot de miel" jail_chan_mode = self.ctx.Config.SALON_JAIL_MODES # Mode du salon "pot de miel"
color_green = self.ctx.Config.COLORS.green
color_red = self.ctx.Config.COLORS.red
color_black = self.ctx.Config.COLORS.black
color_nogc = self.ctx.Config.COLORS.nogc
match command: match command:
case 'show_reputation': case 'show_reputation':
if self.mod_config.reputation == 0:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="Reputation system if off!")
return None
if not self.ctx.Reputation.UID_REPUTATION_DB: if not self.ctx.Reputation.UID_REPUTATION_DB:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="No one is suspected") await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser, msg="No one is suspected")
@@ -664,9 +673,6 @@ class Defender(IModule):
# .proxy_scan set psutil_scan on/off --> Active les informations de connexion a la machine locale # .proxy_scan set psutil_scan on/off --> Active les informations de connexion a la machine locale
# .proxy_scan set abuseipdb_scan on/off --> Active le scan via l'api abuseipdb # .proxy_scan set abuseipdb_scan on/off --> Active le scan via l'api abuseipdb
len_cmd = len(cmd) len_cmd = len(cmd)
color_green = self.ctx.Config.COLORS.green
color_red = self.ctx.Config.COLORS.red
color_black = self.ctx.Config.COLORS.black
if len_cmd == 4: if len_cmd == 4:
set_key = str(cmd[1]).lower() set_key = str(cmd[1]).lower()
@@ -943,6 +949,7 @@ class Defender(IModule):
if chan.name not in channel_to_dont_quit: if chan.name not in channel_to_dont_quit:
await self.ctx.Irc.Protocol.send_join_chan(uidornickname=dnickname, channel=chan.name) await self.ctx.Irc.Protocol.send_join_chan(uidornickname=dnickname, channel=chan.name)
await self.ctx.Irc.Protocol.send_priv_msg(dnickname, f"Sentinel mode activated on {channel}", channel=chan.name) await self.ctx.Irc.Protocol.send_priv_msg(dnickname, f"Sentinel mode activated on {channel}", channel=chan.name)
await self.ctx.Irc.Protocol.send_priv_msg(dnickname, f"[ {color_green}SENTINEL{color_nogc} ] Activated by {fromuser}", channel=self.ctx.Config.SERVICE_CHANLOG)
return None return None
if activation == 'off': if activation == 'off':
@@ -957,7 +964,7 @@ class Defender(IModule):
await self.ctx.Irc.Protocol.send_priv_msg(dnickname, f"Sentinel mode deactivated on {channel}", channel=chan.name) await self.ctx.Irc.Protocol.send_priv_msg(dnickname, f"Sentinel mode deactivated on {channel}", channel=chan.name)
await self.join_saved_channels() await self.join_saved_channels()
await self.ctx.Irc.Protocol.send_priv_msg(dnickname, f"[ {color_red}SENTINEL{color_nogc} ] Deactivated by {fromuser}", channel=self.ctx.Config.SERVICE_CHANLOG)
return None return None
case _: case _: