test/test_wireguard.py

   1 #!/usr/bin/env python3
   2 """ Wg tests """
   3
   4 import datetime
   5 import base64
   6 import os
   7 import struct
   8
   9 from hashlib import blake2s
  10 from config import config
  11 from scapy.packet import Raw
  12 from scapy.layers.l2 import Ether
  13 from scapy.layers.inet import IP, UDP
  14 from scapy.layers.inet6 import IPv6
  15 from scapy.layers.vxlan import VXLAN
  16 from scapy.contrib.wireguard import (
  17     Wireguard,
  18     WireguardResponse,
  19     WireguardInitiation,
  20     WireguardTransport,
  21     WireguardCookieReply,
  22 )
  23 from cryptography.hazmat.primitives.asymmetric.x25519 import (
  24     X25519PrivateKey,
  25     X25519PublicKey,
  26 )
  27 from cryptography.hazmat.primitives.serialization import (
  28     Encoding,
  29     PrivateFormat,
  30     PublicFormat,
  31     NoEncryption,
  32 )
  33 from noise.connection import NoiseConnection, Keypair
  34
  35 from Crypto.Cipher import ChaCha20_Poly1305
  36 from Crypto.Random import get_random_bytes
  37
  38 from vpp_interface import VppInterface
  39 from vpp_pg_interface import is_ipv6_misc
  40 from vpp_ip_route import VppIpRoute, VppRoutePath
  41 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
  42 from vpp_vxlan_tunnel import VppVxlanTunnel
  43 from vpp_object import VppObject
  44 from vpp_papi import VppEnum
  45 from asfframework import tag_run_solo, tag_fixme_vpp_debug
  46 from framework import VppTestCase
  47 from re import compile
  48 import unittest
  49
  50 """ TestWg is a subclass of  VPPTestCase classes.
  51
  52 Wg test.
  53
  54 """
  55
  56
  57 def private_key_bytes(k):
  58     return k.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())
  59
  60
  61 def public_key_bytes(k):
  62     return k.public_bytes(Encoding.Raw, PublicFormat.Raw)
  63
  64
  65 def get_field_bytes(pkt, name):
  66     fld, val = pkt.getfield_and_val(name)
  67     return fld.i2m(pkt, val)
  68
  69
  70 class VppWgInterface(VppInterface):
  71     """
  72     VPP WireGuard interface
  73     """
  74
  75     def __init__(self, test, src, port):
  76         super(VppWgInterface, self).__init__(test)
  77
  78         self.port = port
  79         self.src = src
  80         self.private_key = X25519PrivateKey.generate()
  81         self.public_key = self.private_key.public_key()
  82
  83         # cookie related params
  84         self.cookie_key = blake2s(b"cookie--" + self.public_key_bytes()).digest()
  85
  86     def public_key_bytes(self):
  87         return public_key_bytes(self.public_key)
  88
  89     def private_key_bytes(self):
  90         return private_key_bytes(self.private_key)
  91
  92     def add_vpp_config(self):
  93         r = self.test.vapi.wireguard_interface_create(
  94             interface={
  95                 "user_instance": 0xFFFFFFFF,
  96                 "port": self.port,
  97                 "src_ip": self.src,
  98                 "private_key": private_key_bytes(self.private_key),
  99                 "generate_key": False,
 100             }
 101         )
 102         self.set_sw_if_index(r.sw_if_index)
 103         self.test.registry.register(self, self.test.logger)
 104         return self
 105
 106     def remove_vpp_config(self):
 107         self.test.vapi.wireguard_interface_delete(sw_if_index=self._sw_if_index)
 108
 109     def query_vpp_config(self):
 110         ts = self.test.vapi.wireguard_interface_dump(sw_if_index=0xFFFFFFFF)
 111         for t in ts:
 112             if (
 113                 t.interface.sw_if_index == self._sw_if_index
 114                 and str(t.interface.src_ip) == self.src
 115                 and t.interface.port == self.port
 116                 and t.interface.private_key == private_key_bytes(self.private_key)
 117             ):
 118                 return True
 119         return False
 120
 121     def want_events(self, peer_index=0xFFFFFFFF):
 122         self.test.vapi.want_wireguard_peer_events(
 123             enable_disable=1,
 124             pid=os.getpid(),
 125             sw_if_index=self._sw_if_index,
 126             peer_index=peer_index,
 127         )
 128
 129     def wait_events(self, expect, peers, timeout=5):
 130         for i in range(len(peers)):
 131             rv = self.test.vapi.wait_for_event(timeout, "wireguard_peer_event")
 132             self.test.assertEqual(rv.peer_index, peers[i])
 133             self.test.assertEqual(rv.flags, expect)
 134
 135     def __str__(self):
 136         return self.object_id()
 137
 138     def object_id(self):
 139         return "wireguard-%d" % self._sw_if_index
 140
 141
 142 NOISE_HANDSHAKE_NAME = b"Noise_IKpsk2_25519_ChaChaPoly_BLAKE2s"
 143 NOISE_IDENTIFIER_NAME = b"WireGuard v1 zx2c4 Jason@zx2c4.com"
 144
 145 HANDSHAKE_COUNTING_INTERVAL = 0.5
 146 UNDER_LOAD_INTERVAL = 1.0
 147 HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD = 40
 148 HANDSHAKE_NUM_BEFORE_RATELIMITING = 5
 149
 150 HANDSHAKE_JITTER = 0.5
 151
 152
 153 class VppWgPeer(VppObject):
 154     def __init__(self, test, itf, endpoint, port, allowed_ips, persistent_keepalive=15):
 155         self._test = test
 156         self.itf = itf
 157         self.endpoint = endpoint
 158         self.port = port
 159         self.allowed_ips = allowed_ips
 160         self.persistent_keepalive = persistent_keepalive
 161
 162         # remote peer's public
 163         self.private_key = X25519PrivateKey.generate()
 164         self.public_key = self.private_key.public_key()
 165
 166         # cookie related params
 167         self.cookie_key = blake2s(b"cookie--" + self.public_key_bytes()).digest()
 168         self.last_sent_cookie = None
 169         self.last_mac1 = None
 170         self.last_received_cookie = None
 171
 172         self.noise = NoiseConnection.from_name(NOISE_HANDSHAKE_NAME)
 173
 174     def change_endpoint(self, endpoint, port):
 175         self.endpoint = endpoint
 176         self.port = port
 177
 178     def add_vpp_config(self):
 179         rv = self._test.vapi.wireguard_peer_add(
 180             peer={
 181                 "public_key": self.public_key_bytes(),
 182                 "port": self.port,
 183                 "endpoint": self.endpoint,
 184                 "n_allowed_ips": len(self.allowed_ips),
 185                 "allowed_ips": self.allowed_ips,
 186                 "sw_if_index": self.itf.sw_if_index,
 187                 "persistent_keepalive": self.persistent_keepalive,
 188             }
 189         )
 190         self.index = rv.peer_index
 191         self.receiver_index = self.index + 1
 192         self._test.registry.register(self, self._test.logger)
 193         return self
 194
 195     def remove_vpp_config(self):
 196         self._test.vapi.wireguard_peer_remove(peer_index=self.index)
 197
 198     def object_id(self):
 199         return "wireguard-peer-%s" % self.index
 200
 201     def public_key_bytes(self):
 202         return public_key_bytes(self.public_key)
 203
 204     def query_vpp_config(self):
 205         peers = self._test.vapi.wireguard_peers_dump()
 206
 207         for p in peers:
 208             # "::" endpoint will be returned as "0.0.0.0" in peer's details
 209             endpoint = "0.0.0.0" if self.endpoint == "::" else self.endpoint
 210             if (
 211                 p.peer.public_key == self.public_key_bytes()
 212                 and p.peer.port == self.port
 213                 and str(p.peer.endpoint) == endpoint
 214                 and p.peer.sw_if_index == self.itf.sw_if_index
 215                 and len(self.allowed_ips) == p.peer.n_allowed_ips
 216             ):
 217                 self.allowed_ips.sort()
 218                 p.peer.allowed_ips.sort()
 219
 220                 for a1, a2 in zip(self.allowed_ips, p.peer.allowed_ips):
 221                     if str(a1) != str(a2):
 222                         return False
 223                 return True
 224         return False
 225
 226     def mk_tunnel_header(self, tx_itf, is_ip6=False):
 227         if is_ip6 is False:
 228             return (
 229                 Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
 230                 / IP(src=self.endpoint, dst=self.itf.src)
 231                 / UDP(sport=self.port, dport=self.itf.port)
 232             )
 233         else:
 234             return (
 235                 Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
 236                 / IPv6(src=self.endpoint, dst=self.itf.src)
 237                 / UDP(sport=self.port, dport=self.itf.port)
 238             )
 239
 240     def noise_reset(self):
 241         self.noise = NoiseConnection.from_name(NOISE_HANDSHAKE_NAME)
 242
 243     def noise_init(self, public_key=None):
 244         self.noise.set_prologue(NOISE_IDENTIFIER_NAME)
 245         self.noise.set_psks(psk=bytes(bytearray(32)))
 246
 247         if not public_key:
 248             public_key = self.itf.public_key
 249
 250         # local/this private
 251         self.noise.set_keypair_from_private_bytes(
 252             Keypair.STATIC, private_key_bytes(self.private_key)
 253         )
 254         # remote's public
 255         self.noise.set_keypair_from_public_bytes(
 256             Keypair.REMOTE_STATIC, public_key_bytes(public_key)
 257         )
 258
 259         self.noise.start_handshake()
 260
 261     def mk_cookie(self, p, tx_itf, is_resp=False, is_ip6=False):
 262         self.verify_header(p, is_ip6)
 263
 264         wg_pkt = Wireguard(p[Raw])
 265
 266         if is_resp:
 267             self._test.assertEqual(wg_pkt[Wireguard].message_type, 2)
 268             self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
 269             self._test.assertEqual(wg_pkt[WireguardResponse].mac2, bytes([0] * 16))
 270         else:
 271             self._test.assertEqual(wg_pkt[Wireguard].message_type, 1)
 272             self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
 273             self._test.assertEqual(wg_pkt[WireguardInitiation].mac2, bytes([0] * 16))
 274
 275         # collect info from wg packet (initiation or response)
 276         src = get_field_bytes(p[IPv6 if is_ip6 else IP], "src")
 277         sport = p[UDP].sport.to_bytes(2, byteorder="big")
 278         if is_resp:
 279             mac1 = wg_pkt[WireguardResponse].mac1
 280             sender_index = wg_pkt[WireguardResponse].sender_index
 281         else:
 282             mac1 = wg_pkt[WireguardInitiation].mac1
 283             sender_index = wg_pkt[WireguardInitiation].sender_index
 284
 285         # make cookie reply
 286         cookie_reply = Wireguard() / WireguardCookieReply()
 287         cookie_reply[Wireguard].message_type = 3
 288         cookie_reply[Wireguard].reserved_zero = 0
 289         cookie_reply[WireguardCookieReply].receiver_index = sender_index
 290         nonce = get_random_bytes(24)
 291         cookie_reply[WireguardCookieReply].nonce = nonce
 292
 293         # generate cookie data
 294         changing_secret = get_random_bytes(32)
 295         self.last_sent_cookie = blake2s(
 296             src + sport, digest_size=16, key=changing_secret
 297         ).digest()
 298
 299         # encrypt cookie data
 300         cipher = ChaCha20_Poly1305.new(key=self.cookie_key, nonce=nonce)
 301         cipher.update(mac1)
 302         ciphertext, tag = cipher.encrypt_and_digest(self.last_sent_cookie)
 303         cookie_reply[WireguardCookieReply].encrypted_cookie = ciphertext + tag
 304
 305         # prepare cookie reply to be sent
 306         cookie_reply = self.mk_tunnel_header(tx_itf, is_ip6) / cookie_reply
 307
 308         return cookie_reply
 309
 310     def consume_cookie(self, p, is_ip6=False):
 311         self.verify_header(p, is_ip6)
 312
 313         cookie_reply = Wireguard(p[Raw])
 314
 315         self._test.assertEqual(cookie_reply[Wireguard].message_type, 3)
 316         self._test.assertEqual(cookie_reply[Wireguard].reserved_zero, 0)
 317         self._test.assertEqual(
 318             cookie_reply[WireguardCookieReply].receiver_index, self.receiver_index
 319         )
 320
 321         # collect info from cookie reply
 322         nonce = cookie_reply[WireguardCookieReply].nonce
 323         encrypted_cookie = cookie_reply[WireguardCookieReply].encrypted_cookie
 324         ciphertext, tag = encrypted_cookie[:16], encrypted_cookie[16:]
 325
 326         # decrypt cookie data
 327         cipher = ChaCha20_Poly1305.new(key=self.itf.cookie_key, nonce=nonce)
 328         cipher.update(self.last_mac1)
 329         self.last_received_cookie = cipher.decrypt_and_verify(ciphertext, tag)
 330
 331     def mk_handshake(self, tx_itf, is_ip6=False, public_key=None):
 332         self.noise.set_as_initiator()
 333         self.noise_init(public_key)
 334
 335         p = Wireguard() / WireguardInitiation()
 336
 337         p[Wireguard].message_type = 1
 338         p[Wireguard].reserved_zero = 0
 339         p[WireguardInitiation].sender_index = self.receiver_index
 340
 341         # some random data for the message
 342         #  lifted from the noise protocol's wireguard example
 343         now = datetime.datetime.now()
 344         tai = struct.pack(
 345             "!qi",
 346             4611686018427387914 + int(now.timestamp()),
 347             int(now.microsecond * 1e3),
 348         )
 349         b = self.noise.write_message(payload=tai)
 350
 351         # load noise into init message
 352         p[WireguardInitiation].unencrypted_ephemeral = b[0:32]
 353         p[WireguardInitiation].encrypted_static = b[32:80]
 354         p[WireguardInitiation].encrypted_timestamp = b[80:108]
 355
 356         # generate the mac1 hash
 357         mac_key = blake2s(b"mac1----" + self.itf.public_key_bytes()).digest()
 358         mac1 = blake2s(bytes(p)[0:116], digest_size=16, key=mac_key).digest()
 359         p[WireguardInitiation].mac1 = mac1
 360         self.last_mac1 = mac1
 361
 362         # generate the mac2 hash
 363         if self.last_received_cookie:
 364             mac2 = blake2s(
 365                 bytes(p)[0:132], digest_size=16, key=self.last_received_cookie
 366             ).digest()
 367             p[WireguardInitiation].mac2 = mac2
 368             self.last_received_cookie = None
 369         else:
 370             p[WireguardInitiation].mac2 = bytearray(16)
 371
 372         p = self.mk_tunnel_header(tx_itf, is_ip6) / p
 373
 374         return p
 375
 376     def verify_header(self, p, is_ip6=False):
 377         if is_ip6 is False:
 378             self._test.assertEqual(p[IP].src, self.itf.src)
 379             self._test.assertEqual(p[IP].dst, self.endpoint)
 380             self._test.assert_packet_checksums_valid(p)
 381         else:
 382             self._test.assertEqual(p[IPv6].src, self.itf.src)
 383             self._test.assertEqual(p[IPv6].dst, self.endpoint)
 384             self._test.assert_packet_checksums_valid(p, False)
 385         self._test.assertEqual(p[UDP].sport, self.itf.port)
 386         self._test.assertEqual(p[UDP].dport, self.port)
 387
 388     def consume_init(self, p, tx_itf, is_ip6=False, is_mac2=False):
 389         self.noise.set_as_responder()
 390         self.noise_init(self.itf.public_key)
 391         self.verify_header(p, is_ip6)
 392
 393         init = Wireguard(p[Raw])
 394
 395         self._test.assertEqual(init[Wireguard].message_type, 1)
 396         self._test.assertEqual(init[Wireguard].reserved_zero, 0)
 397
 398         self.sender = init[WireguardInitiation].sender_index
 399
 400         # validate the mac1 hash
 401         mac_key = blake2s(b"mac1----" + public_key_bytes(self.public_key)).digest()
 402         mac1 = blake2s(bytes(init)[0:-32], digest_size=16, key=mac_key).digest()
 403         self._test.assertEqual(init[WireguardInitiation].mac1, mac1)
 404
 405         # validate the mac2 hash
 406         if is_mac2:
 407             self._test.assertNotEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
 408             self._test.assertNotEqual(self.last_sent_cookie, None)
 409             mac2 = blake2s(
 410                 bytes(init)[0:-16], digest_size=16, key=self.last_sent_cookie
 411             ).digest()
 412             self._test.assertEqual(init[WireguardInitiation].mac2, mac2)
 413             self.last_sent_cookie = None
 414         else:
 415             self._test.assertEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
 416
 417         # this passes only unencrypted_ephemeral, encrypted_static,
 418         # encrypted_timestamp fields of the init
 419         payload = self.noise.read_message(bytes(init)[8:-32])
 420
 421         # build the response
 422         b = self.noise.write_message()
 423         mac_key = blake2s(b"mac1----" + public_key_bytes(self.itf.public_key)).digest()
 424         resp = Wireguard(message_type=2, reserved_zero=0) / WireguardResponse(
 425             sender_index=self.receiver_index,
 426             receiver_index=self.sender,
 427             unencrypted_ephemeral=b[0:32],
 428             encrypted_nothing=b[32:],
 429         )
 430         mac1 = blake2s(bytes(resp)[:-32], digest_size=16, key=mac_key).digest()
 431         resp[WireguardResponse].mac1 = mac1
 432         self.last_mac1 = mac1
 433
 434         resp = self.mk_tunnel_header(tx_itf, is_ip6) / resp
 435         self._test.assertTrue(self.noise.handshake_finished)
 436
 437         return resp
 438
 439     def consume_response(self, p, is_ip6=False):
 440         self.verify_header(p, is_ip6)
 441
 442         resp = Wireguard(p[Raw])
 443
 444         self._test.assertEqual(resp[Wireguard].message_type, 2)
 445         self._test.assertEqual(resp[Wireguard].reserved_zero, 0)
 446         self._test.assertEqual(
 447             resp[WireguardResponse].receiver_index, self.receiver_index
 448         )
 449
 450         self.sender = resp[Wireguard].sender_index
 451
 452         payload = self.noise.read_message(bytes(resp)[12:60])
 453         self._test.assertEqual(payload, b"")
 454         self._test.assertTrue(self.noise.handshake_finished)
 455
 456     def decrypt_transport(self, p, is_ip6=False):
 457         self.verify_header(p, is_ip6)
 458
 459         p = Wireguard(p[Raw])
 460         self._test.assertEqual(p[Wireguard].message_type, 4)
 461         self._test.assertEqual(p[Wireguard].reserved_zero, 0)
 462         self._test.assertEqual(
 463             p[WireguardTransport].receiver_index, self.receiver_index
 464         )
 465
 466         d = self.noise.decrypt(p[WireguardTransport].encrypted_encapsulated_packet)
 467         return d
 468
 469     def encrypt_transport(self, p):
 470         return self.noise.encrypt(bytes(p))
 471
 472     def validate_encapped(self, rxs, tx, is_tunnel_ip6=False, is_transport_ip6=False):
 473         ret_rxs = []
 474         for rx in rxs:
 475             rx = self.decrypt_transport(rx, is_tunnel_ip6)
 476             if is_transport_ip6 is False:
 477                 rx = IP(rx)
 478                 # check the original packet is present
 479                 self._test.assertEqual(rx[IP].dst, tx[IP].dst)
 480                 self._test.assertEqual(rx[IP].ttl, tx[IP].ttl - 1)
 481             else:
 482                 rx = IPv6(rx)
 483                 # check the original packet is present
 484                 self._test.assertEqual(rx[IPv6].dst, tx[IPv6].dst)
 485                 self._test.assertEqual(rx[IPv6].hlim, tx[IPv6].hlim - 1)
 486             ret_rxs.append(rx)
 487         return ret_rxs
 488
 489     def want_events(self):
 490         self._test.vapi.want_wireguard_peer_events(
 491             enable_disable=1,
 492             pid=os.getpid(),
 493             peer_index=self.index,
 494             sw_if_index=self.itf.sw_if_index,
 495         )
 496
 497     def wait_event(self, expect, timeout=5):
 498         rv = self._test.vapi.wait_for_event(timeout, "wireguard_peer_event")
 499         self._test.assertEqual(rv.flags, expect)
 500         self._test.assertEqual(rv.peer_index, self.index)
 501
 502
 503 def is_handshake_init(p):
 504     wg_p = Wireguard(p[Raw])
 505
 506     return wg_p[Wireguard].message_type == 1
 507
 508
 509 @unittest.skipIf(
 510     "wireguard" in config.excluded_plugins, "Exclude Wireguard plugin tests"
 511 )
 512 @tag_run_solo
 513 class TestWg(VppTestCase):
 514     """Wireguard Test Case"""
 515
 516     error_str = compile(r"Error")
 517
 518     wg4_output_node_name = "/err/wg4-output-tun/"
 519     wg4_input_node_name = "/err/wg4-input/"
 520     wg6_output_node_name = "/err/wg6-output-tun/"
 521     wg6_input_node_name = "/err/wg6-input/"
 522     kp4_error = wg4_output_node_name + "Keypair error"
 523     mac4_error = wg4_input_node_name + "Invalid MAC handshake"
 524     peer4_in_err = wg4_input_node_name + "Peer error"
 525     peer4_out_err = wg4_output_node_name + "Peer error"
 526     kp6_error = wg6_output_node_name + "Keypair error"
 527     mac6_error = wg6_input_node_name + "Invalid MAC handshake"
 528     peer6_in_err = wg6_input_node_name + "Peer error"
 529     peer6_out_err = wg6_output_node_name + "Peer error"
 530     cookie_dec4_err = wg4_input_node_name + "Failed during Cookie decryption"
 531     cookie_dec6_err = wg6_input_node_name + "Failed during Cookie decryption"
 532     ratelimited4_err = wg4_input_node_name + "Handshake ratelimited"
 533     ratelimited6_err = wg6_input_node_name + "Handshake ratelimited"
 534
 535     @classmethod
 536     def setUpClass(cls):
 537         super(TestWg, cls).setUpClass()
 538         try:
 539             cls.create_pg_interfaces(range(3))
 540             for i in cls.pg_interfaces:
 541                 i.admin_up()
 542                 i.config_ip4()
 543                 i.config_ip6()
 544                 i.resolve_arp()
 545                 i.resolve_ndp()
 546
 547         except Exception:
 548             super(TestWg, cls).tearDownClass()
 549             raise
 550
 551     @classmethod
 552     def tearDownClass(cls):
 553         super(TestWg, cls).tearDownClass()
 554
 555     def setUp(self):
 556         super(VppTestCase, self).setUp()
 557         self.base_kp4_err = self.statistics.get_err_counter(self.kp4_error)
 558         self.base_mac4_err = self.statistics.get_err_counter(self.mac4_error)
 559         self.base_peer4_in_err = self.statistics.get_err_counter(self.peer4_in_err)
 560         self.base_peer4_out_err = self.statistics.get_err_counter(self.peer4_out_err)
 561         self.base_kp6_err = self.statistics.get_err_counter(self.kp6_error)
 562         self.base_mac6_err = self.statistics.get_err_counter(self.mac6_error)
 563         self.base_peer6_in_err = self.statistics.get_err_counter(self.peer6_in_err)
 564         self.base_peer6_out_err = self.statistics.get_err_counter(self.peer6_out_err)
 565         self.base_cookie_dec4_err = self.statistics.get_err_counter(
 566             self.cookie_dec4_err
 567         )
 568         self.base_cookie_dec6_err = self.statistics.get_err_counter(
 569             self.cookie_dec6_err
 570         )
 571         self.base_ratelimited4_err = self.statistics.get_err_counter(
 572             self.ratelimited4_err
 573         )
 574         self.base_ratelimited6_err = self.statistics.get_err_counter(
 575             self.ratelimited6_err
 576         )
 577
 578     def send_and_assert_no_replies_ignoring_init(
 579         self, intf, pkts, remark="", timeout=None
 580     ):
 581         self.pg_send(intf, pkts)
 582
 583         def _filter_out_fn(p):
 584             return is_ipv6_misc(p) or is_handshake_init(p)
 585
 586         try:
 587             if not timeout:
 588                 timeout = 1
 589             for i in self.pg_interfaces:
 590                 i.assert_nothing_captured(
 591                     timeout=timeout, remark=remark, filter_out_fn=_filter_out_fn
 592                 )
 593                 timeout = 0.1
 594         finally:
 595             pass
 596
 597     def test_wg_interface(self):
 598         """Simple interface creation"""
 599         port = 12312
 600
 601         # Create interface
 602         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 603
 604         self.logger.info(self.vapi.cli("sh int"))
 605
 606         # delete interface
 607         wg0.remove_vpp_config()
 608
 609     def test_handshake_hash(self):
 610         """test hashing an init message"""
 611         # a init packet generated by linux given the key below
 612         h = (
 613             "0100000098b9032b"
 614             "55cc4b39e73c3d24"
 615             "a2a1ab884b524a81"
 616             "1808bb86640fb70d"
 617             "e93154fec1879125"
 618             "ab012624a27f0b75"
 619             "c0a2582f438ddb5f"
 620             "8e768af40b4ab444"
 621             "02f9ff473e1b797e"
 622             "80d39d93c5480c82"
 623             "a3d4510f70396976"
 624             "586fb67300a5167b"
 625             "ae6ca3ff3dfd00eb"
 626             "59be198810f5aa03"
 627             "6abc243d2155ee4f"
 628             "2336483900aef801"
 629             "08752cd700000000"
 630             "0000000000000000"
 631             "00000000"
 632         )
 633
 634         b = bytearray.fromhex(h)
 635         tgt = Wireguard(b)
 636
 637         pubb = base64.b64decode("aRuHFTTxICIQNefp05oKWlJv3zgKxb8+WW7JJMh0jyM=")
 638         pub = X25519PublicKey.from_public_bytes(pubb)
 639
 640         self.assertEqual(pubb, public_key_bytes(pub))
 641
 642         # strip the macs and build a new packet
 643         init = b[0:-32]
 644         mac_key = blake2s(b"mac1----" + public_key_bytes(pub)).digest()
 645         init += blake2s(init, digest_size=16, key=mac_key).digest()
 646         init += b"\x00" * 16
 647
 648         act = Wireguard(init)
 649
 650         self.assertEqual(tgt, act)
 651
 652     def _test_wg_send_cookie_tmpl(self, is_resp, is_ip6):
 653         port = 12323
 654
 655         # create wg interface
 656         if is_ip6:
 657             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 658             wg0.admin_up()
 659             wg0.config_ip6()
 660         else:
 661             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 662             wg0.admin_up()
 663             wg0.config_ip4()
 664
 665         self.pg_enable_capture(self.pg_interfaces)
 666         self.pg_start()
 667
 668         # create a peer
 669         if is_ip6:
 670             peer_1 = VppWgPeer(
 671                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 672             ).add_vpp_config()
 673         else:
 674             peer_1 = VppWgPeer(
 675                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 676             ).add_vpp_config()
 677         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 678
 679         if is_resp:
 680             # skip the first automatic handshake
 681             self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 682
 683             # prepare and send a handshake initiation
 684             # expect the peer to send a handshake response
 685             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 686             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 687         else:
 688             # wait for the peer to send a handshake initiation
 689             rxs = self.pg1.get_capture(1, timeout=2)
 690
 691         # prepare and send a wrong cookie reply
 692         # expect no replies and the cookie error incremented
 693         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 694         cookie.nonce = b"1234567890"
 695         self.send_and_assert_no_replies(self.pg1, [cookie], timeout=0.1)
 696         if is_ip6:
 697             self.assertEqual(
 698                 self.base_cookie_dec6_err + 1,
 699                 self.statistics.get_err_counter(self.cookie_dec6_err),
 700             )
 701         else:
 702             self.assertEqual(
 703                 self.base_cookie_dec4_err + 1,
 704                 self.statistics.get_err_counter(self.cookie_dec4_err),
 705             )
 706
 707         # prepare and send a correct cookie reply
 708         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 709         self.pg_send(self.pg1, [cookie])
 710
 711         # wait for the peer to send a handshake initiation with mac2 set
 712         rxs = self.pg1.get_capture(1, timeout=6)
 713
 714         # verify the initiation and its mac2
 715         peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6, is_mac2=True)
 716
 717         # remove configs
 718         peer_1.remove_vpp_config()
 719         wg0.remove_vpp_config()
 720
 721     def test_wg_send_cookie_on_init_v4(self):
 722         """Send cookie on handshake initiation (v4)"""
 723         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=False)
 724
 725     def test_wg_send_cookie_on_init_v6(self):
 726         """Send cookie on handshake initiation (v6)"""
 727         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=True)
 728
 729     def test_wg_send_cookie_on_resp_v4(self):
 730         """Send cookie on handshake response (v4)"""
 731         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=False)
 732
 733     def test_wg_send_cookie_on_resp_v6(self):
 734         """Send cookie on handshake response (v6)"""
 735         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=True)
 736
 737     def _test_wg_receive_cookie_tmpl(self, is_resp, is_ip6):
 738         port = 12323
 739
 740         # create wg interface
 741         if is_ip6:
 742             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 743             wg0.admin_up()
 744             wg0.config_ip6()
 745         else:
 746             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 747             wg0.admin_up()
 748             wg0.config_ip4()
 749
 750         self.pg_enable_capture(self.pg_interfaces)
 751         self.pg_start()
 752
 753         # create a peer
 754         if is_ip6:
 755             peer_1 = VppWgPeer(
 756                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 757             ).add_vpp_config()
 758         else:
 759             peer_1 = VppWgPeer(
 760                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 761             ).add_vpp_config()
 762         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 763
 764         if is_resp:
 765             # wait for the peer to send a handshake initiation
 766             rxs = self.pg1.get_capture(1, timeout=2)
 767             # prepare and send a bunch of handshake responses
 768             # expect to switch to under load state
 769             resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
 770             txs = [resp] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 771             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 772             # reset noise to be able to turn into initiator later
 773             peer_1.noise_reset()
 774         else:
 775             # skip the first automatic handshake
 776             self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 777
 778             # prepare and send a bunch of handshake initiations
 779             # expect to switch to under load state
 780             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 781             txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 782             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 783
 784         # expect the peer to send a cookie reply
 785         peer_1.consume_cookie(rxs[-1], is_ip6=is_ip6)
 786
 787         # prepare and send a handshake initiation with wrong mac2
 788         # expect a cookie reply
 789         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 790         init.mac2 = b"1234567890"
 791         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 792         peer_1.consume_cookie(rxs[0], is_ip6=is_ip6)
 793
 794         # prepare and send a handshake initiation with correct mac2
 795         # expect a handshake response
 796         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 797         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 798
 799         # verify the response
 800         peer_1.consume_response(rxs[0], is_ip6=is_ip6)
 801
 802         # clear up under load state
 803         self.sleep(UNDER_LOAD_INTERVAL)
 804
 805         # remove configs
 806         peer_1.remove_vpp_config()
 807         wg0.remove_vpp_config()
 808
 809     def test_wg_receive_cookie_on_init_v4(self):
 810         """Receive cookie on handshake initiation (v4)"""
 811         self._test_wg_receive_cookie_tmpl(is_resp=False, is_ip6=False)
 812
 813     def test_wg_receive_cookie_on_init_v6(self):
 814         """Receive cookie on handshake initiation (v6)"""
 815         self._test_wg_receive_cookie_tmpl(is_resp=False, is_ip6=True)
 816
 817     def test_wg_receive_cookie_on_resp_v4(self):
 818         """Receive cookie on handshake response (v4)"""
 819         self._test_wg_receive_cookie_tmpl(is_resp=True, is_ip6=False)
 820
 821     def test_wg_receive_cookie_on_resp_v6(self):
 822         """Receive cookie on handshake response (v6)"""
 823         self._test_wg_receive_cookie_tmpl(is_resp=True, is_ip6=True)
 824
 825     def test_wg_under_load_interval(self):
 826         """Under load interval"""
 827         port = 12323
 828
 829         # create wg interface
 830         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 831         wg0.admin_up()
 832         wg0.config_ip4()
 833
 834         self.pg_enable_capture(self.pg_interfaces)
 835         self.pg_start()
 836
 837         # create a peer
 838         peer_1 = VppWgPeer(
 839             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 840         ).add_vpp_config()
 841         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 842
 843         # skip the first automatic handshake
 844         self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 845
 846         # prepare and send a bunch of handshake initiations
 847         # expect to switch to under load state
 848         init = peer_1.mk_handshake(self.pg1)
 849         txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 850         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 851
 852         # expect the peer to send a cookie reply
 853         peer_1.consume_cookie(rxs[-1])
 854
 855         # sleep till the next counting interval
 856         # expect under load state is still active
 857         self.sleep(HANDSHAKE_COUNTING_INTERVAL)
 858
 859         # prepare and send a handshake initiation with wrong mac2
 860         # expect a cookie reply
 861         init = peer_1.mk_handshake(self.pg1)
 862         init.mac2 = b"1234567890"
 863         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 864         peer_1.consume_cookie(rxs[0])
 865
 866         # sleep till the end of being under load
 867         # expect under load state is over
 868         self.sleep(UNDER_LOAD_INTERVAL - HANDSHAKE_COUNTING_INTERVAL)
 869
 870         # prepare and send a handshake initiation with wrong mac2
 871         # expect a handshake response
 872         init = peer_1.mk_handshake(self.pg1)
 873         init.mac2 = b"1234567890"
 874         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 875
 876         # verify the response
 877         peer_1.consume_response(rxs[0])
 878
 879         # remove configs
 880         peer_1.remove_vpp_config()
 881         wg0.remove_vpp_config()
 882
 883     def _test_wg_handshake_ratelimiting_tmpl(self, is_ip6):
 884         port = 12323
 885
 886         # create wg interface
 887         if is_ip6:
 888             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 889             wg0.admin_up()
 890             wg0.config_ip6()
 891         else:
 892             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 893             wg0.admin_up()
 894             wg0.config_ip4()
 895
 896         self.pg_enable_capture(self.pg_interfaces)
 897         self.pg_start()
 898
 899         # create a peer
 900         if is_ip6:
 901             peer_1 = VppWgPeer(
 902                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 903             ).add_vpp_config()
 904         else:
 905             peer_1 = VppWgPeer(
 906                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 907             ).add_vpp_config()
 908         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 909
 910         # skip the first automatic handshake
 911         self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 912
 913         # prepare and send a bunch of handshake initiations
 914         # expect to switch to under load state
 915         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 916         txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 917         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 918
 919         # expect the peer to send a cookie reply
 920         peer_1.consume_cookie(rxs[-1], is_ip6=is_ip6)
 921
 922         # prepare and send a bunch of handshake initiations with correct mac2
 923         # expect a handshake response and then ratelimiting
 924         NUM_TO_REJECT = 10
 925         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 926         txs = [init] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + NUM_TO_REJECT)
 927
 928         # TODO: Deterimine why no handshake response is sent back if test is
 929         #       not run in as part of the test suite.  It fails only very occasionally
 930         #       when run solo.
 931         #
 932         #       Until then, if no response, don't fail trying to verify it.
 933         #       The error counter test still verifies that the correct number of
 934         #       handshake initiaions are ratelimited.
 935         try:
 936             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 937         except:
 938             self.logger.debug(
 939                 f"{self._testMethodDoc}: send_and_expect_some() failed to get any response packets."
 940             )
 941             rxs = None
 942             pass
 943
 944         if is_ip6:
 945             self.assertEqual(
 946                 self.base_ratelimited6_err + NUM_TO_REJECT,
 947                 self.statistics.get_err_counter(self.ratelimited6_err),
 948             )
 949         else:
 950             self.assertEqual(
 951                 self.base_ratelimited4_err + NUM_TO_REJECT,
 952                 self.statistics.get_err_counter(self.ratelimited4_err),
 953             )
 954
 955         # verify the response
 956         if rxs is not None:
 957             peer_1.consume_response(rxs[0], is_ip6=is_ip6)
 958
 959         # clear up under load state
 960         self.sleep(UNDER_LOAD_INTERVAL)
 961
 962         # remove configs
 963         peer_1.remove_vpp_config()
 964         wg0.remove_vpp_config()
 965
 966     def test_wg_handshake_ratelimiting_v4(self):
 967         """Handshake ratelimiting (v4)"""
 968         self._test_wg_handshake_ratelimiting_tmpl(is_ip6=False)
 969
 970     def test_wg_handshake_ratelimiting_v6(self):
 971         """Handshake ratelimiting (v6)"""
 972         self._test_wg_handshake_ratelimiting_tmpl(is_ip6=True)
 973
 974     def test_wg_handshake_ratelimiting_multi_peer(self):
 975         """Handshake ratelimiting (multiple peer)"""
 976         port = 12323
 977
 978         # create wg interface
 979         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 980         wg0.admin_up()
 981         wg0.config_ip4()
 982
 983         self.pg_enable_capture(self.pg_interfaces)
 984         self.pg_start()
 985
 986         # create two peers
 987         NUM_PEERS = 2
 988         self.pg1.generate_remote_hosts(NUM_PEERS)
 989         self.pg1.configure_ipv4_neighbors()
 990
 991         peer_1 = VppWgPeer(
 992             self, wg0, self.pg1.remote_hosts[0].ip4, port + 1, ["10.11.3.0/24"]
 993         ).add_vpp_config()
 994         peer_2 = VppWgPeer(
 995             self, wg0, self.pg1.remote_hosts[1].ip4, port + 1, ["10.11.4.0/24"]
 996         ).add_vpp_config()
 997         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 2)
 998
 999         # skip the first automatic handshake
1000         self.pg1.get_capture(NUM_PEERS, timeout=HANDSHAKE_JITTER)
1001
1002         # (peer_1) prepare and send a bunch of handshake initiations
1003         # expect not to switch to under load state
1004         init_1 = peer_1.mk_handshake(self.pg1)
1005         txs = [init_1] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
1006         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
1007
1008         # (peer_1) expect the peer to send a handshake response
1009         peer_1.consume_response(rxs[0])
1010         peer_1.noise_reset()
1011
1012         # (peer_1) send another bunch of handshake initiations
1013         # expect to switch to under load state
1014         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
1015
1016         # (peer_1) expect the peer to send a cookie reply
1017         peer_1.consume_cookie(rxs[-1])
1018
1019         # (peer_2) prepare and send a handshake initiation
1020         # expect a cookie reply
1021         init_2 = peer_2.mk_handshake(self.pg1)
1022         rxs = self.send_and_expect(self.pg1, [init_2], self.pg1)
1023         peer_2.consume_cookie(rxs[0])
1024
1025         # (peer_1) (peer_2) prepare and send a bunch of handshake initiations with correct mac2
1026         # expect a handshake response and then ratelimiting
1027         PEER_1_NUM_TO_REJECT = 2
1028         PEER_2_NUM_TO_REJECT = 5
1029         init_1 = peer_1.mk_handshake(self.pg1)
1030         txs = [init_1] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + PEER_1_NUM_TO_REJECT)
1031         init_2 = peer_2.mk_handshake(self.pg1)
1032         txs += [init_2] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + PEER_2_NUM_TO_REJECT)
1033         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
1034
1035         self.assertTrue(
1036             self.base_ratelimited4_err + PEER_1_NUM_TO_REJECT
1037             < self.statistics.get_err_counter(self.ratelimited4_err)
1038             <= self.base_ratelimited4_err + PEER_1_NUM_TO_REJECT + PEER_2_NUM_TO_REJECT
1039         )
1040
1041         # (peer_1) (peer_2) verify the response
1042         peer_1.consume_response(rxs[0])
1043         peer_2.consume_response(rxs[1])
1044
1045         # clear up under load state
1046         self.sleep(UNDER_LOAD_INTERVAL)
1047
1048         # remove configs
1049         peer_1.remove_vpp_config()
1050         peer_2.remove_vpp_config()
1051         wg0.remove_vpp_config()
1052
1053     def _test_wg_peer_roaming_on_handshake_tmpl(self, is_endpoint_set, is_resp, is_ip6):
1054         port = 12323
1055
1056         # create wg interface
1057         if is_ip6:
1058             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1059             wg0.admin_up()
1060             wg0.config_ip6()
1061         else:
1062             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1063             wg0.admin_up()
1064             wg0.config_ip4()
1065
1066         self.pg_enable_capture(self.pg_interfaces)
1067         self.pg_start()
1068
1069         # create more remote hosts
1070         NUM_REMOTE_HOSTS = 2
1071         self.pg1.generate_remote_hosts(NUM_REMOTE_HOSTS)
1072         if is_ip6:
1073             self.pg1.configure_ipv6_neighbors()
1074         else:
1075             self.pg1.configure_ipv4_neighbors()
1076
1077         # create a peer
1078         if is_ip6:
1079             peer_1 = VppWgPeer(
1080                 test=self,
1081                 itf=wg0,
1082                 endpoint=self.pg1.remote_hosts[0].ip6 if is_endpoint_set else "::",
1083                 port=port + 1 if is_endpoint_set else 0,
1084                 allowed_ips=["1::3:0/112"],
1085             ).add_vpp_config()
1086         else:
1087             peer_1 = VppWgPeer(
1088                 test=self,
1089                 itf=wg0,
1090                 endpoint=self.pg1.remote_hosts[0].ip4 if is_endpoint_set else "0.0.0.0",
1091                 port=port + 1 if is_endpoint_set else 0,
1092                 allowed_ips=["10.11.3.0/24"],
1093             ).add_vpp_config()
1094         self.assertTrue(peer_1.query_vpp_config())
1095
1096         if is_resp:
1097             # wait for the peer to send a handshake initiation
1098             rxs = self.pg1.get_capture(1, timeout=2)
1099             # prepare a handshake response
1100             resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
1101             # change endpoint
1102             if is_ip6:
1103                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1104                 resp[IPv6].src, resp[UDP].sport = peer_1.endpoint, peer_1.port
1105             else:
1106                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1107                 resp[IP].src, resp[UDP].sport = peer_1.endpoint, peer_1.port
1108             # send the handshake response
1109             # expect a keepalive message sent to the new endpoint
1110             rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1111             # verify the keepalive message
1112             b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
1113             self.assertEqual(0, len(b))
1114         else:
1115             # change endpoint
1116             if is_ip6:
1117                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1118             else:
1119                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1120             # prepare and send a handshake initiation
1121             # expect a handshake response sent to the new endpoint
1122             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
1123             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
1124             # verify the response
1125             peer_1.consume_response(rxs[0], is_ip6=is_ip6)
1126         self.assertTrue(peer_1.query_vpp_config())
1127
1128         # remove configs
1129         peer_1.remove_vpp_config()
1130         wg0.remove_vpp_config()
1131
1132     def test_wg_peer_roaming_on_init_v4(self):
1133         """Peer roaming on handshake initiation (v4)"""
1134         self._test_wg_peer_roaming_on_handshake_tmpl(
1135             is_endpoint_set=False, is_resp=False, is_ip6=False
1136         )
1137
1138     def test_wg_peer_roaming_on_init_v6(self):
1139         """Peer roaming on handshake initiation (v6)"""
1140         self._test_wg_peer_roaming_on_handshake_tmpl(
1141             is_endpoint_set=False, is_resp=False, is_ip6=True
1142         )
1143
1144     def test_wg_peer_roaming_on_resp_v4(self):
1145         """Peer roaming on handshake response (v4)"""
1146         self._test_wg_peer_roaming_on_handshake_tmpl(
1147             is_endpoint_set=True, is_resp=True, is_ip6=False
1148         )
1149
1150     def test_wg_peer_roaming_on_resp_v6(self):
1151         """Peer roaming on handshake response (v6)"""
1152         self._test_wg_peer_roaming_on_handshake_tmpl(
1153             is_endpoint_set=True, is_resp=True, is_ip6=True
1154         )
1155
1156     def _test_wg_peer_roaming_on_data_tmpl(self, is_async, is_ip6):
1157         self.vapi.wg_set_async_mode(is_async)
1158         port = 12323
1159
1160         # create wg interface
1161         if is_ip6:
1162             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1163             wg0.admin_up()
1164             wg0.config_ip6()
1165         else:
1166             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1167             wg0.admin_up()
1168             wg0.config_ip4()
1169
1170         self.pg_enable_capture(self.pg_interfaces)
1171         self.pg_start()
1172
1173         # create more remote hosts
1174         NUM_REMOTE_HOSTS = 2
1175         self.pg1.generate_remote_hosts(NUM_REMOTE_HOSTS)
1176         if is_ip6:
1177             self.pg1.configure_ipv6_neighbors()
1178         else:
1179             self.pg1.configure_ipv4_neighbors()
1180
1181         # create a peer
1182         if is_ip6:
1183             peer_1 = VppWgPeer(
1184                 self, wg0, self.pg1.remote_hosts[0].ip6, port + 1, ["1::3:0/112"]
1185             ).add_vpp_config()
1186         else:
1187             peer_1 = VppWgPeer(
1188                 self, wg0, self.pg1.remote_hosts[0].ip4, port + 1, ["10.11.3.0/24"]
1189             ).add_vpp_config()
1190         self.assertTrue(peer_1.query_vpp_config())
1191
1192         # create a route to rewrite traffic into the wg interface
1193         if is_ip6:
1194             r1 = VppIpRoute(
1195                 self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1196             ).add_vpp_config()
1197         else:
1198             r1 = VppIpRoute(
1199                 self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1200             ).add_vpp_config()
1201
1202         # wait for the peer to send a handshake initiation
1203         rxs = self.pg1.get_capture(1, timeout=2)
1204
1205         # prepare and send a handshake response
1206         # expect a keepalive message
1207         resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
1208         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1209
1210         # verify the keepalive message
1211         b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
1212         self.assertEqual(0, len(b))
1213
1214         # change endpoint
1215         if is_ip6:
1216             peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1217         else:
1218             peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1219
1220         # prepare and send a data packet
1221         # expect endpoint change
1222         if is_ip6:
1223             ip_header = IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1224         else:
1225             ip_header = IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1226         data = (
1227             peer_1.mk_tunnel_header(self.pg1, is_ip6=is_ip6)
1228             / Wireguard(message_type=4, reserved_zero=0)
1229             / WireguardTransport(
1230                 receiver_index=peer_1.sender,
1231                 counter=0,
1232                 encrypted_encapsulated_packet=peer_1.encrypt_transport(
1233                     ip_header / UDP(sport=222, dport=223) / Raw()
1234                 ),
1235             )
1236         )
1237         rxs = self.send_and_expect(self.pg1, [data], self.pg0)
1238         if is_ip6:
1239             self.assertEqual(rxs[0][IPv6].dst, self.pg0.remote_ip6)
1240             self.assertEqual(rxs[0][IPv6].hlim, 19)
1241         else:
1242             self.assertEqual(rxs[0][IP].dst, self.pg0.remote_ip4)
1243             self.assertEqual(rxs[0][IP].ttl, 19)
1244         self.assertTrue(peer_1.query_vpp_config())
1245
1246         # prepare and send a packet that will be rewritten into the wg interface
1247         # expect a data packet sent to the new endpoint
1248         if is_ip6:
1249             ip_header = IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1250         else:
1251             ip_header = IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1252         p = (
1253             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1254             / ip_header
1255             / UDP(sport=555, dport=556)
1256             / Raw()
1257         )
1258         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
1259
1260         # verify the data packet
1261         peer_1.validate_encapped(rxs, p, is_tunnel_ip6=is_ip6, is_transport_ip6=is_ip6)
1262
1263         # remove configs
1264         r1.remove_vpp_config()
1265         peer_1.remove_vpp_config()
1266         wg0.remove_vpp_config()
1267
1268     def test_wg_peer_roaming_on_data_v4_sync(self):
1269         """Peer roaming on data packet (v4, sync)"""
1270         self._test_wg_peer_roaming_on_data_tmpl(is_async=False, is_ip6=False)
1271
1272     def test_wg_peer_roaming_on_data_v6_sync(self):
1273         """Peer roaming on data packet (v6, sync)"""
1274         self._test_wg_peer_roaming_on_data_tmpl(is_async=False, is_ip6=True)
1275
1276     def test_wg_peer_roaming_on_data_v4_async(self):
1277         """Peer roaming on data packet (v4, async)"""
1278         self._test_wg_peer_roaming_on_data_tmpl(is_async=True, is_ip6=False)
1279
1280     def test_wg_peer_roaming_on_data_v6_async(self):
1281         """Peer roaming on data packet (v6, async)"""
1282         self._test_wg_peer_roaming_on_data_tmpl(is_async=True, is_ip6=True)
1283
1284     def test_wg_peer_resp(self):
1285         """Send handshake response IPv4 tunnel"""
1286         port = 12323
1287
1288         # Create interfaces
1289         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1290         wg0.admin_up()
1291         wg0.config_ip4()
1292
1293         self.pg_enable_capture(self.pg_interfaces)
1294         self.pg_start()
1295
1296         peer_1 = VppWgPeer(
1297             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
1298         ).add_vpp_config()
1299         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1300
1301         r1 = VppIpRoute(
1302             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1303         ).add_vpp_config()
1304
1305         # wait for the peer to send a handshake
1306         rx = self.pg1.get_capture(1, timeout=2)
1307
1308         # consume the handshake in the noise protocol and
1309         # generate the response
1310         resp = peer_1.consume_init(rx[0], self.pg1)
1311
1312         # send the response, get keepalive
1313         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1314
1315         for rx in rxs:
1316             b = peer_1.decrypt_transport(rx)
1317             self.assertEqual(0, len(b))
1318
1319         # send a packets that are routed into the tunnel
1320         p = (
1321             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1322             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1323             / UDP(sport=555, dport=556)
1324             / Raw(b"\x00" * 80)
1325         )
1326
1327         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1328
1329         peer_1.validate_encapped(rxs, p)
1330
1331         # send packets into the tunnel, expect to receive them on
1332         # the other side
1333         p = [
1334             (
1335                 peer_1.mk_tunnel_header(self.pg1)
1336                 / Wireguard(message_type=4, reserved_zero=0)
1337                 / WireguardTransport(
1338                     receiver_index=peer_1.sender,
1339                     counter=ii,
1340                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1341                         (
1342                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1343                             / UDP(sport=222, dport=223)
1344                             / Raw()
1345                         )
1346                     ),
1347                 )
1348             )
1349             for ii in range(255)
1350         ]
1351
1352         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1353
1354         for rx in rxs:
1355             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1356             self.assertEqual(rx[IP].ttl, 19)
1357
1358         r1.remove_vpp_config()
1359         peer_1.remove_vpp_config()
1360         wg0.remove_vpp_config()
1361
1362     def test_wg_peer_resp_ipv6(self):
1363         """Send handshake response IPv6 tunnel"""
1364         port = 12323
1365
1366         # Create interfaces
1367         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1368         wg0.admin_up()
1369         wg0.config_ip4()
1370
1371         self.pg_enable_capture(self.pg_interfaces)
1372         self.pg_start()
1373
1374         peer_1 = VppWgPeer(
1375             self, wg0, self.pg1.remote_ip6, port + 1, ["10.11.3.0/24"]
1376         ).add_vpp_config()
1377         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1378
1379         r1 = VppIpRoute(
1380             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1381         ).add_vpp_config()
1382
1383         # wait for the peer to send a handshake
1384         rx = self.pg1.get_capture(1, timeout=2)
1385
1386         # consume the handshake in the noise protocol and
1387         # generate the response
1388         resp = peer_1.consume_init(rx[0], self.pg1, is_ip6=True)
1389
1390         # send the response, get keepalive
1391         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1392
1393         for rx in rxs:
1394             b = peer_1.decrypt_transport(rx, True)
1395             self.assertEqual(0, len(b))
1396
1397         # send a packets that are routed into the tunnel
1398         p = (
1399             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1400             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1401             / UDP(sport=555, dport=556)
1402             / Raw(b"\x00" * 80)
1403         )
1404
1405         rxs = self.send_and_expect(self.pg0, p * 2, self.pg1)
1406         peer_1.validate_encapped(rxs, p, True)
1407
1408         # send packets into the tunnel, expect to receive them on
1409         # the other side
1410         p = [
1411             (
1412                 peer_1.mk_tunnel_header(self.pg1, True)
1413                 / Wireguard(message_type=4, reserved_zero=0)
1414                 / WireguardTransport(
1415                     receiver_index=peer_1.sender,
1416                     counter=ii,
1417                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1418                         (
1419                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1420                             / UDP(sport=222, dport=223)
1421                             / Raw()
1422                         )
1423                     ),
1424                 )
1425             )
1426             for ii in range(255)
1427         ]
1428
1429         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1430
1431         for rx in rxs:
1432             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1433             self.assertEqual(rx[IP].ttl, 19)
1434
1435         r1.remove_vpp_config()
1436         peer_1.remove_vpp_config()
1437         wg0.remove_vpp_config()
1438
1439     def test_wg_peer_v4o4(self):
1440         """Test v4o4"""
1441
1442         port = 12333
1443
1444         # Create interfaces
1445         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1446         wg0.admin_up()
1447         wg0.config_ip4()
1448
1449         peer_1 = VppWgPeer(
1450             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
1451         ).add_vpp_config()
1452         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1453
1454         r1 = VppIpRoute(
1455             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1456         ).add_vpp_config()
1457         r2 = VppIpRoute(
1458             self, "20.22.3.0", 24, [VppRoutePath("20.22.3.1", wg0.sw_if_index)]
1459         ).add_vpp_config()
1460
1461         # route a packet into the wg interface
1462         #  use the allowed-ip prefix
1463         #  this is dropped because the peer is not initiated
1464         p = (
1465             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1466             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1467             / UDP(sport=555, dport=556)
1468             / Raw()
1469         )
1470         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1471         self.assertEqual(
1472             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1473         )
1474
1475         # route a packet into the wg interface
1476         #  use a not allowed-ip prefix
1477         #  this is dropped because there is no matching peer
1478         p = (
1479             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1480             / IP(src=self.pg0.remote_ip4, dst="20.22.3.2")
1481             / UDP(sport=555, dport=556)
1482             / Raw()
1483         )
1484         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1485         self.assertEqual(
1486             self.base_peer4_out_err + 1,
1487             self.statistics.get_err_counter(self.peer4_out_err),
1488         )
1489
1490         # send a handsake from the peer with an invalid MAC
1491         p = peer_1.mk_handshake(self.pg1)
1492         p[WireguardInitiation].mac1 = b"foobar"
1493         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1494         self.assertEqual(
1495             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1496         )
1497
1498         # send a handsake from the peer but signed by the wrong key.
1499         p = peer_1.mk_handshake(
1500             self.pg1, False, X25519PrivateKey.generate().public_key()
1501         )
1502         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1503         self.assertEqual(
1504             self.base_peer4_in_err + 1,
1505             self.statistics.get_err_counter(self.peer4_in_err),
1506         )
1507
1508         # send a valid handsake init for which we expect a response
1509         p = peer_1.mk_handshake(self.pg1)
1510
1511         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1512
1513         peer_1.consume_response(rx[0])
1514
1515         # route a packet into the wg interface
1516         #  this is dropped because the peer is still not initiated
1517         p = (
1518             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1519             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1520             / UDP(sport=555, dport=556)
1521             / Raw()
1522         )
1523         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1524         self.assertEqual(
1525             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1526         )
1527
1528         # send a data packet from the peer through the tunnel
1529         # this completes the handshake
1530         p = (
1531             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1532             / UDP(sport=222, dport=223)
1533             / Raw()
1534         )
1535         d = peer_1.encrypt_transport(p)
1536         p = peer_1.mk_tunnel_header(self.pg1) / (
1537             Wireguard(message_type=4, reserved_zero=0)
1538             / WireguardTransport(
1539                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1540             )
1541         )
1542         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1543
1544         for rx in rxs:
1545             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1546             self.assertEqual(rx[IP].ttl, 19)
1547
1548         # send a packets that are routed into the tunnel
1549         p = (
1550             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1551             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1552             / UDP(sport=555, dport=556)
1553             / Raw(b"\x00" * 80)
1554         )
1555
1556         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1557
1558         for rx in rxs:
1559             rx = IP(peer_1.decrypt_transport(rx))
1560
1561             # check the original packet is present
1562             self.assertEqual(rx[IP].dst, p[IP].dst)
1563             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1564
1565         # send packets into the tunnel, expect to receive them on
1566         # the other side
1567         p = [
1568             (
1569                 peer_1.mk_tunnel_header(self.pg1)
1570                 / Wireguard(message_type=4, reserved_zero=0)
1571                 / WireguardTransport(
1572                     receiver_index=peer_1.sender,
1573                     counter=ii + 1,
1574                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1575                         (
1576                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1577                             / UDP(sport=222, dport=223)
1578                             / Raw()
1579                         )
1580                     ),
1581                 )
1582             )
1583             for ii in range(255)
1584         ]
1585
1586         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1587
1588         for rx in rxs:
1589             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1590             self.assertEqual(rx[IP].ttl, 19)
1591
1592         r1.remove_vpp_config()
1593         r2.remove_vpp_config()
1594         peer_1.remove_vpp_config()
1595         wg0.remove_vpp_config()
1596
1597     def test_wg_peer_v6o6(self):
1598         """Test v6o6"""
1599
1600         port = 12343
1601
1602         # Create interfaces
1603         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1604         wg0.admin_up()
1605         wg0.config_ip6()
1606
1607         peer_1 = VppWgPeer(
1608             self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
1609         ).add_vpp_config()
1610         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1611
1612         r1 = VppIpRoute(
1613             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1614         ).add_vpp_config()
1615         r2 = VppIpRoute(
1616             self, "22::3:0", 112, [VppRoutePath("22::3:1", wg0.sw_if_index)]
1617         ).add_vpp_config()
1618
1619         # route a packet into the wg interface
1620         #  use the allowed-ip prefix
1621         #  this is dropped because the peer is not initiated
1622
1623         p = (
1624             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1625             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1626             / UDP(sport=555, dport=556)
1627             / Raw()
1628         )
1629         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1630
1631         self.assertEqual(
1632             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1633         )
1634
1635         # route a packet into the wg interface
1636         #  use a not allowed-ip prefix
1637         #  this is dropped because there is no matching peer
1638         p = (
1639             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1640             / IPv6(src=self.pg0.remote_ip6, dst="22::3:2")
1641             / UDP(sport=555, dport=556)
1642             / Raw()
1643         )
1644         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1645         self.assertEqual(
1646             self.base_peer6_out_err + 1,
1647             self.statistics.get_err_counter(self.peer6_out_err),
1648         )
1649
1650         # send a handsake from the peer with an invalid MAC
1651         p = peer_1.mk_handshake(self.pg1, True)
1652         p[WireguardInitiation].mac1 = b"foobar"
1653         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1654
1655         self.assertEqual(
1656             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1657         )
1658
1659         # send a handsake from the peer but signed by the wrong key.
1660         p = peer_1.mk_handshake(
1661             self.pg1, True, X25519PrivateKey.generate().public_key()
1662         )
1663         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1664         self.assertEqual(
1665             self.base_peer6_in_err + 1,
1666             self.statistics.get_err_counter(self.peer6_in_err),
1667         )
1668
1669         # send a valid handsake init for which we expect a response
1670         p = peer_1.mk_handshake(self.pg1, True)
1671
1672         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1673
1674         peer_1.consume_response(rx[0], True)
1675
1676         # route a packet into the wg interface
1677         #  this is dropped because the peer is still not initiated
1678         p = (
1679             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1680             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1681             / UDP(sport=555, dport=556)
1682             / Raw()
1683         )
1684         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1685         self.assertEqual(
1686             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1687         )
1688
1689         # send a data packet from the peer through the tunnel
1690         # this completes the handshake
1691         p = (
1692             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1693             / UDP(sport=222, dport=223)
1694             / Raw()
1695         )
1696         d = peer_1.encrypt_transport(p)
1697         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1698             Wireguard(message_type=4, reserved_zero=0)
1699             / WireguardTransport(
1700                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1701             )
1702         )
1703         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1704
1705         for rx in rxs:
1706             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1707             self.assertEqual(rx[IPv6].hlim, 19)
1708
1709         # send a packets that are routed into the tunnel
1710         p = (
1711             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1712             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1713             / UDP(sport=555, dport=556)
1714             / Raw(b"\x00" * 80)
1715         )
1716
1717         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1718
1719         for rx in rxs:
1720             rx = IPv6(peer_1.decrypt_transport(rx, True))
1721
1722             # check the original packet is present
1723             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1724             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1725
1726         # send packets into the tunnel, expect to receive them on
1727         # the other side
1728         p = [
1729             (
1730                 peer_1.mk_tunnel_header(self.pg1, True)
1731                 / Wireguard(message_type=4, reserved_zero=0)
1732                 / WireguardTransport(
1733                     receiver_index=peer_1.sender,
1734                     counter=ii + 1,
1735                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1736                         (
1737                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1738                             / UDP(sport=222, dport=223)
1739                             / Raw()
1740                         )
1741                     ),
1742                 )
1743             )
1744             for ii in range(255)
1745         ]
1746
1747         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1748
1749         for rx in rxs:
1750             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1751             self.assertEqual(rx[IPv6].hlim, 19)
1752
1753         r1.remove_vpp_config()
1754         r2.remove_vpp_config()
1755         peer_1.remove_vpp_config()
1756         wg0.remove_vpp_config()
1757
1758     def test_wg_peer_v6o4(self):
1759         """Test v6o4"""
1760
1761         port = 12353
1762
1763         # Create interfaces
1764         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1765         wg0.admin_up()
1766         wg0.config_ip6()
1767
1768         peer_1 = VppWgPeer(
1769             self, wg0, self.pg1.remote_ip4, port + 1, ["1::3:0/112"]
1770         ).add_vpp_config()
1771         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1772
1773         r1 = VppIpRoute(
1774             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1775         ).add_vpp_config()
1776
1777         # route a packet into the wg interface
1778         #  use the allowed-ip prefix
1779         #  this is dropped because the peer is not initiated
1780         p = (
1781             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1782             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1783             / UDP(sport=555, dport=556)
1784             / Raw()
1785         )
1786         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1787         self.assertEqual(
1788             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1789         )
1790
1791         # send a handsake from the peer with an invalid MAC
1792         p = peer_1.mk_handshake(self.pg1)
1793         p[WireguardInitiation].mac1 = b"foobar"
1794         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1795
1796         self.assertEqual(
1797             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1798         )
1799
1800         # send a handsake from the peer but signed by the wrong key.
1801         p = peer_1.mk_handshake(
1802             self.pg1, False, X25519PrivateKey.generate().public_key()
1803         )
1804         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1805         self.assertEqual(
1806             self.base_peer4_in_err + 1,
1807             self.statistics.get_err_counter(self.peer4_in_err),
1808         )
1809
1810         # send a valid handsake init for which we expect a response
1811         p = peer_1.mk_handshake(self.pg1)
1812
1813         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1814
1815         peer_1.consume_response(rx[0])
1816
1817         # route a packet into the wg interface
1818         #  this is dropped because the peer is still not initiated
1819         p = (
1820             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1821             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1822             / UDP(sport=555, dport=556)
1823             / Raw()
1824         )
1825         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1826         self.assertEqual(
1827             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1828         )
1829
1830         # send a data packet from the peer through the tunnel
1831         # this completes the handshake
1832         p = (
1833             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1834             / UDP(sport=222, dport=223)
1835             / Raw()
1836         )
1837         d = peer_1.encrypt_transport(p)
1838         p = peer_1.mk_tunnel_header(self.pg1) / (
1839             Wireguard(message_type=4, reserved_zero=0)
1840             / WireguardTransport(
1841                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1842             )
1843         )
1844         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1845
1846         for rx in rxs:
1847             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1848             self.assertEqual(rx[IPv6].hlim, 19)
1849
1850         # send a packets that are routed into the tunnel
1851         p = (
1852             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1853             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1854             / UDP(sport=555, dport=556)
1855             / Raw(b"\x00" * 80)
1856         )
1857
1858         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1859
1860         for rx in rxs:
1861             rx = IPv6(peer_1.decrypt_transport(rx))
1862
1863             # check the original packet is present
1864             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1865             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1866
1867         # send packets into the tunnel, expect to receive them on
1868         # the other side
1869         p = [
1870             (
1871                 peer_1.mk_tunnel_header(self.pg1)
1872                 / Wireguard(message_type=4, reserved_zero=0)
1873                 / WireguardTransport(
1874                     receiver_index=peer_1.sender,
1875                     counter=ii + 1,
1876                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1877                         (
1878                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1879                             / UDP(sport=222, dport=223)
1880                             / Raw()
1881                         )
1882                     ),
1883                 )
1884             )
1885             for ii in range(255)
1886         ]
1887
1888         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1889
1890         for rx in rxs:
1891             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1892             self.assertEqual(rx[IPv6].hlim, 19)
1893
1894         r1.remove_vpp_config()
1895         peer_1.remove_vpp_config()
1896         wg0.remove_vpp_config()
1897
1898     def test_wg_peer_v4o6(self):
1899         """Test v4o6"""
1900
1901         port = 12363
1902
1903         # Create interfaces
1904         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1905         wg0.admin_up()
1906         wg0.config_ip4()
1907
1908         peer_1 = VppWgPeer(
1909             self, wg0, self.pg1.remote_ip6, port + 1, ["10.11.3.0/24"]
1910         ).add_vpp_config()
1911         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1912
1913         r1 = VppIpRoute(
1914             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1915         ).add_vpp_config()
1916
1917         # route a packet into the wg interface
1918         #  use the allowed-ip prefix
1919         #  this is dropped because the peer is not initiated
1920         p = (
1921             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1922             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1923             / UDP(sport=555, dport=556)
1924             / Raw()
1925         )
1926         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1927         self.assertEqual(
1928             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1929         )
1930
1931         # send a handsake from the peer with an invalid MAC
1932         p = peer_1.mk_handshake(self.pg1, True)
1933         p[WireguardInitiation].mac1 = b"foobar"
1934         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1935         self.assertEqual(
1936             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1937         )
1938
1939         # send a handsake from the peer but signed by the wrong key.
1940         p = peer_1.mk_handshake(
1941             self.pg1, True, X25519PrivateKey.generate().public_key()
1942         )
1943         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1944         self.assertEqual(
1945             self.base_peer6_in_err + 1,
1946             self.statistics.get_err_counter(self.peer6_in_err),
1947         )
1948
1949         # send a valid handsake init for which we expect a response
1950         p = peer_1.mk_handshake(self.pg1, True)
1951
1952         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1953
1954         peer_1.consume_response(rx[0], True)
1955
1956         # route a packet into the wg interface
1957         #  this is dropped because the peer is still not initiated
1958         p = (
1959             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1960             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1961             / UDP(sport=555, dport=556)
1962             / Raw()
1963         )
1964         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1965         self.assertEqual(
1966             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1967         )
1968
1969         # send a data packet from the peer through the tunnel
1970         # this completes the handshake
1971         p = (
1972             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1973             / UDP(sport=222, dport=223)
1974             / Raw()
1975         )
1976         d = peer_1.encrypt_transport(p)
1977         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1978             Wireguard(message_type=4, reserved_zero=0)
1979             / WireguardTransport(
1980                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1981             )
1982         )
1983         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1984
1985         for rx in rxs:
1986             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1987             self.assertEqual(rx[IP].ttl, 19)
1988
1989         # send a packets that are routed into the tunnel
1990         p = (
1991             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1992             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1993             / UDP(sport=555, dport=556)
1994             / Raw(b"\x00" * 80)
1995         )
1996
1997         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1998
1999         for rx in rxs:
2000             rx = IP(peer_1.decrypt_transport(rx, True))
2001
2002             # check the original packet is present
2003             self.assertEqual(rx[IP].dst, p[IP].dst)
2004             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
2005
2006         # send packets into the tunnel, expect to receive them on
2007         # the other side
2008         p = [
2009             (
2010                 peer_1.mk_tunnel_header(self.pg1, True)
2011                 / Wireguard(message_type=4, reserved_zero=0)
2012                 / WireguardTransport(
2013                     receiver_index=peer_1.sender,
2014                     counter=ii + 1,
2015                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
2016                         (
2017                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2018                             / UDP(sport=222, dport=223)
2019                             / Raw()
2020                         )
2021                     ),
2022                 )
2023             )
2024             for ii in range(255)
2025         ]
2026
2027         rxs = self.send_and_expect(self.pg1, p, self.pg0)
2028
2029         for rx in rxs:
2030             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2031             self.assertEqual(rx[IP].ttl, 19)
2032
2033         r1.remove_vpp_config()
2034         peer_1.remove_vpp_config()
2035         wg0.remove_vpp_config()
2036
2037     def test_wg_multi_peer(self):
2038         """multiple peer setup"""
2039         port = 12373
2040
2041         # Create interfaces
2042         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2043         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
2044         wg0.admin_up()
2045         wg1.admin_up()
2046
2047         # Check peer counter
2048         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
2049
2050         self.pg_enable_capture(self.pg_interfaces)
2051         self.pg_start()
2052
2053         # Create many peers on sencond interface
2054         NUM_PEERS = 16
2055         self.pg2.generate_remote_hosts(NUM_PEERS)
2056         self.pg2.configure_ipv4_neighbors()
2057         self.pg1.generate_remote_hosts(NUM_PEERS)
2058         self.pg1.configure_ipv4_neighbors()
2059
2060         peers_1 = []
2061         peers_2 = []
2062         routes_1 = []
2063         routes_2 = []
2064         for i in range(NUM_PEERS):
2065             peers_1.append(
2066                 VppWgPeer(
2067                     self,
2068                     wg0,
2069                     self.pg1.remote_hosts[i].ip4,
2070                     port + 1 + i,
2071                     ["10.0.%d.4/32" % i],
2072                 ).add_vpp_config()
2073             )
2074             routes_1.append(
2075                 VppIpRoute(
2076                     self,
2077                     "10.0.%d.4" % i,
2078                     32,
2079                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
2080                 ).add_vpp_config()
2081             )
2082
2083             peers_2.append(
2084                 VppWgPeer(
2085                     self,
2086                     wg1,
2087                     self.pg2.remote_hosts[i].ip4,
2088                     port + 100 + i,
2089                     ["10.100.%d.4/32" % i],
2090                 ).add_vpp_config()
2091             )
2092             routes_2.append(
2093                 VppIpRoute(
2094                     self,
2095                     "10.100.%d.4" % i,
2096                     32,
2097                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
2098                 ).add_vpp_config()
2099             )
2100
2101         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
2102
2103         self.logger.info(self.vapi.cli("show wireguard peer"))
2104         self.logger.info(self.vapi.cli("show wireguard interface"))
2105         self.logger.info(self.vapi.cli("show adj 37"))
2106         self.logger.info(self.vapi.cli("sh ip fib 172.16.3.17"))
2107         self.logger.info(self.vapi.cli("sh ip fib 10.11.3.0"))
2108
2109         # remove routes
2110         for r in routes_1:
2111             r.remove_vpp_config()
2112         for r in routes_2:
2113             r.remove_vpp_config()
2114
2115         # remove peers
2116         for p in peers_1:
2117             self.assertTrue(p.query_vpp_config())
2118             p.remove_vpp_config()
2119         for p in peers_2:
2120             self.assertTrue(p.query_vpp_config())
2121             p.remove_vpp_config()
2122
2123         wg0.remove_vpp_config()
2124         wg1.remove_vpp_config()
2125
2126     def test_wg_multi_interface(self):
2127         """Multi-tunnel on the same port"""
2128         port = 12500
2129
2130         # Create many wireguard interfaces
2131         NUM_IFS = 4
2132         self.pg1.generate_remote_hosts(NUM_IFS)
2133         self.pg1.configure_ipv4_neighbors()
2134         self.pg0.generate_remote_hosts(NUM_IFS)
2135         self.pg0.configure_ipv4_neighbors()
2136
2137         self.pg_enable_capture(self.pg_interfaces)
2138         self.pg_start()
2139
2140         # Create interfaces with a peer on each
2141         peers = []
2142         routes = []
2143         wg_ifs = []
2144         for i in range(NUM_IFS):
2145             # Use the same port for each interface
2146             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2147             wg0.admin_up()
2148             wg0.config_ip4()
2149             wg_ifs.append(wg0)
2150             peers.append(
2151                 VppWgPeer(
2152                     self,
2153                     wg0,
2154                     self.pg1.remote_hosts[i].ip4,
2155                     port + 1 + i,
2156                     ["10.0.%d.0/24" % i],
2157                 ).add_vpp_config()
2158             )
2159
2160             routes.append(
2161                 VppIpRoute(
2162                     self,
2163                     "10.0.%d.0" % i,
2164                     24,
2165                     [VppRoutePath("10.0.%d.4" % i, wg0.sw_if_index)],
2166                 ).add_vpp_config()
2167             )
2168
2169         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_IFS)
2170
2171         # skip the first automatic handshake
2172         self.pg1.get_capture(NUM_IFS, timeout=HANDSHAKE_JITTER)
2173
2174         for i in range(NUM_IFS):
2175             # send a valid handsake init for which we expect a response
2176             p = peers[i].mk_handshake(self.pg1)
2177             rx = self.send_and_expect(self.pg1, [p], self.pg1)
2178             peers[i].consume_response(rx[0])
2179
2180             # send a data packet from the peer through the tunnel
2181             # this completes the handshake
2182             p = (
2183                 IP(src="10.0.%d.4" % i, dst=self.pg0.remote_hosts[i].ip4, ttl=20)
2184                 / UDP(sport=222, dport=223)
2185                 / Raw()
2186             )
2187             d = peers[i].encrypt_transport(p)
2188             p = peers[i].mk_tunnel_header(self.pg1) / (
2189                 Wireguard(message_type=4, reserved_zero=0)
2190                 / WireguardTransport(
2191                     receiver_index=peers[i].sender,
2192                     counter=0,
2193                     encrypted_encapsulated_packet=d,
2194                 )
2195             )
2196             rxs = self.send_and_expect(self.pg1, [p], self.pg0)
2197             for rx in rxs:
2198                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
2199                 self.assertEqual(rx[IP].ttl, 19)
2200
2201         # send a packets that are routed into the tunnel
2202         for i in range(NUM_IFS):
2203             p = (
2204                 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2205                 / IP(src=self.pg0.remote_hosts[i].ip4, dst="10.0.%d.4" % i)
2206                 / UDP(sport=555, dport=556)
2207                 / Raw(b"\x00" * 80)
2208             )
2209
2210             rxs = self.send_and_expect(self.pg0, p * 64, self.pg1)
2211
2212             for rx in rxs:
2213                 rx = IP(peers[i].decrypt_transport(rx))
2214
2215                 # check the oringial packet is present
2216                 self.assertEqual(rx[IP].dst, p[IP].dst)
2217                 self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
2218
2219         # send packets into the tunnel
2220         for i in range(NUM_IFS):
2221             p = [
2222                 (
2223                     peers[i].mk_tunnel_header(self.pg1)
2224                     / Wireguard(message_type=4, reserved_zero=0)
2225                     / WireguardTransport(
2226                         receiver_index=peers[i].sender,
2227                         counter=ii + 1,
2228                         encrypted_encapsulated_packet=peers[i].encrypt_transport(
2229                             (
2230                                 IP(
2231                                     src="10.0.%d.4" % i,
2232                                     dst=self.pg0.remote_hosts[i].ip4,
2233                                     ttl=20,
2234                                 )
2235                                 / UDP(sport=222, dport=223)
2236                                 / Raw()
2237                             )
2238                         ),
2239                     )
2240                 )
2241                 for ii in range(64)
2242             ]
2243
2244             rxs = self.send_and_expect(self.pg1, p, self.pg0)
2245
2246             for rx in rxs:
2247                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
2248                 self.assertEqual(rx[IP].ttl, 19)
2249
2250         for r in routes:
2251             r.remove_vpp_config()
2252         for p in peers:
2253             p.remove_vpp_config()
2254         for i in wg_ifs:
2255             i.remove_vpp_config()
2256
2257     def test_wg_event(self):
2258         """Test events"""
2259         port = 12600
2260         ESTABLISHED_FLAG = (
2261             VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_ESTABLISHED
2262         )
2263         DEAD_FLAG = VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_STATUS_DEAD
2264
2265         # Create interfaces
2266         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2267         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
2268         wg0.admin_up()
2269         wg1.admin_up()
2270
2271         # Check peer counter
2272         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
2273
2274         self.pg_enable_capture(self.pg_interfaces)
2275         self.pg_start()
2276
2277         # Create peers
2278         NUM_PEERS = 2
2279         self.pg2.generate_remote_hosts(NUM_PEERS)
2280         self.pg2.configure_ipv4_neighbors()
2281         self.pg1.generate_remote_hosts(NUM_PEERS)
2282         self.pg1.configure_ipv4_neighbors()
2283
2284         peers_0 = []
2285         peers_1 = []
2286         routes_0 = []
2287         routes_1 = []
2288         for i in range(NUM_PEERS):
2289             peers_0.append(
2290                 VppWgPeer(
2291                     self,
2292                     wg0,
2293                     self.pg1.remote_hosts[i].ip4,
2294                     port + 1 + i,
2295                     ["10.0.%d.4/32" % i],
2296                 ).add_vpp_config()
2297             )
2298             routes_0.append(
2299                 VppIpRoute(
2300                     self,
2301                     "10.0.%d.4" % i,
2302                     32,
2303                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
2304                 ).add_vpp_config()
2305             )
2306
2307             peers_1.append(
2308                 VppWgPeer(
2309                     self,
2310                     wg1,
2311                     self.pg2.remote_hosts[i].ip4,
2312                     port + 100 + i,
2313                     ["10.100.%d.4/32" % i],
2314                 ).add_vpp_config()
2315             )
2316             routes_1.append(
2317                 VppIpRoute(
2318                     self,
2319                     "10.100.%d.4" % i,
2320                     32,
2321                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
2322                 ).add_vpp_config()
2323             )
2324
2325         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
2326
2327         # skip the first automatic handshake
2328         self.pg1.get_capture(NUM_PEERS, timeout=HANDSHAKE_JITTER)
2329         self.pg2.get_capture(NUM_PEERS, timeout=HANDSHAKE_JITTER)
2330
2331         # Want events from the first perr of wg0
2332         # and from all wg1 peers
2333         peers_0[0].want_events()
2334         wg1.want_events()
2335
2336         for i in range(NUM_PEERS):
2337             # wg0 peers: send a valid handsake init for which we expect a response
2338             p = peers_0[i].mk_handshake(self.pg1)
2339             rx = self.send_and_expect(self.pg1, [p], self.pg1)
2340             peers_0[i].consume_response(rx[0])
2341
2342             # wg0 peers: send empty packet, it means successful connection (WIREGUARD_PEER_ESTABLISHED)
2343             keepalive = peers_0[i].encrypt_transport(0)
2344             p = peers_0[i].mk_tunnel_header(self.pg1) / (
2345                 Wireguard(message_type=4, reserved_zero=0)
2346                 / WireguardTransport(
2347                     receiver_index=peers_0[i].sender,
2348                     counter=0,
2349                     encrypted_encapsulated_packet=keepalive,
2350                 )
2351             )
2352             # TODO: Figure out wny there are sometimes wg packets received here
2353             # self.send_and_assert_no_replies(self.pg1, [p])
2354             self.pg_send(self.pg1, [p])
2355
2356             # wg0 peers: wait for established flag
2357             if i == 0:
2358                 peers_0[0].wait_event(ESTABLISHED_FLAG)
2359
2360             # wg1 peers: send a valid handsake init for which we expect a response
2361             p = peers_1[i].mk_handshake(self.pg2)
2362             rx = self.send_and_expect(self.pg2, [p], self.pg2)
2363             peers_1[i].consume_response(rx[0])
2364
2365             # wg1 peers: send empty packet, it means successful connection (WIREGUARD_PEER_ESTABLISHED)
2366             keepalive = peers_1[i].encrypt_transport(0)
2367             p = peers_1[i].mk_tunnel_header(self.pg2) / (
2368                 Wireguard(message_type=4, reserved_zero=0)
2369                 / WireguardTransport(
2370                     receiver_index=peers_1[i].sender,
2371                     counter=0,
2372                     encrypted_encapsulated_packet=keepalive,
2373                 )
2374             )
2375             self.send_and_assert_no_replies(self.pg2, [p])
2376
2377         # wg1 peers: wait for established flag
2378         wg1.wait_events(ESTABLISHED_FLAG, [peers_1[0].index, peers_1[1].index])
2379
2380         # remove routes
2381         for r in routes_0:
2382             r.remove_vpp_config()
2383         for r in routes_1:
2384             r.remove_vpp_config()
2385
2386         # remove peers
2387         for i in range(NUM_PEERS):
2388             self.assertTrue(peers_0[i].query_vpp_config())
2389             peers_0[i].remove_vpp_config()
2390             if i == 0:
2391                 peers_0[i].wait_event(0)
2392                 peers_0[i].wait_event(DEAD_FLAG)
2393         for p in peers_1:
2394             self.assertTrue(p.query_vpp_config())
2395             p.remove_vpp_config()
2396             p.wait_event(0)
2397             p.wait_event(DEAD_FLAG)
2398
2399         wg0.remove_vpp_config()
2400         wg1.remove_vpp_config()
2401
2402     def test_wg_sending_handshake_when_admin_down(self):
2403         """Sending handshake when admin down"""
2404         port = 12323
2405
2406         # create wg interface
2407         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2408         wg0.config_ip4()
2409
2410         # create a peer
2411         peer_1 = VppWgPeer(
2412             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2413         ).add_vpp_config()
2414         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2415
2416         self.pg_enable_capture(self.pg_interfaces)
2417         self.pg_start()
2418
2419         # wait for the peer to send a handshake initiation
2420         # expect no handshakes
2421         for i in range(2):
2422             self.pg1.assert_nothing_captured(remark="handshake packet(s) sent")
2423
2424         self.pg_enable_capture(self.pg_interfaces)
2425         self.pg_start()
2426
2427         # administratively enable the wg interface
2428         # expect the peer to send a handshake initiation
2429         wg0.admin_up()
2430         rxs = self.pg1.get_capture(1, timeout=2)
2431         peer_1.consume_init(rxs[0], self.pg1)
2432
2433         self.pg_enable_capture(self.pg_interfaces)
2434         self.pg_start()
2435
2436         # administratively disable the wg interface
2437         # expect no handshakes
2438         wg0.admin_down()
2439         for i in range(6):
2440             self.pg1.assert_nothing_captured(remark="handshake packet(s) sent")
2441
2442         # remove configs
2443         peer_1.remove_vpp_config()
2444         wg0.remove_vpp_config()
2445
2446     def test_wg_sending_data_when_admin_down(self):
2447         """Sending data when admin down"""
2448         port = 12323
2449
2450         # create wg interface
2451         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2452         wg0.admin_up()
2453         wg0.config_ip4()
2454
2455         self.pg_enable_capture(self.pg_interfaces)
2456         self.pg_start()
2457
2458         # create a peer
2459         peer_1 = VppWgPeer(
2460             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2461         ).add_vpp_config()
2462         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2463
2464         # create a route to rewrite traffic into the wg interface
2465         r1 = VppIpRoute(
2466             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2467         ).add_vpp_config()
2468
2469         # wait for the peer to send a handshake initiation
2470         rxs = self.pg1.get_capture(1, timeout=2)
2471
2472         # prepare and send a handshake response
2473         # expect a keepalive message
2474         resp = peer_1.consume_init(rxs[0], self.pg1)
2475         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2476
2477         # verify the keepalive message
2478         b = peer_1.decrypt_transport(rxs[0])
2479         self.assertEqual(0, len(b))
2480
2481         # prepare and send a packet that will be rewritten into the wg interface
2482         # expect a data packet sent
2483         p = (
2484             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2485             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2486             / UDP(sport=555, dport=556)
2487             / Raw()
2488         )
2489         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2490
2491         # verify the data packet
2492         peer_1.validate_encapped(rxs, p)
2493
2494         # administratively disable the wg interface
2495         wg0.admin_down()
2496
2497         # send a packet that will be rewritten into the wg interface
2498         # expect no data packets sent
2499         self.send_and_assert_no_replies(self.pg0, [p])
2500
2501         # administratively enable the wg interface
2502         # expect the peer to send a handshake initiation
2503         wg0.admin_up()
2504         peer_1.noise_reset()
2505         rxs = self.pg1.get_capture(1, timeout=2)
2506         resp = peer_1.consume_init(rxs[0], self.pg1)
2507
2508         # send a packet that will be rewritten into the wg interface
2509         # expect no data packets sent because the peer is not initiated
2510         self.send_and_assert_no_replies(self.pg0, [p])
2511         self.assertEqual(
2512             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
2513         )
2514
2515         # send a handshake response and expect a keepalive message
2516         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2517
2518         # verify the keepalive message
2519         b = peer_1.decrypt_transport(rxs[0])
2520         self.assertEqual(0, len(b))
2521
2522         # send a packet that will be rewritten into the wg interface
2523         # expect a data packet sent
2524         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2525
2526         # verify the data packet
2527         peer_1.validate_encapped(rxs, p)
2528
2529         # remove configs
2530         r1.remove_vpp_config()
2531         peer_1.remove_vpp_config()
2532         wg0.remove_vpp_config()
2533
2534     def _test_wg_large_packet_tmpl(self, is_async, is_ip6):
2535         self.vapi.wg_set_async_mode(is_async)
2536         port = 12323
2537
2538         # create wg interface
2539         if is_ip6:
2540             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
2541             wg0.admin_up()
2542             wg0.config_ip6()
2543         else:
2544             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2545             wg0.admin_up()
2546             wg0.config_ip4()
2547
2548         self.pg_enable_capture(self.pg_interfaces)
2549         self.pg_start()
2550
2551         # create a peer
2552         if is_ip6:
2553             peer_1 = VppWgPeer(
2554                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
2555             ).add_vpp_config()
2556         else:
2557             peer_1 = VppWgPeer(
2558                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2559             ).add_vpp_config()
2560         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2561
2562         # create a route to rewrite traffic into the wg interface
2563         if is_ip6:
2564             r1 = VppIpRoute(
2565                 self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
2566             ).add_vpp_config()
2567         else:
2568             r1 = VppIpRoute(
2569                 self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2570             ).add_vpp_config()
2571
2572         # wait for the peer to send a handshake initiation
2573         rxs = self.pg1.get_capture(1, timeout=2)
2574
2575         # prepare and send a handshake response
2576         # expect a keepalive message
2577         resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
2578         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2579
2580         # verify the keepalive message
2581         b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
2582         self.assertEqual(0, len(b))
2583
2584         # prepare and send data packets
2585         # expect to receive them decrypted
2586         if is_ip6:
2587             ip_header = IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
2588         else:
2589             ip_header = IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2590         packet_len_opts = (
2591             2500,  # two buffers
2592             1500,  # one buffer
2593             4500,  # three buffers
2594             1910 if is_ip6 else 1950,  # auth tag is not contiguous
2595         )
2596         txs = []
2597         for l in packet_len_opts:
2598             txs.append(
2599                 peer_1.mk_tunnel_header(self.pg1, is_ip6=is_ip6)
2600                 / Wireguard(message_type=4, reserved_zero=0)
2601                 / WireguardTransport(
2602                     receiver_index=peer_1.sender,
2603                     counter=len(txs),
2604                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
2605                         ip_header / UDP(sport=222, dport=223) / Raw(b"\xfe" * l)
2606                     ),
2607                 )
2608             )
2609         rxs = self.send_and_expect(self.pg1, txs, self.pg0)
2610
2611         # verify decrypted packets
2612         for i, l in enumerate(packet_len_opts):
2613             if is_ip6:
2614                 self.assertEqual(rxs[i][IPv6].dst, self.pg0.remote_ip6)
2615                 self.assertEqual(rxs[i][IPv6].hlim, ip_header.hlim - 1)
2616             else:
2617                 self.assertEqual(rxs[i][IP].dst, self.pg0.remote_ip4)
2618                 self.assertEqual(rxs[i][IP].ttl, ip_header.ttl - 1)
2619             self.assertEqual(len(rxs[i][Raw]), l)
2620             self.assertEqual(bytes(rxs[i][Raw]), b"\xfe" * l)
2621
2622         # prepare and send packets that will be rewritten into the wg interface
2623         # expect data packets sent
2624         if is_ip6:
2625             ip_header = IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
2626         else:
2627             ip_header = IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2628         packet_len_opts = (
2629             2500,  # two buffers
2630             1500,  # one buffer
2631             4500,  # three buffers
2632             1980 if is_ip6 else 2000,  # no free space to write auth tag
2633         )
2634         txs = []
2635         for l in packet_len_opts:
2636             txs.append(
2637                 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2638                 / ip_header
2639                 / UDP(sport=555, dport=556)
2640                 / Raw(b"\xfe" * l)
2641             )
2642         rxs = self.send_and_expect(self.pg0, txs, self.pg1)
2643
2644         # verify the data packets
2645         rxs_decrypted = peer_1.validate_encapped(
2646             rxs, ip_header, is_tunnel_ip6=is_ip6, is_transport_ip6=is_ip6
2647         )
2648
2649         for i, l in enumerate(packet_len_opts):
2650             self.assertEqual(len(rxs_decrypted[i][Raw]), l)
2651             self.assertEqual(bytes(rxs_decrypted[i][Raw]), b"\xfe" * l)
2652
2653         # remove configs
2654         r1.remove_vpp_config()
2655         peer_1.remove_vpp_config()
2656         wg0.remove_vpp_config()
2657
2658     def test_wg_large_packet_v4_sync(self):
2659         """Large packet (v4, sync)"""
2660         self._test_wg_large_packet_tmpl(is_async=False, is_ip6=False)
2661
2662     def test_wg_large_packet_v6_sync(self):
2663         """Large packet (v6, sync)"""
2664         self._test_wg_large_packet_tmpl(is_async=False, is_ip6=True)
2665
2666     def test_wg_large_packet_v4_async(self):
2667         """Large packet (v4, async)"""
2668         self._test_wg_large_packet_tmpl(is_async=True, is_ip6=False)
2669
2670     def test_wg_large_packet_v6_async(self):
2671         """Large packet (v6, async)"""
2672         self._test_wg_large_packet_tmpl(is_async=True, is_ip6=True)
2673
2674     def test_wg_lack_of_buf_headroom(self):
2675         """Lack of buffer's headroom (v6 vxlan over v6 wg)"""
2676         port = 12323
2677
2678         # create wg interface
2679         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
2680         wg0.admin_up()
2681         wg0.config_ip6()
2682
2683         self.pg_enable_capture(self.pg_interfaces)
2684         self.pg_start()
2685
2686         # create a peer
2687         peer_1 = VppWgPeer(
2688             self, wg0, self.pg1.remote_ip6, port + 1, ["::/0"]
2689         ).add_vpp_config()
2690         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2691
2692         # create a route to enable communication between wg interface addresses
2693         r1 = VppIpRoute(
2694             self, wg0.remote_ip6, 128, [VppRoutePath("0.0.0.0", wg0.sw_if_index)]
2695         ).add_vpp_config()
2696
2697         # wait for the peer to send a handshake initiation
2698         rxs = self.pg1.get_capture(1, timeout=2)
2699
2700         # prepare and send a handshake response
2701         # expect a keepalive message
2702         resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=True)
2703         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2704
2705         # verify the keepalive message
2706         b = peer_1.decrypt_transport(rxs[0], is_ip6=True)
2707         self.assertEqual(0, len(b))
2708
2709         # create vxlan interface over the wg interface
2710         vxlan0 = VppVxlanTunnel(self, src=wg0.local_ip6, dst=wg0.remote_ip6, vni=1111)
2711         vxlan0.add_vpp_config()
2712
2713         # create bridge domain
2714         bd1 = VppBridgeDomain(self, bd_id=1)
2715         bd1.add_vpp_config()
2716
2717         # add the vxlan interface and pg0 to the bridge domain
2718         bd1_ports = (
2719             VppBridgeDomainPort(self, bd1, vxlan0).add_vpp_config(),
2720             VppBridgeDomainPort(self, bd1, self.pg0).add_vpp_config(),
2721         )
2722
2723         # prepare and send packets that will be rewritten into the vxlan interface
2724         # expect they to be rewritten into the wg interface then and data packets sent
2725         tx = (
2726             Ether(dst="00:00:00:00:00:01", src="00:00:00:00:00:02")
2727             / IPv6(src="::1", dst="::2", hlim=20)
2728             / UDP(sport=1111, dport=1112)
2729             / Raw(b"\xfe" * 1900)
2730         )
2731         rxs = self.send_and_expect(self.pg0, [tx] * 5, self.pg1)
2732
2733         # verify the data packet
2734         for rx in rxs:
2735             rx_decrypted = IPv6(peer_1.decrypt_transport(rx, is_ip6=True))
2736
2737             self.assertEqual(rx_decrypted[VXLAN].vni, vxlan0.vni)
2738             inner = rx_decrypted[VXLAN].payload
2739
2740             # check the original packet is present
2741             self.assertEqual(inner[IPv6].dst, tx[IPv6].dst)
2742             self.assertEqual(inner[IPv6].hlim, tx[IPv6].hlim)
2743             self.assertEqual(len(inner[Raw]), len(tx[Raw]))
2744             self.assertEqual(bytes(inner[Raw]), bytes(tx[Raw]))
2745
2746         # remove configs
2747         for bdp in bd1_ports:
2748             bdp.remove_vpp_config()
2749         bd1.remove_vpp_config()
2750         vxlan0.remove_vpp_config()
2751         r1.remove_vpp_config()
2752         peer_1.remove_vpp_config()
2753         wg0.remove_vpp_config()
2754
2755
2756 @tag_fixme_vpp_debug
2757 class WireguardHandoffTests(TestWg):
2758     """Wireguard Tests in multi worker setup"""
2759
2760     vpp_worker_count = 2
2761
2762     def test_wg_peer_init(self):
2763         """Handoff"""
2764
2765         port = 12383
2766
2767         # Create interfaces
2768         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2769         wg0.admin_up()
2770         wg0.config_ip4()
2771
2772         self.pg_enable_capture(self.pg_interfaces)
2773         self.pg_start()
2774
2775         peer_1 = VppWgPeer(
2776             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.2.0/24", "10.11.3.0/24"]
2777         ).add_vpp_config()
2778         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2779
2780         r1 = VppIpRoute(
2781             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2782         ).add_vpp_config()
2783
2784         # skip the first automatic handshake
2785         self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
2786
2787         # send a valid handsake init for which we expect a response
2788         p = peer_1.mk_handshake(self.pg1)
2789
2790         rx = self.send_and_expect(self.pg1, [p], self.pg1)
2791
2792         peer_1.consume_response(rx[0])
2793
2794         # send a data packet from the peer through the tunnel
2795         # this completes the handshake and pins the peer to worker 0
2796         p = (
2797             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2798             / UDP(sport=222, dport=223)
2799             / Raw()
2800         )
2801         d = peer_1.encrypt_transport(p)
2802         p = peer_1.mk_tunnel_header(self.pg1) / (
2803             Wireguard(message_type=4, reserved_zero=0)
2804             / WireguardTransport(
2805                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
2806             )
2807         )
2808         rxs = self.send_and_expect(self.pg1, [p], self.pg0, worker=0)
2809
2810         for rx in rxs:
2811             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2812             self.assertEqual(rx[IP].ttl, 19)
2813
2814         # send a packets that are routed into the tunnel
2815         # and pins the peer tp worker 1
2816         pe = (
2817             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2818             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2819             / UDP(sport=555, dport=556)
2820             / Raw(b"\x00" * 80)
2821         )
2822         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=1)
2823         peer_1.validate_encapped(rxs, pe)
2824
2825         # send packets into the tunnel, from the other worker
2826         p = [
2827             (
2828                 peer_1.mk_tunnel_header(self.pg1)
2829                 / Wireguard(message_type=4, reserved_zero=0)
2830                 / WireguardTransport(
2831                     receiver_index=peer_1.sender,
2832                     counter=ii + 1,
2833                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
2834                         (
2835                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2836                             / UDP(sport=222, dport=223)
2837                             / Raw()
2838                         )
2839                     ),
2840                 )
2841             )
2842             for ii in range(255)
2843         ]
2844
2845         rxs = self.send_and_expect(self.pg1, p, self.pg0, worker=1)
2846
2847         for rx in rxs:
2848             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2849             self.assertEqual(rx[IP].ttl, 19)
2850
2851         # send a packets that are routed into the tunnel
2852         # from worker 0
2853         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=0)
2854
2855         peer_1.validate_encapped(rxs, pe)
2856
2857         r1.remove_vpp_config()
2858         peer_1.remove_vpp_config()
2859         wg0.remove_vpp_config()
2860
2861     @unittest.skip("test disabled")
2862     def test_wg_multi_interface(self):
2863         """Multi-tunnel on the same port"""
2864
2865
2866 @unittest.skipIf(
2867     "wireguard" in config.excluded_plugins, "Exclude Wireguard plugin tests"
2868 )
2869 @tag_run_solo
2870 class TestWgFIB(VppTestCase):
2871     """Wireguard FIB Test Case"""
2872
2873     @classmethod
2874     def setUpClass(cls):
2875         super(TestWgFIB, cls).setUpClass()
2876
2877     @classmethod
2878     def tearDownClass(cls):
2879         super(TestWgFIB, cls).tearDownClass()
2880
2881     def setUp(self):
2882         super(TestWgFIB, self).setUp()
2883
2884         self.create_pg_interfaces(range(2))
2885
2886         for i in self.pg_interfaces:
2887             i.admin_up()
2888             i.config_ip4()
2889
2890     def tearDown(self):
2891         for i in self.pg_interfaces:
2892             i.unconfig_ip4()
2893             i.admin_down()
2894         super(TestWgFIB, self).tearDown()
2895
2896     def test_wg_fib_tracking(self):
2897         """FIB tracking"""
2898         port = 12323
2899
2900         # create wg interface
2901         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2902         wg0.admin_up()
2903         wg0.config_ip4()
2904
2905         self.pg_enable_capture(self.pg_interfaces)
2906         self.pg_start()
2907
2908         # create a peer
2909         peer_1 = VppWgPeer(
2910             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2911         ).add_vpp_config()
2912         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2913
2914         # create a route to rewrite traffic into the wg interface
2915         r1 = VppIpRoute(
2916             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2917         ).add_vpp_config()
2918
2919         # resolve ARP and expect the adjacency to update
2920         self.pg1.resolve_arp()
2921
2922         # wait for the peer to send a handshake initiation
2923         rxs = self.pg1.get_capture(2, timeout=6)
2924
2925         # prepare and send a handshake response
2926         # expect a keepalive message
2927         resp = peer_1.consume_init(rxs[1], self.pg1)
2928         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2929
2930         # verify the keepalive message
2931         b = peer_1.decrypt_transport(rxs[0])
2932         self.assertEqual(0, len(b))
2933
2934         # prepare and send a packet that will be rewritten into the wg interface
2935         # expect a data packet sent
2936         p = (
2937             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2938             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2939             / UDP(sport=555, dport=556)
2940             / Raw()
2941         )
2942         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2943
2944         # verify the data packet
2945         peer_1.validate_encapped(rxs, p)
2946
2947         # remove configs
2948         r1.remove_vpp_config()
2949         peer_1.remove_vpp_config()
2950         wg0.remove_vpp_config()