fix(operations): handle password prompt from sudo-rs (eg. recent Ubuntu) (#1734)

Author: egnor

src/pyinfra/connectors/util.py

@@ -24,6 +24,15 @@
 SUDO_ASKPASS_ENV_VAR = "PYINFRA_SUDO_PASSWORD"
 SU_ASKPASS_ENV_VAR = "PYINFRA_SU_PASSWORD"
 
+# Output lines that indicate sudo could not prompt for a password and we should retry with one.
+# - sudo (Todd C. Miller's): "sudo: a password is required"
+# - sudo-rs (Trifecta Tech): "sudo-rs: interactive authentication is required"
+#   https://github.com/trifectatechfoundation/sudo-rs (default sudo on Ubuntu 25.10+)
+SUDO_PASSWORD_REQUIRED_LINES = (
+    "sudo: a password is required",
+    "sudo-rs: interactive authentication is required",
+)
+
 
 ASKPASS_COMMAND = r"""
 temp=$(mktemp "${{TMPDIR:={0}}}/pyinfra-sudo-askpass-XXXXXXXXXXXX")
@@ -204,7 +213,7 @@ def execute_command_with_sudo_retry(
     # https://github.com/pyinfra-dev/pyinfra/issues/1292
     if return_code != 0 and output and output.combined_lines:
         for line in reversed(output.combined_lines):
-            if line.line.strip() == "sudo: a password is required":
+            if line.line.strip() in SUDO_PASSWORD_REQUIRED_LINES:
                 # If we need a password, ask the user for it and attach to the host
                 # internal connector data for use when executing future commands.
                 sudo_password = getpass(f"{host.print_prefix}sudo password: ")

tests/test_connectors/test_ssh.py

@@ -636,6 +636,45 @@ def test_run_shell_command_retry_for_sudo_password(
             get_pty=False,
         )
 
+    @mock.patch("pyinfra.connectors.ssh.SSHClient")
+    @mock.patch("pyinfra.connectors.util.getpass")
+    def test_run_shell_command_retry_for_sudo_rs_password(
+        self,
+        fake_getpass,
+        fake_ssh_client,
+    ):
+        # sudo-rs (the Rust replacement, default in Ubuntu 25.10+) prints a different message
+        # when it cannot prompt non-interactively; the retry path should recognize it too.
+        fake_getpass.return_value = "PASSWORD"
+
+        fake_ssh = mock.MagicMock()
+        fake_stdin = mock.MagicMock()
+        fake_stdout = mock.MagicMock()
+        fake_stderr = ["sudo-rs: interactive authentication is required"]
+        fake_ssh.exec_command.return_value = fake_stdin, fake_stdout, fake_stderr
+
+        fake_ssh_client.return_value = fake_ssh
+
+        inventory = make_inventory(hosts=("somehost",))
+        state = State(inventory, Config())
+        host = inventory.get_host("somehost")
+        host.connect(state)
+        host.connector_data["sudo_askpass_path"] = "/tmp/pyinfra-sudo-askpass-XXXXXXXXXXXX"
+
+        command = "echo hi"
+        return_values = [1, 0]  # return 0 on the second call
+        fake_stdout.channel.recv_exit_status.side_effect = lambda: return_values.pop(0)
+
+        out = host.run_shell_command(command, _sudo=True)
+        assert len(out) == 2
+        assert out[0] is True
+        assert fake_getpass.called
+        fake_ssh.exec_command.assert_called_with(
+            "env SUDO_ASKPASS=/tmp/pyinfra-sudo-askpass-XXXXXXXXXXXX "
+            "PYINFRA_SUDO_PASSWORD=PASSWORD sudo -H -A -k sh -c 'echo hi'",
+            get_pty=False,
+        )
+
     # SSH file put/get tests
     #