Skip to content
Closed
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
13 changes: 9 additions & 4 deletions src/portkeydrop/protocols.py
Original file line number Diff line number Diff line change
Expand Up @@ -444,12 +444,10 @@ def connect(self) -> None:
f"SFTP connection failed: key file not found: {self._info.key_path}"
)
# asyncssh handles OpenSSH + PPK v2/v3 natively
passphrase = self._info.password if self._info.password else None
connect_kwargs["client_keys"] = [key_path]
connect_kwargs["passphrase"] = passphrase
connect_kwargs["agent_path"] = None # disable agent
auth_methods = [f"key-file:{self._info.key_path}"]
elif self._info.password:
if self._info.password:
connect_kwargs["password"] = self._info.password
auth_methods.append("password")

Expand Down Expand Up @@ -488,7 +486,14 @@ async def _connect():
self._info.host,
auth_methods,
)
if self._info.key_path:
if self._info.key_path and self._info.password:
message = (
f"Authentication failed with key file '{self._info.key_path}' "
"and password fallback. "
"Check key permissions, server authorized_keys, and verify "
"username/password."
)
elif self._info.key_path:
message = (
f"Authentication failed with key file '{self._info.key_path}'. "
"Check key permissions, passphrase, and server authorized_keys."
Expand Down
72 changes: 72 additions & 0 deletions tests/test_sftp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,57 @@ def test_connect_with_key_file(self, mock_connect: AsyncMock, _mock_exists: Magi
call_kwargs = mock_connect.call_args[1]
assert call_kwargs["client_keys"] == ["/path/to/key"]
assert call_kwargs["agent_path"] is None
assert "passphrase" not in call_kwargs
assert "password" not in call_kwargs

@patch("os.path.exists", return_value=True)
@patch("asyncssh.connect", new_callable=AsyncMock)
def test_connect_with_key_and_password_fallback(
self, mock_connect: AsyncMock, _mock_exists: MagicMock
) -> None:
"""When both key_path and password are set, password is passed for
fallback auth and passphrase is omitted."""
info = ConnectionInfo(
protocol=Protocol.SFTP,
host="example.com",
username="user",
password="secret",
key_path="/path/to/key",
)
mock_conn, mock_sftp = _make_mock_conn()
mock_connect.return_value = mock_conn

client = SFTPClient(info)
client.connect()

call_kwargs = mock_connect.call_args[1]
assert call_kwargs["client_keys"] == ["/path/to/key"]
assert call_kwargs["password"] == "secret"
assert "passphrase" not in call_kwargs
assert client._connected is True

@patch("os.path.exists", return_value=True)
@patch("asyncssh.connect", new_callable=AsyncMock)
def test_connect_with_key_and_password_logs_both_methods(
self, mock_connect: AsyncMock, _mock_exists: MagicMock, caplog: pytest.LogCaptureFixture
) -> None:
info = ConnectionInfo(
protocol=Protocol.SFTP,
host="example.com",
username="user",
password="secret",
key_path="/path/to/key",
)
mock_conn, mock_sftp = _make_mock_conn()
mock_connect.return_value = mock_conn

caplog.set_level(logging.DEBUG, logger="portkeydrop.protocols")

client = SFTPClient(info)
client.connect()

assert "key-file:/path/to/key" in caplog.text
assert "password" in caplog.text

@patch("asyncssh.connect", new_callable=AsyncMock)
def test_connect_agent_only(self, mock_connect: AsyncMock) -> None:
Expand Down Expand Up @@ -163,6 +214,27 @@ def test_auth_failure_message_for_agent_and_password(self, mock_connect: AsyncMo
):
client.connect()

@patch("os.path.exists", return_value=True)
@patch("asyncssh.connect", new_callable=AsyncMock)
def test_auth_failure_message_for_key_and_password(
self, mock_connect: AsyncMock, _mock_exists: MagicMock
) -> None:
import asyncssh

mock_connect.side_effect = asyncssh.PermissionDenied("denied")

info = ConnectionInfo(
protocol=Protocol.SFTP,
host="example.com",
username="user",
password="secret",
key_path="/path/to/key",
)
client = SFTPClient(info)

with pytest.raises(ConnectionError, match="and password fallback"):
client.connect()

@patch("asyncssh.connect", new_callable=AsyncMock)
def test_auth_failure_message_for_agent_only(self, mock_connect: AsyncMock) -> None:
import asyncssh
Expand Down
Loading