import datetime
import base64
+import os
from hashlib import blake2s
from scapy.packet import Packet
from scapy.layers.l2 import Ether, ARP
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
-from scapy.contrib.wireguard import Wireguard, WireguardResponse, \
- WireguardInitiation, WireguardTransport
-from cryptography.hazmat.primitives.asymmetric.x25519 import \
- X25519PrivateKey, X25519PublicKey
-from cryptography.hazmat.primitives.serialization import Encoding, \
- PrivateFormat, PublicFormat, NoEncryption
+from scapy.contrib.wireguard import (
+ Wireguard,
+ WireguardResponse,
+ WireguardInitiation,
+ WireguardTransport,
+ WireguardCookieReply,
+)
+from cryptography.hazmat.primitives.asymmetric.x25519 import (
+ X25519PrivateKey,
+ X25519PublicKey,
+)
+from cryptography.hazmat.primitives.serialization import (
+ Encoding,
+ PrivateFormat,
+ PublicFormat,
+ NoEncryption,
+)
from cryptography.hazmat.primitives.hashes import BLAKE2s, Hash
from cryptography.hazmat.primitives.hmac import HMAC
from cryptography.hazmat.backends import default_backend
from noise.connection import NoiseConnection, Keypair
+from Crypto.Cipher import ChaCha20_Poly1305
+from Crypto.Random import get_random_bytes
+
from vpp_ipip_tun_interface import VppIpIpTunInterface
from vpp_interface import VppInterface
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_object import VppObject
+from vpp_papi import VppEnum
from framework import VppTestCase
from re import compile
import unittest
def private_key_bytes(k):
- return k.private_bytes(Encoding.Raw,
- PrivateFormat.Raw,
- NoEncryption())
+ return k.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())
def public_key_bytes(k):
- return k.public_bytes(Encoding.Raw,
- PublicFormat.Raw)
+ return k.public_bytes(Encoding.Raw, PublicFormat.Raw)
+
+
+def get_field_bytes(pkt, name):
+ fld, val = pkt.getfield_and_val(name)
+ return fld.i2m(pkt, val)
class VppWgInterface(VppInterface):
return private_key_bytes(self.private_key)
def add_vpp_config(self):
- r = self.test.vapi.wireguard_interface_create(interface={
- 'user_instance': 0xffffffff,
- 'port': self.port,
- 'src_ip': self.src,
- 'private_key': private_key_bytes(self.private_key),
- 'generate_key': False
- })
+ r = self.test.vapi.wireguard_interface_create(
+ interface={
+ "user_instance": 0xFFFFFFFF,
+ "port": self.port,
+ "src_ip": self.src,
+ "private_key": private_key_bytes(self.private_key),
+ "generate_key": False,
+ }
+ )
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.wireguard_interface_delete(
- sw_if_index=self._sw_if_index)
+ self.test.vapi.wireguard_interface_delete(sw_if_index=self._sw_if_index)
def query_vpp_config(self):
- ts = self.test.vapi.wireguard_interface_dump(sw_if_index=0xffffffff)
+ ts = self.test.vapi.wireguard_interface_dump(sw_if_index=0xFFFFFFFF)
for t in ts:
- if t.interface.sw_if_index == self._sw_if_index and \
- str(t.interface.src_ip) == self.src and \
- t.interface.port == self.port and \
- t.interface.private_key == private_key_bytes(self.private_key):
+ if (
+ t.interface.sw_if_index == self._sw_if_index
+ and str(t.interface.src_ip) == self.src
+ and t.interface.port == self.port
+ and t.interface.private_key == private_key_bytes(self.private_key)
+ ):
return True
return False
+ def want_events(self, peer_index=0xFFFFFFFF):
+ self.test.vapi.want_wireguard_peer_events(
+ enable_disable=1,
+ pid=os.getpid(),
+ sw_if_index=self._sw_if_index,
+ peer_index=peer_index,
+ )
+
+ def wait_events(self, expect, peers, timeout=5):
+ for i in range(len(peers)):
+ rv = self.test.vapi.wait_for_event(timeout, "wireguard_peer_event")
+ self.test.assertEqual(rv.peer_index, peers[i])
+ self.test.assertEqual(rv.flags, expect)
+
def __str__(self):
return self.object_id()
routes = test.vapi.ip_route_dump(table_id, is_ip6)
for e in routes:
- if table_id == e.route.table_id \
- and str(e.route.prefix) == str(prefix):
+ if table_id == e.route.table_id and str(e.route.prefix) == str(prefix):
return True
return False
class VppWgPeer(VppObject):
-
- def __init__(self,
- test,
- itf,
- endpoint,
- port,
- allowed_ips,
- persistent_keepalive=15):
+ def __init__(self, test, itf, endpoint, port, allowed_ips, persistent_keepalive=15):
self._test = test
self.itf = itf
self.endpoint = endpoint
self.private_key = X25519PrivateKey.generate()
self.public_key = self.private_key.public_key()
+ # cookie related params
+ self.cookie_key = blake2s(b"cookie--" + self.public_key_bytes()).digest()
+ self.last_sent_cookie = None
+
self.noise = NoiseConnection.from_name(NOISE_HANDSHAKE_NAME)
def add_vpp_config(self, is_ip6=False):
rv = self._test.vapi.wireguard_peer_add(
peer={
- 'public_key': self.public_key_bytes(),
- 'port': self.port,
- 'endpoint': self.endpoint,
- 'n_allowed_ips': len(self.allowed_ips),
- 'allowed_ips': self.allowed_ips,
- 'sw_if_index': self.itf.sw_if_index,
- 'persistent_keepalive': self.persistent_keepalive})
+ "public_key": self.public_key_bytes(),
+ "port": self.port,
+ "endpoint": self.endpoint,
+ "n_allowed_ips": len(self.allowed_ips),
+ "allowed_ips": self.allowed_ips,
+ "sw_if_index": self.itf.sw_if_index,
+ "persistent_keepalive": self.persistent_keepalive,
+ }
+ )
self.index = rv.peer_index
self.receiver_index = self.index + 1
self._test.registry.register(self, self._test.logger)
self._test.vapi.wireguard_peer_remove(peer_index=self.index)
def object_id(self):
- return ("wireguard-peer-%s" % self.index)
+ return "wireguard-peer-%s" % self.index
def public_key_bytes(self):
return public_key_bytes(self.public_key)
peers = self._test.vapi.wireguard_peers_dump()
for p in peers:
- if p.peer.public_key == self.public_key_bytes() and \
- p.peer.port == self.port and \
- str(p.peer.endpoint) == self.endpoint and \
- p.peer.sw_if_index == self.itf.sw_if_index and \
- len(self.allowed_ips) == p.peer.n_allowed_ips:
+ if (
+ p.peer.public_key == self.public_key_bytes()
+ and p.peer.port == self.port
+ and str(p.peer.endpoint) == self.endpoint
+ and p.peer.sw_if_index == self.itf.sw_if_index
+ and len(self.allowed_ips) == p.peer.n_allowed_ips
+ ):
self.allowed_ips.sort()
p.peer.allowed_ips.sort()
return True
return False
- def set_responder(self):
- self.noise.set_as_responder()
-
def mk_tunnel_header(self, tx_itf, is_ip6=False):
if is_ip6 is False:
- return (Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac) /
- IP(src=self.endpoint, dst=self.itf.src) /
- UDP(sport=self.port, dport=self.itf.port))
+ return (
+ Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
+ / IP(src=self.endpoint, dst=self.itf.src)
+ / UDP(sport=self.port, dport=self.itf.port)
+ )
else:
- return (Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac) /
- IPv6(src=self.endpoint, dst=self.itf.src) /
- UDP(sport=self.port, dport=self.itf.port))
+ return (
+ Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
+ / IPv6(src=self.endpoint, dst=self.itf.src)
+ / UDP(sport=self.port, dport=self.itf.port)
+ )
def noise_init(self, public_key=None):
self.noise.set_prologue(NOISE_IDENTIFIER_NAME)
# local/this private
self.noise.set_keypair_from_private_bytes(
- Keypair.STATIC,
- private_key_bytes(self.private_key))
+ Keypair.STATIC, private_key_bytes(self.private_key)
+ )
# remote's public
self.noise.set_keypair_from_public_bytes(
- Keypair.REMOTE_STATIC,
- public_key_bytes(public_key))
+ Keypair.REMOTE_STATIC, public_key_bytes(public_key)
+ )
self.noise.start_handshake()
+ def mk_cookie(self, p, tx_itf, is_resp=False, is_ip6=False):
+ self.verify_header(p, is_ip6)
+
+ wg_pkt = Wireguard(p[Raw])
+
+ if is_resp:
+ self._test.assertEqual(wg_pkt[Wireguard].message_type, 2)
+ self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
+ self._test.assertEqual(wg_pkt[WireguardResponse].mac2, bytes([0] * 16))
+ else:
+ self._test.assertEqual(wg_pkt[Wireguard].message_type, 1)
+ self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
+ self._test.assertEqual(wg_pkt[WireguardInitiation].mac2, bytes([0] * 16))
+
+ # collect info from wg packet (initiation or response)
+ src = get_field_bytes(p[IPv6 if is_ip6 else IP], "src")
+ sport = p[UDP].sport.to_bytes(2, byteorder="big")
+ if is_resp:
+ mac1 = wg_pkt[WireguardResponse].mac1
+ sender_index = wg_pkt[WireguardResponse].sender_index
+ else:
+ mac1 = wg_pkt[WireguardInitiation].mac1
+ sender_index = wg_pkt[WireguardInitiation].sender_index
+
+ # make cookie reply
+ cookie_reply = Wireguard() / WireguardCookieReply()
+ cookie_reply[Wireguard].message_type = 3
+ cookie_reply[Wireguard].reserved_zero = 0
+ cookie_reply[WireguardCookieReply].receiver_index = sender_index
+ nonce = get_random_bytes(24)
+ cookie_reply[WireguardCookieReply].nonce = nonce
+
+ # generate cookie data
+ changing_secret = get_random_bytes(32)
+ self.last_sent_cookie = blake2s(
+ src + sport, digest_size=16, key=changing_secret
+ ).digest()
+
+ # encrypt cookie data
+ cipher = ChaCha20_Poly1305.new(key=self.cookie_key, nonce=nonce)
+ cipher.update(mac1)
+ ciphertext, tag = cipher.encrypt_and_digest(self.last_sent_cookie)
+ cookie_reply[WireguardCookieReply].encrypted_cookie = ciphertext + tag
+
+ # prepare cookie reply to be sent
+ cookie_reply = self.mk_tunnel_header(tx_itf, is_ip6) / cookie_reply
+
+ return cookie_reply
+
def mk_handshake(self, tx_itf, is_ip6=False, public_key=None):
self.noise.set_as_initiator()
self.noise_init(public_key)
- p = (Wireguard() / WireguardInitiation())
+ p = Wireguard() / WireguardInitiation()
p[Wireguard].message_type = 1
p[Wireguard].reserved_zero = 0
# some random data for the message
# lifted from the noise protocol's wireguard example
now = datetime.datetime.now()
- tai = struct.pack('!qi', 4611686018427387914 + int(now.timestamp()),
- int(now.microsecond * 1e3))
+ tai = struct.pack(
+ "!qi",
+ 4611686018427387914 + int(now.timestamp()),
+ int(now.microsecond * 1e3),
+ )
b = self.noise.write_message(payload=tai)
# load noise into init message
p[WireguardInitiation].encrypted_timestamp = b[80:108]
# generate the mac1 hash
- mac_key = blake2s(b'mac1----' +
- self.itf.public_key_bytes()).digest()
- p[WireguardInitiation].mac1 = blake2s(bytes(p)[0:116],
- digest_size=16,
- key=mac_key).digest()
+ mac_key = blake2s(b"mac1----" + self.itf.public_key_bytes()).digest()
+ p[WireguardInitiation].mac1 = blake2s(
+ bytes(p)[0:116], digest_size=16, key=mac_key
+ ).digest()
p[WireguardInitiation].mac2 = bytearray(16)
- p = (self.mk_tunnel_header(tx_itf, is_ip6) / p)
+ p = self.mk_tunnel_header(tx_itf, is_ip6) / p
return p
self._test.assertEqual(p[UDP].dport, self.port)
self._test.assert_packet_checksums_valid(p)
- def consume_init(self, p, tx_itf, is_ip6=False):
+ def consume_init(self, p, tx_itf, is_ip6=False, is_mac2=False):
self.noise.set_as_responder()
self.noise_init(self.itf.public_key)
self.verify_header(p, is_ip6)
self.sender = init[WireguardInitiation].sender_index
- # validate the hash
- mac_key = blake2s(b'mac1----' +
- public_key_bytes(self.public_key)).digest()
- mac1 = blake2s(bytes(init)[0:-32],
- digest_size=16,
- key=mac_key).digest()
+ # validate the mac1 hash
+ mac_key = blake2s(b"mac1----" + public_key_bytes(self.public_key)).digest()
+ mac1 = blake2s(bytes(init)[0:-32], digest_size=16, key=mac_key).digest()
self._test.assertEqual(init[WireguardInitiation].mac1, mac1)
+ # validate the mac2 hash
+ if is_mac2:
+ self._test.assertNotEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
+ self._test.assertNotEqual(self.last_sent_cookie, None)
+ mac2 = blake2s(
+ bytes(init)[0:-16], digest_size=16, key=self.last_sent_cookie
+ ).digest()
+ self._test.assertEqual(init[WireguardInitiation].mac2, mac2)
+ self.last_sent_cookie = None
+ else:
+ self._test.assertEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
+
# this passes only unencrypted_ephemeral, encrypted_static,
# encrypted_timestamp fields of the init
payload = self.noise.read_message(bytes(init)[8:-32])
# build the response
b = self.noise.write_message()
- mac_key = blake2s(b'mac1----' +
- public_key_bytes(self.itf.public_key)).digest()
- resp = (Wireguard(message_type=2, reserved_zero=0) /
- WireguardResponse(sender_index=self.receiver_index,
- receiver_index=self.sender,
- unencrypted_ephemeral=b[0:32],
- encrypted_nothing=b[32:]))
- mac1 = blake2s(bytes(resp)[:-32],
- digest_size=16,
- key=mac_key).digest()
+ mac_key = blake2s(b"mac1----" + public_key_bytes(self.itf.public_key)).digest()
+ resp = Wireguard(message_type=2, reserved_zero=0) / WireguardResponse(
+ sender_index=self.receiver_index,
+ receiver_index=self.sender,
+ unencrypted_ephemeral=b[0:32],
+ encrypted_nothing=b[32:],
+ )
+ mac1 = blake2s(bytes(resp)[:-32], digest_size=16, key=mac_key).digest()
resp[WireguardResponse].mac1 = mac1
- resp = (self.mk_tunnel_header(tx_itf, is_ip6) / resp)
+ resp = self.mk_tunnel_header(tx_itf, is_ip6) / resp
self._test.assertTrue(self.noise.handshake_finished)
return resp
self._test.assertEqual(resp[Wireguard].message_type, 2)
self._test.assertEqual(resp[Wireguard].reserved_zero, 0)
- self._test.assertEqual(resp[WireguardResponse].receiver_index,
- self.receiver_index)
+ self._test.assertEqual(
+ resp[WireguardResponse].receiver_index, self.receiver_index
+ )
self.sender = resp[Wireguard].sender_index
payload = self.noise.read_message(bytes(resp)[12:60])
- self._test.assertEqual(payload, b'')
+ self._test.assertEqual(payload, b"")
self._test.assertTrue(self.noise.handshake_finished)
def decrypt_transport(self, p, is_ip6=False):
p = Wireguard(p[Raw])
self._test.assertEqual(p[Wireguard].message_type, 4)
self._test.assertEqual(p[Wireguard].reserved_zero, 0)
- self._test.assertEqual(p[WireguardTransport].receiver_index,
- self.receiver_index)
+ self._test.assertEqual(
+ p[WireguardTransport].receiver_index, self.receiver_index
+ )
- d = self.noise.decrypt(
- p[WireguardTransport].encrypted_encapsulated_packet)
+ d = self.noise.decrypt(p[WireguardTransport].encrypted_encapsulated_packet)
return d
def encrypt_transport(self, p):
# chech the oringial packet is present
self._test.assertEqual(rx[IP].dst, tx[IP].dst)
- self._test.assertEqual(rx[IP].ttl, tx[IP].ttl-1)
+ self._test.assertEqual(rx[IP].ttl, tx[IP].ttl - 1)
else:
rx = IPv6(self.decrypt_transport(rx))
# chech the oringial packet is present
self._test.assertEqual(rx[IPv6].dst, tx[IPv6].dst)
- self._test.assertEqual(rx[IPv6].ttl, tx[IPv6].ttl-1)
+ self._test.assertEqual(rx[IPv6].ttl, tx[IPv6].ttl - 1)
+
+ def want_events(self):
+ self._test.vapi.want_wireguard_peer_events(
+ enable_disable=1,
+ pid=os.getpid(),
+ peer_index=self.index,
+ sw_if_index=self.itf.sw_if_index,
+ )
+
+ def wait_event(self, expect, timeout=5):
+ rv = self._test.vapi.wait_for_event(timeout, "wireguard_peer_event")
+ self._test.assertEqual(rv.flags, expect)
+ self._test.assertEqual(rv.peer_index, self.index)
class TestWg(VppTestCase):
- """ Wireguard Test Case """
+ """Wireguard Test Case"""
error_str = compile(r"Error")
- 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/'
+ 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_error = wg4_input_node_name + "Peer error"
+ 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_error = wg6_input_node_name + "Peer error"
+ 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"
@classmethod
def setUpClass(cls):
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_err = self.statistics.get_err_counter(self.peer4_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_err = self.statistics.get_err_counter(self.peer6_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
+ )
def test_wg_interface(self):
- """ Simple interface creation """
+ """Simple interface creation"""
port = 12312
# Create interface
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
self.logger.info(self.vapi.cli("sh int"))
wg0.remove_vpp_config()
def test_handshake_hash(self):
- """ test hashing an init message """
+ """test hashing an init message"""
# a init packet generated by linux given the key below
- h = "0100000098b9032b" \
- "55cc4b39e73c3d24" \
- "a2a1ab884b524a81" \
- "1808bb86640fb70d" \
- "e93154fec1879125" \
- "ab012624a27f0b75" \
- "c0a2582f438ddb5f" \
- "8e768af40b4ab444" \
- "02f9ff473e1b797e" \
- "80d39d93c5480c82" \
- "a3d4510f70396976" \
- "586fb67300a5167b" \
- "ae6ca3ff3dfd00eb" \
- "59be198810f5aa03" \
- "6abc243d2155ee4f" \
- "2336483900aef801" \
- "08752cd700000000" \
- "0000000000000000" \
+ h = (
+ "0100000098b9032b"
+ "55cc4b39e73c3d24"
+ "a2a1ab884b524a81"
+ "1808bb86640fb70d"
+ "e93154fec1879125"
+ "ab012624a27f0b75"
+ "c0a2582f438ddb5f"
+ "8e768af40b4ab444"
+ "02f9ff473e1b797e"
+ "80d39d93c5480c82"
+ "a3d4510f70396976"
+ "586fb67300a5167b"
+ "ae6ca3ff3dfd00eb"
+ "59be198810f5aa03"
+ "6abc243d2155ee4f"
+ "2336483900aef801"
+ "08752cd700000000"
+ "0000000000000000"
"00000000"
+ )
b = bytearray.fromhex(h)
tgt = Wireguard(b)
# strip the macs and build a new packet
init = b[0:-32]
- mac_key = blake2s(b'mac1----' + public_key_bytes(pub)).digest()
- init += blake2s(init,
- digest_size=16,
- key=mac_key).digest()
- init += b'\x00' * 16
+ mac_key = blake2s(b"mac1----" + public_key_bytes(pub)).digest()
+ init += blake2s(init, digest_size=16, key=mac_key).digest()
+ init += b"\x00" * 16
act = Wireguard(init)
self.assertEqual(tgt, act)
+ def _test_wg_send_cookie_tmpl(self, is_resp, is_ip6):
+ port = 12323
+
+ # create wg interface
+ if is_ip6:
+ wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
+ wg0.admin_up()
+ wg0.config_ip6()
+ else:
+ 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()
+
+ # create a peer
+ if is_ip6:
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
+ ).add_vpp_config()
+ else:
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
+ ).add_vpp_config()
+ self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
+
+ if is_resp:
+ # prepare and send a handshake initiation
+ # expect the peer to send a handshake response
+ init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
+ rxs = self.send_and_expect(self.pg1, [init], self.pg1)
+ else:
+ # wait for the peer to send a handshake initiation
+ rxs = self.pg1.get_capture(1, timeout=2)
+
+ # prepare and send a wrong cookie reply
+ # expect no replies and the cookie error incremented
+ cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
+ cookie.nonce = b"1234567890"
+ self.send_and_assert_no_replies(self.pg1, [cookie], timeout=0.1)
+ if is_ip6:
+ self.assertEqual(
+ self.base_cookie_dec6_err + 1,
+ self.statistics.get_err_counter(self.cookie_dec6_err),
+ )
+ else:
+ self.assertEqual(
+ self.base_cookie_dec4_err + 1,
+ self.statistics.get_err_counter(self.cookie_dec4_err),
+ )
+
+ # prepare and send a correct cookie reply
+ cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
+ self.pg_send(self.pg1, [cookie])
+
+ # wait for the peer to send a handshake initiation with mac2 set
+ rxs = self.pg1.get_capture(1, timeout=6)
+
+ # verify the initiation and its mac2
+ peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6, is_mac2=True)
+
+ # remove configs
+ peer_1.remove_vpp_config()
+ wg0.remove_vpp_config()
+
+ def test_wg_send_cookie_on_init_v4(self):
+ """Send cookie on handshake initiation (v4)"""
+ self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=False)
+
+ def test_wg_send_cookie_on_init_v6(self):
+ """Send cookie on handshake initiation (v6)"""
+ self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=True)
+
+ def test_wg_send_cookie_on_resp_v4(self):
+ """Send cookie on handshake response (v4)"""
+ self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=False)
+
+ def test_wg_send_cookie_on_resp_v6(self):
+ """Send cookie on handshake response (v6)"""
+ self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=True)
+
def test_wg_peer_resp(self):
- """ Send handshake response """
+ """Send handshake response"""
port = 12323
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
+ 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.3.0/24"]).add_vpp_config()
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip4, port + 1, ["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()
+ r1 = VppIpRoute(
+ self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
+ ).add_vpp_config()
# wait for the peer to send a handshake
rx = self.pg1.get_capture(1, timeout=2)
self.assertEqual(0, len(b))
# send a packets that are routed into the tunnel
- p = (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))
+ p = (
+ 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, p * 255, self.pg1)
# send packets into the tunnel, expect to receive them on
# the other side
- p = [(peer_1.mk_tunnel_header(self.pg1) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- receiver_index=peer_1.sender,
- counter=ii,
- 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)]
+ p = [
+ (
+ peer_1.mk_tunnel_header(self.pg1)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender,
+ counter=ii,
+ 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)
wg0.remove_vpp_config()
def test_wg_peer_v4o4(self):
- """ Test v4o4"""
+ """Test v4o4"""
port = 12333
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip4()
- peer_1 = VppWgPeer(self,
- wg0,
- self.pg1.remote_ip4,
- port+1,
- ["10.11.3.0/24"]).add_vpp_config()
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip4, port + 1, ["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()
+ r1 = VppIpRoute(
+ self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
+ ).add_vpp_config()
+ r2 = VppIpRoute(
+ self, "20.22.3.0", 24, [VppRoutePath("20.22.3.1", wg0.sw_if_index)]
+ ).add_vpp_config()
# route a packet into the wg interface
# use the allowed-ip prefix
# this is dropped because the peer is not initiated
- p = (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())
+ p = (
+ 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()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp4_err + 1,
- self.statistics.get_err_counter(self.kp4_error))
+ self.assertEqual(
+ self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
+ )
+
+ # route a packet into the wg interface
+ # use a not allowed-ip prefix
+ # this is dropped because there is no matching peer
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_ip4, dst="20.22.3.2")
+ / UDP(sport=555, dport=556)
+ / Raw()
+ )
+ self.send_and_assert_no_replies(self.pg0, [p])
+ self.assertEqual(
+ self.base_peer4_out_err + 1,
+ self.statistics.get_err_counter(self.peer4_out_err),
+ )
# send a handsake from the peer with an invalid MAC
p = peer_1.mk_handshake(self.pg1)
- p[WireguardInitiation].mac1 = b'foobar'
+ p[WireguardInitiation].mac1 = b"foobar"
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_mac4_err + 1,
- self.statistics.get_err_counter(self.mac4_error))
+ self.assertEqual(
+ self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
+ )
# send a handsake from the peer but signed by the wrong key.
- p = peer_1.mk_handshake(self.pg1,
- False,
- X25519PrivateKey.generate().public_key())
+ p = peer_1.mk_handshake(
+ self.pg1, False, X25519PrivateKey.generate().public_key()
+ )
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_peer4_err + 1,
- self.statistics.get_err_counter(self.peer4_error))
+ self.assertEqual(
+ self.base_peer4_in_err + 1,
+ self.statistics.get_err_counter(self.peer4_in_err),
+ )
# send a valid handsake init for which we expect a response
p = peer_1.mk_handshake(self.pg1)
# route a packet into the wg interface
# this is dropped because the peer is still not initiated
- p = (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())
+ p = (
+ 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()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp4_err + 2,
- self.statistics.get_err_counter(self.kp4_error))
+ self.assertEqual(
+ self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
+ )
# send a data packet from the peer through the tunnel
# this completes the handshake
- p = (IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20) /
- UDP(sport=222, dport=223) /
- Raw())
+ 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) /
- (Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(receiver_index=peer_1.sender,
- counter=0,
- encrypted_encapsulated_packet=d)))
+ p = peer_1.mk_tunnel_header(self.pg1) / (
+ Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
+ )
+ )
rxs = self.send_and_expect(self.pg1, [p], self.pg0)
for rx in rxs:
self.assertEqual(rx[IP].ttl, 19)
# send a packets that are routed into the tunnel
- p = (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))
+ p = (
+ 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, p * 255, self.pg1)
# chech the oringial packet is present
self.assertEqual(rx[IP].dst, p[IP].dst)
- self.assertEqual(rx[IP].ttl, p[IP].ttl-1)
+ self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
# send packets into the tunnel, expect to receive them on
# the other side
- p = [(peer_1.mk_tunnel_header(self.pg1) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- 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)]
+ p = [
+ (
+ peer_1.mk_tunnel_header(self.pg1)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ 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)
self.assertEqual(rx[IP].ttl, 19)
r1.remove_vpp_config()
+ r2.remove_vpp_config()
peer_1.remove_vpp_config()
wg0.remove_vpp_config()
def test_wg_peer_v6o6(self):
- """ Test v6o6"""
+ """Test v6o6"""
port = 12343
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip6,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip6()
- peer_1 = VppWgPeer(self,
- wg0,
- self.pg1.remote_ip6,
- port+1,
- ["1::3:0/112"]).add_vpp_config(True)
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
+ ).add_vpp_config(True)
self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
- r1 = VppIpRoute(self, "1::3:0", 112,
- [VppRoutePath("1::3:1",
- wg0.sw_if_index)]).add_vpp_config()
+ r1 = VppIpRoute(
+ self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
+ ).add_vpp_config()
+ r2 = VppIpRoute(
+ self, "22::3:0", 112, [VppRoutePath("22::3:1", wg0.sw_if_index)]
+ ).add_vpp_config()
# route a packet into the wg interface
# use the allowed-ip prefix
# this is dropped because the peer is not initiated
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst="1::3:2") /
- UDP(sport=555, dport=556) /
- Raw())
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp6_err + 1,
- self.statistics.get_err_counter(self.kp6_error))
+ self.assertEqual(
+ self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
+ )
+
+ # route a packet into the wg interface
+ # use a not allowed-ip prefix
+ # this is dropped because there is no matching peer
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="22::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw()
+ )
+ self.send_and_assert_no_replies(self.pg0, [p])
+ self.assertEqual(
+ self.base_peer6_out_err + 1,
+ self.statistics.get_err_counter(self.peer6_out_err),
+ )
# send a handsake from the peer with an invalid MAC
p = peer_1.mk_handshake(self.pg1, True)
- p[WireguardInitiation].mac1 = b'foobar'
+ p[WireguardInitiation].mac1 = b"foobar"
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_mac6_err + 1,
- self.statistics.get_err_counter(self.mac6_error))
+ self.assertEqual(
+ self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
+ )
# send a handsake from the peer but signed by the wrong key.
- p = peer_1.mk_handshake(self.pg1,
- True,
- X25519PrivateKey.generate().public_key())
+ p = peer_1.mk_handshake(
+ self.pg1, True, X25519PrivateKey.generate().public_key()
+ )
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_peer6_err + 1,
- self.statistics.get_err_counter(self.peer6_error))
+ self.assertEqual(
+ self.base_peer6_in_err + 1,
+ self.statistics.get_err_counter(self.peer6_in_err),
+ )
# send a valid handsake init for which we expect a response
p = peer_1.mk_handshake(self.pg1, True)
# route a packet into the wg interface
# this is dropped because the peer is still not initiated
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst="1::3:2") /
- UDP(sport=555, dport=556) /
- Raw())
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp6_err + 2,
- self.statistics.get_err_counter(self.kp6_error))
+ self.assertEqual(
+ self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
+ )
# send a data packet from the peer through the tunnel
# this completes the handshake
- p = (IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20) /
- UDP(sport=222, dport=223) /
- Raw())
+ p = (
+ IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
+ / UDP(sport=222, dport=223)
+ / Raw()
+ )
d = peer_1.encrypt_transport(p)
- p = (peer_1.mk_tunnel_header(self.pg1, True) /
- (Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(receiver_index=peer_1.sender,
- counter=0,
- encrypted_encapsulated_packet=d)))
+ p = peer_1.mk_tunnel_header(self.pg1, True) / (
+ Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
+ )
+ )
rxs = self.send_and_expect(self.pg1, [p], self.pg0)
for rx in rxs:
self.assertEqual(rx[IPv6].hlim, 19)
# send a packets that are routed into the tunnel
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst="1::3:2") /
- UDP(sport=555, dport=556) /
- Raw(b'\x00' * 80))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw(b"\x00" * 80)
+ )
rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
# chech the oringial packet is present
self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
- self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim-1)
+ self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
# send packets into the tunnel, expect to receive them on
# the other side
- p = [(peer_1.mk_tunnel_header(self.pg1, True) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- receiver_index=peer_1.sender,
- counter=ii+1,
- encrypted_encapsulated_packet=peer_1.encrypt_transport(
- (IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20) /
- UDP(sport=222, dport=223) /
- Raw())))) for ii in range(255)]
+ p = [
+ (
+ peer_1.mk_tunnel_header(self.pg1, True)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender,
+ counter=ii + 1,
+ encrypted_encapsulated_packet=peer_1.encrypt_transport(
+ (
+ IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
+ / UDP(sport=222, dport=223)
+ / Raw()
+ )
+ ),
+ )
+ )
+ for ii in range(255)
+ ]
rxs = self.send_and_expect(self.pg1, p, self.pg0)
self.assertEqual(rx[IPv6].hlim, 19)
r1.remove_vpp_config()
+ r2.remove_vpp_config()
peer_1.remove_vpp_config()
wg0.remove_vpp_config()
def test_wg_peer_v6o4(self):
- """ Test v6o4"""
+ """Test v6o4"""
port = 12353
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip6()
- peer_1 = VppWgPeer(self,
- wg0,
- self.pg1.remote_ip4,
- port+1,
- ["1::3:0/112"]).add_vpp_config(True)
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip4, port + 1, ["1::3:0/112"]
+ ).add_vpp_config(True)
self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
- r1 = VppIpRoute(self, "1::3:0", 112,
- [VppRoutePath("1::3:1",
- wg0.sw_if_index)]).add_vpp_config()
+ r1 = VppIpRoute(
+ self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
+ ).add_vpp_config()
# route a packet into the wg interface
# use the allowed-ip prefix
# this is dropped because the peer is not initiated
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst="1::3:2") /
- UDP(sport=555, dport=556) /
- Raw())
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp6_err + 1,
- self.statistics.get_err_counter(self.kp6_error))
+ self.assertEqual(
+ self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
+ )
# send a handsake from the peer with an invalid MAC
p = peer_1.mk_handshake(self.pg1)
- p[WireguardInitiation].mac1 = b'foobar'
+ p[WireguardInitiation].mac1 = b"foobar"
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_mac4_err + 1,
- self.statistics.get_err_counter(self.mac4_error))
+ self.assertEqual(
+ self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
+ )
# send a handsake from the peer but signed by the wrong key.
- p = peer_1.mk_handshake(self.pg1,
- False,
- X25519PrivateKey.generate().public_key())
+ p = peer_1.mk_handshake(
+ self.pg1, False, X25519PrivateKey.generate().public_key()
+ )
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_peer4_err + 1,
- self.statistics.get_err_counter(self.peer4_error))
+ self.assertEqual(
+ self.base_peer4_in_err + 1,
+ self.statistics.get_err_counter(self.peer4_in_err),
+ )
# send a valid handsake init for which we expect a response
p = peer_1.mk_handshake(self.pg1)
# route a packet into the wg interface
# this is dropped because the peer is still not initiated
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst="1::3:2") /
- UDP(sport=555, dport=556) /
- Raw())
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp6_err + 2,
- self.statistics.get_err_counter(self.kp6_error))
+ self.assertEqual(
+ self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
+ )
# send a data packet from the peer through the tunnel
# this completes the handshake
- p = (IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20) /
- UDP(sport=222, dport=223) /
- Raw())
+ p = (
+ IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
+ / UDP(sport=222, dport=223)
+ / Raw()
+ )
d = peer_1.encrypt_transport(p)
- p = (peer_1.mk_tunnel_header(self.pg1) /
- (Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(receiver_index=peer_1.sender,
- counter=0,
- encrypted_encapsulated_packet=d)))
+ p = peer_1.mk_tunnel_header(self.pg1) / (
+ Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
+ )
+ )
rxs = self.send_and_expect(self.pg1, [p], self.pg0)
for rx in rxs:
self.assertEqual(rx[IPv6].hlim, 19)
# send a packets that are routed into the tunnel
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst="1::3:2") /
- UDP(sport=555, dport=556) /
- Raw(b'\x00' * 80))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
+ / UDP(sport=555, dport=556)
+ / Raw(b"\x00" * 80)
+ )
rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
# chech the oringial packet is present
self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
- self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim-1)
+ self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
# send packets into the tunnel, expect to receive them on
# the other side
- p = [(peer_1.mk_tunnel_header(self.pg1) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- receiver_index=peer_1.sender,
- counter=ii+1,
- encrypted_encapsulated_packet=peer_1.encrypt_transport(
- (IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20) /
- UDP(sport=222, dport=223) /
- Raw())))) for ii in range(255)]
+ p = [
+ (
+ peer_1.mk_tunnel_header(self.pg1)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender,
+ counter=ii + 1,
+ encrypted_encapsulated_packet=peer_1.encrypt_transport(
+ (
+ IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
+ / UDP(sport=222, dport=223)
+ / Raw()
+ )
+ ),
+ )
+ )
+ for ii in range(255)
+ ]
rxs = self.send_and_expect(self.pg1, p, self.pg0)
wg0.remove_vpp_config()
def test_wg_peer_v4o6(self):
- """ Test v4o6"""
+ """Test v4o6"""
port = 12363
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip6,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip4()
- peer_1 = VppWgPeer(self,
- wg0,
- self.pg1.remote_ip6,
- port+1,
- ["10.11.3.0/24"]).add_vpp_config()
+ peer_1 = VppWgPeer(
+ self, wg0, self.pg1.remote_ip6, port + 1, ["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()
+ r1 = VppIpRoute(
+ self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
+ ).add_vpp_config()
# route a packet into the wg interface
# use the allowed-ip prefix
# this is dropped because the peer is not initiated
- p = (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())
+ p = (
+ 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()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp4_err + 1,
- self.statistics.get_err_counter(self.kp4_error))
+ self.assertEqual(
+ self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
+ )
# send a handsake from the peer with an invalid MAC
p = peer_1.mk_handshake(self.pg1, True)
- p[WireguardInitiation].mac1 = b'foobar'
+ p[WireguardInitiation].mac1 = b"foobar"
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_mac6_err + 1,
- self.statistics.get_err_counter(self.mac6_error))
+ self.assertEqual(
+ self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
+ )
# send a handsake from the peer but signed by the wrong key.
- p = peer_1.mk_handshake(self.pg1,
- True,
- X25519PrivateKey.generate().public_key())
+ p = peer_1.mk_handshake(
+ self.pg1, True, X25519PrivateKey.generate().public_key()
+ )
self.send_and_assert_no_replies(self.pg1, [p])
- self.assertEqual(self.base_peer6_err + 1,
- self.statistics.get_err_counter(self.peer6_error))
+ self.assertEqual(
+ self.base_peer6_in_err + 1,
+ self.statistics.get_err_counter(self.peer6_in_err),
+ )
# send a valid handsake init for which we expect a response
p = peer_1.mk_handshake(self.pg1, True)
# route a packet into the wg interface
# this is dropped because the peer is still not initiated
- p = (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())
+ p = (
+ 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()
+ )
self.send_and_assert_no_replies(self.pg0, [p])
- self.assertEqual(self.base_kp4_err + 2,
- self.statistics.get_err_counter(self.kp4_error))
+ self.assertEqual(
+ self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
+ )
# send a data packet from the peer through the tunnel
# this completes the handshake
- p = (IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20) /
- UDP(sport=222, dport=223) /
- Raw())
+ 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, True) /
- (Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(receiver_index=peer_1.sender,
- counter=0,
- encrypted_encapsulated_packet=d)))
+ p = peer_1.mk_tunnel_header(self.pg1, True) / (
+ Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
+ )
+ )
rxs = self.send_and_expect(self.pg1, [p], self.pg0)
for rx in rxs:
self.assertEqual(rx[IP].ttl, 19)
# send a packets that are routed into the tunnel
- p = (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))
+ p = (
+ 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, p * 255, self.pg1)
# chech the oringial packet is present
self.assertEqual(rx[IP].dst, p[IP].dst)
- self.assertEqual(rx[IP].ttl, p[IP].ttl-1)
+ self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
# send packets into the tunnel, expect to receive them on
# the other side
- p = [(peer_1.mk_tunnel_header(self.pg1, True) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- 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)]
+ p = [
+ (
+ peer_1.mk_tunnel_header(self.pg1, True)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ 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)
wg0.remove_vpp_config()
def test_wg_multi_peer(self):
- """ multiple peer setup """
+ """multiple peer setup"""
port = 12373
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
- wg1 = VppWgInterface(self,
- self.pg2.local_ip4,
- port+1).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
+ wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
wg0.admin_up()
wg1.admin_up()
routes_1 = []
routes_2 = []
for i in range(NUM_PEERS):
- peers_1.append(VppWgPeer(self,
- wg0,
- self.pg1.remote_hosts[i].ip4,
- port+1+i,
- ["10.0.%d.4/32" % i]).add_vpp_config())
- routes_1.append(VppIpRoute(self, "10.0.%d.4" % i, 32,
- [VppRoutePath(self.pg1.remote_hosts[i].ip4,
- wg0.sw_if_index)]).add_vpp_config())
-
- peers_2.append(VppWgPeer(self,
- wg1,
- self.pg2.remote_hosts[i].ip4,
- port+100+i,
- ["10.100.%d.4/32" % i]).add_vpp_config())
- routes_2.append(VppIpRoute(self, "10.100.%d.4" % i, 32,
- [VppRoutePath(self.pg2.remote_hosts[i].ip4,
- wg1.sw_if_index)]).add_vpp_config())
-
- self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS*2)
+ peers_1.append(
+ VppWgPeer(
+ self,
+ wg0,
+ self.pg1.remote_hosts[i].ip4,
+ port + 1 + i,
+ ["10.0.%d.4/32" % i],
+ ).add_vpp_config()
+ )
+ routes_1.append(
+ VppIpRoute(
+ self,
+ "10.0.%d.4" % i,
+ 32,
+ [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
+ ).add_vpp_config()
+ )
+
+ peers_2.append(
+ VppWgPeer(
+ self,
+ wg1,
+ self.pg2.remote_hosts[i].ip4,
+ port + 100 + i,
+ ["10.100.%d.4/32" % i],
+ ).add_vpp_config()
+ )
+ routes_2.append(
+ VppIpRoute(
+ self,
+ "10.100.%d.4" % i,
+ 32,
+ [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
+ ).add_vpp_config()
+ )
+
+ self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
self.logger.info(self.vapi.cli("show wireguard peer"))
self.logger.info(self.vapi.cli("show wireguard interface"))
wg1.remove_vpp_config()
def test_wg_multi_interface(self):
- """ Multi-tunnel on the same port """
+ """Multi-tunnel on the same port"""
port = 12500
# Create many wireguard interfaces
wg_ifs = []
for i in range(NUM_IFS):
# Use the same port for each interface
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip4()
wg_ifs.append(wg0)
- peers.append(VppWgPeer(self,
- wg0,
- self.pg1.remote_hosts[i].ip4,
- port+1+i,
- ["10.0.%d.0/24" % i]).add_vpp_config())
-
- routes.append(VppIpRoute(self, "10.0.%d.0" % i, 24,
- [VppRoutePath("10.0.%d.4" % i,
- wg0.sw_if_index)]).add_vpp_config())
+ peers.append(
+ VppWgPeer(
+ self,
+ wg0,
+ self.pg1.remote_hosts[i].ip4,
+ port + 1 + i,
+ ["10.0.%d.0/24" % i],
+ ).add_vpp_config()
+ )
+
+ routes.append(
+ VppIpRoute(
+ self,
+ "10.0.%d.0" % i,
+ 24,
+ [VppRoutePath("10.0.%d.4" % i, wg0.sw_if_index)],
+ ).add_vpp_config()
+ )
self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_IFS)
# send a data packet from the peer through the tunnel
# this completes the handshake
- p = (IP(src="10.0.%d.4" % i,
- dst=self.pg0.remote_hosts[i].ip4, ttl=20) /
- UDP(sport=222, dport=223) /
- Raw())
+ p = (
+ IP(src="10.0.%d.4" % i, dst=self.pg0.remote_hosts[i].ip4, ttl=20)
+ / UDP(sport=222, dport=223)
+ / Raw()
+ )
d = peers[i].encrypt_transport(p)
- p = (peers[i].mk_tunnel_header(self.pg1) /
- (Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(receiver_index=peers[i].sender,
- counter=0,
- encrypted_encapsulated_packet=d)))
+ p = peers[i].mk_tunnel_header(self.pg1) / (
+ Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peers[i].sender,
+ counter=0,
+ encrypted_encapsulated_packet=d,
+ )
+ )
rxs = self.send_and_expect(self.pg1, [p], self.pg0)
for rx in rxs:
self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
# send a packets that are routed into the tunnel
for i in range(NUM_IFS):
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_hosts[i].ip4, dst="10.0.%d.4" % i) /
- UDP(sport=555, dport=556) /
- Raw(b'\x00' * 80))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_hosts[i].ip4, dst="10.0.%d.4" % i)
+ / UDP(sport=555, dport=556)
+ / Raw(b"\x00" * 80)
+ )
rxs = self.send_and_expect(self.pg0, p * 64, self.pg1)
# check the oringial packet is present
self.assertEqual(rx[IP].dst, p[IP].dst)
- self.assertEqual(rx[IP].ttl, p[IP].ttl-1)
+ self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
# send packets into the tunnel
for i in range(NUM_IFS):
- p = [(peers[i].mk_tunnel_header(self.pg1) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- receiver_index=peers[i].sender,
- counter=ii+1,
- encrypted_encapsulated_packet=peers[i].encrypt_transport(
- (IP(src="10.0.%d.4" % i,
- dst=self.pg0.remote_hosts[i].ip4, ttl=20) /
- UDP(sport=222, dport=223) /
- Raw())))) for ii in range(64)]
+ p = [
+ (
+ peers[i].mk_tunnel_header(self.pg1)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ receiver_index=peers[i].sender,
+ counter=ii + 1,
+ encrypted_encapsulated_packet=peers[i].encrypt_transport(
+ (
+ IP(
+ src="10.0.%d.4" % i,
+ dst=self.pg0.remote_hosts[i].ip4,
+ ttl=20,
+ )
+ / UDP(sport=222, dport=223)
+ / Raw()
+ )
+ ),
+ )
+ )
+ for ii in range(64)
+ ]
rxs = self.send_and_expect(self.pg1, p, self.pg0)
for i in wg_ifs:
i.remove_vpp_config()
+ def test_wg_event(self):
+ """Test events"""
+ port = 12600
+ ESTABLISHED_FLAG = (
+ VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_ESTABLISHED
+ )
+ DEAD_FLAG = VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_STATUS_DEAD
+
+ # Create interfaces
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
+ wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
+ wg0.admin_up()
+ wg1.admin_up()
+
+ # Check peer counter
+ self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Create peers
+ NUM_PEERS = 2
+ self.pg2.generate_remote_hosts(NUM_PEERS)
+ self.pg2.configure_ipv4_neighbors()
+ self.pg1.generate_remote_hosts(NUM_PEERS)
+ self.pg1.configure_ipv4_neighbors()
+
+ peers_0 = []
+ peers_1 = []
+ routes_0 = []
+ routes_1 = []
+ for i in range(NUM_PEERS):
+ peers_0.append(
+ VppWgPeer(
+ self,
+ wg0,
+ self.pg1.remote_hosts[i].ip4,
+ port + 1 + i,
+ ["10.0.%d.4/32" % i],
+ ).add_vpp_config()
+ )
+ routes_0.append(
+ VppIpRoute(
+ self,
+ "10.0.%d.4" % i,
+ 32,
+ [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
+ ).add_vpp_config()
+ )
+
+ peers_1.append(
+ VppWgPeer(
+ self,
+ wg1,
+ self.pg2.remote_hosts[i].ip4,
+ port + 100 + i,
+ ["10.100.%d.4/32" % i],
+ ).add_vpp_config()
+ )
+ routes_1.append(
+ VppIpRoute(
+ self,
+ "10.100.%d.4" % i,
+ 32,
+ [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
+ ).add_vpp_config()
+ )
+
+ self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
+
+ # Want events from the first perr of wg0
+ # and from all wg1 peers
+ peers_0[0].want_events()
+ wg1.want_events()
+
+ for i in range(NUM_PEERS):
+ # send a valid handsake init for which we expect a response
+ p = peers_0[i].mk_handshake(self.pg1)
+ rx = self.send_and_expect(self.pg1, [p], self.pg1)
+ peers_0[i].consume_response(rx[0])
+ if i == 0:
+ peers_0[0].wait_event(ESTABLISHED_FLAG)
+
+ p = peers_1[i].mk_handshake(self.pg2)
+ rx = self.send_and_expect(self.pg2, [p], self.pg2)
+ peers_1[i].consume_response(rx[0])
+
+ wg1.wait_events(ESTABLISHED_FLAG, [peers_1[0].index, peers_1[1].index])
+
+ # remove routes
+ for r in routes_0:
+ r.remove_vpp_config()
+ for r in routes_1:
+ r.remove_vpp_config()
+
+ # remove peers
+ for i in range(NUM_PEERS):
+ self.assertTrue(peers_0[i].query_vpp_config())
+ peers_0[i].remove_vpp_config()
+ if i == 0:
+ peers_0[i].wait_event(0)
+ peers_0[i].wait_event(DEAD_FLAG)
+ for p in peers_1:
+ self.assertTrue(p.query_vpp_config())
+ p.remove_vpp_config()
+ p.wait_event(0)
+ p.wait_event(DEAD_FLAG)
+
+ wg0.remove_vpp_config()
+ wg1.remove_vpp_config()
+
class WireguardHandoffTests(TestWg):
- """ Wireguard Tests in multi worker setup """
+ """Wireguard Tests in multi worker setup"""
+
vpp_worker_count = 2
def test_wg_peer_init(self):
- """ Handoff """
+ """Handoff"""
port = 12383
# Create interfaces
- wg0 = VppWgInterface(self,
- self.pg1.local_ip4,
- port).add_vpp_config()
+ wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
wg0.admin_up()
wg0.config_ip4()
- peer_1 = VppWgPeer(self,
- wg0,
- self.pg1.remote_ip4,
- port+1,
- ["10.11.2.0/24",
- "10.11.3.0/24"]).add_vpp_config()
+ 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()
+ r1 = VppIpRoute(
+ self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
+ ).add_vpp_config()
# send a valid handsake init for which we expect a response
p = peer_1.mk_handshake(self.pg1)
# 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())
+ 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) /
- (Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(receiver_index=peer_1.sender,
- counter=0,
- encrypted_encapsulated_packet=d)))
- rxs = self.send_and_expect(self.pg1, [p], self.pg0,
- worker=0)
+ p = peer_1.mk_tunnel_header(self.pg1) / (
+ Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ 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)
# 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))
+ 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) /
- Wireguard(message_type=4, reserved_zero=0) /
- WireguardTransport(
- 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)]
+ p = [
+ (
+ peer_1.mk_tunnel_header(self.pg1)
+ / Wireguard(message_type=4, reserved_zero=0)
+ / WireguardTransport(
+ 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)
@unittest.skip("test disabled")
def test_wg_multi_interface(self):
- """ Multi-tunnel on the same port """
+ """Multi-tunnel on the same port"""
Code Review / vpp.git / blobdiff