+ def gen_encrypt_pkts_ext_hdrs6(self, sa, sw_intf, src, dst, count=1,
+ payload_size=54):
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ sa.encrypt(IPv6(src=src, dst=dst) /
+ ICMPv6EchoRequest(id=0, seq=1,
+ data='X' * payload_size))
+ for i in range(count)]
+
+ def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IPv6(src=src, dst=dst) /
+ IPv6ExtHdrHopByHop() /
+ IPv6ExtHdrFragment(id=2, offset=200) /
+ Raw(b'\xff' * 200)
+ for i in range(count)]
+
+ def verify_tra_encrypted6(self, p, sa, rxs):
+ decrypted = []
+ for rx in rxs:
+ self.assert_packet_checksums_valid(rx)
+ try:
+ decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
+ decrypted.append(decrypt_pkt)
+ self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
+ self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
+ except:
+ self.logger.debug(ppp("Unexpected packet:", rx))
+ try:
+ self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
+ except:
+ pass
+ raise
+ return decrypted
+
+ def verify_tra_66_ext_hdrs(self, p):
+ count = 63
+
+ #
+ # check we can decrypt with options
+ #
+ tx = self.gen_encrypt_pkts_ext_hdrs6(p.scapy_tra_sa, self.tra_if,
+ src=self.tra_if.remote_ip6,
+ dst=self.tra_if.local_ip6,
+ count=count)
+ self.send_and_expect(self.tra_if, tx, self.tra_if)
+
+ #
+ # injecting a packet from ourselves to be routed of box is a hack
+ # but it matches an outbout policy, alors je ne regrette rien
+ #
+
+ # one extension before ESP
+ tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
+ IPv6(src=self.tra_if.local_ip6,
+ dst=self.tra_if.remote_ip6) /
+ IPv6ExtHdrFragment(id=2, offset=200) /
+ Raw(b'\xff' * 200))
+
+ rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
+ dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
+
+ for dc in dcs:
+ # for reasons i'm not going to investigate scapy does not
+ # created the correct headers after decrypt. but reparsing
+ # the ipv6 packet fixes it
+ dc = IPv6(raw(dc[IPv6]))
+ self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
+
+ # two extensions before ESP
+ tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
+ IPv6(src=self.tra_if.local_ip6,
+ dst=self.tra_if.remote_ip6) /
+ IPv6ExtHdrHopByHop() /
+ IPv6ExtHdrFragment(id=2, offset=200) /
+ Raw(b'\xff' * 200))
+
+ rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
+ dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
+
+ for dc in dcs:
+ dc = IPv6(raw(dc[IPv6]))
+ self.assertTrue(dc[IPv6ExtHdrHopByHop])
+ self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
+
+ # two extensions before ESP, one after
+ tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
+ IPv6(src=self.tra_if.local_ip6,
+ dst=self.tra_if.remote_ip6) /
+ IPv6ExtHdrHopByHop() /
+ IPv6ExtHdrFragment(id=2, offset=200) /
+ IPv6ExtHdrDestOpt() /
+ Raw(b'\xff' * 200))
+
+ rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
+ dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
+
+ for dc in dcs:
+ dc = IPv6(raw(dc[IPv6]))
+ self.assertTrue(dc[IPv6ExtHdrDestOpt])
+ self.assertTrue(dc[IPv6ExtHdrHopByHop])
+ self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
+
+
+class IpsecTra6Tests(IpsecTra6):
+ """ UT test methods for Transport v6 """
+ def test_tra_basic6(self):
+ """ ipsec v6 transport basic test """
+ self.verify_tra_basic6(count=1)
+