Date: Thu, 9 Feb 2023 14:50:48 GMT From: "Alexander V. Chernikov" <melifaro@FreeBSD.org> To: src-committers@FreeBSD.org, dev-commits-src-all@FreeBSD.org, dev-commits-src-main@FreeBSD.org Subject: git: 6332ef894199 - main - testing: handling non-root users with VNETs in pytest-based tests. Message-ID: <202302091450.319Eomiq088008@gitrepo.freebsd.org>
next in thread | raw e-mail | index | archive | help
The branch main has been updated by melifaro: URL: https://cgit.FreeBSD.org/src/commit/?id=6332ef8941999b0c074d1ece0e1e108447c70b98 commit 6332ef8941999b0c074d1ece0e1e108447c70b98 Author: Alexander V. Chernikov <melifaro@FreeBSD.org> AuthorDate: 2023-02-09 14:31:34 +0000 Commit: Alexander V. Chernikov <melifaro@FreeBSD.org> CommitDate: 2023-02-09 14:49:27 +0000 testing: handling non-root users with VNETs in pytest-based tests. Currently isolation and resource requirements are handled directly by the kyua runner, based on the requirements specified by the test. It works well for simple tests, but may cause discrepancy with tests doing complex pre-setups. For example, all tests that perform VNET setups require root access to properly function. This change adds additional handling of the "require_user" property within the python testing framework. Specifically, it requests root access if the test class signals its root requirements and drops privileges to the desired user after performing the pre-setup. Differential Revision: https://reviews.freebsd.org/D37923 MFC after: 2 weeks --- tests/atf_python/atf_pytest.py | 54 ++++++++++++++++++++++++++++++++++++---- tests/atf_python/sys/net/vnet.py | 3 +++ tests/atf_python/utils.py | 23 ++++++++++++++++- tests/conftest.py | 26 +++++++++++++------ 4 files changed, 92 insertions(+), 14 deletions(-) diff --git a/tests/atf_python/atf_pytest.py b/tests/atf_python/atf_pytest.py index d530c7b4515c..db7244d3234b 100644 --- a/tests/atf_python/atf_pytest.py +++ b/tests/atf_python/atf_pytest.py @@ -3,6 +3,7 @@ from typing import Any from typing import Dict from typing import List from typing import NamedTuple +from typing import Optional from typing import Tuple import pytest @@ -51,10 +52,32 @@ class ATFTestObj(object): return line return obj.name + @staticmethod + def _convert_user_mark(mark, obj, ret: Dict): + username = mark.args[0] + if username == "unprivileged": + # Special unprivileged user requested. + # First, require the unprivileged-user config option presence + key = "require.config" + if key not in ret: + ret[key] = "unprivileged_user" + else: + ret[key] = "{} {}".format(ret[key], "unprivileged_user") + # Check if the framework requires root + test_cls = ATFHandler.get_test_class(obj) + if test_cls and getattr(test_cls, "NEED_ROOT", False): + # Yes, so we ask kyua to run us under root instead + # It is up to the implementation to switch back to the desired + # user + ret["require.user"] = "root" + else: + ret["require.user"] = username + + def _convert_marks(self, obj) -> Dict[str, Any]: wj_func = lambda x: " ".join(x) # noqa: E731 _map: Dict[str, Dict] = { - "require_user": {"name": "require.user"}, + "require_user": {"handler": self._convert_user_mark}, "require_arch": {"name": "require.arch", "fmt": wj_func}, "require_diskspace": {"name": "require.diskspace"}, "require_files": {"name": "require.files", "fmt": wj_func}, @@ -66,6 +89,9 @@ class ATFTestObj(object): ret = {} for mark in obj.iter_markers(): if mark.name in _map: + if "handler" in _map[mark.name]: + _map[mark.name]["handler"](mark, obj, ret) + continue name = _map[mark.name].get("name", mark.name) if "fmt" in _map[mark.name]: val = _map[mark.name]["fmt"](mark.args[0]) @@ -91,8 +117,24 @@ class ATFHandler(object): state: str reason: str - def __init__(self): + def __init__(self, report_file_name: Optional[str]): self._tests_state_map: Dict[str, ReportStatus] = {} + self._report_file_name = report_file_name + self._report_file_handle = None + + def setup_configure(self): + fname = self._report_file_name + if fname: + self._report_file_handle = open(fname, mode="w") + + def setup_method_pre(self, item): + """Called before actually running the test setup_method""" + # Check if we need to manually drop the privileges + for mark in item.iter_markers(): + if mark.name == "require_user": + cls = self.get_test_class(item) + cls.TARGET_USER = mark.args[0] + break def override_runtest(self, obj): # Override basic runtest command @@ -220,7 +262,9 @@ class ATFHandler(object): # global failure self.set_report_state(test_name, state, reason) - def write_report(self, path): + def write_report(self): + if self._report_file_handle is None: + return if self._tests_state_map: # If we're executing in ATF mode, there has to be just one test # Anyway, deterministically pick the first one @@ -230,8 +274,8 @@ class ATFHandler(object): line = test.state else: line = "{}: {}".format(test.state, test.reason) - with open(path, mode="w") as f: - print(line, file=f) + print(line, file=self._report_file_handle) + self._report_file_handle.close() @staticmethod def get_atf_vars() -> Dict[str, str]: diff --git a/tests/atf_python/sys/net/vnet.py b/tests/atf_python/sys/net/vnet.py index 1f61269ffe6c..c0e0a24f6687 100644 --- a/tests/atf_python/sys/net/vnet.py +++ b/tests/atf_python/sys/net/vnet.py @@ -329,6 +329,7 @@ class ObjectsMap(NamedTuple): class VnetTestTemplate(BaseTest): + NEED_ROOT: bool = True TOPOLOGY = {} def _get_vnet_handler(self, vnet_alias: str): @@ -374,6 +375,7 @@ class VnetTestTemplate(BaseTest): # Do unbuffered stdout for children # so the logs are present if the child hangs sys.stdout.reconfigure(line_buffering=True) + self.drop_privileges() handler(vnet) def setup_topology(self, topo: Dict, topology_id: str): @@ -465,6 +467,7 @@ class VnetTestTemplate(BaseTest): # Save state for the main handler self.iface_map = obj_map.iface_map self.vnet_map = obj_map.vnet_map + self.drop_privileges() def cleanup(self, test_id: str): # pytest test id: file::class::test_name diff --git a/tests/atf_python/utils.py b/tests/atf_python/utils.py index fddfadac9a56..c8146b943ce9 100644 --- a/tests/atf_python/utils.py +++ b/tests/atf_python/utils.py @@ -1,8 +1,10 @@ #!/usr/bin/env python3 import os +import pwd from ctypes import CDLL from ctypes import get_errno from ctypes.util import find_library +from typing import Dict from typing import List from typing import Optional @@ -31,6 +33,8 @@ libc = LibCWrapper() class BaseTest(object): + NEED_ROOT: bool = False # True if the class needs root privileges for the setup + TARGET_USER = None # Set to the target user by the framework REQUIRED_MODULES: List[str] = [] def _check_modules(self): @@ -41,9 +45,26 @@ class BaseTest(object): pytest.skip( "kernel module '{}' not available: {}".format(mod_name, err_str) ) + @property + def atf_vars(self) -> Dict[str, str]: + px = "_ATF_VAR_" + return {k[len(px):]: v for k, v in os.environ.items() if k.startswith(px)} + + def drop_privileges_user(self, user: str): + uid = pwd.getpwnam(user)[2] + print("Dropping privs to {}/{}".format(user, uid)) + os.setuid(uid) + + def drop_privileges(self): + if self.TARGET_USER: + if self.TARGET_USER == "unprivileged": + user = self.atf_vars["unprivileged-user"] + else: + user = self.TARGET_USER + self.drop_privileges_user(user) @property - def test_id(self): + def test_id(self) -> str: # 'test_ip6_output.py::TestIP6Output::test_output6_pktinfo[ipandif] (setup)' return os.environ.get("PYTEST_CURRENT_TEST").split(" ")[0] diff --git a/tests/conftest.py b/tests/conftest.py index b5ce5ae2286b..687f6bde375e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,10 +7,14 @@ PLUGIN_ENABLED = False DEFAULT_HANDLER = None +def set_handler(config): + global DEFAULT_HANDLER, PLUGIN_ENABLED + DEFAULT_HANDLER = ATFHandler(report_file_name=config.option.atf_file) + PLUGIN_ENABLED = True + return DEFAULT_HANDLER + + def get_handler(): - global DEFAULT_HANDLER - if DEFAULT_HANDLER is None: - DEFAULT_HANDLER = ATFHandler() return DEFAULT_HANDLER @@ -81,11 +85,9 @@ def pytest_configure(config): "markers", "timeout(dur): int/float with max duration in sec" ) - global PLUGIN_ENABLED - PLUGIN_ENABLED = config.option.atf - if not PLUGIN_ENABLED: + if not config.option.atf: return - get_handler() + handler = set_handler(config) if config.option.collectonly: # Need to output list of tests to stdout, hence override @@ -93,6 +95,8 @@ def pytest_configure(config): reporter = config.pluginmanager.getplugin("terminalreporter") if reporter: config.pluginmanager.unregister(reporter) + else: + handler.setup_configure() def pytest_collection_modifyitems(session, config, items): @@ -114,6 +118,12 @@ def pytest_collection_finish(session): handler.list_tests(session.items) +def pytest_runtest_setup(item): + if PLUGIN_ENABLED: + handler = get_handler() + handler.setup_method_pre(item) + + def pytest_runtest_logreport(report): if PLUGIN_ENABLED: handler = get_handler() @@ -123,4 +133,4 @@ def pytest_runtest_logreport(report): def pytest_unconfigure(config): if PLUGIN_ENABLED and config.option.atf_file: handler = get_handler() - handler.write_report(config.option.atf_file) + handler.write_report()
Want to link to this message? Use this URL: <https://mail-archive.FreeBSD.org/cgi/mid.cgi?202302091450.319Eomiq088008>