Refine SSH auth ordering and password fallback

dvc_ssh/__init__.py

@@ -79,6 +79,17 @@ def _prepare_credentials(self, **config):
         if keyfile := config.get("keyfile"):
             login_info["client_keys"] = [os.path.expanduser(keyfile)]
 
+        # Limit auth attempts to methods backed by configured credentials.
+        # This avoids prompting for unrelated default keys while preserving
+        # keyboard-interactive fallback for password-based logins.
+        preferred_auth = []
+        if login_info.get("client_keys") or login_info.get("passphrase") is not None:
+            preferred_auth.append("publickey")
+        if login_info.get("password") is not None:
+            preferred_auth.extend(("password", "keyboard-interactive"))
+        if preferred_auth:
+            login_info["preferred_auth"] = preferred_auth
+
         login_info["timeout"] = config.get("timeout", 1800)
 
         # These two settings fine tune the asyncssh to use the

dvc_ssh/client.py

@@ -95,7 +95,12 @@ async def _read_private_key_interactive(self, path: "FilePath") -> "SSHKey":
                     pass
         raise KeyImportError("Incorrect passphrase")
 
-    def kbdint_auth_requested(self) -> str:
+    def kbdint_auth_requested(self):
+        # Let asyncssh handle password-backed keyboard-interactive auth so it
+        # can reuse a configured password instead of triggering our prompt path.
+        if self._conn._options.password is not None:
+            return NotImplemented
+
         return ""
 
     async def kbdint_challenge_received(

dvc_ssh/tests/docker-compose.yml

@@ -4,7 +4,9 @@ services:
   openssh-server:
     image: ghcr.io/linuxserver/openssh-server
     environment:
+      - PASSWORD_ACCESS=true
       - USER_NAME=user
+      - USER_PASSWORD=password
       - PUBLIC_KEY_FILE=/tmp/key
     ports:
       - 2222

dvc_ssh/tests/test_client.py

@@ -0,0 +1,44 @@
+from types import SimpleNamespace
+
+import pytest
+
+import dvc_ssh.client
+from dvc_ssh import SSHFileSystem
+from dvc_ssh.client import InteractiveSSHClient
+from dvc_ssh.tests.cloud import TEST_SSH_USER
+
+
+@pytest.mark.parametrize(
+    "password,expected",
+    [("secret", NotImplemented), (None, "")],
+)
+def test_kbdint_auth_requested(password, expected):
+    client = InteractiveSSHClient()
+    client._conn = SimpleNamespace(_options=SimpleNamespace(password=password))
+
+    result = client.kbdint_auth_requested()
+
+    if expected is NotImplemented:
+        assert result is NotImplemented
+    else:
+        assert result == expected
+
+
+def test_password_auth_uses_configured_password(ssh_server, monkeypatch, tmp_path):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
+
+    async def fail_getpass(*args, **kwargs):
+        raise AssertionError("_getpass should not be called")
+
+    monkeypatch.setattr(dvc_ssh.client, "_getpass", fail_getpass)
+
+    fs = SSHFileSystem(
+        host=ssh_server["host"],
+        port=ssh_server["port"],
+        user=TEST_SSH_USER,
+        password="password",
+    )
+
+    assert fs.fs_args["preferred_auth"] == ["password", "keyboard-interactive"]
+    assert fs.exists("/tmp")

dvc_ssh/tests/test_fs.py

@@ -43,14 +43,26 @@ def test_init():
     assert fs.fs_args["port"] == "1234"
     assert fs.fs_args["password"] == "xxx"
     assert fs.fs_args["passphrase"] == "yyy"
+    assert fs.fs_args["preferred_auth"] == [
+        "publickey",
+        "password",
+        "keyboard-interactive",
+    ]
 
 
-@pytest.mark.parametrize("option", ("password", "passphrase"))
-def test_ssh_ask_password(mocker, option):
+@pytest.mark.parametrize(
+    "option,expected_auth",
+    [
+        ("password", ["password", "keyboard-interactive"]),
+        ("passphrase", ["publickey"]),
+    ],
+)
+def test_ssh_ask_password(mocker, option, expected_auth):
     mocker.patch("dvc_ssh.ask_password", return_value="fish")
     args = {f"ask_{option}": True}
     fs = SSHFileSystem(user="test", host="2.2.2.2", **args)
     assert fs.fs_args[option] == "fish"
+    assert fs.fs_args["preferred_auth"] == expected_auth
 
 
 @pytest.mark.parametrize("password", [None, "foo"])
@@ -64,6 +76,16 @@ def test_passphrase(mocker, password, passphrase):
     assert connect.call_args[1]["password"] == password
     assert connect.call_args[1]["passphrase"] == passphrase
 
+    expected_preferred_auth = []
+    if passphrase is not None:
+        expected_preferred_auth.append("publickey")
+    if password is not None:
+        expected_preferred_auth.extend(("password", "keyboard-interactive"))
+
+    assert connect.call_args[1].get("preferred_auth") == (
+        expected_preferred_auth or None
+    )
+
 
 def test_ssh_user():
     fs = SSHFileSystem(host="example.com", user="test")
@@ -92,6 +114,26 @@ def test_ssh_keyfile(config, expected_keyfile):
     assert fs.fs_args.get("client_keys") == expected_keyfiles
 
 
+@pytest.mark.parametrize(
+    "config,expected_preferred_auth",
+    [
+        ({"host": "example.com"}, None),
+        (
+            {"host": "example.com", "password": "secret"},
+            ["password", "keyboard-interactive"],
+        ),
+        ({"host": "example.com", "keyfile": "id_test"}, ["publickey"]),
+        (
+            {"host": "example.com", "keyfile": "id_test", "password": "secret"},
+            ["publickey", "password", "keyboard-interactive"],
+        ),
+    ],
+)
+def test_ssh_preferred_auth(config, expected_preferred_auth):
+    fs = SSHFileSystem(**config)
+    assert fs.fs_args.get("preferred_auth") == expected_preferred_auth
+
+
 @pytest.mark.parametrize(
     "config,expected_gss_auth",
     [