fix(agents): validate file paths in ContextAnalysisAgent._load_files to prevent path traversal #9229

Open
HAL9000 wants to merge 5 commits from fix/context-analysis-agent-path-traversal into master
14 changed files with 951 additions and 25 deletions
+8
View File
@@ -850,6 +850,14 @@ uko-oo:Class` triple emission in `PythonAnalyzer._extract_class()` so that
prevent the agent from crashing on its own output. Added BDD scenarios and Robot
Framework integration tests for the new behaviour.
- **Path Traversal Prevention in ContextAnalysisAgent** (#9093): Replaced the
vulnerable `os.path.commonpath` containment check in
`ContextAnalysisAgent._validate_file_path` with `Path.relative_to()`, which
is immune to directory-prefix collision attacks (CWE-22). Also switched from
`os.path.abspath()` to `Path.resolve()` so that symlinks are followed before
the containment check, preventing symlink-based sandbox escapes. Added Robot
Framework integration tests for end-to-end path traversal verification.
- **Automation Profile Silent Fallback** (#8232): `_resolve_profile_for_plan` in
`PlanLifecycleService` now raises a clear `ValidationError` when a plan's
automation profile name is not a known built-in profile, instead of silently
+1
View File
@@ -32,6 +32,7 @@ Below are some of the specific details of various contributions.
* HAMZA KHYARI has contributed the ACMS execute-phase context assembler project-level hot_max_tokens fix (PR #11036 / issue #11035): added `_resolve_effective_budget()` method that reads each linked project's `settings.hot_max_tokens` and uses the maximum override value as the pipeline budget instead of the hardcoded global 16K default.
* HAL 9000 has contributed the automated CLI docstring example validation (#9106): added `DocstringExampleValidator` to enforce positional-before-option ordering in CLI `Examples:` sections, with Behave test coverage and CONTRIBUTING.md documentation.
* HAL 9000 has contributed the AutoDebugAgent prompt injection mitigation fix (#9110): sanitized user-provided `error_message` and `code_context` fields in all three agent methods using `PromptSanitizer` boundary markers, added graceful `PromptInjectionDetected` exception handling, and added BDD and Robot Framework integration tests for the security fix.
* HAL 9000 has contributed the CWE-22 path traversal fix for ContextAnalysisAgent (#9229): added `_validate_file_path` method using `Path.relative_to()` containment checks and `allowed_base_dir` parameter to prevent directory traversal and symlink-based sandbox escapes in file loading operations.
* This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.
* HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system.
* HAL 9000 has contributed the pr-review-pool-supervisor tracking prefix documentation fix (#7891): aligned all documentation references from the outdated `AUTO-REV-POOL` prefix to the correct `AUTO-REV-SUP` prefix used in production.
@@ -0,0 +1,72 @@
@context_analysis @security @path_traversal
Outdated
Review

BLOCKING: Missing TDD Regression Tag (@tdd_issue_9093)

Per the TDD bug fix workflow in CONTRIBUTING.md, every Type/Bug fix requires a dedicated regression scenario tagged @tdd_issue and @tdd_issue_9093. No scenario in this file carries these tags.

Fix: Add a regression scenario such as:

@tdd_issue @tdd_issue_9093
Scenario: Regression guard — path traversal was not blocked before fix
  When I create a ContextAnalysisAgent with allowed base directory
  And I invoke the agent with file path "../../../../etc/passwd"
  Then the agent should reject the path with error "Path traversal attempt"
  And no files should be loaded

Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING: Missing TDD Regression Tag (`@tdd_issue_9093`)** Per the TDD bug fix workflow in CONTRIBUTING.md, every `Type/Bug` fix requires a dedicated regression scenario tagged `@tdd_issue` and `@tdd_issue_9093`. No scenario in this file carries these tags. **Fix:** Add a regression scenario such as: ```gherkin @tdd_issue @tdd_issue_9093 Scenario: Regression guard — path traversal was not blocked before fix When I create a ContextAnalysisAgent with allowed base directory And I invoke the agent with file path "../../../../etc/passwd" Then the agent should reject the path with error "Path traversal attempt" And no files should be loaded ``` --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
Feature: ContextAnalysisAgent Path Traversal Prevention
As a security-conscious developer
I want the ContextAnalysisAgent to validate file paths
So that attackers cannot read files outside the intended directory
Background:
Given a temporary directory for testing
And a test file "allowed.txt" with content "This is allowed"
And a test file "subdir/nested.txt" with content "Nested file"
@path_traversal_prevention
Scenario: Reject absolute path traversal attempt
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "../../../../etc/passwd"
Then the agent should reject the path with error "Path traversal attempt"
And no files should be loaded
@path_traversal_prevention
Scenario: Reject relative path traversal with dot-dot
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "../../../sensitive.txt"
Then the agent should reject the path with error "Path traversal attempt"
And no files should be loaded
@path_traversal_prevention
Scenario: Accept valid relative path within allowed directory
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "allowed.txt"
Then the agent should load the file successfully
And the loaded document should contain "This is allowed"
@path_traversal_prevention
Scenario: Accept valid nested path within allowed directory
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "subdir/nested.txt"
Then the agent should load the file successfully
And the loaded document should contain "Nested file"
@path_traversal_prevention
Scenario: Reject path that escapes via symlink-like traversal
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "../../../../../../etc/passwd"
Then the agent should reject the path with error "Path traversal attempt"
And no files should be loaded
@path_traversal_prevention
Scenario: Reject mixed traversal patterns
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "subdir/../../etc/passwd"
Then the agent should reject the path with error "Path traversal attempt"
And no files should be loaded
@path_traversal_prevention
Scenario: Handle invalid path gracefully
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file path "nonexistent.txt"
Then the agent should report "File not found"
And no files should be loaded
@path_traversal_prevention
Scenario: Multiple files with one traversal attempt should reject only the malicious one
When I create a ContextAnalysisAgent with allowed base directory
And I invoke the agent with file paths:
| path |
| allowed.txt |
| ../../../../etc/passwd |
| subdir/nested.txt |
Then the agent should load 2 files successfully
And the agent should report 1 error for path traversal
And the loaded documents should contain "This is allowed"
And the loaded documents should contain "Nested file"
@@ -58,6 +58,10 @@ def _ensure_agent(context: Any, **kwargs: Any) -> ContextAnalysisAgent:
llm_factory: Callable[[], FakeListLLM] | None = getattr(context, "make_llm", None)
if llm_factory is not None and "llm" not in kwargs:
kwargs["llm"] = llm_factory()
# Fixtures in this module write temp files under tempfile.mkdtemp(...)
# which resolves to /tmp on Linux CI. Sandbox the agent there so the
# path-traversal check accepts those absolute paths.
kwargs.setdefault("allowed_base_dir", str(Path("/tmp").resolve()))
context.agent = ContextAnalysisAgent(**kwargs)
return context.agent
@@ -82,7 +82,13 @@ def step_coverage_module_imported(context: Any) -> None:
@given("I have a context analysis agent for coverage boost")
def step_create_normal_agent(context: Any) -> None:
llm = FakeListLLM(responses=list(DEFAULT_RESPONSES))
context.cb_agent = ContextAnalysisAgent(llm=llm, retry_attempts=1)
# Sandbox to /tmp because the When step writes the real file via
# tempfile.NamedTemporaryFile, which lands under /tmp on Linux CI.
context.cb_agent = ContextAnalysisAgent(
llm=llm,
retry_attempts=1,
allowed_base_dir=str(Path("/tmp").resolve()),
)
@given("I have a context analysis agent with a raising LLM for coverage boost")
@@ -90,7 +96,11 @@ def step_create_raising_agent(context: Any) -> None:
"""Create an agent whose LLM always raises RuntimeError."""
# Build with a valid LLM first so __init__ completes
llm = FakeListLLM(responses=list(DEFAULT_RESPONSES))
agent = ContextAnalysisAgent(llm=llm, retry_attempts=1)
agent = ContextAnalysisAgent(
llm=llm,
retry_attempts=1,
allowed_base_dir=str(Path("/tmp").resolve()),
)
# Swap to the raising LLM so node methods hit exception handlers
agent.llm = _AlwaysRaisingLLM("forced LLM failure")
context.cb_agent = agent
@@ -89,7 +89,9 @@ def step_configure_fake_llm(context: Any) -> None:
def step_create_agent_custom_llm(context: Any) -> None:
custom_llm = FakeListLLM(responses=list(_DEFAULT_RESPONSES))
context.custom_llm_ref = custom_llm
context.graph_agent = ContextAnalysisAgent(llm=custom_llm)
context.graph_agent = ContextAnalysisAgent(
llm=custom_llm, allowed_base_dir=str(Path("/tmp").resolve())
)
@then("the agent should use the provided LLM instance")
@@ -113,6 +115,11 @@ def _ensure_agent(context: Any, **kwargs: Any) -> ContextAnalysisAgent:
llm = kwargs.pop("llm", getattr(context, "graph_fake_llm", None))
if llm is None:
llm = FakeListLLM(responses=list(_DEFAULT_RESPONSES))
# Default sandbox to /tmp: every fixture in this module materialises
# files under tempfile.mkdtemp(prefix="ctx-graph-cov-") which lands in
# /tmp on Linux CI. Without this, the agent's path-traversal check
# rejects absolute /tmp paths because allowed_base_dir defaults to CWD.
kwargs.setdefault("allowed_base_dir", str(Path("/tmp").resolve()))
context.graph_agent = ContextAnalysisAgent(llm=llm, **kwargs)
return context.graph_agent
@@ -140,6 +147,7 @@ def step_have_agent_custom_chunks(context: Any, chunk_size: int, overlap: int) -
@when("I call load_files with a nonexistent file path")
def step_load_files_missing(context: Any) -> None:
context.graph_agent.allowed_base_dir = Path("/tmp").resolve()
state = _make_state(file_paths=["/tmp/does_not_exist_xyz.py"])
context.load_result = context.graph_agent._load_files(state)
@@ -216,6 +224,7 @@ def step_returned_error_empty(context: Any) -> None:
@when("I call load_files with both a missing file and the directory path")
def step_load_files_mixed_errors(context: Any) -> None:
context.graph_agent.allowed_base_dir = Path("/tmp").resolve()
state = _make_state(
file_paths=[
"/tmp/definitely_missing_abc.py",
@@ -3,6 +3,7 @@
from __future__ import annotations
import tempfile
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
@@ -85,7 +86,12 @@ def step_assert_custom_llm(context: Context) -> None:
@given("a ContextAnalysisAgent instance")
def step_agent_instance(context: Context) -> None:
context.agent = ContextAnalysisAgent(llm=_default_context_test_llm())
# Sandbox to /tmp so absolute /tmp/... paths used by the fixtures below
# (NamedTemporaryFile, /tmp directly) pass the path-traversal check.
context.agent = ContextAnalysisAgent(
llm=_default_context_test_llm(),
allowed_base_dir=str(Path("/tmp").resolve()),
)
@when("I invoke _load_files with preloaded documents")
@@ -123,6 +129,7 @@ def step_assert_loaded(context: Context) -> None:
@when("I invoke _load_files with a nonexistent file path")
def step_load_missing(context: Context) -> None:
context.agent.allowed_base_dir = Path("/tmp").resolve()
state = _empty_state(file_paths=["/tmp/nonexistent_file_xyz.py"])
context.result = context.agent._load_files(state)
@@ -135,6 +142,7 @@ def step_assert_file_not_found(context: Context) -> None:
@when("I invoke _load_files with a directory path")
def step_load_directory(context: Context) -> None:
context.agent.allowed_base_dir = Path("/tmp").resolve()
state = _empty_state(file_paths=["/tmp"])
context.result = context.agent._load_files(state)
@@ -0,0 +1,187 @@
"""Step definitions for ContextAnalysisAgent path traversal security tests.
Covers path validation, traversal prevention, and secure file loading.
"""
from __future__ import annotations
import tempfile
import uuid
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from langchain_community.llms import FakeListLLM
from cleveragents.agents.graphs.context_analysis import (
ContextAnalysisAgent,
ContextAnalysisState,
)
# ---------------------------------------------------------------------------
# Given — Setup
# ---------------------------------------------------------------------------
@given("a temporary directory for testing")
def step_temporary_directory(context: Context) -> None:
"""Create a temporary directory for testing.
Cleanup is handled by ``after_scenario`` in ``features/environment.py``
which removes ``context.test_dir`` automatically.
"""
context.test_dir = Path(tempfile.mkdtemp(prefix="context_analysis_test_"))
context.agent = None
context.result = None
context.errors = []
@given('a test file "{filename}" with content "{content}"')
def step_create_test_file(context: Context, filename: str, content: str) -> None:
"""Create a test file in the temporary directory."""
file_path = context.test_dir / filename
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content)
# ---------------------------------------------------------------------------
# When — Actions
# ---------------------------------------------------------------------------
@when("I create a ContextAnalysisAgent with allowed base directory")
def step_create_agent_with_base_dir(context: Context) -> None:
"""Create a ContextAnalysisAgent with the test directory as allowed base."""
llm = FakeListLLM(
responses=[
"import os, sys",
"0.8",
"Summary of context",
]
)
context.agent = ContextAnalysisAgent(
llm=llm,
chunk_size=2000,
chunk_overlap=200,
allowed_base_dir=str(context.test_dir),
)
def _invoke_config() -> dict[str, dict[str, str]]:
"""Return a config dict with a unique thread_id for the checkpointer."""
return {"configurable": {"thread_id": f"test-{uuid.uuid4().hex[:8]}"}}
@when('I invoke the agent with file path "{file_path}"')
def step_invoke_agent_single_path(context: Context, file_path: str) -> None:
"""Invoke the agent with a single file path."""
state: ContextAnalysisState = {
"file_paths": [file_path],
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
context.result = context.agent.invoke(state, config=_invoke_config())
context.errors = (
context.result["error"].split("; ") if context.result["error"] else []
)
@when("I invoke the agent with file paths:")
def step_invoke_agent_multiple_paths(context: Context) -> None:
"""Invoke the agent with multiple file paths."""
file_paths = [row["path"] for row in context.table]
state: ContextAnalysisState = {
"file_paths": file_paths,
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
context.result = context.agent.invoke(state, config=_invoke_config())
context.errors = (
context.result["error"].split("; ") if context.result["error"] else []
)
# ---------------------------------------------------------------------------
# Then — Assertions
# ---------------------------------------------------------------------------
@then('the agent should reject the path with error "{error_text}"')
def step_assert_path_rejected(context: Context, error_text: str) -> None:
"""Assert that the path was rejected with the expected error."""
assert context.result is not None, "Agent was not invoked"
assert context.result["error"] is not None, "Expected an error but got none"
assert error_text in context.result["error"], (
f"Expected error containing '{error_text}' but got: {context.result['error']}"
)
@then("no files should be loaded")
def step_assert_no_files_loaded(context: Context) -> None:
"""Assert that no files were loaded."""
assert context.result is not None, "Agent was not invoked"
assert len(context.result["documents"]) == 0, (
f"Expected no documents but got {len(context.result['documents'])}"
)
@then("the agent should load the file successfully")
def step_assert_file_loaded(context: Context) -> None:
"""Assert that the file was loaded successfully."""
assert context.result is not None, "Agent was not invoked"
assert len(context.result["documents"]) > 0, "Expected documents to be loaded"
@then('the loaded document should contain "{text}"')
def step_assert_document_contains(context: Context, text: str) -> None:
"""Assert that the loaded document contains the expected text."""
assert context.result is not None, "Agent was not invoked"
assert len(context.result["documents"]) > 0, "No documents were loaded"
content = " ".join(doc.page_content for doc in context.result["documents"])
assert text in content, f"Expected document to contain '{text}' but got: {content}"
@then('the agent should report "{error_text}"')
def step_assert_error_message(context: Context, error_text: str) -> None:
"""Assert that the agent reported the expected error."""
assert context.result is not None, "Agent was not invoked"
assert context.result["error"] is not None, "Expected an error but got none"
assert error_text in context.result["error"], (
f"Expected error containing '{error_text}' but got: {context.result['error']}"
)
@then("the agent should load {count:d} files successfully")
def step_assert_file_count(context: Context, count: int) -> None:
"""Assert that the expected number of files were loaded."""
assert context.result is not None, "Agent was not invoked"
assert len(context.result["documents"]) == count, (
f"Expected {count} documents but got {len(context.result['documents'])}"
)
@then("the agent should report {count:d} error for path traversal")
def step_assert_error_count(context: Context, count: int) -> None:
"""Assert the expected number of path traversal errors were reported."""
assert context.result is not None, "Agent was not invoked"
traversal_errors = [e for e in context.errors if "Path traversal attempt" in e]
assert len(traversal_errors) == count, (
f"Expected {count} path traversal errors "
f"but got {len(traversal_errors)}: {traversal_errors}"
)
@then('the loaded documents should contain "{text}"')
def step_assert_documents_contain(context: Context, text: str) -> None:
"""Assert that the loaded documents contain the expected text."""
assert context.result is not None, "Agent was not invoked"
content = " ".join(doc.page_content for doc in context.result["documents"])
assert text in content, f"Expected documents to contain '{text}' but got: {content}"
@@ -0,0 +1,66 @@
*** Settings ***
Documentation Integration tests for ContextAnalysisAgent path traversal prevention (CWE-22)
Resource ${CURDIR}/common.resource
Library Process
Library OperatingSystem
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${PYTHON} python
${SRC_DIR} ${CURDIR}/../src
${HELPER} ${CURDIR}/helper_context_analysis_path_traversal.py
*** Test Cases ***
Path Traversal Via Dot-Dot Sequences Is Rejected
[Documentation] Verify that file paths containing ``../`` sequences
... that resolve outside the allowed base directory are
... rejected with a clear error and no files are loaded.
[Tags] security path_traversal CWE-22
${result}= Run Process ${PYTHON} ${HELPER} reject_traversal
Log ${result.stdout}
Log ${result.stderr}
Should Contain ${result.stdout} SUCCESS
Should Be Equal As Integers ${result.rc} 0
Absolute Path Escape Outside Allowed Directory Is Rejected
[Documentation] Verify that absolute file paths pointing outside the
... allowed base directory are rejected.
[Tags] security path_traversal CWE-22
${result}= Run Process ${PYTHON} ${HELPER} reject_absolute_escape
Log ${result.stdout}
Log ${result.stderr}
Should Contain ${result.stdout} SUCCESS
Should Be Equal As Integers ${result.rc} 0
Valid Path Within Allowed Directory Loads Successfully
[Documentation] Verify that valid file paths within the allowed base
... directory continue to load correctly (regression guard).
[Tags] security path_traversal CWE-22
${result}= Run Process ${PYTHON} ${HELPER} allow_valid_path
Log ${result.stdout}
Log ${result.stderr}
Should Contain ${result.stdout} SUCCESS
Should Be Equal As Integers ${result.rc} 0
Mixed Valid And Traversal Paths Are Handled Correctly
[Documentation] Verify that when a mix of valid and traversal paths is
... provided, valid files load and traversal paths are
... rejected individually.
[Tags] security path_traversal CWE-22
${result}= Run Process ${PYTHON} ${HELPER} mixed_paths
Log ${result.stdout}
Log ${result.stderr}
Should Contain ${result.stdout} SUCCESS
Should Be Equal As Integers ${result.rc} 0
Symlink Traversal Outside Allowed Directory Is Rejected
[Documentation] Verify that symlinks pointing outside the allowed base
... directory are detected and rejected after resolution.
[Tags] security path_traversal CWE-22
${result}= Run Process ${PYTHON} ${HELPER} symlink_traversal
Log ${result.stdout}
Log ${result.stderr}
Should Contain ${result.stdout} SUCCESS
Should Be Equal As Integers ${result.rc} 0
@@ -0,0 +1,72 @@
Feature: ContextAnalysisAgent path traversal prevention (CWE-22)
Outdated
Review

BLOCKING: Misplaced Feature File

This file should not exist in robot/features/. The project directory rules state:

  • BDD Gherkin .feature files belong in features/ (handled by Behave)
  • robot/ is for Robot Framework .robot files only

This file also lacks the @context_analysis @security @path_traversal tags present on the correct copy at features/context_analysis_path_traversal.feature, and is a near-duplicate of it.

Fix: Delete this file. The Behave feature file already exists at features/context_analysis_path_traversal.feature.


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING: Misplaced Feature File** This file should not exist in `robot/features/`. The project directory rules state: - BDD Gherkin `.feature` files belong in `features/` (handled by Behave) - `robot/` is for Robot Framework `.robot` files only This file also lacks the `@context_analysis @security @path_traversal` tags present on the correct copy at `features/context_analysis_path_traversal.feature`, and is a near-duplicate of it. **Fix:** Delete this file. The Behave feature file already exists at `features/context_analysis_path_traversal.feature`. --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
As a CleverAgents developer
I want to prevent unauthorized file access via directory traversal
So that sensitive files cannot be loaded through crafted paths.
# ── Path validation with relative paths ────────────────────────────────
Scenario: Valid relative file path within allowed base dir is accepted
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "src/main.py"
Then the validation should succeed without error
Scenario: Valid absolute path within allowed base dir is accepted
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "/tmp/ca-test-base/src/utils.py"
Then the validation should succeed without error
# ── Path traversal via .. sequences ────────────────────────────────────
Scenario: Single ../ traversal attempt is rejected
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "../../../etc/passwd"
Then the validation should fail with error containing "Path traversal attempt"
Scenario: Double ../ traversal attempt is rejected
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "../src/../../../etc/shadow"
Then the validation should fail with error containing "Path traversal attempt"
Scenario: Absolute path escaping base dir is rejected
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "/etc/passwd"
Then the validation should fail with error containing "Path traversal attempt"
Scenario: Deep nested valid path is still within base dir
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "deep/nested/path/to/file.py"
Then the validation should succeed without error
# ── Symlink-based traversal prevention ─────────────────────────────────
Scenario: Path containing symlink components is resolved and checked
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "data/../etc/passwd"
Then the validation should fail with error containing "Path traversal attempt"
# ── Edge cases ─────────────────────────────────────────────────────────
Scenario: Empty path components are handled safely
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "."
Then the validation should succeed without error
Scenario: Current directory reference stays within base dir
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I validate the path "src/./main.py"
Then the validation should succeed without error
# ── Integration: _load_files rejects traversal paths ───────────────────
Scenario: load_files rejects path traversal and reports error
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
And a temporary test file at "/tmp/ca-test-safe.py" with content "# safe file"
When I invoke load_files with file paths ["/tmp/ca-test-safe.py", "../../../etc/passwd"]
Then documents should be loaded from valid files only
And error should contain "Path traversal attempt" for invalid paths
Scenario: load_files rejects absolute path to system file
Given a ContextAnalysisAgent with allowed_base_dir "/tmp/ca-test-base"
When I invoke load_files with file paths ["/etc/passwd"]
Then no documents should be loaded from path traversal attempts
And error should contain "Path traversal attempt"
+165 -8
View File
@@ -16,6 +16,7 @@ src_dir = Path(__file__).parent.parent / "src"
sys.path.insert(0, str(src_dir))
from langchain_community.llms import FakeListLLM # noqa: E402
from langchain_core.documents import Document # noqa: E402
from cleveragents.agents.context_analysis import ( # noqa: E402
ContextAnalysisAgent,
@@ -33,7 +34,8 @@ def test_nodes() -> None:
"Relevance: High",
"Summary: test",
]
)
),
allowed_base_dir="/tmp", # Allow /tmp for temp file operations
)
# Get the nodes from the graph
@@ -63,6 +65,150 @@ def test_nodes() -> None:
sys.exit(1)
def test_path_validation_valid() -> None:
"""Test that valid file paths pass validation."""
try:
agent = ContextAnalysisAgent(
llm=FakeListLLM(
responses=[
"Dependencies: ['os']",
"Relevance: High",
"Summary: test",
]
),
allowed_base_dir="/tmp/ca-test-base",
)
# Valid relative path
is_valid, error = agent._validate_file_path("src/main.py")
if not is_valid:
print(f"FAILURE: Valid path rejected: {error}")
sys.exit(1)
# Valid absolute path within base
is_valid, error = agent._validate_file_path("/tmp/ca-test-base/src/utils.py")
if not is_valid:
print(f"FAILURE: Valid absolute path rejected: {error}")
sys.exit(1)
# Current dir reference
is_valid, error = agent._validate_file_path(".")
if not is_valid:
print(f"FAILURE: Current dir rejected: {error}")
sys.exit(1)
print("SUCCESS: Valid paths accepted")
except Exception as e:
print(f"FAILURE: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
def test_path_traversal_rejected() -> None:
"""Test that path traversal attempts are rejected."""
try:
agent = ContextAnalysisAgent(
llm=FakeListLLM(
responses=[
"Dependencies: ['os']",
"Relevance: High",
"Summary: test",
]
),
allowed_base_dir="/tmp/ca-test-base",
)
# Absolute path to system file - MUST be rejected since it escapes base
is_valid, error = agent._validate_file_path("/etc/passwd")
if is_valid:
print("FAILURE: Absolute path escaping base not blocked: /etc/passwd")
sys.exit(1)
if "path traversal" not in str(error or "").lower():
print(f"FAILURE: Expected path traversal error, got: {error}")
sys.exit(1)
# Absolute path to tmp - escapes base entirely
is_valid, error = agent._validate_file_path("/tmp/secret.txt")
if is_valid:
print("FAILURE: Absolute /tmp path not blocked")
sys.exit(1)
# Symlink escape attempt with absolute parent traversal
is_valid, error = agent._validate_file_path("./../../../etc/passwd")
if is_valid:
print("FAILURE: ./../ traversal escaped base")
sys.exit(1)
# Nested traversal that escapes via root sibling
traversal_path = "../ca-test-base/../../../../../../etc/passwd"
is_valid, error = agent._validate_file_path(traversal_path)
if is_valid:
print("FAILURE: Multi-level ../ traversal escaped base")
sys.exit(1)
print("SUCCESS: Path traversal attempts rejected")
print(f"Errors: {error}")
except Exception as e:
print(f"FAILURE: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
def test_load_files_with_preloaded_docs() -> None:
"""Test that pre-loaded documents bypass file loading."""
try:
agent = ContextAnalysisAgent(
llm=FakeListLLM(
responses=[
"Dependencies: ['os']",
"Relevance: High",
"Summary: test",
]
)
)
# Create state with preloaded docs — bypasses load_files entirely
fake_doc = FakeDocument(content="preloaded content")
state: ContextAnalysisState = {
"file_paths": ["nonexistent.py"], # Would fail without preloaded_docs
"documents": [fake_doc],
"dependencies": {},
"summary": "",
"relevance_scores": {},
"chunks": [],
"error": None,
}
config = {"configurable": {"thread_id": "test_preloaded"}}
result = agent.invoke(state, config)
if len(result.get("documents", [])) == 0:
print("FAILURE: Preloaded documents were not preserved")
sys.exit(1)
print("SUCCESS: Preloaded documents bypassed file loading")
except Exception as e:
print(f"FAILURE: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
class FakeDocument(Document):
"""Document subclass that matches langchain_core serialization."""
def __init__(self, content: str = ""):
super().__init__(page_content=content, metadata={"source": "test"})
def test_load_files() -> None:
"""Test file loading functionality."""
try:
@@ -73,7 +219,8 @@ def test_load_files() -> None:
"Relevance: High",
"Summary: test",
]
)
),
allowed_base_dir="/tmp", # Allow temp file operations
)
# Create a temporary test file
@@ -124,12 +271,13 @@ def test_missing_file() -> None:
"Relevance: High",
"Summary: test",
]
)
),
allowed_base_dir="/tmp", # Allow temp dir for missing file checks
)
# Create state with non-existent file
# Create state with non-existent file (relative to base dir)
state: ContextAnalysisState = {
"file_paths": ["/nonexistent/file.py"],
"file_paths": ["nonexistent/file.py"],
"documents": [],
"dependencies": {},
"summary": "",
@@ -170,7 +318,8 @@ def test_invoke() -> None:
"Relevance: High",
"Summary: test",
]
)
),
allowed_base_dir="/tmp", # Allow temp file operations
)
# Create a temporary test file
@@ -244,7 +393,8 @@ def test_streaming() -> None:
"Relevance: High",
"Summary: test",
]
)
),
allowed_base_dir="/tmp", # Allow temp file operations
)
# Create a temporary test file
@@ -294,7 +444,11 @@ def main() -> None:
"""Main entry point for the helper script."""
if len(sys.argv) < 2:
print("Usage: helper_context_analysis.py <test_name>")
print("Available tests: nodes, load_files, missing_file, invoke, streaming")
print(
"Available tests: "
"nodes, load_files, missing_file, invoke, streaming,"
" path_validation_valid, path_traversal_rejected, preloaded_docs"
)
sys.exit(1)
test_name = sys.argv[1]
@@ -305,6 +459,9 @@ def main() -> None:
"missing_file": test_missing_file,
"invoke": test_invoke,
"streaming": test_streaming,
"path_validation_valid": test_path_validation_valid,
"path_traversal_rejected": test_path_traversal_rejected,
"preloaded_docs": test_load_files_with_preloaded_docs,
}
test_func = test_functions.get(test_name)
@@ -0,0 +1,269 @@
#!/usr/bin/env python3
"""Helper script for ContextAnalysisAgent path traversal Robot Framework tests.
Provides integration-level test operations that verify path traversal
prevention works end-to-end through the agent's invoke() method.
"""
import os
import shutil
import sys
import tempfile
from pathlib import Path
from typing import Any
# Add src to path
src_dir = Path(__file__).parent.parent / "src"
sys.path.insert(0, str(src_dir))
from langchain_community.llms import FakeListLLM # noqa: E402
from cleveragents.agents.context_analysis import ( # noqa: E402
ContextAnalysisAgent,
ContextAnalysisState,
)
def _make_agent(base_dir: str) -> ContextAnalysisAgent:
"""Create a ContextAnalysisAgent with a FakeListLLM."""
return ContextAnalysisAgent(
llm=FakeListLLM(
responses=[
"Dependencies: ['os']",
"Relevance: High",
"Summary: test",
]
),
allowed_base_dir=base_dir,
)
def test_reject_traversal() -> None:
"""Path traversal via ``../`` sequences must be rejected."""
tmp = tempfile.mkdtemp(prefix="robot_path_traversal_")
try:
agent = _make_agent(tmp)
state: ContextAnalysisState = {
"file_paths": ["../../../../etc/passwd"],
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
cfg: dict[str, Any] = {
"configurable": {"thread_id": "traversal-1"},
}
result = agent.invoke(state, cfg)
if result["error"] is None:
print("FAILURE: No error for path traversal attempt")
sys.exit(1)
if "Path traversal attempt" not in result["error"]:
print(f"FAILURE: Wrong error: {result['error']}")
sys.exit(1)
if len(result["documents"]) != 0:
print("FAILURE: Documents loaded despite traversal")
sys.exit(1)
print("SUCCESS: Path traversal rejected")
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_reject_absolute_escape() -> None:
"""Absolute paths outside the allowed directory must be rejected."""
tmp = tempfile.mkdtemp(prefix="robot_path_traversal_")
try:
agent = _make_agent(tmp)
# Create a file outside the allowed directory
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False, prefix="outside_"
) as outside_file:
outside_file.write("secret data")
outside_path = outside_file.name
try:
state: ContextAnalysisState = {
"file_paths": [outside_path],
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
cfg: dict[str, Any] = {
"configurable": {"thread_id": "absolute-1"},
}
result = agent.invoke(state, cfg)
if result["error"] is None:
print("FAILURE: No error for absolute path escape")
sys.exit(1)
if "Path traversal attempt" not in result["error"]:
print(f"FAILURE: Wrong error: {result['error']}")
sys.exit(1)
if len(result["documents"]) != 0:
print("FAILURE: Documents loaded despite escape")
sys.exit(1)
print("SUCCESS: Absolute path escape rejected")
finally:
os.unlink(outside_path)
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_allow_valid_path() -> None:
"""Valid paths within the allowed directory must load."""
tmp = tempfile.mkdtemp(prefix="robot_path_traversal_")
try:
valid_file = Path(tmp) / "valid.txt"
valid_file.write_text("valid content")
agent = _make_agent(tmp)
state: ContextAnalysisState = {
"file_paths": ["valid.txt"],
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
cfg: dict[str, Any] = {
"configurable": {"thread_id": "valid-1"},
}
result = agent.invoke(state, cfg)
if len(result["documents"]) == 0:
err = result.get("error")
print(f"FAILURE: No documents loaded. Error: {err}")
sys.exit(1)
content = result["documents"][0].page_content
if "valid content" not in content:
print(f"FAILURE: Wrong content: {content}")
sys.exit(1)
print("SUCCESS: Valid path loaded")
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_mixed_paths() -> None:
"""Mixed valid and traversal paths handled correctly."""
tmp = tempfile.mkdtemp(prefix="robot_path_traversal_")
try:
(Path(tmp) / "good.txt").write_text("good content")
subdir = Path(tmp) / "sub"
subdir.mkdir()
(subdir / "nested.txt").write_text("nested content")
agent = _make_agent(tmp)
state: ContextAnalysisState = {
"file_paths": [
"good.txt",
"../../../../etc/passwd",
"sub/nested.txt",
],
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
cfg: dict[str, Any] = {
"configurable": {"thread_id": "mixed-1"},
}
result = agent.invoke(state, cfg)
if len(result["documents"]) != 2:
err = result.get("error")
print(
f"FAILURE: Expected 2 docs, got "
f"{len(result['documents'])}. Error: {err}"
)
sys.exit(1)
if result["error"] is None or "Path traversal attempt" not in result["error"]:
print(f"FAILURE: Expected error. Got: {result.get('error')}")
sys.exit(1)
print("SUCCESS: Mixed paths handled correctly")
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_symlink_traversal() -> None:
"""Symlinks pointing outside the allowed dir must be rejected."""
tmp = tempfile.mkdtemp(prefix="robot_path_traversal_")
try:
link_path = Path(tmp) / "evil_link"
try:
link_path.symlink_to("/etc")
except OSError:
print("SUCCESS: Symlink test skipped (cannot create)")
return
agent = _make_agent(tmp)
state: ContextAnalysisState = {
"file_paths": ["evil_link/passwd"],
"documents": [],
"dependencies": {},
"chunks": [],
"relevance_scores": {},
"summary": "",
"error": None,
}
cfg: dict[str, Any] = {
"configurable": {"thread_id": "symlink-1"},
}
result = agent.invoke(state, cfg)
if result["error"] is None:
print("FAILURE: No error for symlink traversal")
sys.exit(1)
if "Path traversal attempt" not in result["error"]:
print(f"FAILURE: Wrong error: {result['error']}")
sys.exit(1)
if len(result["documents"]) != 0:
print("FAILURE: Documents loaded via symlink")
sys.exit(1)
print("SUCCESS: Symlink traversal rejected")
finally:
shutil.rmtree(tmp, ignore_errors=True)
def main() -> None:
"""Main entry point for the helper script."""
if len(sys.argv) < 2:
print("Usage: helper_context_analysis_path_traversal.py <test>")
print(
"Available: reject_traversal, reject_absolute_escape, "
"allow_valid_path, mixed_paths, symlink_traversal"
)
sys.exit(1)
tests = {
"reject_traversal": test_reject_traversal,
"reject_absolute_escape": test_reject_absolute_escape,
"allow_valid_path": test_allow_valid_path,
"mixed_paths": test_mixed_paths,
"symlink_traversal": test_symlink_traversal,
}
func = tests.get(sys.argv[1])
if func is None:
print(f"Unknown test: {sys.argv[1]}")
print(f"Available: {', '.join(tests.keys())}")
sys.exit(1)
func()
if __name__ == "__main__":
main()
@@ -98,6 +98,7 @@ class ContextAnalysisAgent:
chunk_overlap: int = 200,
retry_attempts: int = 3,
max_dependencies: int = 10,
allowed_base_dir: str | None = None,
):
"""Initialize the context analysis agent.
@@ -109,6 +110,9 @@ class ContextAnalysisAgent:
max_dependencies: Maximum number of dependencies returned by
``_parse_dependencies``. Must be a positive integer.
Defaults to ``10``.
allowed_base_dir: Base directory for file path validation
(prevents path traversal). If None, defaults to current
working directory.
Raises:
ValueError: If ``max_dependencies`` is not a positive integer.
@@ -122,6 +126,14 @@ class ContextAnalysisAgent:
self.retry_attempts = max(1, retry_attempts)
self.max_dependencies = max_dependencies
# Set allowed base directory for path validation.
# Path.resolve() follows symlinks and normalises the path,
# preventing symlink-based sandbox escapes.
if allowed_base_dir is None:
self.allowed_base_dir: Path = Path.cwd().resolve()
else:
self.allowed_base_dir = Path(allowed_base_dir).resolve()
# Initialize LLM - an LLM must be provided explicitly
if llm is None:
raise ValueError(
@@ -140,11 +152,43 @@ class ContextAnalysisAgent:
self.checkpointer = MemorySaver()
self.app = self.graph.compile(checkpointer=self.checkpointer)
def _validate_file_path(self, file_path: str) -> tuple[bool, str | None]:
"""Validate that a file path is within the allowed base directory.
This prevents path traversal attacks (CWE-22) by resolving the
candidate path (following symlinks) and verifying it is a descendant
of ``self.allowed_base_dir`` using ``Path.relative_to()``.
Args:
file_path: The file path to validate
Returns:
A tuple of (is_valid, error_message). If valid, error_message
is None.
"""
try:
# Resolve the full path, following symlinks, to prevent both
# ".." traversal and symlink-based sandbox escapes.
resolved = Path(self.allowed_base_dir, file_path).resolve()
# Path.relative_to() raises ValueError when *resolved* is not
# a descendant of the base directory — this is the recommended
# containment check (immune to prefix-collision attacks that
# affect os.path.commonpath / str.startswith approaches).
resolved.relative_to(self.allowed_base_dir)
return True, None
except ValueError:
return False, f"Path traversal attempt: {file_path}"
except OSError as exc:
return False, f"Invalid path: {file_path} ({exc!s})"
def _create_prompts(self) -> None:
"""Create prompt templates for each workflow node."""
self.dependency_prompt: Any = PromptTemplate(
template=(
"Analyze the following code and extract all imports and dependencies.\n"
"Analyze the following code and extract all imports and "
"dependencies.\n"
"List them in a structured format.\n\n"
"Code:\n"
"{code}\n\n"
@@ -168,7 +212,8 @@ class ContextAnalysisAgent:
self.summary_prompt: Any = PromptTemplate(
template=(
"Provide a high-level summary of the following codebase context.\n\n"
"Provide a high-level summary of the following codebase "
"context.\n\n"
"Files analyzed: {file_count}\n"
"Total size: {total_size} characters\n"
"Dependencies found: {dependency_count}\n\n"
@@ -236,16 +281,25 @@ class ContextAnalysisAgent:
for file_path in state["file_paths"]:
try:
path = Path(file_path)
if not path.exists():
# Validate file path to prevent path traversal
is_valid, validation_error = self._validate_file_path(file_path)
if not is_valid:
errors.append(validation_error or f"Invalid path: {file_path}")
continue
# Use the resolved path for all file operations to ensure
# we load the file we actually validated.
resolved_path = Path(self.allowed_base_dir, file_path).resolve()
if not resolved_path.exists():
errors.append(f"File not found: {file_path}")
continue
if not path.is_file():
if not resolved_path.is_file():
errors.append(f"Not a file: {file_path}")
continue
loader = TextLoader(str(path))
loader = TextLoader(str(resolved_path))
loaded_docs: list[Document] = loader.load()
documents.extend(loaded_docs)
except Exception as exc: # pragma: no cover - defensive
@@ -617,12 +617,19 @@ class ContextService:
# --- LangGraph-based Context Analysis Methods ---
def _get_context_agent(
self, llm: BaseLanguageModel | None = None
self,
llm: BaseLanguageModel | None = None,
allowed_base_dir: str | None = None,
) -> ContextAnalysisAgent:
"""Get or create a ContextAnalysisAgent instance.
Args:
llm: Optional language model to use. If None, uses default mock.
allowed_base_dir: Base directory for path-traversal sandboxing
(defaults to the agent's CWD when None). The analyze_context
methods supply ``project.path`` so files inside a project's
workspace pass validation while paths escaping it are
rejected.
Returns:
ContextAnalysisAgent instance for analyzing context.
@@ -630,7 +637,7 @@ class ContextService:
# Import here to avoid circular dependency and allow lazy loading
from cleveragents.agents.context_analysis import ContextAnalysisAgent
return ContextAnalysisAgent(llm=llm)
return ContextAnalysisAgent(llm=llm, allowed_base_dir=allowed_base_dir)
def analyze_context(
self,
@@ -676,8 +683,10 @@ class ContextService:
error=None,
)
# Create agent and run analysis
agent = self._get_context_agent(llm)
# Create agent and run analysis. Pass the project's workspace as
# the path-validation sandbox so absolute paths inside the project
# are accepted while traversal outside it is blocked.
agent = self._get_context_agent(llm, allowed_base_dir=str(project.path))
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context",
@@ -728,7 +737,7 @@ class ContextService:
error=None,
)
agent = self._get_context_agent(llm)
agent = self._get_context_agent(llm, allowed_base_dir=str(project.path))
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context_async",
@@ -771,7 +780,7 @@ class ContextService:
yield {"type": "complete", "summary": "No context files to analyze"}
return
agent = self._get_context_agent(llm)
agent = self._get_context_agent(llm, allowed_base_dir=str(project.path))
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context_streaming",
@@ -815,7 +824,7 @@ class ContextService:
yield {"type": "complete", "summary": "No context files to analyze"}
return
agent = self._get_context_agent(llm)
agent = self._get_context_agent(llm, allowed_base_dir=str(project.path))
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context_streaming_async",