fix(python/ssh): confine known_hosts_file to cwd or ~/.ssh

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

@@ -0,0 +1,36 @@
+"""Local path confinement for SSH file actions."""
+
+from __future__ import annotations
+
+import os
+
+
+def resolve_safe_local_path(path: str, *, allowed_roots: list[str] | None = None) -> str:
+    """Resolve ``path`` and require it to stay under an allowed root.
+
+    Default root is ``os.getcwd()``. Callers may add extra roots (e.g. ``~/.ssh``
+    for known_hosts). Twin of the TypeScript twitter/flaunch/zora cwd confine.
+    """
+    if not path or not str(path).strip():
+        raise ValueError("Local path must be within an allowed directory")
+
+    cwd = os.path.realpath(os.getcwd())
+    roots = [cwd]
+    if allowed_roots:
+        for root in allowed_roots:
+            roots.append(os.path.realpath(os.path.expanduser(root)))
+
+    expanded = os.path.expanduser(path)
+    # Match path.resolve(cwd, relativeOrAbsolute): absolute inputs ignore cwd.
+    resolved = os.path.normpath(
+        os.path.join(cwd, expanded) if not os.path.isabs(expanded) else expanded
+    )
+
+    if not any(resolved == root or resolved.startswith(root + os.sep) for root in roots):
+        raise ValueError("Local path must be within an allowed directory")
+
+    # realpath closes symlink escapes for existing path prefixes.
+    real = os.path.realpath(resolved)
+    if not any(real == root or real.startswith(root + os.sep) for root in roots):
+        raise ValueError("Local path escapes the allowed directories")
+    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):
@@ -413,6 +414,8 @@ def ssh_upload(self, args: dict[str, Any]) -> str:
             return f"Error: I/O operation: {e!s}"
         except ValidationError as e:
             return f"Error: Invalid input parameters: {e!s}"
+        except ValueError as e:
+            return f"Error: {e!s}"
         except Exception as e:
             return f"Error: File upload: {e!s}"
 
@@ -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)
@@ -491,6 +492,8 @@ def ssh_download(self, args: dict[str, Any]) -> str:
             return f"Error: I/O operation: {e!s}"
         except ValidationError as e:
             return f"Error: Invalid input parameters: {e!s}"
+        except ValueError as e:
+            return f"Error: {e!s}"
         except Exception as e:
             return f"Error: File download: {e!s}"
 
@@ -538,7 +541,11 @@ def ssh_add_host_key(self, args: dict[str, Any]) -> str:
             host = validated_args.host
             key = validated_args.key
             key_type = validated_args.key_type
-            known_hosts_file = os.path.expanduser(validated_args.known_hosts_file)
+            ssh_dir = os.path.join(os.path.expanduser("~"), ".ssh")
+            known_hosts_file = resolve_safe_local_path(
+                validated_args.known_hosts_file,
+                allowed_roots=[ssh_dir],
+            )
 
             host_entry = host
             entry = f"{host_entry} {key_type} {key}\n"
@@ -571,6 +578,8 @@ def ssh_add_host_key(self, args: dict[str, Any]) -> str:
 
         except ValidationError as e:
             return f"Error: Invalid input parameters: {e!s}"
+        except ValueError as e:
+            return f"Error: {e!s}"
         except FileNotFoundError as e:
             return f"Error: Unable to access known_hosts file: {e!s}"
         except PermissionError as e:

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

@@ -5,24 +5,22 @@
 """
 
 import os
-import tempfile
 from unittest import mock
 
 import pytest
 
 
 @pytest.fixture
-def temp_known_hosts():
-    """Create a temporary known_hosts file for testing."""
-    with tempfile.NamedTemporaryFile(mode="w+", delete=False) as temp_file:
-        temp_file.write("existing.example.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQ==\n")
-        temp_file.write("other.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIHRVs==\n")
-        temp_file_path = temp_file.name
-
-    yield temp_file_path
-
-    if os.path.exists(temp_file_path):
-        os.unlink(temp_file_path)
+def temp_known_hosts(tmp_path, monkeypatch):
+    """Create a temporary known_hosts file under cwd for testing."""
+    monkeypatch.chdir(tmp_path)
+    hosts = tmp_path / "known_hosts"
+    hosts.write_text(
+        "existing.example.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQ==\n"
+        "other.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIHRVs==\n",
+        encoding="utf-8",
+    )
+    return str(hosts.resolve())
 
 
 def test_add_host_key_basic(ssh_provider, temp_known_hosts):
@@ -102,27 +100,42 @@ def test_add_host_key_with_custom_key_type(ssh_provider, temp_known_hosts):
     assert "keytype.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIHRVs==" in content
 
 
-def test_add_host_key_create_file(ssh_provider):
+def test_add_host_key_create_file(ssh_provider, tmp_path, monkeypatch):
     """Test adding a host key when the known_hosts file doesn't exist."""
-    with tempfile.TemporaryDirectory() as temp_dir:
-        new_file_path = os.path.join(temp_dir, "new_known_hosts")
+    monkeypatch.chdir(tmp_path)
+    new_file_path = str((tmp_path / "new_known_hosts").resolve())
 
-        result = ssh_provider.ssh_add_host_key(
-            {
-                "host": "new.example.com",
-                "key": "AAAAB3NzaC1yc2EAAAADAQABAAABAQ==",
-                "known_hosts_file": new_file_path,
-            }
-        )
+    result = ssh_provider.ssh_add_host_key(
+        {
+            "host": "new.example.com",
+            "key": "AAAAB3NzaC1yc2EAAAADAQABAAABAQ==",
+            "known_hosts_file": "new_known_hosts",
+        }
+    )
 
-        assert "successfully added" in result
-        assert "Host key for 'new.example.com'" in result
+    assert "successfully added" in result
+    assert "Host key for 'new.example.com'" in result
+
+    assert os.path.exists(new_file_path)
+    with open(new_file_path) as f:
+        content = f.read()
 
-        assert os.path.exists(new_file_path)
-        with open(new_file_path) as f:
-            content = f.read()
+    assert "new.example.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQ==" in content
 
-        assert "new.example.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQ==" in content
+
+def test_add_host_key_rejects_path_outside_allowed_roots(ssh_provider, tmp_path, monkeypatch):
+    """Test known_hosts_file outside cwd and ~/.ssh is rejected."""
+    monkeypatch.chdir(tmp_path)
+
+    result = ssh_provider.ssh_add_host_key(
+        {
+            "host": "evil.example.com",
+            "key": "AAAAB3NzaC1yc2EAAAADAQABAAABAQ==",
+            "known_hosts_file": "/etc/agentkit-known-hosts",
+        }
+    )
+
+    assert "allowed directory" in result
 
 
 def test_add_host_key_invalid_params(ssh_provider):
@@ -136,21 +149,25 @@ def test_add_host_key_invalid_params(ssh_provider):
     assert "Invalid input parameters" in result
 
 
-def test_add_host_key_file_error(ssh_provider):
+def test_add_host_key_file_error(ssh_provider, tmp_path, monkeypatch):
     """Test handling file access errors."""
+    monkeypatch.chdir(tmp_path)
+    hosts = tmp_path / "known_hosts"
+    hosts.write_text("", encoding="utf-8")
+
     with (
         mock.patch("os.path.exists") as mock_exists,
         mock.patch("os.makedirs"),
         mock.patch("builtins.open") as mock_open,
     ):
         mock_exists.return_value = True
-
         mock_open.side_effect = OSError("Permission denied")
 
         result = ssh_provider.ssh_add_host_key(
             {
                 "host": "error.example.com",
                 "key": "AAAAB3NzaC1yc2EAAAADAQABAAABAQ==",
+                "known_hosts_file": str(hosts),
             }
         )
 

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 allowed 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 "allowed 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)