mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-10-05 22:18:16 +08:00
Code Issues Releases Wiki Activity

switch pcap libraries #137

Browse Source
This commit is contained in:
Joel Bender 2018-03-05 14:37:36 -05:00
parent 3130ca3d9c
commit 22ce545404
3 changed files with 63 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
py25/bacpypes/analysis.py
View File

@ -2,6 +2,14 @@
""" """
Analysis - Decoding pcap files Analysis - Decoding pcap files
Before analyzing files, install libpcap-dev:
$ sudo apt install libpcap-dev
then install pypcap:
https://github.com/pynetwork/pypcap
""" """
import sys import sys
@ -15,7 +23,7 @@ try:
except: except:
pass pass
from .debugging import ModuleLogger, DebugContents, bacpypes_debugging from .debugging import ModuleLogger, DebugContents, bacpypes_debugging, btox
from .pdu import PDU, Address from .pdu import PDU, Address
from .bvll import BVLPDU, bvl_pdu_types, ForwardedNPDU, \ from .bvll import BVLPDU, bvl_pdu_types, ForwardedNPDU, \
@ -33,13 +41,6 @@ _protocols={socket.IPPROTO_TCP:'tcp',
socket.IPPROTO_UDP:'udp', socket.IPPROTO_UDP:'udp',
socket.IPPROTO_ICMP:'icmp'} socket.IPPROTO_ICMP:'icmp'}
#
# _hexify
#
def _hexify(s, sep='.'):
return sep.join('%02X' % ord(c) for c in s)
# #
# strftimestamp # strftimestamp
# #
@ -53,11 +54,11 @@ def strftimestamp(ts):
# #
def decode_ethernet(s): def decode_ethernet(s):
if _debug: decode_ethernet._debug("decode_ethernet %s...", _hexify(s[:14])) if _debug: decode_ethernet._debug("decode_ethernet %s...", btox(s[:14]))
d={} d={}
d['destination_address'] = _hexify(s[0:6], ':') d['destination_address'] = btox(s[0:6], ':')
d['source_address'] = _hexify(s[6:12], ':') d['source_address'] = btox(s[6:12], ':')
d['type'] = struct.unpack('!H',s[12:14])[0] d['type'] = struct.unpack('!H',s[12:14])[0]
d['data'] = s[14:] d['data'] = s[14:]
@ -70,7 +71,7 @@ bacpypes_debugging(decode_ethernet)
# #
def decode_vlan(s): def decode_vlan(s):
if _debug: decode_vlan._debug("decode_vlan %s...", _hexify(s[:4])) if _debug: decode_vlan._debug("decode_vlan %s...", btox(s[:4]))
d = {} d = {}
x = struct.unpack('!H',s[0:2])[0] x = struct.unpack('!H',s[0:2])[0]
@ -89,7 +90,7 @@ bacpypes_debugging(decode_vlan)
# #
def decode_ip(s): def decode_ip(s):
if _debug: decode_ip._debug("decode_ip %r", _hexify(s[:20])) if _debug: decode_ip._debug("decode_ip %r", btox(s[:20]))
d = {} d = {}
d['version'] = (ord(s[0]) & 0xf0) >> 4 d['version'] = (ord(s[0]) & 0xf0) >> 4
@ -119,7 +120,7 @@ bacpypes_debugging(decode_ip)
# #
def decode_udp(s): def decode_udp(s):
if _debug: decode_udp._debug("decode_udp %s...", _hexify(s[:8])) if _debug: decode_udp._debug("decode_udp %s...", btox(s[:8]))
d = {} d = {}
d['source_port'] = struct.unpack('!H',s[0:2])[0] d['source_port'] = struct.unpack('!H',s[0:2])[0]
@ -225,7 +226,7 @@ def decode_packet(data):
# check for version number # check for version number
if (pdu.pduData[0] != '\x01'): if (pdu.pduData[0] != '\x01'):
if _debug: decode_packet._debug(" - not a version 1 packet: %s...", _hexify(pdu.pduData[:30])) if _debug: decode_packet._debug(" - not a version 1 packet: %s...", btox(pdu.pduData[:30]))
return None return None
# it's an NPDU # it's an NPDU
@ -355,33 +356,7 @@ def decode_file(fname):
"""Given the name of a pcap file, open it, decode the contents and yield each packet.""" """Given the name of a pcap file, open it, decode the contents and yield each packet."""
if _debug: decode_file._debug("decode_file %r", fname) if _debug: decode_file._debug("decode_file %r", fname)
if not pcap: raise NotImplementedError("not implemented")
raise RuntimeError("failed to import pcap")
# create a pcap object
p = pcap.pcapObject()
p.open_offline(fname)
i = 0
while 1:
# the object acts like an iterator
pkt = p.next()
if not pkt:
break
# returns a tuple
pktlen, data, timestamp = pkt
pkt = decode_packet(data)
if not pkt:
continue
# save the index and timestamp in the packet
pkt._index = i
pkt._timestamp = timestamp
yield pkt
i += 1
bacpypes_debugging(decode_file) bacpypes_debugging(decode_file)

48
py27/bacpypes/analysis.py
View File

@ -2,6 +2,14 @@
""" """
Analysis - Decoding pcap files Analysis - Decoding pcap files
Before analyzing files, install libpcap-dev:
$ sudo apt install libpcap-dev
then install pypcap:
https://github.com/pynetwork/pypcap
""" """
import sys import sys
@ -15,7 +23,7 @@ try:
except: except:
pass pass
from .debugging import ModuleLogger, DebugContents, bacpypes_debugging from .debugging import ModuleLogger, DebugContents, bacpypes_debugging, btox
from .pdu import PDU, Address from .pdu import PDU, Address
from .bvll import BVLPDU, bvl_pdu_types, ForwardedNPDU, \ from .bvll import BVLPDU, bvl_pdu_types, ForwardedNPDU, \
@ -33,13 +41,6 @@ _protocols={socket.IPPROTO_TCP:'tcp',
socket.IPPROTO_UDP:'udp', socket.IPPROTO_UDP:'udp',
socket.IPPROTO_ICMP:'icmp'} socket.IPPROTO_ICMP:'icmp'}
#
# _hexify
#
def _hexify(s, sep='.'):
return sep.join('%02X' % ord(c) for c in s)
# #
# strftimestamp # strftimestamp
# #
@ -54,11 +55,11 @@ def strftimestamp(ts):
@bacpypes_debugging @bacpypes_debugging
def decode_ethernet(s): def decode_ethernet(s):
if _debug: decode_ethernet._debug("decode_ethernet %s...", _hexify(s[:14])) if _debug: decode_ethernet._debug("decode_ethernet %s...", btox(s[:14]))
d={} d={}
d['destination_address'] = _hexify(s[0:6], ':') d['destination_address'] = btox(s[0:6], ':')
d['source_address'] = _hexify(s[6:12], ':') d['source_address'] = btox(s[6:12], ':')
d['type'] = struct.unpack('!H',s[12:14])[0] d['type'] = struct.unpack('!H',s[12:14])[0]
d['data'] = s[14:] d['data'] = s[14:]
@ -70,7 +71,7 @@ def decode_ethernet(s):
@bacpypes_debugging @bacpypes_debugging
def decode_vlan(s): def decode_vlan(s):
if _debug: decode_vlan._debug("decode_vlan %s...", _hexify(s[:4])) if _debug: decode_vlan._debug("decode_vlan %s...", btox(s[:4]))
d = {} d = {}
x = struct.unpack('!H',s[0:2])[0] x = struct.unpack('!H',s[0:2])[0]
@ -88,7 +89,7 @@ def decode_vlan(s):
@bacpypes_debugging @bacpypes_debugging
def decode_ip(s): def decode_ip(s):
if _debug: decode_ip._debug("decode_ip %r", _hexify(s[:20])) if _debug: decode_ip._debug("decode_ip %r", btox(s[:20]))
d = {} d = {}
d['version'] = (ord(s[0]) & 0xf0) >> 4 d['version'] = (ord(s[0]) & 0xf0) >> 4
@ -117,7 +118,7 @@ def decode_ip(s):
@bacpypes_debugging @bacpypes_debugging
def decode_udp(s): def decode_udp(s):
if _debug: decode_udp._debug("decode_udp %s...", _hexify(s[:8])) if _debug: decode_udp._debug("decode_udp %s...", btox(s[:8]))
d = {} d = {}
d['source_port'] = struct.unpack('!H',s[0:2])[0] d['source_port'] = struct.unpack('!H',s[0:2])[0]
@ -222,7 +223,7 @@ def decode_packet(data):
# check for version number # check for version number
if (pdu.pduData[0] != '\x01'): if (pdu.pduData[0] != '\x01'):
if _debug: decode_packet._debug(" - not a version 1 packet: %s...", _hexify(pdu.pduData[:30])) if _debug: decode_packet._debug(" - not a version 1 packet: %s...", btox(pdu.pduData[:30]))
return None return None
# it's an NPDU # it's an NPDU
@ -355,30 +356,19 @@ def decode_file(fname):
raise RuntimeError("failed to import pcap") raise RuntimeError("failed to import pcap")
# create a pcap object # create a pcap object
p = pcap.pcapObject() p = pcap.pcap(fname)
p.open_offline(fname)
i = 0 for timestamp, data in p:
while 1:
# the object acts like an iterator
pkt = p.next()
if not pkt:
break
# returns a tuple
pktlen, data, timestamp = pkt
pkt = decode_packet(data) pkt = decode_packet(data)
if not pkt: if not pkt:
continue continue
# save the index and timestamp in the packet # save the index and timestamp in the packet
pkt._index = i # pkt._index = i
pkt._timestamp = timestamp pkt._timestamp = timestamp
yield pkt yield pkt
i += 1
# #
# Tracer # Tracer
# #

64
py34/bacpypes/analysis.py
View File

@ -2,6 +2,14 @@
""" """
Analysis - Decoding pcap files Analysis - Decoding pcap files
Before analyzing files, install libpcap-dev:
$ sudo apt install libpcap-dev
then install pypcap:
https://github.com/pynetwork/pypcap
""" """
import sys import sys
@ -15,7 +23,7 @@ try:
except: except:
pass pass
from .debugging import ModuleLogger, DebugContents, bacpypes_debugging from .debugging import ModuleLogger, DebugContents, bacpypes_debugging, btox
from .pdu import PDU, Address from .pdu import PDU, Address
from .bvll import BVLPDU, bvl_pdu_types, ForwardedNPDU, \ from .bvll import BVLPDU, bvl_pdu_types, ForwardedNPDU, \
@ -33,13 +41,6 @@ _protocols={socket.IPPROTO_TCP:'tcp',
socket.IPPROTO_UDP:'udp', socket.IPPROTO_UDP:'udp',
socket.IPPROTO_ICMP:'icmp'} socket.IPPROTO_ICMP:'icmp'}
#
# _hexify
#
def _hexify(s, sep='.'):
return sep.join('%02X' % ord(c) for c in s)
# #
# strftimestamp # strftimestamp
# #
@ -54,11 +55,11 @@ def strftimestamp(ts):
@bacpypes_debugging @bacpypes_debugging
def decode_ethernet(s): def decode_ethernet(s):
if _debug: decode_ethernet._debug("decode_ethernet %s...", _hexify(s[:14])) if _debug: decode_ethernet._debug("decode_ethernet %s...", btox(s[:14], '.'))
d={} d={}
d['destination_address'] = _hexify(s[0:6], ':') d['destination_address'] = btox(s[0:6], ':')
d['source_address'] = _hexify(s[6:12], ':') d['source_address'] = btox(s[6:12], ':')
d['type'] = struct.unpack('!H',s[12:14])[0] d['type'] = struct.unpack('!H',s[12:14])[0]
d['data'] = s[14:] d['data'] = s[14:]
@ -70,7 +71,7 @@ def decode_ethernet(s):
@bacpypes_debugging @bacpypes_debugging
def decode_vlan(s): def decode_vlan(s):
if _debug: decode_vlan._debug("decode_vlan %s...", _hexify(s[:4])) if _debug: decode_vlan._debug("decode_vlan %s...", btox(s[:4]))
d = {} d = {}
x = struct.unpack('!H',s[0:2])[0] x = struct.unpack('!H',s[0:2])[0]
@ -88,18 +89,18 @@ def decode_vlan(s):
@bacpypes_debugging @bacpypes_debugging
def decode_ip(s): def decode_ip(s):
if _debug: decode_ip._debug("decode_ip %r", _hexify(s[:20])) if _debug: decode_ip._debug("decode_ip %r", btox(s[:20], '.'))
d = {} d = {}
d['version'] = (ord(s[0]) & 0xf0) >> 4 d['version'] = (s[0] & 0xf0) >> 4
d['header_len'] = ord(s[0]) & 0x0f d['header_len'] = s[0] & 0x0f
d['tos'] = ord(s[1]) d['tos'] = s[1]
d['total_len'] = struct.unpack('!H',s[2:4])[0] d['total_len'] = struct.unpack('!H',s[2:4])[0]
d['id'] = struct.unpack('!H',s[4:6])[0] d['id'] = struct.unpack('!H',s[4:6])[0]
d['flags'] = (ord(s[6]) & 0xe0) >> 5 d['flags'] = (s[6] & 0xe0) >> 5
d['fragment_offset'] = struct.unpack('!H',s[6:8])[0] & 0x1f d['fragment_offset'] = struct.unpack('!H',s[6:8])[0] & 0x1f
d['ttl'] = ord(s[8]) d['ttl'] = s[8]
d['protocol'] = _protocols.get(ord(s[9]), '0x%.2x ?' % ord(s[9])) d['protocol'] = _protocols.get(s[9], '0x%.2x ?' % s[9])
d['checksum'] = struct.unpack('!H',s[10:12])[0] d['checksum'] = struct.unpack('!H',s[10:12])[0]
d['source_address'] = socket.inet_ntoa(s[12:16]) d['source_address'] = socket.inet_ntoa(s[12:16])
d['destination_address'] = socket.inet_ntoa(s[16:20]) d['destination_address'] = socket.inet_ntoa(s[16:20])
@ -117,7 +118,7 @@ def decode_ip(s):
@bacpypes_debugging @bacpypes_debugging
def decode_udp(s): def decode_udp(s):
if _debug: decode_udp._debug("decode_udp %s...", _hexify(s[:8])) if _debug: decode_udp._debug("decode_udp %s...", btox(s[:8]))
d = {} d = {}
d['source_port'] = struct.unpack('!H',s[0:2])[0] d['source_port'] = struct.unpack('!H',s[0:2])[0]
@ -187,7 +188,7 @@ def decode_packet(data):
pdu = PDU(data, source=pduSource, destination=pduDestination) pdu = PDU(data, source=pduSource, destination=pduDestination)
# check for a BVLL header # check for a BVLL header
if (pdu.pduData[0] == '\x81'): if (pdu.pduData[0] == 0x81):
if _debug: decode_packet._debug(" - BVLL header found") if _debug: decode_packet._debug(" - BVLL header found")
xpdu = BVLPDU() xpdu = BVLPDU()
@ -221,8 +222,8 @@ def decode_packet(data):
return xpdu return xpdu
# check for version number # check for version number
if (pdu.pduData[0] != '\x01'): if (pdu.pduData[0] != 0x01):
if _debug: decode_packet._debug(" - not a version 1 packet: %s...", _hexify(pdu.pduData[:30])) if _debug: decode_packet._debug(" - not a version 1 packet: %s...", btox(pdu.pduData[:30], '.'))
return None return None
# it's an NPDU # it's an NPDU
@ -355,30 +356,19 @@ def decode_file(fname):
raise RuntimeError("failed to import pcap") raise RuntimeError("failed to import pcap")
# create a pcap object # create a pcap object
p = pcap.pcapObject() p = pcap.pcap(fname)
p.open_offline(fname)
i = 0 for timestamp, data in p:
while 1:
# the object acts like an iterator
pkt = p.next()
if not pkt:
break
# returns a tuple
pktlen, data, timestamp = pkt
pkt = decode_packet(data) pkt = decode_packet(data)
if not pkt: if not pkt:
continue continue
# save the index and timestamp in the packet # save the index and timestamp in the packet
pkt._index = i # pkt._index = i
pkt._timestamp = timestamp pkt._timestamp = timestamp
yield pkt yield pkt
i += 1
# #
# Tracer # Tracer
# #