From: Aleš Mrázek Date: Mon, 2 Mar 2026 13:21:39 +0000 (+0100) Subject: python: modeling: types: added path types X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=7bae1980af0714f79680bffe43650cb43a71d753;p=thirdparty%2Fknot-resolver.git python: modeling: types: added path types --- diff --git a/pyproject.toml b/pyproject.toml index a7ff9ed3b..995d8b8f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -196,7 +196,8 @@ ignore = [ "TCH003", # *Move standard library import into a type-checking block ] exclude = [ - "tests/python/knot_resolver*" + "tests/python/knot_resolver*", + "tests/python/knot_resolver/utils/modeling/types/path_testing*", ] [tool.ruff.lint.isort] diff --git a/python/knot_resolver/utils/modeling/types/__init__.py b/python/knot_resolver/utils/modeling/types/__init__.py index eaea0e935..25737dfbe 100644 --- a/python/knot_resolver/utils/modeling/types/__init__.py +++ b/python/knot_resolver/utils/modeling/types/__init__.py @@ -9,8 +9,19 @@ from .integer_types import ( Percent, PortNumber, ) +from .path_types import ( + Directory, + File, + FilePath, + ReadableFile, + WritableDirectory, + WritableFilePath, +) __all__ = [ + "Directory", + "File", + "FilePath", "FloatNonNegative", "Integer0_32", "Integer0_512", @@ -21,4 +32,7 @@ __all__ = [ "Transformed", "Percent", "PortNumber", + "ReadableFile", + "WritableDirectory", + "WritableFilePath", ] diff --git a/python/knot_resolver/utils/modeling/types/path_types.py b/python/knot_resolver/utils/modeling/types/path_types.py new file mode 100644 index 000000000..2a1c4cb3c --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/path_types.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import os +import stat +from enum import Flag, auto +from grp import getgrnam +from pwd import getpwnam, getpwuid +from typing import TYPE_CHECKING + +from knot_resolver.utils.modeling.context import Context, Strictness +from knot_resolver.utils.modeling.errors import DataValidationError, DataValueError + +from .base_path_types import BasePath + +if TYPE_CHECKING: + from pathlib import Path + + +class Directory(BasePath): + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness > Strictness.NORMAL: + try: + path = self._path_absolute + if not path.is_dir(): + msg = f"path '{path}' does not point to an existing directory" + raise DataValueError(msg) + except PermissionError as e: + msg = f"insufficient permissions for '{e.filename}'" + raise DataValidationError(msg, self._tree_path) from None + + +class File(BasePath): + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness > Strictness.BASIC: + try: + path = self._path_absolute + if context.strictness > Strictness.NORMAL and not path.is_file(): + msg = f"path '{path}' does not point to an existing file" + raise DataValueError(msg) + except PermissionError as e: + msg = f"insufficient permissions for '{e.filename}'" + raise DataValidationError(msg, self._tree_path) from None + + +class FilePath(BasePath): + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness > Strictness.BASIC: + try: + path = self._path_absolute + if context.strictness > Strictness.NORMAL: + parent = path.parent + if not parent.is_dir(): + msg = f"parent '{parent}' does not point an existing directory" + raise DataValueError(msg) + if path.is_dir(): + msg = f"path '{path}' points to a directory when we expected a file" + raise DataValueError(msg) + except PermissionError as e: + msg = f"insufficient permissions for '{e.filename}'" + raise DataValidationError(msg, self._tree_path) from None + + +class _PermissionMode(Flag): + READ = auto() + WRITE = auto() + EXECUTE = auto() + + +def _check_path_permission(context: Context, dest_path: Path, perm_mode: _PermissionMode) -> bool: + chflags = { + _PermissionMode.READ: [stat.S_IRUSR, stat.S_IRGRP, stat.S_IROTH], + _PermissionMode.WRITE: [stat.S_IWUSR, stat.S_IWGRP, stat.S_IWOTH], + _PermissionMode.EXECUTE: [stat.S_IXUSR, stat.S_IXGRP, stat.S_IXOTH], + } + + # running outside (client, ...) + if context.username and context.groupname: + user_uid = getpwnam(context.username).pw_uid + user_gid = getgrnam(context.groupname).gr_gid + username = context.username + # running under root privileges + elif os.geteuid() == 0: + return True + # running normally under an unprivileged user + else: + user_uid = os.getuid() + user_gid = os.getgid() + username = getpwuid(user_uid).pw_name + + try: + dest_stat = dest_path.stat() + except PermissionError: + return False + + dest_uid = dest_stat.st_uid + dest_gid = dest_stat.st_gid + dest_mode = dest_stat.st_mode + + def accessible(perm: _PermissionMode) -> bool: + if user_uid == dest_uid: + return bool(dest_mode & chflags[perm][0]) + b_groups = os.getgrouplist(username, user_gid) + if user_gid == dest_gid or dest_gid in b_groups: + return bool(dest_mode & chflags[perm][1]) + return bool(dest_mode & chflags[perm][2]) + + # __iter__ for class enum.Flag added in python3.11 + # 'for perm in perm_mode:' fails for <=python3.11 + return all(not (perm in perm_mode and not accessible(perm)) for perm in _PermissionMode) + + +class ReadableFile(File): + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness is Strictness.STRICT: + path = self._path_absolute + msg = f"insufficient permissions to read '{path}'" + if not (_check_path_permission(context, path, _PermissionMode.READ) or os.access(self._value, os.R_OK)): + raise DataValidationError(msg, self._tree_path) + + +class WritableDirectory(Directory): + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness is Strictness.STRICT: + path = self._path_absolute + if not ( + _check_path_permission(context, path, _PermissionMode.WRITE | _PermissionMode.EXECUTE) + and os.access(path.parent, os.W_OK | os.X_OK) + ): + msg = f"insufficient permissions to write/execute '{path}'" + raise DataValidationError(msg, self._tree_path) + + +class WritableFilePath(FilePath): + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness is Strictness.STRICT: + path = self._path_absolute + # check that parent dir is writable + if not ( + _check_path_permission(context, path.parent, _PermissionMode.WRITE | _PermissionMode.EXECUTE) + or os.access(path.parent, os.W_OK | os.X_OK) + ): + msg = f"insufficient permissions to write/execute '{path.parent}'" + raise DataValidationError(msg, self._tree_path) + # check that existing file is writable + if path.exists() and not ( + _check_path_permission(context, path, _PermissionMode.WRITE | _PermissionMode.EXECUTE) + or os.access(path, os.W_OK) + ): + msg = f"insufficient permissions to write/execute '{path}'" + raise DataValidationError(msg, self._tree_path) diff --git a/scripts/poe-tasks/fix-format b/scripts/poe-tasks/fix-format index 23c9f4f75..fecd543a7 100755 --- a/scripts/poe-tasks/fix-format +++ b/scripts/poe-tasks/fix-format @@ -14,7 +14,7 @@ echo # format python code echo -e "${yellow}Formatting Python code using ruff...${reset}" -ruff format $dirs +ruff format --exclude=tests/python/knot_resolver/utils/modeling/types/path_testing $dirs check_rv $? echo diff --git a/scripts/poe-tasks/test-unit b/scripts/poe-tasks/test-unit index 9560374e0..7519f41d0 100755 --- a/scripts/poe-tasks/test-unit +++ b/scripts/poe-tasks/test-unit @@ -4,5 +4,16 @@ src_dir="$(dirname "$(realpath "$0")")" source $src_dir/utils/_env.sh +test_dir="tests/python/knot_resolver" + +# we cannot test path types permissions if running under root +if [ "$(id -u)" -eq 0 ]; then + ignore="$test_dir/utils/modeling/types/test_path_types.py" +else + # prepare permissions + bash $test_dir/utils/modeling/types/prepare_path_types_testing + ignore="$test_dir/utils/modeling/types/path_types_testing" +fi + # run pytest -env PYTHONPATH=. pytest --junitxml=unit.junit.xml --cov=python/knot_resolver --show-capture=all tests/python/knot_resolver +env PYTHONPATH=. pytest --junitxml=unit.junit.xml --cov=python/knot_resolver --show-capture=all --ignore=$ignore $test_dir diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/readable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/readable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/readable_dir/readable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/readable_dir/readable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unreadable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unreadable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unreadable_dir/unreadable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unreadable_dir/unreadable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unwritable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unwritable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unwritable_dir/unwritable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/unwritable_dir/unwritable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/writable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/writable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/path_types_testing/writable_dir/writable.file b/tests/python/knot_resolver/utils/modeling/types/path_types_testing/writable_dir/writable.file new file mode 100755 index 000000000..e69de29bb diff --git a/tests/python/knot_resolver/utils/modeling/types/prepare_path_types_testing b/tests/python/knot_resolver/utils/modeling/types/prepare_path_types_testing new file mode 100755 index 000000000..bee4d92d7 --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/prepare_path_types_testing @@ -0,0 +1,18 @@ +#!/usr/bin/env bash + +# ensure consistent behaviour +src_dir="$(dirname "$(realpath "$0")")" +cd $src_dir + +testing_dir=path_types_testing + +# default +chmod -R 755 $testing_dir +# readable +chmod 555 $testing_dir/readable* +# unreadable +chmod 000 $testing_dir/unreadable* +# writable +chmod 333 $testing_dir/unwritable* +# unwritable +chmod 555 $testing_dir/unwritable* diff --git a/tests/python/knot_resolver/utils/modeling/types/test_path_types.py b/tests/python/knot_resolver/utils/modeling/types/test_path_types.py new file mode 100644 index 000000000..2c28b8d80 --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/test_path_types.py @@ -0,0 +1,153 @@ +from pathlib import Path +from typing import Any + +import pytest + +from knot_resolver.utils.modeling.context import Context, Strictness +from knot_resolver.utils.modeling.errors import DataModelingError +from knot_resolver.utils.modeling.types.path_types import ( + Directory, + File, + FilePath, + ReadableFile, + WritableDirectory, + WritableFilePath, +) + +context_default = Context(strictness=Strictness.STRICT) +base_path = Path(__file__).parent / "path_types_testing" + +readable_dirs = [ + "readable_dir", # relative + str(base_path) + "/readable_dir", # absolute +] + +unreadable_dirs = [ + "unreadable_dir", + str(base_path) + "/unreadable_dir", +] + +writable_dirs = [ + "writable_dir", + str(base_path) + "/writable_dir", +] + +unwritable_dirs = [ + "unwritable_dir", + str(base_path) + "/unwritable_dir", +] + +readable_files = [ + "readable.file", + str(base_path) + "/readable.file", +] + +unreadable_files = [ + "unreadable.file", + str(base_path) + "/unreadable.file", +] + +writable_files = [ + "writable.file", + str(base_path) + "/writable.file", +] + +unwritable_files = [ + "unwritable.file", + str(base_path) + "/unwritable.file", +] + +nonexisting_files = [ + "nonexisting.file", + str(base_path) + "/nonexisting.file", +] + + +@pytest.mark.parametrize("value", readable_dirs + writable_dirs) +def test_directory(value: str): + obj = Directory(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", readable_files + writable_files) +def test_directory_invalid(value: Any): + obj = Directory(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default) + + +@pytest.mark.parametrize("value", readable_files + writable_files) +def test_file(value: str): + obj = File(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", readable_dirs + writable_dirs) +def test_file_invalid(value: Any): + obj = File(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default) + + +@pytest.mark.parametrize("value", readable_files + writable_files + nonexisting_files) +def test_filepath(value: str): + obj = FilePath(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", readable_dirs + writable_dirs) +def test_filepath_invalid(value: Any): + obj = File(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default) + + +@pytest.mark.parametrize("value", readable_files) +def test_readablefile(value: str): + obj = ReadableFile(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", unreadable_files + readable_dirs + writable_dirs) +def test_readablefile_invalid(value: Any): + obj = ReadableFile(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default) + + +@pytest.mark.parametrize("value", writable_dirs) +def test_writabledirectory(value: str): + obj = WritableDirectory(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", readable_files + writable_files + readable_dirs + unwritable_dirs) +def test_writabledirectory_invalid(value: Any): + obj = WritableDirectory(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default) + + +@pytest.mark.parametrize("value", writable_files + nonexisting_files) +def test_writablefilepath(value: str): + obj = WritableFilePath(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", readable_files + readable_dirs + writable_dirs) +def test_writablefilepath_invalid(value: Any): + obj = WritableFilePath(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default)