test/test_wireguard.py

   1 #!/usr/bin/env python3
   2 """ Wg tests """
   3
   4 import datetime
   5 import base64
   6 import os
   7
   8 from hashlib import blake2s
   9 from scapy.packet import Packet
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether, ARP
  12 from scapy.layers.inet import IP, UDP
  13 from scapy.layers.inet6 import IPv6
  14 from scapy.contrib.wireguard import (
  15     Wireguard,
  16     WireguardResponse,
  17     WireguardInitiation,
  18     WireguardTransport,
  19     WireguardCookieReply,
  20 )
  21 from cryptography.hazmat.primitives.asymmetric.x25519 import (
  22     X25519PrivateKey,
  23     X25519PublicKey,
  24 )
  25 from cryptography.hazmat.primitives.serialization import (
  26     Encoding,
  27     PrivateFormat,
  28     PublicFormat,
  29     NoEncryption,
  30 )
  31 from cryptography.hazmat.primitives.hashes import BLAKE2s, Hash
  32 from cryptography.hazmat.primitives.hmac import HMAC
  33 from cryptography.hazmat.backends import default_backend
  34 from noise.connection import NoiseConnection, Keypair
  35
  36 from Crypto.Cipher import ChaCha20_Poly1305
  37 from Crypto.Random import get_random_bytes
  38
  39 from vpp_ipip_tun_interface import VppIpIpTunInterface
  40 from vpp_interface import VppInterface
  41 from vpp_ip_route import VppIpRoute, VppRoutePath
  42 from vpp_object import VppObject
  43 from vpp_papi import VppEnum
  44 from framework import VppTestCase
  45 from re import compile
  46 import unittest
  47
  48 """ TestWg is a subclass of  VPPTestCase classes.
  49
  50 Wg test.
  51
  52 """
  53
  54
  55 def private_key_bytes(k):
  56     return k.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())
  57
  58
  59 def public_key_bytes(k):
  60     return k.public_bytes(Encoding.Raw, PublicFormat.Raw)
  61
  62
  63 def get_field_bytes(pkt, name):
  64     fld, val = pkt.getfield_and_val(name)
  65     return fld.i2m(pkt, val)
  66
  67
  68 class VppWgInterface(VppInterface):
  69     """
  70     VPP WireGuard interface
  71     """
  72
  73     def __init__(self, test, src, port):
  74         super(VppWgInterface, self).__init__(test)
  75
  76         self.port = port
  77         self.src = src
  78         self.private_key = X25519PrivateKey.generate()
  79         self.public_key = self.private_key.public_key()
  80
  81     def public_key_bytes(self):
  82         return public_key_bytes(self.public_key)
  83
  84     def private_key_bytes(self):
  85         return private_key_bytes(self.private_key)
  86
  87     def add_vpp_config(self):
  88         r = self.test.vapi.wireguard_interface_create(
  89             interface={
  90                 "user_instance": 0xFFFFFFFF,
  91                 "port": self.port,
  92                 "src_ip": self.src,
  93                 "private_key": private_key_bytes(self.private_key),
  94                 "generate_key": False,
  95             }
  96         )
  97         self.set_sw_if_index(r.sw_if_index)
  98         self.test.registry.register(self, self.test.logger)
  99         return self
 100
 101     def remove_vpp_config(self):
 102         self.test.vapi.wireguard_interface_delete(sw_if_index=self._sw_if_index)
 103
 104     def query_vpp_config(self):
 105         ts = self.test.vapi.wireguard_interface_dump(sw_if_index=0xFFFFFFFF)
 106         for t in ts:
 107             if (
 108                 t.interface.sw_if_index == self._sw_if_index
 109                 and str(t.interface.src_ip) == self.src
 110                 and t.interface.port == self.port
 111                 and t.interface.private_key == private_key_bytes(self.private_key)
 112             ):
 113                 return True
 114         return False
 115
 116     def want_events(self, peer_index=0xFFFFFFFF):
 117         self.test.vapi.want_wireguard_peer_events(
 118             enable_disable=1,
 119             pid=os.getpid(),
 120             sw_if_index=self._sw_if_index,
 121             peer_index=peer_index,
 122         )
 123
 124     def wait_events(self, expect, peers, timeout=5):
 125         for i in range(len(peers)):
 126             rv = self.test.vapi.wait_for_event(timeout, "wireguard_peer_event")
 127             self.test.assertEqual(rv.peer_index, peers[i])
 128             self.test.assertEqual(rv.flags, expect)
 129
 130     def __str__(self):
 131         return self.object_id()
 132
 133     def object_id(self):
 134         return "wireguard-%d" % self._sw_if_index
 135
 136
 137 def find_route(test, prefix, is_ip6, table_id=0):
 138     routes = test.vapi.ip_route_dump(table_id, is_ip6)
 139
 140     for e in routes:
 141         if table_id == e.route.table_id and str(e.route.prefix) == str(prefix):
 142             return True
 143     return False
 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
 150 class VppWgPeer(VppObject):
 151     def __init__(self, test, itf, endpoint, port, allowed_ips, persistent_keepalive=15):
 152         self._test = test
 153         self.itf = itf
 154         self.endpoint = endpoint
 155         self.port = port
 156         self.allowed_ips = allowed_ips
 157         self.persistent_keepalive = persistent_keepalive
 158
 159         # remote peer's public
 160         self.private_key = X25519PrivateKey.generate()
 161         self.public_key = self.private_key.public_key()
 162
 163         # cookie related params
 164         self.cookie_key = blake2s(b"cookie--" + self.public_key_bytes()).digest()
 165         self.last_sent_cookie = None
 166
 167         self.noise = NoiseConnection.from_name(NOISE_HANDSHAKE_NAME)
 168
 169     def add_vpp_config(self, is_ip6=False):
 170         rv = self._test.vapi.wireguard_peer_add(
 171             peer={
 172                 "public_key": self.public_key_bytes(),
 173                 "port": self.port,
 174                 "endpoint": self.endpoint,
 175                 "n_allowed_ips": len(self.allowed_ips),
 176                 "allowed_ips": self.allowed_ips,
 177                 "sw_if_index": self.itf.sw_if_index,
 178                 "persistent_keepalive": self.persistent_keepalive,
 179             }
 180         )
 181         self.index = rv.peer_index
 182         self.receiver_index = self.index + 1
 183         self._test.registry.register(self, self._test.logger)
 184         return self
 185
 186     def remove_vpp_config(self):
 187         self._test.vapi.wireguard_peer_remove(peer_index=self.index)
 188
 189     def object_id(self):
 190         return "wireguard-peer-%s" % self.index
 191
 192     def public_key_bytes(self):
 193         return public_key_bytes(self.public_key)
 194
 195     def query_vpp_config(self):
 196         peers = self._test.vapi.wireguard_peers_dump()
 197
 198         for p in peers:
 199             if (
 200                 p.peer.public_key == self.public_key_bytes()
 201                 and p.peer.port == self.port
 202                 and str(p.peer.endpoint) == self.endpoint
 203                 and p.peer.sw_if_index == self.itf.sw_if_index
 204                 and len(self.allowed_ips) == p.peer.n_allowed_ips
 205             ):
 206                 self.allowed_ips.sort()
 207                 p.peer.allowed_ips.sort()
 208
 209                 for (a1, a2) in zip(self.allowed_ips, p.peer.allowed_ips):
 210                     if str(a1) != str(a2):
 211                         return False
 212                 return True
 213         return False
 214
 215     def mk_tunnel_header(self, tx_itf, is_ip6=False):
 216         if is_ip6 is False:
 217             return (
 218                 Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
 219                 / IP(src=self.endpoint, dst=self.itf.src)
 220                 / UDP(sport=self.port, dport=self.itf.port)
 221             )
 222         else:
 223             return (
 224                 Ether(dst=tx_itf.local_mac, src=tx_itf.remote_mac)
 225                 / IPv6(src=self.endpoint, dst=self.itf.src)
 226                 / UDP(sport=self.port, dport=self.itf.port)
 227             )
 228
 229     def noise_init(self, public_key=None):
 230         self.noise.set_prologue(NOISE_IDENTIFIER_NAME)
 231         self.noise.set_psks(psk=bytes(bytearray(32)))
 232
 233         if not public_key:
 234             public_key = self.itf.public_key
 235
 236         # local/this private
 237         self.noise.set_keypair_from_private_bytes(
 238             Keypair.STATIC, private_key_bytes(self.private_key)
 239         )
 240         # remote's public
 241         self.noise.set_keypair_from_public_bytes(
 242             Keypair.REMOTE_STATIC, public_key_bytes(public_key)
 243         )
 244
 245         self.noise.start_handshake()
 246
 247     def mk_cookie(self, p, tx_itf, is_resp=False, is_ip6=False):
 248         self.verify_header(p, is_ip6)
 249
 250         wg_pkt = Wireguard(p[Raw])
 251
 252         if is_resp:
 253             self._test.assertEqual(wg_pkt[Wireguard].message_type, 2)
 254             self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
 255             self._test.assertEqual(wg_pkt[WireguardResponse].mac2, bytes([0] * 16))
 256         else:
 257             self._test.assertEqual(wg_pkt[Wireguard].message_type, 1)
 258             self._test.assertEqual(wg_pkt[Wireguard].reserved_zero, 0)
 259             self._test.assertEqual(wg_pkt[WireguardInitiation].mac2, bytes([0] * 16))
 260
 261         # collect info from wg packet (initiation or response)
 262         src = get_field_bytes(p[IPv6 if is_ip6 else IP], "src")
 263         sport = p[UDP].sport.to_bytes(2, byteorder="big")
 264         if is_resp:
 265             mac1 = wg_pkt[WireguardResponse].mac1
 266             sender_index = wg_pkt[WireguardResponse].sender_index
 267         else:
 268             mac1 = wg_pkt[WireguardInitiation].mac1
 269             sender_index = wg_pkt[WireguardInitiation].sender_index
 270
 271         # make cookie reply
 272         cookie_reply = Wireguard() / WireguardCookieReply()
 273         cookie_reply[Wireguard].message_type = 3
 274         cookie_reply[Wireguard].reserved_zero = 0
 275         cookie_reply[WireguardCookieReply].receiver_index = sender_index
 276         nonce = get_random_bytes(24)
 277         cookie_reply[WireguardCookieReply].nonce = nonce
 278
 279         # generate cookie data
 280         changing_secret = get_random_bytes(32)
 281         self.last_sent_cookie = blake2s(
 282             src + sport, digest_size=16, key=changing_secret
 283         ).digest()
 284
 285         # encrypt cookie data
 286         cipher = ChaCha20_Poly1305.new(key=self.cookie_key, nonce=nonce)
 287         cipher.update(mac1)
 288         ciphertext, tag = cipher.encrypt_and_digest(self.last_sent_cookie)
 289         cookie_reply[WireguardCookieReply].encrypted_cookie = ciphertext + tag
 290
 291         # prepare cookie reply to be sent
 292         cookie_reply = self.mk_tunnel_header(tx_itf, is_ip6) / cookie_reply
 293
 294         return cookie_reply
 295
 296     def mk_handshake(self, tx_itf, is_ip6=False, public_key=None):
 297         self.noise.set_as_initiator()
 298         self.noise_init(public_key)
 299
 300         p = Wireguard() / WireguardInitiation()
 301
 302         p[Wireguard].message_type = 1
 303         p[Wireguard].reserved_zero = 0
 304         p[WireguardInitiation].sender_index = self.receiver_index
 305
 306         # some random data for the message
 307         #  lifted from the noise protocol's wireguard example
 308         now = datetime.datetime.now()
 309         tai = struct.pack(
 310             "!qi",
 311             4611686018427387914 + int(now.timestamp()),
 312             int(now.microsecond * 1e3),
 313         )
 314         b = self.noise.write_message(payload=tai)
 315
 316         # load noise into init message
 317         p[WireguardInitiation].unencrypted_ephemeral = b[0:32]
 318         p[WireguardInitiation].encrypted_static = b[32:80]
 319         p[WireguardInitiation].encrypted_timestamp = b[80:108]
 320
 321         # generate the mac1 hash
 322         mac_key = blake2s(b"mac1----" + self.itf.public_key_bytes()).digest()
 323         p[WireguardInitiation].mac1 = blake2s(
 324             bytes(p)[0:116], digest_size=16, key=mac_key
 325         ).digest()
 326         p[WireguardInitiation].mac2 = bytearray(16)
 327
 328         p = self.mk_tunnel_header(tx_itf, is_ip6) / p
 329
 330         return p
 331
 332     def verify_header(self, p, is_ip6=False):
 333         if is_ip6 is False:
 334             self._test.assertEqual(p[IP].src, self.itf.src)
 335             self._test.assertEqual(p[IP].dst, self.endpoint)
 336         else:
 337             self._test.assertEqual(p[IPv6].src, self.itf.src)
 338             self._test.assertEqual(p[IPv6].dst, self.endpoint)
 339         self._test.assertEqual(p[UDP].sport, self.itf.port)
 340         self._test.assertEqual(p[UDP].dport, self.port)
 341         self._test.assert_packet_checksums_valid(p)
 342
 343     def consume_init(self, p, tx_itf, is_ip6=False, is_mac2=False):
 344         self.noise.set_as_responder()
 345         self.noise_init(self.itf.public_key)
 346         self.verify_header(p, is_ip6)
 347
 348         init = Wireguard(p[Raw])
 349
 350         self._test.assertEqual(init[Wireguard].message_type, 1)
 351         self._test.assertEqual(init[Wireguard].reserved_zero, 0)
 352
 353         self.sender = init[WireguardInitiation].sender_index
 354
 355         # validate the mac1 hash
 356         mac_key = blake2s(b"mac1----" + public_key_bytes(self.public_key)).digest()
 357         mac1 = blake2s(bytes(init)[0:-32], digest_size=16, key=mac_key).digest()
 358         self._test.assertEqual(init[WireguardInitiation].mac1, mac1)
 359
 360         # validate the mac2 hash
 361         if is_mac2:
 362             self._test.assertNotEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
 363             self._test.assertNotEqual(self.last_sent_cookie, None)
 364             mac2 = blake2s(
 365                 bytes(init)[0:-16], digest_size=16, key=self.last_sent_cookie
 366             ).digest()
 367             self._test.assertEqual(init[WireguardInitiation].mac2, mac2)
 368             self.last_sent_cookie = None
 369         else:
 370             self._test.assertEqual(init[WireguardInitiation].mac2, bytes([0] * 16))
 371
 372         # this passes only unencrypted_ephemeral, encrypted_static,
 373         # encrypted_timestamp fields of the init
 374         payload = self.noise.read_message(bytes(init)[8:-32])
 375
 376         # build the response
 377         b = self.noise.write_message()
 378         mac_key = blake2s(b"mac1----" + public_key_bytes(self.itf.public_key)).digest()
 379         resp = Wireguard(message_type=2, reserved_zero=0) / WireguardResponse(
 380             sender_index=self.receiver_index,
 381             receiver_index=self.sender,
 382             unencrypted_ephemeral=b[0:32],
 383             encrypted_nothing=b[32:],
 384         )
 385         mac1 = blake2s(bytes(resp)[:-32], digest_size=16, key=mac_key).digest()
 386         resp[WireguardResponse].mac1 = mac1
 387
 388         resp = self.mk_tunnel_header(tx_itf, is_ip6) / resp
 389         self._test.assertTrue(self.noise.handshake_finished)
 390
 391         return resp
 392
 393     def consume_response(self, p, is_ip6=False):
 394         self.verify_header(p, is_ip6)
 395
 396         resp = Wireguard(p[Raw])
 397
 398         self._test.assertEqual(resp[Wireguard].message_type, 2)
 399         self._test.assertEqual(resp[Wireguard].reserved_zero, 0)
 400         self._test.assertEqual(
 401             resp[WireguardResponse].receiver_index, self.receiver_index
 402         )
 403
 404         self.sender = resp[Wireguard].sender_index
 405
 406         payload = self.noise.read_message(bytes(resp)[12:60])
 407         self._test.assertEqual(payload, b"")
 408         self._test.assertTrue(self.noise.handshake_finished)
 409
 410     def decrypt_transport(self, p, is_ip6=False):
 411         self.verify_header(p, is_ip6)
 412
 413         p = Wireguard(p[Raw])
 414         self._test.assertEqual(p[Wireguard].message_type, 4)
 415         self._test.assertEqual(p[Wireguard].reserved_zero, 0)
 416         self._test.assertEqual(
 417             p[WireguardTransport].receiver_index, self.receiver_index
 418         )
 419
 420         d = self.noise.decrypt(p[WireguardTransport].encrypted_encapsulated_packet)
 421         return d
 422
 423     def encrypt_transport(self, p):
 424         return self.noise.encrypt(bytes(p))
 425
 426     def validate_encapped(self, rxs, tx, is_ip6=False):
 427         for rx in rxs:
 428             if is_ip6 is False:
 429                 rx = IP(self.decrypt_transport(rx))
 430
 431                 # chech the oringial packet is present
 432                 self._test.assertEqual(rx[IP].dst, tx[IP].dst)
 433                 self._test.assertEqual(rx[IP].ttl, tx[IP].ttl - 1)
 434             else:
 435                 rx = IPv6(self.decrypt_transport(rx))
 436
 437                 # chech the oringial packet is present
 438                 self._test.assertEqual(rx[IPv6].dst, tx[IPv6].dst)
 439                 self._test.assertEqual(rx[IPv6].ttl, tx[IPv6].ttl - 1)
 440
 441     def want_events(self):
 442         self._test.vapi.want_wireguard_peer_events(
 443             enable_disable=1,
 444             pid=os.getpid(),
 445             peer_index=self.index,
 446             sw_if_index=self.itf.sw_if_index,
 447         )
 448
 449     def wait_event(self, expect, timeout=5):
 450         rv = self._test.vapi.wait_for_event(timeout, "wireguard_peer_event")
 451         self._test.assertEqual(rv.flags, expect)
 452         self._test.assertEqual(rv.peer_index, self.index)
 453
 454
 455 class TestWg(VppTestCase):
 456     """Wireguard Test Case"""
 457
 458     error_str = compile(r"Error")
 459
 460     wg4_output_node_name = "/err/wg4-output-tun/"
 461     wg4_input_node_name = "/err/wg4-input/"
 462     wg6_output_node_name = "/err/wg6-output-tun/"
 463     wg6_input_node_name = "/err/wg6-input/"
 464     kp4_error = wg4_output_node_name + "Keypair error"
 465     mac4_error = wg4_input_node_name + "Invalid MAC handshake"
 466     peer4_in_err = wg4_input_node_name + "Peer error"
 467     peer4_out_err = wg4_output_node_name + "Peer error"
 468     kp6_error = wg6_output_node_name + "Keypair error"
 469     mac6_error = wg6_input_node_name + "Invalid MAC handshake"
 470     peer6_in_err = wg6_input_node_name + "Peer error"
 471     peer6_out_err = wg6_output_node_name + "Peer error"
 472     cookie_dec4_err = wg4_input_node_name + "Failed during Cookie decryption"
 473     cookie_dec6_err = wg6_input_node_name + "Failed during Cookie decryption"
 474
 475     @classmethod
 476     def setUpClass(cls):
 477         super(TestWg, cls).setUpClass()
 478         try:
 479             cls.create_pg_interfaces(range(3))
 480             for i in cls.pg_interfaces:
 481                 i.admin_up()
 482                 i.config_ip4()
 483                 i.config_ip6()
 484                 i.resolve_arp()
 485                 i.resolve_ndp()
 486
 487         except Exception:
 488             super(TestWg, cls).tearDownClass()
 489             raise
 490
 491     @classmethod
 492     def tearDownClass(cls):
 493         super(TestWg, cls).tearDownClass()
 494
 495     def setUp(self):
 496         super(VppTestCase, self).setUp()
 497         self.base_kp4_err = self.statistics.get_err_counter(self.kp4_error)
 498         self.base_mac4_err = self.statistics.get_err_counter(self.mac4_error)
 499         self.base_peer4_in_err = self.statistics.get_err_counter(self.peer4_in_err)
 500         self.base_peer4_out_err = self.statistics.get_err_counter(self.peer4_out_err)
 501         self.base_kp6_err = self.statistics.get_err_counter(self.kp6_error)
 502         self.base_mac6_err = self.statistics.get_err_counter(self.mac6_error)
 503         self.base_peer6_in_err = self.statistics.get_err_counter(self.peer6_in_err)
 504         self.base_peer6_out_err = self.statistics.get_err_counter(self.peer6_out_err)
 505         self.base_cookie_dec4_err = self.statistics.get_err_counter(
 506             self.cookie_dec4_err
 507         )
 508         self.base_cookie_dec6_err = self.statistics.get_err_counter(
 509             self.cookie_dec6_err
 510         )
 511
 512     def test_wg_interface(self):
 513         """Simple interface creation"""
 514         port = 12312
 515
 516         # Create interface
 517         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 518
 519         self.logger.info(self.vapi.cli("sh int"))
 520
 521         # delete interface
 522         wg0.remove_vpp_config()
 523
 524     def test_handshake_hash(self):
 525         """test hashing an init message"""
 526         # a init packet generated by linux given the key below
 527         h = (
 528             "0100000098b9032b"
 529             "55cc4b39e73c3d24"
 530             "a2a1ab884b524a81"
 531             "1808bb86640fb70d"
 532             "e93154fec1879125"
 533             "ab012624a27f0b75"
 534             "c0a2582f438ddb5f"
 535             "8e768af40b4ab444"
 536             "02f9ff473e1b797e"
 537             "80d39d93c5480c82"
 538             "a3d4510f70396976"
 539             "586fb67300a5167b"
 540             "ae6ca3ff3dfd00eb"
 541             "59be198810f5aa03"
 542             "6abc243d2155ee4f"
 543             "2336483900aef801"
 544             "08752cd700000000"
 545             "0000000000000000"
 546             "00000000"
 547         )
 548
 549         b = bytearray.fromhex(h)
 550         tgt = Wireguard(b)
 551
 552         pubb = base64.b64decode("aRuHFTTxICIQNefp05oKWlJv3zgKxb8+WW7JJMh0jyM=")
 553         pub = X25519PublicKey.from_public_bytes(pubb)
 554
 555         self.assertEqual(pubb, public_key_bytes(pub))
 556
 557         # strip the macs and build a new packet
 558         init = b[0:-32]
 559         mac_key = blake2s(b"mac1----" + public_key_bytes(pub)).digest()
 560         init += blake2s(init, digest_size=16, key=mac_key).digest()
 561         init += b"\x00" * 16
 562
 563         act = Wireguard(init)
 564
 565         self.assertEqual(tgt, act)
 566
 567     def _test_wg_send_cookie_tmpl(self, is_resp, is_ip6):
 568         port = 12323
 569
 570         # create wg interface
 571         if is_ip6:
 572             wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 573             wg0.admin_up()
 574             wg0.config_ip6()
 575         else:
 576             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 577             wg0.admin_up()
 578             wg0.config_ip4()
 579
 580         self.pg_enable_capture(self.pg_interfaces)
 581         self.pg_start()
 582
 583         # create a peer
 584         if is_ip6:
 585             peer_1 = VppWgPeer(
 586                 self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 587             ).add_vpp_config()
 588         else:
 589             peer_1 = VppWgPeer(
 590                 self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 591             ).add_vpp_config()
 592         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 593
 594         if is_resp:
 595             # prepare and send a handshake initiation
 596             # expect the peer to send a handshake response
 597             init = peer_1.mk_handshake(self.pg1, is_ip6=is_ip6)
 598             rxs = self.send_and_expect(self.pg1, [init], self.pg1)
 599         else:
 600             # wait for the peer to send a handshake initiation
 601             rxs = self.pg1.get_capture(1, timeout=2)
 602
 603         # prepare and send a wrong cookie reply
 604         # expect no replies and the cookie error incremented
 605         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 606         cookie.nonce = b"1234567890"
 607         self.send_and_assert_no_replies(self.pg1, [cookie], timeout=0.1)
 608         if is_ip6:
 609             self.assertEqual(
 610                 self.base_cookie_dec6_err + 1,
 611                 self.statistics.get_err_counter(self.cookie_dec6_err),
 612             )
 613         else:
 614             self.assertEqual(
 615                 self.base_cookie_dec4_err + 1,
 616                 self.statistics.get_err_counter(self.cookie_dec4_err),
 617             )
 618
 619         # prepare and send a correct cookie reply
 620         cookie = peer_1.mk_cookie(rxs[0], self.pg1, is_resp=is_resp, is_ip6=is_ip6)
 621         self.pg_send(self.pg1, [cookie])
 622
 623         # wait for the peer to send a handshake initiation with mac2 set
 624         rxs = self.pg1.get_capture(1, timeout=6)
 625
 626         # verify the initiation and its mac2
 627         peer_1.consume_init(rxs[0], self.pg1, is_ip6=is_ip6, is_mac2=True)
 628
 629         # remove configs
 630         peer_1.remove_vpp_config()
 631         wg0.remove_vpp_config()
 632
 633     def test_wg_send_cookie_on_init_v4(self):
 634         """Send cookie on handshake initiation (v4)"""
 635         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=False)
 636
 637     def test_wg_send_cookie_on_init_v6(self):
 638         """Send cookie on handshake initiation (v6)"""
 639         self._test_wg_send_cookie_tmpl(is_resp=False, is_ip6=True)
 640
 641     def test_wg_send_cookie_on_resp_v4(self):
 642         """Send cookie on handshake response (v4)"""
 643         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=False)
 644
 645     def test_wg_send_cookie_on_resp_v6(self):
 646         """Send cookie on handshake response (v6)"""
 647         self._test_wg_send_cookie_tmpl(is_resp=True, is_ip6=True)
 648
 649     def test_wg_peer_resp(self):
 650         """Send handshake response"""
 651         port = 12323
 652
 653         # Create interfaces
 654         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 655         wg0.admin_up()
 656         wg0.config_ip4()
 657
 658         self.pg_enable_capture(self.pg_interfaces)
 659         self.pg_start()
 660
 661         peer_1 = VppWgPeer(
 662             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 663         ).add_vpp_config()
 664         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 665
 666         r1 = VppIpRoute(
 667             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
 668         ).add_vpp_config()
 669
 670         # wait for the peer to send a handshake
 671         rx = self.pg1.get_capture(1, timeout=2)
 672
 673         # consume the handshake in the noise protocol and
 674         # generate the response
 675         resp = peer_1.consume_init(rx[0], self.pg1)
 676
 677         # send the response, get keepalive
 678         rxs = self.send_and_expect(self.pg1, [resp], self.pg1)
 679
 680         for rx in rxs:
 681             b = peer_1.decrypt_transport(rx)
 682             self.assertEqual(0, len(b))
 683
 684         # send a packets that are routed into the tunnel
 685         p = (
 686             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 687             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
 688             / UDP(sport=555, dport=556)
 689             / Raw(b"\x00" * 80)
 690         )
 691
 692         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
 693
 694         peer_1.validate_encapped(rxs, p)
 695
 696         # send packets into the tunnel, expect to receive them on
 697         # the other side
 698         p = [
 699             (
 700                 peer_1.mk_tunnel_header(self.pg1)
 701                 / Wireguard(message_type=4, reserved_zero=0)
 702                 / WireguardTransport(
 703                     receiver_index=peer_1.sender,
 704                     counter=ii,
 705                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
 706                         (
 707                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
 708                             / UDP(sport=222, dport=223)
 709                             / Raw()
 710                         )
 711                     ),
 712                 )
 713             )
 714             for ii in range(255)
 715         ]
 716
 717         rxs = self.send_and_expect(self.pg1, p, self.pg0)
 718
 719         for rx in rxs:
 720             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 721             self.assertEqual(rx[IP].ttl, 19)
 722
 723         r1.remove_vpp_config()
 724         peer_1.remove_vpp_config()
 725         wg0.remove_vpp_config()
 726
 727     def test_wg_peer_v4o4(self):
 728         """Test v4o4"""
 729
 730         port = 12333
 731
 732         # Create interfaces
 733         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
 734         wg0.admin_up()
 735         wg0.config_ip4()
 736
 737         peer_1 = VppWgPeer(
 738             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.3.0/24"]
 739         ).add_vpp_config()
 740         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 741
 742         r1 = VppIpRoute(
 743             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
 744         ).add_vpp_config()
 745         r2 = VppIpRoute(
 746             self, "20.22.3.0", 24, [VppRoutePath("20.22.3.1", wg0.sw_if_index)]
 747         ).add_vpp_config()
 748
 749         # route a packet into the wg interface
 750         #  use the allowed-ip prefix
 751         #  this is dropped because the peer is not initiated
 752         p = (
 753             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 754             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
 755             / UDP(sport=555, dport=556)
 756             / Raw()
 757         )
 758         self.send_and_assert_no_replies(self.pg0, [p])
 759         self.assertEqual(
 760             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
 761         )
 762
 763         # route a packet into the wg interface
 764         #  use a not allowed-ip prefix
 765         #  this is dropped because there is no matching peer
 766         p = (
 767             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 768             / IP(src=self.pg0.remote_ip4, dst="20.22.3.2")
 769             / UDP(sport=555, dport=556)
 770             / Raw()
 771         )
 772         self.send_and_assert_no_replies(self.pg0, [p])
 773         self.assertEqual(
 774             self.base_peer4_out_err + 1,
 775             self.statistics.get_err_counter(self.peer4_out_err),
 776         )
 777
 778         # send a handsake from the peer with an invalid MAC
 779         p = peer_1.mk_handshake(self.pg1)
 780         p[WireguardInitiation].mac1 = b"foobar"
 781         self.send_and_assert_no_replies(self.pg1, [p])
 782         self.assertEqual(
 783             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
 784         )
 785
 786         # send a handsake from the peer but signed by the wrong key.
 787         p = peer_1.mk_handshake(
 788             self.pg1, False, X25519PrivateKey.generate().public_key()
 789         )
 790         self.send_and_assert_no_replies(self.pg1, [p])
 791         self.assertEqual(
 792             self.base_peer4_in_err + 1,
 793             self.statistics.get_err_counter(self.peer4_in_err),
 794         )
 795
 796         # send a valid handsake init for which we expect a response
 797         p = peer_1.mk_handshake(self.pg1)
 798
 799         rx = self.send_and_expect(self.pg1, [p], self.pg1)
 800
 801         peer_1.consume_response(rx[0])
 802
 803         # route a packet into the wg interface
 804         #  this is dropped because the peer is still not initiated
 805         p = (
 806             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 807             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
 808             / UDP(sport=555, dport=556)
 809             / Raw()
 810         )
 811         self.send_and_assert_no_replies(self.pg0, [p])
 812         self.assertEqual(
 813             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
 814         )
 815
 816         # send a data packet from the peer through the tunnel
 817         # this completes the handshake
 818         p = (
 819             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
 820             / UDP(sport=222, dport=223)
 821             / Raw()
 822         )
 823         d = peer_1.encrypt_transport(p)
 824         p = peer_1.mk_tunnel_header(self.pg1) / (
 825             Wireguard(message_type=4, reserved_zero=0)
 826             / WireguardTransport(
 827                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
 828             )
 829         )
 830         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
 831
 832         for rx in rxs:
 833             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 834             self.assertEqual(rx[IP].ttl, 19)
 835
 836         # send a packets that are routed into the tunnel
 837         p = (
 838             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 839             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
 840             / UDP(sport=555, dport=556)
 841             / Raw(b"\x00" * 80)
 842         )
 843
 844         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
 845
 846         for rx in rxs:
 847             rx = IP(peer_1.decrypt_transport(rx))
 848
 849             # chech the oringial packet is present
 850             self.assertEqual(rx[IP].dst, p[IP].dst)
 851             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
 852
 853         # send packets into the tunnel, expect to receive them on
 854         # the other side
 855         p = [
 856             (
 857                 peer_1.mk_tunnel_header(self.pg1)
 858                 / Wireguard(message_type=4, reserved_zero=0)
 859                 / WireguardTransport(
 860                     receiver_index=peer_1.sender,
 861                     counter=ii + 1,
 862                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
 863                         (
 864                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
 865                             / UDP(sport=222, dport=223)
 866                             / Raw()
 867                         )
 868                     ),
 869                 )
 870             )
 871             for ii in range(255)
 872         ]
 873
 874         rxs = self.send_and_expect(self.pg1, p, self.pg0)
 875
 876         for rx in rxs:
 877             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 878             self.assertEqual(rx[IP].ttl, 19)
 879
 880         r1.remove_vpp_config()
 881         r2.remove_vpp_config()
 882         peer_1.remove_vpp_config()
 883         wg0.remove_vpp_config()
 884
 885     def test_wg_peer_v6o6(self):
 886         """Test v6o6"""
 887
 888         port = 12343
 889
 890         # Create interfaces
 891         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
 892         wg0.admin_up()
 893         wg0.config_ip6()
 894
 895         peer_1 = VppWgPeer(
 896             self, wg0, self.pg1.remote_ip6, port + 1, ["1::3:0/112"]
 897         ).add_vpp_config(True)
 898         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
 899
 900         r1 = VppIpRoute(
 901             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
 902         ).add_vpp_config()
 903         r2 = VppIpRoute(
 904             self, "22::3:0", 112, [VppRoutePath("22::3:1", wg0.sw_if_index)]
 905         ).add_vpp_config()
 906
 907         # route a packet into the wg interface
 908         #  use the allowed-ip prefix
 909         #  this is dropped because the peer is not initiated
 910
 911         p = (
 912             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 913             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
 914             / UDP(sport=555, dport=556)
 915             / Raw()
 916         )
 917         self.send_and_assert_no_replies(self.pg0, [p])
 918
 919         self.assertEqual(
 920             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
 921         )
 922
 923         # route a packet into the wg interface
 924         #  use a not allowed-ip prefix
 925         #  this is dropped because there is no matching peer
 926         p = (
 927             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 928             / IPv6(src=self.pg0.remote_ip6, dst="22::3:2")
 929             / UDP(sport=555, dport=556)
 930             / Raw()
 931         )
 932         self.send_and_assert_no_replies(self.pg0, [p])
 933         self.assertEqual(
 934             self.base_peer6_out_err + 1,
 935             self.statistics.get_err_counter(self.peer6_out_err),
 936         )
 937
 938         # send a handsake from the peer with an invalid MAC
 939         p = peer_1.mk_handshake(self.pg1, True)
 940         p[WireguardInitiation].mac1 = b"foobar"
 941         self.send_and_assert_no_replies(self.pg1, [p])
 942
 943         self.assertEqual(
 944             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
 945         )
 946
 947         # send a handsake from the peer but signed by the wrong key.
 948         p = peer_1.mk_handshake(
 949             self.pg1, True, X25519PrivateKey.generate().public_key()
 950         )
 951         self.send_and_assert_no_replies(self.pg1, [p])
 952         self.assertEqual(
 953             self.base_peer6_in_err + 1,
 954             self.statistics.get_err_counter(self.peer6_in_err),
 955         )
 956
 957         # send a valid handsake init for which we expect a response
 958         p = peer_1.mk_handshake(self.pg1, True)
 959
 960         rx = self.send_and_expect(self.pg1, [p], self.pg1)
 961
 962         peer_1.consume_response(rx[0], True)
 963
 964         # route a packet into the wg interface
 965         #  this is dropped because the peer is still not initiated
 966         p = (
 967             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 968             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
 969             / UDP(sport=555, dport=556)
 970             / Raw()
 971         )
 972         self.send_and_assert_no_replies(self.pg0, [p])
 973         self.assertEqual(
 974             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
 975         )
 976
 977         # send a data packet from the peer through the tunnel
 978         # this completes the handshake
 979         p = (
 980             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
 981             / UDP(sport=222, dport=223)
 982             / Raw()
 983         )
 984         d = peer_1.encrypt_transport(p)
 985         p = peer_1.mk_tunnel_header(self.pg1, True) / (
 986             Wireguard(message_type=4, reserved_zero=0)
 987             / WireguardTransport(
 988                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
 989             )
 990         )
 991         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
 992
 993         for rx in rxs:
 994             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 995             self.assertEqual(rx[IPv6].hlim, 19)
 996
 997         # send a packets that are routed into the tunnel
 998         p = (
 999             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1000             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1001             / UDP(sport=555, dport=556)
1002             / Raw(b"\x00" * 80)
1003         )
1004
1005         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1006
1007         for rx in rxs:
1008             rx = IPv6(peer_1.decrypt_transport(rx, True))
1009
1010             # chech the oringial packet is present
1011             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1012             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1013
1014         # send packets into the tunnel, expect to receive them on
1015         # the other side
1016         p = [
1017             (
1018                 peer_1.mk_tunnel_header(self.pg1, True)
1019                 / Wireguard(message_type=4, reserved_zero=0)
1020                 / WireguardTransport(
1021                     receiver_index=peer_1.sender,
1022                     counter=ii + 1,
1023                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1024                         (
1025                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1026                             / UDP(sport=222, dport=223)
1027                             / Raw()
1028                         )
1029                     ),
1030                 )
1031             )
1032             for ii in range(255)
1033         ]
1034
1035         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1036
1037         for rx in rxs:
1038             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1039             self.assertEqual(rx[IPv6].hlim, 19)
1040
1041         r1.remove_vpp_config()
1042         r2.remove_vpp_config()
1043         peer_1.remove_vpp_config()
1044         wg0.remove_vpp_config()
1045
1046     def test_wg_peer_v6o4(self):
1047         """Test v6o4"""
1048
1049         port = 12353
1050
1051         # Create interfaces
1052         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1053         wg0.admin_up()
1054         wg0.config_ip6()
1055
1056         peer_1 = VppWgPeer(
1057             self, wg0, self.pg1.remote_ip4, port + 1, ["1::3:0/112"]
1058         ).add_vpp_config(True)
1059         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1060
1061         r1 = VppIpRoute(
1062             self, "1::3:0", 112, [VppRoutePath("1::3:1", wg0.sw_if_index)]
1063         ).add_vpp_config()
1064
1065         # route a packet into the wg interface
1066         #  use the allowed-ip prefix
1067         #  this is dropped because the peer is not initiated
1068         p = (
1069             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1070             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1071             / UDP(sport=555, dport=556)
1072             / Raw()
1073         )
1074         self.send_and_assert_no_replies(self.pg0, [p])
1075         self.assertEqual(
1076             self.base_kp6_err + 1, self.statistics.get_err_counter(self.kp6_error)
1077         )
1078
1079         # send a handsake from the peer with an invalid MAC
1080         p = peer_1.mk_handshake(self.pg1)
1081         p[WireguardInitiation].mac1 = b"foobar"
1082         self.send_and_assert_no_replies(self.pg1, [p])
1083
1084         self.assertEqual(
1085             self.base_mac4_err + 1, self.statistics.get_err_counter(self.mac4_error)
1086         )
1087
1088         # send a handsake from the peer but signed by the wrong key.
1089         p = peer_1.mk_handshake(
1090             self.pg1, False, X25519PrivateKey.generate().public_key()
1091         )
1092         self.send_and_assert_no_replies(self.pg1, [p])
1093         self.assertEqual(
1094             self.base_peer4_in_err + 1,
1095             self.statistics.get_err_counter(self.peer4_in_err),
1096         )
1097
1098         # send a valid handsake init for which we expect a response
1099         p = peer_1.mk_handshake(self.pg1)
1100
1101         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1102
1103         peer_1.consume_response(rx[0])
1104
1105         # route a packet into the wg interface
1106         #  this is dropped because the peer is still not initiated
1107         p = (
1108             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1109             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1110             / UDP(sport=555, dport=556)
1111             / Raw()
1112         )
1113         self.send_and_assert_no_replies(self.pg0, [p])
1114         self.assertEqual(
1115             self.base_kp6_err + 2, self.statistics.get_err_counter(self.kp6_error)
1116         )
1117
1118         # send a data packet from the peer through the tunnel
1119         # this completes the handshake
1120         p = (
1121             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1122             / UDP(sport=222, dport=223)
1123             / Raw()
1124         )
1125         d = peer_1.encrypt_transport(p)
1126         p = peer_1.mk_tunnel_header(self.pg1) / (
1127             Wireguard(message_type=4, reserved_zero=0)
1128             / WireguardTransport(
1129                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1130             )
1131         )
1132         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1133
1134         for rx in rxs:
1135             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1136             self.assertEqual(rx[IPv6].hlim, 19)
1137
1138         # send a packets that are routed into the tunnel
1139         p = (
1140             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1141             / IPv6(src=self.pg0.remote_ip6, dst="1::3:2")
1142             / UDP(sport=555, dport=556)
1143             / Raw(b"\x00" * 80)
1144         )
1145
1146         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1147
1148         for rx in rxs:
1149             rx = IPv6(peer_1.decrypt_transport(rx))
1150
1151             # chech the oringial packet is present
1152             self.assertEqual(rx[IPv6].dst, p[IPv6].dst)
1153             self.assertEqual(rx[IPv6].hlim, p[IPv6].hlim - 1)
1154
1155         # send packets into the tunnel, expect to receive them on
1156         # the other side
1157         p = [
1158             (
1159                 peer_1.mk_tunnel_header(self.pg1)
1160                 / Wireguard(message_type=4, reserved_zero=0)
1161                 / WireguardTransport(
1162                     receiver_index=peer_1.sender,
1163                     counter=ii + 1,
1164                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1165                         (
1166                             IPv6(src="1::3:1", dst=self.pg0.remote_ip6, hlim=20)
1167                             / UDP(sport=222, dport=223)
1168                             / Raw()
1169                         )
1170                     ),
1171                 )
1172             )
1173             for ii in range(255)
1174         ]
1175
1176         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1177
1178         for rx in rxs:
1179             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1180             self.assertEqual(rx[IPv6].hlim, 19)
1181
1182         r1.remove_vpp_config()
1183         peer_1.remove_vpp_config()
1184         wg0.remove_vpp_config()
1185
1186     def test_wg_peer_v4o6(self):
1187         """Test v4o6"""
1188
1189         port = 12363
1190
1191         # Create interfaces
1192         wg0 = VppWgInterface(self, self.pg1.local_ip6, port).add_vpp_config()
1193         wg0.admin_up()
1194         wg0.config_ip4()
1195
1196         peer_1 = VppWgPeer(
1197             self, wg0, self.pg1.remote_ip6, port + 1, ["10.11.3.0/24"]
1198         ).add_vpp_config()
1199         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1200
1201         r1 = VppIpRoute(
1202             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1203         ).add_vpp_config()
1204
1205         # route a packet into the wg interface
1206         #  use the allowed-ip prefix
1207         #  this is dropped because the peer is not initiated
1208         p = (
1209             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1210             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1211             / UDP(sport=555, dport=556)
1212             / Raw()
1213         )
1214         self.send_and_assert_no_replies(self.pg0, [p])
1215         self.assertEqual(
1216             self.base_kp4_err + 1, self.statistics.get_err_counter(self.kp4_error)
1217         )
1218
1219         # send a handsake from the peer with an invalid MAC
1220         p = peer_1.mk_handshake(self.pg1, True)
1221         p[WireguardInitiation].mac1 = b"foobar"
1222         self.send_and_assert_no_replies(self.pg1, [p])
1223         self.assertEqual(
1224             self.base_mac6_err + 1, self.statistics.get_err_counter(self.mac6_error)
1225         )
1226
1227         # send a handsake from the peer but signed by the wrong key.
1228         p = peer_1.mk_handshake(
1229             self.pg1, True, X25519PrivateKey.generate().public_key()
1230         )
1231         self.send_and_assert_no_replies(self.pg1, [p])
1232         self.assertEqual(
1233             self.base_peer6_in_err + 1,
1234             self.statistics.get_err_counter(self.peer6_in_err),
1235         )
1236
1237         # send a valid handsake init for which we expect a response
1238         p = peer_1.mk_handshake(self.pg1, True)
1239
1240         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1241
1242         peer_1.consume_response(rx[0], True)
1243
1244         # route a packet into the wg interface
1245         #  this is dropped because the peer is still not initiated
1246         p = (
1247             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1248             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1249             / UDP(sport=555, dport=556)
1250             / Raw()
1251         )
1252         self.send_and_assert_no_replies(self.pg0, [p])
1253         self.assertEqual(
1254             self.base_kp4_err + 2, self.statistics.get_err_counter(self.kp4_error)
1255         )
1256
1257         # send a data packet from the peer through the tunnel
1258         # this completes the handshake
1259         p = (
1260             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1261             / UDP(sport=222, dport=223)
1262             / Raw()
1263         )
1264         d = peer_1.encrypt_transport(p)
1265         p = peer_1.mk_tunnel_header(self.pg1, True) / (
1266             Wireguard(message_type=4, reserved_zero=0)
1267             / WireguardTransport(
1268                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1269             )
1270         )
1271         rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1272
1273         for rx in rxs:
1274             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1275             self.assertEqual(rx[IP].ttl, 19)
1276
1277         # send a packets that are routed into the tunnel
1278         p = (
1279             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1280             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1281             / UDP(sport=555, dport=556)
1282             / Raw(b"\x00" * 80)
1283         )
1284
1285         rxs = self.send_and_expect(self.pg0, p * 255, self.pg1)
1286
1287         for rx in rxs:
1288             rx = IP(peer_1.decrypt_transport(rx, True))
1289
1290             # chech the oringial packet is present
1291             self.assertEqual(rx[IP].dst, p[IP].dst)
1292             self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1293
1294         # send packets into the tunnel, expect to receive them on
1295         # the other side
1296         p = [
1297             (
1298                 peer_1.mk_tunnel_header(self.pg1, True)
1299                 / Wireguard(message_type=4, reserved_zero=0)
1300                 / WireguardTransport(
1301                     receiver_index=peer_1.sender,
1302                     counter=ii + 1,
1303                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1304                         (
1305                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1306                             / UDP(sport=222, dport=223)
1307                             / Raw()
1308                         )
1309                     ),
1310                 )
1311             )
1312             for ii in range(255)
1313         ]
1314
1315         rxs = self.send_and_expect(self.pg1, p, self.pg0)
1316
1317         for rx in rxs:
1318             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1319             self.assertEqual(rx[IP].ttl, 19)
1320
1321         r1.remove_vpp_config()
1322         peer_1.remove_vpp_config()
1323         wg0.remove_vpp_config()
1324
1325     def test_wg_multi_peer(self):
1326         """multiple peer setup"""
1327         port = 12373
1328
1329         # Create interfaces
1330         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1331         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
1332         wg0.admin_up()
1333         wg1.admin_up()
1334
1335         # Check peer counter
1336         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
1337
1338         self.pg_enable_capture(self.pg_interfaces)
1339         self.pg_start()
1340
1341         # Create many peers on sencond interface
1342         NUM_PEERS = 16
1343         self.pg2.generate_remote_hosts(NUM_PEERS)
1344         self.pg2.configure_ipv4_neighbors()
1345         self.pg1.generate_remote_hosts(NUM_PEERS)
1346         self.pg1.configure_ipv4_neighbors()
1347
1348         peers_1 = []
1349         peers_2 = []
1350         routes_1 = []
1351         routes_2 = []
1352         for i in range(NUM_PEERS):
1353             peers_1.append(
1354                 VppWgPeer(
1355                     self,
1356                     wg0,
1357                     self.pg1.remote_hosts[i].ip4,
1358                     port + 1 + i,
1359                     ["10.0.%d.4/32" % i],
1360                 ).add_vpp_config()
1361             )
1362             routes_1.append(
1363                 VppIpRoute(
1364                     self,
1365                     "10.0.%d.4" % i,
1366                     32,
1367                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
1368                 ).add_vpp_config()
1369             )
1370
1371             peers_2.append(
1372                 VppWgPeer(
1373                     self,
1374                     wg1,
1375                     self.pg2.remote_hosts[i].ip4,
1376                     port + 100 + i,
1377                     ["10.100.%d.4/32" % i],
1378                 ).add_vpp_config()
1379             )
1380             routes_2.append(
1381                 VppIpRoute(
1382                     self,
1383                     "10.100.%d.4" % i,
1384                     32,
1385                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
1386                 ).add_vpp_config()
1387             )
1388
1389         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
1390
1391         self.logger.info(self.vapi.cli("show wireguard peer"))
1392         self.logger.info(self.vapi.cli("show wireguard interface"))
1393         self.logger.info(self.vapi.cli("show adj 37"))
1394         self.logger.info(self.vapi.cli("sh ip fib 172.16.3.17"))
1395         self.logger.info(self.vapi.cli("sh ip fib 10.11.3.0"))
1396
1397         # remove routes
1398         for r in routes_1:
1399             r.remove_vpp_config()
1400         for r in routes_2:
1401             r.remove_vpp_config()
1402
1403         # remove peers
1404         for p in peers_1:
1405             self.assertTrue(p.query_vpp_config())
1406             p.remove_vpp_config()
1407         for p in peers_2:
1408             self.assertTrue(p.query_vpp_config())
1409             p.remove_vpp_config()
1410
1411         wg0.remove_vpp_config()
1412         wg1.remove_vpp_config()
1413
1414     def test_wg_multi_interface(self):
1415         """Multi-tunnel on the same port"""
1416         port = 12500
1417
1418         # Create many wireguard interfaces
1419         NUM_IFS = 4
1420         self.pg1.generate_remote_hosts(NUM_IFS)
1421         self.pg1.configure_ipv4_neighbors()
1422         self.pg0.generate_remote_hosts(NUM_IFS)
1423         self.pg0.configure_ipv4_neighbors()
1424
1425         # Create interfaces with a peer on each
1426         peers = []
1427         routes = []
1428         wg_ifs = []
1429         for i in range(NUM_IFS):
1430             # Use the same port for each interface
1431             wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1432             wg0.admin_up()
1433             wg0.config_ip4()
1434             wg_ifs.append(wg0)
1435             peers.append(
1436                 VppWgPeer(
1437                     self,
1438                     wg0,
1439                     self.pg1.remote_hosts[i].ip4,
1440                     port + 1 + i,
1441                     ["10.0.%d.0/24" % i],
1442                 ).add_vpp_config()
1443             )
1444
1445             routes.append(
1446                 VppIpRoute(
1447                     self,
1448                     "10.0.%d.0" % i,
1449                     24,
1450                     [VppRoutePath("10.0.%d.4" % i, wg0.sw_if_index)],
1451                 ).add_vpp_config()
1452             )
1453
1454         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_IFS)
1455
1456         for i in range(NUM_IFS):
1457             # send a valid handsake init for which we expect a response
1458             p = peers[i].mk_handshake(self.pg1)
1459             rx = self.send_and_expect(self.pg1, [p], self.pg1)
1460             peers[i].consume_response(rx[0])
1461
1462             # send a data packet from the peer through the tunnel
1463             # this completes the handshake
1464             p = (
1465                 IP(src="10.0.%d.4" % i, dst=self.pg0.remote_hosts[i].ip4, ttl=20)
1466                 / UDP(sport=222, dport=223)
1467                 / Raw()
1468             )
1469             d = peers[i].encrypt_transport(p)
1470             p = peers[i].mk_tunnel_header(self.pg1) / (
1471                 Wireguard(message_type=4, reserved_zero=0)
1472                 / WireguardTransport(
1473                     receiver_index=peers[i].sender,
1474                     counter=0,
1475                     encrypted_encapsulated_packet=d,
1476                 )
1477             )
1478             rxs = self.send_and_expect(self.pg1, [p], self.pg0)
1479             for rx in rxs:
1480                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
1481                 self.assertEqual(rx[IP].ttl, 19)
1482
1483         # send a packets that are routed into the tunnel
1484         for i in range(NUM_IFS):
1485             p = (
1486                 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1487                 / IP(src=self.pg0.remote_hosts[i].ip4, dst="10.0.%d.4" % i)
1488                 / UDP(sport=555, dport=556)
1489                 / Raw(b"\x00" * 80)
1490             )
1491
1492             rxs = self.send_and_expect(self.pg0, p * 64, self.pg1)
1493
1494             for rx in rxs:
1495                 rx = IP(peers[i].decrypt_transport(rx))
1496
1497                 # check the oringial packet is present
1498                 self.assertEqual(rx[IP].dst, p[IP].dst)
1499                 self.assertEqual(rx[IP].ttl, p[IP].ttl - 1)
1500
1501         # send packets into the tunnel
1502         for i in range(NUM_IFS):
1503             p = [
1504                 (
1505                     peers[i].mk_tunnel_header(self.pg1)
1506                     / Wireguard(message_type=4, reserved_zero=0)
1507                     / WireguardTransport(
1508                         receiver_index=peers[i].sender,
1509                         counter=ii + 1,
1510                         encrypted_encapsulated_packet=peers[i].encrypt_transport(
1511                             (
1512                                 IP(
1513                                     src="10.0.%d.4" % i,
1514                                     dst=self.pg0.remote_hosts[i].ip4,
1515                                     ttl=20,
1516                                 )
1517                                 / UDP(sport=222, dport=223)
1518                                 / Raw()
1519                             )
1520                         ),
1521                     )
1522                 )
1523                 for ii in range(64)
1524             ]
1525
1526             rxs = self.send_and_expect(self.pg1, p, self.pg0)
1527
1528             for rx in rxs:
1529                 self.assertEqual(rx[IP].dst, self.pg0.remote_hosts[i].ip4)
1530                 self.assertEqual(rx[IP].ttl, 19)
1531
1532         for r in routes:
1533             r.remove_vpp_config()
1534         for p in peers:
1535             p.remove_vpp_config()
1536         for i in wg_ifs:
1537             i.remove_vpp_config()
1538
1539     def test_wg_event(self):
1540         """Test events"""
1541         port = 12600
1542         ESTABLISHED_FLAG = (
1543             VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_ESTABLISHED
1544         )
1545         DEAD_FLAG = VppEnum.vl_api_wireguard_peer_flags_t.WIREGUARD_PEER_STATUS_DEAD
1546
1547         # Create interfaces
1548         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1549         wg1 = VppWgInterface(self, self.pg2.local_ip4, port + 1).add_vpp_config()
1550         wg0.admin_up()
1551         wg1.admin_up()
1552
1553         # Check peer counter
1554         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 0)
1555
1556         self.pg_enable_capture(self.pg_interfaces)
1557         self.pg_start()
1558
1559         # Create peers
1560         NUM_PEERS = 2
1561         self.pg2.generate_remote_hosts(NUM_PEERS)
1562         self.pg2.configure_ipv4_neighbors()
1563         self.pg1.generate_remote_hosts(NUM_PEERS)
1564         self.pg1.configure_ipv4_neighbors()
1565
1566         peers_0 = []
1567         peers_1 = []
1568         routes_0 = []
1569         routes_1 = []
1570         for i in range(NUM_PEERS):
1571             peers_0.append(
1572                 VppWgPeer(
1573                     self,
1574                     wg0,
1575                     self.pg1.remote_hosts[i].ip4,
1576                     port + 1 + i,
1577                     ["10.0.%d.4/32" % i],
1578                 ).add_vpp_config()
1579             )
1580             routes_0.append(
1581                 VppIpRoute(
1582                     self,
1583                     "10.0.%d.4" % i,
1584                     32,
1585                     [VppRoutePath(self.pg1.remote_hosts[i].ip4, wg0.sw_if_index)],
1586                 ).add_vpp_config()
1587             )
1588
1589             peers_1.append(
1590                 VppWgPeer(
1591                     self,
1592                     wg1,
1593                     self.pg2.remote_hosts[i].ip4,
1594                     port + 100 + i,
1595                     ["10.100.%d.4/32" % i],
1596                 ).add_vpp_config()
1597             )
1598             routes_1.append(
1599                 VppIpRoute(
1600                     self,
1601                     "10.100.%d.4" % i,
1602                     32,
1603                     [VppRoutePath(self.pg2.remote_hosts[i].ip4, wg1.sw_if_index)],
1604                 ).add_vpp_config()
1605             )
1606
1607         self.assertEqual(len(self.vapi.wireguard_peers_dump()), NUM_PEERS * 2)
1608
1609         # Want events from the first perr of wg0
1610         # and from all wg1 peers
1611         peers_0[0].want_events()
1612         wg1.want_events()
1613
1614         for i in range(NUM_PEERS):
1615             # send a valid handsake init for which we expect a response
1616             p = peers_0[i].mk_handshake(self.pg1)
1617             rx = self.send_and_expect(self.pg1, [p], self.pg1)
1618             peers_0[i].consume_response(rx[0])
1619             if i == 0:
1620                 peers_0[0].wait_event(ESTABLISHED_FLAG)
1621
1622             p = peers_1[i].mk_handshake(self.pg2)
1623             rx = self.send_and_expect(self.pg2, [p], self.pg2)
1624             peers_1[i].consume_response(rx[0])
1625
1626         wg1.wait_events(ESTABLISHED_FLAG, [peers_1[0].index, peers_1[1].index])
1627
1628         # remove routes
1629         for r in routes_0:
1630             r.remove_vpp_config()
1631         for r in routes_1:
1632             r.remove_vpp_config()
1633
1634         # remove peers
1635         for i in range(NUM_PEERS):
1636             self.assertTrue(peers_0[i].query_vpp_config())
1637             peers_0[i].remove_vpp_config()
1638             if i == 0:
1639                 peers_0[i].wait_event(0)
1640                 peers_0[i].wait_event(DEAD_FLAG)
1641         for p in peers_1:
1642             self.assertTrue(p.query_vpp_config())
1643             p.remove_vpp_config()
1644             p.wait_event(0)
1645             p.wait_event(DEAD_FLAG)
1646
1647         wg0.remove_vpp_config()
1648         wg1.remove_vpp_config()
1649
1650
1651 class WireguardHandoffTests(TestWg):
1652     """Wireguard Tests in multi worker setup"""
1653
1654     vpp_worker_count = 2
1655
1656     def test_wg_peer_init(self):
1657         """Handoff"""
1658
1659         port = 12383
1660
1661         # Create interfaces
1662         wg0 = VppWgInterface(self, self.pg1.local_ip4, port).add_vpp_config()
1663         wg0.admin_up()
1664         wg0.config_ip4()
1665
1666         peer_1 = VppWgPeer(
1667             self, wg0, self.pg1.remote_ip4, port + 1, ["10.11.2.0/24", "10.11.3.0/24"]
1668         ).add_vpp_config()
1669         self.assertEqual(len(self.vapi.wireguard_peers_dump()), 1)
1670
1671         r1 = VppIpRoute(
1672             self, "10.11.3.0", 24, [VppRoutePath("10.11.3.1", wg0.sw_if_index)]
1673         ).add_vpp_config()
1674
1675         # send a valid handsake init for which we expect a response
1676         p = peer_1.mk_handshake(self.pg1)
1677
1678         rx = self.send_and_expect(self.pg1, [p], self.pg1)
1679
1680         peer_1.consume_response(rx[0])
1681
1682         # send a data packet from the peer through the tunnel
1683         # this completes the handshake and pins the peer to worker 0
1684         p = (
1685             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1686             / UDP(sport=222, dport=223)
1687             / Raw()
1688         )
1689         d = peer_1.encrypt_transport(p)
1690         p = peer_1.mk_tunnel_header(self.pg1) / (
1691             Wireguard(message_type=4, reserved_zero=0)
1692             / WireguardTransport(
1693                 receiver_index=peer_1.sender, counter=0, encrypted_encapsulated_packet=d
1694             )
1695         )
1696         rxs = self.send_and_expect(self.pg1, [p], self.pg0, worker=0)
1697
1698         for rx in rxs:
1699             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1700             self.assertEqual(rx[IP].ttl, 19)
1701
1702         # send a packets that are routed into the tunnel
1703         # and pins the peer tp worker 1
1704         pe = (
1705             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1706             / IP(src=self.pg0.remote_ip4, dst="10.11.3.2")
1707             / UDP(sport=555, dport=556)
1708             / Raw(b"\x00" * 80)
1709         )
1710         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=1)
1711         peer_1.validate_encapped(rxs, pe)
1712
1713         # send packets into the tunnel, from the other worker
1714         p = [
1715             (
1716                 peer_1.mk_tunnel_header(self.pg1)
1717                 / Wireguard(message_type=4, reserved_zero=0)
1718                 / WireguardTransport(
1719                     receiver_index=peer_1.sender,
1720                     counter=ii + 1,
1721                     encrypted_encapsulated_packet=peer_1.encrypt_transport(
1722                         (
1723                             IP(src="10.11.3.1", dst=self.pg0.remote_ip4, ttl=20)
1724                             / UDP(sport=222, dport=223)
1725                             / Raw()
1726                         )
1727                     ),
1728                 )
1729             )
1730             for ii in range(255)
1731         ]
1732
1733         rxs = self.send_and_expect(self.pg1, p, self.pg0, worker=1)
1734
1735         for rx in rxs:
1736             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1737             self.assertEqual(rx[IP].ttl, 19)
1738
1739         # send a packets that are routed into the tunnel
1740         # from owrker 0
1741         rxs = self.send_and_expect(self.pg0, pe * 255, self.pg1, worker=0)
1742
1743         peer_1.validate_encapped(rxs, pe)
1744
1745         r1.remove_vpp_config()
1746         peer_1.remove_vpp_config()
1747         wg0.remove_vpp_config()
1748
1749     @unittest.skip("test disabled")
1750     def test_wg_multi_interface(self):
1751         """Multi-tunnel on the same port"""