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

V5.3.5:

Browse Source 
- core/irc.py : Nothing special except
        a notice sent to the user that the command
        is not available
    - mod_clone:
        When the user unload the module, the
        server will unset modes of #clone channel
    - mod_defender:
        * adding a try block to catch errors
        * adding a mode block to check if
        the jail channel is not having +b mode
        * ensure defender is +o in jail channel
    - mod_test:
        * adding some try block to catch errors
This commit is contained in:
adator
2024-10-02 00:01:13 +02:00
parent 3cdee5fddf
commit 12c7e5e832
4 changed files with 181 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
core/irc.py
View File

@@ -1010,7 +1010,8 @@ class Irc:
arg.remove(f':{self.Config.SERVICE_PREFIX}') arg.remove(f':{self.Config.SERVICE_PREFIX}')
if not arg[0].lower() in self.commands: if not arg[0].lower() in self.commands:
self.Base.logs.debug(f"This command {arg[0]} is not available") self.Base.logs.debug(f"This command {arg[0]} is not available")
return False self.sendNotice(f"This command [{self.Config.COLORS.bold}{arg[0]}{self.Config.COLORS.bold}] is not available", user_trigger)
return None
cmd_to_send = convert_to_string.replace(':','') cmd_to_send = convert_to_string.replace(':','')
self.Base.log_cmd(user_trigger, cmd_to_send) self.Base.log_cmd(user_trigger, cmd_to_send)

3
mods/mod_clone.py
View File

@@ -60,6 +60,7 @@ class Clone():
self.Base.db_query_channel(action='add', module_name=self.module_name, channel_name=self.Config.CLONE_CHANNEL) self.Base.db_query_channel(action='add', module_name=self.module_name, channel_name=self.Config.CLONE_CHANNEL)
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} JOIN {self.Config.CLONE_CHANNEL}") self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} JOIN {self.Config.CLONE_CHANNEL}")
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} SAMODE {self.Config.CLONE_CHANNEL} +o {self.Config.SERVICE_NICKNAME}")
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} +nts") self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} +nts")
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} +k {self.Config.CLONE_CHANNEL_PASSWORD}") self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} +k {self.Config.CLONE_CHANNEL_PASSWORD}")
@@ -126,6 +127,8 @@ class Clone():
self.Irc.send2socket(f':{self.Config.SERVICE_NICKNAME} PRIVMSG {clone} :KILL') self.Irc.send2socket(f':{self.Config.SERVICE_NICKNAME} PRIVMSG {clone} :KILL')
self.Base.db_query_channel(action='del', module_name=self.module_name, channel_name=self.Config.CLONE_CHANNEL) self.Base.db_query_channel(action='del', module_name=self.module_name, channel_name=self.Config.CLONE_CHANNEL)
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} -nts")
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} MODE {self.Config.CLONE_CHANNEL} -k {self.Config.CLONE_CHANNEL_PASSWORD}")
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} PART {self.Config.CLONE_CHANNEL}") self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} PART {self.Config.CLONE_CHANNEL}")
return None return None

321
mods/mod_defender.py
View File

@@ -147,7 +147,8 @@ class Defender():
self.Base.create_thread(func=self.thread_reputation_timer) self.Base.create_thread(func=self.thread_reputation_timer)
if self.ModConfig.reputation == 1: if self.ModConfig.reputation == 1:
self.Irc.send2socket(f":{self.Config.SERVEUR_ID} SJOIN {self.Base.get_unixtime()} {self.Config.SALON_JAIL} + :{self.Config.SERVICE_ID}") self.Irc.send2socket(f":{self.Config.SERVEUR_ID} SJOIN {self.Base.get_unixtime()} {self.Config.SALON_JAIL} + :{self.Config.SERVICE_NICKNAME}")
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} SAMODE {self.Config.SALON_JAIL} +o {self.Config.SERVICE_NICKNAME}")
return None return None
@@ -963,184 +964,206 @@ class Defender():
self.Logs.error(f"Thread_cloudfilt_scan Error : {ve}") self.Logs.error(f"Thread_cloudfilt_scan Error : {ve}")
def cmd(self, data: list) -> None: def cmd(self, data: list) -> None:
try:
service_id = self.Config.SERVICE_ID # Defender serveur id
cmd = list(data).copy()
service_id = self.Config.SERVICE_ID # Defender serveur id match cmd[1]:
cmd = list(data).copy()
match cmd[1]: case 'REPUTATION':
# :001 REPUTATION 91.168.141.239 118
try:
self.reputation_first_connexion['ip'] = cmd[2]
self.reputation_first_connexion['score'] = cmd[3]
if str(cmd[3]).find('*') != -1:
# If the reputation changed, we do not need to scan the IP
return None
case 'REPUTATION': if not self.Base.is_valid_ip(cmd[2]):
# :001 REPUTATION 91.168.141.239 118 return None
try:
self.reputation_first_connexion['ip'] = cmd[2] # Possibilité de déclancher les bans a ce niveau.
self.reputation_first_connexion['score'] = cmd[3] except IndexError as ie:
if str(cmd[3]).find('*') != -1: self.Logs.error(f'cmd reputation: index error: {ie}')
# If the reputation changed, we do not need to scan the IP
if len(cmd) < 3:
return None
match cmd[2]:
case 'MODE':
# ['...', ':001XSCU0Q', 'MODE', '#jail', '+b', '~security-group:unknown-users']
channel = str(cmd[3])
mode = str(cmd[4])
group_to_check = str(cmd[5:])
group_to_unban = '~security-group:unknown-users'
if self.Config.SALON_JAIL == channel:
if mode == '+b' and group_to_unban in group_to_check:
self.Irc.send2socket(f":{service_id} MODE {self.Config.SALON_JAIL} -b ~security-group:unknown-users")
self.Irc.send2socket(f":{service_id} MODE {self.Config.SALON_JAIL} -eee ~security-group:webirc-users ~security-group:known-users ~security-group:websocket-users")
case 'PRIVMSG':
cmd.pop(0)
user_trigger = str(cmd[0]).replace(':','')
channel = cmd[2]
find_nickname = self.User.get_nickname(user_trigger)
self.flood(find_nickname, channel)
case 'UID':
# If Init then do nothing
if self.Irc.INIT == 1:
return None return None
if not self.Base.is_valid_ip(cmd[2]): # Supprimer la premiere valeur et finir le code normalement
return None cmd.pop(0)
# Possibilité de déclancher les bans a ce niveau. # Get User information
except IndexError as ie: _User = self.User.get_User(str(cmd[7]))
self.Logs.error(f'cmd reputation: index error: {ie}')
match cmd[2]: # If user is not service or IrcOp then scan them
case 'PRIVMSG':
cmd.pop(0)
user_trigger = str(cmd[0]).replace(':','')
channel = cmd[2]
find_nickname = self.User.get_nickname(user_trigger)
self.flood(find_nickname, channel)
case 'UID':
# If Init then do nothing
if self.Irc.INIT == 1:
return None
# Supprimer la premiere valeur et finir le code normalement
cmd.pop(0)
# Get User information
_User = self.User.get_User(str(cmd[7]))
# If user is not service or IrcOp then scan them
if not re.match(fr'^.*[S|o?].*$', _User.umodes):
self.abuseipdb_UserModel.append(_User) if self.ModConfig.abuseipdb_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
self.freeipapi_UserModel.append(_User) if self.ModConfig.freeipapi_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
self.cloudfilt_UserModel.append(_User) if self.ModConfig.cloudfilt_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
self.psutil_UserModel.append(_User) if self.ModConfig.psutil_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
self.localscan_UserModel.append(_User) if self.ModConfig.local_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
if _User is None:
self.Logs.critical(f'This UID: [{cmd[7]}] is not available please check why')
return None
reputation_flag = self.ModConfig.reputation
reputation_seuil = self.ModConfig.reputation_seuil
if self.Irc.INIT == 0:
# Si le user n'es pas un service ni un IrcOP
if not re.match(fr'^.*[S|o?].*$', _User.umodes): if not re.match(fr'^.*[S|o?].*$', _User.umodes):
if reputation_flag == 1 and _User.score_connexion <= reputation_seuil: self.abuseipdb_UserModel.append(_User) if self.ModConfig.abuseipdb_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
currentDateTime = self.Base.get_datetime() self.freeipapi_UserModel.append(_User) if self.ModConfig.freeipapi_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
self.reputation_insert( self.cloudfilt_UserModel.append(_User) if self.ModConfig.cloudfilt_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
self.ReputationModel( self.psutil_UserModel.append(_User) if self.ModConfig.psutil_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
uid=_User.uid, nickname=_User.nickname, username=_User.username, realname=_User.realname, self.localscan_UserModel.append(_User) if self.ModConfig.local_scan == 1 and not _User.remote_ip in self.Config.WHITELISTED_IP else None
hostname=_User.hostname, umodes=_User.umodes, vhost=_User.vhost, ip=_User.remote_ip, score=_User.score_connexion,
secret_code=self.Base.get_random(8), isWebirc=_User.isWebirc, isWebsocket=_User.isWebsocket, connected_datetime=currentDateTime, if _User is None:
updated_datetime=currentDateTime self.Logs.critical(f'This UID: [{cmd[7]}] is not available please check why')
return None
reputation_flag = self.ModConfig.reputation
reputation_seuil = self.ModConfig.reputation_seuil
if self.Irc.INIT == 0:
# Si le user n'es pas un service ni un IrcOP
if not re.match(fr'^.*[S|o?].*$', _User.umodes):
if reputation_flag == 1 and _User.score_connexion <= reputation_seuil:
currentDateTime = self.Base.get_datetime()
self.reputation_insert(
self.ReputationModel(
uid=_User.uid, nickname=_User.nickname, username=_User.username, realname=_User.realname,
hostname=_User.hostname, umodes=_User.umodes, vhost=_User.vhost, ip=_User.remote_ip, score=_User.score_connexion,
secret_code=self.Base.get_random(8), isWebirc=_User.isWebirc, isWebsocket=_User.isWebsocket, connected_datetime=currentDateTime,
updated_datetime=currentDateTime
)
) )
) # self.Irc.send2socket(f":{service_id} WHOIS {nickname}")
# self.Irc.send2socket(f":{service_id} WHOIS {nickname}") if self.reputation_check(_User.uid):
if self.reputation_check(_User.uid): if reputation_flag == 1 and _User.score_connexion <= reputation_seuil:
if reputation_flag == 1 and _User.score_connexion <= reputation_seuil: self.system_reputation(_User.uid)
self.system_reputation(_User.uid) self.Logs.info('Démarrer le systeme de reputation')
self.Logs.info('Démarrer le systeme de reputation')
case 'SJOIN': case 'SJOIN':
# ['@msgid=F9B7JeHL5pj9nN57cJ5pEr;time=2023-12-28T20:47:24.305Z', ':001', 'SJOIN', '1702138958', '#welcome', ':0015L1AHL'] # ['@msgid=F9B7JeHL5pj9nN57cJ5pEr;time=2023-12-28T20:47:24.305Z', ':001', 'SJOIN', '1702138958', '#welcome', ':0015L1AHL']
try: try:
cmd.pop(0)
parsed_chan = cmd[3]
if self.ModConfig.reputation == 1:
parsed_UID = cmd[4]
pattern = fr'^:[@|%|\+|~|\*]*'
parsed_UID = re.sub(pattern, '', parsed_UID)
get_reputation = self.reputation_get_Reputation(parsed_UID)
if parsed_chan != self.Config.SALON_JAIL:
self.Irc.send2socket(f":{service_id} MODE {parsed_chan} +b ~security-group:unknown-users")
self.Irc.send2socket(f":{service_id} MODE {parsed_chan} +eee ~security-group:webirc-users ~security-group:known-users ~security-group:websocket-users")
if not get_reputation is None:
isWebirc = get_reputation.isWebirc
if not isWebirc:
if parsed_chan != self.Config.SALON_JAIL:
self.Irc.send2socket(f":{service_id} SAPART {get_reputation.nickname} {parsed_chan}")
if self.ModConfig.reputation_ban_all_chan == 1 and not isWebirc:
if parsed_chan != self.Config.SALON_JAIL:
self.Irc.send2socket(f":{service_id} MODE {parsed_chan} +b {get_reputation.nickname}!*@*")
self.Irc.send2socket(f":{service_id} KICK {parsed_chan} {get_reputation.nickname}")
self.Logs.debug(f'SJOIN parsed_uid : {parsed_UID}')
except KeyError as ke:
self.Logs.error(f"key error SJOIN : {ke}")
case 'SLOG':
# self.Base.scan_ports(cmd[7])
cmd.pop(0) cmd.pop(0)
parsed_chan = cmd[3]
if self.ModConfig.reputation == 1: if not self.Base.is_valid_ip(cmd[7]):
parsed_UID = cmd[4] return None
pattern = fr'^:[@|%|\+|~|\*]*'
parsed_UID = re.sub(pattern, '', parsed_UID)
get_reputation = self.reputation_get_Reputation(parsed_UID) # if self.ModConfig.local_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
# self.localscan_remote_ip.append(cmd[7])
if parsed_chan != self.Config.SALON_JAIL: # if self.ModConfig.psutil_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
self.Irc.send2socket(f":{service_id} MODE {parsed_chan} +b ~security-group:unknown-users") # self.psutil_remote_ip.append(cmd[7])
self.Irc.send2socket(f":{service_id} MODE {parsed_chan} +eee ~security-group:webirc-users ~security-group:known-users ~security-group:websocket-users")
if not get_reputation is None: # if self.ModConfig.abuseipdb_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
isWebirc = get_reputation.isWebirc # self.abuseipdb_remote_ip.append(cmd[7])
if not isWebirc: # if self.ModConfig.freeipapi_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
if parsed_chan != self.Config.SALON_JAIL: # self.freeipapi_remote_ip.append(cmd[7])
self.Irc.send2socket(f":{service_id} SAPART {get_reputation.nickname} {parsed_chan}")
if self.ModConfig.reputation_ban_all_chan == 1 and not isWebirc: # if self.ModConfig.cloudfilt_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
if parsed_chan != self.Config.SALON_JAIL: # self.cloudfilt_remote_ip.append(cmd[7])
self.Irc.send2socket(f":{service_id} MODE {parsed_chan} +b {get_reputation.nickname}!*@*")
self.Irc.send2socket(f":{service_id} KICK {parsed_chan} {get_reputation.nickname}")
self.Logs.debug(f'SJOIN parsed_uid : {parsed_UID}') case 'NICK':
except KeyError as ke: # :0010BS24L NICK [NEWNICK] 1697917711
self.Logs.error(f"key error SJOIN : {ke}") # Changement de nickname
try:
cmd.pop(0)
uid = str(cmd[0]).replace(':','')
get_Reputation = self.reputation_get_Reputation(uid)
jail_salon = self.Config.SALON_JAIL
service_id = self.Config.SERVICE_ID
case 'SLOG': if get_Reputation is None:
# self.Base.scan_ports(cmd[7]) self.Logs.debug(f'This UID: {uid} is not listed in the reputation dataclass')
cmd.pop(0) return None
if not self.Base.is_valid_ip(cmd[7]): # Update the new nickname
return None oldnick = get_Reputation.nickname
newnickname = cmd[2]
get_Reputation.nickname = newnickname
# if self.ModConfig.local_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP: # If ban in all channel is ON then unban old nickname an ban the new nickname
# self.localscan_remote_ip.append(cmd[7]) if self.ModConfig.reputation_ban_all_chan == 1:
for chan in self.Channel.UID_CHANNEL_DB:
if chan.name != jail_salon:
self.Irc.send2socket(f":{service_id} MODE {chan.name} -b {oldnick}!*@*")
self.Irc.send2socket(f":{service_id} MODE {chan.name} +b {newnickname}!*@*")
# if self.ModConfig.psutil_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP: except KeyError as ke:
# self.psutil_remote_ip.append(cmd[7]) self.Logs.error(f'cmd - NICK - KeyError: {ke}')
# if self.ModConfig.abuseipdb_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP: case 'QUIT':
# self.abuseipdb_remote_ip.append(cmd[7]) # :001N1WD7L QUIT :Quit: free_znc_1
# if self.ModConfig.freeipapi_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
# self.freeipapi_remote_ip.append(cmd[7])
# if self.ModConfig.cloudfilt_scan == 1 and not cmd[7] in self.Config.WHITELISTED_IP:
# self.cloudfilt_remote_ip.append(cmd[7])
case 'NICK':
# :0010BS24L NICK [NEWNICK] 1697917711
# Changement de nickname
try:
cmd.pop(0) cmd.pop(0)
uid = str(cmd[0]).replace(':','') ban_all_chan = self.Base.int_if_possible(self.ModConfig.reputation_ban_all_chan)
get_Reputation = self.reputation_get_Reputation(uid) user_id = str(cmd[0]).replace(':','')
final_UID = user_id
jail_salon = self.Config.SALON_JAIL jail_salon = self.Config.SALON_JAIL
service_id = self.Config.SERVICE_ID service_id = self.Config.SERVICE_ID
if get_Reputation is None: get_user_reputation = self.reputation_get_Reputation(final_UID)
self.Logs.debug(f'This UID: {uid} is not listed in the reputation dataclass')
return None
# Update the new nickname if not get_user_reputation is None:
oldnick = get_Reputation.nickname final_nickname = get_user_reputation.nickname
newnickname = cmd[2]
get_Reputation.nickname = newnickname
# If ban in all channel is ON then unban old nickname an ban the new nickname
if self.ModConfig.reputation_ban_all_chan == 1:
for chan in self.Channel.UID_CHANNEL_DB: for chan in self.Channel.UID_CHANNEL_DB:
if chan.name != jail_salon: if chan.name != jail_salon and ban_all_chan == 1:
self.Irc.send2socket(f":{service_id} MODE {chan.name} -b {oldnick}!*@*") self.Irc.send2socket(f":{service_id} MODE {chan.name} -b {final_nickname}!*@*")
self.Irc.send2socket(f":{service_id} MODE {chan.name} +b {newnickname}!*@*") self.reputation_delete(final_UID)
except KeyError as ke: except KeyError as ke:
self.Logs.error(f'cmd - NICK - KeyError: {ke}') self.Logs.error(f"{ke} / {cmd} / length {str(len(cmd))}")
except IndexError as ie:
case 'QUIT': self.Logs.error(f"{ie} / {cmd} / length {str(len(cmd))}")
# :001N1WD7L QUIT :Quit: free_znc_1 except Exception as err:
cmd.pop(0) self.Logs.error(f"General Error: {err}")
ban_all_chan = self.Base.int_if_possible(self.ModConfig.reputation_ban_all_chan)
user_id = str(cmd[0]).replace(':','')
final_UID = user_id
jail_salon = self.Config.SALON_JAIL
service_id = self.Config.SERVICE_ID
get_user_reputation = self.reputation_get_Reputation(final_UID)
if not get_user_reputation is None:
final_nickname = get_user_reputation.nickname
for chan in self.Channel.UID_CHANNEL_DB:
if chan.name != jail_salon and ban_all_chan == 1:
self.Irc.send2socket(f":{service_id} MODE {chan.name} -b {final_nickname}!*@*")
self.reputation_delete(final_UID)
def _hcmds(self, user:str, channel: any, cmd: list, fullcmd: list = []) -> None: def _hcmds(self, user:str, channel: any, cmd: list, fullcmd: list = []) -> None:

4
mods/mod_test.py
View File

@@ -128,6 +128,10 @@ class Test():
cmd = list(data).copy() cmd = list(data).copy()
return None return None
except KeyError as ke:
self.Base.logs.error(f"Key Error: {ke}")
except IndexError as ie:
self.Base.logs.error(f"{ie} / {cmd} / length {str(len(cmd))}")
except Exception as err: except Exception as err:
self.Base.logs.error(f"General Error: {err}") self.Base.logs.error(f"General Error: {err}")