From 916e885636c13a06d1a7e0909b86d22f10d342c6 Mon Sep 17 00:00:00 2001 From: SashaMIT Date: Sun, 9 Aug 2026 15:51:13 +0700 Subject: [PATCH] fix(python/ssh): confine upload/download local paths to cwd Resolve agent-supplied local_path under process cwd (realpath) before SFTP upload/download so prompt-injected paths cannot read or write arbitrary files. --- .../action_providers/ssh/path_utils.py | 28 +++ .../ssh/ssh_action_provider.py | 11 +- .../action_providers/ssh/test_download.py | 118 ++++++------ .../action_providers/ssh/test_path_utils.py | 27 +++ .../tests/action_providers/ssh/test_upload.py | 174 ++++++++++-------- 5 files changed, 222 insertions(+), 136 deletions(-) create mode 100644 python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/path_utils.py create mode 100644 python/coinbase-agentkit/tests/action_providers/ssh/test_path_utils.py diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/path_utils.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/path_utils.py new file mode 100644 index 000000000..21db5af55 --- /dev/null +++ b/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 diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/ssh_action_provider.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/ssh_action_provider.py index 274f15375..ecc76fecb 100644 --- a/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/ssh_action_provider.py +++ b/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: diff --git a/python/coinbase-agentkit/tests/action_providers/ssh/test_download.py b/python/coinbase-agentkit/tests/action_providers/ssh/test_download.py index bcd1efa96..c5be595ba 100644 --- a/python/coinbase-agentkit/tests/action_providers/ssh/test_download.py +++ b/python/coinbase-agentkit/tests/action_providers/ssh/test_download.py @@ -9,45 +9,63 @@ 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", } ) @@ -55,11 +73,11 @@ def test_ssh_download_connection_not_found(ssh_provider): 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) diff --git a/python/coinbase-agentkit/tests/action_providers/ssh/test_path_utils.py b/python/coinbase-agentkit/tests/action_providers/ssh/test_path_utils.py new file mode 100644 index 000000000..fec32e5f0 --- /dev/null +++ b/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") diff --git a/python/coinbase-agentkit/tests/action_providers/ssh/test_upload.py b/python/coinbase-agentkit/tests/action_providers/ssh/test_upload.py index 6066639ed..258ed88d5 100644 --- a/python/coinbase-agentkit/tests/action_providers/ssh/test_upload.py +++ b/python/coinbase-agentkit/tests/action_providers/ssh/test_upload.py @@ -9,103 +9,117 @@ from coinbase_agentkit.action_providers.ssh.connection import SSHConnectionError -def test_ssh_upload_success(ssh_provider): +def test_ssh_upload_success(ssh_provider, tmp_path, monkeypatch): """Test successful file upload.""" + monkeypatch.chdir(tmp_path) + local = tmp_path / "payload.txt" + local.write_text("x", encoding="utf-8") + local_path = str(local.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 + + result = ssh_provider.ssh_upload( + { + "connection_id": "test-conn", + "local_path": "payload.txt", + "remote_path": "/remote/path", + } + ) + + assert "File upload successful" in result + assert local_path in result + assert "/remote/path" in result + mock_connection.upload_file.assert_called_once_with(local_path, "/remote/path") + + +def test_ssh_upload_rejects_path_outside_cwd(ssh_provider, tmp_path, monkeypatch): + """Test file upload 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_upload( + { + "connection_id": "test-conn", + "local_path": "/etc/passwd", + "remote_path": "/remote/path", + } + ) + + assert "working directory" in result + mock_connection.upload_file.assert_not_called() - with ( - mock.patch("os.path.exists", return_value=True), - mock.patch("os.path.isfile", return_value=True), - ): - 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_upload( - { - "connection_id": "test-conn", - "local_path": "/local/path", - "remote_path": "/remote/path", - } - ) - - assert "File upload successful" in result - assert "/local/path" in result - assert "/remote/path" in result - mock_connection.upload_file.assert_called_once_with("/local/path", "/remote/path") - - -def test_ssh_upload_connection_not_found(ssh_provider): + +def test_ssh_upload_connection_not_found(ssh_provider, tmp_path, monkeypatch): """Test file upload with connection not found.""" + monkeypatch.chdir(tmp_path) + (tmp_path / "payload.txt").write_text("x", encoding="utf-8") mock_pool = ssh_provider.connection_pool + mock_pool.has_connection.return_value = False - with ( - mock.patch("os.path.exists", return_value=True), - mock.patch("os.path.isfile", return_value=True), - ): - mock_pool.has_connection.return_value = False - - result = ssh_provider.ssh_upload( - { - "connection_id": "test-conn", - "local_path": "/local/path", - "remote_path": "/remote/path", - } - ) + result = ssh_provider.ssh_upload( + { + "connection_id": "test-conn", + "local_path": "payload.txt", + "remote_path": "/remote/path", + } + ) - assert "Error: Connection ID 'test-conn' not found" in result - mock_pool.has_connection.assert_called_once_with("test-conn") + assert "Error: Connection ID 'test-conn' not found" in result + mock_pool.has_connection.assert_called_once_with("test-conn") -def test_ssh_upload_not_connected(ssh_provider): +def test_ssh_upload_not_connected(ssh_provider, tmp_path, monkeypatch): """Test file upload with inactive connection.""" + monkeypatch.chdir(tmp_path) + (tmp_path / "payload.txt").write_text("x", encoding="utf-8") 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 - with ( - mock.patch("os.path.exists", return_value=True), - mock.patch("os.path.isfile", return_value=True), - ): - mock_pool.has_connection.return_value = True - mock_pool.get_connection.return_value = mock_connection - mock_connection.is_connected.return_value = False - - result = ssh_provider.ssh_upload( - { - "connection_id": "test-conn", - "local_path": "/local/path", - "remote_path": "/remote/path", - } - ) + result = ssh_provider.ssh_upload( + { + "connection_id": "test-conn", + "local_path": "payload.txt", + "remote_path": "/remote/path", + } + ) - assert "Error: Connection 'test-conn' is not currently active" in result - mock_pool.get_connection.assert_called_once_with("test-conn") - mock_connection.is_connected.assert_called_once() + assert "Error: Connection 'test-conn' is not currently active" in result + mock_pool.get_connection.assert_called_once_with("test-conn") + mock_connection.is_connected.assert_called_once() -def test_ssh_upload_error(ssh_provider): +def test_ssh_upload_error(ssh_provider, tmp_path, monkeypatch): """Test file upload with error.""" + monkeypatch.chdir(tmp_path) + local = tmp_path / "payload.txt" + local.write_text("x", encoding="utf-8") + local_path = str(local.resolve()) + mock_pool = ssh_provider.connection_pool mock_connection = mock.Mock() - - with ( - mock.patch("os.path.exists", return_value=True), - mock.patch("os.path.isfile", return_value=True), - ): - mock_pool.has_connection.return_value = True - mock_pool.get_connection.return_value = mock_connection - mock_connection.is_connected.return_value = True - mock_connection.upload_file.side_effect = SSHConnectionError("Upload failed") - - result = ssh_provider.ssh_upload( - { - "connection_id": "test-conn", - "local_path": "/local/path", - "remote_path": "/remote/path", - } - ) - - assert "Error: SSH connection:" in result - assert "Upload failed" in result - mock_connection.upload_file.assert_called_once_with("/local/path", "/remote/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.upload_file.side_effect = SSHConnectionError("Upload failed") + + result = ssh_provider.ssh_upload( + { + "connection_id": "test-conn", + "local_path": "payload.txt", + "remote_path": "/remote/path", + } + ) + + assert "Error: SSH connection" in result + mock_connection.upload_file.assert_called_once_with(local_path, "/remote/path")