groups:
- manager
network:
- interfaces:
- - listen:
- ip: 127.0.0.1
- port: 5353
+ listen:
+ - ip-address: 127.0.0.1@5353
server:
workers: 1
rundir: etc/knot-resolver/runtime
management:
listen:
ip: 127.0.0.1
- port: 5000
+ port: 5000
\ No newline at end of file
-from typing import List, Optional
+from typing import List, Optional, Union
from typing_extensions import Literal
from knot_resolver_manager.datamodel.types import (
CheckedPath,
+ InterfacePort,
IPAddress,
+ IPAddressPort,
IPNetwork,
IPv4Address,
IPv6Address,
- Listen,
SizeUnit,
)
from knot_resolver_manager.utils import SchemaNode
-KindEnum = Literal["dns", "xdp", "dot", "doh"]
-
-
-class InterfaceSchema(SchemaNode):
- class Raw(SchemaNode):
- listen: Listen
- kind: KindEnum = "dns"
- freebind: bool = False
-
- _PREVIOUS_SCHEMA = Raw
-
- listen: Listen
- kind: KindEnum
- freebind: bool
-
- def _listen(self, origin: Raw) -> Listen:
- if not origin.listen.port:
- if origin.kind == "dot":
- origin.listen.port = 853
- elif origin.kind == "doh":
- origin.listen.port = 443
- else:
- origin.listen.port = 53
- return origin.listen
+KindEnum = Literal["dns", "xdp", "dot", "doh2"]
class EdnsBufferSizeSchema(SchemaNode):
raise ValueError("'padding' must be number in range<0-512>")
+class ListenSchema(SchemaNode):
+ class Raw(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
+ kind: KindEnum = "dns"
+ freebind: bool = False
+
+ _PREVIOUS_SCHEMA = Raw
+
+ unix_socket: Union[None, CheckedPath, List[CheckedPath]]
+ ip_address: Union[None, IPAddressPort, IPAddressPort, List[IPAddressPort]]
+ interface: Union[None, InterfacePort, List[InterfacePort]]
+ port: Optional[int]
+ kind: KindEnum
+ freebind: bool
+
+ def _port(self, origin: Raw) -> Optional[int]:
+ if origin.port:
+ return origin.port
+ elif origin.ip_address or origin.interface:
+ if origin.kind == "dot":
+ return 853
+ elif origin.kind == "doh2":
+ return 443
+ return 53
+ return None
+
+ def _validate(self) -> None:
+ present = {
+ "ip_address" if self.ip_address is not None else ...,
+ "unix_socket" if self.unix_socket is not None else ...,
+ "interface" if self.interface is not None else ...,
+ }
+ if not (present == {"ip_address", ...} or present == {"unix_socket", ...} or present == {"interface", ...}):
+ raise ValueError(
+ "Listen configuration contains multiple incompatible options at once. "
+ "Only one of 'ip-address', 'interface' and 'unix-socket' optionscan be configured at once."
+ )
+ if self.port and self.unix_socket:
+ raise ValueError(
+ "'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):
do_ipv4: bool = True
do_ipv6: bool = True
edns_buffer_size: EdnsBufferSizeSchema = EdnsBufferSizeSchema()
address_renumbering: Optional[List[AddressRenumberingSchema]] = None
tls: TLSSchema = TLSSchema()
- interfaces: List[InterfaceSchema] = [
- InterfaceSchema({"listen": {"ip": "127.0.0.1", "port": 53}}),
- InterfaceSchema({"listen": {"ip": "::1", "port": 53}, "freebind": True}),
+ listen: List[ListenSchema] = [
+ ListenSchema({"ip-address": "127.0.0.1"}),
+ ListenSchema({"ip-address": "::1", "freebind": True}),
]
def _validate(self):
-{% macro net_listen(interface) -%}
-net.listen(
-{%- if interface.listen.ip -%}
-'{{ interface.listen.ip|string }}',
-{%- if interface.listen.port -%}
-{{ interface.listen.port|int }},
+{% macro listen_interface(interface) -%}
+net.{{ interface }}
+{%- endmacro %}
+
+
+{% macro listen_kind(kind) -%}
+'{{ 'tls' if kind == 'dot' else kind }}'
+{%- endmacro %}
+
+
+{% macro net_listen_unix_socket(socket, kind, freebind) -%}
+net.listen('{{ socket }}',nil,{kind={{ listen_kind(kind) }},freebind={{ 'true' if freebind else 'false'}}})
+{%- endmacro %}
+
+
+{% macro net_listen_ip_address(ip_address, kind, freebind, port) -%}
+net.listen('{{ ip_address.addr }}',
+{%- if ip_address.port -%}
+{{ ip_address.port }},
+{%- else -%}
+{{ port }},
{%- endif -%}
-{%- elif interface.listen.unix_socket -%}
-'{{ interface.listen.unix_socket|string }}',nil,
-{%- elif interface.listen.interface -%}
-net.{{ interface.listen.interface|string }},
-{%- if interface.listen.port -%}
-{{ interface.listen.port|int }},
+{kind={{ listen_kind(kind) }},freebind={{ 'true' if freebind else 'false'}}})
+{%- endmacro %}
+
+
+{% macro net_listen_interface(interface, kind, freebind, port) -%}
+net.listen({{ listen_interface(interface.intrfc) }},
+{%- if interface.port -%}
+{{ interface.port }},
+{%- else -%}
+{{ port }},
{%- endif -%}
+{kind={{ listen_kind(kind) }},freebind={{ 'true' if freebind else 'false'}}})
+{%- endmacro %}
+
+
+{% macro network_listen(listen) -%}
+{%- if listen.unix_socket -%}
+ {%- if listen.unix_socket is iterable-%}
+ {% for socket in listen.unix_socket -%}
+ {{ net_listen_unix_socket(socket, listen.kind, listen.freebind) }}
+ {% endfor -%}
+ {%- else -%}
+ {{ net_listen_unix_socket(listen.unix_socket, listen.kind, listen.freebind) }}
+ {%- endif -%}
+{%- elif listen.ip_address -%}
+ {%- if listen.ip_address is iterable-%}
+ {% for address in listen.ip_address -%}
+ {{ net_listen_ip_address(address, listen.kind, listen.freebind, listen.port) }}
+ {% endfor -%}
+ {%- else -%}
+ {{ net_listen_ip_address(listen.ip_address, listen.kind, listen.freebind, listen.port) }}
+ {%- endif -%}
+{%- elif listen.interface -%}
+ {%- if listen.interface is iterable-%}
+ {% for interface in listen.interface -%}
+ {{ net_listen_interface(interface, listen.kind, listen.freebind, listen.port) }}
+ {% endfor -%}
+ {%- else -%}
+ {{ net_listen_interface(listen.interface, listen.kind, listen.freebind, listen.port) }}
+ {%- endif -%}
{%- endif -%}
-{kind='{{ 'tls' if interface.kind == 'dot' else interface.kind }}',freebind={{ 'true' if interface.freebind else 'false'}}})
{%- endmacro %}
\ No newline at end of file
-{% from 'macros/network_macros.lua.j2' import net_listen %}
+{% from 'macros/network_macros.lua.j2' import network_listen %}
-- network.do-ipv4/6
net.ipv4 = {{ 'true' if cfg.network.do_ipv4 else 'false' }}
}
{% endif %}
--- network.interfaces
-{% for interface in cfg.network.interfaces %}
-{{ net_listen(interface) }}
+-- network.listen
+{% for listen in cfg.network.listen %}
+{{ network_listen(listen) }}
{% endfor %}
\ No newline at end of file
}
+class InterfacePort(CustomValueType):
+ intrfc: str
+ port: Optional[int] = None
+
+ def __init__(self, source_value: Any, object_path: str = "/") -> None:
+ super().__init__(source_value)
+ if isinstance(source_value, str):
+ if "@" in source_value:
+ sep = source_value.split("@", 1)
+ try:
+ self.port = 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
+ self._value = source_value
+
+ else:
+ raise SchemaException(
+ f"Unexpected value for a '<interface>[@<port>]'. Expected string, got '{source_value}'"
+ f" with type '{type(source_value)}'",
+ object_path,
+ )
+
+ def to_std(self) -> str:
+ return self._value
+
+ def __str__(self) -> str:
+ return self._value
+
+ def __int__(self) -> int:
+ raise ValueError("Can't convert InterfacePort to an integer")
+
+ def __eq__(self, o: object) -> bool:
+ """
+ Two instances of InterfacePort are equal when they represent same string.
+ """
+ return isinstance(o, InterfacePort) and str(o._value) == str(self._value)
+
+ def serialize(self) -> Any:
+ return str(self._value)
+
+ @classmethod
+ def json_schema(cls: Type["InterfacePort"]) -> Dict[Any, Any]:
+ return {
+ "type": "string",
+ }
+
+
class IPAddressPort(CustomValueType):
+ addr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
+ port: Optional[int] = None
+
def __init__(self, source_value: Any, object_path: str = "/") -> None:
super().__init__(source_value)
if isinstance(source_value, str):
- addr = source_value
if "@" in source_value:
sep = source_value.split("@", 1)
- addr = sep[0]
try:
- port = int(sep[1])
+ self.port = int(sep[1])
except ValueError as e:
raise SchemaException("Failed to parse port.", object_path) from e
- if not 0 <= port <= 65_535:
- raise SchemaException(f"Port value '{port}' out of range of usual 2-byte port value", object_path)
- try:
- ipaddress.ip_address(addr)
- except ValueError as e:
- raise SchemaException("Failed to parse IP address.", 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:
+ raise SchemaException("Failed to parse IP address.", object_path) from e
+ else:
+ try:
+ self.addr = ipaddress.ip_address(source_value)
+ except ValueError as e:
+ raise SchemaException("Failed to parse IP address.", object_path) from e
+ self._value = source_value
- self._value: str = source_value
else:
raise SchemaException(
- f"Unexpected value for a '<ip-address>@<port>'. Expected string, got '{source_value}'"
+ f"Unexpected value for a '<ip-address>[@<port>]'. Expected string, got '{source_value}'"
f" with type '{type(source_value)}'",
object_path,
)
network:
- interfaces:
- - listen:
- ip: 127.0.0.1
- port: 5353
+ listen:
+ - ip-address: 127.0.0.1@5353
server:
workers: 1
rundir: tests/integration/run
cache:
storage: cache
logging:
- level: debug
+ level: debug
\ No newline at end of file
from knot_resolver_manager.datamodel.config_schema import template_from_str
-from knot_resolver_manager.datamodel.network_schema import InterfaceSchema
+from knot_resolver_manager.datamodel.network_schema import ListenSchema
-def test_net_listen():
- ip = InterfaceSchema({"listen": {"ip": "::1", "port": 53}, "freebind": True})
- soc = InterfaceSchema({"listen": {"unix-socket": "/tmp/kresd-socket"}, "kind": "dot"})
- infc = InterfaceSchema({"listen": {"interface": "eth0"}, "kind": "doh"})
-
- tmpl_str = """{% from 'macros/network_macros.lua.j2' import net_listen %}
-{{ net_listen(interface) }}"""
-
+def test_network_listen():
+ tmpl_str = """{% from 'macros/network_macros.lua.j2' import network_listen %}
+{{ network_listen(listen) }}"""
tmpl = template_from_str(tmpl_str)
+
+ soc = ListenSchema({"unix-socket": "/tmp/kresd-socket", "kind": "dot"})
+ assert tmpl.render(listen=soc) == "net.listen('/tmp/kresd-socket',nil,{kind='tls',freebind=false})"
+ soc_list = ListenSchema({"unix-socket": [soc.unix_socket, "/tmp/kresd-socket2"], "kind": "dot"})
assert (
- tmpl.render(interface=ip)
- == f"net.listen('{ip.listen.ip}',{ip.listen.port},{{kind='dns',freebind={str(ip.freebind).lower()}}})"
+ tmpl.render(listen=soc_list)
+ == "net.listen('/tmp/kresd-socket',nil,{kind='tls',freebind=false})\n"
+ + "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_list = ListenSchema({"ip-address": [ip.ip_address, "127.0.0.1@5353"]})
assert (
- tmpl.render(interface=soc)
- == f"net.listen('{soc.listen.unix_socket}',nil,{{kind='tls',freebind={str(soc.freebind).lower()}}})"
+ tmpl.render(listen=ip_list)
+ == "net.listen('::1',53,{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"})
assert (
- tmpl.render(interface=infc)
- == f"net.listen(net.{infc.listen.interface},443,{{kind='doh',freebind={str(soc.freebind).lower()}}})"
+ 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"
)
-import ipaddress
+from pytest import raises
-from knot_resolver_manager.datamodel.network_schema import NetworkSchema
-from knot_resolver_manager.datamodel.types import IPv4Address, IPv6Address
+from knot_resolver_manager.datamodel.network_schema import ListenSchema, NetworkSchema
+from knot_resolver_manager.datamodel.types import IPAddressPort
+from knot_resolver_manager.exceptions import KresManagerException
-def test_interfaces_default():
+def test_listen_defaults():
o = NetworkSchema()
- assert len(o.interfaces) == 2
- # {"listen": {"ip": "127.0.0.1", "port": 53}}
- assert o.interfaces[0].listen.ip == IPv4Address("127.0.0.1")
- assert o.interfaces[0].listen.port == 53
- assert o.interfaces[0].kind == "dns"
- assert o.interfaces[0].freebind == False
- # {"listen": {"ip": "::1", "port": 53}, "freebind": True}
- assert o.interfaces[1].listen.ip == IPv6Address("::1")
- assert o.interfaces[1].listen.port == 53
- assert o.interfaces[1].kind == "dns"
- assert o.interfaces[1].freebind == True
+ 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].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].kind == "dns"
+ assert o.listen[1].freebind == True
+
+
+def test_listen_kind_port_defaults():
+ soc = ListenSchema({"unix-socket": "/tmp/kresd-socket"})
+ dns = ListenSchema({"ip-address": "::1"})
+ dot = ListenSchema({"ip-address": "::1", "kind": "dot"})
+ doh2 = ListenSchema({"ip-address": "::1", "kind": "doh2"})
+
+ assert soc.port == None
+ assert dns.port == 53
+ assert dot.port == 853
+ assert doh2.port == 443
+
+
+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"})
projects / thirdparty / knot-resolver.git / commitdiff
