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
-from scapy.layers.l2 import Ether, ARP
+from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6
+from scapy.layers.l2 import Ether, ARP, GRE
from scapy.data import IP_PROTOS
from scapy.packet import bind_layers
from util import ppp
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, ttl=64):
"""
Create packet stream for inside network
self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
+ self.check_ip_checksum(packet)
self.assertEqual(packet[IP].src, nat_ip)
if dst_ip is not None:
self.assertEqual(packet[IP].dst, dst_ip)
self.assertNotEqual(
packet[TCP].sport, self.tcp_port_in)
self.tcp_port_out = packet[TCP].sport
+ self.check_tcp_checksum(packet)
elif packet.haslayer(UDP):
if same_port:
self.assertEqual(packet[UDP].sport, self.udp_port_in)
else:
self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
self.icmp_id_out = packet[ICMP].id
+ self.check_icmp_checksum(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet "
"(outside network):", packet))
self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
+ self.check_ip_checksum(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))
self.assertEqual(packet[IPv6].dst, dst_ip)
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))
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)
host_out_port = tcp.sport
except:
self.logger.error(ppp("Unexpected or invalid packet:", p))
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)
except:
self.logger.error(ppp("Unexpected or invalid packet:"), p)
raise
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)
elif packet.haslayer(UDP):
self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
self.assertEqual(packet[UDP].dport, server_udp_port)
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)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
self.assertEqual(packet[UDP].sport, server_udp_port)
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)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].sport, self.udp_port_in)
self.assertEqual(packet[UDP].dport, server_udp_port)
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)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
self.assertEqual(packet[UDP].sport, server_udp_port)
capture = self.pg8.get_capture(len(pkts))
self.verify_capture_out(capture)
+ def test_static_unknown_proto(self):
+ """ 1:1 NAT translate packet with unknown protocol """
+ nat_ip = "10.0.0.10"
+ self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
+ self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ # in2out
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ GRE() /
+ IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg1.get_capture(1)
+ packet = p[0]
+ try:
+ 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)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # out2in
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=nat_ip) /
+ GRE() /
+ IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ 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)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
def tearDown(self):
super(TestSNAT, self).tearDown()
if not self.vpp_dead:
cls.icmp_id_out = 6305
cls.nat_addr = '10.0.0.3'
cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
+ cls.vrf1_id = 10
+ cls.vrf1_nat_addr = '10.0.10.3'
+ cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
+ cls.vrf1_nat_addr)
- cls.create_pg_interfaces(range(2))
+ cls.create_pg_interfaces(range(3))
cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
+ cls.ip6_interfaces.append(cls.pg_interfaces[2])
cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
+ cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
+
+ cls.pg0.generate_remote_hosts(2)
+
for i in cls.ip6_interfaces:
i.admin_up()
i.config_ip6()
- i.resolve_ndp()
+ i.configure_ipv6_neighbors()
for i in cls.ip4_interfaces:
i.admin_up()
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture(3)
- self.verify_capture_out(capture, packet_num=3, nat_ip=self.nat_addr,
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=self.nat_addr,
dst_ip=self.pg1.remote_ip4)
# out2in
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture(3)
+ capture = self.pg0.get_capture(len(pkts))
ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture(3)
- self.verify_capture_out(capture, packet_num=3, nat_ip=self.nat_addr,
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=self.nat_addr,
dst_ip=self.pg1.remote_ip4)
# out2in
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture(3)
- ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
+ capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
ses_num_end = self.nat64_get_ses_num()
self.assertEqual(ses_num_end - ses_num_start, 3)
+ # tenant with specific VRF
+ self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n,
+ self.vrf1_nat_addr_n,
+ vrf_id=self.vrf1_id)
+ self.vapi.nat64_add_del_interface(self.pg2.sw_if_index)
+
+ pkts = self.create_stream_in_ip6(self.pg2, self.pg1)
+ self.pg2.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr,
+ dst_ip=self.pg1.remote_ip4)
+
+ pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg2.remote_ip6)
+
def test_static(self):
""" NAT64 static translation test """
self.tcp_port_in = 60303
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture(3)
- self.verify_capture_out(capture, packet_num=3, nat_ip=self.nat_addr,
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=self.nat_addr,
dst_ip=self.pg1.remote_ip4, same_port=True)
# out2in
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture(3)
+ capture = self.pg0.get_capture(len(pkts))
ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture(3)
+ capture = self.pg1.get_capture(len(pkts))
ses_num_before_timeout = self.nat64_get_ses_num()
ses_num_after_timeout = self.nat64_get_ses_num()
self.assertNotEqual(ses_num_before_timeout, ses_num_after_timeout)
+ def test_icmp_error(self):
+ """ NAT64 ICMP Error message translation """
+ self.tcp_port_in = 6303
+ 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)
+ self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
+
+ # send some packets to create sessions
+ pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture_ip4 = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture_ip4,
+ nat_ip=self.nat_addr,
+ dst_ip=self.pg1.remote_ip4)
+
+ pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture_ip6 = self.pg0.get_capture(len(pkts))
+ ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
+ self.verify_capture_in_ip6(capture_ip6, ip[IPv6].src,
+ self.pg0.remote_ip6)
+
+ # in2out
+ pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=ip[IPv6].src) /
+ ICMPv6DestUnreach(code=1) /
+ packet[IPv6] for packet in capture_ip6]
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, self.nat_addr)
+ self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
+ self.assertEqual(packet[ICMP].type, 3)
+ self.assertEqual(packet[ICMP].code, 13)
+ inner = packet[IPerror]
+ self.assertEqual(inner.src, self.pg1.remote_ip4)
+ self.assertEqual(inner.dst, self.nat_addr)
+ self.check_icmp_checksum(packet)
+ if inner.haslayer(TCPerror):
+ self.assertEqual(inner[TCPerror].dport, self.tcp_port_out)
+ elif inner.haslayer(UDPerror):
+ self.assertEqual(inner[UDPerror].dport, self.udp_port_out)
+ else:
+ self.assertEqual(inner[ICMPerror].id, self.icmp_id_out)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # out2in
+ pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
+ ICMP(type=3, code=13) /
+ packet[IP] for packet in capture_ip4]
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IPv6].src, ip.src)
+ self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
+ icmp = packet[ICMPv6DestUnreach]
+ self.assertEqual(icmp.code, 1)
+ inner = icmp[IPerror6]
+ self.assertEqual(inner.src, self.pg0.remote_ip6)
+ self.assertEqual(inner.dst, ip.src)
+ self.check_icmpv6_checksum(packet)
+ if inner.haslayer(TCPerror):
+ self.assertEqual(inner[TCPerror].sport, self.tcp_port_in)
+ elif inner.haslayer(UDPerror):
+ self.assertEqual(inner[UDPerror].sport, self.udp_port_in)
+ else:
+ self.assertEqual(inner[ICMPv6EchoRequest].id,
+ self.icmp_id_in)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ def test_hairpinning(self):
+ """ NAT64 hairpinning """
+
+ client = self.pg0.remote_hosts[0]
+ server = self.pg0.remote_hosts[1]
+ server_tcp_in_port = 22
+ server_tcp_out_port = 4022
+ server_udp_in_port = 23
+ server_udp_out_port = 4023
+ client_tcp_in_port = 1234
+ client_udp_in_port = 1235
+ client_tcp_out_port = 0
+ client_udp_out_port = 0
+ ip = IPv6(src=''.join(['64:ff9b::', self.nat_addr]))
+ nat_addr_ip6 = ip.src
+
+ 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)
+ self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
+
+ self.vapi.nat64_add_del_static_bib(server.ip6n,
+ self.nat_addr_n,
+ server_tcp_in_port,
+ server_tcp_out_port,
+ IP_PROTOS.tcp)
+ self.vapi.nat64_add_del_static_bib(server.ip6n,
+ self.nat_addr_n,
+ server_udp_in_port,
+ server_udp_out_port,
+ IP_PROTOS.udp)
+
+ # client to server
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=client.ip6, dst=nat_addr_ip6) /
+ TCP(sport=client_tcp_in_port, dport=server_tcp_out_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=client.ip6, dst=nat_addr_ip6) /
+ UDP(sport=client_udp_in_port, dport=server_udp_out_port))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IPv6].src, nat_addr_ip6)
+ self.assertEqual(packet[IPv6].dst, server.ip6)
+ 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))
+ raise
+
+ # server to client
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=server.ip6, dst=nat_addr_ip6) /
+ TCP(sport=server_tcp_in_port, dport=client_tcp_out_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=server.ip6, dst=nat_addr_ip6) /
+ UDP(sport=server_udp_in_port, dport=client_udp_out_port))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IPv6].src, nat_addr_ip6)
+ self.assertEqual(packet[IPv6].dst, client.ip6)
+ 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
+
+ # ICMP error
+ pkts = []
+ pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=client.ip6, dst=nat_addr_ip6) /
+ ICMPv6DestUnreach(code=1) /
+ packet[IPv6] for packet in capture]
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IPv6].src, nat_addr_ip6)
+ self.assertEqual(packet[IPv6].dst, server.ip6)
+ icmp = packet[ICMPv6DestUnreach]
+ self.assertEqual(icmp.code, 1)
+ inner = icmp[IPerror6]
+ self.assertEqual(inner.src, server.ip6)
+ self.assertEqual(inner.dst, nat_addr_ip6)
+ self.check_icmpv6_checksum(packet)
+ if inner.haslayer(TCPerror):
+ self.assertEqual(inner[TCPerror].sport, server_tcp_in_port)
+ self.assertEqual(inner[TCPerror].dport,
+ client_tcp_out_port)
+ else:
+ self.assertEqual(inner[UDPerror].sport, server_udp_in_port)
+ self.assertEqual(inner[UDPerror].dport,
+ client_udp_out_port)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
def nat64_get_ses_num(self):
"""
Return number of active NAT64 sessions.
for addr in adresses:
self.vapi.nat64_add_del_pool_addr_range(addr.address,
addr.address,
+ vrf_id=addr.vrf_id,
is_add=0)
def tearDown(self):
Code Review / vpp.git / blobdiff