File: network_schema.py

package info (click to toggle)
knot-resolver 6.0.17-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 16,376 kB
  • sloc: javascript: 42,732; ansic: 40,311; python: 12,580; cpp: 2,121; sh: 1,988; xml: 193; makefile: 181
file content (203 lines) | stat: -rw-r--r-- 7,736 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
from typing import Any, List, Literal, Optional, Union

from knot_resolver.constants import WATCHDOG_LIB
from knot_resolver.datamodel.types import (
    EscapedStr32B,
    Int0_512,
    Int0_65535,
    InterfaceOptionalPort,
    IPAddress,
    IPAddressEM,
    IPNetwork,
    IPv4Address,
    IPv6Address,
    ListOrItem,
    PortNumber,
    ReadableFile,
    SizeUnit,
    WritableFilePath,
)
from knot_resolver.utils.modeling import ConfigSchema

KindEnum = Literal["dns", "xdp", "dot", "doh-legacy", "doh2"]


class EdnsBufferSizeSchema(ConfigSchema):
    """
    EDNS payload size advertised in DNS packets.

    ---
    upstream: Maximum EDNS upstream (towards other DNS servers) payload size.
    downstream: Maximum EDNS downstream (towards clients) payload size for communication.
    """

    upstream: SizeUnit = SizeUnit("1232B")
    downstream: SizeUnit = SizeUnit("1232B")


class AddressRenumberingSchema(ConfigSchema):
    """
    Renumbers addresses in answers to different address space.

    ---
    source: Source subnet.
    destination: Destination address prefix.
    """

    source: IPNetwork
    destination: Union[IPAddressEM, IPAddress]


class TLSSchema(ConfigSchema):
    class Raw(ConfigSchema):
        """
        TLS configuration, also affects DNS over TLS and DNS over HTTPS.

        ---
        watchdog: Enables watchdog of changes in TLS certificate files. Requires the optional 'watchdog' dependency.
        cert_file: Path to certificate file.
        key_file: Path to certificate key file.
        sticket_secret: Secret for TLS session resumption via tickets. (RFC 5077).
        sticket_secret_file: Path to file with secret for TLS session resumption via tickets. (RFC 5077).
        padding: EDNS(0) padding of queries and answers sent over an encrypted channel.
        """

        watchdog: Union[Literal["auto"], bool] = "auto"
        cert_file: Optional[ReadableFile] = None
        key_file: Optional[ReadableFile] = None
        sticket_secret: Optional[EscapedStr32B] = None
        sticket_secret_file: Optional[ReadableFile] = None
        padding: Union[bool, Int0_512] = True

    _LAYER = Raw

    watchdog: bool
    cert_file: Optional[ReadableFile] = None
    key_file: Optional[ReadableFile] = None
    sticket_secret: Optional[EscapedStr32B] = None
    sticket_secret_file: Optional[ReadableFile] = None
    padding: Union[bool, Int0_512] = True

    def _watchdog(self, obj: Raw) -> Any:
        if obj.watchdog == "auto":
            return WATCHDOG_LIB
        return obj.watchdog

    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 bool(self.cert_file) != bool(self.key_file):
            raise ValueError("'cert-file' and 'key-file' must be configured together")
        if self.cert_file and self.key_file and self.watchdog and not WATCHDOG_LIB:
            raise ValueError(
                "'files-watchdog' is enabled, but the required 'watchdog' dependency (optional) is not installed"
            )


class ListenSchema(ConfigSchema):
    class Raw(ConfigSchema):
        """
        Configuration of listening interface.

        ---
        unix_socket: Path to unix domain socket to listen to.
        interface: IP address or interface name with optional port number to listen to.
        port: Port number to listen to.
        kind: Specifies DNS query transport protocol.
        freebind: Used for binding to non-local address.
        """

        interface: Optional[ListOrItem[InterfaceOptionalPort]] = None
        unix_socket: Optional[ListOrItem[WritableFilePath]] = None
        port: Optional[PortNumber] = None
        kind: KindEnum = "dns"
        freebind: bool = False

    _LAYER = Raw

    interface: Optional[ListOrItem[InterfaceOptionalPort]]
    unix_socket: Optional[ListOrItem[WritableFilePath]]
    port: Optional[PortNumber]
    kind: KindEnum
    freebind: bool

    def _interface(self, origin: Raw) -> Optional[ListOrItem[InterfaceOptionalPort]]:
        if origin.interface:
            port_set: Optional[bool] = None
            for intrfc in origin.interface:  # type: ignore[attr-defined]
                if origin.port and intrfc.port:
                    raise ValueError("The port number is defined in two places ('port' option and '@<port>' syntax).")
                if port_set is not None and (bool(intrfc.port) != port_set):
                    raise ValueError(
                        "The '@<port>' syntax must be used either for all or none of the interface in the list."
                    )
                port_set = bool(intrfc.port)
        return origin.interface

    def _port(self, origin: Raw) -> Optional[PortNumber]:
        if origin.port:
            return origin.port
        # default port number based on kind
        if origin.interface:
            if origin.kind == "dot":
                return PortNumber(853)
            if origin.kind in ["doh-legacy", "doh2"]:
                return PortNumber(443)
            return PortNumber(53)
        return None

    def _validate(self) -> None:
        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 'interface' option."
            )


class ProxyProtocolSchema(ConfigSchema):
    """
    PROXYv2 protocol configuration.

    ---
    enable: Enable/disable PROXYv2 protocol.
    allow: Allow usage of the PROXYv2 protocol headers by clients on the specified addresses.
    """

    enable: bool = False
    allow: Optional[List[Union[IPAddress, IPNetwork]]] = None


class NetworkSchema(ConfigSchema):
    """
    Network connections and protocols configuration.

    ---
    do_ipv4: Enable/disable using IPv4 for contacting upstream nameservers.
    do_ipv6: Enable/disable using IPv6 for contacting upstream nameservers.
    out_interface_v4: IPv4 address used to perform queries. Not set by default, which lets the OS choose any address.
    out_interface_v6: IPv6 address used to perform queries. Not set by default, which lets the OS choose any address.
    tcp_pipeline: TCP pipeline limit. The number of outstanding queries that a single client connection can make in parallel.
    edns_tcp_keepalive: Allows clients to discover the connection timeout. (RFC 7828)
    edns_buffer_size: Maximum EDNS payload size advertised in DNS packets. Different values can be configured for communication downstream (towards clients) and upstream (towards other DNS servers).
    address_renumbering: Renumbers addresses in answers to different address space.
    tls: TLS configuration, also affects DNS over TLS and DNS over HTTPS.
    proxy_protocol: PROXYv2 protocol configuration.
    listen: List of interfaces to listen to and its configuration.
    """

    do_ipv4: bool = True
    do_ipv6: bool = True
    out_interface_v4: Optional[IPv4Address] = None
    out_interface_v6: Optional[IPv6Address] = None
    tcp_pipeline: Int0_65535 = Int0_65535(100)
    edns_tcp_keepalive: bool = True
    edns_buffer_size: EdnsBufferSizeSchema = EdnsBufferSizeSchema()
    address_renumbering: Optional[List[AddressRenumberingSchema]] = None
    tls: TLSSchema = TLSSchema()
    proxy_protocol: ProxyProtocolSchema = ProxyProtocolSchema()
    listen: List[ListenSchema] = [
        ListenSchema({"interface": "127.0.0.1"}),
        ListenSchema({"interface": "::1", "freebind": True}),
    ]