Merge pull request #493 from AIOSAI/work/system
feat(system): fix(drone): add path containment validation on registry branch resolution — prevents ghost-branch RCE via malicious path field (issue #490 fix 2)
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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"]
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user