datamodel: types: custom type for port number

- IntRange: base type for integer ranges

diff --git a/manager/knot_resolver_manager/datamodel/network_schema.py b/manager/knot_resolver_manager/datamodel/network_schema.py
index 06aaa8051563dd48c3079fcc57b3fa9b9614bf0f..b2851f60c5d66dc9daa6415a639fa5739e5c8738 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/network_schema.py
+++ b/manager/knot_resolver_manager/datamodel/network_schema.py
@@ -1,3 +1,4 @@
+import os
 from typing import List, Optional, Union
 
 from typing_extensions import Literal
@@ -10,6 +11,7 @@ from knot_resolver_manager.datamodel.types import (
     IPNetwork,
     IPv4Address,
     IPv6Address,
+    PortNumber,
     SizeUnit,
 )
 from knot_resolver_manager.utils import SchemaNode
@@ -47,7 +49,7 @@ class ListenSchema(SchemaNode):
         unix_socket: Union[None, CheckedPath, List[CheckedPath]] = None
         ip_address: Union[None, IPAddressPort, List[IPAddressPort]] = None
         interface: Union[None, InterfacePort, List[InterfacePort]] = None
-        port: Optional[int] = None
+        port: Optional[PortNumber] = None
         kind: KindEnum = "dns"
         freebind: bool = False
 
@@ -56,7 +58,7 @@ class ListenSchema(SchemaNode):
     unix_socket: Union[None, CheckedPath, List[CheckedPath]]
     ip_address: Union[None, IPAddressPort, List[IPAddressPort]]
     interface: Union[None, InterfacePort, List[InterfacePort]]
-    port: Optional[int]
+    port: Optional[PortNumber]
     kind: KindEnum
     freebind: bool
 
@@ -90,15 +92,15 @@ class ListenSchema(SchemaNode):
             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]:
+    def _port(self, origin: Raw) -> Optional[PortNumber]:
         if origin.port:
             return origin.port
         elif origin.ip_address or origin.interface:
             if origin.kind == "dot":
-                return 853
+                return PortNumber(853)
             elif origin.kind == "doh2":
-                return 443
-            return 53
+                return PortNumber(443)
+            return PortNumber(53)
         return None
 
     def _validate(self) -> None:
@@ -117,8 +119,6 @@ class ListenSchema(SchemaNode):
                 "'unix-socket' and 'port' are not compatible options. "
                 "Port configuration can only be used with 'ip-address' or 'interface'."
             )
-        if self.port and not 0 <= self.port <= 65_535:
-            raise ValueError(f"Port value {self.port} out of range of usual 2-byte port value")
 
 
 class NetworkSchema(SchemaNode):

diff --git a/manager/knot_resolver_manager/datamodel/types.py b/manager/knot_resolver_manager/datamodel/types.py
index c97f136458ca635b91351bfade5b723f90d0dd8a..c545739bb454cff0ce2fb408c840b1cf32424118 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/types.py
+++ b/manager/knot_resolver_manager/datamodel/types.py
@@ -1,5 +1,6 @@
 import ipaddress
 import logging
+import os
 import re
 from enum import Enum, auto
 from pathlib import Path
@@ -164,6 +165,47 @@ RecordTypeEnum = Literal[
 ]
 
 
+class IntRange(CustomValueType):
+    _value: int
+    _min: int
+    _max: int
+
+    def __init__(self, source_value: Any, object_path: str = "/") -> None:
+        if isinstance(source_value, int):
+            if not self._min <= source_value <= self._max:
+                raise SchemaException(
+                    f"Integer value {source_value} out of range <{self._min}, {self._max}>", object_path
+                )
+            self._value = source_value
+        else:
+            raise SchemaException(
+                f"Unexpected input type for integer - {type(source_value)}."
+                " Cause might be invalid format or invalid type.",
+                object_path,
+            )
+
+    def __int__(self) -> int:
+        return self._value
+
+    def __str__(self) -> str:
+        return str(self._value)
+
+    def __eq__(self, o: object) -> bool:
+        return isinstance(o, IntRange) and o._value == self._value
+
+    def serialize(self) -> Any:
+        return self._value
+
+    @classmethod
+    def json_schema(cls: Type["IntRange"]) -> Dict[Any, Any]:
+        return {"type": "integer", "minimum": cls._min, "maximum": cls._max}
+
+
+class PortNumber(IntRange):
+    _min: int = 1
+    _max: int = 65_535
+
+
 class Unit(CustomValueType):
     _re: Pattern[str]
     _units: Dict[str, int]
@@ -355,7 +397,7 @@ class DomainName(CustomValueType):
 
 class InterfacePort(CustomValueType):
     intrfc: str
-    port: Optional[int] = None
+    port: Optional[PortNumber] = None
 
     def __init__(self, source_value: Any, object_path: str = "/") -> None:
         super().__init__(source_value)
@@ -363,14 +405,9 @@ class InterfacePort(CustomValueType):
             if "@" in source_value:
                 sep = source_value.split("@", 1)
                 try:
-                    self.port = int(sep[1])
+                    self.port = PortNumber(int(sep[1]))
                 except ValueError as e:
                     raise SchemaException("Failed to parse port.", object_path) from e
-
-                if not 0 <= self.port <= 65_535:
-                    raise SchemaException(
-                        f"Port value '{self.port}' out of range of usual 2-byte port value", object_path
-                    )
                 self.intrfc = sep[0]
             else:
                 self.intrfc = source_value
@@ -410,7 +447,7 @@ class InterfacePort(CustomValueType):
 
 class IPAddressPort(CustomValueType):
     addr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
-    port: Optional[int] = None
+    port: Optional[PortNumber] = None
 
     def __init__(self, source_value: Any, object_path: str = "/") -> None:
         super().__init__(source_value)
@@ -418,15 +455,10 @@ class IPAddressPort(CustomValueType):
             if "@" in source_value:
                 sep = source_value.split("@", 1)
                 try:
-                    self.port = int(sep[1])
+                    self.port = PortNumber(int(sep[1]))
                 except ValueError as e:
                     raise SchemaException("Failed to parse port.", object_path) from e
 
-                if not 0 <= self.port <= 65_535:
-                    raise SchemaException(
-                        f"Port value '{self.port}' out of range of usual 2-byte port value", object_path
-                    )
-
                 try:
                     self.addr = ipaddress.ip_address(sep[0])
                 except ValueError as e:

diff --git a/manager/tests/unit/datamodel/test_network_schema.py b/manager/tests/unit/datamodel/test_network_schema.py
index 1b8d9f83b749c05b4fc2cb491f196deac52e13d1..10f09b42b5537fc055d6b37d5592dd18ab2416b8 100644 (file)
--- a/manager/tests/unit/datamodel/test_network_schema.py
+++ b/manager/tests/unit/datamodel/test_network_schema.py
@@ -1,7 +1,7 @@
 from pytest import raises
 
 from knot_resolver_manager.datamodel.network_schema import ListenSchema, NetworkSchema
-from knot_resolver_manager.datamodel.types import IPAddressPort
+from knot_resolver_manager.datamodel.types import IPAddressPort, PortNumber
 from knot_resolver_manager.exceptions import KresManagerException
 
 
@@ -11,12 +11,12 @@ def test_listen_defaults():
     assert len(o.listen) == 2
     # {"ip-address": "127.0.0.1"}
     assert o.listen[0].ip_address == IPAddressPort("127.0.0.1")
-    assert o.listen[0].port == 53
+    assert o.listen[0].port == PortNumber(53)
     assert o.listen[0].kind == "dns"
     assert o.listen[0].freebind == False
     # {"ip-address": "::1", "freebind": True}
     assert o.listen[1].ip_address == IPAddressPort("::1")
-    assert o.listen[1].port == 53
+    assert o.listen[1].port == PortNumber(53)
     assert o.listen[1].kind == "dns"
     assert o.listen[1].freebind == True
 
@@ -28,9 +28,9 @@ def test_listen_kind_port_defaults():
     doh2 = ListenSchema({"ip-address": "::1", "kind": "doh2"})
 
     assert soc.port == None
-    assert dns.port == 53
-    assert dot.port == 853
-    assert doh2.port == 443
+    assert dns.port == PortNumber(53)
+    assert dot.port == PortNumber(853)
+    assert doh2.port == PortNumber(443)
 
 
 def test_listen_unix_socket():