- manager
network:
listen:
- - ip-address: 127.0.0.1@5353
+ - interface: lo@5353
server:
workers: 1
rundir: etc/knot-resolver/runtime
management:
- ip-address: 127.0.0.1@5000
\ No newline at end of file
+ interface: 127.0.0.1@5000
\ No newline at end of file
CheckedPath,
InterfaceOptionalPort,
IPAddress,
- IPAddressOptionalPort,
IPNetwork,
IPv4Address,
IPv6Address,
raise ValueError("'padding' must be number in range<0-512>")
-def listen_config_validate(obj: object) -> None:
- present = {
- "ip-address" if hasattr(obj, "ip_address") and getattr(obj, "ip_address") is not None else ...,
- "unix-socket" if hasattr(obj, "unix_socket") and getattr(obj, "unix_socket") is not None else ...,
- "interface" if hasattr(obj, "interface") and getattr(obj, "interface") is not None else ...,
- }
- if not (present == {"ip-address", ...} or present == {"unix-socket", ...} or present == {"interface", ...}):
- raise ValueError(
- "Listen configuration contains incompatible configuration options."
- f" Expected one of 'ip-address', 'interface' and 'unix-socket' options, got '{present}'."
- )
-
-
class ListenSchema(SchemaNode):
class Raw(SchemaNode):
- unix_socket: Union[None, CheckedPath, List[CheckedPath]] = None
- ip_address: Union[None, IPAddressOptionalPort, List[IPAddressOptionalPort]] = None
interface: Union[None, InterfaceOptionalPort, List[InterfaceOptionalPort]] = None
+ unix_socket: Union[None, CheckedPath, List[CheckedPath]] = None
port: Optional[PortNumber] = None
kind: KindEnum = "dns"
freebind: bool = False
_PREVIOUS_SCHEMA = Raw
- unix_socket: Union[None, CheckedPath, List[CheckedPath]]
- ip_address: Union[None, IPAddressOptionalPort, List[IPAddressOptionalPort]]
interface: Union[None, InterfaceOptionalPort, List[InterfaceOptionalPort]]
+ unix_socket: Union[None, CheckedPath, List[CheckedPath]]
port: Optional[PortNumber]
kind: KindEnum
freebind: bool
- def _ip_address(self, origin: Raw) -> Union[None, IPAddressOptionalPort, List[IPAddressOptionalPort]]:
- 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>' syntax must be used either for all or none of the IP addresses in the list."
- )
- port_set = bool(addr.port)
- elif isinstance(origin.ip_address, IPAddressOptionalPort) 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, InterfaceOptionalPort, List[InterfaceOptionalPort]]:
if isinstance(origin.interface, list):
port_set: Optional[bool] = None
def _port(self, origin: Raw) -> Optional[PortNumber]:
if origin.port:
return origin.port
- elif origin.ip_address or origin.interface:
+ # default port number based on kind
+ elif origin.interface:
if origin.kind == "dot":
return PortNumber(853)
elif origin.kind == "doh2":
return None
def _validate(self) -> None:
- listen_config_validate(self)
+ if bool(self.unix_socket) == bool(self.interface):
+ raise ValueError("One of 'interface' or 'unix-socket' must be configured.")
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'."
+ "'unix-socket' and 'port' are not compatible options."
+ " Port configuration can only be used with 'interface' option."
)
address_renumbering: Optional[List[AddressRenumberingSchema]] = None
tls: TLSSchema = TLSSchema()
listen: List[ListenSchema] = [
- ListenSchema({"ip-address": "127.0.0.1"}),
- ListenSchema({"ip-address": "::1", "freebind": True}),
+ ListenSchema({"interface": "127.0.0.1"}),
+ ListenSchema({"interface": "::1", "freebind": True}),
]
def _validate(self):
from typing_extensions import Literal
-from knot_resolver_manager.datamodel.network_schema import listen_config_validate
from knot_resolver_manager.datamodel.types import (
CheckedPath,
DNSRecordTypeEnum,
class ManagementSchema(SchemaNode):
unix_socket: Optional[CheckedPath] = None
- ip_address: Optional[IPAddressPort] = None
+ interface: Optional[IPAddressPort] = None
def _validate(self) -> None:
- if bool(self.unix_socket) == bool(self.ip_address):
- raise ValueError("One of 'ip-address' or 'unix-socket' must be configured..")
+ if bool(self.unix_socket) == bool(self.interface):
+ raise ValueError("One of 'interface' or 'unix-socket' must be configured.")
class WebmgmtSchema(SchemaNode):
unix_socket: Optional[CheckedPath] = None
- ip_address: Optional[IPAddressPort] = None
interface: Optional[InterfacePort] = None
tls: bool = False
cert_file: Optional[CheckedPath] = None
key_file: Optional[CheckedPath] = None
def _validate(self) -> None:
- listen_config_validate(self)
+ if bool(self.unix_socket) == bool(self.interface):
+ raise ValueError("One of 'interface' or 'unix-socket' must be configured.")
class ServerSchema(SchemaNode):
-{% macro listen_interface(interface) -%}
-net.{{ interface }}
-{%- endmacro %}
-
-
{% macro listen_kind(kind) -%}
'{{ 'tls' if kind == 'dot' else kind }}'
{%- endmacro %}
{%- 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 -%}
-{kind={{ listen_kind(kind) }},freebind={{ 'true' if freebind else 'false'}}})
-{%- endmacro %}
-
-
{% macro net_listen_interface(interface, kind, freebind, port) -%}
-net.listen({{ listen_interface(interface.if_name) }},
+net.listen(
+{%- if interface.addr -%}
+'{{ interface.addr }}',
+{%- elif interface.if_name -%}
+net.{{ interface.if_name }},
+{%- endif -%}
{%- if interface.port -%}
{{ interface.port }},
{%- else -%}
{%- 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 -%}
{% if cfg.server.webmgmt -%}
-- server.webmgmt
modules.load('http')
-http.config({
- tls = {{ 'true' if cfg.server.webmgmt.tls else 'false'}},
- {{ "cert = '"+cfg.server.webmgmt.cert_file+"'," if cfg.server.webmgmt.cert_file }}
- {{ "key = '"+cfg.server.webmgmt.key_file+"'," if cfg.server.webmgmt.key_file }}
+http.config({tls = {{ 'true' if cfg.server.webmgmt.tls else 'false'}},
+{%- if cfg.server.webmgmt.cert_file -%}
+ cert = '{{ cfg.server.webmgmt.cert_file }}',
+{%- endif -%}
+{%- if cfg.server.webmgmt.cert_file -%}
+ key = '{{ cfg.server.webmgmt.key_file }}',
+{%- endif -%}
}, 'webmgmt')
net.listen(
{%- if cfg.server.webmgmt.unix_socket -%}
'{{ cfg.server.webmgmt.unix_socket }}',nil,
-{%- elif cfg.server.webmgmt.ip_address -%}
- '{{ cfg.server.webmgmt.ip_address.addr }}',{{ cfg.server.webmgmt.ip_address.port }},
{%- elif cfg.server.webmgmt.interface -%}
- net.{{ cfg.server.webmgmt.interface.if_name }},{{ cfg.server.webmgmt.interface.port }},
+ {%- if cfg.server.webmgmt.interface.addr -%}
+ '{{ cfg.server.webmgmt.interface.addr }}',{{ cfg.server.webmgmt.interface.port }},
+ {%- elif cfg.server.webmgmt.interface.if_name -%}
+ net.{{ cfg.server.webmgmt.interface.if_name }},{{ cfg.server.webmgmt.interface.port }},
+ {%- endif -%}
{%- endif -%}
{ kind = 'webmgmt' })
{%- endif %}
\ No newline at end of file
class InterfacePort(StrBase):
+ addr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
if_name: InterfaceName
port: PortNumber
if isinstance(source_value, str):
parts = source_value.split("@")
if len(parts) == 2:
+ try:
+ self.addr = ipaddress.ip_address(parts[0])
+ except ValueError as e1:
+ try:
+ self.if_name = InterfaceName(parts[0])
+ except SchemaException as e2:
+ raise SchemaException(
+ f"Expected IP address or interface name, got '{parts[0]}'.", object_path
+ ) from e1 and e2
self.port = PortNumber.from_str(parts[1], object_path)
- self.if_name = InterfaceName(parts[0], object_path)
else:
- raise SchemaException(f"Expected '<interface>@<port>', got '{source_value}'.", object_path)
+ raise SchemaException(
+ f"Expected '<ip-address|interface-name>@<port>', got '{source_value}'.", object_path
+ )
self._value = source_value
else:
raise SchemaException(
- "Unexpected value for '<i
nterface>@<port>'."
+ "Unexpected value for '<i
p-address|interface-name>@<port>'."
f" Expected string, got '{source_value}' with type '{type(source_value)}'",
object_path,
)
class InterfaceOptionalPort(StrBase):
- if_name: InterfaceName
+ addr: Union[None, ipaddress.IPv4Address, ipaddress.IPv6Address] = None
+ if_name: Optional[InterfaceName] = None
port: Optional[PortNumber] = None
def __init__(self, source_value: Any, object_path: str = "/") -> None:
if isinstance(source_value, str):
parts = source_value.split("@")
if 0 < len(parts) < 3:
- self.if_name = InterfaceName(parts[0], object_path)
+ try:
+ self.addr = ipaddress.ip_address(parts[0])
+ except ValueError as e1:
+ try:
+ self.if_name = InterfaceName(parts[0])
+ except SchemaException as e2:
+ raise SchemaException(
+ f"Expected IP address or interface name, got '{parts[0]}'.", object_path
+ ) from e1 and e2
if len(parts) == 2:
self.port = PortNumber.from_str(parts[1], object_path)
else:
- raise SchemaException(f"Expected '<i
nterface>[@<port>]', got '{parts}'.", object_path)
+ raise SchemaException(f"Expected '<i
p-address|interface-name>[@<port>]', got '{parts}'.", object_path)
self._value = source_value
else:
raise SchemaException(
- "Unexpected value for '<i
nterface>[@<port>]'."
+ "Unexpected value for '<i
p-address|interface-name>[@<port>]'."
f" Expected string, got '{source_value}' with type '{type(source_value)}'",
object_path,
)
if mgn.unix_socket:
nsite = web.UnixSite(self.runner, str(mgn.unix_socket))
logger.info(f"Starting API HTTP server on http+unix://{mgn.unix_socket}")
- elif mgn.ip_address:
- nsite = web.TCPSite(self.runner, str(mgn.ip_address.addr), int(mgn.ip_address.port))
- logger.info(f"Starting API HTTP server on http://{mgn.ip_address.addr}:{mgn.ip_address.port}")
+ elif mgn.interface:
+ nsite = web.TCPSite(self.runner, str(mgn.interface.addr), int(mgn.interface.port))
+ logger.info(f"Starting API HTTP server on http://{mgn.interface.addr}:{mgn.interface.port}")
else:
raise KresManagerException("Requested API on unsupported configuration format.")
await nsite.start()
if self.listen is not None and self.site is not None:
if self.listen.unix_socket:
logger.info(f"Stopping API HTTP server on http+unix://{mgn.unix_socket}")
- elif self.listen.ip_address:
+ elif self.listen.interface:
logger.info(
- f"Stopping API HTTP server on http://{self.listen.ip_address.addr}:{self.listen.ip_address.port}"
+ f"Stopping API HTTP server on http://{self.listen.interface.addr}:{self.listen.interface.port}"
)
await self.site.stop()
network:
listen:
- - ip-address: 127.0.0.1@5353
+ - interface: 127.0.0.1@5353
server:
workers: 1
rundir: tests/integration/run
management:
- ip-address: 127.0.0.1@5001
+ interface: 127.0.0.1@5001
cache:
storage: cache
logging:
+ "net.listen('/tmp/kresd-socket2',nil,{kind='tls',freebind=false})\n"
)
- ip = ListenSchema({"ip-address": "::1@55", "freebind": True})
+ ip = ListenSchema({"interface": "::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"]})
+ ip_list = ListenSchema({"interface": [ip.interface, "127.0.0.1@5353"]})
assert (
tmpl.render(listen=ip_list)
== "net.listen('::1',55,{kind='dns',freebind=false})\n"
IPv4Address,
IPv6Address,
IPv6Network96,
+ PortNumber,
SizeUnit,
TimeUnit,
)
from knot_resolver_manager.utils import SchemaNode
+def test_port_number():
+ assert PortNumber(1)
+ assert PortNumber(65_535)
+ assert PortNumber(5353)
+ assert PortNumber(5000)
+
+ with raises(KresManagerException):
+ PortNumber(0)
+ with raises(KresManagerException):
+ PortNumber(65_636)
+ with raises(KresManagerException):
+ PortNumber(-1)
+
+
def test_size_unit():
assert SizeUnit("5368709120B") == SizeUnit("5242880K") == SizeUnit("5120M") == SizeUnit("5G")
def test_interface_port():
- class TestSchema(SchemaNode):
- interface: InterfacePort
+ o = InterfacePort("lo@5353")
+ assert str(o) == "lo@5353"
+ assert o == InterfacePort("lo@5353")
+ assert str(o.if_name) == "lo"
+ assert o.port == PortNumber(5353)
- o = TestSchema({"interface": "lo@5353"})
- assert str(o.interface) == "lo@5353"
- assert o.interface == InterfacePort("lo@5353")
+ o = InterfacePort("2001:db8::1000@5001")
+ assert str(o) == "2001:db8::1000@5001"
+ assert o == InterfacePort("2001:db8::1000@5001")
+ assert str(o.addr) == "2001:db8::1000"
+ assert o.port == PortNumber(5001)
with raises(KresManagerException):
- TestSchema({"interface": "lo"})
+ InterfacePort("lo")
with raises(KresManagerException):
- TestSchema({"interface": "lo@"})
- with raises(KresManagerException):
- TestSchema({"interface": "lo@-1"})
- with raises(KresManagerException):
- TestSchema({"interface": "lo@65536"})
+ InterfacePort("53")
def test_interface_optional_port():
- class TestSchema(SchemaNode):
- interface: InterfaceOptionalPort
+ o = InterfaceOptionalPort("lo")
+ assert str(o) == "lo"
+ assert o == InterfaceOptionalPort("lo")
+ assert str(o.if_name) == "lo"
+ assert o.port == None
- o = TestSchema({"interface": "lo"})
- assert str(o.interface) == "lo"
- assert o.interface == InterfaceOptionalPort("lo")
+ o = InterfaceOptionalPort("123.4.5.6")
+ assert str(o) == "123.4.5.6"
+ assert o == InterfaceOptionalPort("123.4.5.6")
+ assert str(o.addr) == "123.4.5.6"
+ assert o.port == None
- o = TestSchema({"interface": "lo@5353"})
- assert str(o.interface) == "lo@5353"
- assert o.interface == InterfaceOptionalPort("lo@5353")
+ o = InterfaceOptionalPort("lo@5353")
+ assert str(o) == "lo@5353"
+ assert o == InterfaceOptionalPort("lo@5353")
+ assert str(o.if_name) == "lo"
+ assert o.port == PortNumber(5353)
+
+ o = InterfaceOptionalPort("2001:db8::1000@5001")
+ assert str(o) == "2001:db8::1000@5001"
+ assert o == InterfaceOptionalPort("2001:db8::1000@5001")
+ assert str(o.addr) == "2001:db8::1000"
+ assert o.port == PortNumber(5001)
with raises(KresManagerException):
- TestSchema({"ip-port": "lo@"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "lo@-1"})
+ InterfaceOptionalPort("lo@")
with raises(KresManagerException):
- TestSchema({"ip-port": "lo@65536"})
+ InterfaceOptionalPort("@53")
def test_ip_address_port():
o = TestSchema({"ip-port": "123.4.5.6@5353"})
assert str(o.ip_port) == "123.4.5.6@5353"
- assert o.ip_port == IPAddressOptionalPort("123.4.5.6@5353")
+ assert o.ip_port == IPAddressPort("123.4.5.6@5353")
o = TestSchema({"ip-port": "2001:db8::1000@53"})
assert str(o.ip_port) == "2001:db8::1000@53"
- assert o.ip_port == IPAddressOptionalPort("2001:db8::1000@53")
+ assert o.ip_port == IPAddressPort("2001:db8::1000@53")
with raises(KresManagerException):
TestSchema({"ip-port": "123.4.5.6"})
from pytest import raises
from knot_resolver_manager.datamodel.network_schema import ListenSchema, NetworkSchema
-from knot_resolver_manager.datamodel.types import IPAddressOptionalPort, PortNumber
+from knot_resolver_manager.datamodel.types import PortNumber
+from knot_resolver_manager.datamodel.types.types import InterfaceOptionalPort
from knot_resolver_manager.exceptions import KresManagerException
assert len(o.listen) == 2
# {"ip-address": "127.0.0.1"}
- assert o.listen[0].ip_address == IPAddressOptionalPort("127.0.0.1")
+ assert o.listen[0].interface == InterfaceOptionalPort("127.0.0.1")
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 == IPAddressOptionalPort("::1")
+ assert o.listen[1].interface == InterfaceOptionalPort("::1")
assert o.listen[1].port == PortNumber(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 == PortNumber(53)
- assert dot.port == PortNumber(853)
- assert doh2.port == PortNumber(443)
+ assert ListenSchema({"unix-socket": "/tmp/kresd-socket"}).port == None
+ assert ListenSchema({"interface": "::1"}).port == PortNumber(53)
+ assert ListenSchema({"interface": "::1", "kind": "dot"}).port == PortNumber(853)
+ assert ListenSchema({"interface": "::1", "kind": "doh2"}).port == PortNumber(443)
def test_listen_unix_socket_valid():
def test_listen_ip_address_valid():
- 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})
+ assert ListenSchema({"interface": "::1"})
+ assert ListenSchema({"interface": "::1@5353"})
+ assert ListenSchema({"interface": "::1", "port": 5353})
+ assert ListenSchema({"interface": ["127.0.0.1", "::1"]})
+ assert ListenSchema({"interface": ["127.0.0.1@5353", "::1@5353"]})
+ assert ListenSchema({"interface": ["127.0.0.1", "::1"], "port": 5353})
def test_listen_ip_address_invalid():
def test_management():
- assert ManagementSchema({"ip-address": "::1@53"})
+ assert ManagementSchema({"interface": "::1@53"})
assert ManagementSchema({"unix-socket": "/path/socket"})
with raises(KresManagerException):
projects / thirdparty / knot-resolver.git / commitdiff
