from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP, ICMP, TCP
-from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded, IPv6ExtHdrFragment
+from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded, IPv6ExtHdrFragment, \
+ ICMPv6EchoRequest, ICMPv6DestUnreach
class TestMAP(VppTestCase):
self.pg0.admin_up()
self.pg0.config_ip4()
self.pg0.resolve_arp()
+ self.pg0.generate_remote_hosts(2)
+ self.pg0.configure_ipv4_neighbors()
# pg1 is 'outside' IPv6
self.pg1.admin_up()
for p in rx:
self.validate(p[1], p4_translated)
- # IPv4 TTL
+ # IPv4 TTL=0
ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
p4 = (p_ether / ip4_ttl_expired / payload)
for p in rx:
self.validate(p[1], icmp4_reply)
- '''
- This one is broken, cause it would require hairpinning...
- # IPv4 TTL TTL1
+ # IPv4 TTL=1
ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
p4 = (p_ether / ip4_ttl_expired / payload)
- icmp4_reply = IP(id=0, ttl=254, src=self.pg0.local_ip4,
- dst=self.pg0.remote_ip4) / \
- ICMP(type='time-exceeded', code='ttl-zero-during-transit' ) / \
- IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0) / payload
+ icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
+ dst=self.pg0.remote_ip4) /
+ ICMP(type='time-exceeded',
+ code='ttl-zero-during-transit') /
+ IP(src=self.pg0.remote_ip4,
+ dst='192.168.0.1', ttl=1) / payload)
rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
for p in rx:
self.validate(p[1], icmp4_reply)
- '''
# IPv6 Hop limit
ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
p6 = (p_ether6 / p_ip6 / payload)
self.send_and_assert_no_replies(self.pg1, p6*1)
- # Packet fragmentation
+ # UDP packet fragmentation
payload_len = 1453
payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
p4 = (p_ether / p_ip4 / payload)
self.validate_frag_payload_len(rx, UDP, payload_len)
- # Packet fragmentation send fragments
+ # UDP packet fragmentation send fragments
payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
p4 = (p_ether / p_ip4 / payload)
frags = fragment_rfc791(p4, fragsize=1000)
self.validate_frag_payload_len(rx, UDP, payload_len)
- # reass_pkt = reassemble(rx)
- # p4_reply.ttl -= 1
- # p4_reply.id = 256
- # self.validate(reass_pkt, p4_reply)
+ # ICMP packet fragmentation
+ payload = ICMP(id=6529) / self.payload(payload_len)
+ p4 = (p_ether / p_ip4 / payload)
+ self.pg_enable_capture()
+ self.pg0.add_stream(p4)
+ self.pg_start()
+ rx = self.pg1.get_capture(2)
+
+ p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
+ dst='2001:db8:160::c0a8:1:6')
+ for p in rx:
+ self.validate_frag(p, p_ip6_translated)
+
+ self.validate_frag_payload_len(rx, ICMPv6EchoRequest, payload_len)
+
+ # ICMP packet fragmentation send fragments
+ payload = ICMP(id=6529) / self.payload(payload_len)
+ p4 = (p_ether / p_ip4 / payload)
+ frags = fragment_rfc791(p4, fragsize=1000)
+ self.pg_enable_capture()
+ self.pg0.add_stream(frags)
+ self.pg_start()
+ rx = self.pg1.get_capture(2)
+
+ for p in rx:
+ self.validate_frag(p, p_ip6_translated)
+
+ self.validate_frag_payload_len(rx, ICMPv6EchoRequest, payload_len)
# TCP MSS clamping
self.vapi.map_param_set_tcp(1300)
for p in rx:
self.validate(p[1], p4_translated)
+ # TCP MSS clamping cleanup
+ self.vapi.map_param_set_tcp(0)
+
+ # Enable icmp6 param to get back ICMPv6 unreachable messages in case
+ # of security check fails
+ self.vapi.map_param_set_icmp6(enable_unreachable=1)
+
+ # Send back an IPv6 packet that will be droppped due to security
+ # check fail
+ p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ p_ip6_sec_check_fail = IPv6(src='2001:db8:1fe::c0a8:1:f',
+ dst='1234:5678:90ab:cdef:ac:1001:200:0')
+ payload = TCP(sport=0xabcd, dport=0xabcd)
+ p6 = (p_ether6 / p_ip6_sec_check_fail / payload)
+
+ self.pg_send(self.pg1, p6*1)
+ self.pg0.get_capture(0, timeout=1)
+ rx = self.pg1.get_capture(1)
+
+ icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
+ dst='2001:db8:1fe::c0a8:1:f') /
+ ICMPv6DestUnreach(code=5) /
+ p_ip6_sec_check_fail / payload)
+
+ for p in rx:
+ self.validate(p[1], icmp6_reply)
+
+ # ICMPv6 unreachable messages cleanup
+ self.vapi.map_param_set_icmp6(enable_unreachable=0)
+
def test_map_t_ip6_psid(self):
""" MAP-T v6->v4 PSID validation"""
p6 = (p_ether6 / p_ip6 / payload)
self.send_and_assert_no_replies(self.pg1, p6*1)
+ def test_map_t_pre_resolve(self):
+ """ MAP-T pre-resolve"""
+
+ # Add a domain that maps from pg0 to pg1
+ map_dst = '2001:db8::/32'
+ map_src = '1234:5678:90ab:cdef::/64'
+ ip4_pfx = '192.168.0.0/24'
+ tag = 'MAP-T Test Domain.'
+
+ self.vapi.map_add_domain(ip6_prefix=map_dst,
+ ip4_prefix=ip4_pfx,
+ ip6_src=map_src,
+ ea_bits_len=16,
+ psid_offset=6,
+ psid_length=4,
+ mtu=1500,
+ tag=tag)
+
+ # Enable MAP-T on interfaces.
+ self.vapi.map_if_enable_disable(is_enable=1,
+ sw_if_index=self.pg0.sw_if_index,
+ is_translation=1)
+ self.vapi.map_if_enable_disable(is_enable=1,
+ sw_if_index=self.pg1.sw_if_index,
+ is_translation=1)
+
+ # Enable pre-resolve option
+ self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
+ ip6_nh_address="4001::1",
+ is_add=1)
+
+ # Add a route to 4001::1 and expect the translated traffic to be
+ # sent via that route next-hop.
+ pre_res_route6 = VppIpRoute(self, "4001::1", 128,
+ [VppRoutePath(self.pg1.remote_hosts[2].ip6,
+ self.pg1.sw_if_index)])
+ pre_res_route6.add_vpp_config()
+
+ # Add a route to 10.1.2.3 and expect the "untranslated" traffic to be
+ # sent via that route next-hop.
+ pre_res_route4 = VppIpRoute(self, "10.1.2.3", 32,
+ [VppRoutePath(self.pg0.remote_hosts[1].ip4,
+ self.pg0.sw_if_index)])
+ pre_res_route4.add_vpp_config()
+
+ # Send an IPv4 packet that will be translated
+ p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
+ payload = TCP(sport=0xabcd, dport=0xabcd)
+ p4 = (p_ether / p_ip4 / payload)
+
+ p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
+ dst="2001:db8:1f0::c0a8:1:f") / payload)
+ p6_translated.hlim -= 1
+
+ rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
+ for p in rx:
+ self.assertEqual(p[Ether].dst, self.pg1.remote_hosts[2].mac)
+ self.validate(p[1], p6_translated)
+
+ # Send back an IPv6 packet that will be "untranslated"
+ p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
+ dst='1234:5678:90ab:cdef:ac:1001:200:0')
+ p6 = (p_ether6 / p_ip6 / payload)
+
+ p4_translated = (IP(src='192.168.0.1',
+ dst=self.pg0.remote_ip4) / payload)
+ p4_translated.id = 0
+ p4_translated.ttl -= 1
+
+ rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
+ for p in rx:
+ self.assertEqual(p[Ether].dst, self.pg0.remote_hosts[1].mac)
+ self.validate(p[1], p4_translated)
+
+ # Cleanup pre-resolve option
+ self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
+ ip6_nh_address="4001::1",
+ is_add=0)
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)