make test: unify packet checksum verifications
Change-Id: If9cc7c5e32ebecff398fd38b39e8f485754a4ad4
Signed-off-by: Klement Sekera <ksekera@cisco.com>
diff --git a/test/framework.py b/test/framework.py
index 9732145..f90197b 100644
--- a/test/framework.py
+++ b/test/framework.py
@@ -25,6 +25,9 @@
from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
getLogger, colorize
from vpp_object import VppObjectRegistry
+from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
+from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
+from scapy.layers.inet6 import ICMPv6EchoReply
if os.name == 'posix' and sys.version_info[0] < 3:
# using subprocess32 is recommended by python official documentation
# @ https://docs.python.org/2/library/subprocess.html
@@ -32,6 +35,7 @@
else:
import subprocess
+
debug_framework = False
if os.getenv('TEST_DEBUG', "0") == "1":
debug_framework = True
@@ -729,6 +733,87 @@
name, real_value, expected_min, expected_max)
self.assertTrue(expected_min <= real_value <= expected_max, msg)
+ def assert_packet_checksums_valid(self, packet,
+ ignore_zero_udp_checksums=True):
+ udp_layers = ['UDP', 'UDPerror']
+ checksum_fields = ['cksum', 'chksum']
+ checksums = []
+ counter = 0
+ temp = packet.__class__(str(packet))
+ while True:
+ layer = temp.getlayer(counter)
+ if layer:
+ for cf in checksum_fields:
+ if hasattr(layer, cf):
+ if ignore_zero_udp_checksums and \
+ 0 == getattr(layer, cf) and \
+ layer.name in udp_layers:
+ continue
+ delattr(layer, cf)
+ checksums.append((counter, cf))
+ else:
+ break
+ counter = counter + 1
+ temp = temp.__class__(str(temp))
+ for layer, cf in checksums:
+ self.assert_equal(getattr(packet[layer], cf),
+ getattr(temp[layer], cf),
+ "packet checksum on layer #%d: %s" % (
+ layer, temp[layer].name))
+
+ def assert_checksum_valid(self, received_packet, layer,
+ field_name='chksum',
+ ignore_zero_checksum=False):
+ """ Check checksum of received packet on given layer """
+ received_packet_checksum = getattr(received_packet[layer], field_name)
+ if ignore_zero_checksum and 0 == received_packet_checksum:
+ return
+ recalculated = received_packet.__class__(str(received_packet))
+ delattr(recalculated[layer], field_name)
+ recalculated = recalculated.__class__(str(recalculated))
+ self.assert_equal(received_packet_checksum,
+ getattr(recalculated[layer], field_name),
+ "packet checksum on layer: %s" % layer)
+
+ def assert_ip_checksum_valid(self, received_packet,
+ ignore_zero_checksum=False):
+ self.assert_checksum_valid(received_packet, 'IP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_tcp_checksum_valid(self, received_packet,
+ ignore_zero_checksum=False):
+ self.assert_checksum_valid(received_packet, 'TCP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_udp_checksum_valid(self, received_packet,
+ ignore_zero_checksum=True):
+ self.assert_checksum_valid(received_packet, 'UDP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_embedded_icmp_checksum_valid(self, received_packet):
+ if received_packet.haslayer(IPerror):
+ self.assert_checksum_valid(received_packet, 'IPerror')
+ if received_packet.haslayer(TCPerror):
+ self.assert_checksum_valid(received_packet, 'TCPerror')
+ if received_packet.haslayer(UDPerror):
+ self.assert_checksum_valid(received_packet, 'UDPerror',
+ ignore_zero_checksum=True)
+ if received_packet.haslayer(ICMPerror):
+ self.assert_checksum_valid(received_packet, 'ICMPerror')
+
+ def assert_icmp_checksum_valid(self, received_packet):
+ self.assert_checksum_valid(received_packet, 'ICMP')
+ self.assert_embedded_icmp_checksum_valid(received_packet)
+
+ def assert_icmpv6_checksum_valid(self, pkt):
+ if pkt.haslayer(ICMPv6DestUnreach):
+ self.assert_checksum_valid(pkt, 'ICMPv6DestUnreach', 'cksum')
+ self.assert_embedded_icmp_checksum_valid(pkt)
+ if pkt.haslayer(ICMPv6EchoRequest):
+ self.assert_checksum_valid(pkt, 'ICMPv6EchoRequest', 'cksum')
+ if pkt.haslayer(ICMPv6EchoReply):
+ self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
+
@classmethod
def sleep(cls, timeout, remark=None):
if hasattr(cls, 'logger'):
diff --git a/test/test_ipsec_nat.py b/test/test_ipsec_nat.py
index 9c22fbb..cebbfc8 100644
--- a/test/test_ipsec_nat.py
+++ b/test/test_ipsec_nat.py
@@ -84,29 +84,10 @@
ICMP(id=self.icmp_id_out, type='echo-request'))
]
- def check_checksum(self, pkt, layer):
- """ Check checksum of the packet on given layer """
- new = pkt.__class__(str(pkt))
- del new[layer].chksum
- new = new.__class__(str(new))
- self.assertEqual(new[layer].chksum, pkt[layer].chksum)
-
- def check_ip_checksum(self, pkt):
- return self.check_checksum(pkt, 'IP')
-
- def check_tcp_checksum(self, pkt):
- return self.check_checksum(pkt, 'TCP')
-
- def check_udp_checksum(self, pkt):
- return self.check_checksum(pkt, 'UDP')
-
- def check_icmp_checksum(self, pkt):
- return self.check_checksum(pkt, 'ICMP')
-
def verify_capture_plain(self, capture):
for packet in capture:
try:
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
self.assert_equal(packet[IP].src, self.pg0.remote_ip4,
"decrypted packet source address")
self.assert_equal(packet[IP].dst, self.pg1.remote_ip4,
@@ -117,7 +98,6 @@
"unexpected UDP header in decrypted packet")
self.assert_equal(packet[TCP].dport, self.tcp_port_in,
"decrypted packet TCP destination port")
- self.check_tcp_checksum(packet)
elif packet.haslayer(UDP):
if packet[UDP].payload:
self.assertFalse(
@@ -131,7 +111,6 @@
"unexpected UDP header in decrypted packet")
self.assert_equal(packet[ICMP].id, self.icmp_id_in,
"decrypted packet ICMP ID")
- self.check_icmp_checksum(packet)
except Exception:
self.logger.error(
ppp("Unexpected or invalid plain packet:", packet))
diff --git a/test/test_nat.py b/test/test_nat.py
index c2e36be..59641a2 100644
--- a/test/test_nat.py
+++ b/test/test_nat.py
@@ -7,7 +7,6 @@
import random
from framework import VppTestCase, VppTestRunner, running_extended_tests
-from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
@@ -26,114 +25,6 @@
class MethodHolder(VppTestCase):
""" NAT create capture and verify method holder """
- @classmethod
- def setUpClass(cls):
- super(MethodHolder, cls).setUpClass()
-
- def tearDown(self):
- super(MethodHolder, self).tearDown()
-
- def check_ip_checksum(self, pkt):
- """
- Check IP checksum of the packet
-
- :param pkt: Packet to check IP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['IP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['IP'].chksum, pkt['IP'].chksum)
-
- def check_tcp_checksum(self, pkt):
- """
- Check TCP checksum in IP packet
-
- :param pkt: Packet to check TCP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['TCP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['TCP'].chksum, pkt['TCP'].chksum)
-
- def check_udp_checksum(self, pkt):
- """
- Check UDP checksum in IP packet
-
- :param pkt: Packet to check UDP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['UDP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['UDP'].chksum, pkt['UDP'].chksum)
-
- def check_icmp_errror_embedded(self, pkt):
- """
- Check ICMP error embeded packet checksum
-
- :param pkt: Packet to check ICMP error embeded packet checksum
- """
- if pkt.haslayer(IPerror):
- new = pkt.__class__(str(pkt))
- del new['IPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['IPerror'].chksum, pkt['IPerror'].chksum)
-
- if pkt.haslayer(TCPerror):
- new = pkt.__class__(str(pkt))
- del new['TCPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['TCPerror'].chksum, pkt['TCPerror'].chksum)
-
- if pkt.haslayer(UDPerror):
- if pkt['UDPerror'].chksum != 0:
- new = pkt.__class__(str(pkt))
- del new['UDPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['UDPerror'].chksum,
- pkt['UDPerror'].chksum)
-
- if pkt.haslayer(ICMPerror):
- del new['ICMPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPerror'].chksum, pkt['ICMPerror'].chksum)
-
- def check_icmp_checksum(self, pkt):
- """
- Check ICMP checksum in IPv4 packet
-
- :param pkt: Packet to check ICMP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['ICMP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMP'].chksum, pkt['ICMP'].chksum)
- if pkt.haslayer(IPerror):
- self.check_icmp_errror_embedded(pkt)
-
- def check_icmpv6_checksum(self, pkt):
- """
- Check ICMPv6 checksum in IPv4 packet
-
- :param pkt: Packet to check ICMPv6 checksum
- """
- new = pkt.__class__(str(pkt))
- if pkt.haslayer(ICMPv6DestUnreach):
- del new['ICMPv6DestUnreach'].cksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPv6DestUnreach'].cksum,
- pkt['ICMPv6DestUnreach'].cksum)
- self.check_icmp_errror_embedded(pkt)
- if pkt.haslayer(ICMPv6EchoRequest):
- del new['ICMPv6EchoRequest'].cksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPv6EchoRequest'].cksum,
- pkt['ICMPv6EchoRequest'].cksum)
- if pkt.haslayer(ICMPv6EchoReply):
- del new['ICMPv6EchoReply'].cksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPv6EchoReply'].cksum,
- pkt['ICMPv6EchoReply'].cksum)
-
def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64):
"""
Create packet stream for inside network
@@ -383,7 +274,7 @@
for packet in capture:
try:
if not is_ip6:
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
self.assertEqual(packet[IP46].src, nat_ip)
if dst_ip is not None:
self.assertEqual(packet[IP46].dst, dst_ip)
@@ -394,7 +285,7 @@
self.assertNotEqual(
packet[TCP].sport, self.tcp_port_in)
self.tcp_port_out = packet[TCP].sport
- self.check_tcp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
elif packet.haslayer(UDP):
if same_port:
self.assertEqual(packet[UDP].sport, self.udp_port_in)
@@ -408,10 +299,7 @@
else:
self.assertNotEqual(packet[ICMP46].id, self.icmp_id_in)
self.icmp_id_out = packet[ICMP46].id
- if is_ip6:
- self.check_icmpv6_checksum(packet)
- else:
- self.check_icmp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet "
"(outside network):", packet))
@@ -442,16 +330,14 @@
self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
self.assertEqual(packet[IP].dst, in_if.remote_ip4)
if packet.haslayer(TCP):
self.assertEqual(packet[TCP].dport, self.tcp_port_in)
- self.check_tcp_checksum(packet)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
else:
self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- self.check_icmp_checksum(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet "
"(inside network):", packet))
@@ -471,16 +357,14 @@
try:
self.assertEqual(packet[IPv6].src, src_ip)
self.assertEqual(packet[IPv6].dst, dst_ip)
+ self.assert_packet_checksums_valid(packet)
if packet.haslayer(TCP):
self.assertEqual(packet[TCP].dport, self.tcp_port_in)
- self.check_tcp_checksum(packet)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
- self.check_udp_checksum(packet)
else:
self.assertEqual(packet[ICMPv6EchoReply].id,
self.icmp_id_in)
- self.check_icmpv6_checksum(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet "
"(inside network):", packet))
@@ -655,7 +539,7 @@
for p in frags:
self.assertEqual(p[IP].src, src)
self.assertEqual(p[IP].dst, dst)
- self.check_ip_checksum(p)
+ self.assert_ip_checksum_valid(p)
buffer.seek(p[IP].frag * 8)
buffer.write(p[IP].payload)
ip = frags[0].getlayer(IP)
@@ -663,7 +547,7 @@
proto=frags[0][IP].proto)
if ip.proto == IP_PROTOS.tcp:
p = (ip / TCP(buffer.getvalue()))
- self.check_tcp_checksum(p)
+ self.assert_tcp_checksum_valid(p)
elif ip.proto == IP_PROTOS.udp:
p = (ip / UDP(buffer.getvalue()))
return p
@@ -688,9 +572,9 @@
nh=frags[0][IPv6ExtHdrFragment].nh)
if ip.nh == IP_PROTOS.tcp:
p = (ip / TCP(buffer.getvalue()))
- self.check_tcp_checksum(p)
elif ip.nh == IP_PROTOS.udp:
p = (ip / UDP(buffer.getvalue()))
+ self.assert_packet_checksums_valid(p)
return p
def initiate_tcp_session(self, in_if, out_if):
@@ -1589,14 +1473,12 @@
self.pg_start()
capture = self.pg0.get_capture(1)
p = capture[0]
- server = None
try:
ip = p[IP]
tcp = p[TCP]
self.assertEqual(ip.dst, self.pg0.remote_ip4)
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1633,8 +1515,7 @@
tcp = p[TCP]
self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(tcp.sport, external_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1666,14 +1547,12 @@
self.pg_start()
capture = self.pg0.get_capture(1)
p = capture[0]
- server = None
try:
ip = p[IP]
tcp = p[TCP]
self.assertEqual(ip.dst, self.pg0.remote_ip4)
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1692,8 +1571,7 @@
tcp = p[TCP]
self.assertEqual(ip.src, self.pg0.remote_ip4)
self.assertEqual(tcp.sport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1786,8 +1664,7 @@
self.assertEqual(ip.src, self.pg1.remote_ip4)
self.assertEqual(tcp.dport, 56789)
self.assertEqual(tcp.sport, 12345)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1836,8 +1713,7 @@
else:
server = server2
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1856,8 +1732,7 @@
tcp = p[TCP]
self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(tcp.sport, external_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1954,8 +1829,7 @@
else:
server = server2
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1974,8 +1848,7 @@
tcp = p[TCP]
self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(tcp.sport, external_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -1995,8 +1868,7 @@
tcp = p[TCP]
self.assertEqual(ip.dst, server1.ip4)
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -2015,8 +1887,7 @@
tcp = p[TCP]
self.assertEqual(ip.src, server1.ip4)
self.assertEqual(tcp.sport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -2261,7 +2132,7 @@
self.assertEqual(ip.dst, server.ip4)
self.assertNotEqual(tcp.sport, host_in_port)
self.assertEqual(tcp.dport, server_in_port)
- self.check_tcp_checksum(p)
+ self.assert_packet_checksums_valid(p)
host_out_port = tcp.sport
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
@@ -2283,7 +2154,7 @@
self.assertEqual(ip.dst, host.ip4)
self.assertEqual(tcp.sport, server_out_port)
self.assertEqual(tcp.dport, host_in_port)
- self.check_tcp_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -2334,7 +2205,7 @@
self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
self.assertEqual(packet[TCP].dport, server_tcp_port)
self.tcp_port_out = packet[TCP].sport
- self.check_tcp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
elif packet.haslayer(UDP):
self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
self.assertEqual(packet[UDP].dport, server_udp_port)
@@ -2371,7 +2242,7 @@
if packet.haslayer(TCP):
self.assertEqual(packet[TCP].dport, self.tcp_port_in)
self.assertEqual(packet[TCP].sport, server_tcp_port)
- self.check_tcp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
self.assertEqual(packet[UDP].sport, server_udp_port)
@@ -2407,7 +2278,7 @@
self.assertEqual(packet[TCP].sport, self.tcp_port_in)
self.assertEqual(packet[TCP].dport, server_tcp_port)
self.tcp_port_out = packet[TCP].sport
- self.check_tcp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].sport, self.udp_port_in)
self.assertEqual(packet[UDP].dport, server_udp_port)
@@ -2444,7 +2315,7 @@
if packet.haslayer(TCP):
self.assertEqual(packet[TCP].dport, self.tcp_port_in)
self.assertEqual(packet[TCP].sport, server_tcp_port)
- self.check_tcp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
self.assertEqual(packet[UDP].sport, server_udp_port)
@@ -3062,7 +2933,7 @@
self.assertEqual(packet[IP].src, nat_ip)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3082,7 +2953,7 @@
self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3117,7 +2988,7 @@
self.assertEqual(packet[IP].src, host_nat_ip)
self.assertEqual(packet[IP].dst, server.ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3137,7 +3008,7 @@
self.assertEqual(packet[IP].src, server_nat_ip)
self.assertEqual(packet[IP].dst, host.ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3172,7 +3043,7 @@
self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3192,7 +3063,7 @@
self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3202,8 +3073,6 @@
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
host_in_port = 1234
- host_out_port = 0
- server_in_port = 5678
server_out_port = 8765
server_nat_ip = "10.0.0.11"
@@ -3238,7 +3107,7 @@
self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, server.ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3258,7 +3127,7 @@
self.assertEqual(packet[IP].src, server_nat_ip)
self.assertEqual(packet[IP].dst, host.ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -3385,7 +3254,7 @@
self.assertEqual(ip.dst, server.ip4)
self.assertNotEqual(tcp.sport, host_in_port)
self.assertEqual(tcp.dport, server_in_port)
- self.check_tcp_checksum(p)
+ self.assert_packet_checksums_valid(p)
host_out_port = tcp.sport
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
@@ -3407,7 +3276,7 @@
self.assertEqual(ip.dst, host.ip4)
self.assertEqual(tcp.sport, server_out_port)
self.assertEqual(tcp.dport, host_in_port)
- self.check_tcp_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3439,14 +3308,12 @@
self.pg_start()
capture = self.pg0.get_capture(1)
p = capture[0]
- server = None
try:
ip = p[IP]
tcp = p[TCP]
self.assertEqual(ip.dst, self.pg0.remote_ip4)
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3465,8 +3332,7 @@
tcp = p[TCP]
self.assertEqual(ip.src, external_addr)
self.assertEqual(tcp.sport, external_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3515,10 +3381,6 @@
capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
- tcp_port_out = self.tcp_port_out
- udp_port_out = self.udp_port_out
- icmp_id_out = self.icmp_id_out
-
# session initiaded from remote host - do not translate
pkts = self.create_stream_out(self.pg1,
self.pg0.remote_ip4,
@@ -3569,8 +3431,7 @@
self.assertEqual(tcp.sport, 12345)
self.assertEqual(ip.dst, self.pg1.remote_ip4)
self.assertEqual(tcp.dport, local_port)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3590,8 +3451,7 @@
self.assertEqual(tcp.sport, external_port)
self.assertEqual(ip.dst, self.pg0.remote_ip4)
self.assertEqual(tcp.dport, 12345)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3624,8 +3484,7 @@
self.assertNotEqual(tcp.sport, 12345)
external_port = tcp.sport
self.assertEqual(tcp.dport, 80)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3646,8 +3505,7 @@
self.assertEqual(ip.dst, local_host.ip4)
self.assertEqual(tcp.sport, 80)
self.assertEqual(tcp.dport, 12345)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3679,7 +3537,6 @@
self.pg_start()
capture = self.pg9.get_capture(1)
p = capture[0]
- server = None
try:
ip = p[IP]
tcp = p[TCP]
@@ -3688,8 +3545,7 @@
self.assertEqual(tcp.dport, local_port)
self.assertNotEqual(tcp.sport, 12345)
eh_port_in = tcp.sport
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3710,8 +3566,7 @@
self.assertEqual(ip.dst, remote_host.ip4)
self.assertEqual(tcp.sport, external_port)
self.assertEqual(tcp.dport, 12345)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -3815,10 +3670,8 @@
def test_reass_hairpinning(self):
""" NAT44 fragments hairpinning """
- host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
host_in_port = random.randint(1025, 65535)
- host_out_port = 0
server_in_port = random.randint(1025, 65535)
server_out_port = random.randint(1025, 65535)
data = "A" * 4 + "B" * 16 + "C" * 3
@@ -3921,8 +3774,7 @@
self.assertEqual(tcp.dport, 22)
self.assertNotEqual(tcp.sport, 4567)
self.assertEqual((tcp.sport >> 6) & 63, 10)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -4031,8 +3883,7 @@
eh_addr_in = ip.src
eh_port_in = tcp.sport
saved_port_in = tcp.dport
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -4052,8 +3903,7 @@
self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(tcp.dport, eh_port_out)
self.assertEqual(tcp.sport, port_out)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -5395,8 +5245,6 @@
self.udp_port_in = 6304
self.icmp_id_in = 6305
- ses_num_start = self.nat64_get_ses_num()
-
self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
self.nat_addr_n)
self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
@@ -5439,7 +5287,7 @@
inner = packet[IPerror]
self.assertEqual(inner.src, self.pg1.remote_ip4)
self.assertEqual(inner.dst, self.nat_addr)
- self.check_icmp_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
if inner.haslayer(TCPerror):
self.assertEqual(inner[TCPerror].dport, self.tcp_port_out)
elif inner.haslayer(UDPerror):
@@ -5468,7 +5316,7 @@
inner = icmp[IPerror6]
self.assertEqual(inner.src, self.pg0.remote_ip6)
self.assertEqual(inner.dst, ip.src)
- self.check_icmpv6_checksum(packet)
+ self.assert_icmpv6_checksum_valid(packet)
if inner.haslayer(TCPerror):
self.assertEqual(inner[TCPerror].sport, self.tcp_port_in)
elif inner.haslayer(UDPerror):
@@ -5530,15 +5378,14 @@
try:
self.assertEqual(packet[IPv6].src, nat_addr_ip6)
self.assertEqual(packet[IPv6].dst, server.ip6)
+ self.assert_packet_checksums_valid(packet)
if packet.haslayer(TCP):
self.assertNotEqual(packet[TCP].sport, client_tcp_in_port)
self.assertEqual(packet[TCP].dport, server_tcp_in_port)
- self.check_tcp_checksum(packet)
client_tcp_out_port = packet[TCP].sport
else:
self.assertNotEqual(packet[UDP].sport, client_udp_in_port)
self.assertEqual(packet[UDP].dport, server_udp_in_port)
- self.check_udp_checksum(packet)
client_udp_out_port = packet[UDP].sport
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
@@ -5562,14 +5409,13 @@
try:
self.assertEqual(packet[IPv6].src, nat_addr_ip6)
self.assertEqual(packet[IPv6].dst, client.ip6)
+ self.assert_packet_checksums_valid(packet)
if packet.haslayer(TCP):
self.assertEqual(packet[TCP].sport, server_tcp_out_port)
self.assertEqual(packet[TCP].dport, client_tcp_in_port)
- self.check_tcp_checksum(packet)
else:
self.assertEqual(packet[UDP].sport, server_udp_out_port)
self.assertEqual(packet[UDP].dport, client_udp_in_port)
- self.check_udp_checksum(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -5593,7 +5439,7 @@
inner = icmp[IPerror6]
self.assertEqual(inner.src, server.ip6)
self.assertEqual(inner.dst, nat_addr_ip6)
- self.check_icmpv6_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
if inner.haslayer(TCPerror):
self.assertEqual(inner[TCPerror].sport, server_tcp_in_port)
self.assertEqual(inner[TCPerror].dport,
@@ -5716,7 +5562,7 @@
self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -5854,8 +5700,7 @@
self.assertNotEqual(tcp.sport, 12345)
external_port = tcp.sport
self.assertEqual(tcp.dport, 80)
- self.check_tcp_checksum(p)
- self.check_ip_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -5876,7 +5721,7 @@
self.assertEqual(ip.dst, self.pg3.remote_ip6)
self.assertEqual(tcp.sport, 80)
self.assertEqual(tcp.dport, 12345)
- self.check_tcp_checksum(p)
+ self.assert_packet_checksums_valid(p)
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
@@ -5934,7 +5779,6 @@
def test_reass_hairpinning(self):
""" NAT64 fragments hairpinning """
data = 'a' * 200
- client = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
server_in_port = random.randint(1025, 65535)
server_out_port = random.randint(1025, 65535)
@@ -6360,7 +6204,7 @@
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertNotEqual(capture[UDP].sport, 20000)
self.assertEqual(capture[UDP].dport, 10000)
- self.check_ip_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
out_port = capture[UDP].sport
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@@ -6377,7 +6221,7 @@
self.assertEqual(capture[IP].dst, '192.168.1.1')
self.assertEqual(capture[UDP].sport, 10000)
self.assertEqual(capture[UDP].dport, 20000)
- self.check_ip_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
# TCP
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
@@ -6394,8 +6238,7 @@
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertNotEqual(capture[TCP].sport, 20001)
self.assertEqual(capture[TCP].dport, 10001)
- self.check_ip_checksum(capture)
- self.check_tcp_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
out_port = capture[TCP].sport
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@@ -6412,8 +6255,7 @@
self.assertEqual(capture[IP].dst, '192.168.1.1')
self.assertEqual(capture[TCP].sport, 10001)
self.assertEqual(capture[TCP].dport, 20001)
- self.check_ip_checksum(capture)
- self.check_tcp_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
# ICMP
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
@@ -6429,8 +6271,7 @@
self.assertEqual(capture[IP].src, self.nat_addr)
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertNotEqual(capture[ICMP].id, 4000)
- self.check_ip_checksum(capture)
- self.check_icmp_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
out_id = capture[ICMP].id
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@@ -6446,8 +6287,7 @@
self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
self.assertEqual(capture[IP].dst, '192.168.1.1')
self.assertEqual(capture[ICMP].id, 4000)
- self.check_ip_checksum(capture)
- self.check_icmp_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
# ping DS-Lite AFTR tunnel endpoint address
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
@@ -6534,7 +6374,7 @@
self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
self.assertEqual(capture[UDP].sport, 10000)
self.assertEqual(capture[UDP].dport, 20000)
- self.check_ip_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
# UDP decapsulation
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
@@ -6551,7 +6391,7 @@
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertEqual(capture[UDP].sport, 20000)
self.assertEqual(capture[UDP].dport, 10000)
- self.check_ip_checksum(capture)
+ self.assert_packet_checksums_valid(capture)
# ping DS-Lite B4 tunnel endpoint address
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
@@ -6632,12 +6472,7 @@
try:
self.assertEqual(packet[IPv6].src, self.nat_addr)
self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
- if packet.haslayer(TCP):
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.check_udp_checksum(packet)
- elif packet.haslayer(ICMPv6EchoRequest):
- self.check_icmpv6_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -6668,12 +6503,7 @@
try:
self.assertEqual(packet[IPv6].src, self.pg1.remote_ip6)
self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
- if packet.haslayer(TCP):
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.check_udp_checksum(packet)
- elif packet.haslayer(ICMPv6EchoReply):
- self.check_icmpv6_checksum(packet)
+ self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
@@ -6729,5 +6559,6 @@
self.logger.info(self.vapi.cli("show nat66 static mappings"))
self.clear_nat66()
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)