+ def test_tun_44_police(self):
+ """IPSEC interface IPv4 with input policer"""
+ n_pkts = 127
+ p = self.ipv4_params
+
+ self.config_network(p)
+ self.config_sa_tun(p,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4)
+ self.config_protect(p)
+
+ action_tx = PolicerAction(
+ VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
+ 0)
+ policer = VppPolicer(self, "pol1", 80, 0, 1000, 0,
+ conform_action=action_tx,
+ exceed_action=action_tx,
+ violate_action=action_tx)
+ policer.add_vpp_config()
+
+ # Start policing on tun
+ policer.apply_vpp_config(p.tun_if.sw_if_index, True)
+
+ self.verify_tun_44(p, count=n_pkts)
+ c = p.tun_if.get_rx_stats()
+ self.assertEqual(c['packets'], n_pkts)
+ c = p.tun_if.get_tx_stats()
+ self.assertEqual(c['packets'], n_pkts)
+
+ stats = policer.get_stats()
+
+ # Single rate, 2 colour policer - expect conform, violate but no exceed
+ self.assertGreater(stats['conform_packets'], 0)
+ self.assertEqual(stats['exceed_packets'], 0)
+ self.assertGreater(stats['violate_packets'], 0)
+
+ # Stop policing on tun
+ policer.apply_vpp_config(p.tun_if.sw_if_index, False)
+ self.verify_tun_44(p, count=n_pkts)
+
+ # No new policer stats
+ statsnew = policer.get_stats()
+ self.assertEqual(stats, statsnew)
+
+ # teardown
+ policer.remove_vpp_config()
+ self.unconfig_protect(p)
+ self.unconfig_sa(p)
+ self.unconfig_network(p)
+
+
+class TestIpsecItf4MPLS(TemplateIpsec,
+ TemplateIpsecItf4,
+ IpsecTun4):
+ """ IPsec Interface MPLSoIPv4 """
+
+ tun4_encrypt_node_name = "esp-mpls-encrypt-tun"
+
+ def setUp(self):
+ super(TestIpsecItf4MPLS, self).setUp()
+
+ self.tun_if = self.pg0
+
+ def tearDown(self):
+ super(TestIpsecItf4MPLS, self).tearDown()
+
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
+ payload_size=100):
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ sa.encrypt(MPLS(label=44, ttl=3) /
+ IP(src=src, dst=dst) /
+ UDP(sport=1166, dport=2233) /
+ Raw(b'X' * payload_size))
+ for i in range(count)]
+
+ def verify_encrypted(self, p, sa, rxs):
+ for rx in rxs:
+ try:
+ pkt = sa.decrypt(rx[IP])
+ if not pkt.haslayer(IP):
+ pkt = IP(pkt[Raw].load)
+ self.assert_packet_checksums_valid(pkt)
+ self.assert_equal(pkt[MPLS].label, 44)
+ self.assert_equal(pkt[IP].dst, p.remote_tun_if_host)
+ except (IndexError, AssertionError):
+ self.logger.debug(ppp("Unexpected packet:", rx))
+ try:
+ self.logger.debug(ppp("Decrypted packet:", pkt))
+ except:
+ pass
+ raise
+
+ def test_tun_mpls_o_ip4(self):
+ """IPSEC interface MPLS over IPv4"""
+
+ n_pkts = 127
+ p = self.ipv4_params
+ f = FibPathProto
+
+ tbl = VppMplsTable(self, 0)
+ tbl.add_vpp_config()
+
+ self.config_network(p)
+ # deag MPLS routes from the tunnel
+ r4 = VppMplsRoute(self, 44, 1,
+ [VppRoutePath(
+ self.pg1.remote_ip4,
+ self.pg1.sw_if_index)]).add_vpp_config()
+ p.route.modify([VppRoutePath(p.tun_if.remote_ip4,
+ p.tun_if.sw_if_index,
+ labels=[VppMplsLabel(44)])])
+ p.tun_if.enable_mpls()
+
+ self.config_sa_tun(p,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4)
+ self.config_protect(p)
+
+ self.verify_tun_44(p, count=n_pkts)
+
+ # cleanup
+ p.tun_if.disable_mpls()
+ self.unconfig_protect(p)
+ self.unconfig_sa(p)
+ self.unconfig_network(p)
+
Code Review / vpp.git / blobdiff