From 916e885636c13a06d1a7e0909b86d22f10d342c6 Mon Sep 17 00:00:00 2001 From: SashaMIT Date: Sun, 9 Aug 2026 15:51:13 +0700 Subject: [PATCH 1/2] 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") From f32734cbd0c74492ccd7f0719ae5221f3e8ae528 Mon Sep 17 00:00:00 2001 From: SashaMIT Date: Sun, 9 Aug 2026 15:53:11 +0700 Subject: [PATCH 2/2] fix(python/ssh): confine known_hosts_file to cwd or ~/.ssh Extend resolve_safe_local_path with allowed_roots and apply it to ssh_add_host_key so agent paths cannot write arbitrary host-key files. Also catch ValidationError before ValueError (Pydantic subclass). --- .../action_providers/ssh/path_utils.py | 32 +++++--- .../ssh/ssh_action_provider.py | 16 ++-- .../action_providers/ssh/test_add_host_key.py | 77 +++++++++++-------- .../action_providers/ssh/test_download.py | 4 +- .../action_providers/ssh/test_path_utils.py | 21 ++++- .../tests/action_providers/ssh/test_upload.py | 4 +- 6 files changed, 100 insertions(+), 54 deletions(-) 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 index 21db5af55..2daadb1be 100644 --- a/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/path_utils.py +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/ssh/path_utils.py @@ -5,24 +5,32 @@ import os -def resolve_safe_local_path(path: str) -> str: - """Resolve ``path`` and require it to stay under ``os.getcwd()``. +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. - Twin of the TypeScript twitter/flaunch/zora cwd confine: agent-supplied - local paths must not read or write arbitrary process-readable files. + 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 the working directory") + 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))) - 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") + # 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 real != root and not real.startswith(root + os.sep): - raise ValueError("Local path escapes the working directory") + 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 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 ecc76fecb..b9d4d99ee 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 @@ -412,10 +412,10 @@ 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 ValueError as e: + return f"Error: {e!s}" except Exception as e: return f"Error: File upload: {e!s}" @@ -490,10 +490,10 @@ 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 ValueError as e: + return f"Error: {e!s}" except Exception as e: return f"Error: File download: {e!s}" @@ -541,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" @@ -574,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: diff --git a/python/coinbase-agentkit/tests/action_providers/ssh/test_add_host_key.py b/python/coinbase-agentkit/tests/action_providers/ssh/test_add_host_key.py index 3583f0422..625dc90eb 100644 --- a/python/coinbase-agentkit/tests/action_providers/ssh/test_add_host_key.py +++ b/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), } ) 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 c5be595ba..a6878a4f5 100644 --- a/python/coinbase-agentkit/tests/action_providers/ssh/test_download.py +++ b/python/coinbase-agentkit/tests/action_providers/ssh/test_download.py @@ -35,7 +35,7 @@ def test_ssh_download_success(ssh_provider, tmp_path, monkeypatch): def test_ssh_download_rejects_path_outside_cwd(ssh_provider, tmp_path, monkeypatch): - """Test file download rejects paths outside the working directory.""" + """Test file download rejects paths outside the allowed directory.""" monkeypatch.chdir(tmp_path) mock_pool = ssh_provider.connection_pool mock_connection = mock.Mock() @@ -51,7 +51,7 @@ def test_ssh_download_rejects_path_outside_cwd(ssh_provider, tmp_path, monkeypat } ) - assert "working directory" in result + assert "allowed directory" in result mock_connection.download_file.assert_not_called() 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 index fec32e5f0..9c87a5e9a 100644 --- a/python/coinbase-agentkit/tests/action_providers/ssh/test_path_utils.py +++ b/python/coinbase-agentkit/tests/action_providers/ssh/test_path_utils.py @@ -1,7 +1,6 @@ """Tests for SSH local path confinement.""" import os -from pathlib import Path import pytest @@ -20,8 +19,24 @@ def test_resolve_safe_local_path_allows_under_cwd(tmp_path, monkeypatch): def test_resolve_safe_local_path_rejects_escapes(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) - with pytest.raises(ValueError, match="working directory"): + with pytest.raises(ValueError, match="allowed directory"): resolve_safe_local_path("../outside.txt") - with pytest.raises(ValueError, match="working directory"): + with pytest.raises(ValueError, match="allowed directory"): resolve_safe_local_path("/etc/passwd") + + +def test_resolve_safe_local_path_allows_extra_root(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + ssh_dir = tmp_path / "fake-ssh" + ssh_dir.mkdir() + hosts = ssh_dir / "known_hosts" + hosts.write_text("", encoding="utf-8") + + assert resolve_safe_local_path( + str(hosts), + allowed_roots=[str(ssh_dir)], + ) == str(hosts.resolve()) + + with pytest.raises(ValueError, match="allowed directory"): + resolve_safe_local_path("/etc/passwd", allowed_roots=[str(ssh_dir)]) 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 258ed88d5..6f794b608 100644 --- a/python/coinbase-agentkit/tests/action_providers/ssh/test_upload.py +++ b/python/coinbase-agentkit/tests/action_providers/ssh/test_upload.py @@ -37,7 +37,7 @@ def test_ssh_upload_success(ssh_provider, tmp_path, monkeypatch): def test_ssh_upload_rejects_path_outside_cwd(ssh_provider, tmp_path, monkeypatch): - """Test file upload rejects paths outside the working directory.""" + """Test file upload rejects paths outside the allowed directory.""" monkeypatch.chdir(tmp_path) mock_pool = ssh_provider.connection_pool mock_connection = mock.Mock() @@ -53,7 +53,7 @@ def test_ssh_upload_rejects_path_outside_cwd(ssh_provider, tmp_path, monkeypatch } ) - assert "working directory" in result + assert "allowed directory" in result mock_connection.upload_file.assert_not_called()