Merge pull request #4200 from sskaje/T4930-1

T4930: Allow WireGuard peers via DNS hostname

1 files changed, 72 insertions, 36 deletions

diff --git a/smoketest/scripts/cli/test_interfaces_wireguard.py b/smoketest/scripts/cli/test_interfaces_wireguard.py
index 4b994a659..f8cd18cf2 100755
--- a/smoketest/scripts/cli/test_interfaces_wireguard.py
+++ b/smoketest/scripts/cli/test_interfaces_wireguard.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 #
-# Copyright (C) 2020-2023 VyOS maintainers and contributors
+# Copyright (C) 2020-2025 VyOS maintainers and contributors
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 2 or later as
@@ -17,27 +17,33 @@
 import os
 import unittest
 
-from base_vyostest_shim import VyOSUnitTestSHIM
+from base_interfaces_test import BasicInterfaceTest
 from vyos.configsession import ConfigSessionError
 from vyos.utils.file import read_file
 from vyos.utils.process import cmd
+from vyos.utils.process import is_systemd_service_running
 
 base_path = ['interfaces', 'wireguard']
-
-class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase):
+domain_resolver = 'vyos-domain-resolver.service'
+class WireGuardInterfaceTest(BasicInterfaceTest.TestCase):
     @classmethod
     def setUpClass(cls):
-        super(WireGuardInterfaceTest, cls).setUpClass()
-
-        cls._test_addr = ['192.0.2.1/26', '192.0.2.255/31', '192.0.2.64/32',
-                          '2001:db8:1::ffff/64', '2001:db8:101::1/112']
-        cls._interfaces = ['wg0', 'wg1']
+        cls._base_path = ['interfaces', 'wireguard']
+        cls._options = {
+            'wg0': ['private-key wBbGJJXYllwDcw63AFjiIR6ZlsvqvAf3eDwog64Dp0Q=',
+                    'peer RED public-key 6hkkfxN4VUQLu36NLZr47I7ST/FkQl2clPWr+9a6ZH8=',
+                    'peer RED allowed-ips 169.254.0.0/16',
+                    'port 5678'],
+            'wg1': ['private-key QFwnBHlHYspehvpklBKb7cikM+QMkEy2p6gfsg06S08=',
+                    'peer BLUE public-key hRJLmP8SVU9/MLmPmYmpOa+RTB4F/zhDqA+/QDuW1Hg=',
+                    'peer BLUE allowed-ips 169.254.0.0/16',
+                    'port 4567'],
+        }
+        cls._interfaces = list(cls._options)
 
-    def tearDown(self):
-        self.cli_delete(base_path)
-        self.cli_commit()
+        super(WireGuardInterfaceTest, cls).setUpClass()
 
-    def test_01_wireguard_peer(self):
+    def test_wireguard_peer(self):
         # Create WireGuard interfaces with associated peers
         for intf in self._interfaces:
             peer = 'foo-' + intf
@@ -64,7 +70,7 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase):
 
             self.assertTrue(os.path.isdir(f'/sys/class/net/{intf}'))
 
-    def test_02_wireguard_add_remove_peer(self):
+    def test_wireguard_add_remove_peer(self):
         # T2939: Create WireGuard interfaces with associated peers.
         # Remove one of the configured peers.
         # T4774: Test prevention of duplicate peer public keys
@@ -102,7 +108,7 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase):
         self.cli_delete(base_path + [interface, 'peer', 'PEER01'])
         self.cli_commit()
 
-    def test_03_wireguard_same_public_key(self):
+    def test_wireguard_same_public_key(self):
         # T5413: Test prevention of equality interface public key and peer's
         #        public key
         interface = 'wg0'
@@ -115,45 +121,40 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase):
         self.cli_set(base_path + [interface, 'private-key', privkey])
 
         self.cli_set(base_path + [interface, 'peer', 'PEER01', 'public-key', pubkey_fail])
-        self.cli_set(base_path + [interface, 'peer', 'PEER01', 'port', port])
         self.cli_set(base_path + [interface, 'peer', 'PEER01', 'allowed-ips', '10.205.212.10/32'])
         self.cli_set(base_path + [interface, 'peer', 'PEER01', 'address', '192.0.2.1'])
 
         # The same pubkey as the interface wg0
         with self.assertRaises(ConfigSessionError):
             self.cli_commit()
-
         self.cli_set(base_path + [interface, 'peer', 'PEER01', 'public-key', pubkey_ok])
 
+        # If address is defined for a peer, so must be the peer port
+        with self.assertRaises(ConfigSessionError):
+            self.cli_commit()
+        self.cli_set(base_path + [interface, 'peer', 'PEER01', 'port', port])
+
         # Commit peers
         self.cli_commit()
 
         self.assertTrue(os.path.isdir(f'/sys/class/net/{interface}'))
 
-    def test_04_wireguard_threaded(self):
+    def test_wireguard_threaded(self):
         # T5409: Test adding threaded option on interface.
-        #        Test prevention for adding threaded
-        #        if no enabled peer is configured.
-        interface = 'wg0'
-        port = '12345'
-        privkey = 'OOjcXGfgQlAuM6q8Z9aAYduCua7pxf7UKYvIqoUPoGQ='
-        pubkey = 'ebFx/1G0ti8tvuZd94sEIosAZZIznX+dBAKG/8DFm0I='
+        for intf in self._interfaces:
+            for option in self._options.get(intf, []):
+                self.cli_set(self._base_path + [intf] + option.split())
 
-        self.cli_set(base_path + [interface, 'address', '172.16.0.1/24'])
-        self.cli_set(base_path + [interface, 'private-key', privkey])
-
-        self.cli_set(base_path + [interface, 'peer', 'PEER01', 'port', port])
-        self.cli_set(base_path + [interface, 'peer', 'PEER01', 'public-key', pubkey])
-        self.cli_set(base_path + [interface, 'peer', 'PEER01', 'allowed-ips', '10.205.212.10/32'])
-        self.cli_set(base_path + [interface, 'peer', 'PEER01', 'address', '192.0.2.1'])
-        self.cli_set(base_path + [interface, 'per-client-thread'])
+            self.cli_set(base_path + [intf, 'per-client-thread'])
 
         # Commit peers
         self.cli_commit()
-        tmp = read_file(f'/sys/class/net/{interface}/threaded')
-        self.assertTrue(tmp, "1")
 
-    def test_05_wireguard_peer_pubkey_change(self):
+        for intf in self._interfaces:
+            tmp = read_file(f'/sys/class/net/{intf}/threaded')
+            self.assertTrue(tmp, "1")
+
+    def test_wireguard_peer_pubkey_change(self):
         # T5707 changing WireGuard CLI public key of a peer - it's not removed
 
         def get_peers(interface) -> list:
@@ -171,7 +172,6 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase):
                     peers.append(items[0])
             return peers
 
-
         interface = 'wg1337'
         port = '1337'
         privkey = 'iJi4lb2HhkLx2KSAGOjji2alKkYsJjSPkHkrcpxgEVU='
@@ -200,5 +200,41 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase):
         self.assertNotIn(pubkey_1, peers)
         self.assertIn(pubkey_2, peers)
 
+    def test_wireguard_hostname(self):
+        # T4930: Test dynamic endpoint support
+        interface = 'wg1234'
+        port = '54321'
+        privkey = 'UOWIeZKNzijhgu0bPRy2PB3gnuOBLfQax5GiYfkmU3A='
+        pubkey = '4nG5NfhHBQUq/DnwT0RjRoBCqAh3VrRHqdQgzC/xujk='
+
+        base_interface_path = base_path + [interface]
+        self.cli_set(base_interface_path + ['address', '172.16.0.1/24'])
+        self.cli_set(base_interface_path + ['private-key', privkey])
+
+        peer_base_path = base_interface_path + ['peer', 'dynamic01']
+        self.cli_set(peer_base_path + ['port', port])
+        self.cli_set(peer_base_path + ['public-key', pubkey])
+        self.cli_set(peer_base_path + ['allowed-ips', '169.254.0.0/16'])
+        self.cli_set(peer_base_path + ['address', '192.0.2.1'])
+        self.cli_set(peer_base_path + ['host-name', 'wg.vyos.net'])
+
+        # Peer address and host-name are mutually exclusive
+        with self.assertRaises(ConfigSessionError):
+            self.cli_commit()
+
+        self.cli_delete(peer_base_path + ['address'])
+
+        # Commit peers
+        self.cli_commit()
+
+        # Ensure the service is running which checks for DNS changes
+        self.assertTrue(is_systemd_service_running(domain_resolver))
+
+        self.cli_delete(base_interface_path)
+        self.cli_commit()
+
+        # Ensure the service is no longer running after WireGuard interface is deleted
+        self.assertFalse(is_systemd_service_running(domain_resolver))
+
 if __name__ == '__main__':
     unittest.main(verbosity=2)