import sys
import logging
logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
-from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
+from resources.libraries.python.PacketVerifier import RxQueue, TxQueue,\
+ checksum_equal
from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6ND_NS
from scapy.layers.inet6 import ICMPv6NDOptDstLLAddr, ICMPv6NDOptSrcLLAddr
# receive ICMPv6 neighbor advertisement message
ether = rxq.recv(2, sent_packets)
if ether is None:
- rxq._proc.terminate()
raise RuntimeError('ICMPv6 echo reply Rx timeout')
if not ether.haslayer(IPv6):
- rxq._proc.terminate()
raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format(
ether.__repr__()))
ipv6 = ether['IPv6']
if not ipv6.haslayer(ICMPv6ND_NA):
- rxq._proc.terminate()
raise RuntimeError(
'Unexpected packet with no ICMPv6 ND-NA received {0}'.format(
ipv6.__repr__()))
# verify target address
if icmpv6_na.tgt != dst_ip:
- rxq._proc.terminate()
raise RuntimeError('Invalid target address {0} should be {1}'.format(
icmpv6_na.tgt, dst_ip))
if not icmpv6_na.haslayer(ICMPv6NDOptDstLLAddr):
- rxq._proc.terminate()
raise RuntimeError(
'Missing Destination Link-Layer Address option in ICMPv6 ' +
'Neighbor Advertisement {0}'.format(icmpv6_na.__repr__()))
# verify destination link-layer address field
if dst_ll_addr.lladdr != dst_mac:
- rxq._proc.terminate()
raise RuntimeError('Invalid lladdr {0} should be {1}'.format(
dst_ll_addr.lladdr, dst_mac))
cksum = icmpv6_na.cksum
del icmpv6_na.cksum
tmp = ICMPv6ND_NA(str(icmpv6_na))
- if tmp.cksum != cksum:
- rxq._proc.terminate()
+ if not checksum_equal(tmp.cksum, cksum):
raise RuntimeError(
'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum))
- rxq._proc.terminate()
sys.exit(0)
if __name__ == "__main__":