class IPsecIPv4Params:
-
addr_type = socket.AF_INET
addr_any = "0.0.0.0"
addr_bcast = "255.255.255.255"
class IPsecIPv6Params:
-
addr_type = socket.AF_INET6
addr_any = "0::0"
addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
def mk_scapy_crypt_key(p):
- if p.crypt_algo in ("AES-GCM", "AES-CTR"):
+ if p.crypt_algo in ("AES-GCM", "AES-CTR", "AES-NULL-GMAC"):
return p.crypt_key + struct.pack("!I", p.salt)
else:
return p.crypt_key
crypt_key = mk_scapy_crypt_key(p)
p.scapy_tun_sa = SecurityAssociation(
encryption_type,
- spi=p.vpp_tun_spi,
+ spi=p.scapy_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo,
)
p.vpp_tun_sa = SecurityAssociation(
encryption_type,
- spi=p.scapy_tun_spi,
+ spi=p.vpp_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo,
crypt_key = mk_scapy_crypt_key(p)
p.scapy_tra_sa = SecurityAssociation(
encryption_type,
- spi=p.vpp_tra_spi,
+ spi=p.scapy_tra_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo,
)
p.vpp_tra_sa = SecurityAssociation(
encryption_type,
- spi=p.scapy_tra_spi,
+ spi=p.vpp_tra_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo,
"""verify methods for Transport v4"""
def get_replay_counts(self, p):
- replay_node_name = "/err/%s/SA replayed packet" % self.tra4_decrypt_node_name[0]
+ replay_node_name = "/err/%s/replay" % self.tra4_decrypt_node_name[0]
count = self.statistics.get_err_counter(replay_node_name)
if p.async_mode:
replay_post_node_name = (
- "/err/%s/SA replayed packet" % self.tra4_decrypt_node_name[p.async_mode]
+ "/err/%s/replay" % self.tra4_decrypt_node_name[p.async_mode]
)
count += self.statistics.get_err_counter(replay_post_node_name)
return count
def get_hash_failed_counts(self, p):
- if ESP == self.encryption_type and p.crypt_algo == "AES-GCM":
+ if ESP == self.encryption_type and p.crypt_algo in ("AES-GCM", "AES-NULL-GMAC"):
hash_failed_node_name = (
- "/err/%s/ESP decryption failed"
- % self.tra4_decrypt_node_name[p.async_mode]
+ "/err/%s/decryption_failed" % self.tra4_decrypt_node_name[p.async_mode]
)
else:
hash_failed_node_name = (
- "/err/%s/Integrity check failed"
- % self.tra4_decrypt_node_name[p.async_mode]
+ "/err/%s/integ_error" % self.tra4_decrypt_node_name[p.async_mode]
)
count = self.statistics.get_err_counter(hash_failed_node_name)
esn_on = p.vpp_tra_sa.esn_en
ar_on = p.flags & saf.IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY
- seq_cycle_node_name = (
- "/err/%s/sequence number cycled (packet dropped)"
- % self.tra4_encrypt_node_name
- )
+ seq_cycle_node_name = "/err/%s/seq_cycled" % self.tra4_encrypt_node_name
replay_count = self.get_replay_counts(p)
hash_failed_count = self.get_hash_failed_counts(p)
seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
]
recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
+ # a replayed packet, then an out of window, then a legit
+ # tests that a early failure on the batch doesn't affect subsequent packets.
+ pkts = [
+ (
+ Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
+ / p.scapy_tra_sa.encrypt(
+ IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
+ seq_num=203,
+ )
+ ),
+ (
+ Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
+ / p.scapy_tra_sa.encrypt(
+ IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
+ seq_num=81,
+ )
+ ),
+ (
+ Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
+ / p.scapy_tra_sa.encrypt(
+ IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
+ seq_num=204,
+ )
+ ),
+ ]
+ n_rx = 1 if ar_on else 3
+ recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
+
# move the window over half way to a wrap
pkts = [
(
p = self.params[socket.AF_INET]
esn_en = p.vpp_tra_sa.esn_en
- seq_cycle_node_name = (
- "/err/%s/sequence number cycled (packet dropped)"
- % self.tra4_encrypt_node_name
- )
+ seq_cycle_node_name = "/err/%s/seq_cycled" % self.tra4_encrypt_node_name
replay_count = self.get_replay_counts(p)
hash_failed_count = self.get_hash_failed_counts(p)
seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
+ hash_err = "integ_error"
if ESP == self.encryption_type:
- undersize_node_name = (
- "/err/%s/undersized packet" % self.tra4_decrypt_node_name[0]
- )
+ undersize_node_name = "/err/%s/runt" % self.tra4_decrypt_node_name[0]
undersize_count = self.statistics.get_err_counter(undersize_node_name)
+ # For AES-GCM an error in the hash is reported as a decryption failure
+ if p.crypt_algo in ("AES-GCM", "AES-NULL-GMAC"):
+ hash_err = "decryption_failed"
+ # In async mode, we don't report errors in the hash.
+ if p.async_mode:
+ hash_err = ""
#
# send packets with seq numbers 1->34
self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
replay_count += len(pkts)
self.assertEqual(self.get_replay_counts(p), replay_count)
+ err = p.tra_sa_in.get_err("replay")
+ self.assertEqual(err, replay_count)
#
# now send a batch of packets all with the same sequence number
recv_pkts = self.send_and_expect(self.tra_if, pkts * 8, self.tra_if, n_rx=1)
replay_count += 7
self.assertEqual(self.get_replay_counts(p), replay_count)
+ err = p.tra_sa_in.get_err("replay")
+ self.assertEqual(err, replay_count)
#
# now move the window over to 257 (more than one byte) and into Case A
self.send_and_assert_no_replies(self.tra_if, pkt * 3, timeout=0.2)
replay_count += 3
self.assertEqual(self.get_replay_counts(p), replay_count)
+ err = p.tra_sa_in.get_err("replay")
+ self.assertEqual(err, replay_count)
# the window size is 64 packets
# in window are still accepted
# a packet that does not decrypt does not move the window forward
bogus_sa = SecurityAssociation(
self.encryption_type,
- p.vpp_tra_spi,
+ p.scapy_tra_spi,
crypt_algo=p.crypt_algo,
crypt_key=mk_scapy_crypt_key(p)[::-1],
auth_algo=p.auth_algo,
hash_failed_count += 17
self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
+ if hash_err != "":
+ err = p.tra_sa_in.get_err(hash_err)
+ self.assertEqual(err, hash_failed_count)
# a malformed 'runt' packet
# created by a mis-constructed SA
if ESP == self.encryption_type and p.crypt_algo != "NULL":
- bogus_sa = SecurityAssociation(self.encryption_type, p.vpp_tra_spi)
+ bogus_sa = SecurityAssociation(self.encryption_type, p.scapy_tra_spi)
pkt = Ether(
src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
) / bogus_sa.encrypt(
undersize_count += 17
self.assert_error_counter_equal(undersize_node_name, undersize_count)
+ err = p.tra_sa_in.get_err("runt")
+ self.assertEqual(err, undersize_count)
# which we can determine since this packet is still in the window
pkt = Ether(
# wrap. but since it isn't then the verify will fail.
hash_failed_count += 17
self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
+ if hash_err != "":
+ err = p.tra_sa_in.get_err(hash_err)
+ self.assertEqual(err, hash_failed_count)
else:
replay_count += 17
self.assertEqual(self.get_replay_counts(p), replay_count)
+ err = p.tra_sa_in.get_err("replay")
+ self.assertEqual(err, replay_count)
# valid packet moves the window over to 258
pkt = Ether(
# causes the TX seq number to wrap; unless we're using extened sequence
# numbers.
#
- self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.scapy_tra_sa_id)
+ self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.vpp_tra_sa_id)
self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
hash_failed_count += 1
self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
+ if hash_err != "":
+ err = p.tra_sa_in.get_err(hash_err)
+ self.assertEqual(err, hash_failed_count)
#
# but if we move the window forward to case B, then we can wrap
self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
seq_cycle_count += len(pkts)
self.assert_error_counter_equal(seq_cycle_node_name, seq_cycle_count)
+ err = p.tra_sa_out.get_err("seq_cycled")
+ self.assertEqual(err, seq_cycle_count)
# move the security-associations seq number on to the last we used
self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
]
self.send_and_expect(self.tra_if, pkts, self.tra_if)
- self.assertEqual(p.tra_sa_out.get_lost(), 0)
+ self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
# skip a sequence number
pkts = [
]
self.send_and_expect(self.tra_if, pkts, self.tra_if)
- self.assertEqual(p.tra_sa_out.get_lost(), 0)
+ self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
# the lost packet are counted untill we get up past the first
# sizeof(replay_window) packets
]
self.send_and_expect(self.tra_if, pkts, self.tra_if)
- self.assertEqual(p.tra_sa_out.get_lost(), 1)
+ self.assertEqual(p.tra_sa_in.get_err("lost"), 1)
# lost of holes in the sequence
pkts = [
]
self.send_and_expect(self.tra_if, pkts, self.tra_if)
- self.assertEqual(p.tra_sa_out.get_lost(), 51)
+ self.assertEqual(p.tra_sa_in.get_err("lost"), 51)
# a big hole in the seq number space
pkts = [
]
self.send_and_expect(self.tra_if, pkts, self.tra_if)
- self.assertEqual(p.tra_sa_out.get_lost(), 151)
+ self.assertEqual(p.tra_sa_in.get_err("lost"), 151)
def verify_tra_basic4(self, count=1, payload_size=54):
"""ipsec v4 transport basic test"""
self.assertEqual(
pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
)
- self.assertEqual(p.tra_sa_out.get_lost(), 0)
- self.assertEqual(p.tra_sa_in.get_lost(), 0)
+ self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
+ self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tra4_decrypt_node_name[0], count)
decrypt_pkts = []
for rx in rxs:
if p.nat_header:
- self.assertEqual(rx[UDP].dport, 4500)
+ self.assertEqual(rx[UDP].dport, p.nat_header.dport)
self.assert_packet_checksums_valid(rx)
self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
try:
)
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
+ "/err/%s/nat_keepalive" % self.tun4_input_node, 31
)
pkt = (
/ 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)
+ self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 31)
pkt = (
Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
/ Padding(0 * 21)
)
self.send_and_assert_no_replies(self.tun_if, pkt * 31)
- self.assert_error_counter_equal("/err/%s/Too Short" % self.tun4_input_node, 62)
+ self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 62)
class IpsecTun4Tests(IpsecTun4):
self.logger.info(self.vapi.ppcli("show ipsec all"))
self.verify_counters6(p, p, count)
+ def verify_keepalive(self, p):
+ # the sizeof Raw is calculated to pad to the minimum ehternet
+ # frame size of 64 btyes
+ pkt = (
+ Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
+ / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
+ / UDP(sport=333, dport=4500)
+ / Raw(b"\xff")
+ / Padding(0 * 1)
+ )
+ self.send_and_assert_no_replies(self.tun_if, pkt * 31)
+ self.assert_error_counter_equal(
+ "/err/%s/nat_keepalive" % self.tun6_input_node, 31
+ )
+
+ pkt = (
+ Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
+ / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
+ / UDP(sport=333, dport=4500)
+ / Raw(b"\xfe")
+ )
+ self.send_and_assert_no_replies(self.tun_if, pkt * 31)
+ self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 31)
+
+ pkt = (
+ Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
+ / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
+ / UDP(sport=333, dport=4500)
+ / Raw(b"\xfe")
+ / Padding(0 * 21)
+ )
+ self.send_and_assert_no_replies(self.tun_if, pkt * 31)
+ self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 62)
+
class IpsecTun6Tests(IpsecTun6):
"""UT test methods for Tunnel v6"""
Code Review / vpp.git / blobdiff