diff --git a/envs/finqa_env/server/tools.py b/envs/finqa_env/server/tools.py index 6d827f722..b51e818cb 100644 --- a/envs/finqa_env/server/tools.py +++ b/envs/finqa_env/server/tools.py @@ -48,6 +48,34 @@ def get_available_companies(self) -> List[str]: if os.path.isdir(os.path.join(self.companies_path, d)) ] + def _resolve_within(self, *parts: str) -> str | None: + """ + Join agent-supplied path components under ``companies_path`` and confirm the + result stays inside it. + + The tool arguments (``company_name``, ``table_name``) are supplied by the agent + over the MCP boundary. Joining them into a filesystem path without a containment + check allows ``..`` traversal to read files outside the data directory. This + rejects any such escape. + + Args: + parts (`str`): + Path components to join beneath ``companies_path``. + + Returns: + `str` or `None`: the resolved absolute path if it stays within + ``companies_path``, otherwise `None`. + """ + root = os.path.realpath(self.companies_path) + candidate = os.path.realpath(os.path.join(self.companies_path, *parts)) + try: + if candidate != root and os.path.commonpath([root, candidate]) == root: + return candidate + except ValueError: + # Raised when paths cannot be compared (e.g. different drives on Windows). + pass + return None + def execute_tool( self, tool_name: str, tool_args: Dict[str, Any] ) -> Tuple[str, bool]: @@ -82,9 +110,9 @@ def get_descriptions(self, company_name: str) -> str: Returns: JSON list of table names """ - company_path = os.path.join(self.companies_path, company_name) + company_path = self._resolve_within(company_name) - if not os.path.isdir(company_path): + if company_path is None or not os.path.isdir(company_path): available = self.get_available_companies() return ( f"Error: '{company_name}' not found. Available companies: {available}" @@ -109,9 +137,9 @@ def get_table_info(self, company_name: str, table_name: str) -> str: Returns: JSON string with table metadata (description, columns, dtypes, unique values) """ - company_path = os.path.join(self.companies_path, company_name) + company_path = self._resolve_within(company_name) - if not os.path.isdir(company_path): + if company_path is None or not os.path.isdir(company_path): available = self.get_available_companies() return ( f"Error: '{company_name}' not found. Available companies: {available}" @@ -211,12 +239,10 @@ def sql_query(self, company_name: str, table_name: str, query: str) -> str: # Clean table name cleaned_table_name = table_name.replace(".txt", "").replace(".json", "") - table_path = os.path.join( - self.companies_path, company_name, f"{cleaned_table_name}.json" - ) + table_path = self._resolve_within(company_name, f"{cleaned_table_name}.json") - if not os.path.isfile(table_path): - return f"Error: Table file not found at {table_path}" + if table_path is None or not os.path.isfile(table_path): + return f"Error: Table file not found for '{company_name}/{table_name}'" try: # Load table and execute query diff --git a/tests/envs/test_finqa_environment.py b/tests/envs/test_finqa_environment.py index 5396dff1a..366307dc4 100644 --- a/tests/envs/test_finqa_environment.py +++ b/tests/envs/test_finqa_environment.py @@ -553,6 +553,52 @@ def test_sql_query_no_filter(self, tools): assert "Error" in result +class TestToolsPathTraversal: + """Agent-supplied company/table names must not escape the data directory (CWE-22). + + Self-contained (synthetic data, no downloaded dataset needed) so it runs in CI. + """ + + @pytest.fixture + def tools(self, tmp_path): + pytest.importorskip("pandas") + from envs.finqa_env.server.tools import FinQATools + + company = tmp_path / "input_companies" / "acme" + company.mkdir(parents=True) + (company / "revenue.json").write_text('[{"x": 1}]') + # A file the agent must never be able to reach, placed outside input_companies/. + (tmp_path / "secret.json").write_text('[{"password": "leaked-secret"}]') + return FinQATools(str(tmp_path)) + + def test_legit_access_still_works(self, tools): + assert "Error" not in tools.get_descriptions("acme") + assert "leaked" not in tools.sql_query( + "acme", "revenue", "SELECT x FROM revenue WHERE x = 1" + ) + + def test_get_descriptions_rejects_traversal(self, tools): + for evil in ["..", "../..", "../../../../etc"]: + result = tools.get_descriptions(evil) + assert "Error" in result + assert "secret" not in result + + def test_sql_query_rejects_company_traversal(self, tools): + result = tools.sql_query( + "..", "secret", "SELECT password FROM secret WHERE 1 = 1" + ) + assert "leaked-secret" not in result + + def test_sql_query_rejects_table_traversal(self, tools): + result = tools.sql_query( + "acme", "../../secret", "SELECT password FROM x WHERE 1 = 1" + ) + assert "leaked-secret" not in result + + def test_get_table_info_rejects_traversal(self, tools): + assert "Error" in tools.get_table_info("../..", "whatever") + + @pytest.mark.skipif(_integration_skip, reason=_integration_reason) class TestEnvironment: """Test environment logic using MCP actions."""