From babef04e07b9ac91ca28994b18a6f81b26d1192a Mon Sep 17 00:00:00 2001 From: CloakHQ Date: Mon, 11 May 2026 21:29:50 +0200 Subject: [PATCH] fix(cloakserve): sanitize fingerprint seed to prevent path traversal (#217) Validate seed format with strict regex, add path containment check before rmtree, and bind to 127.0.0.1 by default on bare metal. --- bin/cloakserve | 37 +++++++++++++--- tests/test_cloakserve.py | 94 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 125 insertions(+), 6 deletions(-) diff --git a/bin/cloakserve b/bin/cloakserve index c1e07a8..c51242e 100755 --- a/bin/cloakserve +++ b/bin/cloakserve @@ -22,6 +22,7 @@ import json import logging import os import random +import re import shutil import socket import subprocess @@ -61,6 +62,9 @@ BASE_CHROME_ARGS = [ BASE_CDP_PORT = 5100 +SAFE_SEED_RE = re.compile(r"^[A-Za-z0-9_-]{1,128}$") +RESERVED_SEEDS = {"__default__"} + # --------------------------------------------------------------------------- # ChromeProcess — one running Chrome instance @@ -111,6 +115,14 @@ class ChromePool: self._locks[seed] = asyncio.Lock() return self._locks[seed] + def _safe_rmtree(self, path: str) -> None: + resolved = Path(path).resolve() + data_resolved = Path(self._data_dir).resolve() + if resolved == data_resolved or not resolved.is_relative_to(data_resolved): + logger.error("Refusing to delete path outside data_dir: %s", resolved) + return + shutil.rmtree(path, True) + def _allocate_port(self) -> int: """Find a free port starting from _next_port.""" for _ in range(100): @@ -159,6 +171,11 @@ class ChromePool: seed_key = "__default__" actual_seed = str(random.randint(10000, 99999)) else: + if not SAFE_SEED_RE.match(seed) or seed in RESERVED_SEEDS: + raise web.HTTPBadRequest( + text=json.dumps({"error": "Invalid fingerprint seed"}), + content_type="application/json", + ) seed_key = seed actual_seed = seed @@ -232,7 +249,7 @@ class ChromePool: if not await self._wait_for_cdp(port): process.kill() await asyncio.to_thread(process.wait, timeout=5) - await asyncio.to_thread(shutil.rmtree, user_data_dir, True) + await asyncio.to_thread(self._safe_rmtree, user_data_dir) raise web.HTTPBadGateway( text=json.dumps({"error": "Chrome failed to start"}), content_type="application/json", @@ -266,8 +283,7 @@ class ChromePool: await asyncio.to_thread(proc.process.wait, timeout=5) except subprocess.TimeoutExpired: proc.process.kill() - # Clean up user data dir (can be slow for large profiles) - await asyncio.to_thread(shutil.rmtree, proc.user_data_dir, True) + await asyncio.to_thread(self._safe_rmtree, proc.user_data_dir) if self._default is proc: self._default = None self._locks.pop(key, None) @@ -559,8 +575,8 @@ async def on_shutdown(app: web.Application) -> None: # --------------------------------------------------------------------------- def _default_data_dir() -> str: - """Smart default: Docker → /tmp/cloakserve, bare metal → ~/.cloakbrowser/cloakserve.""" - if os.path.exists("/.dockerenv"): + """Smart default: container → /tmp/cloakserve, bare metal → ~/.cloakbrowser/cloakserve.""" + if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"): return "/tmp/cloakserve" return str(Path.home() / ".cloakbrowser" / "cloakserve") @@ -624,6 +640,13 @@ def main() -> None: binary = ensure_binary() config, global_args = parse_cli_args(sys.argv[1:]) + if config["default_seed"] and ( + not SAFE_SEED_RE.match(config["default_seed"]) + or config["default_seed"] in RESERVED_SEEDS + ): + logger.error("Invalid --fingerprint seed: %s", config["default_seed"]) + sys.exit(1) + pool = ChromePool( binary=binary, global_args=global_args, @@ -662,7 +685,9 @@ def main() -> None: port, ) - web.run_app(app, host="0.0.0.0", port=port, print=None) + in_container = os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv") + host = "0.0.0.0" if in_container else "127.0.0.1" + web.run_app(app, host=host, port=port, print=None) if __name__ == "__main__": diff --git a/tests/test_cloakserve.py b/tests/test_cloakserve.py index 74c519f..a1dccb1 100644 --- a/tests/test_cloakserve.py +++ b/tests/test_cloakserve.py @@ -22,6 +22,8 @@ parse_connection_params = _mod.parse_connection_params parse_cli_args = _mod.parse_cli_args ChromePool = _mod.ChromePool _default_data_dir = _mod._default_data_dir +SAFE_SEED_RE = _mod.SAFE_SEED_RE +RESERVED_SEEDS = _mod.RESERVED_SEEDS # --------------------------------------------------------------------------- @@ -244,3 +246,95 @@ class TestConnectionTracking: pool.disconnect("a") assert pool._connections["a"] == 1 assert pool._connections["b"] == 1 + + +# --------------------------------------------------------------------------- +# Seed validation (CVE fix — path traversal via fingerprint param) +# --------------------------------------------------------------------------- + + +class TestSeedValidation: + """Verify SAFE_SEED_RE rejects path traversal and reserved names.""" + + @pytest.mark.parametrize("seed", [ + "../foo", "../../etc", "/etc/passwd", "..", ".", "foo/bar", + "foo\\bar", "\x00evil", "", "a" * 129, + ]) + def test_malicious_seeds_rejected(self, seed): + assert not SAFE_SEED_RE.match(seed) + + @pytest.mark.parametrize("seed", [ + "__default__", + ]) + def test_reserved_seeds_rejected(self, seed): + assert seed in RESERVED_SEEDS + + @pytest.mark.parametrize("seed", [ + "12345", "my-seed_01", "ABC", "a" * 128, "0", "test-seed", + ]) + def test_valid_seeds_accepted(self, seed): + assert SAFE_SEED_RE.match(seed) + assert seed not in RESERVED_SEEDS + + +# --------------------------------------------------------------------------- +# Path containment (_safe_rmtree) +# --------------------------------------------------------------------------- + + +class TestSafeRmtree: + """Verify _safe_rmtree refuses to delete outside data_dir.""" + + def _make_pool(self, data_dir: str): + return ChromePool( + binary="/fake/chrome", + global_args=[], + headless=True, + data_dir=data_dir, + ) + + def test_refuses_path_outside_data_dir(self, tmp_path): + data_dir = tmp_path / "profiles" + data_dir.mkdir() + victim = tmp_path / "victim" + victim.mkdir() + (victim / "sentinel").touch() + + pool = self._make_pool(str(data_dir)) + pool._safe_rmtree(str(victim)) + + assert victim.exists(), "Directory outside data_dir must not be deleted" + + def test_refuses_data_dir_itself(self, tmp_path): + data_dir = tmp_path / "profiles" + data_dir.mkdir() + (data_dir / "sentinel").touch() + + pool = self._make_pool(str(data_dir)) + pool._safe_rmtree(str(data_dir)) + + assert data_dir.exists(), "data_dir itself must not be deleted" + + def test_deletes_valid_subdirectory(self, tmp_path): + data_dir = tmp_path / "profiles" + data_dir.mkdir() + subdir = data_dir / "seed-12345" + subdir.mkdir() + (subdir / "data").touch() + + pool = self._make_pool(str(data_dir)) + pool._safe_rmtree(str(subdir)) + + assert not subdir.exists(), "Valid subdirectory should be deleted" + + def test_refuses_traversal_path(self, tmp_path): + data_dir = tmp_path / "profiles" + data_dir.mkdir() + victim = tmp_path / "victim" + victim.mkdir() + + traversal = str(data_dir / ".." / "victim") + pool = self._make_pool(str(data_dir)) + pool._safe_rmtree(traversal) + + assert victim.exists(), "Traversal path must not be deleted"