fix(finqa_env): prevent path traversal via agent-supplied company/table names

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

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."""