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