Oxbian/SIDPS
1
0
Fork 0
mirror of https://github.com/Oxbian/SIDPS.git synced 2025-07-07 12:24:38 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: adding cooldown to alerts + more precise pattern matching for TCP

Browse Source
This commit is contained in:
Oxbian
2024-11-19 16:42:08 -05:00
parent 48e9554d38
commit 3e75cb8927
3 changed files with 77 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
idps/tcp.py
View File

@ -19,74 +19,81 @@ class TCP:
self.packets[ip_src] = []
if flags == "S":
self.packets[ip_src].append([port_src, ip_dst, port_dst, [flags], timestamp])
self.packets[ip_src].append([port_src, ip_dst, port_dst, ["S"], timestamp])
return
elif flags == "SA":
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "S", True)
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "S")
if i is not None:
self.packets[ip_dst][i][3].append("SA")
self.packets[ip_dst][i][4] = timestamp
print(f"i: {i}, {ip_src}:{port_src}->{ip_dst}:{port_dst}, paquets: \n{self.packets}")
self.packets[ip][i][3].append("SA")
self.packets[ip][i][4] = timestamp
return
else:
self.packets[ip_src].append([port_src, ip_dst, port_dst, [flags], timestamp])
self.packets[ip_src].append([port_src, ip_dst, port_dst, ["SA"], timestamp])
return
elif flags == "A":
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "SA")
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "SA")
if i is None:
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "R", True)
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "R")
if i is not None:
self.packets[ip_src][i][3].append("A")
self.packets[ip_src][i][4] = timestamp
print(f"i: {i}, {ip_src}:{port_src}->{ip_dst}:{port_dst}, paquets: \n{self.packets}")
self.packets[ip][i][3].append("A")
self.packets[ip][i][4] = timestamp
return
else:
self.packets[ip_src].append([port_src, ip_dst, port_dst, [flags], timestamp])
self.packets[ip_src].append([port_src, ip_dst, port_dst, ["A"], timestamp])
return
elif flags == "RA":
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "A")
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "A")
if i is None:
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "S")
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "S")
if i is not None:
self.packets[ip_src][i][3].append("RA")
self.packets[ip_src][i][4] = timestamp
print(f"i: {i}, {ip_src}:{port_src}->{ip_dst}:{port_dst}, paquets: \n{self.packets}")
self.packets[ip][i][3].append("RA")
self.packets[ip][i][4] = timestamp
return
else:
self.packets[ip_src].append([port_src, ip_dst, port_dst, [flags], timestamp])
self.packets[ip_src].append([port_src, ip_dst, port_dst, ["RA"], timestamp])
return
elif flags == "R":
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "A")
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "A")
if i is None:
i = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "S")
i, ip = self.find_packet_to_replace(ip_src, port_src, ip_dst, port_dst, "S")
if i is not None:
self.packets[ip_src][i][3].append("R")
self.packets[ip_src][i][4] = timestamp
print(f"i: {i}, {ip_src}:{port_src}->{ip_dst}:{port_dst}, paquets: \n{self.packets}")
self.packets[ip][i][3].append("R")
self.packets[ip][i][4] = timestamp
return
else:
self.packets[ip_src].append([port_src, ip_dst, port_dst, [flags], timestamp])
self.packets[ip_src].append([port_src, ip_dst, port_dst, ["R"], timestamp])
return
def find_packet_to_replace(self, ip_src, port_src, ip_dst, port_dst, flags, reverse=False):
"""Cherche l'indice du paquet dont le flag doit être remplacé"""
if reverse is True:
ip_src, ip_dst = ip_dst, ip_src
port_src, port_dst = port_dst, port_src
def find_packet_to_replace(self, ip_src, port_src, ip_dst, port_dst, flags):
"""Cherche l'indice et le port de source du paquet dont le flag doit être remplacé"""
if ip_src not in self.packets.keys():
return None
# Recherche dans le sens src->dst
if ip_src in self.packets.keys():
for i, [p_s, ip_d, p_d, f, stamp] in enumerate(self.packets[ip_src]):
if p_s == port_src and ip_d == ip_dst and p_d == port_dst and flags in f:
return i, ip_src
for i, [p_s, ip_d, p_d, f, stamp] in enumerate(self.packets[ip_src]):
if p_s == port_src and ip_d == ip_dst and p_d == port_dst and flags in f:
return i
return None
# Recherche dans le sens dst->src
if ip_dst in self.packets.keys():
for i, [p_d, ip_s, p_s, f, stamp] in enumerate(self.packets[ip_dst]):
if p_s == port_src and ip_s == ip_src and p_d == port_dst and flags in f:
return i, ip_dst
return None, None
def clean_old_packets(self):
"""Supprime les paquets qui date de plus longtemps que le temps de clean"""
@ -108,14 +115,16 @@ class TCP:
if not self.packets[ip_src]:
del self.packets[ip_src]
def count_packet_of_type(self, flag, time_treshold):
def count_packet_of_type(self, flag, time_treshold, isList = False):
"""Compte les paquets qui ont le flag choisi et qui sont dans la fenêtre de temps"""
count = 0
current_timestamp = time.time()
for ip in list(self.packets.keys()):
for packet in self.packets[ip]:
if flag in packet[3] and packet[4] >= current_timestamp - time_treshold:
if isList and set(flag) == set(packet[3]) and packet[4] >= current_timestamp - time_treshold:
count += 1
elif not isList and flag in packet[3] and packet[4] >= current_timestamp - time_treshold:
count += 1
return count