wireguard: T5413: Blocked adding the peer with the router's public key

python/vyos/validate.py

@@ -1,4 +1,4 @@
-# Copyright 2018-2021 VyOS maintainers and contributors <[email protected]>
+# Copyright 2018-2023 VyOS maintainers and contributors <[email protected]>
 #
 # This library is free software; you can redistribute it and/or
 # modify it under the terms of the GNU Lesser General Public
@@ -297,3 +297,20 @@ def has_vrf_configured(conf, intf):
 
     conf.set_level(old_level)
     return ret
+
+def is_wireguard_key_pair(private_key: str, public_key:str) -> bool:
+    """
+     Checks if public/private keys are keypair
+    :param private_key: Wireguard private key
+    :type private_key: str
+    :param public_key: Wireguard public key
+    :type public_key: str
+    :return: If public/private keys are keypair returns True else False
+    :rtype: bool
+    """
+    from vyos.util import cmd
+    gen_public_key = cmd('wg pubkey', input=private_key)
+    if gen_public_key == public_key:
+        return True
+    else:
+        return False

smoketest/scripts/cli/test_interfaces_wireguard.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 #
-# Copyright (C) 2020-2021 VyOS maintainers and contributors
+# Copyright (C) 2020-2023 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
@@ -91,5 +91,39 @@ def test_wireguard_add_remove_peer(self):
         self.cli_delete(base_path + [interface, 'peer', 'PEER01'])
         self.cli_commit()
 
+    def test_wireguard_same_public_key(self):
+        # T5413: Test prevention of using peer own public key.
+        interface = 'wg0'
+        port = '12345'
+        pubkey_ok = 'ebFx/1G0ti8tvuZd94sEIosAZZIznX+dBAKG/8DFm0I='
+
+        public_key_path = f'/config/auth/wireguard/default/public.key'
+        with open(public_key_path, 'r') as file:
+            pubkey_fail = file.read().rstrip()
+
+        self.cli_set(base_path + [interface, 'address', '172.16.0.1/24'])
+        self.cli_set(base_path + [interface, 'private-key', 'default'])
+
+        self.cli_set(
+            base_path + [interface, 'peer', 'PEER01', 'pubkey', 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', 'pubkey', pubkey_ok])
+
+        # Commit peers
+        self.cli_commit()
+
+        self.assertTrue(os.path.isdir(f'/sys/class/net/{interface}'))
+
+
 if __name__ == '__main__':
     unittest.main(verbosity=2)

src/conf_mode/interfaces-wireguard.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 #
-# Copyright (C) 2018-2020 VyOS maintainers and contributors
+# Copyright (C) 2018-2023 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
@@ -28,6 +28,7 @@
 from vyos.configverify import verify_mtu_ipv6
 from vyos.configverify import verify_bond_bridge_member
 from vyos.ifconfig import WireGuardIf
+from vyos.validate import is_wireguard_key_pair
 from vyos.util import check_kmod
 from vyos import ConfigError
 from vyos import airbag
@@ -94,6 +95,11 @@ def verify(wireguard):
             raise ConfigError('Both Wireguard port and address must be defined '
                               f'for peer "{tmp}" if either one of them is set!')
 
+        if 'disable' not in peer:
+            with open(wireguard['private_key'], 'r') as file:
+                private_key = file.read().rstrip()
+                if is_wireguard_key_pair(private_key, peer['pubkey']):
+                    raise ConfigError(f'Peer "{tmp}" has the same public key as the interface "{wireguard["ifname"]}"')
 def apply(wireguard):
     tmp = WireGuardIf(wireguard['ifname'])
     if 'deleted' in wireguard:

src/migration-scripts/interfaces/22-to-23

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2023 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
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import sys
+from vyos.configtree import ConfigTree
+from vyos.validate import is_wireguard_key_pair
+
+if __name__ == '__main__':
+    if len(sys.argv) < 2:
+        print("Must specify file name!")
+        sys.exit(1)
+
+    file_name = sys.argv[1]
+
+    with open(file_name, 'r') as f:
+        config_file = f.read()
+
+    config = ConfigTree(config_file)
+    base = ['interfaces', 'wireguard']
+    if not config.exists(base):
+        # Nothing to do
+        sys.exit(0)
+    for interface in config.list_nodes(base):
+        private_key_name = config.return_value(
+            base + [interface, 'private-key'])
+        private_key_path = f'/config/auth/wireguard/{private_key_name}/private.key'
+        with open(private_key_path, 'r') as file:
+            private_key = file.read().rstrip()
+        interface_base = base + [interface]
+        if config.exists(interface_base + ['peer']):
+            for peer in config.list_nodes(interface_base + ['peer']):
+                peer_base = interface_base + ['peer', peer]
+                peer_public_key = config.return_value(peer_base + ['pubkey'])
+                if config.exists(peer_base + ['pubkey']):
+                    if not config.exists(peer_base + ['disable']) \
+                            and is_wireguard_key_pair(private_key,
+                                                      peer_public_key):
+                        config.set(peer_base + ['disable'])
+
+    try:
+        with open(file_name, 'w') as f:
+            f.write(config.to_string())
+    except OSError as e:
+        print("Failed to save the modified config: {}".format(e))
+        sys.exit(1)