Make sure files created by PTF can be read by the invoking user.

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 = []

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.

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:

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.

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)]