From d75735c926d1fd3e5a8555a74114a751deadf0c3 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Tue, 28 Apr 2026 18:11:49 -0700 Subject: [PATCH] =?UTF-8?q?feat(system):=20fix(drone):=20add=20path=20cont?= =?UTF-8?q?ainment=20validation=20on=20registry=20branch=20resolution=20?= =?UTF-8?q?=E2=80=94=20prevents=20ghost-branch=20RCE=20via=20malicious=20p?= =?UTF-8?q?ath=20field=20(issue=20#490=20fix=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: @devpulse --- .../drone/apps/handlers/registry_handler.py | 34 +++++++++++ src/aipass/drone/apps/modules/resolver.py | 8 +++ .../drone/tests/test_registry_handler.py | 60 ++++++++++++++++++- src/aipass/drone/tests/test_resolver.py | 39 ++++++++++++ 4 files changed, 139 insertions(+), 2 deletions(-) diff --git a/src/aipass/drone/apps/handlers/registry_handler.py b/src/aipass/drone/apps/handlers/registry_handler.py index 982bba34..2130ad9f 100644 --- a/src/aipass/drone/apps/handlers/registry_handler.py +++ b/src/aipass/drone/apps/handlers/registry_handler.py @@ -28,6 +28,38 @@ from .exceptions import ( from aipass.drone.apps.handlers.json import json_handler +# --------------------------------------------------------------------------- +# Path containment validation +# --------------------------------------------------------------------------- + + +def _validate_branch_path(branch_path: Path, project_root: Path, branch_name: str) -> bool: + """Validate that a resolved branch path is contained within the project root. + + Returns True if the path is safe. Returns False and logs a warning if + the path escapes the project boundary (path-traversal / ghost-branch). + """ + try: + resolved = branch_path.resolve() + root = project_root.resolve() + if not resolved.is_relative_to(root): + logger.warning( + "SECURITY: branch '%s' path escapes project root: %s (root: %s)", + branch_name, + resolved, + root, + ) + return False + except (OSError, ValueError) as exc: + logger.warning( + "SECURITY: branch '%s' path validation failed: %s", + branch_name, + exc, + ) + return False + return True + + # --------------------------------------------------------------------------- # Registry path resolution # --------------------------------------------------------------------------- @@ -218,6 +250,8 @@ def _load_registry_data(registry_path: Path) -> Dict[str, Any]: branch_path = Path(raw_path) if not branch_path.is_absolute(): branch_path = (registry_dir / branch_path).resolve() + if not _validate_branch_path(branch_path, registry_dir, name): + continue entry = dict(branch) entry["name"] = name entry["path"] = str(branch_path) diff --git a/src/aipass/drone/apps/modules/resolver.py b/src/aipass/drone/apps/modules/resolver.py index 71b7f787..960e2907 100644 --- a/src/aipass/drone/apps/modules/resolver.py +++ b/src/aipass/drone/apps/modules/resolver.py @@ -13,6 +13,7 @@ Thin orchestrator for resolving symbolic @branch names to paths and metadata. Delegates registry access to the handler layer. """ +from pathlib import Path from typing import Any, Dict, List, Optional from aipass.prax import logger @@ -24,6 +25,8 @@ from aipass.drone.apps.handlers.registry_handler import ( load_registry, get_all_branches, get_branch_by_name, + get_registry_path, + _validate_branch_path, ) @@ -158,6 +161,11 @@ def resolve_branch(symbolic_name: str) -> str: if branch is None: raise BranchNotFoundError(f"Branch '{symbolic_name}' not found in registry") + branch_path = Path(branch["path"]) + project_root = get_registry_path().parent + if not _validate_branch_path(branch_path, project_root, name): + raise BranchNotFoundError(f"Branch '{symbolic_name}' path escapes project root — blocked for security") + system_logger.info("Resolved @%s → %s", name, branch["path"]) return branch["path"] diff --git a/src/aipass/drone/tests/test_registry_handler.py b/src/aipass/drone/tests/test_registry_handler.py index 10cf0122..f883a954 100644 --- a/src/aipass/drone/tests/test_registry_handler.py +++ b/src/aipass/drone/tests/test_registry_handler.py @@ -19,6 +19,7 @@ import pytest from aipass.drone.apps.handlers.registry_handler import ( _first_registry_in, + _validate_branch_path, _verify_registry_credential, find_registry, get_all_branches, @@ -126,8 +127,8 @@ class TestLoadRegistry: assert str(branch_path).startswith(str(registry_dir)) def test_absolute_path_not_re_resolved(self, registry_dir: Path): - """A branch with an already-absolute path is NOT re-resolved against registry_dir.""" - abs_path = "/opt/custom/my_branch" + """A branch with an already-absolute path inside project root is preserved as-is.""" + abs_path = str(registry_dir / "my_branch") reg = _minimal_registry( branches=[ { @@ -532,3 +533,58 @@ class TestRegistryPathManagement: with patch.dict(os.environ, {"AIPASS_REGISTRY": env_path}): result = get_registry_path() assert result == Path(env_path) + + +# =================================================================== +# Path containment validation +# =================================================================== + + +class TestValidateBranchPath: + """_validate_branch_path() security boundary checks.""" + + def test_valid_path_inside_project(self, registry_dir: Path): + """Path within project root passes validation.""" + branch = registry_dir / "src" / "aipass" / "trigger" + branch.mkdir(parents=True) + assert _validate_branch_path(branch, registry_dir, "trigger") is True + + def test_path_traversal_blocked(self, registry_dir: Path): + """Path with ../ escaping project root is rejected.""" + evil_path = registry_dir / ".." / ".." / ".." / "tmp" / "evil" + assert _validate_branch_path(evil_path, registry_dir, "evil") is False + + def test_absolute_path_outside_root_blocked(self, registry_dir: Path): + """Absolute path outside project root is rejected.""" + assert _validate_branch_path(Path("/tmp/evil"), registry_dir, "evil") is False + + def test_exact_project_root_passes(self, registry_dir: Path): + """Path equal to project root passes (edge case).""" + assert _validate_branch_path(registry_dir, registry_dir, "root") is True + + def test_symlink_resolved(self, registry_dir: Path): + """Symlink pointing outside project root is rejected after resolution.""" + external = Path(tempfile.mkdtemp(prefix="external_")) + try: + link = registry_dir / "sneaky_link" + link.symlink_to(external) + assert _validate_branch_path(link, registry_dir, "sneaky") is False + finally: + shutil.rmtree(external, ignore_errors=True) + + def test_registry_rejects_traversal_entry(self, registry_dir: Path, monkeypatch): + """A registry entry with path traversal is silently dropped during load.""" + reg = _minimal_registry( + branches=[ + {"name": "legit", "path": "src/legit", "status": "active", "type": "lib"}, + {"name": "evil", "path": "../../../tmp/evil", "status": "active", "type": "lib"}, + ] + ) + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + monkeypatch.chdir(registry_dir) + result = load_registry() + branch_names = list(result["branches"].keys()) + assert "legit" in branch_names + assert "evil" not in branch_names diff --git a/src/aipass/drone/tests/test_resolver.py b/src/aipass/drone/tests/test_resolver.py index 64ab5490..6becf4f8 100644 --- a/src/aipass/drone/tests/test_resolver.py +++ b/src/aipass/drone/tests/test_resolver.py @@ -408,3 +408,42 @@ class TestWithSampleRegistry: assert info["status"] == "active" finally: reset_registry_path() + + +# =========================================================================== +# Path containment — resolver side +# =========================================================================== + + +class TestResolverPathContainment: + """resolve_branch() rejects registry entries with path traversal.""" + + def test_traversal_path_raises(self, tmp_path: Path): + """Registry entry with ../../../tmp/evil is rejected (filtered at load or resolve).""" + reg_file = tmp_path / "AIPASS_REGISTRY.json" + _write_registry( + reg_file, + [_make_branch("evil", "../../../tmp/evil")], + ) + set_registry_path(reg_file) + try: + with pytest.raises(BranchNotFoundError): + resolve_branch("@evil") + finally: + reset_registry_path() + + def test_valid_path_resolves(self, tmp_path: Path): + """Registry entry within project root resolves normally.""" + branch_dir = tmp_path / "src" / "aipass" / "legit" + branch_dir.mkdir(parents=True) + reg_file = tmp_path / "AIPASS_REGISTRY.json" + _write_registry( + reg_file, + [_make_branch("legit", "src/aipass/legit")], + ) + set_registry_path(reg_file) + try: + result = resolve_branch("@legit") + assert "legit" in result + finally: + reset_registry_path()