From 1fa044bde5dbfea16b2af13e1345b1981d9712e5 Mon Sep 17 00:00:00 2001 From: Fabian Ruffy <212419330+fruffy-xsight@users.noreply.github.com> Date: Wed, 29 Jul 2026 05:54:24 -0400 Subject: [PATCH 1/2] Make sure files created by PTF can be read by the invoking user. ptf needs root to open raw sockets and it is normally run with sudo. Every file it wrote was then owned by root: the log directory, log files, pcaps, xUnit results, and the profiler dump. That makes the files unreadable in normal operations. Chown those files to SUDO_UID/SUDO_GID as they are created. This only happens when running as root under sudo, and does nothing otherwise. Signed-off-by: Fabian Ruffy <212419330+fruffy-xsight@users.noreply.github.com> --- ptf | 11 +++++++---- src/ptf/__init__.py | 3 +++ src/ptf/dataplane.py | 1 + src/ptf/ptfutils.py | 44 ++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 55 insertions(+), 4 deletions(-) diff --git a/ptf b/ptf index 85c33a1..124520b 100755 --- a/ptf +++ b/ptf @@ -29,6 +29,7 @@ import random import signal import fnmatch import copy +import shutil import types from collections import OrderedDict @@ -470,10 +471,9 @@ def logging_setup(config): if config["log_dir"] != None: if os.path.exists(config["log_dir"]): - import shutil - shutil.rmtree(config["log_dir"]) os.makedirs(config["log_dir"]) + ptf.ptfutils.chown_to_invoking_user(config["log_dir"]) else: if os.path.exists(config["log_file"]): os.remove(config["log_file"]) @@ -490,10 +490,9 @@ def xunit_setup(config): return if os.path.exists(config["xunit_dir"]): - import shutil - shutil.rmtree(config["xunit_dir"]) os.makedirs(config["xunit_dir"]) + ptf.ptfutils.chown_to_invoking_user(config["xunit_dir"]) def pcap_setup(config): @@ -535,6 +534,7 @@ def profiler_teardown(profiler): profiler.disable() profiler.dump_stats(config["profile_file"]) + ptf.ptfutils.chown_to_invoking_user(config["profile_file"]) def load_test_modules(config): @@ -926,6 +926,9 @@ if __name__ == "__main__": else: runner = unittest.TextTestRunner(verbosity=2) result = runner.run(test_suite) + if config["xunit"]: + # The XML result files are only written once the run completes. + ptf.ptfutils.chown_to_invoking_user(config["xunit_dir"], recursive=True) run_failures = result.failures run_errors = result.errors run_timeouts = [] diff --git a/src/ptf/__init__.py b/src/ptf/__init__.py index a7cb9fc..bae606e 100644 --- a/src/ptf/__init__.py +++ b/src/ptf/__init__.py @@ -13,6 +13,8 @@ import os import logging +from . import ptfutils + try: from ._version import __version__ except ImportError: @@ -58,6 +60,7 @@ def open_logfile(name): handler = logging.FileHandler(filename, mode="a") handler.setFormatter(formatter) logger.addHandler(handler) + ptfutils.chown_to_invoking_user(filename) # We log all ERROR and CRITICAL messages to stdout as well as to the # logfile. diff --git a/src/ptf/dataplane.py b/src/ptf/dataplane.py index 7fae161..71741e0 100644 --- a/src/ptf/dataplane.py +++ b/src/ptf/dataplane.py @@ -1039,6 +1039,7 @@ def start_pcap(self, filename): with self.cvar: assert self.pcap_writer == None self.pcap_writer = PcapWriter(filename) + ptfutils.chown_to_invoking_user(filename) def stop_pcap(self): if self.pcap_writer: diff --git a/src/ptf/ptfutils.py b/src/ptf/ptfutils.py index 97464aa..4f90d77 100644 --- a/src/ptf/ptfutils.py +++ b/src/ptf/ptfutils.py @@ -26,6 +26,50 @@ def gen_xid(): return random.randrange(1, 0xFFFFFFFF) +def chown_to_invoking_user(path, recursive=False): + """ + Give a file or directory created by ptf back to the user who ran sudo. + + ptf needs root to open raw sockets, so it is normally started with sudo. + Everything it writes (logs, pcaps, xUnit results) would then be owned by + root, leaving the invoking user with artifacts they can neither read nor + remove. This is a no-op when ptf is not running under sudo. + """ + + # Only root can hand a file to somebody else. Without sudo there is + # nothing to undo either: the files already belong to whoever ran ptf. + if os.geteuid() != 0: + return + + sudo_uid = os.environ.get("SUDO_UID") + sudo_gid = os.environ.get("SUDO_GID") + if sudo_uid is None or sudo_gid is None: + return + + try: + uid, gid = int(sudo_uid), int(sudo_gid) + except ValueError: + logging.warning( + "Ignoring malformed SUDO_UID/SUDO_GID: %s/%s", sudo_uid, sudo_gid + ) + return + + paths = [path] + if recursive: + for dirpath, dirnames, filenames in os.walk(path): + paths.extend(os.path.join(dirpath, name) for name in dirnames + filenames) + + for target in paths: + try: + # lchown, not chown: the output directories are owned by an + # unprivileged user by the time we walk them, so following a + # symlink placed there would let that user have any file on the + # system chowned to themselves. + os.lchown(target, uid, gid) + except OSError as e: + logging.warning("Could not change ownership of %s: %s", target, e) + + """ Wait on a condition variable until the given function returns non-None or a timeout expires. The condition variable must already be acquired. From e7ded661b7d9b62c04fbd53d579395ca6730935e Mon Sep 17 00:00:00 2001 From: Fabian Ruffy <212419330+fruffy-xsight@users.noreply.github.com> Date: Wed, 29 Jul 2026 05:57:46 -0400 Subject: [PATCH 2/2] Add a couple tests for handing sudo-created files to the invoking user Signed-off-by: Fabian Ruffy <212419330+fruffy-xsight@users.noreply.github.com> --- utests/tests/ptf/test_ptfutils.py | 165 ++++++++++++++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 utests/tests/ptf/test_ptfutils.py diff --git a/utests/tests/ptf/test_ptfutils.py b/utests/tests/ptf/test_ptfutils.py new file mode 100644 index 0000000..50fe5ad --- /dev/null +++ b/utests/tests/ptf/test_ptfutils.py @@ -0,0 +1,165 @@ +# Copyright 2026 Xsight Labs +# SPDX-License-Identifier: Apache-2.0 + +import logging +import os + +import pytest + +import ptf +from ptf.ptfutils import chown_to_invoking_user + +INVOKING_UID = 1234 +INVOKING_GID = 5678 + + +def _reject_chown(*args, **kwargs): + raise AssertionError("ownership must be changed with lchown, not chown") + + +@pytest.fixture +def chown_calls(monkeypatch): + """ + Record ownership changes instead of performing them. + + Fails the test if plain chown is used. Following a symlink placed in an + output directory would let an unprivileged user have any file on the + system chowned to themselves. + """ + + calls = [] + monkeypatch.setattr( + os, "lchown", lambda path, uid, gid: calls.append((path, uid, gid)) + ) + monkeypatch.setattr(os, "chown", _reject_chown) + return calls + + +@pytest.fixture +def sudo_env(monkeypatch): + """ + Pretend to be root under sudo, whoever is really running the tests. + """ + + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.setenv("SUDO_UID", str(INVOKING_UID)) + monkeypatch.setenv("SUDO_GID", str(INVOKING_GID)) + + +@pytest.fixture +def restore_root_logger(): + """ + open_logfile() replaces the handlers on the root logger. Put back the ones + pytest installed, so log capturing keeps working for later tests. + """ + + logger = logging.getLogger() + original_handlers = list(logger.handlers) + original_level = logger.level + + yield + + for handler in list(logger.handlers): + logger.removeHandler(handler) + if handler not in original_handlers: + handler.close() + for handler in original_handlers: + logger.addHandler(handler) + logger.setLevel(original_level) + + +class TestChownToInvokingUser: + def test_chowns_to_the_user_who_ran_sudo(self, tmp_path, chown_calls, sudo_env): + logfile = tmp_path / "ptf.log" + logfile.touch() + + chown_to_invoking_user(str(logfile)) + + assert chown_calls == [(str(logfile), INVOKING_UID, INVOKING_GID)] + + def test_does_nothing_when_not_running_as_root( + self, tmp_path, chown_calls, sudo_env, monkeypatch + ): + # sudo -u someuser ptf: SUDO_UID is set, but we cannot give a file away. + monkeypatch.setattr(os, "geteuid", lambda: 1000) + + chown_to_invoking_user(str(tmp_path)) + + assert chown_calls == [] + + def test_does_nothing_when_root_without_sudo( + self, tmp_path, chown_calls, monkeypatch + ): + # Running as root directly, for instance in a container. The files + # are meant to belong to root. + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.delenv("SUDO_UID", raising=False) + monkeypatch.delenv("SUDO_GID", raising=False) + + chown_to_invoking_user(str(tmp_path)) + + assert chown_calls == [] + + def test_does_nothing_when_only_the_uid_is_known( + self, tmp_path, chown_calls, sudo_env, monkeypatch + ): + monkeypatch.delenv("SUDO_GID") + + chown_to_invoking_user(str(tmp_path)) + + assert chown_calls == [] + + def test_warns_and_gives_up_on_a_malformed_sudo_uid( + self, tmp_path, chown_calls, sudo_env, monkeypatch, caplog + ): + monkeypatch.setenv("SUDO_UID", "not-a-number") + + with caplog.at_level(logging.WARNING): + chown_to_invoking_user(str(tmp_path)) + + assert chown_calls == [] + assert "not-a-number" in caplog.text + + def test_recursive_covers_the_whole_tree(self, tmp_path, chown_calls, sudo_env): + (tmp_path / "TEST-results.xml").touch() + (tmp_path / "nested").mkdir() + (tmp_path / "nested" / "TEST-more.xml").touch() + + chown_to_invoking_user(str(tmp_path), recursive=True) + + assert sorted(path for path, _, _ in chown_calls) == [ + str(tmp_path), + str(tmp_path / "TEST-results.xml"), + str(tmp_path / "nested"), + str(tmp_path / "nested" / "TEST-more.xml"), + ] + assert all( + (uid, gid) == (INVOKING_UID, INVOKING_GID) for _, uid, gid in chown_calls + ) + + def test_a_failed_chown_is_reported_but_does_not_stop_the_run( + self, tmp_path, sudo_env, monkeypatch, caplog + ): + def refuse(path, uid, gid): + raise OSError(1, "Operation not permitted") + + monkeypatch.setattr(os, "lchown", refuse) + + with caplog.at_level(logging.WARNING): + chown_to_invoking_user(str(tmp_path)) + + assert "Could not change ownership" in caplog.text + + +class TestLogFileOwnership: + def test_open_logfile_hands_the_log_to_the_invoking_user( + self, tmp_path, chown_calls, sudo_env, monkeypatch, restore_root_logger + ): + logfile = tmp_path / "ptf.log" + monkeypatch.setitem(ptf.config, "log_dir", None) + monkeypatch.setitem(ptf.config, "log_file", str(logfile)) + + ptf.open_logfile("main") + + assert logfile.exists() + assert chown_calls == [(str(logfile), INVOKING_UID, INVOKING_GID)]