ssh client: implement password_auth_requested (#403)

Author: skshetry

src/scmrepo/git/backend/dulwich/asyncssh_vendor.py

@@ -41,6 +41,12 @@ async def _read_all(read: Callable[[int], Coroutine], n: Optional[int] = None) -
     return b"".join(result)
 
 
+async def _getpass(*args, **kwargs) -> str:
+    from getpass import getpass
+
+    return await asyncio.to_thread(getpass, *args, **kwargs)
+
+
 class _StderrWrapper:
     def __init__(self, stderr: "SSHReader", loop: asyncio.AbstractEventLoop) -> None:
         self.stderr = stderr
@@ -164,8 +170,6 @@ async def public_key_auth_requested(  # noqa: C901
         return None
 
     async def _read_private_key_interactive(self, path: "FilePath") -> "SSHKey":
-        from getpass import getpass
-
         from asyncssh.public_key import (
             KeyEncryptionError,
             KeyImportError,
@@ -177,11 +181,8 @@ async def _read_private_key_interactive(self, path: "FilePath") -> "SSHKey":
         if passphrase:
             return read_private_key(path, passphrase=passphrase)
 
-        loop = asyncio.get_running_loop()
         for _ in range(3):
-            passphrase = await loop.run_in_executor(
-                None, getpass, f"Enter passphrase for key '{path}': "
-            )
+            passphrase = await _getpass(f"Enter passphrase for key {path!r}: ")
             if passphrase:
                 try:
                     key = read_private_key(path, passphrase=passphrase)
@@ -201,23 +202,20 @@ async def kbdint_challenge_received(  # pylint: disable=invalid-overridden-metho
         lang: str,
         prompts: "KbdIntPrompts",
     ) -> Optional["KbdIntResponse"]:
-        from getpass import getpass
-
         if os.environ.get("GIT_TERMINAL_PROMPT") == "0":
             return None
 
-        def _getpass(prompt: str) -> str:
-            return getpass(prompt=prompt).rstrip()
-
         if instructions:
             pass
-        loop = asyncio.get_running_loop()
-        return [
-            await loop.run_in_executor(
-                None, _getpass, f"({name}) {prompt}" if name else prompt
-            )
-            for prompt, _ in prompts
-        ]
+
+        response: list[str] = []
+        for prompt, _echo in prompts:
+            p = await _getpass(f"({name}) {prompt}" if name else prompt)
+            response.append(p.rstrip())
+        return response
+
+    async def password_auth_requested(self) -> str:
+        return await _getpass()
 
 
 class AsyncSSHVendor(BaseAsyncObject, SSHVendor):

tests/test_dulwich.py

@@ -8,6 +8,7 @@
 import asyncssh
 import paramiko
 import pytest
+from paramiko.server import InteractiveQuery
 from pytest_mock import MockerFixture
 from pytest_test_utils.waiters import wait_until
 
@@ -52,13 +53,24 @@ class Server(paramiko.ServerInterface):
     """http://docs.paramiko.org/en/2.4/api/server.html."""
 
     def __init__(self, commands, *args, **kwargs) -> None:
-        super().__init__(*args, **kwargs)
+        super().__init__()
         self.commands = commands
+        self.allowed_auths = kwargs.get("allowed_auths", "publickey,password")
 
     def check_channel_exec_request(self, channel, command):
         self.commands.append(command)
         return True
 
+    def check_auth_interactive(self, username: str, submethods: str):
+        return InteractiveQuery(
+            "Password", "Enter the password", f"Password for user {USER}:"
+        )
+
+    def check_auth_interactive_response(self, responses):
+        if responses[0] == PASSWORD:
+            return paramiko.AUTH_SUCCESSFUL
+        return paramiko.AUTH_FAILED
+
     def check_auth_password(self, username, password):
         if username == USER and password == PASSWORD:
             return paramiko.AUTH_SUCCESSFUL
@@ -76,12 +88,12 @@ def check_channel_request(self, kind, chanid):
         return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
 
     def get_allowed_auths(self, username):
-        return "password,publickey"
+        return self.allowed_auths
 
 
 @pytest.fixture
 def ssh_conn(request: pytest.FixtureRequest) -> dict[str, Any]:
-    server = Server([])
+    server = Server([], **getattr(request, "param", {}))
 
     socket.setdefaulttimeout(10)
     request.addfinalizer(lambda: socket.setdefaulttimeout(None))
@@ -133,7 +145,8 @@ def test_run_command_password(server: Server, ssh_port: int):
     assert b"test_run_command_password" in server.commands
 
 
-def test_run_command_no_password(server: Server, ssh_port: int):
+@pytest.mark.parametrize("ssh_conn", [{"allowed_auths": "publickey"}], indirect=True)
+def test_run_command_no_password(ssh_port: int):
     vendor = AsyncSSHVendor()
     with pytest.raises(AuthError):
         vendor.run_command(
@@ -145,6 +158,28 @@ def test_run_command_no_password(server: Server, ssh_port: int):
         )
 
 
+@pytest.mark.parametrize(
+    "ssh_conn",
+    [{"allowed_auths": "password"}, {"allowed_auths": "keyboard-interactive"}],
+    indirect=True,
+    ids=["password", "interactive"],
+)
+def test_should_prompt_for_password_when_no_password_passed(
+    mocker: MockerFixture, server: Server, ssh_port: int
+):
+    mocked_getpass = mocker.patch("getpass.getpass", return_value=PASSWORD)
+    vendor = AsyncSSHVendor()
+    vendor.run_command(
+        "127.0.0.1",
+        "test_run_command_password",
+        username=USER,
+        port=ssh_port,
+        password=None,
+    )
+    assert server.commands == [b"test_run_command_password"]
+    mocked_getpass.asssert_called_once()
+
+
 def test_run_command_with_privkey(server: Server, ssh_port: int):
     key = asyncssh.import_private_key(CLIENT_KEY)