import unittest
import socket
-from scapy.layers.inet import IP, ICMP, TCP
+from scapy.layers.inet import IP, ICMP, TCP, UDP
from scapy.layers.ipsec import SecurityAssociation
from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
from framework import VppTestCase, VppTestRunner
from util import ppp
+from vpp_papi import VppEnum
class IPsecIPv4Params(object):
+
addr_type = socket.AF_INET
addr_any = "0.0.0.0"
addr_bcast = "255.255.255.255"
addr_len = 32
is_ipv6 = 0
- remote_tun_if_host = '1.1.1.1'
- scapy_tun_sa_id = 10
- scapy_tun_spi = 1001
- vpp_tun_sa_id = 20
- vpp_tun_spi = 1000
+ def __init__(self):
+ self.remote_tun_if_host = '1.1.1.1'
+
+ self.scapy_tun_sa_id = 10
+ self.scapy_tun_spi = 1001
+ self.vpp_tun_sa_id = 20
+ self.vpp_tun_spi = 1000
- scapy_tra_sa_id = 30
- scapy_tra_spi = 2001
- vpp_tra_sa_id = 40
- vpp_tra_spi = 2000
+ self.scapy_tra_sa_id = 30
+ self.scapy_tra_spi = 2001
+ self.vpp_tra_sa_id = 40
+ self.vpp_tra_spi = 2000
- auth_algo_vpp_id = 2 # internal VPP enum value for SHA1_96
- auth_algo = 'HMAC-SHA1-96' # scapy name
- auth_key = 'C91KUR9GYMm5GfkEvNjX'
+ 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'
- crypt_algo_vpp_id = 1 # internal VPP enum value for AES_CBC_128
- crypt_algo = 'AES-CBC' # scapy name
- crypt_key = 'JPjyOWBeVEQiMe7h'
+ 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.flags = 0
+ self.nat_header = None
class IPsecIPv6Params(object):
+
addr_type = socket.AF_INET6
addr_any = "0::0"
addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
addr_len = 128
is_ipv6 = 1
- remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
- scapy_tun_sa_id = 50
- scapy_tun_spi = 3001
- vpp_tun_sa_id = 60
- vpp_tun_spi = 3000
+ def __init__(self):
+ self.remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
- scapy_tra_sa_id = 70
- scapy_tra_spi = 4001
- vpp_tra_sa_id = 80
- vpp_tra_spi = 4000
+ self.scapy_tun_sa_id = 50
+ self.scapy_tun_spi = 3001
+ self.vpp_tun_sa_id = 60
+ self.vpp_tun_spi = 3000
- auth_algo_vpp_id = 4 # internal VPP enum value for SHA_256_128
- auth_algo = 'SHA2-256-128' # scapy name
- auth_key = 'C91KUR9GYMm5GfkEvNjX'
+ self.scapy_tra_sa_id = 70
+ self.scapy_tra_spi = 4001
+ self.vpp_tra_sa_id = 80
+ self.vpp_tra_spi = 4000
- crypt_algo_vpp_id = 3 # internal VPP enum value for AES_CBC_256
- crypt_algo = 'AES-CBC' # scapy name
- crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h'
+ self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
+ IPSEC_API_INTEG_ALG_SHA_256_128)
+ self.auth_algo = 'SHA2-256-128' # scapy name
+ self.auth_key = 'C91KUR9GYMm5GfkEvNjX'
+
+ self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
+ IPSEC_API_CRYPTO_ALG_AES_CBC_256)
+ self.crypt_algo = 'AES-CBC' # scapy name
+ self.crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h'
+ self.flags = 0
+ self.nat_header = None
class TemplateIpsec(VppTestCase):
self.tun_spd_id = 1
self.tra_spd_id = 2
- self.vpp_esp_protocol = 1
- self.vpp_ah_protocol = 0
+ self.vpp_esp_protocol = (VppEnum.vl_api_ipsec_proto_t.
+ IPSEC_API_PROTO_ESP)
+ self.vpp_ah_protocol = (VppEnum.vl_api_ipsec_proto_t.
+ IPSEC_API_PROTO_AH)
self.create_pg_interfaces(range(3))
self.interfaces = list(self.pg_interfaces)
auth_algo=params.auth_algo, auth_key=params.auth_key,
tunnel_header=ip_class_by_addr_type[params.addr_type](
src=self.tun_if.remote_addr[params.addr_type],
- dst=self.tun_if.local_addr[params.addr_type]))
+ dst=self.tun_if.local_addr[params.addr_type]),
+ nat_t_header=params.nat_header)
vpp_tun_sa = SecurityAssociation(
self.encryption_type, spi=params.scapy_tun_spi,
crypt_algo=params.crypt_algo, crypt_key=params.crypt_key,
auth_algo=params.auth_algo, auth_key=params.auth_key,
tunnel_header=ip_class_by_addr_type[params.addr_type](
dst=self.tun_if.remote_addr[params.addr_type],
- src=self.tun_if.local_addr[params.addr_type]))
+ src=self.tun_if.local_addr[params.addr_type]),
+ nat_t_header=params.nat_header)
return vpp_tun_sa, scapy_tun_sa
def configure_sa_tra(self, params):
- params.scapy_tra_sa = SecurityAssociation(self.encryption_type,
- spi=params.vpp_tra_spi,
- crypt_algo=params.crypt_algo,
- crypt_key=params.crypt_key,
- auth_algo=params.auth_algo,
- auth_key=params.auth_key)
- params.vpp_tra_sa = SecurityAssociation(self.encryption_type,
- spi=params.scapy_tra_spi,
- crypt_algo=params.crypt_algo,
- crypt_key=params.crypt_key,
- auth_algo=params.auth_algo,
- auth_key=params.auth_key)
+ params.scapy_tra_sa = SecurityAssociation(
+ self.encryption_type,
+ spi=params.vpp_tra_spi,
+ crypt_algo=params.crypt_algo,
+ crypt_key=params.crypt_key,
+ auth_algo=params.auth_algo,
+ auth_key=params.auth_key,
+ nat_t_header=params.nat_header)
+ params.vpp_tra_sa = SecurityAssociation(
+ self.encryption_type,
+ spi=params.scapy_tra_spi,
+ crypt_algo=params.crypt_algo,
+ crypt_key=params.crypt_key,
+ auth_algo=params.auth_algo,
+ auth_key=params.auth_key,
+ nat_t_header=params.nat_header)
class IpsecTcpTests(object):
self.assert_packet_checksums_valid(decrypted)
-class IpsecTraTests(object):
+class IpsecTra4Tests(object):
def test_tra_anti_replay(self, count=1):
""" ipsec v4 transport anti-reply test """
p = self.params[socket.AF_INET]
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec"))
+ pkts = p.tra_sa_in.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA in counts: expected %d != %d" %
+ (count, pkts))
+ pkts = p.tra_sa_out.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA out counts: expected %d != %d" %
+ (count, pkts))
+
self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tra4_decrypt_node_name, count)
""" ipsec v4 transport burst test """
self.test_tra_basic(count=257)
+
+class IpsecTra6Tests(object):
def test_tra_basic6(self, count=1):
""" ipsec v6 transport basic test """
self.vapi.cli("clear errors")
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec"))
+ pkts = p.tra_sa_in.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA in counts: expected %d != %d" %
+ (count, pkts))
+ pkts = p.tra_sa_out.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA out counts: expected %d != %d" %
+ (count, pkts))
self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count)
self.test_tra_basic6(count=257)
+class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
+ pass
+
+
class IpsecTun4Tests(object):
def test_tun_basic44(self, count=1):
""" ipsec 4o4 tunnel basic test """
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec"))
+ if (hasattr(p, "spd_policy_in_any")):
+ pkts = p.spd_policy_in_any.get_stats()['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']
+ self.assertEqual(pkts, count,
+ "incorrect SA in counts: expected %d != %d" %
+ (count, pkts))
+ pkts = p.tun_sa_out.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA out counts: expected %d != %d" %
+ (count, pkts))
+
self.assert_packet_counter_equal(self.tun4_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tun4_decrypt_node_name, count)
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec"))
+ if (hasattr(p, "tun_sa_in")):
+ pkts = p.tun_sa_in.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA in counts: expected %d != %d" %
+ (count, pkts))
+ pkts = p.tun_sa_out.get_stats()['packets']
+ self.assertEqual(pkts, count,
+ "incorrect SA out counts: expected %d != %d" %
+ (count, pkts))
self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tun6_decrypt_node_name, count)
self.test_tun_basic66(count=257)
-class IpsecTunTests(IpsecTun4Tests, IpsecTun6Tests):
+class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
pass
Code Review / vpp.git / blobdiff