Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions src/borgitory/services/cloud_providers/storage/sftp_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,10 @@ class SFTPStorageConfig(CloudStorageConfig):
private_key: Optional[str] = None
remote_path: str = Field(..., min_length=1)
host_key_checking: bool = Field(default=True)
disable_server_side_checksums: bool = Field(
default=False,
description="Disable MD5/SHA1 hash commands on the remote server. Required for restricted SFTP servers (e.g. Ugreen NAS) that do not allow remote command execution.",
)

@field_validator("host")
@classmethod
Expand Down Expand Up @@ -147,6 +151,7 @@ async def upload_repository(
password=self._config.password,
private_key=self._config.private_key,
path_prefix=remote_path,
disable_hashcheck=self._config.disable_server_side_checksums,
):
if not progress_callback:
continue
Expand Down Expand Up @@ -279,6 +284,7 @@ def get_rclone_mapping(cls) -> RcloneMethodMapping:
"port": "port",
"password": "password",
"private_key": "private_key",
"disable_server_side_checksums": "disable_hashcheck",
},
Comment thread
mlapaglia marked this conversation as resolved.
required_params=["repository", "host", "username"],
optional_params={
Expand Down Expand Up @@ -341,6 +347,7 @@ async def sync_repository_to_sftp(
password: Optional[str] = None,
private_key: Optional[str] = None,
path_prefix: str = "",
disable_hashcheck: bool = False,
) -> AsyncGenerator[ProgressData, None]:
"""Sync a Borg repository to SFTP using Rclone with SFTP backend"""

Expand All @@ -359,6 +366,9 @@ async def sync_repository_to_sftp(
"--verbose",
]

if disable_hashcheck:
command.append("--sftp-disable-hashcheck")

try:
async with self._build_sftp_flags(
host, username, port, password, private_key
Expand Down Expand Up @@ -437,6 +447,7 @@ async def test_sftp_connection(
port: int = 22,
password: Optional[str] = None,
private_key: Optional[str] = None,
disable_hashcheck: bool = False,
) -> ConnectionTestResult:
"""Test SFTP connection by listing remote directory"""
try:
Expand Down Expand Up @@ -602,6 +613,7 @@ async def _test_sftp_write_permissions(
"port": "port",
"password": "password",
"private_key": "private_key",
"disable_server_side_checksums": "disable_hashcheck",
},
required_params=["repository", "host", "username"],
optional_params={
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,25 @@
</option>
</select>
</div>
<div>
<label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
Server-Side Checksums
</label>
<select name="provider_config[disable_server_side_checksums]"
class="select-modern mt-1">
<option value="false"
{% if not config or not config.disable_server_side_checksums %}selected{% endif %}>
Enabled (Recommended)
</option>
<option value="true"
{% if config and config.disable_server_side_checksums %}selected{% endif %}>
Disabled (for restricted SFTP servers, e.g. Ugreen NAS)
</option>
</select>
<p class="mt-1 text-sm text-gray-600 dark:text-gray-400">
Disable if your SFTP server does not allow remote command execution (md5sum/sha1sum).
</p>
</div>
<div>
<label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
Path Prefix (optional)
Expand Down
214 changes: 214 additions & 0 deletions tests/unit/cloud_providers/test_sftp_storage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,214 @@
import pytest
from unittest.mock import AsyncMock, MagicMock
from borgitory.protocols.file_protocols import FileServiceProtocol
from borgitory.services.cloud_providers.storage.sftp_storage import (
SFTPStorageConfig,
SFTPStorage,
)
from borgitory.services.cloud_providers.types import SyncEvent


class TestSFTPStorage:
"""Test SFTP storage runtime behavior"""

@pytest.fixture
def mock_command_executor(self) -> AsyncMock:
return AsyncMock()

@pytest.fixture
def mock_file_service(self) -> AsyncMock:
return AsyncMock(spec=FileServiceProtocol)

@pytest.fixture
def storage_config(self) -> SFTPStorageConfig:
return SFTPStorageConfig(
host="nas.example.com",
username="backup",
remote_path="/backups",
password="secret",
port=22,
)

@pytest.fixture
def storage(
self,
storage_config: SFTPStorageConfig,
mock_command_executor: AsyncMock,
mock_file_service: AsyncMock,
) -> SFTPStorage:
return SFTPStorage(storage_config, mock_command_executor, mock_file_service)

def _make_mock_process(self, return_code: int = 0) -> AsyncMock:
mock_process = AsyncMock()
mock_process.pid = 12345
mock_process.wait.return_value = return_code
mock_process.stdout = AsyncMock()
mock_process.stderr = AsyncMock()
mock_process.stdout.readline = AsyncMock(return_value=b"")
mock_process.stderr.readline = AsyncMock(return_value=b"")
return mock_process

def test_get_sensitive_fields(self, storage: SFTPStorage) -> None:
assert "password" in storage.get_sensitive_fields()
assert "private_key" in storage.get_sensitive_fields()

def test_get_connection_info(self, storage: SFTPStorage) -> None:
info = storage.get_connection_info()
assert info.provider == "sftp"
assert info.details["host"] == "nas.example.com"
assert info.details["port"] == 22
assert info.details["username"] == "backup"
assert info.details["remote_path"] == "/backups"
assert info.details["auth_method"] == "password"

def test_get_connection_info_private_key(
self, mock_command_executor: AsyncMock, mock_file_service: AsyncMock
) -> None:
config = SFTPStorageConfig(
host="nas.example.com",
username="backup",
remote_path="/backups",
private_key="-----BEGIN OPENSSH PRIVATE KEY-----\nfakekey\n-----END OPENSSH PRIVATE KEY-----",
)
storage = SFTPStorage(config, mock_command_executor, mock_file_service)
info = storage.get_connection_info()
assert info.details["auth_method"] == "private_key"

async def test_upload_repository_success(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_command_executor.create_subprocess.return_value = self._make_mock_process(
0
)

progress_events = []

def progress_callback(event: SyncEvent) -> None:
progress_events.append(event)

await storage.upload_repository(
repository_path="/path/to/repo",
remote_path="backups/test",
progress_callback=progress_callback,
)

assert any(event.type.value == "started" for event in progress_events)
assert any(event.type.value == "completed" for event in progress_events)

async def test_upload_repository_failure(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_command_executor.create_subprocess.return_value = self._make_mock_process(
1
)

with pytest.raises(Exception, match="SFTP sync failed"):
await storage.upload_repository(
repository_path="/path/to/repo",
remote_path="backups/test",
progress_callback=lambda e: None,
)

async def test_sync_does_not_include_hashcheck_flag_by_default(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_command_executor.create_subprocess.return_value = self._make_mock_process(
0
)

async for _ in storage.sync_repository_to_sftp(
repository=MagicMock(path="/repo"),
host="nas.example.com",
username="backup",
remote_path="/backups",
password="secret",
disable_hashcheck=False,
):
pass

call_args = mock_command_executor.create_subprocess.call_args
command = call_args.kwargs.get("command") or call_args.args[0]
assert "--sftp-disable-hashcheck" not in command

async def test_sync_includes_hashcheck_flag_when_disabled(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_command_executor.create_subprocess.return_value = self._make_mock_process(
0
)

async for _ in storage.sync_repository_to_sftp(
repository=MagicMock(path="/repo"),
host="nas.example.com",
username="backup",
remote_path="/backups",
password="secret",
disable_hashcheck=True,
):
pass

call_args = mock_command_executor.create_subprocess.call_args
command = call_args.kwargs.get("command") or call_args.args[0]
assert "--sftp-disable-hashcheck" in command

async def test_upload_repository_with_disable_checksums_config(
self, mock_command_executor: AsyncMock, mock_file_service: AsyncMock
) -> None:
config = SFTPStorageConfig(
host="nas.example.com",
username="backup",
remote_path="/backups",
password="secret",
disable_server_side_checksums=True,
)
storage = SFTPStorage(config, mock_command_executor, mock_file_service)
mock_command_executor.create_subprocess.return_value = self._make_mock_process(
0
)

await storage.upload_repository(
repository_path="/path/to/repo",
remote_path="backups/test",
)

call_args = mock_command_executor.create_subprocess.call_args
command = call_args.kwargs.get("command") or call_args.args[0]
assert "--sftp-disable-hashcheck" in command

async def test_upload_repository_without_disable_checksums_config(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_command_executor.create_subprocess.return_value = self._make_mock_process(
0
)

await storage.upload_repository(
repository_path="/path/to/repo",
remote_path="backups/test",
)

call_args = mock_command_executor.create_subprocess.call_args
command = call_args.kwargs.get("command") or call_args.args[0]
assert "--sftp-disable-hashcheck" not in command

async def test_test_connection_success(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_result = AsyncMock()
mock_result.success = True
mock_result.stdout = ""
mock_result.stderr = ""
mock_command_executor.execute_command.return_value = mock_result

result = await storage.test_connection()
assert result is True

async def test_test_connection_failure(
self, storage: SFTPStorage, mock_command_executor: AsyncMock
) -> None:
mock_command_executor.execute_command.side_effect = Exception(
"Connection refused"
)

result = await storage.test_connection()
assert result is False
21 changes: 21 additions & 0 deletions tests/unit/cloud_providers/test_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -495,6 +495,27 @@ def test_empty_remote_path(self) -> None:
remote_path="", # Empty
)

def test_disable_server_side_checksums_defaults_false(self) -> None:
"""Test that server-side checksums are enabled by default"""
config = SFTPStorageConfig(
host="nas.example.com",
username="backup",
password="secret",
remote_path="/backups",
)
assert config.disable_server_side_checksums is False

def test_disable_server_side_checksums_explicit_true(self) -> None:
"""Test that server-side checksums can be disabled (e.g. for Ugreen NAS)"""
config = SFTPStorageConfig(
host="nas.example.com",
username="backup",
password="secret",
remote_path="/backups",
disable_server_side_checksums=True,
)
assert config.disable_server_side_checksums is True


class TestS3Storage:
"""Test S3Storage implementation"""
Expand Down
Loading