From a66b5d93e146fd894dc006f5a62d01cd60706691 Mon Sep 17 00:00:00 2001 From: Subramaniyajothi6 Date: Mon, 13 Jul 2026 21:27:00 +0530 Subject: [PATCH 1/2] fix(parser-sandbox): drop PYTHONPATH/HOME and run parser child unprivileged _sanitised_env() kept PYTHONPATH and HOME in the parser child's environment. An inherited PYTHONPATH pointing at an attacker-controlled directory let the "sandboxed" parser import arbitrary modules; HOME was an equivalent vector via the per-user site directory (~/.local/lib/pythonX.Y/site-packages). - Drop PYTHONPATH and HOME; set PYTHONNOUSERSITE=1 and PYTHONSAFEPATH=1 to also close the user-site and python -c CWD import vectors. Installed dependencies still resolve via the interpreter's own site-packages. - Add _privilege_drop_kwargs(): when the backend runs as root, execute the parser child as the nobody account (fallback uid/gid 65534) via subprocess user/group/extra_groups. No-op when already unprivileged or on Windows. - Apply the same drop to the unshare --user --net capability probe so a host where root can unshare but nobody cannot is detected at probe time. Flip the two helper tests that asserted PYTHONPATH/HOME are retained (they encoded the vulnerable behaviour), add hardening-flag and privilege-drop coverage, and add an end-to-end regression proving an injected PYTHONPATH cannot be imported by the sandboxed parser. --- backend/secuscan/parser_sandbox.py | 76 ++++++++++++++-- testing/backend/unit/test_parser_sandbox.py | 34 +++++++ .../unit/test_parser_sandbox_helpers.py | 90 +++++++++++++++++-- 3 files changed, 186 insertions(+), 14 deletions(-) diff --git a/backend/secuscan/parser_sandbox.py b/backend/secuscan/parser_sandbox.py index 7d7c4d0d0..0aff2b42b 100644 --- a/backend/secuscan/parser_sandbox.py +++ b/backend/secuscan/parser_sandbox.py @@ -10,6 +10,12 @@ other in-process state. - Environment variables (which may contain SECUSCAN_VAULT_KEY, API keys, etc.) are stripped from the child process. + - The child's import path is locked down: neither PYTHONPATH nor HOME is + inherited, and user-site / current-directory imports are disabled, so a + caller cannot point the "sandboxed" interpreter at an attacker-controlled + directory and have the parser import arbitrary modules. + - When the backend runs as root, the parser child drops to an unprivileged + user before it executes, so untrusted parser code never runs as root. - Execution is bounded by a configurable timeout. - Output size is capped so a runaway parser cannot exhaust backend memory. - On Linux with unprivileged user namespaces available, the child runs in @@ -139,14 +145,67 @@ def __init__(self, plugin_id: str, reason: str, stderr: str = "") -> None: def _sanitised_env() -> Dict[str, str]: - """Return a minimal environment for the child process. + """Return a minimal, hardened environment for the parser child. + + Beyond stripping credentials and application secrets, this deliberately + drops the two variables that let a caller inject import search paths into + the "sandboxed" interpreter: + + * ``PYTHONPATH`` — prepends arbitrary directories to ``sys.path``, so an + inherited value pointing at an attacker-controlled directory lets the + parser ``import`` anything placed there. + * ``HOME`` — controls the per-user site directory + (``~/.local/lib/pythonX.Y/site-packages``), which is another + ``sys.path`` entry and therefore an equivalent injection vector. + + Only variables the interpreter genuinely needs to start and to decode text + are kept (``PATH``, temp-dir hints, locale). ``PYTHONNOUSERSITE`` disables + the user-site directory even if ``HOME`` leaks in some other way, and + ``PYTHONSAFEPATH`` (3.11+) stops the process working directory — added to + ``sys.path`` by ``python -c`` — from being used to smuggle in modules. + Installed dependencies still resolve normally: the interpreter's own + ``site-packages`` is on ``sys.path`` regardless of ``PYTHONPATH``. + """ + keep_keys = {"PATH", "TMPDIR", "TEMP", "TMP", "LANG", "LC_ALL"} + env = {k: v for k, v in os.environ.items() if k in keep_keys} + env["PYTHONNOUSERSITE"] = "1" + env["PYTHONSAFEPATH"] = "1" + return env + + +# Fallback ids for the classic unprivileged "nobody"/"nogroup" accounts, used +# only if the names cannot be resolved on this host. +_NOBODY_UID = 65534 +_NOBODY_GID = 65534 + + +def _privilege_drop_kwargs() -> Dict[str, Any]: + """subprocess kwargs that run the parser child as an unprivileged user. + + Untrusted parser code must never execute with root privileges. When the + backend itself runs as root we resolve the ``nobody`` account (falling back + to the conventional 65534 ids) and hand ``user``/``group``/``extra_groups`` + to :class:`subprocess.Popen`, which performs the setgid/setgroups/setuid + drop in the child between ``fork`` and ``exec``. - Retains PATH and PYTHONPATH (needed to locate the interpreter and any - installed packages) while stripping all credentials and application - secrets present in the parent's environment. + Returns an empty dict — i.e. no change in behaviour — when the backend is + not running as root or the platform has no concept of uids (Windows). In + that common case the process is already unprivileged and there is nothing + to drop. """ - keep_keys = {"PATH", "PYTHONPATH", "HOME", "TMPDIR", "TEMP", "TMP", "LANG", "LC_ALL"} - return {k: v for k, v in os.environ.items() if k in keep_keys} + if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0: + return {} + + uid, gid = _NOBODY_UID, _NOBODY_GID + try: + import pwd + + entry = pwd.getpwnam("nobody") + uid, gid = entry.pw_uid, entry.pw_gid + except (ImportError, KeyError): + pass + + return {"user": uid, "group": gid, "extra_groups": []} # --------------------------------------------------------------------------- @@ -200,6 +259,10 @@ def _unshare_net_supported() -> bool: [unshare_path, "--user", "--net", "--", "true"], capture_output=True, timeout=5, + # Probe under the same privilege drop the real run uses, so that a + # host where root can create user namespaces but "nobody" cannot is + # detected here rather than failing at parse time. + **_privilege_drop_kwargs(), ) _unshare_available = probe.returncode == 0 except Exception: @@ -280,6 +343,7 @@ def run_parser_in_sandbox( stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=_sanitised_env(), + **_privilege_drop_kwargs(), ) def _read_stdout() -> None: diff --git a/testing/backend/unit/test_parser_sandbox.py b/testing/backend/unit/test_parser_sandbox.py index 6667b6b9f..19be5f18c 100644 --- a/testing/backend/unit/test_parser_sandbox.py +++ b/testing/backend/unit/test_parser_sandbox.py @@ -377,6 +377,40 @@ def test_sanitised_env_retains_path(self): env = _sanitised_env() assert "PATH" in env + def test_injected_pythonpath_cannot_be_imported(self, tmp_path): + """Regression for the sandbox-escape vector (issue #1804). + + A caller-controlled PYTHONPATH must NOT let the parser import modules + from an attacker-chosen directory. We plant an ``evilmod`` on PYTHONPATH + and confirm the sandboxed parser cannot import it. + """ + evil_dir = tmp_path / "attacker" + evil_dir.mkdir() + (evil_dir / "evilmod.py").write_text("PWNED = True\n") + + p = _write_parser( + tmp_path, + """\ + import evilmod + def parse(output): + return {"pwned": evilmod.PWNED} + """, + ) + + prev = os.environ.get("PYTHONPATH") + os.environ["PYTHONPATH"] = str(evil_dir) + try: + with pytest.raises(ParserSandboxError) as exc_info: + run_parser_in_sandbox(p, "escape_plugin", "data") + finally: + if prev is None: + os.environ.pop("PYTHONPATH", None) + else: + os.environ["PYTHONPATH"] = prev + + # The failure must be the blocked import, not some unrelated error. + assert "evilmod" in exc_info.value.stderr_excerpt + # --------------------------------------------------------------------------- # ParserSandboxError diff --git a/testing/backend/unit/test_parser_sandbox_helpers.py b/testing/backend/unit/test_parser_sandbox_helpers.py index 785d46213..f166f9a9c 100644 --- a/testing/backend/unit/test_parser_sandbox_helpers.py +++ b/testing/backend/unit/test_parser_sandbox_helpers.py @@ -23,20 +23,34 @@ def test_sanitised_env_keeps_path(): assert "PATH" in result -def test_sanitised_env_keeps_pythonpath(): - """PYTHONPATH is retained in the sanitised environment.""" +def test_sanitised_env_drops_pythonpath(): + """PYTHONPATH is dropped: an inherited value is an import-injection vector.""" from backend.secuscan.parser_sandbox import _sanitised_env - with patch.dict("os.environ", {"PYTHONPATH": "/opt/python"}, clear=False): + with patch.dict("os.environ", {"PYTHONPATH": "/opt/attacker"}, clear=False): result = _sanitised_env() - assert "PYTHONPATH" in result + assert "PYTHONPATH" not in result -def test_sanitised_env_keeps_home(): - """HOME is retained in the sanitised environment.""" +def test_sanitised_env_drops_home(): + """HOME is dropped: it controls the ~/.local user-site sys.path entry.""" from backend.secuscan.parser_sandbox import _sanitised_env - with patch.dict("os.environ", {"HOME": "/root"}, clear=False): + with patch.dict("os.environ", {"HOME": "/attacker/home"}, clear=False): result = _sanitised_env() - assert "HOME" in result + assert "HOME" not in result + + +def test_sanitised_env_disables_user_site(): + """PYTHONNOUSERSITE is forced on so ~/.local site-packages is ignored.""" + from backend.secuscan.parser_sandbox import _sanitised_env + result = _sanitised_env() + assert result.get("PYTHONNOUSERSITE") == "1" + + +def test_sanitised_env_sets_safe_path(): + """PYTHONSAFEPATH is forced on so CWD/'' is not prepended to sys.path.""" + from backend.secuscan.parser_sandbox import _sanitised_env + result = _sanitised_env() + assert result.get("PYTHONSAFEPATH") == "1" def test_sanitised_env_strips_vault_key(): @@ -90,3 +104,63 @@ def test_sanitised_env_is_deterministic(): r1 = _sanitised_env() r2 = _sanitised_env() assert r1 == r2 + + +# --------------------------------------------------------------------------- +# _privilege_drop_kwargs() +# --------------------------------------------------------------------------- + + +def test_privilege_drop_noop_when_not_root(): + """When the backend is not root there is nothing to drop; kwargs are empty.""" + from backend.secuscan import parser_sandbox + with patch.object(parser_sandbox.os, "geteuid", return_value=1000, create=True): + assert parser_sandbox._privilege_drop_kwargs() == {} + + +def test_privilege_drop_noop_on_non_posix(): + """On a platform without posix semantics the drop is a no-op.""" + from backend.secuscan import parser_sandbox + with patch.object(parser_sandbox.os, "name", "nt"): + assert parser_sandbox._privilege_drop_kwargs() == {} + + +def test_privilege_drop_targets_nobody_when_root(): + """As root, the child is dropped to the resolved unprivileged account.""" + import sys + import types + from backend.secuscan import parser_sandbox + + class _FakePw: + pw_uid = 12345 + pw_gid = 23456 + + # `pwd` is POSIX-only; inject a fake so this runs on any CI host. + fake_pwd = types.SimpleNamespace(getpwnam=lambda name: _FakePw()) + with patch.object(parser_sandbox.os, "name", "posix"), \ + patch.object(parser_sandbox.os, "geteuid", return_value=0, create=True), \ + patch.dict(sys.modules, {"pwd": fake_pwd}): + kwargs = parser_sandbox._privilege_drop_kwargs() + + assert kwargs["user"] == 12345 + assert kwargs["group"] == 23456 + assert kwargs["extra_groups"] == [] + + +def test_privilege_drop_falls_back_to_65534_when_nobody_missing(): + """A host without a 'nobody' account still gets a safe unprivileged target.""" + import sys + import types + from backend.secuscan import parser_sandbox + + def _missing(name): + raise KeyError(name) + + fake_pwd = types.SimpleNamespace(getpwnam=_missing) + with patch.object(parser_sandbox.os, "name", "posix"), \ + patch.object(parser_sandbox.os, "geteuid", return_value=0, create=True), \ + patch.dict(sys.modules, {"pwd": fake_pwd}): + kwargs = parser_sandbox._privilege_drop_kwargs() + + assert kwargs["user"] == parser_sandbox._NOBODY_UID + assert kwargs["group"] == parser_sandbox._NOBODY_GID From 37ca5dd770c9f71cb7c8174106479ab8e74220bd Mon Sep 17 00:00:00 2001 From: Subramaniyajothi6 Date: Wed, 15 Jul 2026 16:14:26 +0530 Subject: [PATCH 2/2] fix(parser-sandbox): stage parser.py so the privilege-dropped child can read it Dropping the parser child to 'nobody' when the backend runs as root broke parsing whenever parser.py (or its parent directory) was root-owned and not world-readable: the unprivileged child could not read the source and the parse failed. Reads were previously assumed to work because the tests only inspected the subprocess kwargs, never an actual root-owned execution. Stage a private copy of parser.py the drop target can read: the file is copied into a fresh temp dir, chowned to the drop uid/gid and made 0400, while the directory stays backend-owned and 0711 (traversable, not listable) so cleanup works whether or not the backend is root. The copy lives only for the child's lifetime and is removed on exit. When no privilege drop is in effect the original path is used unchanged. The drop is resolved once and reused for both staging and the child so they always target the same account. Tests: _staged_parser is a no-op without a drop and otherwise produces a faithful, 0400, drop-target-readable copy that is cleaned up; an end-to-end run under a simulated drop still parses and leaks no staging dir; and a root-gated regression reproduces the root-private parser.py condition and proves the child reads+runs it unprivileged. Addresses review feedback on #2019. --- backend/secuscan/parser_sandbox.py | 172 ++++++++++++------ testing/backend/unit/test_parser_sandbox.py | 84 +++++++++ .../unit/test_parser_sandbox_helpers.py | 55 ++++++ 3 files changed, 251 insertions(+), 60 deletions(-) diff --git a/backend/secuscan/parser_sandbox.py b/backend/secuscan/parser_sandbox.py index 0aff2b42b..21518c4ba 100644 --- a/backend/secuscan/parser_sandbox.py +++ b/backend/secuscan/parser_sandbox.py @@ -45,6 +45,7 @@ from __future__ import annotations +import contextlib import json import os import platform @@ -53,9 +54,10 @@ import sys import subprocess import string +import tempfile import logging from pathlib import Path -from typing import Any, Dict +from typing import Any, Dict, Iterator logger = logging.getLogger(__name__) @@ -208,6 +210,46 @@ def _privilege_drop_kwargs() -> Dict[str, Any]: return {"user": uid, "group": gid, "extra_groups": []} +@contextlib.contextmanager +def _staged_parser( + parser_path: Path, drop_kwargs: Dict[str, Any] +) -> Iterator[Path]: + """Yield a ``parser.py`` path the (possibly privilege-dropped) child can read. + + When the parser child drops to an unprivileged user (i.e. the backend runs + as root and ``drop_kwargs`` is non-empty), the original ``parser.py`` may + live under a root-owned, non-world-readable directory that the drop target + cannot traverse or read — the child would then fail to import it. Stage a + private copy owned by that account, with the staging directory and file + readable only by it (and root), and remove the copy once the child has + finished. + + When no privilege drop is in effect the original path is yielded unchanged + and nothing is copied — the already-unprivileged backend can read its own + plugin files directly. + """ + if not drop_kwargs: + yield parser_path + return + + uid, gid = drop_kwargs["user"], drop_kwargs["group"] + staging_dir = tempfile.mkdtemp(prefix="secuscan-parser-") + try: + staged = Path(staging_dir) / "parser.py" + shutil.copyfile(parser_path, staged) + # Give the copy to the unprivileged account read-only (0400); other + # local users cannot read it. The directory stays owned by the backend + # and only traversable by others (0711, not readable/listable), which + # lets the dropped child reach the file while keeping cleanup below + # working whether or not the backend is root. + os.chown(staged, uid, gid) + os.chmod(staged, 0o400) + os.chmod(staging_dir, 0o711) + yield staged + finally: + shutil.rmtree(staging_dir, ignore_errors=True) + + # --------------------------------------------------------------------------- # Network isolation for the parser subprocess # --------------------------------------------------------------------------- @@ -318,12 +360,10 @@ def run_parser_in_sandbox( max_input_bytes = max(len(parser_input.encode("utf-8")) + 128, 64 * 1024) - # Use Template.safe_substitute so that any stray $ in parser_path does not - # raise; repr() ensures the path is a valid Python string literal. - bootstrap = _BOOTSTRAP_TEMPLATE.safe_substitute( - parser_path_repr=repr(str(parser_path)), - max_input_bytes=max_input_bytes, - ) + # Resolve the privilege drop once and reuse it for both staging and the + # child process, so the account the file is made readable for is exactly + # the account the child runs as. + drop_kwargs = _privilege_drop_kwargs() envelope = json.dumps({"input": parser_input}) stdin_bytes = envelope.encode("utf-8") @@ -337,61 +377,73 @@ def run_parser_in_sandbox( # Stderr cap: diagnostics only — 64 KB is more than enough for any error message. _MAX_STDERR_BYTES = 65536 - proc = subprocess.Popen( - _sandbox_argv(sys.executable, bootstrap), - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - env=_sanitised_env(), - **_privilege_drop_kwargs(), - ) - - def _read_stdout() -> None: - nonlocal stdout_total, overflow - assert proc.stdout is not None - while True: - chunk = proc.stdout.read(65536) - if not chunk: - break - stdout_total += len(chunk) - if stdout_total > max_output_bytes: - overflow = True - proc.kill() - break - stdout_chunks.append(chunk) - - def _read_stderr() -> None: - assert proc.stderr is not None - total = 0 - while True: - chunk = proc.stderr.read(4096) - if not chunk: - break - total += len(chunk) - if total <= _MAX_STDERR_BYTES: - stderr_chunks.append(chunk) - # Always drain so the child is never blocked on a full pipe. - - t_out = threading.Thread(target=_read_stdout, daemon=True) - t_err = threading.Thread(target=_read_stderr, daemon=True) - t_out.start() - t_err.start() - - try: - proc.stdin.write(stdin_bytes) # type: ignore[union-attr] - proc.stdin.close() # type: ignore[union-attr] - except BrokenPipeError: - pass + # Stage the parser where the (possibly privilege-dropped) child can read it, + # then run it. The staged copy must outlive the child, so the whole run + # happens inside the context; _staged_parser removes it on exit and is a + # no-op when the backend is already unprivileged. + with _staged_parser(parser_path, drop_kwargs) as effective_parser_path: + # Use Template.safe_substitute so that any stray $ in parser_path does + # not raise; repr() ensures the path is a valid Python string literal. + bootstrap = _BOOTSTRAP_TEMPLATE.safe_substitute( + parser_path_repr=repr(str(effective_parser_path)), + max_input_bytes=max_input_bytes, + ) - timed_out = False - try: - proc.wait(timeout=timeout_seconds) - except subprocess.TimeoutExpired: - timed_out = True - proc.kill() + proc = subprocess.Popen( + _sandbox_argv(sys.executable, bootstrap), + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=_sanitised_env(), + **drop_kwargs, + ) - t_out.join(timeout=5) - t_err.join(timeout=5) + def _read_stdout() -> None: + nonlocal stdout_total, overflow + assert proc.stdout is not None + while True: + chunk = proc.stdout.read(65536) + if not chunk: + break + stdout_total += len(chunk) + if stdout_total > max_output_bytes: + overflow = True + proc.kill() + break + stdout_chunks.append(chunk) + + def _read_stderr() -> None: + assert proc.stderr is not None + total = 0 + while True: + chunk = proc.stderr.read(4096) + if not chunk: + break + total += len(chunk) + if total <= _MAX_STDERR_BYTES: + stderr_chunks.append(chunk) + # Always drain so the child is never blocked on a full pipe. + + t_out = threading.Thread(target=_read_stdout, daemon=True) + t_err = threading.Thread(target=_read_stderr, daemon=True) + t_out.start() + t_err.start() + + try: + proc.stdin.write(stdin_bytes) # type: ignore[union-attr] + proc.stdin.close() # type: ignore[union-attr] + except BrokenPipeError: + pass + + timed_out = False + try: + proc.wait(timeout=timeout_seconds) + except subprocess.TimeoutExpired: + timed_out = True + proc.kill() + + t_out.join(timeout=5) + t_err.join(timeout=5) stderr_text = b"".join(stderr_chunks).decode("utf-8", errors="replace") diff --git a/testing/backend/unit/test_parser_sandbox.py b/testing/backend/unit/test_parser_sandbox.py index 19be5f18c..d684815f8 100644 --- a/testing/backend/unit/test_parser_sandbox.py +++ b/testing/backend/unit/test_parser_sandbox.py @@ -412,6 +412,90 @@ def parse(output): assert "evilmod" in exc_info.value.stderr_excerpt +# --------------------------------------------------------------------------- +# Privilege drop: the dropped child must still be able to read + run parser.py +# (PR #2019 review): dropping to nobody must not make the parser unreadable. +# --------------------------------------------------------------------------- + + +class TestPrivilegeDropExecution: + @pytest.mark.skipif(os.name != "posix", reason="privilege drop is POSIX-only") + def test_parser_runs_when_privilege_drop_is_active(self, tmp_path, monkeypatch): + """With a drop in effect the parser is staged and still parses correctly.""" + from backend.secuscan import parser_sandbox + + p = _write_parser( + tmp_path, + """\ + def parse(output): + return {"echo": output} + """, + ) + + # A real drop to another account needs root; drop to our own ids so the + # staging + child-exec path runs unchanged on non-root CI. Omit + # extra_groups (setgroups needs privilege) to keep the child startable. + drop = {"user": os.getuid(), "group": os.getgid()} + monkeypatch.setattr(parser_sandbox, "_privilege_drop_kwargs", lambda: drop) + + result = run_parser_in_sandbox(p, "drop_plugin", "payload") + assert result == {"echo": "payload"} + + @pytest.mark.skipif(os.name != "posix", reason="privilege drop is POSIX-only") + def test_no_staging_dir_leaks_after_dropped_run(self, tmp_path, monkeypatch): + """The staged parser copy is cleaned up after a privilege-dropped run.""" + import tempfile + from pathlib import Path + from backend.secuscan import parser_sandbox + + p = _write_parser( + tmp_path, + """\ + def parse(output): + return {"ok": True} + """, + ) + drop = {"user": os.getuid(), "group": os.getgid()} + monkeypatch.setattr(parser_sandbox, "_privilege_drop_kwargs", lambda: drop) + + before = set(Path(tempfile.gettempdir()).glob("secuscan-parser-*")) + run_parser_in_sandbox(p, "drop_plugin", "data") + after = set(Path(tempfile.gettempdir()).glob("secuscan-parser-*")) + assert after == before + + @pytest.mark.skipif( + os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0, + reason="requires running as root to exercise a real privilege drop", + ) + def test_root_reads_and_runs_root_private_parser_unprivileged(self, tmp_path): + """As root, a root-private parser.py is still read+run by the dropped child. + + Reproduces the exact condition the naive drop broke: parser.py and its + parent directory are root-owned and unreadable to others. The staged + copy must let the unprivileged child read and execute it. + """ + private_dir = tmp_path / "private" + private_dir.mkdir() + p = _write_parser( + private_dir, + """\ + import os + def parse(output): + return {"euid": os.geteuid(), "echo": output} + """, + ) + # Root-private: only root can read the original file / traverse its dir. + os.chmod(p, 0o600) + os.chmod(private_dir, 0o700) + + result = run_parser_in_sandbox(p, "root_plugin", "payload") + + # The parser ran at all → the dropped child could read the staged copy. + assert result["echo"] == "payload" + # And it ran unprivileged, never as root. + assert result["euid"] != 0 + + # --------------------------------------------------------------------------- # ParserSandboxError # --------------------------------------------------------------------------- diff --git a/testing/backend/unit/test_parser_sandbox_helpers.py b/testing/backend/unit/test_parser_sandbox_helpers.py index f166f9a9c..71b4eab5a 100644 --- a/testing/backend/unit/test_parser_sandbox_helpers.py +++ b/testing/backend/unit/test_parser_sandbox_helpers.py @@ -4,6 +4,8 @@ Covers _sanitised_env() from backend.secuscan.parser_sandbox. """ +import os + import pytest from unittest.mock import patch @@ -164,3 +166,56 @@ def _missing(name): assert kwargs["user"] == parser_sandbox._NOBODY_UID assert kwargs["group"] == parser_sandbox._NOBODY_GID + + +# --------------------------------------------------------------------------- +# _staged_parser() — makes parser.py readable to the privilege-dropped child +# (PR #2019 review: prove the drop target can actually read the parser). +# --------------------------------------------------------------------------- + + +def test_staged_parser_is_noop_without_drop(tmp_path): + """With no privilege drop the original path is yielded, nothing is copied.""" + import os + from backend.secuscan.parser_sandbox import _staged_parser + + original = tmp_path / "parser.py" + original.write_text("def parse(x):\n return {}\n") + + before = set(os.listdir(tmp_path)) + with _staged_parser(original, {}) as effective: + assert effective == original + # No staging directory was created next to (or instead of) the original. + assert set(os.listdir(tmp_path)) == before + + +@pytest.mark.skipif(os.name != "posix", reason="chown/file-mode semantics are POSIX-only") +def test_staged_parser_makes_a_private_readable_copy(tmp_path): + """A drop stages a faithful, drop-target-readable, then cleaned-up copy.""" + import os + import stat + from backend.secuscan.parser_sandbox import _staged_parser + + original = tmp_path / "parser.py" + source = "def parse(output):\n return {'echo': output}\n" + original.write_text(source) + + # A real drop needs root; targeting our own ids exercises the same staging + # path (chown to self is always permitted) so this runs on non-root CI. + drop = {"user": os.getuid(), "group": os.getgid(), "extra_groups": []} + + with _staged_parser(original, drop) as staged: + staged_dir = staged.parent + # A distinct copy under a fresh secuscan staging directory. + assert staged != original + assert "secuscan-parser-" in os.path.basename(staged_dir) + # Faithful copy of the source parser. + assert staged.read_text() == source + # File is read-only to the drop target; directory only traversable. + assert stat.S_IMODE(os.stat(staged).st_mode) == 0o400 + assert stat.S_IMODE(os.stat(staged_dir).st_mode) == 0o711 + # The drop target (here: us) can actually read it. + assert os.access(staged, os.R_OK) + + # The staged copy is removed once the context exits. + assert not staged_dir.exists()