Merge pull request #980 from zdc/T3763-sagitta

wireguard: T3763: Added check for listening port availability

2 files changed, 41 insertions, 0 deletions

diff --git a/python/vyos/util.py b/python/vyos/util.py
index 8af46a6ee..93a2f6640 100644
--- a/python/vyos/util.py
+++ b/python/vyos/util.py
@@ -819,3 +819,36 @@ def is_systemd_service_running(service):
     Copied from: https://unix.stackexchange.com/a/435317 """
     tmp = cmd(f'systemctl show --value -p SubState {service}')
     return bool((tmp == 'running'))
+
+def check_port_availability(ipaddress, port, protocol):
+    """
+    Check if port is available and not used by any service
+    Return False if a port is busy or IP address does not exists
+    Should be used carefully for services that can start listening
+    dynamically, because IP address may be dynamic too
+    """
+    from socketserver import TCPServer, UDPServer
+    from ipaddress import ip_address
+
+    # verify arguments
+    try:
+        ipaddress = ip_address(ipaddress).compressed
+    except:
+        raise ValueError(f'The {ipaddress} is not a valid IPv4 or IPv6 address')
+    if port not in range(1, 65536):
+        raise ValueError(f'The port number {port} is not in the 1-65535 range')
+    if protocol not in ['tcp', 'udp']:
+        raise ValueError(
+            f'The protocol {protocol} is not supported. Only tcp and udp are allowed'
+        )
+
+    # check port availability
+    try:
+        if protocol == 'tcp':
+            server = TCPServer((ipaddress, port), None, bind_and_activate=True)
+        if protocol == 'udp':
+            server = UDPServer((ipaddress, port), None, bind_and_activate=True)
+        server.server_close()
+        return True
+    except:
+        return False
diff --git a/src/conf_mode/interfaces-wireguard.py b/src/conf_mode/interfaces-wireguard.py
index 4c566a5ad..68181465e 100755
--- a/src/conf_mode/interfaces-wireguard.py
+++ b/src/conf_mode/interfaces-wireguard.py
@@ -30,6 +30,7 @@ from vyos.configverify import verify_bridge_delete
 from vyos.configverify import verify_mtu_ipv6
 from vyos.ifconfig import WireGuardIf
 from vyos.util import check_kmod
+from vyos.util import check_port_availability
 from vyos import ConfigError
 from vyos import airbag
 airbag.enable()
@@ -73,6 +74,13 @@ def verify(wireguard):
     if 'peer' not in wireguard:
         raise ConfigError('At least one Wireguard peer is required!')
 
+    listen_port = int(wireguard['port'])
+    if 'port' in wireguard and check_port_availability('0.0.0.0', listen_port,
+                                                       'udp') is not True:
+        raise ConfigError(
+            f'The UDP port {listen_port} is busy or unavailable and cannot be used for the interface'
+        )
+
     # run checks on individual configured WireGuard peer
     for tmp in wireguard['peer']:
         peer = wireguard['peer'][tmp]