my-ceg-archive/AUNMS---Network-Monitoring-System
1
0
Fork 0
mirror of https://github.com/20kaushik02/AUNMS---Network-Monitoring-System.git synced 2025-12-06 06:44:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Ping function, some imports and error handling

Browse Source
This commit is contained in:
Kaushik 2021-12-23 23:31:16 +05:30
parent 26d762f1fc
commit 3cc3aa40f8
3 changed files with 91 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
network_monitor_thread.py
View File

@ -1,8 +1,11 @@
import sys
from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
from scapy.all import *
from scapy.layers.inet import *
from scapy.layers.l2 import *
class NetworkMonitorThread(QObject):
def __init__(self, interface, parent=None):
@ -10,19 +13,20 @@ class NetworkMonitorThread(QObject):
self.interface = interface
self.packetList = []
self.end = False
quitBool = pyqtSignal(int)
def endSniff(self):
QApplication.processEvents()
print("Ending")
self.end = True
self.quitBool.emit(1)
def sniffStatus(self):
QApplication.processEvents()
return self.end
def getLayers(self,packet):
def getLayers(self, packet):
QApplication.processEvents()
layers = []
counter = 0
@ -35,9 +39,10 @@ class NetworkMonitorThread(QObject):
break
counter += 1
return layers
packetData = pyqtSignal(tuple)
def handlePacket(self,packet):
def handlePacket(self, packet):
self.packetList.append(packet)
QApplication.processEvents()
tableViewPart = dict()
@ -50,9 +55,9 @@ class NetworkMonitorThread(QObject):
tableViewPart['destination'] = packet.dst
tableViewPart['length'] = len(packet)
tableViewPart['layers'] = self.getLayers(packet)
QApplication.processEvents()
(protocol,info) = self.getInfo(packet)
(protocol, info) = self.getInfo(packet)
tableViewPart['info'] = info
if protocol:
tableViewPart['Protocol'] = protocol
@ -62,17 +67,19 @@ class NetworkMonitorThread(QObject):
tableViewPart['Protocol'] = "Other"
QApplication.processEvents()
self.packetData.emit((packet, tableViewPart))
def getInfo(self,packet):
def getInfo(self, packet):
QApplication.processEvents()
info = ""
protocol = ""
if UDP in packet:
protocol = "UDP"
info = "{} -> {} len={} chksum={}".format(packet[UDP].sport,
packet[UDP].dport,
packet[UDP].len,
packet[UDP].chksum)
info = "{} -> {} len={} chksum={}".format(
packet[UDP].sport,
packet[UDP].dport,
packet[UDP].len,
packet[UDP].chksum
)
elif TCP in packet:
flags = {
'F': 'FIN',
@ -84,26 +91,35 @@ class NetworkMonitorThread(QObject):
'E': 'ECE',
'C': 'CWR',
}
flgs = str([flags[x] for x in packet.sprintf('%TCP.flags%')])
protocol = "TCP"
info = "{} -> {} {} seq={} ack={} window={}".format(packet[TCP].sport,
packet[TCP].dport,
flgs,
packet[TCP].seq,
packet[TCP].ack,
packet[TCP].window)
info = "{} -> {} {} seq={} ack={} window={}".format(
packet[TCP].sport,
packet[TCP].dport,
flgs,
packet[TCP].seq,
packet[TCP].ack,
packet[TCP].window
)
elif ARP in packet:
protocol = "ARP"
info = "hwtype={} ptype={} hwlen={} plen={} op={}".format(packet[ARP].hwtype,
packet[ARP].ptype,
packet[ARP].hwlen,
packet[ARP].plen,
packet[ARP].op)
info = "hwtype={} ptype={} hwlen={} plen={} op={}".format(
packet[ARP].hwtype,
packet[ARP].ptype,
packet[ARP].hwlen,
packet[ARP].plen,
packet[ARP].op
)
QApplication.processEvents()
return (protocol,info)
return (protocol, info)
def startSniff(self):
QApplication.processEvents()
self.pkts = sniff(count=0, iface=self.interface, prn=self.handlePacket, stop_filter=lambda x: self.sniffStatus())
self.pkts = sniff(
count=0,
iface=self.interface,
prn=self.handlePacket,
stop_filter={lambda x: self.sniffStatus()}
)

30
ping.py
View File

@ -1,28 +1,40 @@
import sys
from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
from scapy.all import *
from scapy.layers.inet import IP
from scapy.layers.inet import ICMP
class Ping(QWidget):
def __init__(self):
super().__init__()
self.hostn = QLabel("Enter Address/Hostname to ping:")
self.host = QLineEdit()
self.startPingBtn = QPushButton()
self.startPingBtn.setText('Ping')
self.startPingBtn.clicked.connect(self.startPing)
self.result = QTextEdit()
self.result.setReadOnly(True)
self.layoutPing = QVBoxLayout(self)
self.layoutPing.addWidget(self.hostn)
self.layoutPing.addWidget(self.host)
self.layoutPing.addWidget(self.startPingBtn)
self.layoutPing.addWidget(self.result)
self.setLayout(self.layoutPing)
def startPing():
pass
def startPing(self):
if(len(self.host.text()) >= 1):
try:
packet = IP(dst=self.host.text(), ttl=10)/ICMP()
output = sr1(packet, timeout=2)
if output is not None:
self.result.setText(output.summary())
except socket.gaierror as e:
self.result.setText(
"Invalid address/Could not get address info")

36
trace_route.py
View File

@ -1,35 +1,43 @@
from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
from scapy.all import *
from scapy.layers.inet import traceroute
class TraceRoute(QWidget):
def __init__(self):
super().__init__()
self.hostn = QLabel("Enter Address/Hostname to trace:")
self.host = QLineEdit()
self.startTraceBtn = QPushButton()
self.startTraceBtn.setText('Trace Route')
self.startTraceBtn.clicked.connect(self.startTrace)
self.result = QTextEdit()
self.result.setReadOnly(True)
self.layoutTrace = QVBoxLayout(self)
self.layoutTrace.addWidget(self.hostn)
self.layoutTrace.addWidget(self.host)
self.layoutTrace.addWidget(self.startTraceBtn)
self.layoutTrace.addWidget(self.result)
self.setLayout(self.layoutTrace)
def startTrace(self):
result, unans = traceroute(target=self.host.text(), dport=80, verbose=0)
output = []
output.append("\tRoute path\tResponse time")
for snd, rcv in result:
output.append(str(f"{snd.ttl}\t{rcv.src}\t\t{(int((rcv.time - snd.sent_time)*1000))} ms"))
output.append(f"\nUnanswered packets: {len(unans)}")
self.result.setText('\n'.join(output))
if(len(self.host.text()) >= 1):
try:
result, unans = traceroute(
target=self.host.text(), dport=80, verbose=0)
output = []
output.append("\tRoute path\tResponse time")
for snd, rcv in result:
output.append(
str(f"{snd.ttl}\t{rcv.src}\t\t{(int((rcv.time - snd.sent_time)*1000))} ms"))
output.append(f"\nUnanswered packets: {len(unans)}")
self.result.setText('\n'.join(output))
except socket.gaierror as e:
self.result.setText(
"Invalid address/Could not get address info")