#!/usr/bin/env python3
""" PVTI tests """
import datetime
import base64
import os
import copy
import struct
from hashlib import blake2s
from config import config
from scapy.packet import Raw
from scapy.compat import raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.layers.vxlan import VXLAN
from vpp_interface import VppInterface
from vpp_pg_interface import is_ipv6_misc
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
from vpp_vxlan_tunnel import VppVxlanTunnel
from vpp_object import VppObject
from vpp_papi import VppEnum
from asfframework import tag_run_solo, tag_fixme_vpp_debug
from framework import VppTestCase
from re import compile
import unittest
from scapy.packet import Packet, bind_layers
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.fields import (
FlagsField,
XByteField,
XShortField,
ThreeBytesField,
ConditionalField,
ShortField,
ByteEnumField,
X3BytesField,
LEIntField,
ByteField,
StrLenField,
PacketListField,
LEShortField,
IntField,
ShortField,
XIntField,
)
import sys
def eprint(*args, **kwargs):
print(*args, file=sys.stderr, **kwargs)
#
# A custom decoder for Scapy for PVTI packet format
#
class PVTIChunk(Packet):
name = "PVTIChunk"
fields_desc = [
ShortField("total_chunk_length", None),
XShortField("_pad0", 0),
XIntField("_pad1", 0),
StrLenField("data", "", length_from=lambda pkt: pkt.total_chunk_length - 8),
]
# This prevents the first chunk from consuming the entire remaining
# contents of the packet
def extract_padding(self, s):
return "", s
def post_build(self, p, pay):
if self.total_chunk_length is None and self.data:
chunk_header_size = 8
l = chunk_header_size + len(self.data)
p = struct.pack("!H", l) + p[2:]
return p + pay
class PVTI(Packet):
name = "PVTI"
PVTI_ALIGN_BYTES = 9
fields_desc = [
IntField("seq", 0x0),
ByteField("stream_index", 0),
ByteField("chunk_count", None),
ByteField("reass_chunk_count", 0),
ByteField("mandatory_flags_mask", 0),
ByteField("flags_value", 0),
ByteField("pad_bytes", PVTI_ALIGN_BYTES),
StrLenField(
"pad", b"\xca" * PVTI_ALIGN_BYTES, length_from=lambda pkt: pkt.pad_bytes
),
PacketListField("chunks", [], PVTIChunk, count_from=lambda p: p.chunk_count),
]
def mysummary(self):
return self.sprintf("PVTI (len=%PVTI.total_len%)")
def post_build(self, p, pay):
if self.chunk_count is None:
l = len(self.chunks)
# offset of the chunk count within the fields
offset_of_chunk_count = 5
p = (
p[:offset_of_chunk_count]
+ struct.pack("b", l)
+ p[offset_of_chunk_count + 1 :]
)
return p + pay
bind_layers(UDP, PVTI, dport=12312)
# By default, set both ports to the test
# bind_layers(UDP, PVTI, sport=6192, dport=6192)
# PVTI ENcapsulator/DEcapsulator
class PvtiEnDe(object):
"""
PVTI encapsulator/decapsulator
"""
def __init__(
self,
local_ip,
local_port,
remote_ip,
remote_port,
underlay_mtu=1500,
for_rx_test=False,
):
self.for_rx_test = for_rx_test
self.local_ip = local_ip
self.local_port = local_port
self.remote_ip = remote_ip
self.remote_port = remote_port
self.underlay_mtu = underlay_mtu
self.stream_index = 0
self.tx_chunks = []
self.tx_n_reass_chunks = 0
self.tx_seq = 42
# payload = chunk headers + data
self.max_payload_len = underlay_mtu - len(raw(IP() / UDP() / PVTI()))
self.pvti_header_len = len(raw(PVTI()))
self.chunk_header_len = len(raw(PVTIChunk()))
def get_curr_payload_len(self):
tx_len = 0
for c in self.tx_chunks:
tx_len = tx_len + len(c.data) + self.chunk_header_len
return tx_len
def get_payload_room(self):
return self.max_payload_len - self.get_curr_payload_len()
def flush_tx_chunks(self, more_frags=False):
if self.for_rx_test:
ip_dst = self.local_ip
ip_src = self.remote_ip
else:
ip_src = self.local_ip
ip_dst = self.remote_ip
p = (
IP(
src=ip_src,
dst=ip_dst,
ttl=127,
frag=0,
flags=0,
id=self.tx_seq,
)
/ UDP(sport=self.local_port, dport=self.remote_port, chksum=0)
/ PVTI(
reass_chunk_count=self.tx_n_reass_chunks,
seq=self.tx_seq,
stream_index=self.stream_index,
chunks=self.tx_chunks,
)
)
p = IP(raw(p))
self.tx_n_reass_chunks = 0
self.tx_chunks = []
self.tx_seq = self.tx_seq + 1
return p
def encap_pkt(self, p):
out = []
if IP in p:
p[IP].ttl = p[IP].ttl - 1
payload_wip = p[IP].build()
elif IPv6 in p:
p[IPv6].hlim = p[IPv6].hlim - 1
payload_wip = p[IPv6].build()
split_chunks = False
huge_solo_packet = (
len(payload_wip) + self.chunk_header_len > self.get_payload_room()
) and len(self.tx_chunks) == 0
while True:
available_room = self.get_payload_room()
chunk_wip_len = len(payload_wip) + self.chunk_header_len
xpad0 = 0xABAB
xpad1 = 0xABABABAB
if chunk_wip_len <= available_room:
# happy case - there is enough space to fit the entire chunk
if split_chunks:
self.tx_n_reass_chunks = self.tx_n_reass_chunks + 1
tx = PVTIChunk(data=payload_wip, _pad0=xpad0, _pad1=xpad1)
self.tx_chunks.append(tx)
if chunk_wip_len == available_room:
# an unlikely perfect fit - send this packet.
out.append(self.flush_tx_chunks())
break
elif available_room < self.chunk_header_len + 1:
# Can not fit even a chunk header + 1 byte of data
# Flush and retry
out.append(self.flush_tx_chunks())
continue
else:
# Chop as much as we can from the packet
chop_len = available_room - self.chunk_header_len
if split_chunks:
self.tx_n_reass_chunks = self.tx_n_reass_chunks + 1
tx = PVTIChunk(data=payload_wip[:chop_len], _pad0=xpad0, _pad1=xpad1)
self.tx_chunks.append(tx)
out.append(self.flush_tx_chunks())
split_chunks = True
payload_wip = payload_wip[chop_len:]
continue
return out
def encap_packets(self, pkts):
out = []
self.start_encap()
for p in pkts:
out.extend(self.encap_pkt(p))
last_pkt = self.finish_encap()
if last_pkt != None:
out.append(last_pkt)
return out
def start_encap(self):
return None
def finish_encap(self):
out = None
if len(self.tx_chunks) > 0:
out = self.flush_tx_chunks()
return out
""" TestPvti is a subclass of VPPTestCase classes.
PVTI test.
"""
def get_field_bytes(pkt, name):
fld, val = pkt.getfield_and_val(name)
return fld.i2m(pkt, val)
class VppPvtiInterface(VppInterface):
"""
VPP PVTI interface
"""
def __init__(
self, test, local_ip, local_port, remote_ip, remote_port, underlay_mtu=1500
):
super(VppPvtiInterface, self).__init__(test)
self.local_ip = local_ip
self.local_port = local_port
self.remote_ip = remote_ip
self.remote_port = remote_port
self.underlay_mtu = underlay_mtu
def get_ende(self, for_rx_test=False):
return PvtiEnDe(
self.local_ip,
self.local_port,
self.remote_ip,
self.remote_port,
self.underlay_mtu,
for_rx_test,
)
def verify_encap_packets(self, orig_pkts, recv_pkts):
ende = self.get_ende()
recv2_pkts = ende.encap_packets(orig_pkts)
out1 = []
out2 = []
for i, pkt in enumerate(recv_pkts):
if IP in pkt:
rx_pkt = pkt[IP]
elif IPv6 in pkt:
rx_pkt = pkt[IPv6]
else:
raise "Neither IPv4 nor IPv6"
py_pkt = recv2_pkts[i]
if rx_pkt != py_pkt:
eprint("received packet:")
rx_pkt.show()
eprint("python packet:")
py_pkt.show()
out1.append(rx_pkt)
out2.append(py_pkt)
return (out1, out2)
def add_vpp_config(self):
r = self.test.vapi.pvti_interface_create(
interface={
"local_ip": self.local_ip,
"local_port": self.local_port,
"remote_ip": self.remote_ip,
"remote_port": self.remote_port,
"underlay_mtu": self.underlay_mtu,
}
)
self.set_sw_if_index(r.sw_if_index)
self.test.registry.register(self, self.test.logger)
return self
def remove_vpp_config(self):
self.test.vapi.pvti_interface_delete(sw_if_index=self._sw_if_index)
def query_vpp_config(self):
ts = self.test.vapi.pvti_interface_dump(sw_if_index=0xFFFFFFFF)
for t in ts:
if (
t.interface.sw_if_index == self._sw_if_index
and str(t.interface.local_ip) == self.local_ip
and t.interface.local_port == self.local_port
and t.interface.remote_port == self.remote_port
and str(t.interface.remote_ip) == self.remote_ip
):
self.test.logger.info("QUERY AYXX: true")
return True
return False
def __str__(self):
return self.object_id()
def object_id(self):
return "pvti-%d" % self._sw_if_index
@unittest.skipIf("pvti" in config.excluded_plugins, "Exclude PVTI plugin tests")
# @tag_run_solo
class TestPvti(VppTestCase):
"""Packet Vector Tunnel Interface (PVTI) Test Case"""
error_str = compile(r"Error")
# maxDiff = None
wg4_output_node_name = "/err/wg4-output-tun/"
wg4_input_node_name = "/err/wg4-input/"
wg6_output_node_name = "/err/wg6-output-tun/"
wg6_input_node_name = "/err/wg6-input/"
kp4_error = wg4_output_node_name + "Keypair error"
mac4_error = wg4_input_node_name + "Invalid MAC handshake"
peer4_in_err = wg4_input_node_name + "Peer error"
peer4_out_err = wg4_output_node_name + "Peer error"
kp6_error = wg6_output_node_name + "Keypair error"
mac6_error = wg6_input_node_name + "Invalid MAC handshake"
peer6_in_err = wg6_input_node_name + "Peer error"
peer6_out_err = wg6_output_node_name + "Peer error"
cookie_dec4_err = wg4_input_node_name + "Failed during Cookie decryption"
cookie_dec6_err = wg6_input_node_name + "Failed during Cookie decryption"
ratelimited4_err = wg4_input_node_name + "Handshake ratelimited"
ratelimited6_err = wg6_input_node_name + "Handshake ratelimited"
@classmethod
def setUpClass(cls):
super(TestPvti, cls).setUpClass()
try:
cls.create_pg_interfaces(range(2))
for i in cls.pg_interfaces:
i.admin_up()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
except Exception:
super(TestPvti, cls).tearDownClass()
raise
@classmethod
def tearDownClass(cls):
super(TestPvti, cls).tearDownClass()
def setUp(self):
super(VppTestCase, self).setUp()
self.base_kp4_err = self.statistics.get_err_counter(self.kp4_error)
self.base_mac4_err = self.statistics.get_err_counter(self.mac4_error)
self.base_peer4_in_err = self.statistics.get_err_counter(self.peer4_in_err)
self.base_peer4_out_err = self.statistics.get_err_counter(self.peer4_out_err)
self.base_kp6_err = self.statistics.get_err_counter(self.kp6_error)
self.base_mac6_err = self.statistics.get_err_counter(self.mac6_error)
self.base_peer6_in_err = self.statistics.get_err_counter(self.peer6_in_err)
self.base_peer6_out_err = self.statistics.get_err_counter(self.peer6_out_err)
self.base_cookie_dec4_err = self.statistics.get_err_counter(
self.cookie_dec4_err
)
self.base_cookie_dec6_err = self.statistics.get_err_counter(
self.cookie_dec6_err
)
self.base_ratelimited4_err = self.statistics.get_err_counter(
self.ratelimited4_err
)
self.base_ratelimited6_err = self.statistics.get_err_counter(
self.ratelimited6_err
)
def create_packets(
self, src_ip_if, count=1, size=150, for_rx=False, is_ip6=False, af_mix=False
):
pkts = []
total_packet_count = count
padstr0 = ""
padstr1 = ""
for i in range(0, 2000):
padstr0 = padstr0 + (".%03x" % i)
padstr1 = padstr1 + ("+%03x" % i)
for i in range(0, total_packet_count):
if af_mix:
is_ip6 = i % 2 == 1
dst_mac = src_ip_if.local_mac
src_mac = src_ip_if.remote_mac
if for_rx:
dst_ip4 = src_ip_if.remote_ip4
dst_ip6 = src_ip_if.remote_ip6
src_ip4 = "10.0.%d.4" % i
src_ip6 = "2001:db8::%x" % i
else:
src_ip4 = src_ip_if.remote_ip4
src_ip6 = src_ip_if.remote_ip6
dst_ip4 = "10.0.%d.4" % i
dst_ip6 = "2001:db8::%x" % i
src_l4 = 1234 + i
dst_l4 = 4321 + i
ulp = UDP(sport=src_l4, dport=dst_l4)
payload = "test pkt #%d" % i
if i % 2 == 1:
padstr = padstr1
else:
padstr = padstr0
p = Ether(dst=dst_mac, src=src_mac)
if is_ip6:
p /= IPv6(src=src_ip6, dst=dst_ip6)
else:
p /= IP(src=src_ip4, dst=dst_ip4, frag=0, flags=0)
p /= ulp / Raw(payload)
if i % 2 == 1 or total_packet_count == 1:
self.extend_packet(p, size, padstr)
else:
self.extend_packet(p, 150, padstr)
pkts.append(p)
return pkts
def add_rx_ether_header(self, in_pkts, rx_intf=None):
out = []
if rx_intf is None:
rx_intf = self.pg0
dst_mac = rx_intf.local_mac
src_mac = rx_intf.remote_mac
pkts = []
for p in in_pkts:
p0 = Ether(dst=dst_mac, src=src_mac) / p[IP]
out.append(p0)
return out
def encap_for_rx_test(self, pkts, rx_intf=None):
ende = self.pvti0.get_ende(for_rx_test=True)
encap_pkts = ende.encap_packets(pkts)
return self.add_rx_ether_header(encap_pkts, rx_intf)
def decrement_ttl_and_build(self, send_pkts):
out = []
pkts = copy.deepcopy(send_pkts)
for p in pkts:
p[IP].ttl = p[IP].ttl - 1
out.append(Ether(p.build()))
return out
def create_rx_packets(self, dst_ip_if, rx_intf=None, count=1, size=150):
pkts = []
total_packet_count = count
padstr = ""
if rx_intf is None:
rx_intf = self.pg0
for i in range(0, 2000):
padstr = padstr + (".%03x" % i)
dst_mac = rx_intf.local_mac
src_mac = rx_intf.remote_mac
for i in range(0, total_packet_count):
dst_ip4 = dst_ip_if.remote_ip4
src_ip4 = "10.0.%d.4" % i
src_l4 = 1234 + i
dst_l4 = 4321 + i
ulp = UDP(sport=src_l4, dport=dst_l4)
payload = "test"
# if i % 2 == 1 or total_packet_count == 1:
# self.extend_packet(p, size, padstr)
# else:
# self.extend_packet(p, 150, padstr)
pvti = PVTI(seq=42 + i, chunks=[])
for j in range(0, 32):
p = (
IP(src=src_ip4, dst=dst_ip4, frag=0, flags=0, id=j + 0x4000)
/ ulp
/ Raw(payload)
)
chunk0 = PVTIChunk(data=raw(p))
pvti.chunks.append(chunk0)
p = (
Ether(dst=dst_mac, src=src_mac)
/ IP(src="192.0.2.1", dst=rx_intf.local_ip4, id=0x3000 + i)
/ UDP(sport=12312, dport=12312)
/ pvti
)
# p.show()
# Ether(raw(p)).show()
pkts.append(p)
return pkts
def send_and_assert_no_replies_ignoring_init(
self, intf, pkts, remark="", timeout=None
):
self.pg_send(intf, pkts)
def _filter_out_fn(p):
return is_ipv6_misc(p) or is_handshake_init(p)
try:
if not timeout:
timeout = 1
for i in self.pg_interfaces:
i.assert_nothing_captured(
timeout=timeout, remark=remark, filter_out_fn=_filter_out_fn
)
timeout = 0.1
finally:
pass
def test_0000_pvti_interface(self):
"""Simple interface creation"""
local_port = 12312
peer_addr = self.pg0.remote_ip4 # "192.0.2.1"
peer_port = 31234
peer_port = 12312
# Create interface
pvti0 = VppPvtiInterface(
self, self.pg1.local_ip4, local_port, peer_addr, peer_port
).add_vpp_config()
self.logger.info(self.vapi.cli("sh int"))
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show pvti tx peers"))
self.logger.info(self.vapi.cli("show pvti rx peers"))
# delete interface
pvti0.remove_vpp_config()
# self.logger.info(self.vapi.cli("show pvti interface"))
# pvti0.add_vpp_config()
def test_0001_pvti_send_simple_1pkt(self):
"""v4o4 TX: Simple packet: 1 -> 1"""
self.prepare_for_test("v4o4_1pkt_simple")
pkts = self.create_packets(self.pg1)
recv_pkts = self.send_and_expect(self.pg1, pkts, self.pg0)
for p in recv_pkts:
self.logger.info(p)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(pkts, recv_pkts)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0101_pvti_send_simple_1pkt(self):
"""v6o4 TX: Simple packet: 1 -> 1"""
self.prepare_for_test("v6o4_1pkt_simple")
pkts = self.create_packets(self.pg1, is_ip6=True)
recv_pkts = self.send_and_expect(self.pg1, pkts, self.pg0, n_rx=1)
for p in recv_pkts:
self.logger.info(p)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(pkts, recv_pkts)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0002_pvti_send_simple_2pkt(self):
"""TX: Simple packet: 2 -> 1"""
self.prepare_for_test("2pkt_simple")
send_pkts = self.create_packets(self.pg1, count=2)
pkts = copy.deepcopy(send_pkts)
rx = self.send_and_expect(self.pg1, pkts, self.pg0, n_rx=1)
for p in rx:
self.logger.info(p)
# p.show()
payload0 = rx[0][PVTI].chunks[0].data
payload1 = rx[0][PVTI].chunks[1].data
pktA0 = IP(payload0)
pktA1 = IP(payload1)
p0 = pkts[0][IP]
p0.ttl = p0.ttl - 1
pktB0 = IP(p0.build())
p1 = pkts[1][IP]
p1.ttl = p1.ttl - 1
pktB1 = IP(p1.build())
self.assertEqual(pktA0, pktB0)
self.assertEqual(pktA1, pktB1)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def prepare_for_test(self, test_name, underlay_mtu=1500, is_ip6=False):
local_port = 12312
peer_ip4_addr = "192.0.2.1"
peer_ip6_addr = "2001:db8:dead::1"
peer_port = 31234
peer_port = 12312
for i in self.pg_interfaces:
i.test_name = test_name
if is_ip6:
self.pvti0 = VppPvtiInterface(
self,
self.pg1.local_ip6,
local_port,
peer_ip6_addr,
peer_port,
underlay_mtu,
).add_vpp_config()
else:
self.pvti0 = VppPvtiInterface(
self,
self.pg1.local_ip4,
local_port,
peer_ip4_addr,
peer_port,
underlay_mtu,
).add_vpp_config()
self.pvti0.config_ip4()
self.pvti0.config_ip6()
self.pvti0.admin_up()
self.logger.info(self.vapi.cli("ip route add 0.0.0.0/0 via 172.16.3.3"))
## FIXME: using direct "interface" below results in blackouts. intermittently.
# self.logger.info(self.vapi.cli("ip route 0.0.0.0/0 via pvti0"))
self.logger.info(self.vapi.cli("ip route add ::/0 via pvti0"))
self.logger.info(self.vapi.cli("ip route add 192.0.2.1/32 via pg0"))
self.logger.info(self.vapi.cli("ip neighbor pg0 192.0.2.1 000c.0102.0304"))
self.logger.info(self.vapi.cli("ip route 2001:db8:dead::1/128 via pg0"))
self.logger.info(
self.vapi.cli("ip neighbor pg0 2001:db8:dead::1 000c.0102.0304")
)
self.logger.info(self.vapi.cli("ip neighbor pg1 172.16.2.2 000c.0102.0304"))
self.logger.info(self.vapi.cli("sh int"))
self.logger.info(self.vapi.cli("sh ip fib"))
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("set interface ip pvti-bypass pg0"))
def cleanup_after_test(self):
self.logger.info(self.vapi.cli("ip neighbor del pg0 192.0.2.1 000c.0102.0304"))
self.logger.info(self.vapi.cli("ip neighbor del pg1 172.16.2.2 000c.0102.0304"))
self.logger.info(self.vapi.cli("ip route del 192.0.2.1/32 via pg0"))
# self.logger.info(self.vapi.cli("ip route del 0.0.0.0/0 via pvti0"))
self.logger.info(self.vapi.cli("ip route del ::/0 via pvti0"))
self.logger.info(self.vapi.cli("sh int"))
self.logger.info(self.vapi.cli("show pvti interface"))
self.pvti0.remove_vpp_config()
def test_0003_pvti_send_simple_1pkt_big(self):
"""TX: Simple big packet: 1 -> 2"""
self.prepare_for_test("1big_pkt")
send_pkts = self.create_packets(self.pg1, count=1, size=1900)
pkts = copy.deepcopy(send_pkts)
self.logger.info("count: ")
self.logger.info(len(pkts))
rx = self.send_and_expect(self.pg1, pkts, self.pg0, n_rx=2)
for p in rx:
self.logger.info(p)
self.logger.info(len(p[PVTI].chunks[0].data))
# p.show()
payload = rx[0][PVTI].chunks[0].data + rx[1][PVTI].chunks[0].data
pkt1 = IP(payload)
p0 = pkts[0][IP]
p0.ttl = p0.ttl - 1
pkt0 = IP(p0.build())
self.assertEqual(pkt0, pkt1)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0004_pvti_send_simple_5pkt_big(self):
"""v4o4 TX: Simple big packets: 5 -> 2"""
self.prepare_for_test("v4o4_5big_pkt")
send_pkts = self.create_packets(self.pg1, count=5, size=1050)
self.logger.info("count: %d " % len(send_pkts))
# self.logger.info(len(pkts))
rx = self.send_and_expect(self.pg1, send_pkts, self.pg0, n_rx=2)
for p in rx:
self.logger.info(p)
self.logger.info(len(p[PVTI].chunks[0].data))
# p.show()
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0104_pvti_send_simple_5pkt_big(self):
"""v6o4 TX: Simple big packets: 5 -> 2"""
self.prepare_for_test("v4o4_5big_pkt")
send_pkts = self.create_packets(self.pg1, count=5, size=1050, is_ip6=True)
self.logger.info("count: %d " % len(send_pkts))
# self.logger.info(len(pkts))
rx = self.send_and_expect(self.pg1, send_pkts, self.pg0, n_rx=2)
for p in rx:
self.logger.info(p)
self.logger.info(len(p[PVTI].chunks[0].data))
# p.show()
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def Xtest_0204_pvti_send_simple_5pkt_mix(self):
"""vXo4 TX: Simple packets mix: 5 -> 2"""
# FIXME: This test is disabled for now, but left here, to have this comment
# The mix of IPv4 and IPv6 packets in VPP will forward two
# different graphs, so after encap it will result in two
# PV packets: one with IPv4 chunks, and one with IPv6 chunks.
# The python test encapsulator does not do this, and it is probably
# a useless idea to introduce attempts to mimic this behavior,
# because in any case one can not expect the orderly scheduling
# of IPv4 vs IPv6 graph processing.
self.prepare_for_test("vXo4_5big_pkt")
send_pkts = self.create_packets(self.pg1, count=5, size=1050, af_mix=True)
# self.logger.info(len(pkts))
rx = self.send_and_expect(self.pg1, send_pkts, self.pg0, n_rx=2)
for p in rx:
self.logger.info(p)
self.logger.info(len(p[PVTI].chunks[0].data))
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0005_pvti_send_mix_3pkt_medium_mtu(self):
"""TX: small+big+small packets over medium mtu: 3 -> 3"""
self.prepare_for_test("3pkt_small_mtu", underlay_mtu=400)
send_pkts = self.create_packets(self.pg1, count=3, size=500)
pkts = copy.deepcopy(send_pkts)
self.logger.info("count: %d " % len(send_pkts))
# self.logger.info(len(pkts))
rx = self.send_and_expect(self.pg1, send_pkts, self.pg0, n_rx=3)
for p in rx:
self.logger.info(p)
self.logger.info(len(p[PVTI].chunks[0].data))
# p.show()
# check the middle chunk which is spread across two packets
payload = rx[0][PVTI].chunks[1].data + rx[1][PVTI].chunks[0].data
pkt1 = IP(payload)
p0 = pkts[1][IP]
p0.ttl = p0.ttl - 1
pkt0 = IP(p0.build())
self.assertEqual(pkt0, pkt1)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0006_pvti_send_mix_4pkt_medium_mtu(self):
"""TX: small+big+small packets over 600 mtu: 4 -> 3"""
self.prepare_for_test("6pkt_small_mtu", underlay_mtu=600)
send_pkts = self.create_packets(self.pg1, count=4, size=500)
pkts = copy.deepcopy(send_pkts)
# self.logger.info(len(pkts))
rx = self.send_and_expect(self.pg1, send_pkts, self.pg0, n_rx=3)
for p in rx:
self.logger.info(p)
self.logger.info(len(p[PVTI].chunks[0].data))
# p.show()
# check the middle chunk which is spread across two packets
payload = rx[0][PVTI].chunks[1].data + rx[1][PVTI].chunks[0].data
pkt1 = IP(payload)
p0 = pkts[1][IP]
p0.ttl = p0.ttl - 1
pkt0 = IP(p0.build())
self.assertEqual(pkt0, pkt1)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0007_pvti_send_simple_1_3_pkt(self):
"""TX: Simple packet: 1 -> 3, small mtu"""
self.prepare_for_test("1_3_pkt_simple", underlay_mtu=520)
send_pkts = self.create_packets(self.pg1, count=1, size=1400)
pkts = copy.deepcopy(send_pkts)
rx = self.send_and_expect(self.pg1, pkts, self.pg0, n_rx=3)
for p in rx:
self.logger.info(p)
c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_0008_pvti_chained_1_3_pkt(self):
"""TX: Chained packet: 2700 byte 1 -> 3, mtu 1000"""
self.prepare_for_test("1_3_pkt_simple", underlay_mtu=1000)
send_pkts = self.create_packets(self.pg1, count=1, size=2700)
pkts = copy.deepcopy(send_pkts)
pkt0 = Ether(raw(pkts[0]))[IP]
rx = self.send_and_expect(self.pg1, send_pkts, self.pg0, n_rx=3)
for p in rx:
self.logger.info(p)
p0 = pkts[0][IP]
p0.ttl = p0.ttl - 1
pkt0 = IP(p0.build())
payload = (
rx[0][PVTI].chunks[0].data
+ rx[1][PVTI].chunks[0].data
+ rx[2][PVTI].chunks[0].data
# + rx[2][PVTI].chunks[1].data
)
pkt1 = IP(payload)
self.assertEqual(pkt0, pkt1)
# FIXME: this will fail because the send path
# does not combine the data from two chained blocks.
# when this succeeds, the above checks in this testcase will need to be redone
# c_pkts, py_pkts = self.pvti0.verify_encap_packets(send_pkts, rx)
# self.assertEqual(c_pkts, py_pkts)
self.cleanup_after_test()
def test_1001_pvti_rx_simple_1pkt(self):
"""RX: Simple packet: 1 -> 32"""
self.prepare_for_test("1pkt_rx_simple")
pkts = self.create_rx_packets(self.pg1, rx_intf=self.pg0)
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show udp ports"))
recv_pkts = self.send_and_expect(self.pg0, pkts, self.pg1, n_rx=32)
for p in recv_pkts:
self.logger.info(p)
self.cleanup_after_test()
def test_1002_pvti_rx_big_1buf(self):
"""RX: Orig Big packet, single buf: 2 -> 1"""
self.prepare_for_test("1buf_rx_big")
pkts_orig = self.create_packets(self.pg1, count=1, size=1900, for_rx=True)
pkts = self.encap_for_rx_test(pkts_orig, rx_intf=self.pg0)
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show udp ports"))
known_good_pkts = self.decrement_ttl_and_build(pkts_orig)
recv_pkts = self.send_and_expect(self.pg0, pkts, self.pg1, n_rx=1)
for i, p in enumerate(recv_pkts):
self.logger.info(p)
self.assertEqual(p[IP], known_good_pkts[i][IP])
self.cleanup_after_test()
def test_1003_pvti_rx_big_2buf(self):
"""RX: Very Big packet, chained buf: 3 -> 1"""
self.prepare_for_test("2buf_rx_big")
pkts_orig = self.create_packets(self.pg1, count=1, size=3000, for_rx=True)
pkts = self.encap_for_rx_test(pkts_orig, rx_intf=self.pg0)
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show udp ports"))
known_good_pkts = self.decrement_ttl_and_build(pkts_orig)
recv_pkts = self.send_and_expect(self.pg0, pkts, self.pg1, n_rx=1)
for i, p in enumerate(recv_pkts):
self.logger.info(p)
if p[IP] != known_good_pkts[i][IP]:
p[IP].show()
known_good_pkts[i][IP].show()
self.assertEqual(p[IP], known_good_pkts[i][IP])
self.cleanup_after_test()
def test_1004_pvti_rx_big_2buf_and_small(self):
"""RX: Very Big packet, chained buf: 3 -> 1 + small pkt"""
self.prepare_for_test("2buf_rx_big_and_small")
pkts_orig = self.create_packets(self.pg1, count=2, size=3000, for_rx=True)
pkts = self.encap_for_rx_test(pkts_orig, rx_intf=self.pg0)
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show udp ports"))
known_good_pkts = self.decrement_ttl_and_build(pkts_orig)
recv_pkts = self.send_and_expect(self.pg0, pkts, self.pg1, n_rx=2)
for i, p in enumerate(recv_pkts):
self.logger.info(p)
if p[IP] != known_good_pkts[i][IP]:
p[IP].show()
known_good_pkts[i][IP].show()
self.assertEqual(p[IP], known_good_pkts[i][IP])
self.cleanup_after_test()
def test_1005_pvti_rx_big_2buf_and_small_drop(self):
"""RX: Very Big packet, chained buf: 3 -> 1 + small pkt, encap pkt lost"""
self.prepare_for_test("2buf_rx_big_and_small_drop")
pkts_orig = self.create_packets(self.pg1, count=3, size=3000, for_rx=True)
pkts = self.encap_for_rx_test(pkts_orig, rx_intf=self.pg0)
# drop the second packet after encapsulation (the one with the second frag of the large packet)
pkts.pop(1)
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show udp ports"))
known_good_pkts = self.decrement_ttl_and_build(pkts_orig)
# drop the large original packet, leaving just two small ones
known_good_pkts.pop(1)
recv_pkts = self.send_and_expect(self.pg0, pkts, self.pg1, n_rx=2)
for i, p in enumerate(recv_pkts):
self.logger.info(p)
if p[IP] != known_good_pkts[i][IP]:
p[IP].show()
known_good_pkts[i][IP].show()
self.assertEqual(p[IP], known_good_pkts[i][IP])
self.cleanup_after_test()
def test_1006_pvti_rx_big_2buf_and_small_drop2(self):
"""RX: Very Big packet, chained buf: 3 -> 1 + small pkt, non-initial frag pkt lost"""
self.prepare_for_test("2buf_rx_big_and_small_drop2")
pkts_orig = self.create_packets(self.pg1, count=3, size=6000, for_rx=True)
pkts = self.encap_for_rx_test(pkts_orig, rx_intf=self.pg0)
# drop the second packet after encapsulation (the one with the second frag of the large packet)
pkts.pop(2)
self.logger.info(self.vapi.cli("show pvti interface"))
self.logger.info(self.vapi.cli("show udp ports"))
known_good_pkts = self.decrement_ttl_and_build(pkts_orig)
# drop the large original packet, leaving just two small ones
known_good_pkts.pop(1)
recv_pkts = self.send_and_expect(self.pg0, pkts, self.pg1, n_rx=2)
for i, p in enumerate(recv_pkts):
self.logger.info(p)
if p[IP] != known_good_pkts[i][IP]:
p[IP].show()
known_good_pkts[i][IP].show()
self.assertEqual(p[IP], known_good_pkts[i][IP])
self.cleanup_after_test()
class PvtiHandoffTests(TestPvti):
"""Pvti Tests in multi worker setup"""
vpp_worker_count = 2
def xtest_wg_peer_init(self):
"""Handoff"""
port = 12383
# Create interfaces
wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip4()
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
peer_1 = VppWgPeer(
self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.2.0/24", "10.11.3.0/24"]
).add_vpp_config()
self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
r1 = VppIpRoute(
self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
).add_vpp_config()
# skip the first automatic handshake
self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
# send a valid handsake init for which we expect a response
p = peer_1.mk_handshake(self.pg1)
rx = self.send_and_expect(self.pg1, [p], self.pg1)
peer_1.consume_response(rx[0])
# send a data packet from the peer through the tunnel
# this completes the handshake and pins the peer to worker 0
p = (
IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
/ UDP(sport=222, dport=223)
/ Raw()
)
d = peer_1.encrypt_transport(p)
p = peer_1.mk_tunnel_header(self.pg1) / (
Pvti(message_type=4, reserved_zero=0)
/ PvtiTransport(
receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
)
)
rxs = self.send_and_expect(self.pg1, [p], self.pg0, worker=0)
for rx in rxs:
self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
self.assertEqual(rx[IP].ttl, 19)
# send a packets that are routed into the tunnel
# and pins the peer tp worker 1
pe = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
/ UDP(sport=555, dport=556)
/ Raw(b"\x00" * 80)
)
rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=1)
peer_1.validate_encapped(rxs, pe)
# send packets into the tunnel, from the other worker
p = [
(
peer_1.mk_tunnel_header(self.pg1)
/ Pvti(message_type=4, reserved_zero=0)
/ PvtiTransport(
receiver_index=peer_1.sender,
counter=ii + 1,
encrypted_encapsulated_packet=peer_1.encrypt_transport(
(
IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
/ UDP(sport=222, dport=223)
/ Raw()
)
),
)
)
for ii in range(255)
]
rxs = self.send_and_expect(self.pg1, p, self.pg0, worker=1)
for rx in rxs:
self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
self.assertEqual(rx[IP].ttl, 19)
# send a packets that are routed into the tunnel
# from worker 0
rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=0)
peer_1.validate_encapped(rxs, pe)
r1.remove_vpp_config()
peer_1.remove_vpp_config()
wg0.remove_vpp_config()