Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
IsDNSMalicious("./out_DNS.txt")
sys.exit(0)
if options.pcapfile is not None:
if (options.srcIP or options.dstIP) == False:
print "choose -s or -d"
sys.exit(0)
f = open(options.pcapfile)
try:
pcap = dpkt.pcapng.Reader(f)
except:
print "it is not pcapng format..."
f.close()
finally:
f = open(options.pcapfile)
pcap = dpkt.pcap.Reader(f)
printPcap(pcap, options.srcIP, options.dstIP)
parseIPlistLocation("./out_IP.txt")
if options.checkIP == True:
print "-------------check ip--------------"
IsMalicious("./ip_location.txt")
sys.exit(0)
def initialize(self,path):
f = open(path)
pcap = dpkt.pcap.Reader(f)
for _,buf in pcap:
#data = dpkt.ethernet.Ethernet(buf).ip.data.data
ip = dpkt.ethernet.Ethernet(buf).ip
tcp = ip.data
myip = socket.inet_ntoa(ip.dst)
if myip is self.ip and tcp.dport is self.port and len(tcp.data) is not 0:
self.out_streams.append((len(tcp.data),tcp.data))
elif len(tcp.data) is not 0:
self.in_streams.append((len(tcp.data),tcp.data))
f.close()
def __iter__(self):
if not self.fileobj:
self.fileobj = open(self.name, "rb")
self.fd = dpkt.pcap.Reader(self.fileobj)
self.fditer = iter(self.fd)
self.linktype = self.fd.datalink()
return self
def pcap_parser(fname):
f = open(fname, "rb")
pcap = dpkt.pcap.Reader(f)
index = 0
for _, buf in pcap:
index = index + 1
eth = dpkt.ethernet.Ethernet(buf)
data = eth.data
if isinstance(data, dpkt.cdp.CDP) or isinstance(data, dpkt.stp.STP):
continue
try:
llc = LLC(data)
data = llc.data
classification = llc.classification
if isinstance(data, dpkt.cdp.CDP) or isinstance(data, dpkt.stp.STP):
continue
Raises:
UnableToParseFile: when the file cannot be parsed.
"""
data = file_object.read(dpkt.pcap.FileHdr.__hdr_len__)
try:
file_header = dpkt.pcap.FileHdr(data)
packet_header_class = dpkt.pcap.PktHdr
except (dpkt.NeedData, dpkt.UnpackError) as exception:
raise errors.UnableToParseFile(
'[{0:s}] unable to parse file: {1:s} with error: {2!s}'.format(
self.NAME, parser_mediator.GetDisplayName(), exception))
if file_header.magic == dpkt.pcap.PMUDPCT_MAGIC:
try:
file_header = dpkt.pcap.LEFileHdr(data)
packet_header_class = dpkt.pcap.LEPktHdr
except (dpkt.NeedData, dpkt.UnpackError) as exception:
raise errors.UnableToParseFile(
'[{0:s}] unable to parse file: {1:s} with error: {2!s}'.format(
self.NAME, parser_mediator.GetDisplayName(), exception))
elif file_header.magic != dpkt.pcap.TCPDUMP_MAGIC:
raise errors.UnableToParseFile('Unsupported file signature')
packet_number = 1
connections = {}
other_list = []
trunc_list = []
def packets_for_stream(fobj, offset):
"""Open a PCAP, seek to a packet offset, then get all packets belonging to
the same connection."""
pcap = dpkt.pcap.Reader(fobj)
pcapiter = iter(pcap)
ts, raw = pcapiter.next()
fobj.seek(offset)
for p in next_connection_packets(pcapiter, linktype=pcap.datalink()):
yield p
def __init__(self, fileobj):
self.name = fileobj.name
self.fd = fileobj.fileno()
self.__f = fileobj
buf = self.__f.read(dpkt.pcap.FileHdr.__hdr_len__)
self.__fh = dpkt.pcap.FileHdr(buf)
self.__ph = dpkt.pcap.PktHdr
if self.__fh.magic == dpkt.pcap.PMUDPCT_MAGIC:
self.__fh = dpkt.pcap.LEFileHdr(buf)
self.__ph = dpkt.pcap.LEPktHdr
elif self.__fh.magic != dpkt.pcap.TCPDUMP_MAGIC:
raise ValueError, 'invalid tcpdump header'
self.snaplen = self.__fh.snaplen
self.dloff = dpkt.pcap.dltoff[self.__fh.linktype]
self.filter = ''
def read_from_pcap(file_path):
udp_dict = {}
with open(file_path, 'rb') as f:
pcap = dpkt.pcap.Reader(f)
for ts, buf in pcap:
eth = dpkt.ethernet.Ethernet(buf)
if not isinstance(eth.data, dpkt.ip.IP) or not isinstance(eth.data.data, dpkt.udp.UDP):
continue
ip = eth.data
udp = ip.data
laddr = inet_to_str(ip.src)
lport = udp.sport
faddr = inet_to_str(ip.dst)
fport = udp.dport
if not laddr in udp_dict:
udp_dict[laddr] = []
udp_dict[laddr].append((laddr, faddr, lport, fport, ts))
return udp_dict
def parse_pcap(pcap_file):
streams = dict() # Connections with current buffer
pcap = dpkt.pcap.Reader(pcap_file)
for ts, buf in pcap:
eth = dpkt.ethernet.Ethernet(buf)
if eth.type != dpkt.ethernet.ETH_TYPE_IP:
continue
ip = eth.data
if not isinstance(ip, dpkt.ip.IP):
try:
ip = dpkt.ip.IP(ip)
except:
continue
if ip.p != dpkt.ip.IP_PROTO_TCP:
continue
tcp = ip.data
if not isinstance(tcp, dpkt.tcp.TCP):