from scapy.layers.inet import IP, ICMP, TCP, UDP
from scapy.layers.ipsec import SecurityAssociation, ESP
-from scapy.layers.l2 import Ether, Raw
-from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
+from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, IPv6ExtHdrHopByHop, \
+ IPv6ExtHdrFragment, IPv6ExtHdrDestOpt
+
from framework import VppTestCase, VppTestRunner
from util import ppp, reassemble4, fragment_rfc791, fragment_rfc8200
self.remote_tun_if_host = '1.1.1.1'
self.remote_tun_if_host6 = '1111::1'
- self.scapy_tun_sa_id = 10
+ self.scapy_tun_sa_id = 100
self.scapy_tun_spi = 1001
- self.vpp_tun_sa_id = 20
+ self.vpp_tun_sa_id = 200
self.vpp_tun_spi = 1000
- self.scapy_tra_sa_id = 30
+ self.scapy_tra_sa_id = 300
self.scapy_tra_spi = 2001
- self.vpp_tra_sa_id = 40
+ self.vpp_tra_sa_id = 400
self.vpp_tra_spi = 2000
self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
IPSEC_API_INTEG_ALG_SHA1_96)
self.auth_algo = 'HMAC-SHA1-96' # scapy name
- self.auth_key = 'C91KUR9GYMm5GfkEvNjX'
+ self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
IPSEC_API_CRYPTO_ALG_AES_CBC_128)
self.crypt_algo = 'AES-CBC' # scapy name
- self.crypt_key = 'JPjyOWBeVEQiMe7h'
+ self.crypt_key = b'JPjyOWBeVEQiMe7h'
self.salt = 0
self.flags = 0
self.nat_header = None
self.remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
self.remote_tun_if_host4 = '1.1.1.1'
- self.scapy_tun_sa_id = 50
+ self.scapy_tun_sa_id = 500
self.scapy_tun_spi = 3001
- self.vpp_tun_sa_id = 60
+ self.vpp_tun_sa_id = 600
self.vpp_tun_spi = 3000
- self.scapy_tra_sa_id = 70
+ self.scapy_tra_sa_id = 700
self.scapy_tra_spi = 4001
- self.vpp_tra_sa_id = 80
+ self.vpp_tra_sa_id = 800
self.vpp_tra_spi = 4000
self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
IPSEC_API_INTEG_ALG_SHA1_96)
self.auth_algo = 'HMAC-SHA1-96' # scapy name
- self.auth_key = 'C91KUR9GYMm5GfkEvNjX'
+ self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
IPSEC_API_CRYPTO_ALG_AES_CBC_128)
self.crypt_algo = 'AES-CBC' # scapy name
- self.crypt_key = 'JPjyOWBeVEQiMe7h'
+ self.crypt_key = b'JPjyOWBeVEQiMe7h'
self.salt = 0
self.flags = 0
self.nat_header = None
-def mk_scapy_crpyt_key(p):
+def mk_scapy_crypt_key(p):
if p.crypt_algo == "AES-GCM":
return p.crypt_key + struct.pack("!I", p.salt)
else:
def config_tun_params(p, encryption_type, tun_if):
ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
- use_esn = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_USE_ESN))
- crypt_key = mk_scapy_crpyt_key(p)
+ esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
+ IPSEC_API_SAD_FLAG_USE_ESN))
+ p.tun_dst = tun_if.remote_addr[p.addr_type]
+ p.tun_src = tun_if.local_addr[p.addr_type]
+ crypt_key = mk_scapy_crypt_key(p)
p.scapy_tun_sa = SecurityAssociation(
encryption_type, spi=p.vpp_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo, auth_key=p.auth_key,
tunnel_header=ip_class_by_addr_type[p.addr_type](
- src=tun_if.remote_addr[p.addr_type],
- dst=tun_if.local_addr[p.addr_type]),
+ src=p.tun_dst,
+ dst=p.tun_src),
nat_t_header=p.nat_header,
- use_esn=use_esn)
+ esn_en=esn_en)
p.vpp_tun_sa = SecurityAssociation(
encryption_type, spi=p.scapy_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo, auth_key=p.auth_key,
tunnel_header=ip_class_by_addr_type[p.addr_type](
- dst=tun_if.remote_addr[p.addr_type],
- src=tun_if.local_addr[p.addr_type]),
+ dst=p.tun_dst,
+ src=p.tun_src),
nat_t_header=p.nat_header,
- use_esn=use_esn)
+ esn_en=esn_en)
def config_tra_params(p, encryption_type):
- use_esn = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_USE_ESN))
- crypt_key = mk_scapy_crpyt_key(p)
+ esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
+ IPSEC_API_SAD_FLAG_USE_ESN))
+ crypt_key = mk_scapy_crypt_key(p)
p.scapy_tra_sa = SecurityAssociation(
encryption_type,
spi=p.vpp_tra_spi,
auth_algo=p.auth_algo,
auth_key=p.auth_key,
nat_t_header=p.nat_header,
- use_esn=use_esn)
+ esn_en=esn_en)
p.vpp_tra_sa = SecurityAssociation(
encryption_type,
spi=p.scapy_tra_spi,
auth_algo=p.auth_algo,
auth_key=p.auth_key,
nat_t_header=p.nat_header,
- use_esn=use_esn)
+ esn_en=esn_en)
class TemplateIpsec(VppTestCase):
def show_commands_at_teardown(self):
self.logger.info(self.vapi.cli("show hardware"))
- def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
payload_size=54):
return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
sa.encrypt(IP(src=src, dst=dst) /
- ICMP() / Raw('X' * payload_size))
+ ICMP() / Raw(b'X' * payload_size))
for i in range(count)]
- def gen_encrypt_pkts6(self, sa, sw_intf, src, dst, count=1,
+ def gen_encrypt_pkts6(self, p, 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) /
def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IP(src=src, dst=dst) / ICMP() / Raw('X' * payload_size)
+ IP(src=src, dst=dst) / ICMP() / Raw(b'X' * payload_size)
for i in range(count)]
def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
def verify_tcp_checksum(self):
self.vapi.cli("test http server")
p = self.params[socket.AF_INET]
- config_tun_params(p, self.encryption_type, self.tun_if)
send = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
p.scapy_tun_sa.encrypt(IP(src=p.remote_tun_if_host,
dst=self.tun_if.local_ip4) /
""" verify methods for Transport v4 """
def verify_tra_anti_replay(self):
p = self.params[socket.AF_INET]
- use_esn = p.vpp_tra_sa.use_esn
+ esn_en = p.vpp_tra_sa.esn_en
seq_cycle_node_name = ('/err/%s/sequence number cycled' %
self.tra4_encrypt_node_name)
bogus_sa = SecurityAssociation(self.encryption_type,
p.vpp_tra_spi,
crypt_algo=p.crypt_algo,
- crypt_key=mk_scapy_crpyt_key(p)[::-1],
+ crypt_key=mk_scapy_crypt_key(p)[::-1],
auth_algo=p.auth_algo,
auth_key=p.auth_key[::-1])
pkt = (Ether(src=self.tra_if.remote_mac,
seq_num=17))
self.send_and_assert_no_replies(self.tra_if, pkt * 17)
- if use_esn:
+ if esn_en:
# an out of window error with ESN looks like a high sequence
# wrap. but since it isn't then the verify will fail.
hash_failed_count += 17
seq_num=seq))
for seq in range(259, 280)]
- if use_esn:
+ if esn_en:
rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
#
p.scapy_tra_sa.seq_num = 351
p.vpp_tra_sa.seq_num = 351
- def verify_tra_basic4(self, count=1):
+ def verify_tra_basic4(self, count=1, payload_size=54):
""" ipsec v4 transport basic test """
self.vapi.cli("clear errors")
self.vapi.cli("clear ipsec sa")
try:
p = self.params[socket.AF_INET]
- send_pkts = self.gen_encrypt_pkts(p.scapy_tra_sa, self.tra_if,
+ send_pkts = self.gen_encrypt_pkts(p, p.scapy_tra_sa, self.tra_if,
src=self.tra_if.remote_ip4,
dst=self.tra_if.local_ip4,
- count=count)
+ count=count,
+ payload_size=payload_size)
recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
self.tra_if)
for rx in recv_pkts:
class IpsecTra4Tests(IpsecTra4):
""" UT test methods for Transport v4 """
def test_tra_anti_replay(self):
- """ ipsec v4 transport anti-reply test """
+ """ ipsec v4 transport anti-replay test """
self.verify_tra_anti_replay()
def test_tra_basic(self, count=1):
class IpsecTra6(object):
""" verify methods for Transport v6 """
- def verify_tra_basic6(self, count=1):
+ def verify_tra_basic6(self, count=1, payload_size=54):
self.vapi.cli("clear errors")
+ self.vapi.cli("clear ipsec sa")
try:
p = self.params[socket.AF_INET6]
- send_pkts = self.gen_encrypt_pkts6(p.scapy_tra_sa, self.tra_if,
+ send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tra_sa, self.tra_if,
src=self.tra_if.remote_ip6,
dst=self.tra_if.local_ip6,
- count=count)
+ count=count,
+ payload_size=payload_size)
recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
self.tra_if)
for rx in recv_pkts:
self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count)
+ 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 """
self.verify_tra_basic6(count=257)
+class IpsecTra6ExtTests(IpsecTra6):
+ def test_tra_ext_hdrs_66(self):
+ """ ipsec 6o6 tra extension headers test """
+ self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
+
+
class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
""" UT test methods for Transport v6 and v4"""
pass
class IpsecTun4(object):
""" verify methods for Tunnel v4 """
- def verify_counters4(self, p, count, n_frags=None):
+ def verify_counters4(self, p, count, n_frags=None, worker=None):
if not n_frags:
n_frags = count
if (hasattr(p, "spd_policy_in_any")):
- pkts = p.spd_policy_in_any.get_stats()['packets']
+ pkts = p.spd_policy_in_any.get_stats(worker)['packets']
self.assertEqual(pkts, count,
"incorrect SPD any policy: expected %d != %d" %
(count, pkts))
if (hasattr(p, "tun_sa_in")):
- pkts = p.tun_sa_in.get_stats()['packets']
+ pkts = p.tun_sa_in.get_stats(worker)['packets']
self.assertEqual(pkts, count,
"incorrect SA in counts: expected %d != %d" %
(count, pkts))
- pkts = p.tun_sa_out.get_stats()['packets']
+ pkts = p.tun_sa_out.get_stats(worker)['packets']
self.assertEqual(pkts, count,
"incorrect SA out counts: expected %d != %d" %
(count, pkts))
def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
self.vapi.cli("clear errors")
+ self.vapi.cli("clear ipsec counters")
+ self.vapi.cli("clear ipsec sa")
if not n_rx:
n_rx = count
try:
- config_tun_params(p, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts(p.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
src=p.remote_tun_if_host,
dst=self.pg1.remote_ip4,
- count=count)
+ count=count,
+ payload_size=payload_size)
recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
self.verify_decrypted(p, recv_pkts)
self.tun_if, n_rx)
self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
+ for rx in recv_pkts:
+ self.assertEqual(rx[IP].src, p.tun_src)
+ self.assertEqual(rx[IP].dst, p.tun_dst)
+
finally:
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec all"))
+ self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
+ self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
self.verify_counters4(p, count, n_rx)
+ def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
+ self.vapi.cli("clear errors")
+ if not n_rx:
+ n_rx = count
+ try:
+ send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip4,
+ count=count)
+ self.send_and_assert_no_replies(self.tun_if, send_pkts)
+
+ send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
+ dst=p.remote_tun_if_host, count=count,
+ payload_size=payload_size)
+ self.send_and_assert_no_replies(self.pg1, send_pkts)
+
+ finally:
+ self.logger.info(self.vapi.ppcli("show error"))
+ self.logger.info(self.vapi.ppcli("show ipsec all"))
+
def verify_tun_reass_44(self, p):
self.vapi.cli("clear errors")
self.vapi.ip_reassembly_enable_disable(
sw_if_index=self.tun_if.sw_if_index, enable_ip4=True)
try:
- config_tun_params(p, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts(p.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
src=p.remote_tun_if_host,
dst=self.pg1.remote_ip4,
payload_size=1900,
def verify_tun_64(self, p, count=1):
self.vapi.cli("clear errors")
try:
- config_tun_params(p, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts6(p.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
src=p.remote_tun_if_host6,
dst=self.pg1.remote_ip6,
count=count)
pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
UDP(sport=333, dport=4500) /
- Raw(0xff))
+ Raw(b'\xff'))
self.send_and_assert_no_replies(self.tun_if, pkt*31)
self.assert_error_counter_equal(
'/err/%s/NAT Keepalive' % self.tun4_input_node, 31)
pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
UDP(sport=333, dport=4500) /
- Raw(0xfe))
+ Raw(b'\xfe'))
self.send_and_assert_no_replies(self.tun_if, pkt*31)
self.assert_error_counter_equal(
'/err/%s/Too Short' % self.tun4_input_node, 31)
def test_tun_basic44(self):
""" ipsec 4o4 tunnel basic test """
self.verify_tun_44(self.params[socket.AF_INET], count=1)
+ self.tun_if.admin_down()
+ self.tun_if.resolve_arp()
+ self.tun_if.admin_up()
+ self.verify_tun_44(self.params[socket.AF_INET], count=1)
def test_tun_reass_basic44(self):
""" ipsec 4o4 tunnel basic reassembly test """
def test_tun_burst44(self):
""" ipsec 4o4 tunnel burst test """
- self.verify_tun_44(self.params[socket.AF_INET], count=257)
+ self.verify_tun_44(self.params[socket.AF_INET], count=127)
class IpsecTun6(object):
""" verify methods for Tunnel v6 """
- def verify_counters6(self, p_in, p_out, count):
+ def verify_counters6(self, p_in, p_out, count, worker=None):
if (hasattr(p_in, "tun_sa_in")):
- pkts = p_in.tun_sa_in.get_stats()['packets']
+ pkts = p_in.tun_sa_in.get_stats(worker)['packets']
self.assertEqual(pkts, count,
"incorrect SA in counts: expected %d != %d" %
(count, pkts))
if (hasattr(p_out, "tun_sa_out")):
- pkts = p_out.tun_sa_out.get_stats()['packets']
+ pkts = p_out.tun_sa_out.get_stats(worker)['packets']
self.assertEqual(pkts, count,
"incorrect SA out counts: expected %d != %d" %
(count, pkts))
self.vapi.cli("clear errors")
self.vapi.cli("clear ipsec sa")
- config_tun_params(p_in, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts6(p_in.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
+ self.tun_if,
src=p_in.remote_tun_if_host,
dst=self.pg1.remote_ip6,
count=count)
if not p_out:
p_out = p_in
try:
- config_tun_params(p_in, self.encryption_type, self.tun_if)
- config_tun_params(p_out, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts6(p_in.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
+ self.tun_if,
src=p_in.remote_tun_if_host,
dst=self.pg1.remote_ip6,
- count=count)
+ count=count,
+ payload_size=payload_size)
recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
self.verify_decrypted6(p_in, recv_pkts)
dst=p_out.remote_tun_if_host,
count=count,
payload_size=payload_size)
- recv_pkts = self.send_and_expect(self.pg1, send_pkts,
- self.tun_if)
+ recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
+ for rx in recv_pkts:
+ self.assertEqual(rx[IPv6].src, p_out.tun_src)
+ self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
+
finally:
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec all"))
sw_if_index=self.tun_if.sw_if_index, enable_ip6=True)
try:
- config_tun_params(p, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts6(p.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
src=p.remote_tun_if_host,
dst=self.pg1.remote_ip6,
count=1,
- payload_size=1900)
+ payload_size=1850)
send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
self.pg1, n_rx=1)
""" ipsec 4o6 tunnel basic test """
self.vapi.cli("clear errors")
try:
- config_tun_params(p, self.encryption_type, self.tun_if)
- send_pkts = self.gen_encrypt_pkts(p.scapy_tun_sa, self.tun_if,
+ send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
src=p.remote_tun_if_host4,
dst=self.pg1.remote_ip4,
count=count)
self.verify_tun_66(self.params[socket.AF_INET6], count=257)
+class IpsecTun6HandoffTests(IpsecTun6):
+ """ UT test methods for Tunnel v6 with multiple workers """
+ worker_config = "workers 2"
+
+ def test_tun_handoff_66(self):
+ """ ipsec 6o6 tunnel worker hand-off test """
+ N_PKTS = 15
+ p = self.params[socket.AF_INET6]
+
+ # inject alternately on worker 0 and 1. all counts on the SA
+ # should be against worker 0
+ for worker in [0, 1, 0, 1]:
+ send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip6,
+ count=N_PKTS)
+ recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
+ self.pg1, worker=worker)
+ self.verify_decrypted6(p, recv_pkts)
+
+ send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
+ dst=p.remote_tun_if_host,
+ count=N_PKTS)
+ recv_pkts = self.send_and_expect(self.pg1, send_pkts,
+ self.tun_if, worker=worker)
+ self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
+
+ # all counts against the first worker that was used
+ self.verify_counters6(p, p, 4*N_PKTS, worker=0)
+
+
+class IpsecTun4HandoffTests(IpsecTun4):
+ """ UT test methods for Tunnel v4 with multiple workers """
+ worker_config = "workers 2"
+
+ def test_tun_handooff_44(self):
+ """ ipsec 4o4 tunnel worker hand-off test """
+ N_PKTS = 15
+ p = self.params[socket.AF_INET]
+
+ # inject alternately on worker 0 and 1. all counts on the SA
+ # should be against worker 0
+ for worker in [0, 1, 0, 1]:
+ send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip4,
+ count=N_PKTS)
+ recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
+ self.pg1, worker=worker)
+ self.verify_decrypted(p, recv_pkts)
+
+ send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
+ dst=p.remote_tun_if_host,
+ count=N_PKTS)
+ recv_pkts = self.send_and_expect(self.pg1, send_pkts,
+ self.tun_if, worker=worker)
+ self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
+
+ # all counts against the first worker that was used
+ self.verify_counters4(p, 4*N_PKTS, worker=0)
+
+
class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
""" UT test methods for Tunnel v6 & v4 """
pass
Code Review / vpp.git / blobdiff