From 4050b9b33bc5e4173ee0a539178fd97761972ac6 Mon Sep 17 00:00:00 2001 From: OfficialAbhinavSingh Date: Fri, 24 Jul 2026 10:02:19 +0530 Subject: [PATCH] fix(finqa_env): prevent path traversal via agent-supplied company/table names The finqa MCP tools (get_descriptions, get_table_info, sql_query) joined the agent-controlled company_name/table_name into a filesystem path guarded only by an existence check (os.path.isdir/isfile), not a containment check. An agent could pass '..' components to enumerate directories and read arbitrary .json files outside the data directory (e.g. withheld answer data or mounted secrets) over the MCP boundary. Add a _resolve_within() helper that resolves the joined path and rejects any result that escapes companies_path, mirroring the intent of the existing pointer guards. Applied to all three tools. Adds a self-contained regression test (no downloaded dataset required). --- envs/finqa_env/server/tools.py | 44 ++++++++++++++++++++------ tests/envs/test_finqa_environment.py | 46 ++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 9 deletions(-) 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."""