X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_wireguard.py;h=7395402e27f7170e7da2ea348a3821f1fa9902a2;hb=44ec846f4;hp=65ebd8d2c08a5d40967a8e40216d1e350d308b72;hpb=7dd3b5b5e37a4019ae335296ba9c4bd1e465fd17;p=vpp.git diff --git a/test/test_wireguard.py b/test/test_wireguard.py old mode 100755 new mode 100644 index 65ebd8d2c08..7395402e27f --- a/test/test_wireguard.py +++ b/test/test_wireguard.py @@ -3,6 +3,7 @@ import datetime import base64 +import os from hashlib import blake2s from scapy.packet import Packet @@ -10,21 +11,36 @@ from scapy.packet import Raw 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 @@ -37,14 +53,16 @@ Wg test. 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): @@ -67,31 +85,48 @@ 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() @@ -103,8 +138,7 @@ def find_route(test, prefix, is_ip6, table_id=0): 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 @@ -114,14 +148,7 @@ NOISE_IDENTIFIER_NAME = b"WireGuard v1 zx2c4 Jason@zx2c4.com" 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 @@ -133,18 +160,24 @@ class VppWgPeer(VppObject): 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) @@ -154,7 +187,7 @@ class VppWgPeer(VppObject): 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) @@ -163,11 +196,13 @@ class VppWgPeer(VppObject): 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() @@ -177,18 +212,19 @@ class VppWgPeer(VppObject): 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) @@ -199,20 +235,69 @@ class VppWgPeer(VppObject): # 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 @@ -221,8 +306,11 @@ class VppWgPeer(VppObject): # 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 @@ -231,14 +319,13 @@ class VppWgPeer(VppObject): 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 @@ -253,7 +340,7 @@ class VppWgPeer(VppObject): 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) @@ -265,33 +352,40 @@ class VppWgPeer(VppObject): 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 @@ -303,13 +397,14 @@ class VppWgPeer(VppObject): 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): @@ -318,11 +413,11 @@ class VppWgPeer(VppObject): 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): @@ -335,30 +430,47 @@ class VppWgPeer(VppObject): # 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): @@ -384,19 +496,25 @@ class TestWg(VppTestCase): 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")) @@ -404,27 +522,29 @@ class TestWg(VppTestCase): 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) @@ -436,40 +556,116 @@ class TestWg(VppTestCase): # 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) @@ -486,10 +682,12 @@ class TestWg(VppTestCase): 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) @@ -497,15 +695,24 @@ class TestWg(VppTestCase): # 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) @@ -518,53 +725,73 @@ class TestWg(VppTestCase): 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) @@ -575,25 +802,31 @@ class TestWg(VppTestCase): # 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: @@ -601,10 +834,12 @@ class TestWg(VppTestCase): 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) @@ -613,19 +848,28 @@ class TestWg(VppTestCase): # 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) @@ -634,60 +878,81 @@ class TestWg(VppTestCase): 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) @@ -698,25 +963,31 @@ class TestWg(VppTestCase): # 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: @@ -724,10 +995,12 @@ class TestWg(VppTestCase): 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) @@ -736,19 +1009,28 @@ class TestWg(VppTestCase): # 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) @@ -757,58 +1039,61 @@ class TestWg(VppTestCase): 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) @@ -819,25 +1104,31 @@ class TestWg(VppTestCase): # 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: @@ -845,10 +1136,12 @@ class TestWg(VppTestCase): 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) @@ -857,19 +1150,28 @@ class TestWg(VppTestCase): # 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) @@ -882,53 +1184,55 @@ class TestWg(VppTestCase): 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) @@ -939,25 +1243,31 @@ class TestWg(VppTestCase): # 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: @@ -965,10 +1275,12 @@ class TestWg(VppTestCase): 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) @@ -977,19 +1289,28 @@ class TestWg(VppTestCase): # 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) @@ -1002,16 +1323,12 @@ class TestWg(VppTestCase): 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() @@ -1033,25 +1350,43 @@ class TestWg(VppTestCase): 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")) @@ -1077,7 +1412,7 @@ class TestWg(VppTestCase): 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 @@ -1093,21 +1428,28 @@ class TestWg(VppTestCase): 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) @@ -1119,16 +1461,20 @@ class TestWg(VppTestCase): # 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) @@ -1136,10 +1482,12 @@ class TestWg(VppTestCase): # 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) @@ -1148,20 +1496,32 @@ class TestWg(VppTestCase): # 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) @@ -1176,34 +1536,141 @@ class TestWg(VppTestCase): 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) @@ -1214,17 +1681,19 @@ class WireguardHandoffTests(TestWg): # 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) @@ -1232,23 +1701,34 @@ class WireguardHandoffTests(TestWg): # 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) @@ -1268,4 +1748,4 @@ class WireguardHandoffTests(TestWg): @unittest.skip("test disabled") def test_wg_multi_interface(self): - """ Multi-tunnel on the same port """ + """Multi-tunnel on the same port"""