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_ip_route import VppIpRoute, VppRoutePath
  42 from vpp_object import VppObject
  43 from vpp_papi import VppEnum
  44 from framework import tag_fixme_ubuntu2204, tag_fixme_debian11
  45 from framework import is_distro_ubuntu2204, is_distro_debian11
  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
 151 class VppWgPeer(VppObject):
 152     def __init__(self, test, itf, endpoint, port, allowed_ips, persistent_keepalive=15):
 153         self._test = test
 154         self.itf = itf
 155         self.endpoint = endpoint
 156         self.port = port
 157         self.allowed_ips = allowed_ips
 158         self.persistent_keepalive = persistent_keepalive
 159
 160         # remote peer's public
 161         self.private_key = X25519PrivateKey.generate()
 162         self.public_key = self.private_key.public_key()
 163
 164         # cookie related params
 165         self.cookie_key = blake2s(b"cookie--" + self.public_key_bytes()).digest()
 166         self.last_sent_cookie = None
 167         self.last_mac1 = None
 168         self.last_received_cookie = None
 169
 170         self.noise = NoiseConnection.from_name(NOISE_HANDSHAKE_NAME)
 171
 172     def change_endpoint(self, endpoint, port):
 173         self.endpoint = endpoint
 174         self.port = port
 175
 176     def add_vpp_config(self, is_ip6=False):
 177         rv = self._test.vapi.wireguard_peer_add(
 178             peer={
 179                 "public_key": self.public_key_bytes(),
 180                 "port": self.port,
 181                 "endpoint": self.endpoint,
 182                 "n_allowed_ips": len(self.allowed_ips),
 183                 "allowed_ips": self.allowed_ips,
 184                 "sw_if_index": self.itf.sw_if_index,
 185                 "persistent_keepalive": self.persistent_keepalive,
 186             }
 187         )
 188         self.index = rv.peer_index
 189         self.receiver_index = self.index + 1
 190         self._test.registry.register(self, self._test.logger)
 191         return self
 192
 193     def remove_vpp_config(self):
 194         self._test.vapi.wireguard_peer_remove(peer_index=self.index)
 195
 196     def object_id(self):
 197         return "wireguard-peer-%s" % self.index
 198
 199     def public_key_bytes(self):
 200         return public_key_bytes(self.public_key)
 201
 202     def query_vpp_config(self):
 203         peers = self._test.vapi.wireguard_peers_dump()
 204
 205         for p in peers:
 206             # "::" endpoint will be returned as "0.0.0.0" in peer's details
 207             endpoint = "0.0.0.0" if self.endpoint == "::" else self.endpoint
 208             if (
 209                 p.peer.public_key == self.public_key_bytes()
 210                 and p.peer.port == self.port
 211                 and str(p.peer.endpoint) == endpoint
 212                 and p.peer.sw_if_index == self.itf.sw_if_index
 213                 and len(self.allowed_ips) == p.peer.n_allowed_ips
 214             ):
 215                 self.allowed_ips.sort()
 216                 p.peer.allowed_ips.sort()
 217
 218                 for (a1, a2) in zip(self.allowed_ips, p.peer.allowed_ips):
 219                     if str(a1) != str(a2):
 220                         return False
 221                 return True
 222         return False
 223
 224     def mk_tunnel_header(self, tx_itf, is_ip6=False):
 225         if is_ip6 is False:
 226             return (
 227                 Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
 228                 / IP(src=self.endpoint, dst=self.itf.src)
 229                 / UDP(sport=self.port, dport=self.itf.port)
 230             )
 231         else:
 232             return (
 233                 Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
 234                 / IPv6(src=self.endpoint, dst=self.itf.src)
 235                 / UDP(sport=self.port, dport=self.itf.port)
 236             )
 237
 238     def noise_reset(self):
 239         self.noise = NoiseConnection.from_name(NOISE_HANDSHAKE_NAME)
 240
 241     def noise_init(self, public_key=None):
 242         self.noise.set_prologue(NOISE_IDENTIFIER_NAME)
 243         self.noise.set_psks(psk=bytes(bytearray(32)))
 244
 245         if not public_key:
 246             public_key = self.itf.public_key
 247
 248         # local/this private
 249         self.noise.set_keypair_from_private_bytes(
 250             Keypair.STATIC, private_key_bytes(self.private_key)
 251         )
 252         # remote's public
 253         self.noise.set_keypair_from_public_bytes(
 254             Keypair.REMOTE_STATIC, public_key_bytes(public_key)
 255         )
 256
 257         self.noise.start_handshake()
 258
 259     def mk_cookie(self, p, tx_itf, is_resp=False, is_ip6=False):
 260         self.verify_header(p, is_ip6)
 261
 262         wg_pkt = Wireguard(p[Raw])
 263
 264         if is_resp:
 265             self._test.assertEqual(wg_pkt[Wireguard].message_type, 2)
 266             self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
 267             self._test.assertEqual(wg_pkt[WireguardResponse].mac2, bytes([0] * 16))
 268         else:
 269             self._test.assertEqual(wg_pkt[Wireguard].message_type, 1)
 270             self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
 271             self._test.assertEqual(wg_pkt[WireguardInitiation].mac2, bytes([0] * 16))
 272
 273         # collect info from wg packet (initiation or response)
 274         src = get_field_bytes(p[IPv6 if is_ip6 else IP], "src")
 275         sport = p[UDP].sport.to_bytes(2, byteorder="big")
 276         if is_resp:
 277             mac1 = wg_pkt[WireguardResponse].mac1
 278             sender_index = wg_pkt[WireguardResponse].sender_index
 279         else:
 280             mac1 = wg_pkt[WireguardInitiation].mac1
 281             sender_index = wg_pkt[WireguardInitiation].sender_index
 282
 283         # make cookie reply
 284         cookie_reply = Wireguard() / WireguardCookieReply()
 285         cookie_reply[Wireguard].message_type = 3
 286         cookie_reply[Wireguard].reserved_zero = 0
 287         cookie_reply[WireguardCookieReply].receiver_index = sender_index
 288         nonce = get_random_bytes(24)
 289         cookie_reply[WireguardCookieReply].nonce = nonce
 290
 291         # generate cookie data
 292         changing_secret = get_random_bytes(32)
 293         self.last_sent_cookie = blake2s(
 294             src + sport, digest_size=16, key=changing_secret
 295         ).digest()
 296
 297         # encrypt cookie data
 298         cipher = ChaCha20_Poly1305.new(key=self.cookie_key, nonce=nonce)
 299         cipher.update(mac1)
 300         ciphertext, tag = cipher.encrypt_and_digest(self.last_sent_cookie)
 301         cookie_reply[WireguardCookieReply].encrypted_cookie = ciphertext + tag
 302
 303         # prepare cookie reply to be sent
 304         cookie_reply = self.mk_tunnel_header(tx_itf, is_ip6) / cookie_reply
 305
 306         return cookie_reply
 307
 308     def consume_cookie(self, p, is_ip6=False):
 309         self.verify_header(p, is_ip6)
 310
 311         cookie_reply = Wireguard(p[Raw])
 312
 313         self._test.assertEqual(cookie_reply[Wireguard].message_type, 3)
 314         self._test.assertEqual(cookie_reply[Wireguard].reserved_zero, 0)
 315         self._test.assertEqual(
 316             cookie_reply[WireguardCookieReply].receiver_index, self.receiver_index
 317         )
 318
 319         # collect info from cookie reply
 320         nonce = cookie_reply[WireguardCookieReply].nonce
 321         encrypted_cookie = cookie_reply[WireguardCookieReply].encrypted_cookie
 322         ciphertext, tag = encrypted_cookie[:16], encrypted_cookie[16:]
 323
 324         # decrypt cookie data
 325         cipher = ChaCha20_Poly1305.new(key=self.itf.cookie_key, nonce=nonce)
 326         cipher.update(self.last_mac1)
 327         self.last_received_cookie = cipher.decrypt_and_verify(ciphertext, tag)
 328
 329     def mk_handshake(self, tx_itf, is_ip6=False, public_key=None):
 330         self.noise.set_as_initiator()
 331         self.noise_init(public_key)
 332
 333         p = Wireguard() / WireguardInitiation()
 334
 335         p[Wireguard].message_type = 1
 336         p[Wireguard].reserved_zero = 0
 337         p[WireguardInitiation].sender_index = self.receiver_index
 338
 339         # some random data for the message
 340         #  lifted from the noise protocol's wireguard example
 341         now = datetime.datetime.now()
 342         tai = struct.pack(
 343             "!qi",
 344             4611686018427387914 + int(now.timestamp()),
 345             int(now.microsecond * 1e3),
 346         )
 347         b = self.noise.write_message(payload=tai)
 348
 349         # load noise into init message
 350         p[WireguardInitiation].unencrypted_ephemeral = b[0:32]
 351         p[WireguardInitiation].encrypted_static = b[32:80]
 352         p[WireguardInitiation].encrypted_timestamp = b[80:108]
 353
 354         # generate the mac1 hash
 355         mac_key = blake2s(b"mac1----" + self.itf.public_key_bytes()).digest()
 356         mac1 = blake2s(bytes(p)[0:116], digest_size=16, key=mac_key).digest()
 357         p[WireguardInitiation].mac1 = mac1
 358         self.last_mac1 = mac1
 359
 360         # generate the mac2 hash
 361         if self.last_received_cookie:
 362             mac2 = blake2s(
 363                 bytes(p)[0:132], digest_size=16, key=self.last_received_cookie
 364             ).digest()
 365             p[WireguardInitiation].mac2 = mac2
 366             self.last_received_cookie = None
 367         else:
 368             p[WireguardInitiation].mac2 = bytearray(16)
 369
 370         p = self.mk_tunnel_header(tx_itf, is_ip6) / p
 371
 372         return p
 373
 374     def verify_header(self, p, is_ip6=False):
 375         if is_ip6 is False:
 376             self._test.assertEqual(p[IP].src, self.itf.src)
 377             self._test.assertEqual(p[IP].dst, self.endpoint)
 378         else:
 379             self._test.assertEqual(p[IPv6].src, self.itf.src)
 380             self._test.assertEqual(p[IPv6].dst, self.endpoint)
 381         self._test.assertEqual(p[UDP].sport, self.itf.port)
 382         self._test.assertEqual(p[UDP].dport, self.port)
 383         self._test.assert_packet_checksums_valid(p)
 384
 385     def consume_init(self, p, tx_itf, is_ip6=False, is_mac2=False):
 386         self.noise.set_as_responder()
 387         self.noise_init(self.itf.public_key)
 388         self.verify_header(p, is_ip6)
 389
 390         init = Wireguard(p[Raw])
 391
 392         self._test.assertEqual(init[Wireguard].message_type, 1)
 393         self._test.assertEqual(init[Wireguard].reserved_zero, 0)
 394
 395         self.sender = init[WireguardInitiation].sender_index
 396
 397         # validate the mac1 hash
 398         mac_key = blake2s(b"mac1----" + public_key_bytes(self.public_key)).digest()
 399         mac1 = blake2s(bytes(init)[0:-32], digest_size=16, key=mac_key).digest()
 400         self._test.assertEqual(init[WireguardInitiation].mac1, mac1)
 401
 402         # validate the mac2 hash
 403         if is_mac2:
 404             self._test.assertNotEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
 405             self._test.assertNotEqual(self.last_sent_cookie, None)
 406             mac2 = blake2s(
 407                 bytes(init)[0:-16], digest_size=16, key=self.last_sent_cookie
 408             ).digest()
 409             self._test.assertEqual(init[WireguardInitiation].mac2, mac2)
 410             self.last_sent_cookie = None
 411         else:
 412             self._test.assertEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
 413
 414         # this passes only unencrypted_ephemeral, encrypted_static,
 415         # encrypted_timestamp fields of the init
 416         payload = self.noise.read_message(bytes(init)[8:-32])
 417
 418         # build the response
 419         b = self.noise.write_message()
 420         mac_key = blake2s(b"mac1----" + public_key_bytes(self.itf.public_key)).digest()
 421         resp = Wireguard(message_type=2, reserved_zero=0) / WireguardResponse(
 422             sender_index=self.receiver_index,
 423             receiver_index=self.sender,
 424             unencrypted_ephemeral=b[0:32],
 425             encrypted_nothing=b[32:],
 426         )
 427         mac1 = blake2s(bytes(resp)[:-32], digest_size=16, key=mac_key).digest()
 428         resp[WireguardResponse].mac1 = mac1
 429         self.last_mac1 = mac1
 430
 431         resp = self.mk_tunnel_header(tx_itf, is_ip6) / resp
 432         self._test.assertTrue(self.noise.handshake_finished)
 433
 434         return resp
 435
 436     def consume_response(self, p, is_ip6=False):
 437         self.verify_header(p, is_ip6)
 438
 439         resp = Wireguard(p[Raw])
 440
 441         self._test.assertEqual(resp[Wireguard].message_type, 2)
 442         self._test.assertEqual(resp[Wireguard].reserved_zero, 0)
 443         self._test.assertEqual(
 444             resp[WireguardResponse].receiver_index, self.receiver_index
 445         )
 446
 447         self.sender = resp[Wireguard].sender_index
 448
 449         payload = self.noise.read_message(bytes(resp)[12:60])
 450         self._test.assertEqual(payload, b"")
 451         self._test.assertTrue(self.noise.handshake_finished)
 452
 453     def decrypt_transport(self, p, is_ip6=False):
 454         self.verify_header(p, is_ip6)
 455
 456         p = Wireguard(p[Raw])
 457         self._test.assertEqual(p[Wireguard].message_type, 4)
 458         self._test.assertEqual(p[Wireguard].reserved_zero, 0)
 459         self._test.assertEqual(
 460             p[WireguardTransport].receiver_index, self.receiver_index
 461         )
 462
 463         d = self.noise.decrypt(p[WireguardTransport].encrypted_encapsulated_packet)
 464         return d
 465
 466     def encrypt_transport(self, p):
 467         return self.noise.encrypt(bytes(p))
 468
 469     def validate_encapped(self, rxs, tx, is_ip6=False):
 470         for rx in rxs:
 471             if is_ip6 is False:
 472                 rx = IP(self.decrypt_transport(rx, is_ip6=is_ip6))
 473
 474                 # check the original packet is present
 475                 self._test.assertEqual(rx[IP].dst, tx[IP].dst)
 476                 self._test.assertEqual(rx[IP].ttl, tx[IP].ttl - 1)
 477             else:
 478                 rx = IPv6(self.decrypt_transport(rx, is_ip6=is_ip6))
 479
 480                 # check the original packet is present
 481                 self._test.assertEqual(rx[IPv6].dst, tx[IPv6].dst)
 482                 self._test.assertEqual(rx[IPv6].hlim, tx[IPv6].hlim - 1)
 483
 484     def want_events(self):
 485         self._test.vapi.want_wireguard_peer_events(
 486             enable_disable=1,
 487             pid=os.getpid(),
 488             peer_index=self.index,
 489             sw_if_index=self.itf.sw_if_index,
 490         )
 491
 492     def wait_event(self, expect, timeout=5):
 493         rv = self._test.vapi.wait_for_event(timeout, "wireguard_peer_event")
 494         self._test.assertEqual(rv.flags, expect)
 495         self._test.assertEqual(rv.peer_index, self.index)
 496
 497
 498 @tag_fixme_ubuntu2204
 499 @tag_fixme_debian11
 500 class TestWg(VppTestCase):
 501     """Wireguard Test Case"""
 502
 503     error_str = compile(r"Error")
 504
 505     wg4_output_node_name = "/err/wg4-output-tun/"
 506     wg4_input_node_name = "/err/wg4-input/"
 507     wg6_output_node_name = "/err/wg6-output-tun/"
 508     wg6_input_node_name = "/err/wg6-input/"
 509     kp4_error = wg4_output_node_name + "Keypair error"
 510     mac4_error = wg4_input_node_name + "Invalid MAC handshake"
 511     peer4_in_err = wg4_input_node_name + "Peer error"
 512     peer4_out_err = wg4_output_node_name + "Peer error"
 513     kp6_error = wg6_output_node_name + "Keypair error"
 514     mac6_error = wg6_input_node_name + "Invalid MAC handshake"
 515     peer6_in_err = wg6_input_node_name + "Peer error"
 516     peer6_out_err = wg6_output_node_name + "Peer error"
 517     cookie_dec4_err = wg4_input_node_name + "Failed during Cookie decryption"
 518     cookie_dec6_err = wg6_input_node_name + "Failed during Cookie decryption"
 519     ratelimited4_err = wg4_input_node_name + "Handshake ratelimited"
 520     ratelimited6_err = wg6_input_node_name + "Handshake ratelimited"
 521
 522     @classmethod
 523     def setUpClass(cls):
 524         super(TestWg, cls).setUpClass()
 525         if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
 526             cls, "vpp"
 527         ):
 528             return
 529         try:
 530             cls.create_pg_interfaces(range(3))
 531             for i in cls.pg_interfaces:
 532                 i.admin_up()
 533                 i.config_ip4()
 534                 i.config_ip6()
 535                 i.resolve_arp()
 536                 i.resolve_ndp()
 537
 538         except Exception:
 539             super(TestWg, cls).tearDownClass()
 540             raise
 541
 542     @classmethod
 543     def tearDownClass(cls):
 544         super(TestWg, cls).tearDownClass()
 545
 546     def setUp(self):
 547         super(VppTestCase, self).setUp()
 548         self.base_kp4_err = self.statistics.get_err_counter(self.kp4_error)
 549         self.base_mac4_err = self.statistics.get_err_counter(self.mac4_error)
 550         self.base_peer4_in_err = self.statistics.get_err_counter(self.peer4_in_err)
 551         self.base_peer4_out_err = self.statistics.get_err_counter(self.peer4_out_err)
 552         self.base_kp6_err = self.statistics.get_err_counter(self.kp6_error)
 553         self.base_mac6_err = self.statistics.get_err_counter(self.mac6_error)
 554         self.base_peer6_in_err = self.statistics.get_err_counter(self.peer6_in_err)
 555         self.base_peer6_out_err = self.statistics.get_err_counter(self.peer6_out_err)
 556         self.base_cookie_dec4_err = self.statistics.get_err_counter(
 557             self.cookie_dec4_err
 558         )
 559         self.base_cookie_dec6_err = self.statistics.get_err_counter(
 560             self.cookie_dec6_err
 561         )
 562         self.base_ratelimited4_err = self.statistics.get_err_counter(
 563             self.ratelimited4_err
 564         )
 565         self.base_ratelimited6_err = self.statistics.get_err_counter(
 566             self.ratelimited6_err
 567         )
 568
 569     def test_wg_interface(self):
 570         """Simple interface creation"""
 571         port = 12312
 572
 573         # Create interface
 574         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 575
 576         self.logger.info(self.vapi.cli("sh int"))
 577
 578         # delete interface
 579         wg0.remove_vpp_config()
 580
 581     def test_handshake_hash(self):
 582         """test hashing an init message"""
 583         # a init packet generated by linux given the key below
 584         h = (
 585             "0100000098b9032b"
 586             "55cc4b39e73c3d24"
 587             "a2a1ab884b524a81"
 588             "1808bb86640fb70d"
 589             "e93154fec1879125"
 590             "ab012624a27f0b75"
 591             "c0a2582f438ddb5f"
 592             "8e768af40b4ab444"
 593             "02f9ff473e1b797e"
 594             "80d39d93c5480c82"
 595             "a3d4510f70396976"
 596             "586fb67300a5167b"
 597             "ae6ca3ff3dfd00eb"
 598             "59be198810f5aa03"
 599             "6abc243d2155ee4f"
 600             "2336483900aef801"
 601             "08752cd700000000"
 602             "0000000000000000"
 603             "00000000"
 604         )
 605
 606         b = bytearray.fromhex(h)
 607         tgt = Wireguard(b)
 608
 609         pubb = base64.b64decode("aRuHFTTxICIQNefp05oKWlJv3zgKxb8+WW7JJMh0jyM=")
 610         pub = X25519PublicKey.from_public_bytes(pubb)
 611
 612         self.assertEqual(pubb, public_key_bytes(pub))
 613
 614         # strip the macs and build a new packet
 615         init = b[0:-32]
 616         mac_key = blake2s(b"mac1----" + public_key_bytes(pub)).digest()
 617         init += blake2s(init, digest_size=16, key=mac_key).digest()
 618         init += b"\x00" * 16
 619
 620         act = Wireguard(init)
 621
 622         self.assertEqual(tgt, act)
 623
 624     def _test_wg_send_cookie_tmpl(self, is_resp, is_ip6):
 625         port = 12323
 626
 627         # create wg interface
 628         if is_ip6:
 629             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 630             wg0.admin_up()
 631             wg0.config_ip6()
 632         else:
 633             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 634             wg0.admin_up()
 635             wg0.config_ip4()
 636
 637         self.pg_enable_capture(self.pg_interfaces)
 638         self.pg_start()
 639
 640         # create a peer
 641         if is_ip6:
 642             peer_1 = VppWgPeer(
 643                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 644             ).add_vpp_config()
 645         else:
 646             peer_1 = VppWgPeer(
 647                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 648             ).add_vpp_config()
 649         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 650
 651         if is_resp:
 652             # prepare and send a handshake initiation
 653             # expect the peer to send a handshake response
 654             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 655             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 656         else:
 657             # wait for the peer to send a handshake initiation
 658             rxs = self.pg1.get_capture(1, timeout=2)
 659
 660         # prepare and send a wrong cookie reply
 661         # expect no replies and the cookie error incremented
 662         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 663         cookie.nonce = b"1234567890"
 664         self.send_and_assert_no_replies(self.pg1, [cookie], timeout=0.1)
 665         if is_ip6:
 666             self.assertEqual(
 667                 self.base_cookie_dec6_err + 1,
 668                 self.statistics.get_err_counter(self.cookie_dec6_err),
 669             )
 670         else:
 671             self.assertEqual(
 672                 self.base_cookie_dec4_err + 1,
 673                 self.statistics.get_err_counter(self.cookie_dec4_err),
 674             )
 675
 676         # prepare and send a correct cookie reply
 677         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 678         self.pg_send(self.pg1, [cookie])
 679
 680         # wait for the peer to send a handshake initiation with mac2 set
 681         rxs = self.pg1.get_capture(1, timeout=6)
 682
 683         # verify the initiation and its mac2
 684         peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6, is_mac2=True)
 685
 686         # remove configs
 687         peer_1.remove_vpp_config()
 688         wg0.remove_vpp_config()
 689
 690     def test_wg_send_cookie_on_init_v4(self):
 691         """Send cookie on handshake initiation (v4)"""
 692         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=False)
 693
 694     def test_wg_send_cookie_on_init_v6(self):
 695         """Send cookie on handshake initiation (v6)"""
 696         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=True)
 697
 698     def test_wg_send_cookie_on_resp_v4(self):
 699         """Send cookie on handshake response (v4)"""
 700         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=False)
 701
 702     def test_wg_send_cookie_on_resp_v6(self):
 703         """Send cookie on handshake response (v6)"""
 704         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=True)
 705
 706     def _test_wg_receive_cookie_tmpl(self, is_resp, is_ip6):
 707         port = 12323
 708
 709         # create wg interface
 710         if is_ip6:
 711             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 712             wg0.admin_up()
 713             wg0.config_ip6()
 714         else:
 715             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 716             wg0.admin_up()
 717             wg0.config_ip4()
 718
 719         self.pg_enable_capture(self.pg_interfaces)
 720         self.pg_start()
 721
 722         # create a peer
 723         if is_ip6:
 724             peer_1 = VppWgPeer(
 725                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 726             ).add_vpp_config()
 727         else:
 728             peer_1 = VppWgPeer(
 729                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 730             ).add_vpp_config()
 731         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 732
 733         if is_resp:
 734             # wait for the peer to send a handshake initiation
 735             rxs = self.pg1.get_capture(1, timeout=2)
 736             # prepare and send a bunch of handshake responses
 737             # expect to switch to under load state
 738             resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
 739             txs = [resp] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 740             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 741             # reset noise to be able to turn into initiator later
 742             peer_1.noise_reset()
 743         else:
 744             # prepare and send a bunch of handshake initiations
 745             # expect to switch to under load state
 746             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 747             txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 748             rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 749
 750         # expect the peer to send a cookie reply
 751         peer_1.consume_cookie(rxs[-1], is_ip6=is_ip6)
 752
 753         # prepare and send a handshake initiation with wrong mac2
 754         # expect a cookie reply
 755         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 756         init.mac2 = b"1234567890"
 757         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 758         peer_1.consume_cookie(rxs[0], is_ip6=is_ip6)
 759
 760         # prepare and send a handshake initiation with correct mac2
 761         # expect a handshake response
 762         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 763         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 764
 765         # verify the response
 766         peer_1.consume_response(rxs[0], is_ip6=is_ip6)
 767
 768         # clear up under load state
 769         self.sleep(UNDER_LOAD_INTERVAL)
 770
 771         # remove configs
 772         peer_1.remove_vpp_config()
 773         wg0.remove_vpp_config()
 774
 775     def test_wg_receive_cookie_on_init_v4(self):
 776         """Receive cookie on handshake initiation (v4)"""
 777         self._test_wg_receive_cookie_tmpl(is_resp=False, is_ip6=False)
 778
 779     def test_wg_receive_cookie_on_init_v6(self):
 780         """Receive cookie on handshake initiation (v6)"""
 781         self._test_wg_receive_cookie_tmpl(is_resp=False, is_ip6=True)
 782
 783     def test_wg_receive_cookie_on_resp_v4(self):
 784         """Receive cookie on handshake response (v4)"""
 785         self._test_wg_receive_cookie_tmpl(is_resp=True, is_ip6=False)
 786
 787     def test_wg_receive_cookie_on_resp_v6(self):
 788         """Receive cookie on handshake response (v6)"""
 789         self._test_wg_receive_cookie_tmpl(is_resp=True, is_ip6=True)
 790
 791     def test_wg_under_load_interval(self):
 792         """Under load interval"""
 793         port = 12323
 794
 795         # create wg interface
 796         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 797         wg0.admin_up()
 798         wg0.config_ip4()
 799
 800         self.pg_enable_capture(self.pg_interfaces)
 801         self.pg_start()
 802
 803         # create a peer
 804         peer_1 = VppWgPeer(
 805             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 806         ).add_vpp_config()
 807         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 808
 809         # prepare and send a bunch of handshake initiations
 810         # expect to switch to under load state
 811         init = peer_1.mk_handshake(self.pg1)
 812         txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 813         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 814
 815         # expect the peer to send a cookie reply
 816         peer_1.consume_cookie(rxs[-1])
 817
 818         # sleep till the next counting interval
 819         # expect under load state is still active
 820         self.sleep(HANDSHAKE_COUNTING_INTERVAL)
 821
 822         # prepare and send a handshake initiation with wrong mac2
 823         # expect a cookie reply
 824         init = peer_1.mk_handshake(self.pg1)
 825         init.mac2 = b"1234567890"
 826         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 827         peer_1.consume_cookie(rxs[0])
 828
 829         # sleep till the end of being under load
 830         # expect under load state is over
 831         self.sleep(UNDER_LOAD_INTERVAL - HANDSHAKE_COUNTING_INTERVAL)
 832
 833         # prepare and send a handshake initiation with wrong mac2
 834         # expect a handshake response
 835         init = peer_1.mk_handshake(self.pg1)
 836         init.mac2 = b"1234567890"
 837         rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 838
 839         # verify the response
 840         peer_1.consume_response(rxs[0])
 841
 842         # remove configs
 843         peer_1.remove_vpp_config()
 844         wg0.remove_vpp_config()
 845
 846     def _test_wg_handshake_ratelimiting_tmpl(self, is_ip6):
 847         port = 12323
 848
 849         # create wg interface
 850         if is_ip6:
 851             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 852             wg0.admin_up()
 853             wg0.config_ip6()
 854         else:
 855             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 856             wg0.admin_up()
 857             wg0.config_ip4()
 858
 859         self.pg_enable_capture(self.pg_interfaces)
 860         self.pg_start()
 861
 862         # create a peer
 863         if is_ip6:
 864             peer_1 = VppWgPeer(
 865                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 866             ).add_vpp_config()
 867         else:
 868             peer_1 = VppWgPeer(
 869                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 870             ).add_vpp_config()
 871         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 872
 873         # prepare and send a bunch of handshake initiations
 874         # expect to switch to under load state
 875         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 876         txs = [init] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 877         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 878
 879         # expect the peer to send a cookie reply
 880         peer_1.consume_cookie(rxs[-1], is_ip6=is_ip6)
 881
 882         # prepare and send a bunch of handshake initiations with correct mac2
 883         # expect a handshake response and then ratelimiting
 884         NUM_TO_REJECT = 10
 885         init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 886         txs = [init] * (HANDSHAKE_NUM_BEFORE_RATELIMITING + NUM_TO_REJECT)
 887         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 888
 889         if is_ip6:
 890             self.assertEqual(
 891                 self.base_ratelimited6_err + NUM_TO_REJECT,
 892                 self.statistics.get_err_counter(self.ratelimited6_err),
 893             )
 894         else:
 895             self.assertEqual(
 896                 self.base_ratelimited4_err + NUM_TO_REJECT,
 897                 self.statistics.get_err_counter(self.ratelimited4_err),
 898             )
 899
 900         # verify the response
 901         peer_1.consume_response(rxs[0], is_ip6=is_ip6)
 902
 903         # clear up under load state
 904         self.sleep(UNDER_LOAD_INTERVAL)
 905
 906         # remove configs
 907         peer_1.remove_vpp_config()
 908         wg0.remove_vpp_config()
 909
 910     def test_wg_handshake_ratelimiting_v4(self):
 911         """Handshake ratelimiting (v4)"""
 912         self._test_wg_handshake_ratelimiting_tmpl(is_ip6=False)
 913
 914     def test_wg_handshake_ratelimiting_v6(self):
 915         """Handshake ratelimiting (v6)"""
 916         self._test_wg_handshake_ratelimiting_tmpl(is_ip6=True)
 917
 918     def test_wg_handshake_ratelimiting_multi_peer(self):
 919         """Handshake ratelimiting (multiple peer)"""
 920         port = 12323
 921
 922         # create wg interface
 923         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 924         wg0.admin_up()
 925         wg0.config_ip4()
 926
 927         self.pg_enable_capture(self.pg_interfaces)
 928         self.pg_start()
 929
 930         # create two peers
 931         NUM_PEERS = 2
 932         self.pg1.generate_remote_hosts(NUM_PEERS)
 933         self.pg1.configure_ipv4_neighbors()
 934
 935         peer_1 = VppWgPeer(
 936             self, wg0, self.pg1.remote_hosts[0].ip4, port + 1, ["10.11.3.0/24"]
 937         ).add_vpp_config()
 938         peer_2 = VppWgPeer(
 939             self, wg0, self.pg1.remote_hosts[1].ip4, port + 1, ["10.11.4.0/24"]
 940         ).add_vpp_config()
 941         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 2)
 942
 943         # (peer_1) prepare and send a bunch of handshake initiations
 944         # expect not to switch to under load state
 945         init_1 = peer_1.mk_handshake(self.pg1)
 946         txs = [init_1] * HANDSHAKE_NUM_PER_PEER_UNTIL_UNDER_LOAD
 947         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 948
 949         # (peer_1) expect the peer to send a handshake response
 950         peer_1.consume_response(rxs[0])
 951         peer_1.noise_reset()
 952
 953         # (peer_1) send another bunch of handshake initiations
 954         # expect to switch to under load state
 955         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 956
 957         # (peer_1) expect the peer to send a cookie reply
 958         peer_1.consume_cookie(rxs[-1])
 959
 960         # (peer_2) prepare and send a handshake initiation
 961         # expect a cookie reply
 962         init_2 = peer_2.mk_handshake(self.pg1)
 963         rxs = self.send_and_expect(self.pg1, [init_2], self.pg1)
 964         peer_2.consume_cookie(rxs[0])
 965
 966         # (peer_1) prepare and send a bunch of handshake initiations with correct mac2
 967         # expect no ratelimiting and a handshake response
 968         init_1 = peer_1.mk_handshake(self.pg1)
 969         txs = [init_1] * HANDSHAKE_NUM_BEFORE_RATELIMITING
 970         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 971         self.assertEqual(
 972             self.base_ratelimited4_err,
 973             self.statistics.get_err_counter(self.ratelimited4_err),
 974         )
 975
 976         # (peer_1) verify the response
 977         peer_1.consume_response(rxs[0])
 978         peer_1.noise_reset()
 979
 980         # (peer_1) send another two handshake initiations with correct mac2
 981         # expect ratelimiting
 982         # (peer_2) prepare and send a handshake initiation with correct mac2
 983         # expect no ratelimiting and a handshake response
 984         init_2 = peer_2.mk_handshake(self.pg1)
 985         txs = [init_1, init_2, init_1]
 986         rxs = self.send_and_expect_some(self.pg1, txs, self.pg1)
 987
 988         # (peer_1) verify ratelimiting
 989         self.assertEqual(
 990             self.base_ratelimited4_err + 2,
 991             self.statistics.get_err_counter(self.ratelimited4_err),
 992         )
 993
 994         # (peer_2) verify the response
 995         peer_2.consume_response(rxs[0])
 996
 997         # clear up under load state
 998         self.sleep(UNDER_LOAD_INTERVAL)
 999
1000         # remove configs
1001         peer_1.remove_vpp_config()
1002         peer_2.remove_vpp_config()
1003         wg0.remove_vpp_config()
1004
1005     def _test_wg_peer_roaming_on_handshake_tmpl(self, is_endpoint_set, is_resp, is_ip6):
1006         port = 12323
1007
1008         # create wg interface
1009         if is_ip6:
1010             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1011             wg0.admin_up()
1012             wg0.config_ip6()
1013         else:
1014             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1015             wg0.admin_up()
1016             wg0.config_ip4()
1017
1018         self.pg_enable_capture(self.pg_interfaces)
1019         self.pg_start()
1020
1021         # create more remote hosts
1022         NUM_REMOTE_HOSTS = 2
1023         self.pg1.generate_remote_hosts(NUM_REMOTE_HOSTS)
1024         if is_ip6:
1025             self.pg1.configure_ipv6_neighbors()
1026         else:
1027             self.pg1.configure_ipv4_neighbors()
1028
1029         # create a peer
1030         if is_ip6:
1031             peer_1 = VppWgPeer(
1032                 test=self,
1033                 itf=wg0,
1034                 endpoint=self.pg1.remote_hosts[0].ip6 if is_endpoint_set else "::",
1035                 port=port + 1 if is_endpoint_set else 0,
1036                 allowed_ips=["1::3:0/112"],
1037             ).add_vpp_config()
1038         else:
1039             peer_1 = VppWgPeer(
1040                 test=self,
1041                 itf=wg0,
1042                 endpoint=self.pg1.remote_hosts[0].ip4 if is_endpoint_set else "0.0.0.0",
1043                 port=port + 1 if is_endpoint_set else 0,
1044                 allowed_ips=["10.11.3.0/24"],
1045             ).add_vpp_config()
1046         self.assertTrue(peer_1.query_vpp_config())
1047
1048         if is_resp:
1049             # wait for the peer to send a handshake initiation
1050             rxs = self.pg1.get_capture(1, timeout=2)
1051             # prepare a handshake response
1052             resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
1053             # change endpoint
1054             if is_ip6:
1055                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1056                 resp[IPv6].src, resp[UDP].sport = peer_1.endpoint, peer_1.port
1057             else:
1058                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1059                 resp[IP].src, resp[UDP].sport = peer_1.endpoint, peer_1.port
1060             # send the handshake response
1061             # expect a keepalive message sent to the new endpoint
1062             rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1063             # verify the keepalive message
1064             b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
1065             self.assertEqual(0, len(b))
1066         else:
1067             # change endpoint
1068             if is_ip6:
1069                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1070             else:
1071                 peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1072             # prepare and send a handshake initiation
1073             # expect a handshake response sent to the new endpoint
1074             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
1075             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
1076             # verify the response
1077             peer_1.consume_response(rxs[0], is_ip6=is_ip6)
1078         self.assertTrue(peer_1.query_vpp_config())
1079
1080         # remove configs
1081         peer_1.remove_vpp_config()
1082         wg0.remove_vpp_config()
1083
1084     def test_wg_peer_roaming_on_init_v4(self):
1085         """Peer roaming on handshake initiation (v4)"""
1086         self._test_wg_peer_roaming_on_handshake_tmpl(
1087             is_endpoint_set=False, is_resp=False, is_ip6=False
1088         )
1089
1090     def test_wg_peer_roaming_on_init_v6(self):
1091         """Peer roaming on handshake initiation (v6)"""
1092         self._test_wg_peer_roaming_on_handshake_tmpl(
1093             is_endpoint_set=False, is_resp=False, is_ip6=True
1094         )
1095
1096     def test_wg_peer_roaming_on_resp_v4(self):
1097         """Peer roaming on handshake response (v4)"""
1098         self._test_wg_peer_roaming_on_handshake_tmpl(
1099             is_endpoint_set=True, is_resp=True, is_ip6=False
1100         )
1101
1102     def test_wg_peer_roaming_on_resp_v6(self):
1103         """Peer roaming on handshake response (v6)"""
1104         self._test_wg_peer_roaming_on_handshake_tmpl(
1105             is_endpoint_set=True, is_resp=True, is_ip6=True
1106         )
1107
1108     def _test_wg_peer_roaming_on_data_tmpl(self, is_async, is_ip6):
1109         self.vapi.wg_set_async_mode(is_async)
1110         port = 12323
1111
1112         # create wg interface
1113         if is_ip6:
1114             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1115             wg0.admin_up()
1116             wg0.config_ip6()
1117         else:
1118             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1119             wg0.admin_up()
1120             wg0.config_ip4()
1121
1122         self.pg_enable_capture(self.pg_interfaces)
1123         self.pg_start()
1124
1125         # create more remote hosts
1126         NUM_REMOTE_HOSTS = 2
1127         self.pg1.generate_remote_hosts(NUM_REMOTE_HOSTS)
1128         if is_ip6:
1129             self.pg1.configure_ipv6_neighbors()
1130         else:
1131             self.pg1.configure_ipv4_neighbors()
1132
1133         # create a peer
1134         if is_ip6:
1135             peer_1 = VppWgPeer(
1136                 self, wg0, self.pg1.remote_hosts[0].ip6, port + 1, ["1::3:0/112"]
1137             ).add_vpp_config()
1138         else:
1139             peer_1 = VppWgPeer(
1140                 self, wg0, self.pg1.remote_hosts[0].ip4, port + 1, ["10.11.3.0/24"]
1141             ).add_vpp_config()
1142         self.assertTrue(peer_1.query_vpp_config())
1143
1144         # create a route to rewrite traffic into the wg interface
1145         if is_ip6:
1146             r1 = VppIpRoute(
1147                 self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1148             ).add_vpp_config()
1149         else:
1150             r1 = VppIpRoute(
1151                 self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1152             ).add_vpp_config()
1153
1154         # wait for the peer to send a handshake initiation
1155         rxs = self.pg1.get_capture(1, timeout=2)
1156
1157         # prepare and send a handshake response
1158         # expect a keepalive message
1159         resp = peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6)
1160         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1161
1162         # verify the keepalive message
1163         b = peer_1.decrypt_transport(rxs[0], is_ip6=is_ip6)
1164         self.assertEqual(0, len(b))
1165
1166         # change endpoint
1167         if is_ip6:
1168             peer_1.change_endpoint(self.pg1.remote_hosts[1].ip6, port + 100)
1169         else:
1170             peer_1.change_endpoint(self.pg1.remote_hosts[1].ip4, port + 100)
1171
1172         # prepare and send a data packet
1173         # expect endpoint change
1174         if is_ip6:
1175             ip_header = IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1176         else:
1177             ip_header = IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1178         data = (
1179             peer_1.mk_tunnel_header(self.pg1, is_ip6=is_ip6)
1180             / Wireguard(message_type=4, reserved_zero=0)
1181             / WireguardTransport(
1182                 receiver_index=peer_1.sender,
1183                 counter=0,
1184                 encrypted_encapsulated_packet=peer_1.encrypt_transport(
1185                     ip_header / UDP(sport=222, dport=223) / Raw()
1186                 ),
1187             )
1188         )
1189         rxs = self.send_and_expect(self.pg1, [data], self.pg0)
1190         if is_ip6:
1191             self.assertEqual(rxs[0][IPv6].dst, self.pg0.remote_ip6)
1192             self.assertEqual(rxs[0][IPv6].hlim, 19)
1193         else:
1194             self.assertEqual(rxs[0][IP].dst, self.pg0.remote_ip4)
1195             self.assertEqual(rxs[0][IP].ttl, 19)
1196         self.assertTrue(peer_1.query_vpp_config())
1197
1198         # prepare and send a packet that will be rewritten into the wg interface
1199         # expect a data packet sent to the new endpoint
1200         if is_ip6:
1201             ip_header = IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1202         else:
1203             ip_header = IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1204         p = (
1205             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1206             / ip_header
1207             / UDP(sport=555, dport=556)
1208             / Raw()
1209         )
1210         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
1211
1212         # verify the data packet
1213         peer_1.validate_encapped(rxs, p, is_ip6=is_ip6)
1214
1215         # remove configs
1216         r1.remove_vpp_config()
1217         peer_1.remove_vpp_config()
1218         wg0.remove_vpp_config()
1219
1220     def test_wg_peer_roaming_on_data_v4_sync(self):
1221         """Peer roaming on data packet (v4, sync)"""
1222         self._test_wg_peer_roaming_on_data_tmpl(is_async=False, is_ip6=False)
1223
1224     def test_wg_peer_roaming_on_data_v6_sync(self):
1225         """Peer roaming on data packet (v6, sync)"""
1226         self._test_wg_peer_roaming_on_data_tmpl(is_async=False, is_ip6=True)
1227
1228     def test_wg_peer_roaming_on_data_v4_async(self):
1229         """Peer roaming on data packet (v4, async)"""
1230         self._test_wg_peer_roaming_on_data_tmpl(is_async=True, is_ip6=False)
1231
1232     def test_wg_peer_roaming_on_data_v6_async(self):
1233         """Peer roaming on data packet (v6, async)"""
1234         self._test_wg_peer_roaming_on_data_tmpl(is_async=True, is_ip6=True)
1235
1236     def test_wg_peer_resp(self):
1237         """Send handshake response"""
1238         port = 12323
1239
1240         # Create interfaces
1241         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1242         wg0.admin_up()
1243         wg0.config_ip4()
1244
1245         self.pg_enable_capture(self.pg_interfaces)
1246         self.pg_start()
1247
1248         peer_1 = VppWgPeer(
1249             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
1250         ).add_vpp_config()
1251         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1252
1253         r1 = VppIpRoute(
1254             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1255         ).add_vpp_config()
1256
1257         # wait for the peer to send a handshake
1258         rx = self.pg1.get_capture(1, timeout=2)
1259
1260         # consume the handshake in the noise protocol and
1261         # generate the response
1262         resp = peer_1.consume_init(rx[0], self.pg1)
1263
1264         # send the response, get keepalive
1265         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
1266
1267         for rx in rxs:
1268             b = peer_1.decrypt_transport(rx)
1269             self.assertEqual(0, len(b))
1270
1271         # send a packets that are routed into the tunnel
1272         p = (
1273             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1274             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1275             / UDP(sport=555, dport=556)
1276             / Raw(b"\x00" * 80)
1277         )
1278
1279         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1280
1281         peer_1.validate_encapped(rxs, p)
1282
1283         # send packets into the tunnel, expect to receive them on
1284         # the other side
1285         p = [
1286             (
1287                 peer_1.mk_tunnel_header(self.pg1)
1288                 / Wireguard(message_type=4, reserved_zero=0)
1289                 / WireguardTransport(
1290                     receiver_index=peer_1.sender,
1291                     counter=ii,
1292                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1293                         (
1294                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1295                             / UDP(sport=222, dport=223)
1296                             / Raw()
1297                         )
1298                     ),
1299                 )
1300             )
1301             for ii in range(255)
1302         ]
1303
1304         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1305
1306         for rx in rxs:
1307             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1308             self.assertEqual(rx[IP].ttl, 19)
1309
1310         r1.remove_vpp_config()
1311         peer_1.remove_vpp_config()
1312         wg0.remove_vpp_config()
1313
1314     def test_wg_peer_v4o4(self):
1315         """Test v4o4"""
1316
1317         port = 12333
1318
1319         # Create interfaces
1320         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1321         wg0.admin_up()
1322         wg0.config_ip4()
1323
1324         peer_1 = VppWgPeer(
1325             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
1326         ).add_vpp_config()
1327         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1328
1329         r1 = VppIpRoute(
1330             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1331         ).add_vpp_config()
1332         r2 = VppIpRoute(
1333             self, "20.22.3.0", 24, [VppRoutePath("20.22.3.1", wg0.sw_if_index)]
1334         ).add_vpp_config()
1335
1336         # route a packet into the wg interface
1337         #  use the allowed-ip prefix
1338         #  this is dropped because the peer is not initiated
1339         p = (
1340             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1341             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1342             / UDP(sport=555, dport=556)
1343             / Raw()
1344         )
1345         self.send_and_assert_no_replies(self.pg0, [p])
1346         self.assertEqual(
1347             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1348         )
1349
1350         # route a packet into the wg interface
1351         #  use a not allowed-ip prefix
1352         #  this is dropped because there is no matching peer
1353         p = (
1354             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1355             / IP(src=self.pg0.remote_ip4, dst="20.22.3.2")
1356             / UDP(sport=555, dport=556)
1357             / Raw()
1358         )
1359         self.send_and_assert_no_replies(self.pg0, [p])
1360         self.assertEqual(
1361             self.base_peer4_out_err + 1,
1362             self.statistics.get_err_counter(self.peer4_out_err),
1363         )
1364
1365         # send a handsake from the peer with an invalid MAC
1366         p = peer_1.mk_handshake(self.pg1)
1367         p[WireguardInitiation].mac1 = b"foobar"
1368         self.send_and_assert_no_replies(self.pg1, [p])
1369         self.assertEqual(
1370             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1371         )
1372
1373         # send a handsake from the peer but signed by the wrong key.
1374         p = peer_1.mk_handshake(
1375             self.pg1, False, X25519PrivateKey.generate().public_key()
1376         )
1377         self.send_and_assert_no_replies(self.pg1, [p])
1378         self.assertEqual(
1379             self.base_peer4_in_err + 1,
1380             self.statistics.get_err_counter(self.peer4_in_err),
1381         )
1382
1383         # send a valid handsake init for which we expect a response
1384         p = peer_1.mk_handshake(self.pg1)
1385
1386         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1387
1388         peer_1.consume_response(rx[0])
1389
1390         # route a packet into the wg interface
1391         #  this is dropped because the peer is still not initiated
1392         p = (
1393             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1394             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1395             / UDP(sport=555, dport=556)
1396             / Raw()
1397         )
1398         self.send_and_assert_no_replies(self.pg0, [p])
1399         self.assertEqual(
1400             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1401         )
1402
1403         # send a data packet from the peer through the tunnel
1404         # this completes the handshake
1405         p = (
1406             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1407             / UDP(sport=222, dport=223)
1408             / Raw()
1409         )
1410         d = peer_1.encrypt_transport(p)
1411         p = peer_1.mk_tunnel_header(self.pg1) / (
1412             Wireguard(message_type=4, reserved_zero=0)
1413             / WireguardTransport(
1414                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1415             )
1416         )
1417         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1418
1419         for rx in rxs:
1420             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1421             self.assertEqual(rx[IP].ttl, 19)
1422
1423         # send a packets that are routed into the tunnel
1424         p = (
1425             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1426             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1427             / UDP(sport=555, dport=556)
1428             / Raw(b"\x00" * 80)
1429         )
1430
1431         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1432
1433         for rx in rxs:
1434             rx = IP(peer_1.decrypt_transport(rx))
1435
1436             # check the original packet is present
1437             self.assertEqual(rx[IP].dst, p[IP].dst)
1438             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1439
1440         # send packets into the tunnel, expect to receive them on
1441         # the other side
1442         p = [
1443             (
1444                 peer_1.mk_tunnel_header(self.pg1)
1445                 / Wireguard(message_type=4, reserved_zero=0)
1446                 / WireguardTransport(
1447                     receiver_index=peer_1.sender,
1448                     counter=ii + 1,
1449                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1450                         (
1451                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1452                             / UDP(sport=222, dport=223)
1453                             / Raw()
1454                         )
1455                     ),
1456                 )
1457             )
1458             for ii in range(255)
1459         ]
1460
1461         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1462
1463         for rx in rxs:
1464             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1465             self.assertEqual(rx[IP].ttl, 19)
1466
1467         r1.remove_vpp_config()
1468         r2.remove_vpp_config()
1469         peer_1.remove_vpp_config()
1470         wg0.remove_vpp_config()
1471
1472     def test_wg_peer_v6o6(self):
1473         """Test v6o6"""
1474
1475         port = 12343
1476
1477         # Create interfaces
1478         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1479         wg0.admin_up()
1480         wg0.config_ip6()
1481
1482         peer_1 = VppWgPeer(
1483             self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
1484         ).add_vpp_config(True)
1485         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1486
1487         r1 = VppIpRoute(
1488             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1489         ).add_vpp_config()
1490         r2 = VppIpRoute(
1491             self, "22::3:0", 112, [VppRoutePath("22::3:1", wg0.sw_if_index)]
1492         ).add_vpp_config()
1493
1494         # route a packet into the wg interface
1495         #  use the allowed-ip prefix
1496         #  this is dropped because the peer is not initiated
1497
1498         p = (
1499             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1500             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1501             / UDP(sport=555, dport=556)
1502             / Raw()
1503         )
1504         self.send_and_assert_no_replies(self.pg0, [p])
1505
1506         self.assertEqual(
1507             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1508         )
1509
1510         # route a packet into the wg interface
1511         #  use a not allowed-ip prefix
1512         #  this is dropped because there is no matching peer
1513         p = (
1514             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1515             / IPv6(src=self.pg0.remote_ip6, dst="22::3:2")
1516             / UDP(sport=555, dport=556)
1517             / Raw()
1518         )
1519         self.send_and_assert_no_replies(self.pg0, [p])
1520         self.assertEqual(
1521             self.base_peer6_out_err + 1,
1522             self.statistics.get_err_counter(self.peer6_out_err),
1523         )
1524
1525         # send a handsake from the peer with an invalid MAC
1526         p = peer_1.mk_handshake(self.pg1, True)
1527         p[WireguardInitiation].mac1 = b"foobar"
1528         self.send_and_assert_no_replies(self.pg1, [p])
1529
1530         self.assertEqual(
1531             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1532         )
1533
1534         # send a handsake from the peer but signed by the wrong key.
1535         p = peer_1.mk_handshake(
1536             self.pg1, True, X25519PrivateKey.generate().public_key()
1537         )
1538         self.send_and_assert_no_replies(self.pg1, [p])
1539         self.assertEqual(
1540             self.base_peer6_in_err + 1,
1541             self.statistics.get_err_counter(self.peer6_in_err),
1542         )
1543
1544         # send a valid handsake init for which we expect a response
1545         p = peer_1.mk_handshake(self.pg1, True)
1546
1547         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1548
1549         peer_1.consume_response(rx[0], True)
1550
1551         # route a packet into the wg interface
1552         #  this is dropped because the peer is still not initiated
1553         p = (
1554             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1555             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1556             / UDP(sport=555, dport=556)
1557             / Raw()
1558         )
1559         self.send_and_assert_no_replies(self.pg0, [p])
1560         self.assertEqual(
1561             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1562         )
1563
1564         # send a data packet from the peer through the tunnel
1565         # this completes the handshake
1566         p = (
1567             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1568             / UDP(sport=222, dport=223)
1569             / Raw()
1570         )
1571         d = peer_1.encrypt_transport(p)
1572         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1573             Wireguard(message_type=4, reserved_zero=0)
1574             / WireguardTransport(
1575                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1576             )
1577         )
1578         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1579
1580         for rx in rxs:
1581             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1582             self.assertEqual(rx[IPv6].hlim, 19)
1583
1584         # send a packets that are routed into the tunnel
1585         p = (
1586             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1587             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1588             / UDP(sport=555, dport=556)
1589             / Raw(b"\x00" * 80)
1590         )
1591
1592         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1593
1594         for rx in rxs:
1595             rx = IPv6(peer_1.decrypt_transport(rx, True))
1596
1597             # check the original packet is present
1598             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1599             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1600
1601         # send packets into the tunnel, expect to receive them on
1602         # the other side
1603         p = [
1604             (
1605                 peer_1.mk_tunnel_header(self.pg1, True)
1606                 / Wireguard(message_type=4, reserved_zero=0)
1607                 / WireguardTransport(
1608                     receiver_index=peer_1.sender,
1609                     counter=ii + 1,
1610                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1611                         (
1612                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1613                             / UDP(sport=222, dport=223)
1614                             / Raw()
1615                         )
1616                     ),
1617                 )
1618             )
1619             for ii in range(255)
1620         ]
1621
1622         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1623
1624         for rx in rxs:
1625             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1626             self.assertEqual(rx[IPv6].hlim, 19)
1627
1628         r1.remove_vpp_config()
1629         r2.remove_vpp_config()
1630         peer_1.remove_vpp_config()
1631         wg0.remove_vpp_config()
1632
1633     def test_wg_peer_v6o4(self):
1634         """Test v6o4"""
1635
1636         port = 12353
1637
1638         # Create interfaces
1639         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1640         wg0.admin_up()
1641         wg0.config_ip6()
1642
1643         peer_1 = VppWgPeer(
1644             self, wg0, self.pg1.remote_ip4, port + 1, ["1::3:0/112"]
1645         ).add_vpp_config(True)
1646         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1647
1648         r1 = VppIpRoute(
1649             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1650         ).add_vpp_config()
1651
1652         # route a packet into the wg interface
1653         #  use the allowed-ip prefix
1654         #  this is dropped because the peer is not initiated
1655         p = (
1656             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1657             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1658             / UDP(sport=555, dport=556)
1659             / Raw()
1660         )
1661         self.send_and_assert_no_replies(self.pg0, [p])
1662         self.assertEqual(
1663             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1664         )
1665
1666         # send a handsake from the peer with an invalid MAC
1667         p = peer_1.mk_handshake(self.pg1)
1668         p[WireguardInitiation].mac1 = b"foobar"
1669         self.send_and_assert_no_replies(self.pg1, [p])
1670
1671         self.assertEqual(
1672             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1673         )
1674
1675         # send a handsake from the peer but signed by the wrong key.
1676         p = peer_1.mk_handshake(
1677             self.pg1, False, X25519PrivateKey.generate().public_key()
1678         )
1679         self.send_and_assert_no_replies(self.pg1, [p])
1680         self.assertEqual(
1681             self.base_peer4_in_err + 1,
1682             self.statistics.get_err_counter(self.peer4_in_err),
1683         )
1684
1685         # send a valid handsake init for which we expect a response
1686         p = peer_1.mk_handshake(self.pg1)
1687
1688         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1689
1690         peer_1.consume_response(rx[0])
1691
1692         # route a packet into the wg interface
1693         #  this is dropped because the peer is still not initiated
1694         p = (
1695             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1696             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1697             / UDP(sport=555, dport=556)
1698             / Raw()
1699         )
1700         self.send_and_assert_no_replies(self.pg0, [p])
1701         self.assertEqual(
1702             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1703         )
1704
1705         # send a data packet from the peer through the tunnel
1706         # this completes the handshake
1707         p = (
1708             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1709             / UDP(sport=222, dport=223)
1710             / Raw()
1711         )
1712         d = peer_1.encrypt_transport(p)
1713         p = peer_1.mk_tunnel_header(self.pg1) / (
1714             Wireguard(message_type=4, reserved_zero=0)
1715             / WireguardTransport(
1716                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1717             )
1718         )
1719         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1720
1721         for rx in rxs:
1722             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1723             self.assertEqual(rx[IPv6].hlim, 19)
1724
1725         # send a packets that are routed into the tunnel
1726         p = (
1727             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1728             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1729             / UDP(sport=555, dport=556)
1730             / Raw(b"\x00" * 80)
1731         )
1732
1733         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1734
1735         for rx in rxs:
1736             rx = IPv6(peer_1.decrypt_transport(rx))
1737
1738             # check the original packet is present
1739             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1740             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1741
1742         # send packets into the tunnel, expect to receive them on
1743         # the other side
1744         p = [
1745             (
1746                 peer_1.mk_tunnel_header(self.pg1)
1747                 / Wireguard(message_type=4, reserved_zero=0)
1748                 / WireguardTransport(
1749                     receiver_index=peer_1.sender,
1750                     counter=ii + 1,
1751                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1752                         (
1753                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1754                             / UDP(sport=222, dport=223)
1755                             / Raw()
1756                         )
1757                     ),
1758                 )
1759             )
1760             for ii in range(255)
1761         ]
1762
1763         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1764
1765         for rx in rxs:
1766             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1767             self.assertEqual(rx[IPv6].hlim, 19)
1768
1769         r1.remove_vpp_config()
1770         peer_1.remove_vpp_config()
1771         wg0.remove_vpp_config()
1772
1773     def test_wg_peer_v4o6(self):
1774         """Test v4o6"""
1775
1776         port = 12363
1777
1778         # Create interfaces
1779         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1780         wg0.admin_up()
1781         wg0.config_ip4()
1782
1783         peer_1 = VppWgPeer(
1784             self, wg0, self.pg1.remote_ip6, port + 1, ["10.11.3.0/24"]
1785         ).add_vpp_config()
1786         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1787
1788         r1 = VppIpRoute(
1789             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1790         ).add_vpp_config()
1791
1792         # route a packet into the wg interface
1793         #  use the allowed-ip prefix
1794         #  this is dropped because the peer is not initiated
1795         p = (
1796             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1797             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1798             / UDP(sport=555, dport=556)
1799             / Raw()
1800         )
1801         self.send_and_assert_no_replies(self.pg0, [p])
1802         self.assertEqual(
1803             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1804         )
1805
1806         # send a handsake from the peer with an invalid MAC
1807         p = peer_1.mk_handshake(self.pg1, True)
1808         p[WireguardInitiation].mac1 = b"foobar"
1809         self.send_and_assert_no_replies(self.pg1, [p])
1810         self.assertEqual(
1811             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1812         )
1813
1814         # send a handsake from the peer but signed by the wrong key.
1815         p = peer_1.mk_handshake(
1816             self.pg1, True, X25519PrivateKey.generate().public_key()
1817         )
1818         self.send_and_assert_no_replies(self.pg1, [p])
1819         self.assertEqual(
1820             self.base_peer6_in_err + 1,
1821             self.statistics.get_err_counter(self.peer6_in_err),
1822         )
1823
1824         # send a valid handsake init for which we expect a response
1825         p = peer_1.mk_handshake(self.pg1, True)
1826
1827         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1828
1829         peer_1.consume_response(rx[0], True)
1830
1831         # route a packet into the wg interface
1832         #  this is dropped because the peer is still not initiated
1833         p = (
1834             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1835             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1836             / UDP(sport=555, dport=556)
1837             / Raw()
1838         )
1839         self.send_and_assert_no_replies(self.pg0, [p])
1840         self.assertEqual(
1841             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1842         )
1843
1844         # send a data packet from the peer through the tunnel
1845         # this completes the handshake
1846         p = (
1847             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1848             / UDP(sport=222, dport=223)
1849             / Raw()
1850         )
1851         d = peer_1.encrypt_transport(p)
1852         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1853             Wireguard(message_type=4, reserved_zero=0)
1854             / WireguardTransport(
1855                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1856             )
1857         )
1858         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1859
1860         for rx in rxs:
1861             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1862             self.assertEqual(rx[IP].ttl, 19)
1863
1864         # send a packets that are routed into the tunnel
1865         p = (
1866             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1867             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1868             / UDP(sport=555, dport=556)
1869             / Raw(b"\x00" * 80)
1870         )
1871
1872         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1873
1874         for rx in rxs:
1875             rx = IP(peer_1.decrypt_transport(rx, True))
1876
1877             # check the original packet is present
1878             self.assertEqual(rx[IP].dst, p[IP].dst)
1879             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1880
1881         # send packets into the tunnel, expect to receive them on
1882         # the other side
1883         p = [
1884             (
1885                 peer_1.mk_tunnel_header(self.pg1, True)
1886                 / Wireguard(message_type=4, reserved_zero=0)
1887                 / WireguardTransport(
1888                     receiver_index=peer_1.sender,
1889                     counter=ii + 1,
1890                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1891                         (
1892                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1893                             / UDP(sport=222, dport=223)
1894                             / Raw()
1895                         )
1896                     ),
1897                 )
1898             )
1899             for ii in range(255)
1900         ]
1901
1902         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1903
1904         for rx in rxs:
1905             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1906             self.assertEqual(rx[IP].ttl, 19)
1907
1908         r1.remove_vpp_config()
1909         peer_1.remove_vpp_config()
1910         wg0.remove_vpp_config()
1911
1912     def test_wg_multi_peer(self):
1913         """multiple peer setup"""
1914         port = 12373
1915
1916         # Create interfaces
1917         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1918         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
1919         wg0.admin_up()
1920         wg1.admin_up()
1921
1922         # Check peer counter
1923         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
1924
1925         self.pg_enable_capture(self.pg_interfaces)
1926         self.pg_start()
1927
1928         # Create many peers on sencond interface
1929         NUM_PEERS = 16
1930         self.pg2.generate_remote_hosts(NUM_PEERS)
1931         self.pg2.configure_ipv4_neighbors()
1932         self.pg1.generate_remote_hosts(NUM_PEERS)
1933         self.pg1.configure_ipv4_neighbors()
1934
1935         peers_1 = []
1936         peers_2 = []
1937         routes_1 = []
1938         routes_2 = []
1939         for i in range(NUM_PEERS):
1940             peers_1.append(
1941                 VppWgPeer(
1942                     self,
1943                     wg0,
1944                     self.pg1.remote_hosts[i].ip4,
1945                     port + 1 + i,
1946                     ["10.0.%d.4/32" % i],
1947                 ).add_vpp_config()
1948             )
1949             routes_1.append(
1950                 VppIpRoute(
1951                     self,
1952                     "10.0.%d.4" % i,
1953                     32,
1954                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
1955                 ).add_vpp_config()
1956             )
1957
1958             peers_2.append(
1959                 VppWgPeer(
1960                     self,
1961                     wg1,
1962                     self.pg2.remote_hosts[i].ip4,
1963                     port + 100 + i,
1964                     ["10.100.%d.4/32" % i],
1965                 ).add_vpp_config()
1966             )
1967             routes_2.append(
1968                 VppIpRoute(
1969                     self,
1970                     "10.100.%d.4" % i,
1971                     32,
1972                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
1973                 ).add_vpp_config()
1974             )
1975
1976         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
1977
1978         self.logger.info(self.vapi.cli("show wireguard peer"))
1979         self.logger.info(self.vapi.cli("show wireguard interface"))
1980         self.logger.info(self.vapi.cli("show adj 37"))
1981         self.logger.info(self.vapi.cli("sh ip fib 172.16.3.17"))
1982         self.logger.info(self.vapi.cli("sh ip fib 10.11.3.0"))
1983
1984         # remove routes
1985         for r in routes_1:
1986             r.remove_vpp_config()
1987         for r in routes_2:
1988             r.remove_vpp_config()
1989
1990         # remove peers
1991         for p in peers_1:
1992             self.assertTrue(p.query_vpp_config())
1993             p.remove_vpp_config()
1994         for p in peers_2:
1995             self.assertTrue(p.query_vpp_config())
1996             p.remove_vpp_config()
1997
1998         wg0.remove_vpp_config()
1999         wg1.remove_vpp_config()
2000
2001     def test_wg_multi_interface(self):
2002         """Multi-tunnel on the same port"""
2003         port = 12500
2004
2005         # Create many wireguard interfaces
2006         NUM_IFS = 4
2007         self.pg1.generate_remote_hosts(NUM_IFS)
2008         self.pg1.configure_ipv4_neighbors()
2009         self.pg0.generate_remote_hosts(NUM_IFS)
2010         self.pg0.configure_ipv4_neighbors()
2011
2012         # Create interfaces with a peer on each
2013         peers = []
2014         routes = []
2015         wg_ifs = []
2016         for i in range(NUM_IFS):
2017             # Use the same port for each interface
2018             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2019             wg0.admin_up()
2020             wg0.config_ip4()
2021             wg_ifs.append(wg0)
2022             peers.append(
2023                 VppWgPeer(
2024                     self,
2025                     wg0,
2026                     self.pg1.remote_hosts[i].ip4,
2027                     port + 1 + i,
2028                     ["10.0.%d.0/24" % i],
2029                 ).add_vpp_config()
2030             )
2031
2032             routes.append(
2033                 VppIpRoute(
2034                     self,
2035                     "10.0.%d.0" % i,
2036                     24,
2037                     [VppRoutePath("10.0.%d.4" % i, wg0.sw_if_index)],
2038                 ).add_vpp_config()
2039             )
2040
2041         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_IFS)
2042
2043         for i in range(NUM_IFS):
2044             # send a valid handsake init for which we expect a response
2045             p = peers[i].mk_handshake(self.pg1)
2046             rx = self.send_and_expect(self.pg1, [p], self.pg1)
2047             peers[i].consume_response(rx[0])
2048
2049             # send a data packet from the peer through the tunnel
2050             # this completes the handshake
2051             p = (
2052                 IP(src="10.0.%d.4" % i, dst=self.pg0.remote_hosts[i].ip4, ttl=20)
2053                 / UDP(sport=222, dport=223)
2054                 / Raw()
2055             )
2056             d = peers[i].encrypt_transport(p)
2057             p = peers[i].mk_tunnel_header(self.pg1) / (
2058                 Wireguard(message_type=4, reserved_zero=0)
2059                 / WireguardTransport(
2060                     receiver_index=peers[i].sender,
2061                     counter=0,
2062                     encrypted_encapsulated_packet=d,
2063                 )
2064             )
2065             rxs = self.send_and_expect(self.pg1, [p], self.pg0)
2066             for rx in rxs:
2067                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
2068                 self.assertEqual(rx[IP].ttl, 19)
2069
2070         # send a packets that are routed into the tunnel
2071         for i in range(NUM_IFS):
2072             p = (
2073                 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2074                 / IP(src=self.pg0.remote_hosts[i].ip4, dst="10.0.%d.4" % i)
2075                 / UDP(sport=555, dport=556)
2076                 / Raw(b"\x00" * 80)
2077             )
2078
2079             rxs = self.send_and_expect(self.pg0, p * 64, self.pg1)
2080
2081             for rx in rxs:
2082                 rx = IP(peers[i].decrypt_transport(rx))
2083
2084                 # check the oringial packet is present
2085                 self.assertEqual(rx[IP].dst, p[IP].dst)
2086                 self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
2087
2088         # send packets into the tunnel
2089         for i in range(NUM_IFS):
2090             p = [
2091                 (
2092                     peers[i].mk_tunnel_header(self.pg1)
2093                     / Wireguard(message_type=4, reserved_zero=0)
2094                     / WireguardTransport(
2095                         receiver_index=peers[i].sender,
2096                         counter=ii + 1,
2097                         encrypted_encapsulated_packet=peers[i].encrypt_transport(
2098                             (
2099                                 IP(
2100                                     src="10.0.%d.4" % i,
2101                                     dst=self.pg0.remote_hosts[i].ip4,
2102                                     ttl=20,
2103                                 )
2104                                 / UDP(sport=222, dport=223)
2105                                 / Raw()
2106                             )
2107                         ),
2108                     )
2109                 )
2110                 for ii in range(64)
2111             ]
2112
2113             rxs = self.send_and_expect(self.pg1, p, self.pg0)
2114
2115             for rx in rxs:
2116                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
2117                 self.assertEqual(rx[IP].ttl, 19)
2118
2119         for r in routes:
2120             r.remove_vpp_config()
2121         for p in peers:
2122             p.remove_vpp_config()
2123         for i in wg_ifs:
2124             i.remove_vpp_config()
2125
2126     def test_wg_event(self):
2127         """Test events"""
2128         port = 12600
2129         ESTABLISHED_FLAG = (
2130             VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_ESTABLISHED
2131         )
2132         DEAD_FLAG = VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_STATUS_DEAD
2133
2134         # Create interfaces
2135         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2136         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
2137         wg0.admin_up()
2138         wg1.admin_up()
2139
2140         # Check peer counter
2141         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
2142
2143         self.pg_enable_capture(self.pg_interfaces)
2144         self.pg_start()
2145
2146         # Create peers
2147         NUM_PEERS = 2
2148         self.pg2.generate_remote_hosts(NUM_PEERS)
2149         self.pg2.configure_ipv4_neighbors()
2150         self.pg1.generate_remote_hosts(NUM_PEERS)
2151         self.pg1.configure_ipv4_neighbors()
2152
2153         peers_0 = []
2154         peers_1 = []
2155         routes_0 = []
2156         routes_1 = []
2157         for i in range(NUM_PEERS):
2158             peers_0.append(
2159                 VppWgPeer(
2160                     self,
2161                     wg0,
2162                     self.pg1.remote_hosts[i].ip4,
2163                     port + 1 + i,
2164                     ["10.0.%d.4/32" % i],
2165                 ).add_vpp_config()
2166             )
2167             routes_0.append(
2168                 VppIpRoute(
2169                     self,
2170                     "10.0.%d.4" % i,
2171                     32,
2172                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
2173                 ).add_vpp_config()
2174             )
2175
2176             peers_1.append(
2177                 VppWgPeer(
2178                     self,
2179                     wg1,
2180                     self.pg2.remote_hosts[i].ip4,
2181                     port + 100 + i,
2182                     ["10.100.%d.4/32" % i],
2183                 ).add_vpp_config()
2184             )
2185             routes_1.append(
2186                 VppIpRoute(
2187                     self,
2188                     "10.100.%d.4" % i,
2189                     32,
2190                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
2191                 ).add_vpp_config()
2192             )
2193
2194         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
2195
2196         # Want events from the first perr of wg0
2197         # and from all wg1 peers
2198         peers_0[0].want_events()
2199         wg1.want_events()
2200
2201         for i in range(NUM_PEERS):
2202             # send a valid handsake init for which we expect a response
2203             p = peers_0[i].mk_handshake(self.pg1)
2204             rx = self.send_and_expect(self.pg1, [p], self.pg1)
2205             peers_0[i].consume_response(rx[0])
2206             if i == 0:
2207                 peers_0[0].wait_event(ESTABLISHED_FLAG)
2208
2209             p = peers_1[i].mk_handshake(self.pg2)
2210             rx = self.send_and_expect(self.pg2, [p], self.pg2)
2211             peers_1[i].consume_response(rx[0])
2212
2213         wg1.wait_events(ESTABLISHED_FLAG, [peers_1[0].index, peers_1[1].index])
2214
2215         # remove routes
2216         for r in routes_0:
2217             r.remove_vpp_config()
2218         for r in routes_1:
2219             r.remove_vpp_config()
2220
2221         # remove peers
2222         for i in range(NUM_PEERS):
2223             self.assertTrue(peers_0[i].query_vpp_config())
2224             peers_0[i].remove_vpp_config()
2225             if i == 0:
2226                 peers_0[i].wait_event(0)
2227                 peers_0[i].wait_event(DEAD_FLAG)
2228         for p in peers_1:
2229             self.assertTrue(p.query_vpp_config())
2230             p.remove_vpp_config()
2231             p.wait_event(0)
2232             p.wait_event(DEAD_FLAG)
2233
2234         wg0.remove_vpp_config()
2235         wg1.remove_vpp_config()
2236
2237
2238 @tag_fixme_ubuntu2204
2239 @tag_fixme_debian11
2240 class WireguardHandoffTests(TestWg):
2241     """Wireguard Tests in multi worker setup"""
2242
2243     vpp_worker_count = 2
2244
2245     def test_wg_peer_init(self):
2246         """Handoff"""
2247
2248         port = 12383
2249
2250         # Create interfaces
2251         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2252         wg0.admin_up()
2253         wg0.config_ip4()
2254
2255         peer_1 = VppWgPeer(
2256             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.2.0/24", "10.11.3.0/24"]
2257         ).add_vpp_config()
2258         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2259
2260         r1 = VppIpRoute(
2261             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2262         ).add_vpp_config()
2263
2264         # send a valid handsake init for which we expect a response
2265         p = peer_1.mk_handshake(self.pg1)
2266
2267         rx = self.send_and_expect(self.pg1, [p], self.pg1)
2268
2269         peer_1.consume_response(rx[0])
2270
2271         # send a data packet from the peer through the tunnel
2272         # this completes the handshake and pins the peer to worker 0
2273         p = (
2274             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2275             / UDP(sport=222, dport=223)
2276             / Raw()
2277         )
2278         d = peer_1.encrypt_transport(p)
2279         p = peer_1.mk_tunnel_header(self.pg1) / (
2280             Wireguard(message_type=4, reserved_zero=0)
2281             / WireguardTransport(
2282                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
2283             )
2284         )
2285         rxs = self.send_and_expect(self.pg1, [p], self.pg0, worker=0)
2286
2287         for rx in rxs:
2288             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2289             self.assertEqual(rx[IP].ttl, 19)
2290
2291         # send a packets that are routed into the tunnel
2292         # and pins the peer tp worker 1
2293         pe = (
2294             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2295             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2296             / UDP(sport=555, dport=556)
2297             / Raw(b"\x00" * 80)
2298         )
2299         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=1)
2300         peer_1.validate_encapped(rxs, pe)
2301
2302         # send packets into the tunnel, from the other worker
2303         p = [
2304             (
2305                 peer_1.mk_tunnel_header(self.pg1)
2306                 / Wireguard(message_type=4, reserved_zero=0)
2307                 / WireguardTransport(
2308                     receiver_index=peer_1.sender,
2309                     counter=ii + 1,
2310                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
2311                         (
2312                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
2313                             / UDP(sport=222, dport=223)
2314                             / Raw()
2315                         )
2316                     ),
2317                 )
2318             )
2319             for ii in range(255)
2320         ]
2321
2322         rxs = self.send_and_expect(self.pg1, p, self.pg0, worker=1)
2323
2324         for rx in rxs:
2325             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
2326             self.assertEqual(rx[IP].ttl, 19)
2327
2328         # send a packets that are routed into the tunnel
2329         # from owrker 0
2330         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=0)
2331
2332         peer_1.validate_encapped(rxs, pe)
2333
2334         r1.remove_vpp_config()
2335         peer_1.remove_vpp_config()
2336         wg0.remove_vpp_config()
2337
2338     @unittest.skip("test disabled")
2339     def test_wg_multi_interface(self):
2340         """Multi-tunnel on the same port"""
2341
2342
2343 class TestWgFIB(VppTestCase):
2344     """Wireguard FIB Test Case"""
2345
2346     @classmethod
2347     def setUpClass(cls):
2348         super(TestWgFIB, cls).setUpClass()
2349
2350     @classmethod
2351     def tearDownClass(cls):
2352         super(TestWgFIB, cls).tearDownClass()
2353
2354     def setUp(self):
2355         super(TestWgFIB, self).setUp()
2356
2357         self.create_pg_interfaces(range(2))
2358
2359         for i in self.pg_interfaces:
2360             i.admin_up()
2361             i.config_ip4()
2362
2363     def tearDown(self):
2364         for i in self.pg_interfaces:
2365             i.unconfig_ip4()
2366             i.admin_down()
2367         super(TestWgFIB, self).tearDown()
2368
2369     def test_wg_fib_tracking(self):
2370         """FIB tracking"""
2371         port = 12323
2372
2373         # create wg interface
2374         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
2375         wg0.admin_up()
2376         wg0.config_ip4()
2377
2378         self.pg_enable_capture(self.pg_interfaces)
2379         self.pg_start()
2380
2381         # create a peer
2382         peer_1 = VppWgPeer(
2383             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
2384         ).add_vpp_config()
2385         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
2386
2387         # create a route to rewrite traffic into the wg interface
2388         r1 = VppIpRoute(
2389             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
2390         ).add_vpp_config()
2391
2392         # resolve ARP and expect the adjacency to update
2393         self.pg1.resolve_arp()
2394
2395         # wait for the peer to send a handshake initiation
2396         rxs = self.pg1.get_capture(2, timeout=6)
2397
2398         # prepare and send a handshake response
2399         # expect a keepalive message
2400         resp = peer_1.consume_init(rxs[1], self.pg1)
2401         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
2402
2403         # verify the keepalive message
2404         b = peer_1.decrypt_transport(rxs[0])
2405         self.assertEqual(0, len(b))
2406
2407         # prepare and send a packet that will be rewritten into the wg interface
2408         # expect a data packet sent to the new endpoint
2409         p = (
2410             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2411             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
2412             / UDP(sport=555, dport=556)
2413             / Raw()
2414         )
2415         rxs = self.send_and_expect(self.pg0, [p], self.pg1)
2416
2417         # verify the data packet
2418         peer_1.validate_encapped(rxs, p)
2419
2420         # remove configs
2421         r1.remove_vpp_config()
2422         peer_1.remove_vpp_config()
2423         wg0.remove_vpp_config()