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

Add command handler system. Starting adapt the modules to fit other protocls.

Browse Source
This commit is contained in:
adator
2025-09-09 22:37:41 +02:00
parent 6b7fd16a44
commit fd9643eddc
18 changed files with 1319 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

265
core/classes/protocols/unreal6.py
View File

@@ -1,9 +1,10 @@
from base64 import b64decode
from re import match, findall, search
from datetime import datetime
from typing import TYPE_CHECKING, Optional
from typing import TYPE_CHECKING, Any, Optional
from ssl import SSLEOFError, SSLError
from core.classes.protocols.command_handler import CommandHandler
from core.classes.protocols.interface import IProtocol
from core.utils import tr
@@ -32,13 +33,16 @@ class Unrealircd6(IProtocol):
'PROTOCTL', 'SERVER', 'SMOD', 'TKL', 'NETINFO',
'006', '007', '018'}
self.__Logs.info(f"** Loading protocol [{__name__}]")
self.Handler = CommandHandler(ircInstance.Loader)
def get_ircd_protocol_poisition(self, cmd: list[str]) -> tuple[int, Optional[str]]:
self.__Logs.info(f"[PROTOCOL] Protocol [{__name__}] loaded!")
def get_ircd_protocol_poisition(self, cmd: list[str], log: bool = False) -> tuple[int, Optional[str]]:
"""Get the position of known commands
Args:
cmd (list[str]): The server response
log (bool): If true it will log in the logger
Returns:
tuple[int, Optional[str]]: The position and the command.
@@ -47,10 +51,34 @@ class Unrealircd6(IProtocol):
if token.upper() in self.known_protocol:
return index, token.upper()
self.__Logs.debug(f"[IRCD LOGS] You need to handle this response: {cmd}")
if log:
self.__Logs.debug(f"[IRCD LOGS] You need to handle this response: {cmd}")
return (-1, None)
def register_command(self) -> None:
m = self.__Irc.Loader.Definition.MIrcdCommand
self.Handler.register(m(command_name="PING", func=self.on_server_ping))
self.Handler.register(m(command_name="UID", func=self.on_uid))
self.Handler.register(m(command_name="QUIT", func=self.on_quit))
self.Handler.register(m(command_name="SERVER", func=self.on_server))
self.Handler.register(m(command_name="SJOIN", func=self.on_sjoin))
self.Handler.register(m(command_name="EOS", func=self.on_eos))
self.Handler.register(m(command_name="PROTOCTL", func=self.on_protoctl))
self.Handler.register(m(command_name="SVS2MODE", func=self.on_svs2mode))
self.Handler.register(m(command_name="SQUIT", func=self.on_squit))
self.Handler.register(m(command_name="PART", func=self.on_part))
self.Handler.register(m(command_name="VERSION", func=self.on_version_msg))
self.Handler.register(m(command_name="UMODE2", func=self.on_umode2))
self.Handler.register(m(command_name="NICK", func=self.on_nick))
self.Handler.register(m(command_name="REPUTATION", func=self.on_reputation))
self.Handler.register(m(command_name="SMOD", func=self.on_smod))
self.Handler.register(m(command_name="SASL", func=self.on_sasl))
self.Handler.register(m(command_name="MD", func=self.on_md))
self.Handler.register(m(command_name="PRIVMSG", func=self.on_privmsg))
return None
def parse_server_msg(self, server_msg: list[str]) -> Optional[str]:
"""Parse the server message and return the command
@@ -229,6 +257,42 @@ class Unrealircd6(IProtocol):
self.__Irc.User.update_nickname(userObj.uid, newnickname)
return None
def send_set_mode(self, modes: str, *, nickname: Optional[str] = None, channel_name: Optional[str] = None, params: Optional[str] = None) -> None:
"""Set a mode to channel or to a nickname or for a user in a channel
Args:
modes (str): The selected mode
nickname (Optional[str]): The nickname
channel_name (Optional[str]): The channel name
params (Optional[str]): Parameters like password.
"""
service_id = self.__Config.SERVICE_ID
if modes[0] not in ['+', '-']:
self.__Logs.error(f"[MODE ERROR] The mode you have provided is missing the sign: {modes}")
return None
if nickname and channel_name:
# :98KAAAAAB MODE #services +o defenderdev
if not self.__Irc.Channel.is_valid_channel(channel_name):
self.__Logs.error(f"[MODE ERROR] The channel is not valid: {channel_name}")
return None
self.send2socket(f":{service_id} MODE {channel_name} {modes} {nickname}")
return None
if nickname and channel_name is None:
self.send2socket(f":{service_id} MODE {nickname} {modes}")
return None
if nickname is None and channel_name:
if not self.__Irc.Channel.is_valid_channel(channel_name):
self.__Logs.error(f"[MODE ERROR] The channel is not valid: {channel_name}")
return None
self.send2socket(f":{service_id} MODE {channel_name} {modes} {params}")
return None
return None
def send_squit(self, server_id: str, server_link: str, reason: str) -> None:
if not reason:
@@ -431,7 +495,7 @@ class Unrealircd6(IProtocol):
reason (str): The reason for the quit
"""
user_obj = self.__Irc.User.get_user(uidornickname=uid)
reputationObj = self.__Irc.Reputation.get_Reputation(uidornickname=uid)
reputationObj = self.__Irc.Reputation.get_reputation(uidornickname=uid)
if not user_obj is None:
self.send2socket(f":{user_obj.uid} QUIT :{reason}", print_log=print_log)
@@ -565,6 +629,103 @@ class Unrealircd6(IProtocol):
return None
# ------------------------------------------------------------------------
# COMMON IRC PARSER
# ------------------------------------------------------------------------
def parse_uid(self, serverMsg: list[str]) -> dict[str, str]:
"""Parse UID and return dictionary.
>>> ['@s2s-md/geoip=cc=GBtag...', ':001', 'UID', 'albatros', '0', '1721564597', 'albatros', 'hostname...', '001HB8G04', '0', '+iwxz', 'hostname-vhost', 'hostname-vhost', 'MyZBwg==', ':...']
Args:
serverMsg (list[str]): The UID ircd response
"""
umodes = str(serverMsg[10])
remote_ip = self.__Base.decode_ip(str(serverMsg[13])) if 'S' not in umodes else '127.0.0.1'
# Extract Geoip information
pattern = r'^.*geoip=cc=(\S{2}).*$'
geoip_match = match(pattern, serverMsg[0])
geoip = geoip_match.group(1) if geoip_match else None
response = {
'uid': str(serverMsg[8]),
'nickname': str(serverMsg[3]),
'username': str(serverMsg[6]),
'hostname': str(serverMsg[7]),
'umodes': umodes,
'vhost': str(serverMsg[11]),
'ip': remote_ip,
'realname': ' '.join(serverMsg[12:]).lstrip(':'),
'geoip': geoip,
'reputation_score': 0,
'iswebirc': True if 'webirc' in serverMsg[0] else False,
'iswebsocket': True if 'websocket' in serverMsg[0] else False
}
return response
def parse_quit(self, serverMsg: list[str]) -> dict[str, str]:
"""Parse quit and return dictionary.
>>> # ['@unrealtag...', ':001JKNY0N', 'QUIT', ':Quit:', '....']
Args:
serverMsg (list[str]): The server message to parse
Returns:
dict[str, str]: The dictionary.
"""
scopy = serverMsg.copy()
if scopy[0].startswith('@'):
scopy.pop(0)
response = {
"uid": scopy[0].replace(':', ''),
"reason": " ".join(scopy[3:])
}
return response
def parse_nick(self, serverMsg: list[str]) -> dict[str, str]:
"""Parse nick changes and return dictionary.
>>> ['@unrealircd.org/geoip=FR;unrealircd.org/', ':001OOU2H3', 'NICK', 'WebIrc', '1703795844']
Args:
serverMsg (list[str]): The server message to parse
Returns:
dict[str, str]: The response as dictionary.
"""
scopy = serverMsg.copy()
if scopy[0].startswith('@'):
scopy.pop(0)
response = {
"uid": scopy[0].replace(':', ''),
"newnickname": scopy[2],
"timestamp": scopy[3]
}
return response
def parse_privmsg(self, serverMsg: list[str]) -> dict[str, str]:
"""Parse PRIVMSG message.
>>> ['@....', ':97KAAAAAE', 'PRIVMSG', '#welcome', ':This', 'is', 'my', 'public', 'message']
>>> [':97KAAAAAF', 'PRIVMSG', '98KAAAAAB', ':sasa']
Args:
serverMsg (list[str]): The server message to parse
Returns:
dict[str, str]: The response as dictionary.
"""
scopy = serverMsg.copy()
if scopy[0].startswith('@'):
scopy.pop(0)
response = {
"uid_sender": scopy[0].replace(':', ''),
"uid_reciever": self.__Irc.User.get_uid(scopy[2]),
"channel": scopy[2] if self.__Irc.Channel.is_valid_channel(scopy[2]) else None,
"message": " ".join(scopy[3:])
}
return response
#####################
# HANDLE EVENTS #
#####################
@@ -736,6 +897,7 @@ class Unrealircd6(IProtocol):
self.__Irc.User.update_nickname(uid, newnickname)
self.__Irc.Client.update_nickname(uid, newnickname)
self.__Irc.Admin.update_nickname(uid, newnickname)
self.__Irc.Reputation.update(uid, newnickname)
return None
@@ -856,6 +1018,8 @@ class Unrealircd6(IProtocol):
self.__Logs.info(f"# VERSION : {version} ")
self.__Logs.info(f"################################################")
self.send_sjoin(self.__Config.SERVICE_CHANLOG)
if self.__Base.check_for_new_version(False):
self.send_priv_msg(
nick_from=self.__Config.SERVICE_NICKNAME,
@@ -875,6 +1039,10 @@ class Unrealircd6(IProtocol):
for module in self.__Irc.ModuleUtils.model_get_loaded_modules().copy():
module.class_instance.cmd(server_msg_copy)
# Join saved channels & load existing modules
self.__Irc.join_saved_channels()
self.__Irc.ModuleUtils.db_load_all_existing_modules(self.__Irc)
return None
except IndexError as ie:
self.__Logs.error(f"{__name__} - Key Error: {ie}")
@@ -988,14 +1156,42 @@ class Unrealircd6(IProtocol):
dnickname = self.__Config.SERVICE_NICKNAME
dchanlog = self.__Config.SERVICE_CHANLOG
GREEN = self.__Config.COLORS.green
RED = self.__Config.COLORS.red
NOGC = self.__Config.COLORS.nogc
for module in self.__Irc.ModuleUtils.model_get_loaded_modules().copy():
module.class_instance.cmd(serverMsg)
# SASL authentication
# ['@s2s-md/..', ':001', 'UID', 'adator__', '0', '1755987444', '...', 'desktop-h1qck20.mshome.net', '001XLTT0U', '0', '+iwxz', '*', 'Clk-EC2256B2.mshome.net', 'rBKAAQ==', ':...']
uid = serverMsg[8]
nickname = serverMsg[3]
sasl_obj = self.__Irc.Sasl.get_sasl_obj(uid)
if sasl_obj:
if sasl_obj.auth_success:
self.__Irc.insert_db_admin(sasl_obj.client_uid, sasl_obj.username, sasl_obj.level, sasl_obj.language)
self.send_priv_msg(nick_from=dnickname,
msg=tr("[ %sSASL AUTH%s ] - %s (%s) is now connected successfuly to %s", GREEN, NOGC, nickname, sasl_obj.username, dnickname),
channel=dchanlog)
self.send_notice(nick_from=dnickname, nick_to=nickname, msg=tr("Successfuly connected to %s", dnickname))
else:
self.send_priv_msg(nick_from=dnickname,
msg=tr("[ %sSASL AUTH%s ] - %s provided a wrong password for this username %s", RED, NOGC, nickname, sasl_obj.username),
channel=dchanlog)
self.send_notice(nick_from=dnickname, nick_to=nickname, msg=tr("Wrong password!"))
# Delete sasl object!
self.__Irc.Sasl.delete_sasl_client(uid)
return None
# If no sasl authentication then auto connect via fingerprint
if self.__Irc.Admin.db_auth_admin_via_fingerprint(fingerprint, uid):
admin = self.__Irc.Admin.get_admin(uid)
account = admin.account if admin else ''
self.send_priv_msg(nick_from=dnickname,
msg=tr("[ %sSASL AUTO AUTH%s ] - %s (%s) is now connected successfuly to %s", GREEN, NOGC, nickname, account, dnickname),
channel=dchanlog)
msg=tr("[ %sSASL AUTO AUTH%s ] - %s (%s) is now connected successfuly to %s", GREEN, NOGC, nickname, account, dnickname),
channel=dchanlog)
self.send_notice(nick_from=dnickname, nick_to=nickname, msg=tr("Successfuly connected to %s", dnickname))
return None
@@ -1254,7 +1450,7 @@ class Unrealircd6(IProtocol):
except Exception as err:
self.__Logs.error(f'General Error: {err}')
def on_sasl(self, serverMsg: list[str], psasl: 'Sasl') -> Optional['MSasl']:
def on_sasl(self, serverMsg: list[str]) -> Optional['MSasl']:
"""Handle SASL coming from a server
Args:
@@ -1267,7 +1463,7 @@ class Unrealircd6(IProtocol):
# [':irc.local.org', 'SASL', 'defender-dev.deb.biz.st', '0014ZZH1F', 'S', 'EXTERNAL', 'zzzzzzzkey']
# [':irc.local.org', 'SASL', 'defender-dev.deb.biz.st', '00157Z26U', 'C', 'sasakey==']
# [':irc.local.org', 'SASL', 'defender-dev.deb.biz.st', '00157Z26U', 'D', 'A']
psasl = self.__Irc.Sasl
sasl_enabled = False
for smod in self.__Settings.SMOD_MODULES:
if smod.name == 'sasl':
@@ -1307,6 +1503,7 @@ class Unrealircd6(IProtocol):
sasl_obj.fingerprint = str(sCopy[6])
self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {sasl_obj.client_uid} C +")
self.on_sasl_authentication_process(sasl_obj)
return sasl_obj
case 'C':
@@ -1319,14 +1516,64 @@ class Unrealircd6(IProtocol):
sasl_obj.username = username
sasl_obj.password = password
self.on_sasl_authentication_process(sasl_obj)
return sasl_obj
elif sasl_obj.mechanisme == "EXTERNAL":
sasl_obj.message_type = sasl_message_type
self.on_sasl_authentication_process(sasl_obj)
return sasl_obj
except Exception as err:
self.__Logs.error(f'General Error: {err}', exc_info=True)
def on_sasl_authentication_process(self, sasl_model: 'MSasl') -> bool:
s = sasl_model
if sasl_model:
def db_get_admin_info(*, username: Optional[str] = None, password: Optional[str] = None, fingerprint: Optional[str] = None) -> Optional[dict[str, Any]]:
if fingerprint:
mes_donnees = {'fingerprint': fingerprint}
query = f"SELECT user, level, language FROM {self.__Config.TABLE_ADMIN} WHERE fingerprint = :fingerprint"
else:
mes_donnees = {'user': username, 'password': self.__Utils.hash_password(password)}
query = f"SELECT user, level, language FROM {self.__Config.TABLE_ADMIN} WHERE user = :user AND password = :password"
result = self.__Base.db_execute_query(query, mes_donnees)
user_from_db = result.fetchone()
if user_from_db:
return {'user': user_from_db[0], 'level': user_from_db[1], 'language': user_from_db[2]}
else:
return None
if s.message_type == 'C' and s.mechanisme == 'PLAIN':
# Connection via PLAIN
admin_info = 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')
self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S")
self.send2socket(f":{self.__Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful")
else:
self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F")
self.send2socket(f":{self.__Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed")
elif s.message_type == 'S' and s.mechanisme == 'EXTERNAL':
# Connection using fingerprints
admin_info = 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')
self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D S")
self.send2socket(f":{self.__Config.SERVEUR_LINK} 903 {s.username} :SASL authentication successful")
else:
# "904 <nick> :SASL authentication failed"
self.send2socket(f":{self.__Config.SERVEUR_LINK} SASL {self.__Settings.MAIN_SERVER_HOSTNAME} {s.client_uid} D F")
self.send2socket(f":{self.__Config.SERVEUR_LINK} 904 {s.username} :SASL authentication failed")
def on_md(self, serverMsg: list[str]) -> None:
"""Handle MD responses
[':001', 'MD', 'client', '001MYIZ03', 'certfp', ':d1235648...']