"default": 256
},
"management": {
- "description": "Configuration of management HTTP API. By default, unix-socket is located in 'rundir'.",
+ "description": "Configuration of management HTTP API.",
"type": "object",
"properties": {
"unix-socket": {
}
},
"default": {
- "unix_socket": "kres-api.sock",
+ "unix_socket": "/run/knot-resolver/kres-api.sock",
"interface": null
}
},
from typing import Dict, List, Optional, Set, Tuple, Type, TypeVar
from urllib.parse import quote
-from knot_resolver.constants import API_SOCK_FILE, API_SOCK_NAME, CONFIG_FILE, RUN_DIR
+from knot_resolver.constants import API_SOCK_FILE, CONFIG_FILE
from knot_resolver.datamodel.types import IPAddressPort
from knot_resolver.utils.modeling import parsing
from knot_resolver.utils.modeling.exceptions import DataValidationError
try:
with open(config, "r", encoding="utf8") as f:
data = parsing.try_to_parse(f.read())
-
- rkey = "rundir"
- rundir = Path(data[rkey]) if rkey in data else RUN_DIR
-
mkey = "management"
if mkey in data:
management = data[mkey]
-
- ikey = "interface"
- if ikey in data[mkey]:
- ip = IPAddressPort(data[mkey][ikey], object_path=f"/{mkey}/{ikey}")
+ if "unix-socket" in management:
return SocketDesc(
- f"http://{ip.addr}:{ip.port}",
- f'Key "/management/interface" in "{config}" file',
+ f'http+unix://{quote(management["unix-socket"], safe="")}/',
+ f'Key "/management/unix-socket" in "{config}" file',
)
-
- skey = "unix-socket"
- if skey in management:
- socket = Path(management[skey])
- if not socket.is_absolute():
- socket = rundir / socket
+ if "interface" in management:
+ ip = IPAddressPort(management["interface"], object_path=f"/{mkey}/interface")
return SocketDesc(
- f'http+unix://{quote(str(socket), safe="")}/',
- f'Key "/management/unix-socket" in "{config}" file',
+ f"http://{ip.addr}:{ip.port}",
+ f'Key "/management/interface" in "{config}" file',
)
-
- socket = rundir / API_SOCK_NAME
- return SocketDesc(
- f'http+unix://{quote(str(socket), safe="")}/',
- f'Key "/rundir" in "{config}" file',
- )
-
+ return None
except ValueError as e:
raise DataValidationError(*e.args) from e # pylint: disable=no-value-for-parameter
except OSError as e:
import logging
import os
import socket
-from pathlib import Path
from typing import Any, Dict, List, Literal, Optional, Tuple, Union
-from knot_resolver.constants import API_SOCK_NAME, RUN_DIR, VERSION
+from knot_resolver.constants import API_SOCK_FILE, RUN_DIR, VERSION
from knot_resolver.datamodel.cache_schema import CacheSchema
from knot_resolver.datamodel.defer_schema import DeferSchema
from knot_resolver.datamodel.dns64_schema import Dns64Schema
rundir: Directory where the resolver can create files and which will be it's cwd.
workers: The number of running kresd (Knot Resolver daemon) workers. If set to 'auto', it is equal to number of CPUs available.
max_workers: The maximum number of workers allowed. Cannot be changed in runtime.
- management: Configuration of management HTTP API. By default, unix-socket is located in 'rundir'.
+ management: Configuration of management HTTP API.
webmgmt: Configuration of legacy web management endpoint.
options: Fine-tuning global parameters of DNS resolver operation.
network: Network connections and protocols configuration.
rundir: WritableDir = lazy_default(WritableDir, str(RUN_DIR))
workers: Union[Literal["auto"], IntPositive] = IntPositive(1)
max_workers: IntPositive = IntPositive(WORKERS_MAX)
- management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": str(API_SOCK_NAME)})
+ management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": str(API_SOCK_FILE)})
webmgmt: Optional[WebmgmtSchema] = None
options: OptionsSchema = OptionsSchema()
network: NetworkSchema = NetworkSchema()
)
return obj.workers
- def _management(self, obj: Raw) -> Any:
- if obj.management.unix_socket:
- soc = Path(obj.management.unix_socket.serialize())
- if soc.is_absolute():
- return obj.management
- return ManagementSchema({"unix-socket": str(obj.rundir.to_path() / soc)})
- return obj.management
-
def _dnssec(self, obj: Raw) -> Any:
if obj.dnssec is True:
return DnssecSchema()
"""
context = get_global_validation_context()
- set_global_validation_context(Context(RUN_DIR, False))
+ set_global_validation_context(Context(None, False))
schema = KresConfig.json_schema(
schema_id=f"https://www.knot-resolver.cz/documentation/v{VERSION}/_static/config.schema.json",
import inspect
import json
-import os
from typing import Any, Dict, Type, cast
-from knot_resolver.constants import API_SOCK_FILE, API_SOCK_NAME, RUN_DIR
from knot_resolver.datamodel import KresConfig
from knot_resolver.datamodel.lua_schema import LuaSchema
from knot_resolver.utils.modeling import BaseSchema
def test_config_defaults():
config = KresConfig()
- # Management API default
- assert config.management.unix_socket.to_path() == API_SOCK_FILE
-
# DNS64 default
assert config.dns64 == False
-def test_management_unix_socket():
- cwd = os.getcwd()
- config = KresConfig({"rundir": cwd})
- assert str(config.management.unix_socket) == f"{cwd}/{API_SOCK_NAME}"
-
- my_soc = "my-new.soc"
- config = KresConfig({"management": {"unix-socket": my_soc}})
- assert str(config.management.unix_socket) == f"{RUN_DIR}/{my_soc}"
-
-
-def test_management_interface():
- cwd = os.getcwd()
- config = KresConfig({"rundir": cwd, "management": {"interface": "127.0.0.1@5000"}})
-
- assert config.management.unix_socket == None
- assert str(config.management.interface) == "127.0.0.1@5000"
-
-
def test_dnssec_false():
config = KresConfig({"dnssec": False})
projects / thirdparty / knot-resolver.git / commitdiff
