fix(python/ssh): confine upload/download local paths to cwd

python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/path_utils.py

@@ -0,0 +1,28 @@
+"""Local path confinement for SSH file actions."""
+
+from __future__ import annotations
+
+import os
+
+
+def resolve_safe_local_path(path: str) -> str:
+    """Resolve ``path`` and require it to stay under ``os.getcwd()``.
+
+    Twin of the TypeScript twitter/flaunch/zora cwd confine: agent-supplied
+    local paths must not read or write arbitrary process-readable files.
+    """
+    if not path or not str(path).strip():
+        raise ValueError("Local path must be within the working directory")
+
+    root = os.path.realpath(os.getcwd())
+    expanded = os.path.expanduser(path)
+    # Match path.resolve(root, relativeOrAbsolute): absolute inputs ignore root.
+    resolved = os.path.normpath(os.path.join(root, expanded) if not os.path.isabs(expanded) else expanded)
+    if resolved != root and not resolved.startswith(root + os.sep):
+        raise ValueError("Local path must be within the working directory")
+
+    # realpath closes symlink escapes for existing path prefixes.
+    real = os.path.realpath(resolved)
+    if real != root and not real.startswith(root + os.sep):
+        raise ValueError("Local path escapes the working directory")
+    return real

python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/ssh_action_provider.py

@@ -19,6 +19,7 @@
 from ..action_provider import ActionProvider
 from .connection import SSHConnectionError, SSHKeyError, UnknownHostKeyError
 from .connection_pool import SSHConnectionPool
+from .path_utils import resolve_safe_local_path
 from .schemas import (
     AddHostKeySchema,
     ConnectionStatusSchema,
@@ -380,7 +381,7 @@ def ssh_upload(self, args: dict[str, Any]) -> str:
         try:
             validated_args = FileUploadSchema(**args)
             connection_id = validated_args.connection_id
-            local_path = validated_args.local_path
+            local_path = resolve_safe_local_path(validated_args.local_path)
             remote_path = validated_args.remote_path
 
             if not self.connection_pool.has_connection(connection_id):
@@ -411,6 +412,8 @@ def ssh_upload(self, args: dict[str, Any]) -> str:
             return f"Error: SFTP operation: {e!s}"
         except OSError as e:
             return f"Error: I/O operation: {e!s}"
+        except ValueError as e:
+            return f"Error: {e!s}"
         except ValidationError as e:
             return f"Error: Invalid input parameters: {e!s}"
         except Exception as e:
@@ -459,7 +462,7 @@ def ssh_download(self, args: dict[str, Any]) -> str:
             validated_args = FileDownloadSchema(**args)
             connection_id = validated_args.connection_id
             remote_path = validated_args.remote_path
-            local_path = validated_args.local_path
+            local_path = resolve_safe_local_path(validated_args.local_path)
 
             if not self.connection_pool.has_connection(connection_id):
                 return f"Error: Connection ID '{connection_id}' not found. Use ssh_connect first."
@@ -469,8 +472,6 @@ def ssh_download(self, args: dict[str, Any]) -> str:
             if not connection.is_connected():
                 return f"Error: Connection '{connection_id}' is not currently active. Use ssh_connect to establish the connection."
 
-            local_path = os.path.expanduser(local_path)
-
             local_dir = os.path.dirname(local_path)
             if local_dir and not os.path.exists(local_dir):
                 os.makedirs(local_dir)
@@ -489,6 +490,8 @@ def ssh_download(self, args: dict[str, Any]) -> str:
             return f"Error: SFTP operation: {e!s}"
         except OSError as e:
             return f"Error: I/O operation: {e!s}"
+        except ValueError as e:
+            return f"Error: {e!s}"
         except ValidationError as e:
             return f"Error: Invalid input parameters: {e!s}"
         except Exception as e:

python/coinbase-agentkit/tests/action_providers/ssh/test_download.py

@@ -9,57 +9,75 @@
 from coinbase_agentkit.action_providers.ssh.connection import SSHConnectionError
 
 
-def test_ssh_download_success(ssh_provider):
+def test_ssh_download_success(ssh_provider, tmp_path, monkeypatch):
     """Test successful file download."""
+    monkeypatch.chdir(tmp_path)
+    local_path = str((tmp_path / "out.txt").resolve())
+
     mock_pool = ssh_provider.connection_pool
     mock_connection = mock.Mock()
+    mock_pool.has_connection.return_value = True
+    mock_pool.get_connection.return_value = mock_connection
+    mock_connection.is_connected.return_value = True
 
-    with (
-        mock.patch("os.path.dirname", return_value="/local/directory"),
-        mock.patch("os.makedirs"),
-        mock.patch("os.path.expanduser", return_value="/local/path"),
-    ):
-        mock_pool.has_connection.return_value = True
-        mock_pool.get_connection.return_value = mock_connection
-        mock_connection.is_connected.return_value = True
-
-        result = ssh_provider.ssh_download(
-            {
-                "connection_id": "test-conn",
-                "remote_path": "/remote/path",
-                "local_path": "/local/path",
-            }
-        )
-
-        assert "File download successful" in result
-        assert "/remote/path" in result
-        assert "/local/path" in result
-        mock_connection.download_file.assert_called_once_with("/remote/path", "/local/path")
-
-
-def test_ssh_download_connection_not_found(ssh_provider):
-    """Test file download with connection not found."""
+    result = ssh_provider.ssh_download(
+        {
+            "connection_id": "test-conn",
+            "remote_path": "/remote/path",
+            "local_path": "out.txt",
+        }
+    )
+
+    assert "File download successful" in result
+    assert "/remote/path" in result
+    assert local_path in result
+    mock_connection.download_file.assert_called_once_with("/remote/path", local_path)
+
+
+def test_ssh_download_rejects_path_outside_cwd(ssh_provider, tmp_path, monkeypatch):
+    """Test file download rejects paths outside the working directory."""
+    monkeypatch.chdir(tmp_path)
     mock_pool = ssh_provider.connection_pool
+    mock_connection = mock.Mock()
+    mock_pool.has_connection.return_value = True
+    mock_pool.get_connection.return_value = mock_connection
+    mock_connection.is_connected.return_value = True
+
+    result = ssh_provider.ssh_download(
+        {
+            "connection_id": "test-conn",
+            "remote_path": "/remote/path",
+            "local_path": "/etc/agentkit-ssh-download",
+        }
+    )
+
+    assert "working directory" in result
+    mock_connection.download_file.assert_not_called()
 
+
+def test_ssh_download_connection_not_found(ssh_provider, tmp_path, monkeypatch):
+    """Test file download with connection not found."""
+    monkeypatch.chdir(tmp_path)
+    mock_pool = ssh_provider.connection_pool
     mock_pool.has_connection.return_value = False
 
     result = ssh_provider.ssh_download(
         {
             "connection_id": "test-conn",
             "remote_path": "/remote/path",
-            "local_path": "/local/path",
+            "local_path": "out.txt",
         }
     )
 
     assert "Error: Connection ID 'test-conn' not found" in result
     mock_pool.has_connection.assert_called_once_with("test-conn")
 
 
-def test_ssh_download_not_connected(ssh_provider):
+def test_ssh_download_not_connected(ssh_provider, tmp_path, monkeypatch):
     """Test file download with inactive connection."""
+    monkeypatch.chdir(tmp_path)
     mock_pool = ssh_provider.connection_pool
     mock_connection = mock.Mock()
-
     mock_pool.has_connection.return_value = True
     mock_pool.get_connection.return_value = mock_connection
     mock_connection.is_connected.return_value = False
@@ -68,7 +86,7 @@ def test_ssh_download_not_connected(ssh_provider):
         {
             "connection_id": "test-conn",
             "remote_path": "/remote/path",
-            "local_path": "/local/path",
+            "local_path": "out.txt",
         }
     )
 
@@ -77,29 +95,25 @@ def test_ssh_download_not_connected(ssh_provider):
     mock_connection.is_connected.assert_called_once()
 
 
-def test_ssh_download_error(ssh_provider):
+def test_ssh_download_error(ssh_provider, tmp_path, monkeypatch):
     """Test file download with error."""
+    monkeypatch.chdir(tmp_path)
+    local_path = str((tmp_path / "out.txt").resolve())
+
     mock_pool = ssh_provider.connection_pool
     mock_connection = mock.Mock()
+    mock_pool.has_connection.return_value = True
+    mock_pool.get_connection.return_value = mock_connection
+    mock_connection.is_connected.return_value = True
+    mock_connection.download_file.side_effect = SSHConnectionError("Download failed")
+
+    result = ssh_provider.ssh_download(
+        {
+            "connection_id": "test-conn",
+            "remote_path": "/remote/path",
+            "local_path": "out.txt",
+        }
+    )
 
-    with (
-        mock.patch("os.path.dirname", return_value="/local/directory"),
-        mock.patch("os.makedirs"),
-        mock.patch("os.path.expanduser", return_value="/local/path"),
-    ):
-        mock_pool.has_connection.return_value = True
-        mock_pool.get_connection.return_value = mock_connection
-        mock_connection.is_connected.return_value = True
-        mock_connection.download_file.side_effect = SSHConnectionError("Download failed")
-
-        result = ssh_provider.ssh_download(
-            {
-                "connection_id": "test-conn",
-                "remote_path": "/remote/path",
-                "local_path": "/local/path",
-            }
-        )
-
-        assert "Error: SSH connection:" in result
-        assert "Download failed" in result
-        mock_connection.download_file.assert_called_once_with("/remote/path", "/local/path")
+    assert "Error: SSH connection" in result
+    mock_connection.download_file.assert_called_once_with("/remote/path", local_path)

python/coinbase-agentkit/tests/action_providers/ssh/test_path_utils.py

@@ -0,0 +1,27 @@
+"""Tests for SSH local path confinement."""
+
+import os
+from pathlib import Path
+
+import pytest
+
+from coinbase_agentkit.action_providers.ssh.path_utils import resolve_safe_local_path
+
+
+def test_resolve_safe_local_path_allows_under_cwd(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    inside = tmp_path / "payload.txt"
+    inside.write_text("x", encoding="utf-8")
+
+    assert resolve_safe_local_path("payload.txt") == str(inside.resolve())
+    assert resolve_safe_local_path(str(inside)) == str(inside.resolve())
+
+
+def test_resolve_safe_local_path_rejects_escapes(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ValueError, match="working directory"):
+        resolve_safe_local_path("../outside.txt")
+
+    with pytest.raises(ValueError, match="working directory"):
+        resolve_safe_local_path("/etc/passwd")