diff --git a/.typos.toml b/.typos.toml
index 3a6d0cae5..1377bd812 100644
--- a/.typos.toml
+++ b/.typos.toml
@@ -23,3 +23,7 @@ extend-ignore-re = [
 
 [default.extend-words]
 datas = "datas"
+requiere = "requiere"
+ein = "ein"
+ist = "ist"
+passord = "passord"
diff --git a/src/pyinfra/connectors/util.py b/src/pyinfra/connectors/util.py
index e4dfb65d0..9c79b501f 100644
--- a/src/pyinfra/connectors/util.py
+++ b/src/pyinfra/connectors/util.py
@@ -1,18 +1,18 @@
 from __future__ import annotations
 
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from getpass import getpass
 from queue import Queue
-from gevent.subprocess import PIPE, Popen
 from typing import TYPE_CHECKING
-from collections.abc import Callable, Iterable
 
 import gevent
+from gevent.subprocess import PIPE, Popen
 
 from pyinfra import logger
-from pyinfra.api.output import echo, format_text
 from pyinfra.api import HiddenValue, QuoteString, StringCommand
 from pyinfra.api.exceptions import PyinfraError
+from pyinfra.api.output import echo, format_text
 from pyinfra.api.util import memoize
 
 if TYPE_CHECKING:
@@ -38,12 +38,96 @@
 temp=$(mktemp "${{TMPDIR:={0}}}/pyinfra-sudo-askpass-XXXXXXXXXXXX")
 cat >"$temp"<<'__EOF__'
 #!/bin/sh
+marker="$0.${{PPID}}.called"
+if [ -f "$marker" ]; then
+    rm -f "$marker"
+    echo "sudo: 1 incorrect password attempt" >&2
+    exit 1
+fi
+touch "$marker"
+(
+    while kill -0 "$PPID" 2>/dev/null; do
+        sleep 1
+    done
+    rm -f "$marker"
+) </dev/null >/dev/null 2>&1 &
 printf '%s\n' "${1}"
 __EOF__
 chmod 755 "$temp"
 echo "$temp"
 """
 
+SUDO_PASSWORD_REQUIRED_PHRASES = (
+    "a password is required",
+    "uma senha é necessária",
+    "se requiere una contraseña",
+    "un mot de passe est requis",
+    "ein passwort ist erforderlich",
+    "è richiesta una password",
+    "richiesta una password",
+    "требуется пароль",
+    "wymagane jest hasło",
+    "需要密码",
+    "需要密碼",
+    "パスワードが必要です",
+    "het wachtwoord is vereist",
+    "ett lösenord krävs",
+    "adgangskode er påkrævet",
+    "salasanan antamista vaaditaan",
+    "et passord er påkrevd",
+    "bir parola gerekli",
+    "암호가 필요합니다",
+    "je vyžadováno heslo",
+    "jelszó szükséges",
+    "cal una contrasenya",
+    "interactive authentication is required",  # sudo-rs
+)
+
+SUDO_INCORRECT_PASSWORD_PHRASES = (
+    "incorrect password attempt",
+    "tentativa de senha incorreta",
+    "tentativas de senha incorreta",
+    "intento de contraseña incorrecto",
+    "intentos de contraseña incorrectos",
+    "intento incorrecto de contraseña",
+    "intentos incorrectos de contraseña",
+    "tentative de mot de passe incorrecte",
+    "tentatives de mot de passe incorrectes",
+    "fehlerhafter versuch der passworteingabe",
+    "fehlerhafte versuche der passworteingabe",
+    "fehlgeschlagener passwortversuch",
+    "falscher passwortversuch",
+    "tentativo di password non valida",
+    "tentativo errato",
+    "tentativi errati",
+    "неверная попытка ввода пароля",
+    "неверных попыток ввода пароля",
+    "błędna próba podania hasła",
+    "błędne próby podania hasła",
+    "错误密码尝试",
+    "錯誤密碼嘗試",
+    "パスワードを間違えた回数",
+    "onjuiste wachtwoordpoging",
+    "felaktigt lösenordsförsök",
+    "forkert adgangskodeforsøg",
+    "virheellinen salasanayritys",
+    "feil passordforsøk",
+)
+
+
+def _is_sudo_password_required(line: str) -> bool:
+    line_lower = line.lower()
+    if not line_lower.startswith(("sudo:", "sudo-rs:")):
+        return False
+    return any(phrase in line_lower for phrase in SUDO_PASSWORD_REQUIRED_PHRASES)
+
+
+def _is_sudo_incorrect_password(line: str) -> bool:
+    line_lower = line.lower()
+    if not line_lower.startswith(("sudo:", "sudo-rs:")):
+        return False
+    return any(phrase in line_lower for phrase in SUDO_INCORRECT_PASSWORD_PHRASES)
+
 
 def run_local_process(
     command: str,
@@ -208,12 +292,19 @@ def execute_command_with_sudo_retry(
 ) -> tuple[int, CommandOutput]:
     return_code, output = execute_command()
 
+    attempts = 0
+    max_attempts = 3
     # If we failed look for a sudo password prompt line and re-submit using the sudo password. Look
     # at all lines here in case anything else gets printed, eg in:
     # https://github.com/pyinfra-dev/pyinfra/issues/1292
-    if return_code != 0 and output and output.combined_lines:
+    while return_code != 0 and output and output.combined_lines and attempts < max_attempts:
+        requires_password = False
         for line in reversed(output.combined_lines):
-            if line.line.strip() in SUDO_PASSWORD_REQUIRED_LINES:
+            line_stripped = line.line.strip()
+
+            if _is_sudo_incorrect_password(line_stripped) or _is_sudo_password_required(
+                line_stripped
+            ):
                 # Re-running only helps when the command is executed with sudo, otherwise
                 # the prompted password cannot be applied and the command just runs (and
                 # fails) a second time. Return the failed result so the usual failure
@@ -227,13 +318,21 @@ def execute_command_with_sudo_retry(
                         host.print_prefix,
                     )
                     break
-                # 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: ")
-                host.connector_data["prompted_sudo_password"] = sudo_password
-                return_code, output = execute_command()
+                if command_arguments.get("_sudo_password"):
+                    break
+                requires_password = True
                 break
 
+        if requires_password:
+            # 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: ")
+            host.connector_data["prompted_sudo_password"] = sudo_password
+            attempts += 1
+            return_code, output = execute_command()
+        else:
+            break
+
     return return_code, output
 
 
@@ -276,7 +375,8 @@ def remove_any_sudo_askpass_file(host) -> None:
         if not path:
             continue
         try:
-            host.run_shell_command(StringCommand("rm", "-f", QuoteString(path)))
+            wildcard_path = StringCommand(QuoteString(path), ".*.called", _separator="")
+            host.run_shell_command(StringCommand("rm", "-f", QuoteString(path), wildcard_path))
         except Exception as e:
             logger.debug("Could not remove askpass file %s: %s", path, e)
         host.connector_data[cache_key] = None
diff --git a/tests/test_connectors/test_util.py b/tests/test_connectors/test_util.py
index 901aef483..a046afcfc 100644
--- a/tests/test_connectors/test_util.py
+++ b/tests/test_connectors/test_util.py
@@ -257,9 +257,30 @@ def test_remove_any_sudo_askpass_file_quotes_path(self):
 
         remove_any_sudo_askpass_file(host)
 
-        assert commands == ["rm -f '/tmp/weird path; id'"]
+        assert commands == ["rm -f '/tmp/weird path; id' '/tmp/weird path; id'.*.called"]
         assert host.connector_data["sudo_askpass_path__/tmp"] is None
 
+    def test_explicit_sudo_password_overrides_prompted_sudo_password(self):
+        state = State(make_inventory(), Config())
+        host = state.inventory.get_host("somehost")
+        host.init(state)
+        host.connector_data["prompted_sudo_password"] = "old-password"
+        host.run_shell_command = MagicMock(  # type: ignore[method-assign]
+            return_value=(True, CommandOutput([OutputLine("stdout", "/tmp/askpass-XYZ")])),
+        )
+
+        command = make_unix_command_for_host(
+            state,
+            host,
+            "uptime",
+            _sudo=True,
+            _sudo_password="new-password",
+        )
+        assert command.get_raw_value() == (
+            "env SUDO_ASKPASS=/tmp/askpass-XYZ PYINFRA_SUDO_PASSWORD=new-password "
+            "sudo -H -A -k sh -c uptime"
+        )
+
     def test_command_exists_su_config_only(self):
         """
         This tests covers a bug that appeared when `make_unix_command` is called
@@ -373,6 +394,114 @@ def test_retry_with_sudo(self, fake_getpass):
         fake_getpass.assert_called_once()
         assert host.connector_data["prompted_sudo_password"] == "password"
 
+    @patch("pyinfra.connectors.util.getpass", return_value="secret")
+    def test_retry_on_password_required_portuguese(self, fake_getpass):
+        host = self._make_host()
+        fail_output = CommandOutput([OutputLine("stderr", "sudo: uma senha é necessária")])
+        execute_command = MagicMock(side_effect=[(1, fail_output), (0, CommandOutput([]))])
+
+        return_code, _ = execute_command_with_sudo_retry(host, {"_sudo": True}, execute_command)
+        assert return_code == 0
+        assert execute_command.call_count == 2
+        fake_getpass.assert_called_once()
+        assert host.connector_data["prompted_sudo_password"] == "secret"
+
+    @patch("pyinfra.connectors.util.getpass")
+    def test_retry_on_incorrect_password_attempt(self, fake_getpass):
+        fake_getpass.side_effect = ["wrong", "correct"]
+        host = self._make_host()
+
+        fail_output_1 = CommandOutput([OutputLine("stderr", "sudo: a password is required")])
+        fail_output_2 = CommandOutput([OutputLine("stderr", "sudo: 1 incorrect password attempt")])
+        execute_command = MagicMock(
+            side_effect=[(1, fail_output_1), (1, fail_output_2), (0, CommandOutput([]))]
+        )
+
+        return_code, _ = execute_command_with_sudo_retry(host, {"_sudo": True}, execute_command)
+        assert return_code == 0
+        assert fake_getpass.call_count == 2
+        assert host.connector_data["prompted_sudo_password"] == "correct"
+        assert execute_command.call_count == 3
+
+    @patch("pyinfra.connectors.util.getpass")
+    def test_retry_on_incorrect_password_attempt_portuguese(self, fake_getpass):
+        fake_getpass.side_effect = ["wrong", "correct"]
+        host = self._make_host()
+
+        fail_output_1 = CommandOutput([OutputLine("stderr", "sudo: uma senha é necessária")])
+        fail_output_2 = CommandOutput(
+            [OutputLine("stderr", "sudo: 1 tentativa de senha incorreta")]
+        )
+        execute_command = MagicMock(
+            side_effect=[(1, fail_output_1), (1, fail_output_2), (0, CommandOutput([]))]
+        )
+
+        return_code, _ = execute_command_with_sudo_retry(host, {"_sudo": True}, execute_command)
+        assert return_code == 0
+        assert fake_getpass.call_count == 2
+        assert host.connector_data["prompted_sudo_password"] == "correct"
+        assert execute_command.call_count == 3
+
+    @patch("pyinfra.connectors.util.getpass", return_value="wrong")
+    def test_max_attempts_exhausted(self, fake_getpass):
+        host = self._make_host()
+        fail_output = CommandOutput([OutputLine("stderr", "sudo: 1 incorrect password attempt")])
+        execute_command = MagicMock(return_value=(1, fail_output))
+
+        return_code, output = execute_command_with_sudo_retry(
+            host, {"_sudo": True}, execute_command
+        )
+        assert return_code == 1
+        assert output == fail_output
+        assert fake_getpass.call_count == 3
+        assert execute_command.call_count == 4
+
+    @patch("pyinfra.connectors.util.getpass")
+    def test_no_retry_with_explicit_sudo_password(self, fake_getpass):
+        host = self._make_host()
+        fail_output = CommandOutput([OutputLine("stderr", "sudo: 1 incorrect password attempt")])
+        execute_command = MagicMock(return_value=(1, fail_output))
+
+        return_code, output = execute_command_with_sudo_retry(
+            host, {"_sudo": True, "_sudo_password": "wrongpassword"}, execute_command
+        )
+        assert return_code == 1
+        assert output == fail_output
+        execute_command.assert_called_once_with()
+        fake_getpass.assert_not_called()
+
+    @patch("pyinfra.connectors.util.getpass", return_value="correct")
+    def test_retry_when_use_sudo_password_prompted_wrong(self, fake_getpass):
+        state = State(make_inventory(hosts=("somehost",)), Config(USE_SUDO_PASSWORD=True))
+        host = state.inventory.get_host("somehost")
+        host.init(state)
+        host.run_shell_command = MagicMock(  # type: ignore[method-assign]
+            return_value=(True, CommandOutput([OutputLine("stdout", "/tmp/askpass-XYZ")])),
+        )
+        host.connector_data["prompted_sudo_password"] = "wrong"
+
+        commands_generated = []
+        fail_output = CommandOutput([OutputLine("stderr", "sudo: 1 incorrect password attempt")])
+        outcomes = [(1, fail_output), (0, CommandOutput([]))]
+
+        arguments = {"_sudo": True}
+
+        def execute_command():
+            cmd = make_unix_command_for_host(state, host, "uptime", **arguments)
+            commands_generated.append(cmd.get_raw_value())
+            return outcomes.pop(0)
+
+        return_code, _ = execute_command_with_sudo_retry(host, arguments, execute_command)
+        assert return_code == 0
+        assert fake_getpass.call_count == 1
+        assert host.connector_data["prompted_sudo_password"] == "correct"
+        assert commands_generated == [
+            "env SUDO_ASKPASS=/tmp/askpass-XYZ PYINFRA_SUDO_PASSWORD=wrong "
+            "sudo -H -A -k sh -c uptime",
+            "env SUDO_ASKPASS=/tmp/askpass-XYZ PYINFRA_SUDO_PASSWORD=correct "
+            "sudo -H -A -k sh -c uptime",
+        ]
+
 
 class TestRemoveAnySudoAskpassFile(TestCase):
     def test_clears_state_and_runs_rm(self):
@@ -387,8 +516,8 @@ def test_clears_state_and_runs_rm(self):
         remove_any_sudo_askpass_file(host)
 
         assert sorted(commands) == [
-            "rm -f /tmp/su-askpass",
-            "rm -f /tmp/sudo-askpass",
+            "rm -f /tmp/su-askpass /tmp/su-askpass.*.called",
+            "rm -f /tmp/sudo-askpass /tmp/sudo-askpass.*.called",
         ]
         assert host.connector_data["sudo_askpass_path__/tmp"] is None
         assert host.connector_data["su_askpass_path__/tmp"] is None
@@ -406,9 +535,9 @@ def test_clears_multiple_temp_dir_variants(self):
         remove_any_sudo_askpass_file(host)
 
         assert sorted(commands) == [
-            "rm -f /dev/shm/su-askpass",
-            "rm -f /tmp/sudo-askpass",
-            "rm -f /var/tmp/sudo-askpass",
+            "rm -f /dev/shm/su-askpass /dev/shm/su-askpass.*.called",
+            "rm -f /tmp/sudo-askpass /tmp/sudo-askpass.*.called",
+            "rm -f /var/tmp/sudo-askpass /var/tmp/sudo-askpass.*.called",
         ]
         assert host.connector_data["sudo_askpass_path__/tmp"] is None
         assert host.connector_data["sudo_askpass_path__/var/tmp"] is None