datamodel: network: listen: advanced port configuration validation

diff --git a/manager/knot_resolver_manager/datamodel/network_schema.py b/manager/knot_resolver_manager/datamodel/network_schema.py
index 2abb1ec53bab07ed4ff7690c789f41b7cf48119f..06aaa8051563dd48c3079fcc57b3fa9b9614bf0f 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/network_schema.py
+++ b/manager/knot_resolver_manager/datamodel/network_schema.py
@@ -54,12 +54,42 @@ class ListenSchema(SchemaNode):
     _PREVIOUS_SCHEMA = Raw
 
     unix_socket: Union[None, CheckedPath, List[CheckedPath]]
-    ip_address: Union[None, IPAddressPort, IPAddressPort, List[IPAddressPort]]
+    ip_address: Union[None, IPAddressPort, List[IPAddressPort]]
     interface: Union[None, InterfacePort, List[InterfacePort]]
     port: Optional[int]
     kind: KindEnum
     freebind: bool
 
+    def _ip_address(self, origin: Raw) -> Union[None, IPAddressPort, List[IPAddressPort]]:
+        if isinstance(origin.ip_address, list):
+            port_set: Optional[bool] = None
+            for addr in origin.ip_address:
+                if origin.port and addr.port:
+                    raise ValueError("The port number is defined in two places ('port' option and '@<port>' syntax).")
+                if port_set is not None and (bool(addr.port) != port_set):
+                    raise ValueError(
+                        "The port number specified by '@<port>' syntax must or must not be used for each IP address."
+                    )
+                port_set = True if addr.port else False
+        elif isinstance(origin.ip_address, IPAddressPort) and origin.ip_address.port and origin.port:
+            raise ValueError("The port number is defined in two places ('port' option and '@<port>' syntax).")
+        return origin.ip_address
+
+    def _interface(self, origin: Raw) -> Union[None, InterfacePort, List[InterfacePort]]:
+        if isinstance(origin.interface, list):
+            port_set: Optional[bool] = None
+            for intrfc in origin.interface:
+                if origin.port and intrfc.port:
+                    raise ValueError("The port number is defined in two places ('port' option and '@<port>' syntax).")
+                if port_set is not None and (bool(intrfc.port) != port_set):
+                    raise ValueError(
+                        "The port number specified by '@<port>' syntax must or must not be used for each interface."
+                    )
+                port_set = True if intrfc.port else False
+        elif isinstance(origin.interface, InterfacePort) and origin.interface.port and origin.port:
+            raise ValueError("The port number is defined in two places ('port' option and '@<port>' syntax).")
+        return origin.interface
+
     def _port(self, origin: Raw) -> Optional[int]:
         if origin.port:
             return origin.port

diff --git a/manager/tests/unit/datamodel/templates/test_network_macros.py b/manager/tests/unit/datamodel/templates/test_network_macros.py
index 7f302bf4f0937486cfc0179d2b3740481a72c9de..54eeecdd66a56b384a151cd691cfa3af3a28b0d8 100644 (file)
--- a/manager/tests/unit/datamodel/templates/test_network_macros.py
+++ b/manager/tests/unit/datamodel/templates/test_network_macros.py
@@ -16,20 +16,20 @@ def test_network_listen():
         + "net.listen('/tmp/kresd-socket2',nil,{kind='tls',freebind=false})\n"
     )
 
-    ip = ListenSchema({"ip-address": "::1", "freebind": True})
-    assert tmpl.render(listen=ip) == "net.listen('::1',53,{kind='dns',freebind=true})"
+    ip = ListenSchema({"ip-address": "::1@55", "freebind": True})
+    assert tmpl.render(listen=ip) == "net.listen('::1',55,{kind='dns',freebind=true})"
     ip_list = ListenSchema({"ip-address": [ip.ip_address, "127.0.0.1@5353"]})
     assert (
         tmpl.render(listen=ip_list)
-        == "net.listen('::1',53,{kind='dns',freebind=false})\n"
+        == "net.listen('::1',55,{kind='dns',freebind=false})\n"
         + "net.listen('127.0.0.1',5353,{kind='dns',freebind=false})\n"
     )
 
     intrfc = ListenSchema({"interface": "eth0", "kind": "doh2"})
     assert tmpl.render(listen=intrfc) == "net.listen(net.eth0,443,{kind='doh2',freebind=false})"
-    intrfc_list = ListenSchema({"interface": [intrfc.interface, "lo@5353"], "port": 5555, "kind": "doh2"})
+    intrfc_list = ListenSchema({"interface": [intrfc.interface, "lo"], "port": 5555, "kind": "doh2"})
     assert (
         tmpl.render(listen=intrfc_list)
         == "net.listen(net.eth0,5555,{kind='doh2',freebind=false})\n"
-        + "net.listen(net.lo,5353,{kind='doh2',freebind=false})\n"
+        + "net.listen(net.lo,5555,{kind='doh2',freebind=false})\n"
     )

diff --git a/manager/tests/unit/datamodel/test_network_schema.py b/manager/tests/unit/datamodel/test_network_schema.py
index be41206f1ed3ff23d6de7c85937d510abe6af924..1b8d9f83b749c05b4fc2cb491f196deac52e13d1 100644 (file)
--- a/manager/tests/unit/datamodel/test_network_schema.py
+++ b/manager/tests/unit/datamodel/test_network_schema.py
@@ -33,12 +33,50 @@ def test_listen_kind_port_defaults():
     assert doh2.port == 443
 
 
+def test_listen_unix_socket():
+    assert ListenSchema({"unix-socket": "/tmp/kresd-socket"})
+    assert ListenSchema({"unix-socket": ["/tmp/kresd-socket", "/tmp/kresd-socket2"]})
+
+    with raises(KresManagerException):
+        ListenSchema({"ip-address": "::1", "unix-socket": "/tmp/kresd-socket"})
+    with raises(KresManagerException):
+        ListenSchema({"unit-socket": "/tmp/kresd-socket", "port": "53"})
+
+
+def test_listen_ip_address():
+    assert ListenSchema({"ip-address": "::1"})
+    assert ListenSchema({"ip-address": "::1@5353"})
+    assert ListenSchema({"ip-address": "::1", "port": 5353})
+    assert ListenSchema({"ip-address": ["127.0.0.1", "::1"]})
+    assert ListenSchema({"ip-address": ["127.0.0.1@5353", "::1@5353"]})
+    assert ListenSchema({"ip-address": ["127.0.0.1", "::1"], "port": 5353})
+
+    with raises(KresManagerException):
+        ListenSchema({"ip-address": "::1@5353", "port": 5353})
+    with raises(KresManagerException):
+        ListenSchema({"ip-address": ["127.0.0.1", "::1@5353"]})
+    with raises(KresManagerException):
+        ListenSchema({"ip-address": ["127.0.0.1@5353", "::1@5353"], "port": 5353})
+
+
+def test_listen_interface():
+    assert ListenSchema({"interface": "lo"})
+    assert ListenSchema({"interface": "lo@5353"})
+    assert ListenSchema({"interface": "lo", "port": 5353})
+    assert ListenSchema({"interface": ["lo", "eth0"]})
+    assert ListenSchema({"interface": ["lo@5353", "eth0@5353"]})
+    assert ListenSchema({"interface": ["lo", "eth0"], "port": 5353})
+
+    with raises(KresManagerException):
+        ListenSchema({"interface": "lo@5353", "port": 5353})
+    with raises(KresManagerException):
+        ListenSchema({"interface": ["lo", "eth0@5353"]})
+    with raises(KresManagerException):
+        ListenSchema({"interface": ["lo@5353", "eth0@5353"], "port": 5353})
+
+
 def test_listen_validation():
     with raises(KresManagerException):
         ListenSchema({"ip-address": "::1", "port": -10})
     with raises(KresManagerException):
         ListenSchema({"ip-address": "::1", "interface": "eth0"})
-    with raises(KresManagerException):
-        ListenSchema({"ip-address": "::1", "unit-socket": "/tmp/kresd-socket"})
-    with raises(KresManagerException):
-        ListenSchema({"unit-socket": "/tmp/kresd-socket", "port": "53"})