Merge branch 'iss905' into 'master'

datamodel: file permission checks

See merge request knot/knot-resolver!1595

manager/knot_resolver_manager/constants.py

@@ -35,6 +35,14 @@ def kres_gc_executable() -> Path:
     return knot_resolver.sbin_dir / "kres-cache-gc"
 
 
+def kresd_user():
+    return None if knot_resolver is None else knot_resolver.user
+
+
+def kresd_group():
+    return None if knot_resolver is None else knot_resolver.group
+
+
 def kresd_cache_dir(config: "KresConfig") -> Path:
     return config.cache.storage.to_path()
 

manager/knot_resolver_manager/datamodel/cache_schema.py

@@ -4,16 +4,16 @@
 
 from knot_resolver_manager.datamodel.templates import template_from_str
 from knot_resolver_manager.datamodel.types import (
-    Dir,
     DNSRecordTypeEnum,
     DomainName,
     EscapedStr,
-    File,
     IntNonNegative,
     IntPositive,
     Percent,
+    ReadableFile,
     SizeUnit,
     TimeUnit,
+    WritableDir,
 )
 from knot_resolver_manager.utils.modeling import ConfigSchema
 from knot_resolver_manager.utils.modeling.base_schema import lazy_default
@@ -51,7 +51,7 @@ class PrefillSchema(ConfigSchema):
     origin: DomainName
     url: EscapedStr
     refresh_interval: TimeUnit = TimeUnit("1d")
-    ca_file: Optional[File] = None
+    ca_file: Optional[ReadableFile] = None
 
     def _validate(self) -> None:
         if str(self.origin) != ".":
@@ -125,7 +125,7 @@ class CacheSchema(ConfigSchema):
     prefetch: These options help keep the cache hot by prefetching expiring records or learning usage patterns and repetitive queries.
     """
 
-    storage: Dir = lazy_default(Dir, "/var/cache/knot-resolver")
+    storage: WritableDir = lazy_default(WritableDir, "/var/cache/knot-resolver")
     size_max: SizeUnit = SizeUnit("100M")
     garbage_collector: Union[GarbageCollectorSchema, Literal[False]] = GarbageCollectorSchema()
     ttl_min: TimeUnit = TimeUnit("5s")

manager/knot_resolver_manager/datamodel/config_schema.py

@@ -18,7 +18,7 @@
 from knot_resolver_manager.datamodel.network_schema import NetworkSchema
 from knot_resolver_manager.datamodel.options_schema import OptionsSchema
 from knot_resolver_manager.datamodel.templates import POLICY_CONFIG_TEMPLATE, WORKER_CONFIG_TEMPLATE
-from knot_resolver_manager.datamodel.types import Dir, EscapedStr, IntPositive
+from knot_resolver_manager.datamodel.types import EscapedStr, IntPositive, WritableDir
 from knot_resolver_manager.datamodel.view_schema import ViewSchema
 from knot_resolver_manager.datamodel.webmgmt_schema import WebmgmtSchema
 from knot_resolver_manager.utils.modeling import ConfigSchema
@@ -114,7 +114,7 @@ class Raw(ConfigSchema):
         version: int = 1
         nsid: Optional[EscapedStr] = None
         hostname: Optional[EscapedStr] = None
-        rundir: Dir = lazy_default(Dir, _DEFAULT_RUNDIR)
+        rundir: WritableDir = lazy_default(WritableDir, _DEFAULT_RUNDIR)
         workers: Union[Literal["auto"], IntPositive] = IntPositive(1)
         max_workers: IntPositive = IntPositive(_default_max_worker_count())
         management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": DEFAULT_MANAGER_API_SOCK})
@@ -135,7 +135,7 @@ class Raw(ConfigSchema):
 
     nsid: Optional[EscapedStr]
     hostname: EscapedStr
-    rundir: Dir
+    rundir: WritableDir
     workers: IntPositive
     max_workers: IntPositive
     management: ManagementSchema
@@ -231,12 +231,12 @@ def render_lua_policy(self) -> str:
         return POLICY_CONFIG_TEMPLATE.render(cfg=self, cwd=os.getcwd())
 
 
-def get_rundir_without_validation(data: Dict[str, Any]) -> Dir:
+def get_rundir_without_validation(data: Dict[str, Any]) -> WritableDir:
     """
     Without fully parsing, try to get a rundir from a raw config data, otherwise use default.
     Attempts a dir validation to produce a good error message.
 
     Used for initial manager startup.
     """
 
-    return Dir(data["rundir"] if "rundir" in data else _DEFAULT_RUNDIR, object_path="/rundir")
+    return WritableDir(data["rundir"] if "rundir" in data else _DEFAULT_RUNDIR, object_path="/rundir")

manager/knot_resolver_manager/datamodel/dnssec_schema.py

@@ -1,6 +1,6 @@
 from typing import List, Optional
 
-from knot_resolver_manager.datamodel.types import DomainName, EscapedStr, File, IntNonNegative, TimeUnit
+from knot_resolver_manager.datamodel.types import DomainName, EscapedStr, IntNonNegative, ReadableFile, TimeUnit
 from knot_resolver_manager.utils.modeling import ConfigSchema
 
 
@@ -14,7 +14,7 @@ class TrustAnchorFileSchema(ConfigSchema):
 
     """
 
-    file: File
+    file: ReadableFile
     read_only: bool = False
 
 

manager/knot_resolver_manager/datamodel/forward_schema.py

@@ -2,7 +2,7 @@
 
 from typing_extensions import Literal
 
-from knot_resolver_manager.datamodel.types import DomainName, File, IPAddressOptionalPort, ListOrItem, PinSha256
+from knot_resolver_manager.datamodel.types import DomainName, IPAddressOptionalPort, ListOrItem, PinSha256, ReadableFile
 from knot_resolver_manager.utils.modeling import ConfigSchema
 
 
@@ -22,7 +22,7 @@ class ForwardServerSchema(ConfigSchema):
     transport: Optional[Literal["tls"]] = None
     pin_sha256: Optional[ListOrItem[PinSha256]] = None
     hostname: Optional[DomainName] = None
-    ca_file: Optional[File] = None
+    ca_file: Optional[ReadableFile] = None
 
     def _validate(self) -> None:
         if self.pin_sha256 and (self.hostname or self.ca_file):

manager/knot_resolver_manager/datamodel/local_data_schema.py

@@ -5,10 +5,10 @@
 from knot_resolver_manager.datamodel.types import (
     DomainName,
     EscapedStr,
-    File,
     IDPattern,
     IPAddress,
     ListOrItem,
+    ReadableFile,
     TimeUnit,
 )
 from knot_resolver_manager.utils.modeling import ConfigSchema
@@ -32,7 +32,7 @@ class RuleSchema(ConfigSchema):
     name: Optional[ListOrItem[DomainName]] = None
     subtree: Optional[Literal["empty", "nxdomain", "redirect"]] = None
     address: Optional[ListOrItem[IPAddress]] = None
-    file: Optional[ListOrItem[File]] = None
+    file: Optional[ListOrItem[ReadableFile]] = None
     records: Optional[EscapedStr] = None
     tags: Optional[List[IDPattern]] = None
     ttl: Optional[TimeUnit] = None
@@ -64,7 +64,7 @@ class RPZSchema(ConfigSchema):
     tags: Tags to link with other policy rules.
     """
 
-    file: File
+    file: ReadableFile
     tags: Optional[List[IDPattern]] = None
 
 
@@ -87,9 +87,9 @@ class LocalDataSchema(ConfigSchema):
     ttl: Optional[TimeUnit] = None
     nodata: bool = True
     root_fallback_addresses: Optional[Dict[DomainName, ListOrItem[IPAddress]]] = None
-    root_fallback_addresses_files: Optional[List[File]] = None
+    root_fallback_addresses_files: Optional[List[ReadableFile]] = None
     addresses: Optional[Dict[DomainName, ListOrItem[IPAddress]]] = None
-    addresses_files: Optional[List[File]] = None
+    addresses_files: Optional[List[ReadableFile]] = None
     records: Optional[EscapedStr] = None
     rules: Optional[List[RuleSchema]] = None
     rpz: Optional[List[RPZSchema]] = None

manager/knot_resolver_manager/datamodel/logging_schema.py

@@ -3,7 +3,7 @@
 
 from typing_extensions import Literal
 
-from knot_resolver_manager.datamodel.types import FilePath, TimeUnit
+from knot_resolver_manager.datamodel.types import TimeUnit, WritableFilePath
 from knot_resolver_manager.utils.modeling import ConfigSchema
 from knot_resolver_manager.utils.modeling.base_schema import is_obj_type_valid
 
@@ -84,7 +84,7 @@ class DnstapSchema(ConfigSchema):
     log_tcp_rtt: Log TCP RTT (Round-trip time).
     """
 
-    unix_socket: FilePath
+    unix_socket: WritableFilePath
     log_queries: bool = True
     log_responses: bool = True
     log_tcp_rtt: bool = True

manager/knot_resolver_manager/datamodel/lua_schema.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from knot_resolver_manager.datamodel.types import File
+from knot_resolver_manager.datamodel.types import ReadableFile
 from knot_resolver_manager.utils.modeling import ConfigSchema
 
 
@@ -16,7 +16,7 @@ class LuaSchema(ConfigSchema):
 
     script_only: bool = False
     script: Optional[str] = None
-    script_file: Optional[File] = None
+    script_file: Optional[ReadableFile] = None
 
     def _validate(self) -> None:
         if self.script and self.script_file:

manager/knot_resolver_manager/datamodel/management_schema.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from knot_resolver_manager.datamodel.types import FilePath, IPAddressPort
+from knot_resolver_manager.datamodel.types import WritableFilePath, IPAddressPort
 from knot_resolver_manager.utils.modeling import ConfigSchema
 
 
@@ -13,7 +13,7 @@ class ManagementSchema(ConfigSchema):
     interface: IP address and port number to listen to.
     """
 
-    unix_socket: Optional[FilePath] = None
+    unix_socket: Optional[WritableFilePath] = None
     interface: Optional[IPAddressPort] = None
 
     def _validate(self) -> None:

manager/knot_resolver_manager/datamodel/network_schema.py

@@ -4,8 +4,7 @@
 
 from knot_resolver_manager.datamodel.types import (
     EscapedStr32B,
-    File,
-    FilePath,
+    WritableFilePath,
     Int0_512,
     Int0_65535,
     InterfaceOptionalPort,
@@ -16,6 +15,7 @@
     IPv6Address,
     ListOrItem,
     PortNumber,
+    ReadableFile,
     SizeUnit,
 )
 from knot_resolver_manager.utils.modeling import ConfigSchema
@@ -62,10 +62,10 @@ class TLSSchema(ConfigSchema):
     padding: EDNS(0) padding of queries and answers sent over an encrypted channel.
     """
 
-    cert_file: Optional[File] = None
-    key_file: Optional[File] = None
+    cert_file: Optional[ReadableFile] = None
+    key_file: Optional[ReadableFile] = None
     sticket_secret: Optional[EscapedStr32B] = None
-    sticket_secret_file: Optional[File] = None
+    sticket_secret_file: Optional[ReadableFile] = None
     auto_discovery: bool = False
     padding: Union[bool, Int0_512] = True
 
@@ -88,15 +88,15 @@ class Raw(ConfigSchema):
         """
 
         interface: Optional[ListOrItem[InterfaceOptionalPort]] = None
-        unix_socket: Optional[ListOrItem[FilePath]] = 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[FilePath]]
+    unix_socket: Optional[ListOrItem[WritableFilePath]]
     port: Optional[PortNumber]
     kind: KindEnum
     freebind: bool

manager/knot_resolver_manager/datamodel/rpz_schema.py

@@ -1,6 +1,6 @@
 from typing import List, Optional
 
-from knot_resolver_manager.datamodel.types import File, PolicyActionEnum, PolicyFlagEnum
+from knot_resolver_manager.datamodel.types import PolicyActionEnum, PolicyFlagEnum, ReadableFile
 from knot_resolver_manager.utils.modeling import ConfigSchema
 
 
@@ -18,7 +18,7 @@ class RPZSchema(ConfigSchema):
     """
 
     action: PolicyActionEnum
-    file: File
+    file: ReadableFile
     watch: bool = True
     views: Optional[List[str]] = None
     options: Optional[List[PolicyFlagEnum]] = None

manager/knot_resolver_manager/datamodel/static_hints_schema.py

@@ -1,6 +1,6 @@
 from typing import Dict, List, Optional
 
-from knot_resolver_manager.datamodel.types import DomainName, File, IPAddress, TimeUnit
+from knot_resolver_manager.datamodel.types import DomainName, IPAddress, ReadableFile, TimeUnit
 from knot_resolver_manager.utils.modeling import ConfigSchema
 
 
@@ -22,6 +22,6 @@ class StaticHintsSchema(ConfigSchema):
     nodata: bool = True
     etc_hosts: bool = False
     root_hints: Optional[Dict[DomainName, List[IPAddress]]] = None
-    root_hints_file: Optional[File] = None
+    root_hints_file: Optional[ReadableFile] = None
     hints: Optional[Dict[DomainName, List[IPAddress]]] = None
-    hints_files: Optional[List[File]] = None
+    hints_files: Optional[List[ReadableFile]] = None

manager/knot_resolver_manager/datamodel/types/__init__.py

@@ -1,5 +1,5 @@
 from .enums import DNSRecordTypeEnum, PolicyActionEnum, PolicyFlagEnum
-from .files import AbsoluteDir, Dir, File, FilePath
+from .files import AbsoluteDir, Dir, File, FilePath, ReadableFile, WritableDir, WritableFilePath
 from .generic_types import ListOrItem
 from .types import (
     DomainName,
@@ -60,6 +60,9 @@
     "SizeUnit",
     "TimeUnit",
     "AbsoluteDir",
+    "ReadableFile",
+    "WritableDir",
+    "WritableFilePath",
     "File",
     "FilePath",
     "Dir",