test/test_wireguard.py

   1 #!/usr/bin/env python3
   2 """ Wg tests """
   3
   4 import datetime
   5 import base64
   6 import os
   7
   8 from hashlib import blake2s
   9 from scapy.packet import Packet
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether, ARP
  12 from scapy.layers.inet import IP, UDP
  13 from scapy.layers.inet6 import IPv6
  14 from scapy.contrib.wireguard import (
  15     Wireguard,
  16     WireguardResponse,
  17     WireguardInitiation,
  18     WireguardTransport,
  19     WireguardCookieReply,
  20 )
  21 from cryptography.hazmat.primitives.asymmetric.x25519 import (
  22     X25519PrivateKey,
  23     X25519PublicKey,
  24 )
  25 from cryptography.hazmat.primitives.serialization import (
  26     Encoding,
  27     PrivateFormat,
  28     PublicFormat,
  29     NoEncryption,
  30 )
  31 from cryptography.hazmat.primitives.hashes import BLAKE2s, Hash
  32 from cryptography.hazmat.primitives.hmac import HMAC
  33 from cryptography.hazmat.backends import default_backend
  34 from noise.connection import NoiseConnection, Keypair
  35
  36 from Crypto.Cipher import ChaCha20_Poly1305
  37 from Crypto.Random import get_random_bytes
  38
  39 from vpp_ipip_tun_interface import VppIpIpTunInterface
  40 from vpp_interface import VppInterface
  41 from vpp_pg_interface import is_ipv6_misc
  42 from vpp_ip_route import VppIpRoute, VppRoutePath
  43 from vpp_object import VppObject
  44 from vpp_papi import VppEnum
  45 from framework import is_distro_ubuntu2204, is_distro_debian11, 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         for rx in rxs:
 474             rx = self.decrypt_transport(rx, is_tunnel_ip6)
 475             if is_transport_ip6 is False:
 476                 rx = IP(rx)
 477                 # check the original packet is present
 478                 self._test.assertEqual(rx[IP].dst, tx[IP].dst)
 479                 self._test.assertEqual(rx[IP].ttl, tx[IP].ttl - 1)
 480             else:
 481                 rx = IPv6(rx)
 482                 # check the original packet is present
 483                 self._test.assertEqual(rx[IPv6].dst, tx[IPv6].dst)
 484                 self._test.assertEqual(rx[IPv6].hlim, tx[IPv6].hlim - 1)
 485
 486     def want_events(self):
 487         self._test.vapi.want_wireguard_peer_events(
 488             enable_disable=1,
 489             pid=os.getpid(),
 490             peer_index=self.index,
 491             sw_if_index=self.itf.sw_if_index,
 492         )
 493
 494     def wait_event(self, expect, timeout=5):
 495         rv = self._test.vapi.wait_for_event(timeout, "wireguard_peer_event")
 496         self._test.assertEqual(rv.flags, expect)
 497         self._test.assertEqual(rv.peer_index, self.index)
 498
 499
 500 def is_handshake_init(p):
 501     wg_p = Wireguard(p[Raw])
 502
 503     return wg_p[Wireguard].message_type == 1
 504
 505
 506 class TestWg(VppTestCase):
 507     """Wireguard Test Case"""
 508
 509     error_str = compile(r"Error")
 510
 511     wg4_output_node_name = "/err/wg4-output-tun/"
 512     wg4_input_node_name = "/err/wg4-input/"
 513     wg6_output_node_name = "/err/wg6-output-tun/"
 514     wg6_input_node_name = "/err/wg6-input/"
 515     kp4_error = wg4_output_node_name + "Keypair error"
 516     mac4_error = wg4_input_node_name + "Invalid MAC handshake"
 517     peer4_in_err = wg4_input_node_name + "Peer error"
 518     peer4_out_err = wg4_output_node_name + "Peer error"
 519     kp6_error = wg6_output_node_name + "Keypair error"
 520     mac6_error = wg6_input_node_name + "Invalid MAC handshake"
 521     peer6_in_err = wg6_input_node_name + "Peer error"
 522     peer6_out_err = wg6_output_node_name + "Peer error"
 523     cookie_dec4_err = wg4_input_node_name + "Failed during Cookie decryption"
 524     cookie_dec6_err = wg6_input_node_name + "Failed during Cookie decryption"
 525     ratelimited4_err = wg4_input_node_name + "Handshake ratelimited"
 526     ratelimited6_err = wg6_input_node_name + "Handshake ratelimited"
 527
 528     @classmethod
 529     def setUpClass(cls):
 530         super(TestWg, cls).setUpClass()
 531         if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
 532             cls, "vpp"
 533         ):
 534             return
 535         try:
 536             cls.create_pg_interfaces(range(3))
 537             for i in cls.pg_interfaces:
 538                 i.admin_up()
 539                 i.config_ip4()
 540                 i.config_ip6()
 541                 i.resolve_arp()
 542                 i.resolve_ndp()
 543
 544         except Exception:
 545             super(TestWg, cls).tearDownClass()
 546             raise
 547
 548     @classmethod
 549     def tearDownClass(cls):
 550         super(TestWg, cls).tearDownClass()
 551
 552     def setUp(self):
 553         super(VppTestCase, self).setUp()
 554         self.base_kp4_err = self.statistics.get_err_counter(self.kp4_error)
 555         self.base_mac4_err = self.statistics.get_err_counter(self.mac4_error)
 556         self.base_peer4_in_err = self.statistics.get_err_counter(self.peer4_in_err)
 557         self.base_peer4_out_err = self.statistics.get_err_counter(self.peer4_out_err)
 558         self.base_kp6_err = self.statistics.get_err_counter(self.kp6_error)
 559         self.base_mac6_err = self.statistics.get_err_counter(self.mac6_error)
 560         self.base_peer6_in_err = self.statistics.get_err_counter(self.peer6_in_err)
 561         self.base_peer6_out_err = self.statistics.get_err_counter(self.peer6_out_err)
 562         self.base_cookie_dec4_err = self.statistics.get_err_counter(
 563             self.cookie_dec4_err
 564         )
 565         self.base_cookie_dec6_err = self.statistics.get_err_counter(
 566             self.cookie_dec6_err
 567         )
 568         self.base_ratelimited4_err = self.statistics.get_err_counter(
 569             self.ratelimited4_err
 570         )
 571         self.base_ratelimited6_err = self.statistics.get_err_counter(
 572             self.ratelimited6_err
 573         )
 574
 575     def send_and_assert_no_replies_ignoring_init(
 576         self, intf, pkts, remark="", timeout=None
 577     ):
 578         self.pg_send(intf, pkts)
 579
 580         def _filter_out_fn(p):
 581             return is_ipv6_misc(p) or is_handshake_init(p)
 582
 583         try:
 584             if not timeout:
 585                 timeout = 1
 586             for i in self.pg_interfaces:
 587                 i.assert_nothing_captured(
 588                     timeout=timeout, remark=remark, filter_out_fn=_filter_out_fn
 589                 )
 590                 timeout = 0.1
 591         finally:
 592             pass
 593
 594     def test_wg_interface(self):
 595         """Simple interface creation"""
 596         port = 12312
 597
 598         # Create interface
 599         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 600
 601         self.logger.info(self.vapi.cli("sh int"))
 602
 603         # delete interface
 604         wg0.remove_vpp_config()
 605
 606     def test_handshake_hash(self):
 607         """test hashing an init message"""
 608         # a init packet generated by linux given the key below
 609         h = (
 610             "0100000098b9032b"
 611             "55cc4b39e73c3d24"
 612             "a2a1ab884b524a81"
 613             "1808bb86640fb70d"
 614             "e93154fec1879125"
 615             "ab012624a27f0b75"
 616             "c0a2582f438ddb5f"
 617             "8e768af40b4ab444"
 618             "02f9ff473e1b797e"
 619             "80d39d93c5480c82"
 620             "a3d4510f70396976"
 621             "586fb67300a5167b"
 622             "ae6ca3ff3dfd00eb"
 623             "59be198810f5aa03"
 624             "6abc243d2155ee4f"
 625             "2336483900aef801"
 626             "08752cd700000000"
 627             "0000000000000000"
 628             "00000000"
 629         )
 630
 631         b = bytearray.fromhex(h)
 632         tgt = Wireguard(b)
 633
 634         pubb = base64.b64decode("aRuHFTTxICIQNefp05oKWlJv3zgKxb8+WW7JJMh0jyM=")
 635         pub = X25519PublicKey.from_public_bytes(pubb)
 636
 637         self.assertEqual(pubb, public_key_bytes(pub))
 638
 639         # strip the macs and build a new packet
 640         init = b[0:-32]
 641         mac_key = blake2s(b"mac1----" + public_key_bytes(pub)).digest()
 642         init += blake2s(init, digest_size=16, key=mac_key).digest()
 643         init += b"\x00" * 16
 644
 645         act = Wireguard(init)
 646
 647         self.assertEqual(tgt, act)
 648
 649     def _test_wg_send_cookie_tmpl(self, is_resp, is_ip6):
 650         port = 12323
 651
 652         # create wg interface
 653         if is_ip6:
 654             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 655             wg0.admin_up()
 656             wg0.config_ip6()
 657         else:
 658             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 659             wg0.admin_up()
 660             wg0.config_ip4()
 661
 662         self.pg_enable_capture(self.pg_interfaces)
 663         self.pg_start()
 664
 665         # create a peer
 666         if is_ip6:
 667             peer_1 = VppWgPeer(
 668                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 669             ).add_vpp_config()
 670         else:
 671             peer_1 = VppWgPeer(
 672                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 673             ).add_vpp_config()
 674         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 675
 676         if is_resp:
 677             # skip the first automatic handshake
 678             self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 679
 680             # prepare and send a handshake initiation
 681             # expect the peer to send a handshake response
 682             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 683             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 684         else:
 685             # wait for the peer to send a handshake initiation
 686             rxs = self.pg1.get_capture(1, timeout=2)
 687
 688         # prepare and send a wrong cookie reply
 689         # expect no replies and the cookie error incremented
 690         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 691         cookie.nonce = b"1234567890"
 692         self.send_and_assert_no_replies(self.pg1, [cookie], timeout=0.1)
 693         if is_ip6:
 694             self.assertEqual(
 695                 self.base_cookie_dec6_err + 1,
 696                 self.statistics.get_err_counter(self.cookie_dec6_err),
 697             )
 698         else:
 699             self.assertEqual(
 700                 self.base_cookie_dec4_err + 1,
 701                 self.statistics.get_err_counter(self.cookie_dec4_err),
 702             )
 703
 704         # prepare and send a correct cookie reply
 705         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 706         self.pg_send(self.pg1, [cookie])
 707
 708         # wait for the peer to send a handshake initiation with mac2 set
 709         rxs = self.pg1.get_capture(1, timeout=6)
 710
 711         # verify the initiation and its mac2
 712         peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6, is_mac2=True)
 713
 714         # remove configs
 715         peer_1.remove_vpp_config()
 716         wg0.remove_vpp_config()
 717
 718     def test_wg_send_cookie_on_init_v4(self):
 719         """Send cookie on handshake initiation (v4)"""
 720         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=False)
 721
 722     def test_wg_send_cookie_on_init_v6(self):
 723         """Send cookie on handshake initiation (v6)"""
 724         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=True)
 725
 726     def test_wg_send_cookie_on_resp_v4(self):
 727         """Send cookie on handshake response (v4)"""
 728         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=False)
 729
 730     def test_wg_send_cookie_on_resp_v6(self):
 731         """Send cookie on handshake response (v6)"""
 732         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=True)
 733
 734     def _test_wg_receive_cookie_tmpl(self, is_resp, is_ip6):
 735         port = 12323
 736
 737         # create wg interface
 738         if is_ip6:
 739             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 740             wg0.admin_up()
 741             wg0.config_ip6()
 742         else:
 743             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 744             wg0.admin_up()
 745             wg0.config_ip4()
 746
 747         self.pg_enable_capture(self.pg_interfaces)
 748         self.pg_start()
 749
 750         # create a peer
 751         if is_ip6:
 752             peer_1 = VppWgPeer(
 753                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 754             ).add_vpp_config()
 755         else:
 756             peer_1 = VppWgPeer(
 757                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 758             ).add_vpp_config()
 759         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 760
 761         if is_resp:
 762             # wait for the peer to send a handshake initiation
 763             rxs = self.pg1.get_capture(1, timeout=2)
 764             # prepare and send a bunch of handshake responses
 765             # expect to switch to under load state
 766             resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
 767             txs = [resp] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 768             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 769             # reset noise to be able to turn into initiator later
 770             peer_1.noise_reset()
 771         else:
 772             # skip the first automatic handshake
 773             self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 774
 775             # prepare and send a bunch of handshake initiations
 776             # expect to switch to under load state
 777             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 778             txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 779             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 780
 781         # expect the peer to send a cookie reply
 782         peer_1.consume_cookie(rxs[-1], is_ip6=is_ip6)
 783
 784         # prepare and send a handshake initiation with wrong mac2
 785         # expect a cookie reply
 786         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 787         init.mac2 = b"1234567890"
 788         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 789         peer_1.consume_cookie(rxs[0], is_ip6=is_ip6)
 790
 791         # prepare and send a handshake initiation with correct mac2
 792         # expect a handshake response
 793         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 794         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 795
 796         # verify the response
 797         peer_1.consume_response(rxs[0], is_ip6=is_ip6)
 798
 799         # clear up under load state
 800         self.sleep(UNDER_LOAD_INTERVAL)
 801
 802         # remove configs
 803         peer_1.remove_vpp_config()
 804         wg0.remove_vpp_config()
 805
 806     def test_wg_receive_cookie_on_init_v4(self):
 807         """Receive cookie on handshake initiation (v4)"""
 808         self._test_wg_receive_cookie_tmpl(is_resp=False, is_ip6=False)
 809
 810     def test_wg_receive_cookie_on_init_v6(self):
 811         """Receive cookie on handshake initiation (v6)"""
 812         self._test_wg_receive_cookie_tmpl(is_resp=False, is_ip6=True)
 813
 814     def test_wg_receive_cookie_on_resp_v4(self):
 815         """Receive cookie on handshake response (v4)"""
 816         self._test_wg_receive_cookie_tmpl(is_resp=True, is_ip6=False)
 817
 818     def test_wg_receive_cookie_on_resp_v6(self):
 819         """Receive cookie on handshake response (v6)"""
 820         self._test_wg_receive_cookie_tmpl(is_resp=True, is_ip6=True)
 821
 822     def test_wg_under_load_interval(self):
 823         """Under load interval"""
 824         port = 12323
 825
 826         # create wg interface
 827         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 828         wg0.admin_up()
 829         wg0.config_ip4()
 830
 831         self.pg_enable_capture(self.pg_interfaces)
 832         self.pg_start()
 833
 834         # create a peer
 835         peer_1 = VppWgPeer(
 836             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 837         ).add_vpp_config()
 838         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 839
 840         # skip the first automatic handshake
 841         self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 842
 843         # prepare and send a bunch of handshake initiations
 844         # expect to switch to under load state
 845         init = peer_1.mk_handshake(self.pg1)
 846         txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 847         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 848
 849         # expect the peer to send a cookie reply
 850         peer_1.consume_cookie(rxs[-1])
 851
 852         # sleep till the next counting interval
 853         # expect under load state is still active
 854         self.sleep(HANDSHAKE_COUNTING_INTERVAL)
 855
 856         # prepare and send a handshake initiation with wrong mac2
 857         # expect a cookie reply
 858         init = peer_1.mk_handshake(self.pg1)
 859         init.mac2 = b"1234567890"
 860         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 861         peer_1.consume_cookie(rxs[0])
 862
 863         # sleep till the end of being under load
 864         # expect under load state is over
 865         self.sleep(UNDER_LOAD_INTERVAL - HANDSHAKE_COUNTING_INTERVAL)
 866
 867         # prepare and send a handshake initiation with wrong mac2
 868         # expect a handshake response
 869         init = peer_1.mk_handshake(self.pg1)
 870         init.mac2 = b"1234567890"
 871         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 872
 873         # verify the response
 874         peer_1.consume_response(rxs[0])
 875
 876         # remove configs
 877         peer_1.remove_vpp_config()
 878         wg0.remove_vpp_config()
 879
 880     def _test_wg_handshake_ratelimiting_tmpl(self, is_ip6):
 881         port = 12323
 882
 883         # create wg interface
 884         if is_ip6:
 885             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 886             wg0.admin_up()
 887             wg0.config_ip6()
 888         else:
 889             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 890             wg0.admin_up()
 891             wg0.config_ip4()
 892
 893         self.pg_enable_capture(self.pg_interfaces)
 894         self.pg_start()
 895
 896         # create a peer
 897         if is_ip6:
 898             peer_1 = VppWgPeer(
 899                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 900             ).add_vpp_config()
 901         else:
 902             peer_1 = VppWgPeer(
 903                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 904             ).add_vpp_config()
 905         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 906
 907         # skip the first automatic handshake
 908         self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
 909
 910         # prepare and send a bunch of handshake initiations
 911         # expect to switch to under load state
 912         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 913         txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 914         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 915
 916         # expect the peer to send a cookie reply
 917         peer_1.consume_cookie(rxs[-1], is_ip6=is_ip6)
 918
 919         # prepare and send a bunch of handshake initiations with correct mac2
 920         # expect a handshake response and then ratelimiting
 921         NUM_TO_REJECT = 10
 922         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 923         txs = [init] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + NUM_TO_REJECT)
 924         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 925
 926         if is_ip6:
 927             self.assertEqual(
 928                 self.base_ratelimited6_err + NUM_TO_REJECT,
 929                 self.statistics.get_err_counter(self.ratelimited6_err),
 930             )
 931         else:
 932             self.assertEqual(
 933                 self.base_ratelimited4_err + NUM_TO_REJECT,
 934                 self.statistics.get_err_counter(self.ratelimited4_err),
 935             )
 936
 937         # verify the response
 938         peer_1.consume_response(rxs[0], is_ip6=is_ip6)
 939
 940         # clear up under load state
 941         self.sleep(UNDER_LOAD_INTERVAL)
 942
 943         # remove configs
 944         peer_1.remove_vpp_config()
 945         wg0.remove_vpp_config()
 946
 947     def test_wg_handshake_ratelimiting_v4(self):
 948         """Handshake ratelimiting (v4)"""
 949         self._test_wg_handshake_ratelimiting_tmpl(is_ip6=False)
 950
 951     def test_wg_handshake_ratelimiting_v6(self):
 952         """Handshake ratelimiting (v6)"""
 953         self._test_wg_handshake_ratelimiting_tmpl(is_ip6=True)
 954
 955     def test_wg_handshake_ratelimiting_multi_peer(self):
 956         """Handshake ratelimiting (multiple peer)"""
 957         port = 12323
 958
 959         # create wg interface
 960         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 961         wg0.admin_up()
 962         wg0.config_ip4()
 963
 964         self.pg_enable_capture(self.pg_interfaces)
 965         self.pg_start()
 966
 967         # create two peers
 968         NUM_PEERS = 2
 969         self.pg1.generate_remote_hosts(NUM_PEERS)
 970         self.pg1.configure_ipv4_neighbors()
 971
 972         peer_1 = VppWgPeer(
 973             self, wg0, self.pg1.remote_hosts[0].ip4, port + 1, ["10.11.3.0/24"]
 974         ).add_vpp_config()
 975         peer_2 = VppWgPeer(
 976             self, wg0, self.pg1.remote_hosts[1].ip4, port + 1, ["10.11.4.0/24"]
 977         ).add_vpp_config()
 978         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 2)
 979
 980         # skip the first automatic handshake
 981         self.pg1.get_capture(NUM_PEERS, timeout=HANDSHAKE_JITTER)
 982
 983         # (peer_1) prepare and send a bunch of handshake initiations
 984         # expect not to switch to under load state
 985         init_1 = peer_1.mk_handshake(self.pg1)
 986         txs = [init_1] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 987         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 988
 989         # (peer_1) expect the peer to send a handshake response
 990         peer_1.consume_response(rxs[0])
 991         peer_1.noise_reset()
 992
 993         # (peer_1) send another bunch of handshake initiations
 994         # expect to switch to under load state
 995         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 996
 997         # (peer_1) expect the peer to send a cookie reply
 998         peer_1.consume_cookie(rxs[-1])
 999
1000         # (peer_2) prepare and send a handshake initiation
1001         # expect a cookie reply
1002         init_2 = peer_2.mk_handshake(self.pg1)
1003         rxs = self.send_and_expect(self.pg1, [init_2], self.pg1)
1004         peer_2.consume_cookie(rxs[0])
1005
1006         # (peer_1) (peer_2) prepare and send a bunch of handshake initiations with correct mac2
1007         # expect a handshake response and then ratelimiting
1008         PEER_1_NUM_TO_REJECT = 2
1009         PEER_2_NUM_TO_REJECT = 5
1010         init_1 = peer_1.mk_handshake(self.pg1)
1011         txs = [init_1] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + PEER_1_NUM_TO_REJECT)
1012         init_2 = peer_2.mk_handshake(self.pg1)
1013         txs += [init_2] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + PEER_2_NUM_TO_REJECT)
1014         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
1015
1016         self.assertTrue(
1017             self.base_ratelimited4_err + PEER_1_NUM_TO_REJECT
1018             < self.statistics.get_err_counter(self.ratelimited4_err)
1019             <= self.base_ratelimited4_err + PEER_1_NUM_TO_REJECT + PEER_2_NUM_TO_REJECT
1020         )
1021
1022         # (peer_1) (peer_2) verify the response
1023         peer_1.consume_response(rxs[0])
1024         peer_2.consume_response(rxs[1])
1025
1026         # clear up under load state
1027         self.sleep(UNDER_LOAD_INTERVAL)
1028
1029         # remove configs
1030         peer_1.remove_vpp_config()
1031         peer_2.remove_vpp_config()
1032         wg0.remove_vpp_config()
1033
1034     def _test_wg_peer_roaming_on_handshake_tmpl(self, is_endpoint_set, is_resp, is_ip6):
1035         port = 12323
1036
1037         # create wg interface
1038         if is_ip6:
1039             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1040             wg0.admin_up()
1041             wg0.config_ip6()
1042         else:
1043             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1044             wg0.admin_up()
1045             wg0.config_ip4()
1046
1047         self.pg_enable_capture(self.pg_interfaces)
1048         self.pg_start()
1049
1050         # create more remote hosts
1051         NUM_REMOTE_HOSTS = 2
1052         self.pg1.generate_remote_hosts(NUM_REMOTE_HOSTS)
1053         if is_ip6:
1054             self.pg1.configure_ipv6_neighbors()
1055         else:
1056             self.pg1.configure_ipv4_neighbors()
1057
1058         # create a peer
1059         if is_ip6:
1060             peer_1 = VppWgPeer(
1061                 test=self,
1062                 itf=wg0,
1063                 endpoint=self.pg1.remote_hosts[0].ip6 if is_endpoint_set else "::",
1064                 port=port + 1 if is_endpoint_set else 0,
1065                 allowed_ips=["1::3:0/112"],
1066             ).add_vpp_config()
1067         else:
1068             peer_1 = VppWgPeer(
1069                 test=self,
1070                 itf=wg0,
1071                 endpoint=self.pg1.remote_hosts[0].ip4 if is_endpoint_set else "0.0.0.0",
1072                 port=port + 1 if is_endpoint_set else 0,
1073                 allowed_ips=["10.11.3.0/24"],
1074             ).add_vpp_config()
1075         self.assertTrue(peer_1.query_vpp_config())
1076
1077         if is_resp:
1078             # wait for the peer to send a handshake initiation
1079             rxs = self.pg1.get_capture(1, timeout=2)
1080             # prepare a handshake response
1081             resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
1082             # change endpoint
1083             if is_ip6:
1084                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1085                 resp[IPv6].src, resp[UDP].sport = peer_1.endpoint, peer_1.port
1086             else:
1087                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1088                 resp[IP].src, resp[UDP].sport = peer_1.endpoint, peer_1.port
1089             # send the handshake response
1090             # expect a keepalive message sent to the new endpoint
1091             rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1092             # verify the keepalive message
1093             b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
1094             self.assertEqual(0, len(b))
1095         else:
1096             # change endpoint
1097             if is_ip6:
1098                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1099             else:
1100                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1101             # prepare and send a handshake initiation
1102             # expect a handshake response sent to the new endpoint
1103             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
1104             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
1105             # verify the response
1106             peer_1.consume_response(rxs[0], is_ip6=is_ip6)
1107         self.assertTrue(peer_1.query_vpp_config())
1108
1109         # remove configs
1110         peer_1.remove_vpp_config()
1111         wg0.remove_vpp_config()
1112
1113     def test_wg_peer_roaming_on_init_v4(self):
1114         """Peer roaming on handshake initiation (v4)"""
1115         self._test_wg_peer_roaming_on_handshake_tmpl(
1116             is_endpoint_set=False, is_resp=False, is_ip6=False
1117         )
1118
1119     def test_wg_peer_roaming_on_init_v6(self):
1120         """Peer roaming on handshake initiation (v6)"""
1121         self._test_wg_peer_roaming_on_handshake_tmpl(
1122             is_endpoint_set=False, is_resp=False, is_ip6=True
1123         )
1124
1125     def test_wg_peer_roaming_on_resp_v4(self):
1126         """Peer roaming on handshake response (v4)"""
1127         self._test_wg_peer_roaming_on_handshake_tmpl(
1128             is_endpoint_set=True, is_resp=True, is_ip6=False
1129         )
1130
1131     def test_wg_peer_roaming_on_resp_v6(self):
1132         """Peer roaming on handshake response (v6)"""
1133         self._test_wg_peer_roaming_on_handshake_tmpl(
1134             is_endpoint_set=True, is_resp=True, is_ip6=True
1135         )
1136
1137     def _test_wg_peer_roaming_on_data_tmpl(self, is_async, is_ip6):
1138         self.vapi.wg_set_async_mode(is_async)
1139         port = 12323
1140
1141         # create wg interface
1142         if is_ip6:
1143             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1144             wg0.admin_up()
1145             wg0.config_ip6()
1146         else:
1147             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1148             wg0.admin_up()
1149             wg0.config_ip4()
1150
1151         self.pg_enable_capture(self.pg_interfaces)
1152         self.pg_start()
1153
1154         # create more remote hosts
1155         NUM_REMOTE_HOSTS = 2
1156         self.pg1.generate_remote_hosts(NUM_REMOTE_HOSTS)
1157         if is_ip6:
1158             self.pg1.configure_ipv6_neighbors()
1159         else:
1160             self.pg1.configure_ipv4_neighbors()
1161
1162         # create a peer
1163         if is_ip6:
1164             peer_1 = VppWgPeer(
1165                 self, wg0, self.pg1.remote_hosts[0].ip6, port + 1, ["1::3:0/112"]
1166             ).add_vpp_config()
1167         else:
1168             peer_1 = VppWgPeer(
1169                 self, wg0, self.pg1.remote_hosts[0].ip4, port + 1, ["10.11.3.0/24"]
1170             ).add_vpp_config()
1171         self.assertTrue(peer_1.query_vpp_config())
1172
1173         # create a route to rewrite traffic into the wg interface
1174         if is_ip6:
1175             r1 = VppIpRoute(
1176                 self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1177             ).add_vpp_config()
1178         else:
1179             r1 = VppIpRoute(
1180                 self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1181             ).add_vpp_config()
1182
1183         # wait for the peer to send a handshake initiation
1184         rxs = self.pg1.get_capture(1, timeout=2)
1185
1186         # prepare and send a handshake response
1187         # expect a keepalive message
1188         resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
1189         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1190
1191         # verify the keepalive message
1192         b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
1193         self.assertEqual(0, len(b))
1194
1195         # change endpoint
1196         if is_ip6:
1197             peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1198         else:
1199             peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1200
1201         # prepare and send a data packet
1202         # expect endpoint change
1203         if is_ip6:
1204             ip_header = IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1205         else:
1206             ip_header = IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1207         data = (
1208             peer_1.mk_tunnel_header(self.pg1, is_ip6=is_ip6)
1209             / Wireguard(message_type=4, reserved_zero=0)
1210             / WireguardTransport(
1211                 receiver_index=peer_1.sender,
1212                 counter=0,
1213                 encrypted_encapsulated_packet=peer_1.encrypt_transport(
1214                     ip_header / UDP(sport=222, dport=223) / Raw()
1215                 ),
1216             )
1217         )
1218         rxs = self.send_and_expect(self.pg1, [data], self.pg0)
1219         if is_ip6:
1220             self.assertEqual(rxs[0][IPv6].dst, self.pg0.remote_ip6)
1221             self.assertEqual(rxs[0][IPv6].hlim, 19)
1222         else:
1223             self.assertEqual(rxs[0][IP].dst, self.pg0.remote_ip4)
1224             self.assertEqual(rxs[0][IP].ttl, 19)
1225         self.assertTrue(peer_1.query_vpp_config())
1226
1227         # prepare and send a packet that will be rewritten into the wg interface
1228         # expect a data packet sent to the new endpoint
1229         if is_ip6:
1230             ip_header = IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1231         else:
1232             ip_header = IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1233         p = (
1234             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1235             / ip_header
1236             / UDP(sport=555, dport=556)
1237             / Raw()
1238         )
1239         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
1240
1241         # verify the data packet
1242         peer_1.validate_encapped(rxs, p, is_tunnel_ip6=is_ip6, is_transport_ip6=is_ip6)
1243
1244         # remove configs
1245         r1.remove_vpp_config()
1246         peer_1.remove_vpp_config()
1247         wg0.remove_vpp_config()
1248
1249     def test_wg_peer_roaming_on_data_v4_sync(self):
1250         """Peer roaming on data packet (v4, sync)"""
1251         self._test_wg_peer_roaming_on_data_tmpl(is_async=False, is_ip6=False)
1252
1253     def test_wg_peer_roaming_on_data_v6_sync(self):
1254         """Peer roaming on data packet (v6, sync)"""
1255         self._test_wg_peer_roaming_on_data_tmpl(is_async=False, is_ip6=True)
1256
1257     def test_wg_peer_roaming_on_data_v4_async(self):
1258         """Peer roaming on data packet (v4, async)"""
1259         self._test_wg_peer_roaming_on_data_tmpl(is_async=True, is_ip6=False)
1260
1261     def test_wg_peer_roaming_on_data_v6_async(self):
1262         """Peer roaming on data packet (v6, async)"""
1263         self._test_wg_peer_roaming_on_data_tmpl(is_async=True, is_ip6=True)
1264
1265     def test_wg_peer_resp(self):
1266         """Send handshake response IPv4 tunnel"""
1267         port = 12323
1268
1269         # Create interfaces
1270         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1271         wg0.admin_up()
1272         wg0.config_ip4()
1273
1274         self.pg_enable_capture(self.pg_interfaces)
1275         self.pg_start()
1276
1277         peer_1 = VppWgPeer(
1278             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
1279         ).add_vpp_config()
1280         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1281
1282         r1 = VppIpRoute(
1283             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1284         ).add_vpp_config()
1285
1286         # wait for the peer to send a handshake
1287         rx = self.pg1.get_capture(1, timeout=2)
1288
1289         # consume the handshake in the noise protocol and
1290         # generate the response
1291         resp = peer_1.consume_init(rx[0], self.pg1)
1292
1293         # send the response, get keepalive
1294         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1295
1296         for rx in rxs:
1297             b = peer_1.decrypt_transport(rx)
1298             self.assertEqual(0, len(b))
1299
1300         # send a packets that are routed into the tunnel
1301         p = (
1302             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1303             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1304             / UDP(sport=555, dport=556)
1305             / Raw(b"\x00" * 80)
1306         )
1307
1308         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1309
1310         peer_1.validate_encapped(rxs, p)
1311
1312         # send packets into the tunnel, expect to receive them on
1313         # the other side
1314         p = [
1315             (
1316                 peer_1.mk_tunnel_header(self.pg1)
1317                 / Wireguard(message_type=4, reserved_zero=0)
1318                 / WireguardTransport(
1319                     receiver_index=peer_1.sender,
1320                     counter=ii,
1321                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1322                         (
1323                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1324                             / UDP(sport=222, dport=223)
1325                             / Raw()
1326                         )
1327                     ),
1328                 )
1329             )
1330             for ii in range(255)
1331         ]
1332
1333         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1334
1335         for rx in rxs:
1336             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1337             self.assertEqual(rx[IP].ttl, 19)
1338
1339         r1.remove_vpp_config()
1340         peer_1.remove_vpp_config()
1341         wg0.remove_vpp_config()
1342
1343     def test_wg_peer_resp_ipv6(self):
1344         """Send handshake response IPv6 tunnel"""
1345         port = 12323
1346
1347         # Create interfaces
1348         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1349         wg0.admin_up()
1350         wg0.config_ip4()
1351
1352         self.pg_enable_capture(self.pg_interfaces)
1353         self.pg_start()
1354
1355         peer_1 = VppWgPeer(
1356             self, wg0, self.pg1.remote_ip6, port + 1, ["10.11.3.0/24"]
1357         ).add_vpp_config()
1358         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1359
1360         r1 = VppIpRoute(
1361             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1362         ).add_vpp_config()
1363
1364         # wait for the peer to send a handshake
1365         rx = self.pg1.get_capture(1, timeout=2)
1366
1367         # consume the handshake in the noise protocol and
1368         # generate the response
1369         resp = peer_1.consume_init(rx[0], self.pg1, is_ip6=True)
1370
1371         # send the response, get keepalive
1372         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1373
1374         for rx in rxs:
1375             b = peer_1.decrypt_transport(rx, True)
1376             self.assertEqual(0, len(b))
1377
1378         # send a packets that are routed into the tunnel
1379         p = (
1380             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1381             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1382             / UDP(sport=555, dport=556)
1383             / Raw(b"\x00" * 80)
1384         )
1385
1386         rxs = self.send_and_expect(self.pg0, p * 2, self.pg1)
1387         peer_1.validate_encapped(rxs, p, True)
1388
1389         # send packets into the tunnel, expect to receive them on
1390         # the other side
1391         p = [
1392             (
1393                 peer_1.mk_tunnel_header(self.pg1, True)
1394                 / Wireguard(message_type=4, reserved_zero=0)
1395                 / WireguardTransport(
1396                     receiver_index=peer_1.sender,
1397                     counter=ii,
1398                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1399                         (
1400                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1401                             / UDP(sport=222, dport=223)
1402                             / Raw()
1403                         )
1404                     ),
1405                 )
1406             )
1407             for ii in range(255)
1408         ]
1409
1410         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1411
1412         for rx in rxs:
1413             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1414             self.assertEqual(rx[IP].ttl, 19)
1415
1416         r1.remove_vpp_config()
1417         peer_1.remove_vpp_config()
1418         wg0.remove_vpp_config()
1419
1420     def test_wg_peer_v4o4(self):
1421         """Test v4o4"""
1422
1423         port = 12333
1424
1425         # Create interfaces
1426         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1427         wg0.admin_up()
1428         wg0.config_ip4()
1429
1430         peer_1 = VppWgPeer(
1431             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
1432         ).add_vpp_config()
1433         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1434
1435         r1 = VppIpRoute(
1436             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1437         ).add_vpp_config()
1438         r2 = VppIpRoute(
1439             self, "20.22.3.0", 24, [VppRoutePath("20.22.3.1", wg0.sw_if_index)]
1440         ).add_vpp_config()
1441
1442         # route a packet into the wg interface
1443         #  use the allowed-ip prefix
1444         #  this is dropped because the peer is not initiated
1445         p = (
1446             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1447             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1448             / UDP(sport=555, dport=556)
1449             / Raw()
1450         )
1451         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1452         self.assertEqual(
1453             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1454         )
1455
1456         # route a packet into the wg interface
1457         #  use a not allowed-ip prefix
1458         #  this is dropped because there is no matching peer
1459         p = (
1460             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1461             / IP(src=self.pg0.remote_ip4, dst="20.22.3.2")
1462             / UDP(sport=555, dport=556)
1463             / Raw()
1464         )
1465         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1466         self.assertEqual(
1467             self.base_peer4_out_err + 1,
1468             self.statistics.get_err_counter(self.peer4_out_err),
1469         )
1470
1471         # send a handsake from the peer with an invalid MAC
1472         p = peer_1.mk_handshake(self.pg1)
1473         p[WireguardInitiation].mac1 = b"foobar"
1474         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1475         self.assertEqual(
1476             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1477         )
1478
1479         # send a handsake from the peer but signed by the wrong key.
1480         p = peer_1.mk_handshake(
1481             self.pg1, False, X25519PrivateKey.generate().public_key()
1482         )
1483         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1484         self.assertEqual(
1485             self.base_peer4_in_err + 1,
1486             self.statistics.get_err_counter(self.peer4_in_err),
1487         )
1488
1489         # send a valid handsake init for which we expect a response
1490         p = peer_1.mk_handshake(self.pg1)
1491
1492         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1493
1494         peer_1.consume_response(rx[0])
1495
1496         # route a packet into the wg interface
1497         #  this is dropped because the peer is still not initiated
1498         p = (
1499             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1500             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1501             / UDP(sport=555, dport=556)
1502             / Raw()
1503         )
1504         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1505         self.assertEqual(
1506             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1507         )
1508
1509         # send a data packet from the peer through the tunnel
1510         # this completes the handshake
1511         p = (
1512             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1513             / UDP(sport=222, dport=223)
1514             / Raw()
1515         )
1516         d = peer_1.encrypt_transport(p)
1517         p = peer_1.mk_tunnel_header(self.pg1) / (
1518             Wireguard(message_type=4, reserved_zero=0)
1519             / WireguardTransport(
1520                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1521             )
1522         )
1523         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1524
1525         for rx in rxs:
1526             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1527             self.assertEqual(rx[IP].ttl, 19)
1528
1529         # send a packets that are routed into the tunnel
1530         p = (
1531             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1532             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1533             / UDP(sport=555, dport=556)
1534             / Raw(b"\x00" * 80)
1535         )
1536
1537         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1538
1539         for rx in rxs:
1540             rx = IP(peer_1.decrypt_transport(rx))
1541
1542             # check the original packet is present
1543             self.assertEqual(rx[IP].dst, p[IP].dst)
1544             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1545
1546         # send packets into the tunnel, expect to receive them on
1547         # the other side
1548         p = [
1549             (
1550                 peer_1.mk_tunnel_header(self.pg1)
1551                 / Wireguard(message_type=4, reserved_zero=0)
1552                 / WireguardTransport(
1553                     receiver_index=peer_1.sender,
1554                     counter=ii + 1,
1555                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1556                         (
1557                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1558                             / UDP(sport=222, dport=223)
1559                             / Raw()
1560                         )
1561                     ),
1562                 )
1563             )
1564             for ii in range(255)
1565         ]
1566
1567         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1568
1569         for rx in rxs:
1570             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1571             self.assertEqual(rx[IP].ttl, 19)
1572
1573         r1.remove_vpp_config()
1574         r2.remove_vpp_config()
1575         peer_1.remove_vpp_config()
1576         wg0.remove_vpp_config()
1577
1578     def test_wg_peer_v6o6(self):
1579         """Test v6o6"""
1580
1581         port = 12343
1582
1583         # Create interfaces
1584         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1585         wg0.admin_up()
1586         wg0.config_ip6()
1587
1588         peer_1 = VppWgPeer(
1589             self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
1590         ).add_vpp_config()
1591         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1592
1593         r1 = VppIpRoute(
1594             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1595         ).add_vpp_config()
1596         r2 = VppIpRoute(
1597             self, "22::3:0", 112, [VppRoutePath("22::3:1", wg0.sw_if_index)]
1598         ).add_vpp_config()
1599
1600         # route a packet into the wg interface
1601         #  use the allowed-ip prefix
1602         #  this is dropped because the peer is not initiated
1603
1604         p = (
1605             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1606             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1607             / UDP(sport=555, dport=556)
1608             / Raw()
1609         )
1610         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1611
1612         self.assertEqual(
1613             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1614         )
1615
1616         # route a packet into the wg interface
1617         #  use a not allowed-ip prefix
1618         #  this is dropped because there is no matching peer
1619         p = (
1620             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1621             / IPv6(src=self.pg0.remote_ip6, dst="22::3:2")
1622             / UDP(sport=555, dport=556)
1623             / Raw()
1624         )
1625         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1626         self.assertEqual(
1627             self.base_peer6_out_err + 1,
1628             self.statistics.get_err_counter(self.peer6_out_err),
1629         )
1630
1631         # send a handsake from the peer with an invalid MAC
1632         p = peer_1.mk_handshake(self.pg1, True)
1633         p[WireguardInitiation].mac1 = b"foobar"
1634         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1635
1636         self.assertEqual(
1637             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1638         )
1639
1640         # send a handsake from the peer but signed by the wrong key.
1641         p = peer_1.mk_handshake(
1642             self.pg1, True, X25519PrivateKey.generate().public_key()
1643         )
1644         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1645         self.assertEqual(
1646             self.base_peer6_in_err + 1,
1647             self.statistics.get_err_counter(self.peer6_in_err),
1648         )
1649
1650         # send a valid handsake init for which we expect a response
1651         p = peer_1.mk_handshake(self.pg1, True)
1652
1653         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1654
1655         peer_1.consume_response(rx[0], True)
1656
1657         # route a packet into the wg interface
1658         #  this is dropped because the peer is still not initiated
1659         p = (
1660             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1661             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1662             / UDP(sport=555, dport=556)
1663             / Raw()
1664         )
1665         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1666         self.assertEqual(
1667             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1668         )
1669
1670         # send a data packet from the peer through the tunnel
1671         # this completes the handshake
1672         p = (
1673             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1674             / UDP(sport=222, dport=223)
1675             / Raw()
1676         )
1677         d = peer_1.encrypt_transport(p)
1678         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1679             Wireguard(message_type=4, reserved_zero=0)
1680             / WireguardTransport(
1681                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1682             )
1683         )
1684         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1685
1686         for rx in rxs:
1687             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1688             self.assertEqual(rx[IPv6].hlim, 19)
1689
1690         # send a packets that are routed into the tunnel
1691         p = (
1692             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1693             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1694             / UDP(sport=555, dport=556)
1695             / Raw(b"\x00" * 80)
1696         )
1697
1698         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1699
1700         for rx in rxs:
1701             rx = IPv6(peer_1.decrypt_transport(rx, True))
1702
1703             # check the original packet is present
1704             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1705             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1706
1707         # send packets into the tunnel, expect to receive them on
1708         # the other side
1709         p = [
1710             (
1711                 peer_1.mk_tunnel_header(self.pg1, True)
1712                 / Wireguard(message_type=4, reserved_zero=0)
1713                 / WireguardTransport(
1714                     receiver_index=peer_1.sender,
1715                     counter=ii + 1,
1716                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1717                         (
1718                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1719                             / UDP(sport=222, dport=223)
1720                             / Raw()
1721                         )
1722                     ),
1723                 )
1724             )
1725             for ii in range(255)
1726         ]
1727
1728         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1729
1730         for rx in rxs:
1731             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1732             self.assertEqual(rx[IPv6].hlim, 19)
1733
1734         r1.remove_vpp_config()
1735         r2.remove_vpp_config()
1736         peer_1.remove_vpp_config()
1737         wg0.remove_vpp_config()
1738
1739     def test_wg_peer_v6o4(self):
1740         """Test v6o4"""
1741
1742         port = 12353
1743
1744         # Create interfaces
1745         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1746         wg0.admin_up()
1747         wg0.config_ip6()
1748
1749         peer_1 = VppWgPeer(
1750             self, wg0, self.pg1.remote_ip4, port + 1, ["1::3:0/112"]
1751         ).add_vpp_config()
1752         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1753
1754         r1 = VppIpRoute(
1755             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1756         ).add_vpp_config()
1757
1758         # route a packet into the wg interface
1759         #  use the allowed-ip prefix
1760         #  this is dropped because the peer is not initiated
1761         p = (
1762             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1763             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1764             / UDP(sport=555, dport=556)
1765             / Raw()
1766         )
1767         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1768         self.assertEqual(
1769             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1770         )
1771
1772         # send a handsake from the peer with an invalid MAC
1773         p = peer_1.mk_handshake(self.pg1)
1774         p[WireguardInitiation].mac1 = b"foobar"
1775         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1776
1777         self.assertEqual(
1778             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1779         )
1780
1781         # send a handsake from the peer but signed by the wrong key.
1782         p = peer_1.mk_handshake(
1783             self.pg1, False, X25519PrivateKey.generate().public_key()
1784         )
1785         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1786         self.assertEqual(
1787             self.base_peer4_in_err + 1,
1788             self.statistics.get_err_counter(self.peer4_in_err),
1789         )
1790
1791         # send a valid handsake init for which we expect a response
1792         p = peer_1.mk_handshake(self.pg1)
1793
1794         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1795
1796         peer_1.consume_response(rx[0])
1797
1798         # route a packet into the wg interface
1799         #  this is dropped because the peer is still not initiated
1800         p = (
1801             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1802             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1803             / UDP(sport=555, dport=556)
1804             / Raw()
1805         )
1806         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1807         self.assertEqual(
1808             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1809         )
1810
1811         # send a data packet from the peer through the tunnel
1812         # this completes the handshake
1813         p = (
1814             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1815             / UDP(sport=222, dport=223)
1816             / Raw()
1817         )
1818         d = peer_1.encrypt_transport(p)
1819         p = peer_1.mk_tunnel_header(self.pg1) / (
1820             Wireguard(message_type=4, reserved_zero=0)
1821             / WireguardTransport(
1822                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1823             )
1824         )
1825         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1826
1827         for rx in rxs:
1828             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1829             self.assertEqual(rx[IPv6].hlim, 19)
1830
1831         # send a packets that are routed into the tunnel
1832         p = (
1833             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1834             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1835             / UDP(sport=555, dport=556)
1836             / Raw(b"\x00" * 80)
1837         )
1838
1839         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1840
1841         for rx in rxs:
1842             rx = IPv6(peer_1.decrypt_transport(rx))
1843
1844             # check the original packet is present
1845             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1846             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1847
1848         # send packets into the tunnel, expect to receive them on
1849         # the other side
1850         p = [
1851             (
1852                 peer_1.mk_tunnel_header(self.pg1)
1853                 / Wireguard(message_type=4, reserved_zero=0)
1854                 / WireguardTransport(
1855                     receiver_index=peer_1.sender,
1856                     counter=ii + 1,
1857                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1858                         (
1859                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1860                             / UDP(sport=222, dport=223)
1861                             / Raw()
1862                         )
1863                     ),
1864                 )
1865             )
1866             for ii in range(255)
1867         ]
1868
1869         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1870
1871         for rx in rxs:
1872             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1873             self.assertEqual(rx[IPv6].hlim, 19)
1874
1875         r1.remove_vpp_config()
1876         peer_1.remove_vpp_config()
1877         wg0.remove_vpp_config()
1878
1879     def test_wg_peer_v4o6(self):
1880         """Test v4o6"""
1881
1882         port = 12363
1883
1884         # Create interfaces
1885         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1886         wg0.admin_up()
1887         wg0.config_ip4()
1888
1889         peer_1 = VppWgPeer(
1890             self, wg0, self.pg1.remote_ip6, port + 1, ["10.11.3.0/24"]
1891         ).add_vpp_config()
1892         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1893
1894         r1 = VppIpRoute(
1895             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1896         ).add_vpp_config()
1897
1898         # route a packet into the wg interface
1899         #  use the allowed-ip prefix
1900         #  this is dropped because the peer is not initiated
1901         p = (
1902             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1903             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1904             / UDP(sport=555, dport=556)
1905             / Raw()
1906         )
1907         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1908         self.assertEqual(
1909             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1910         )
1911
1912         # send a handsake from the peer with an invalid MAC
1913         p = peer_1.mk_handshake(self.pg1, True)
1914         p[WireguardInitiation].mac1 = b"foobar"
1915         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1916         self.assertEqual(
1917             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1918         )
1919
1920         # send a handsake from the peer but signed by the wrong key.
1921         p = peer_1.mk_handshake(
1922             self.pg1, True, X25519PrivateKey.generate().public_key()
1923         )
1924         self.send_and_assert_no_replies_ignoring_init(self.pg1, [p])
1925         self.assertEqual(
1926             self.base_peer6_in_err + 1,
1927             self.statistics.get_err_counter(self.peer6_in_err),
1928         )
1929
1930         # send a valid handsake init for which we expect a response
1931         p = peer_1.mk_handshake(self.pg1, True)
1932
1933         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1934
1935         peer_1.consume_response(rx[0], True)
1936
1937         # route a packet into the wg interface
1938         #  this is dropped because the peer is still not initiated
1939         p = (
1940             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1941             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1942             / UDP(sport=555, dport=556)
1943             / Raw()
1944         )
1945         self.send_and_assert_no_replies_ignoring_init(self.pg0, [p])
1946         self.assertEqual(
1947             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1948         )
1949
1950         # send a data packet from the peer through the tunnel
1951         # this completes the handshake
1952         p = (
1953             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1954             / UDP(sport=222, dport=223)
1955             / Raw()
1956         )
1957         d = peer_1.encrypt_transport(p)
1958         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1959             Wireguard(message_type=4, reserved_zero=0)
1960             / WireguardTransport(
1961                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1962             )
1963         )
1964         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1965
1966         for rx in rxs:
1967             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1968             self.assertEqual(rx[IP].ttl, 19)
1969
1970         # send a packets that are routed into the tunnel
1971         p = (
1972             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1973             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1974             / UDP(sport=555, dport=556)
1975             / Raw(b"\x00" * 80)
1976         )
1977
1978         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1979
1980         for rx in rxs:
1981             rx = IP(peer_1.decrypt_transport(rx, True))
1982
1983             # check the original packet is present
1984             self.assertEqual(rx[IP].dst, p[IP].dst)
1985             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1986
1987         # send packets into the tunnel, expect to receive them on
1988         # the other side
1989         p = [
1990             (
1991                 peer_1.mk_tunnel_header(self.pg1, True)
1992                 / Wireguard(message_type=4, reserved_zero=0)
1993                 / WireguardTransport(
1994                     receiver_index=peer_1.sender,
1995                     counter=ii + 1,
1996                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1997                         (
1998                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1999                             / UDP(sport=222, dport=223)
2000                             / Raw()
2001                         )
2002                     ),
2003                 )
2004             )
2005             for ii in range(255)
2006         ]
2007
2008         rxs = self.send_and_expect(self.pg1, p, self.pg0)
2009
2010         for rx in rxs:
2011             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2012             self.assertEqual(rx[IP].ttl, 19)
2013
2014         r1.remove_vpp_config()
2015         peer_1.remove_vpp_config()
2016         wg0.remove_vpp_config()
2017
2018     def test_wg_multi_peer(self):
2019         """multiple peer setup"""
2020         port = 12373
2021
2022         # Create interfaces
2023         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2024         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
2025         wg0.admin_up()
2026         wg1.admin_up()
2027
2028         # Check peer counter
2029         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
2030
2031         self.pg_enable_capture(self.pg_interfaces)
2032         self.pg_start()
2033
2034         # Create many peers on sencond interface
2035         NUM_PEERS = 16
2036         self.pg2.generate_remote_hosts(NUM_PEERS)
2037         self.pg2.configure_ipv4_neighbors()
2038         self.pg1.generate_remote_hosts(NUM_PEERS)
2039         self.pg1.configure_ipv4_neighbors()
2040
2041         peers_1 = []
2042         peers_2 = []
2043         routes_1 = []
2044         routes_2 = []
2045         for i in range(NUM_PEERS):
2046             peers_1.append(
2047                 VppWgPeer(
2048                     self,
2049                     wg0,
2050                     self.pg1.remote_hosts[i].ip4,
2051                     port + 1 + i,
2052                     ["10.0.%d.4/32" % i],
2053                 ).add_vpp_config()
2054             )
2055             routes_1.append(
2056                 VppIpRoute(
2057                     self,
2058                     "10.0.%d.4" % i,
2059                     32,
2060                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
2061                 ).add_vpp_config()
2062             )
2063
2064             peers_2.append(
2065                 VppWgPeer(
2066                     self,
2067                     wg1,
2068                     self.pg2.remote_hosts[i].ip4,
2069                     port + 100 + i,
2070                     ["10.100.%d.4/32" % i],
2071                 ).add_vpp_config()
2072             )
2073             routes_2.append(
2074                 VppIpRoute(
2075                     self,
2076                     "10.100.%d.4" % i,
2077                     32,
2078                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
2079                 ).add_vpp_config()
2080             )
2081
2082         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
2083
2084         self.logger.info(self.vapi.cli("show wireguard peer"))
2085         self.logger.info(self.vapi.cli("show wireguard interface"))
2086         self.logger.info(self.vapi.cli("show adj 37"))
2087         self.logger.info(self.vapi.cli("sh ip fib 172.16.3.17"))
2088         self.logger.info(self.vapi.cli("sh ip fib 10.11.3.0"))
2089
2090         # remove routes
2091         for r in routes_1:
2092             r.remove_vpp_config()
2093         for r in routes_2:
2094             r.remove_vpp_config()
2095
2096         # remove peers
2097         for p in peers_1:
2098             self.assertTrue(p.query_vpp_config())
2099             p.remove_vpp_config()
2100         for p in peers_2:
2101             self.assertTrue(p.query_vpp_config())
2102             p.remove_vpp_config()
2103
2104         wg0.remove_vpp_config()
2105         wg1.remove_vpp_config()
2106
2107     def test_wg_multi_interface(self):
2108         """Multi-tunnel on the same port"""
2109         port = 12500
2110
2111         # Create many wireguard interfaces
2112         NUM_IFS = 4
2113         self.pg1.generate_remote_hosts(NUM_IFS)
2114         self.pg1.configure_ipv4_neighbors()
2115         self.pg0.generate_remote_hosts(NUM_IFS)
2116         self.pg0.configure_ipv4_neighbors()
2117
2118         self.pg_enable_capture(self.pg_interfaces)
2119         self.pg_start()
2120
2121         # Create interfaces with a peer on each
2122         peers = []
2123         routes = []
2124         wg_ifs = []
2125         for i in range(NUM_IFS):
2126             # Use the same port for each interface
2127             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2128             wg0.admin_up()
2129             wg0.config_ip4()
2130             wg_ifs.append(wg0)
2131             peers.append(
2132                 VppWgPeer(
2133                     self,
2134                     wg0,
2135                     self.pg1.remote_hosts[i].ip4,
2136                     port + 1 + i,
2137                     ["10.0.%d.0/24" % i],
2138                 ).add_vpp_config()
2139             )
2140
2141             routes.append(
2142                 VppIpRoute(
2143                     self,
2144                     "10.0.%d.0" % i,
2145                     24,
2146                     [VppRoutePath("10.0.%d.4" % i, wg0.sw_if_index)],
2147                 ).add_vpp_config()
2148             )
2149
2150         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_IFS)
2151
2152         # skip the first automatic handshake
2153         self.pg1.get_capture(NUM_IFS, timeout=HANDSHAKE_JITTER)
2154
2155         for i in range(NUM_IFS):
2156             # send a valid handsake init for which we expect a response
2157             p = peers[i].mk_handshake(self.pg1)
2158             rx = self.send_and_expect(self.pg1, [p], self.pg1)
2159             peers[i].consume_response(rx[0])
2160
2161             # send a data packet from the peer through the tunnel
2162             # this completes the handshake
2163             p = (
2164                 IP(src="10.0.%d.4" % i, dst=self.pg0.remote_hosts[i].ip4, ttl=20)
2165                 / UDP(sport=222, dport=223)
2166                 / Raw()
2167             )
2168             d = peers[i].encrypt_transport(p)
2169             p = peers[i].mk_tunnel_header(self.pg1) / (
2170                 Wireguard(message_type=4, reserved_zero=0)
2171                 / WireguardTransport(
2172                     receiver_index=peers[i].sender,
2173                     counter=0,
2174                     encrypted_encapsulated_packet=d,
2175                 )
2176             )
2177             rxs = self.send_and_expect(self.pg1, [p], self.pg0)
2178             for rx in rxs:
2179                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
2180                 self.assertEqual(rx[IP].ttl, 19)
2181
2182         # send a packets that are routed into the tunnel
2183         for i in range(NUM_IFS):
2184             p = (
2185                 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2186                 / IP(src=self.pg0.remote_hosts[i].ip4, dst="10.0.%d.4" % i)
2187                 / UDP(sport=555, dport=556)
2188                 / Raw(b"\x00" * 80)
2189             )
2190
2191             rxs = self.send_and_expect(self.pg0, p * 64, self.pg1)
2192
2193             for rx in rxs:
2194                 rx = IP(peers[i].decrypt_transport(rx))
2195
2196                 # check the oringial packet is present
2197                 self.assertEqual(rx[IP].dst, p[IP].dst)
2198                 self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
2199
2200         # send packets into the tunnel
2201         for i in range(NUM_IFS):
2202             p = [
2203                 (
2204                     peers[i].mk_tunnel_header(self.pg1)
2205                     / Wireguard(message_type=4, reserved_zero=0)
2206                     / WireguardTransport(
2207                         receiver_index=peers[i].sender,
2208                         counter=ii + 1,
2209                         encrypted_encapsulated_packet=peers[i].encrypt_transport(
2210                             (
2211                                 IP(
2212                                     src="10.0.%d.4" % i,
2213                                     dst=self.pg0.remote_hosts[i].ip4,
2214                                     ttl=20,
2215                                 )
2216                                 / UDP(sport=222, dport=223)
2217                                 / Raw()
2218                             )
2219                         ),
2220                     )
2221                 )
2222                 for ii in range(64)
2223             ]
2224
2225             rxs = self.send_and_expect(self.pg1, p, self.pg0)
2226
2227             for rx in rxs:
2228                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
2229                 self.assertEqual(rx[IP].ttl, 19)
2230
2231         for r in routes:
2232             r.remove_vpp_config()
2233         for p in peers:
2234             p.remove_vpp_config()
2235         for i in wg_ifs:
2236             i.remove_vpp_config()
2237
2238     def test_wg_event(self):
2239         """Test events"""
2240         port = 12600
2241         ESTABLISHED_FLAG = (
2242             VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_ESTABLISHED
2243         )
2244         DEAD_FLAG = VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_STATUS_DEAD
2245
2246         # Create interfaces
2247         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2248         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
2249         wg0.admin_up()
2250         wg1.admin_up()
2251
2252         # Check peer counter
2253         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
2254
2255         self.pg_enable_capture(self.pg_interfaces)
2256         self.pg_start()
2257
2258         # Create peers
2259         NUM_PEERS = 2
2260         self.pg2.generate_remote_hosts(NUM_PEERS)
2261         self.pg2.configure_ipv4_neighbors()
2262         self.pg1.generate_remote_hosts(NUM_PEERS)
2263         self.pg1.configure_ipv4_neighbors()
2264
2265         peers_0 = []
2266         peers_1 = []
2267         routes_0 = []
2268         routes_1 = []
2269         for i in range(NUM_PEERS):
2270             peers_0.append(
2271                 VppWgPeer(
2272                     self,
2273                     wg0,
2274                     self.pg1.remote_hosts[i].ip4,
2275                     port + 1 + i,
2276                     ["10.0.%d.4/32" % i],
2277                 ).add_vpp_config()
2278             )
2279             routes_0.append(
2280                 VppIpRoute(
2281                     self,
2282                     "10.0.%d.4" % i,
2283                     32,
2284                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
2285                 ).add_vpp_config()
2286             )
2287
2288             peers_1.append(
2289                 VppWgPeer(
2290                     self,
2291                     wg1,
2292                     self.pg2.remote_hosts[i].ip4,
2293                     port + 100 + i,
2294                     ["10.100.%d.4/32" % i],
2295                 ).add_vpp_config()
2296             )
2297             routes_1.append(
2298                 VppIpRoute(
2299                     self,
2300                     "10.100.%d.4" % i,
2301                     32,
2302                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
2303                 ).add_vpp_config()
2304             )
2305
2306         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
2307
2308         # skip the first automatic handshake
2309         self.pg1.get_capture(NUM_PEERS, timeout=HANDSHAKE_JITTER)
2310         self.pg2.get_capture(NUM_PEERS, timeout=HANDSHAKE_JITTER)
2311
2312         # Want events from the first perr of wg0
2313         # and from all wg1 peers
2314         peers_0[0].want_events()
2315         wg1.want_events()
2316
2317         for i in range(NUM_PEERS):
2318             # send a valid handsake init for which we expect a response
2319             p = peers_0[i].mk_handshake(self.pg1)
2320             rx = self.send_and_expect(self.pg1, [p], self.pg1)
2321             peers_0[i].consume_response(rx[0])
2322             if i == 0:
2323                 peers_0[0].wait_event(ESTABLISHED_FLAG)
2324
2325             p = peers_1[i].mk_handshake(self.pg2)
2326             rx = self.send_and_expect(self.pg2, [p], self.pg2)
2327             peers_1[i].consume_response(rx[0])
2328
2329         wg1.wait_events(ESTABLISHED_FLAG, [peers_1[0].index, peers_1[1].index])
2330
2331         # remove routes
2332         for r in routes_0:
2333             r.remove_vpp_config()
2334         for r in routes_1:
2335             r.remove_vpp_config()
2336
2337         # remove peers
2338         for i in range(NUM_PEERS):
2339             self.assertTrue(peers_0[i].query_vpp_config())
2340             peers_0[i].remove_vpp_config()
2341             if i == 0:
2342                 peers_0[i].wait_event(0)
2343                 peers_0[i].wait_event(DEAD_FLAG)
2344         for p in peers_1:
2345             self.assertTrue(p.query_vpp_config())
2346             p.remove_vpp_config()
2347             p.wait_event(0)
2348             p.wait_event(DEAD_FLAG)
2349
2350         wg0.remove_vpp_config()
2351         wg1.remove_vpp_config()
2352
2353     def test_wg_sending_handshake_when_admin_down(self):
2354         """Sending handshake when admin down"""
2355         port = 12323
2356
2357         # create wg interface
2358         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2359         wg0.config_ip4()
2360
2361         # create a peer
2362         peer_1 = VppWgPeer(
2363             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2364         ).add_vpp_config()
2365         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2366
2367         self.pg_enable_capture(self.pg_interfaces)
2368         self.pg_start()
2369
2370         # wait for the peer to send a handshake initiation
2371         # expect no handshakes
2372         for i in range(2):
2373             self.pg1.assert_nothing_captured(remark="handshake packet(s) sent")
2374
2375         self.pg_enable_capture(self.pg_interfaces)
2376         self.pg_start()
2377
2378         # administratively enable the wg interface
2379         # expect the peer to send a handshake initiation
2380         wg0.admin_up()
2381         rxs = self.pg1.get_capture(1, timeout=2)
2382         peer_1.consume_init(rxs[0], self.pg1)
2383
2384         self.pg_enable_capture(self.pg_interfaces)
2385         self.pg_start()
2386
2387         # administratively disable the wg interface
2388         # expect no handshakes
2389         wg0.admin_down()
2390         for i in range(6):
2391             self.pg1.assert_nothing_captured(remark="handshake packet(s) sent")
2392
2393         # remove configs
2394         peer_1.remove_vpp_config()
2395         wg0.remove_vpp_config()
2396
2397     def test_wg_sending_data_when_admin_down(self):
2398         """Sending data when admin down"""
2399         port = 12323
2400
2401         # create wg interface
2402         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2403         wg0.admin_up()
2404         wg0.config_ip4()
2405
2406         self.pg_enable_capture(self.pg_interfaces)
2407         self.pg_start()
2408
2409         # create a peer
2410         peer_1 = VppWgPeer(
2411             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2412         ).add_vpp_config()
2413         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2414
2415         # create a route to rewrite traffic into the wg interface
2416         r1 = VppIpRoute(
2417             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2418         ).add_vpp_config()
2419
2420         # wait for the peer to send a handshake initiation
2421         rxs = self.pg1.get_capture(1, timeout=2)
2422
2423         # prepare and send a handshake response
2424         # expect a keepalive message
2425         resp = peer_1.consume_init(rxs[0], self.pg1)
2426         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2427
2428         # verify the keepalive message
2429         b = peer_1.decrypt_transport(rxs[0])
2430         self.assertEqual(0, len(b))
2431
2432         # prepare and send a packet that will be rewritten into the wg interface
2433         # expect a data packet sent
2434         p = (
2435             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2436             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2437             / UDP(sport=555, dport=556)
2438             / Raw()
2439         )
2440         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2441
2442         # verify the data packet
2443         peer_1.validate_encapped(rxs, p)
2444
2445         # administratively disable the wg interface
2446         wg0.admin_down()
2447
2448         # send a packet that will be rewritten into the wg interface
2449         # expect no data packets sent
2450         self.send_and_assert_no_replies(self.pg0, [p])
2451
2452         # administratively enable the wg interface
2453         # expect the peer to send a handshake initiation
2454         wg0.admin_up()
2455         peer_1.noise_reset()
2456         rxs = self.pg1.get_capture(1, timeout=2)
2457         resp = peer_1.consume_init(rxs[0], self.pg1)
2458
2459         # send a packet that will be rewritten into the wg interface
2460         # expect no data packets sent because the peer is not initiated
2461         self.send_and_assert_no_replies(self.pg0, [p])
2462         self.assertEqual(
2463             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
2464         )
2465
2466         # send a handshake response and expect a keepalive message
2467         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2468
2469         # verify the keepalive message
2470         b = peer_1.decrypt_transport(rxs[0])
2471         self.assertEqual(0, len(b))
2472
2473         # send a packet that will be rewritten into the wg interface
2474         # expect a data packet sent
2475         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2476
2477         # verify the data packet
2478         peer_1.validate_encapped(rxs, p)
2479
2480         # remove configs
2481         r1.remove_vpp_config()
2482         peer_1.remove_vpp_config()
2483         wg0.remove_vpp_config()
2484
2485
2486 @tag_fixme_vpp_debug
2487 class WireguardHandoffTests(TestWg):
2488     """Wireguard Tests in multi worker setup"""
2489
2490     vpp_worker_count = 2
2491
2492     def test_wg_peer_init(self):
2493         """Handoff"""
2494
2495         port = 12383
2496
2497         # Create interfaces
2498         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2499         wg0.admin_up()
2500         wg0.config_ip4()
2501
2502         self.pg_enable_capture(self.pg_interfaces)
2503         self.pg_start()
2504
2505         peer_1 = VppWgPeer(
2506             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.2.0/24", "10.11.3.0/24"]
2507         ).add_vpp_config()
2508         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2509
2510         r1 = VppIpRoute(
2511             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2512         ).add_vpp_config()
2513
2514         # skip the first automatic handshake
2515         self.pg1.get_capture(1, timeout=HANDSHAKE_JITTER)
2516
2517         # send a valid handsake init for which we expect a response
2518         p = peer_1.mk_handshake(self.pg1)
2519
2520         rx = self.send_and_expect(self.pg1, [p], self.pg1)
2521
2522         peer_1.consume_response(rx[0])
2523
2524         # send a data packet from the peer through the tunnel
2525         # this completes the handshake and pins the peer to worker 0
2526         p = (
2527             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2528             / UDP(sport=222, dport=223)
2529             / Raw()
2530         )
2531         d = peer_1.encrypt_transport(p)
2532         p = peer_1.mk_tunnel_header(self.pg1) / (
2533             Wireguard(message_type=4, reserved_zero=0)
2534             / WireguardTransport(
2535                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
2536             )
2537         )
2538         rxs = self.send_and_expect(self.pg1, [p], self.pg0, worker=0)
2539
2540         for rx in rxs:
2541             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2542             self.assertEqual(rx[IP].ttl, 19)
2543
2544         # send a packets that are routed into the tunnel
2545         # and pins the peer tp worker 1
2546         pe = (
2547             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2548             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2549             / UDP(sport=555, dport=556)
2550             / Raw(b"\x00" * 80)
2551         )
2552         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=1)
2553         peer_1.validate_encapped(rxs, pe)
2554
2555         # send packets into the tunnel, from the other worker
2556         p = [
2557             (
2558                 peer_1.mk_tunnel_header(self.pg1)
2559                 / Wireguard(message_type=4, reserved_zero=0)
2560                 / WireguardTransport(
2561                     receiver_index=peer_1.sender,
2562                     counter=ii + 1,
2563                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
2564                         (
2565                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2566                             / UDP(sport=222, dport=223)
2567                             / Raw()
2568                         )
2569                     ),
2570                 )
2571             )
2572             for ii in range(255)
2573         ]
2574
2575         rxs = self.send_and_expect(self.pg1, p, self.pg0, worker=1)
2576
2577         for rx in rxs:
2578             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2579             self.assertEqual(rx[IP].ttl, 19)
2580
2581         # send a packets that are routed into the tunnel
2582         # from worker 0
2583         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=0)
2584
2585         peer_1.validate_encapped(rxs, pe)
2586
2587         r1.remove_vpp_config()
2588         peer_1.remove_vpp_config()
2589         wg0.remove_vpp_config()
2590
2591     @unittest.skip("test disabled")
2592     def test_wg_multi_interface(self):
2593         """Multi-tunnel on the same port"""
2594
2595
2596 class TestWgFIB(VppTestCase):
2597     """Wireguard FIB Test Case"""
2598
2599     @classmethod
2600     def setUpClass(cls):
2601         super(TestWgFIB, cls).setUpClass()
2602
2603     @classmethod
2604     def tearDownClass(cls):
2605         super(TestWgFIB, cls).tearDownClass()
2606
2607     def setUp(self):
2608         super(TestWgFIB, self).setUp()
2609
2610         self.create_pg_interfaces(range(2))
2611
2612         for i in self.pg_interfaces:
2613             i.admin_up()
2614             i.config_ip4()
2615
2616     def tearDown(self):
2617         for i in self.pg_interfaces:
2618             i.unconfig_ip4()
2619             i.admin_down()
2620         super(TestWgFIB, self).tearDown()
2621
2622     def test_wg_fib_tracking(self):
2623         """FIB tracking"""
2624         port = 12323
2625
2626         # create wg interface
2627         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2628         wg0.admin_up()
2629         wg0.config_ip4()
2630
2631         self.pg_enable_capture(self.pg_interfaces)
2632         self.pg_start()
2633
2634         # create a peer
2635         peer_1 = VppWgPeer(
2636             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2637         ).add_vpp_config()
2638         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2639
2640         # create a route to rewrite traffic into the wg interface
2641         r1 = VppIpRoute(
2642             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2643         ).add_vpp_config()
2644
2645         # resolve ARP and expect the adjacency to update
2646         self.pg1.resolve_arp()
2647
2648         # wait for the peer to send a handshake initiation
2649         rxs = self.pg1.get_capture(2, timeout=6)
2650
2651         # prepare and send a handshake response
2652         # expect a keepalive message
2653         resp = peer_1.consume_init(rxs[1], self.pg1)
2654         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2655
2656         # verify the keepalive message
2657         b = peer_1.decrypt_transport(rxs[0])
2658         self.assertEqual(0, len(b))
2659
2660         # prepare and send a packet that will be rewritten into the wg interface
2661         # expect a data packet sent
2662         p = (
2663             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2664             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2665             / UDP(sport=555, dport=556)
2666             / Raw()
2667         )
2668         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2669
2670         # verify the data packet
2671         peer_1.validate_encapped(rxs, p)
2672
2673         # remove configs
2674         r1.remove_vpp_config()
2675         peer_1.remove_vpp_config()
2676         wg0.remove_vpp_config()