{{ "'workarounds < iterate'," if cfg.options.violators_workarounds }}
{{ "'serve_stale < cache'," if cfg.options.serve_stale }}
{{ "dns64 = '"+cfg.dns64.prefix+"'," if cfg.dns64 }}
- 'hints > iterate'
+ 'hints > iterate',
+{% if cfg.network.address_renumbering %}
+-- network.address-renumbering
+ renumber = {
+{% for item in cfg.network.address_renumbering %}
+ {'{{ item.source }}', '{{ item.destination }}'},
+{% endfor %}
+ }
+{% endif %}
}
-- SERVER section
{% endif %}
-- NETWORK section
+-- network.do-ipv4/6
+net.ipv4 = {{ 'true' if cfg.network.do_ipv4 else 'false' }}
+net.ipv6 = {{ 'true' if cfg.network.do_ipv6 else 'false' }}
+
+-- network.out-interface-v4/v6
+{% if cfg.network.out_interface_v4 %}
+net.outgoing_v4('{{ cfg.network.out_interface_v4 }}')
+{% endif %}
+{% if cfg.network.out_interface_v6 %}
+net.outgoing_v6('{{ cfg.network.out_interface_v6 }}')
+{% endif %}
+
+-- network.tcp-pipeline
+net.tcp_pipeline({{ cfg.network.tcp_pipeline }})
+
+-- network.edns-keep-alive
+{% if cfg.network.edns_keep_alive %}
+modules.load('edns_keepalive')
+{% else %}
+modules.unload('edns_keepalive')
+{% endif %}
+
+-- network.edns-buffer-size
+net.bufsize({{ cfg.network.edns_buffer_size.upstream.bytes() }}, {{ cfg.network.edns_buffer_size.downstream.bytes() }})
+
+{% if cfg.network.tls.cert_file and cfg.network.tls.key_file %}
+-- network.tls
+net.tls('{{ cfg.network.tls.cert_file }}', '{{ cfg.network.tls.key_file }}')
+{% endif %}
+
+{% if cfg.network.tls.sticket_secret %}
+-- network.tls.sticket-secret
+net.tls_sticket_secret('{{ cfg.network.tls.sticket_secret }}')
+{% endif %}
+
+{% if cfg.network.tls.sticket_secret_file %}
+-- network.tls.sticket-secret-file
+net.tls_sticket_secret_file('{{ cfg.network.tls.sticket_secret_file }}')
+{% endif %}
+
+{% if cfg.network.tls.auto_discovery %}
+-- network.tls.auto-discovery
+modules.load('experimental_dot_auth')
+{% else %}
+-- modules.unload('experimental_dot_auth')
+{% endif %}
+
+-- network.tls.padding
+net.tls_padding({{ cfg.network.tls.padding }})
+
-- network.interfaces
{% for item in cfg.network.interfaces %}
net.listen('{{ item.listen.ip }}', {{ item.listen.port }}, {
-from typing import List
+from typing import List, Optional
-from knot_resolver_manager.datamodel.types import Listen
+from knot_resolver_manager.datamodel.types import (
+ AnyPath,
+ IPAddress,
+ IPv4Address,
+ IPv6Address,
+ Listen,
+ SizeUnit,
+ IPNetwork,
+)
from knot_resolver_manager.utils import SchemaNode
from knot_resolver_manager.utils.types import LiteralEnum
freebind: bool = False
+class EdnsBufferSizeSchema(SchemaNode):
+ upstream: SizeUnit = SizeUnit("1232B")
+ downstream: SizeUnit = SizeUnit("1232B")
+
+
+class AddressRenumberingSchema(SchemaNode):
+ source: IPNetwork
+ destination: IPAddress
+
+
+class TLSSchema(SchemaNode):
+ cert_file: Optional[AnyPath] = None
+ key_file: Optional[AnyPath] = None
+ sticket_secret: Optional[str] = None
+ sticket_secret_file: Optional[AnyPath] = None
+ auto_discovery: bool = False
+ padding: int = 1
+
+ def _validate(self):
+ if self.sticket_secret and self.sticket_secret_file:
+ raise ValueError("'sticket_secret' and 'sticket_secret_file' are both defined, only one can be used")
+ if not 0 <= self.padding <= 512:
+ raise ValueError("'padding' must be number in range<0-512>")
+
+
class NetworkSchema(SchemaNode):
+ do_ipv4: bool = True
+ do_ipv6: bool = True
+ out_interface_v4: Optional[IPv4Address] = None
+ out_interface_v6: Optional[IPv6Address] = None
+ tcp_pipeline: int = 100
+ edns_keep_alive: 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}),
]
+
+ def _validate(self):
+ if self.tcp_pipeline < 0:
+ raise ValueError("'tcp-pipeline' must be nonnegative number")
}
-class ListenType(Enum):
- IP_AND_PORT = auto()
- UNIX_SOCKET = auto()
- INTERFACE_AND_PORT = auto()
+class IPv4Address(CustomValueType):
+ def __init__(self, source_value: Any, object_path: str = "/") -> None:
+ super().__init__(source_value)
+ if isinstance(source_value, str):
+ try:
+ self._value: ipaddress.IPv4Address = ipaddress.IPv4Address(source_value)
+ except ValueError as e:
+ raise SchemaException("Failed to parse IPv4 address.", object_path) from e
+ else:
+ raise SchemaException(
+ f"Unexpected value for a IPv4 address. Expected string, got '{source_value}'"
+ f" with type '{type(source_value)}'",
+ object_path,
+ )
+ def to_std(self) -> ipaddress.IPv4Address:
+ return self._value
-class Listen(SchemaNode, Serializable):
- class Raw(SchemaNode):
- ip: Optional[str] = None
- port: Optional[int] = None
- unix_socket: Optional[AnyPath] = None
- interface: Optional[str] = None
+ def __str__(self) -> str:
+ return str(self._value)
- _PREVIOUS_SCHEMA = Raw
+ def __int__(self) -> int:
+ raise ValueError("Can't convert IPv4 address to an integer")
- typ: ListenType
- ip: Optional[Union[ipaddress.IPv4Address, ipaddress.IPv6Address]]
- port: Optional[int]
- unix_socket: Optional[AnyPath]
- interface: Optional[str]
+ def __eq__(self, o: object) -> bool:
+ """
+ Two instances of IPv4Address are equal when they represent same IPv4 address as string.
+ """
+ return isinstance(o, IPv4Address) and str(o._value) == str(self._value)
- def _typ(self, origin: Raw):
- present = {
- "ip" if origin.ip is not None else ...,
- "port" if origin.port is not None else ...,
- "unix_socket" if origin.unix_socket is not None else ...,
- "interface" if origin.interface is not None else ...,
+ def serialize(self) -> Any:
+ return str(self._value)
+
+ @classmethod
+ def json_schema(cls: Type["IPv4Address"]) -> Dict[Any, Any]:
+ return {
+ "type": "string",
}
- if present == {"ip", "port", ...}:
- return ListenType.IP_AND_PORT
- elif present == {"unix_socket", ...}:
- return ListenType.UNIX_SOCKET
- elif present == {"interface", "port", ...}:
- return ListenType.INTERFACE_AND_PORT
+
+class IPv6Address(CustomValueType):
+ def __init__(self, source_value: Any, object_path: str = "/") -> None:
+ super().__init__(source_value)
+ if isinstance(source_value, str):
+ try:
+ self._value: ipaddress.IPv6Address = ipaddress.IPv6Address(source_value)
+ except ValueError as e:
+ raise SchemaException("Failed to parse IPv6 address.", object_path) from e
else:
- raise ValueError(
- "Listen configuration contains multiple incompatible options at once. "
- "You can use (IP and PORT) or (UNIX_SOCKET) or (INTERFACE and PORT)."
+ raise SchemaException(
+ f"Unexpected value for a IPv6 address. Expected string, got '{source_value}'"
+ f" with type '{type(source_value)}'",
+ object_path,
)
- def _port(self, origin: Raw):
- if origin.port is None:
- return None
- if not 0 <= origin.port <= 65_535:
- raise ValueError(f"Port value {origin.port} out of range of usual 2-byte port value")
- return origin.port
-
- def _ip(self, origin: Raw):
- if origin.ip is None:
- return None
- # throws value error, so that get's caught outside of this function
- return ipaddress.ip_address(origin.ip)
-
- def _validate(self) -> None:
- # we already check that it's there is only one option in the `_typ` method
- pass
+ def to_std(self) -> ipaddress.IPv6Address:
+ return self._value
def __str__(self) -> str:
- if self.typ is ListenType.IP_AND_PORT:
- return f"{self.ip} @ {self.port}"
- elif self.typ is ListenType.UNIX_SOCKET:
- return f"{self.unix_socket}"
- elif self.typ is ListenType.INTERFACE_AND_PORT:
- return f"{self.interface} @ {self.port}"
- else:
- raise NotImplementedError()
+ return str(self._value)
+
+ def __int__(self) -> int:
+ raise ValueError("Can't convert IPv6 address to an integer")
def __eq__(self, o: object) -> bool:
- if not isinstance(o, Listen):
- return False
+ """
+ Two instances of IPv6Address are equal when they represent same IPv6 address as string.
+ """
+ return isinstance(o, IPv6Address) and str(o._value) == str(self._value)
- return (
- self.port == o.port
- and self.ip == o.ip
- and self.typ == o.typ
- and self.unix_socket == o.unix_socket
- and self.interface == o.interface
- )
+ def serialize(self) -> Any:
+ return str(self._value)
+
+ @classmethod
+ def json_schema(cls: Type["IPv6Address"]) -> Dict[Any, Any]:
+ return {
+ "type": "string",
+ }
- def to_dict(self) -> Dict[Any, Any]:
- if self.typ is ListenType.IP_AND_PORT:
- return {"port": self.port, "ip": str(self.ip)}
- elif self.typ is ListenType.UNIX_SOCKET:
- return {"unix_socket": str(self.unix_socket)}
- elif self.typ is ListenType.INTERFACE_AND_PORT:
- return {"interface": self.interface, "port": self.port}
- else:
- raise NotImplementedError()
+
+IPAddress = Union[IPv4Address, IPv6Address]
class IPNetwork(CustomValueType):
@classmethod
def json_schema(cls: Type["IPv6Network96"]) -> Dict[Any, Any]:
return {"type": "string"}
+
+
+class ListenType(Enum):
+ IP_AND_PORT = auto()
+ UNIX_SOCKET = auto()
+ INTERFACE_AND_PORT = auto()
+
+
+class Listen(SchemaNode, Serializable):
+ class Raw(SchemaNode):
+ ip: Optional[IPAddress] = None
+ port: Optional[int] = None
+ unix_socket: Optional[AnyPath] = None
+ interface: Optional[str] = None
+
+ _PREVIOUS_SCHEMA = Raw
+
+ typ: ListenType
+ ip: Optional[IPAddress]
+ port: Optional[int]
+ unix_socket: Optional[AnyPath]
+ interface: Optional[str]
+
+ def _typ(self, origin: Raw):
+ present = {
+ "ip" if origin.ip is not None else ...,
+ "port" if origin.port is not None else ...,
+ "unix_socket" if origin.unix_socket is not None else ...,
+ "interface" if origin.interface is not None else ...,
+ }
+
+ if present == {"ip", "port", ...}:
+ return ListenType.IP_AND_PORT
+ elif present == {"unix_socket", ...}:
+ return ListenType.UNIX_SOCKET
+ elif present == {"interface", "port", ...}:
+ return ListenType.INTERFACE_AND_PORT
+ else:
+ raise ValueError(
+ "Listen configuration contains multiple incompatible options at once. "
+ "You can use (IP and PORT) or (UNIX_SOCKET) or (INTERFACE and PORT)."
+ )
+
+ def _port(self, origin: Raw):
+ if origin.port is None:
+ return None
+ if not 0 <= origin.port <= 65_535:
+ raise ValueError(f"Port value {origin.port} out of range of usual 2-byte port value")
+ return origin.port
+
+ def _validate(self) -> None:
+ # we already check that it's there is only one option in the `_typ` method
+ pass
+
+ def __str__(self) -> str:
+ if self.typ is ListenType.IP_AND_PORT:
+ return f"{self.ip} @ {self.port}"
+ elif self.typ is ListenType.UNIX_SOCKET:
+ return f"{self.unix_socket}"
+ elif self.typ is ListenType.INTERFACE_AND_PORT:
+ return f"{self.interface} @ {self.port}"
+ else:
+ raise NotImplementedError()
+
+ def __eq__(self, o: object) -> bool:
+ if not isinstance(o, Listen):
+ return False
+
+ return (
+ self.port == o.port
+ and self.ip == o.ip
+ and self.typ == o.typ
+ and self.unix_socket == o.unix_socket
+ and self.interface == o.interface
+ )
+
+ def to_dict(self) -> Dict[Any, Any]:
+ if self.typ is ListenType.IP_AND_PORT:
+ return {"port": self.port, "ip": str(self.ip)}
+ elif self.typ is ListenType.UNIX_SOCKET:
+ return {"unix_socket": str(self.unix_socket)}
+ elif self.typ is ListenType.INTERFACE_AND_PORT:
+ return {"interface": self.interface, "port": self.port}
+ else:
+ raise NotImplementedError()
projects / thirdparty / knot-resolver.git / commitdiff
