From: Aleš Date: Thu, 20 Jan 2022 15:14:23 +0000 (+0100) Subject: datamodel: types: custom type for port number X-Git-Tag: v6.0.0a1~45^2~10 X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=1bb547ddba8da6eabaee603fe26e6c3bea21c72a;p=thirdparty%2Fknot-resolver.git 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 06aaa8051..b2851f60c 100644 --- 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 '@' 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 c97f13645..c545739bb 100644 --- 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 1b8d9f83b..10f09b42b 100644 --- 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():