Limit askpass prompts to single attempt (#85364)

* Limit askpass prompts to single attempt

OpenSSH client option NumberOfPasswordPrompts defaults to 3 so in case
an incorrect connection password is provided it is excessively tried 3
times. Not only that but running the `_ssh_askpass` entry point multiple
times (via ssh) results in `json.decoder.JSONDecodeError` as after the
first run the shared memory is zero'd and the subsequent runs end up
calling `json.loads` on empty data.

`json.decoder.JSONDecodeError` does not happen prior to Python 3.13 as
the share memory is unlinked automatically on `.close()` and the
`_ssh_askpass` entry point exits with return code 1 before attempting to
load zero'd memory.

Fixes #85359

* changelog and tests

* Update changelogs/fragments/85359-askpass-incorrect-password-retries.yml

Co-authored-by: Matt Davis <6775756+nitzmahone@users.noreply.github.com>

* Update lib/ansible/cli/_ssh_askpass.py

Co-authored-by: Martin Krizek <martin.krizek@gmail.com>

* Avoid race condition in second unlink

---------

Co-authored-by: Matt Davis <6775756+nitzmahone@users.noreply.github.com>
Co-authored-by: Matt Clay <matt@mystile.com>

changelogs/fragments/85359-askpass-incorrect-password-retries.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - ssh connection plugin - Allow only one password prompt attempt when utilizing ``SSH_ASKPASS`` (https://github.com/ansible/ansible/issues/85359)

lib/ansible/cli/_ssh_askpass.py

@@ -3,45 +3,52 @@
 from __future__ import annotations
 
 import json
+import multiprocessing.resource_tracker
 import os
 import re
 import sys
 import typing as t
-from multiprocessing.shared_memory import SharedMemory
 
-HOST_KEY_RE = re.compile(
+from multiprocessing.shared_memory import SharedMemory
-    r'(The authenticity of host |differs from the key for the IP address)',
-)
 
 
 def main() -> t.Never:
-    try:
+    if len(sys.argv) > 1:
-        if HOST_KEY_RE.search(sys.argv[1]):
+        exit_code = 0 if handle_prompt(sys.argv[1]) else 1
-            sys.stdout.buffer.write(b'no')
+    else:
+        exit_code = 1
+
+    sys.exit(exit_code)
+
+
+def handle_prompt(prompt: str) -> bool:
+    if re.search(r'(The authenticity of host |differs from the key for the IP address)', prompt):
+        sys.stdout.write('no')
         sys.stdout.flush()
-            sys.exit(0)
+        return True
-    except IndexError:
+
-        pass
+    # deprecated: description='Python 3.13 and later support track' python_version='3.12'
-
+    can_track = sys.version_info[:2] >= (3, 13)
-    kwargs: dict[str, bool] = {}
+    kwargs = dict(track=False) if can_track else {}
-    if sys.version_info[:2] >= (3, 13):
+
-        # deprecated: description='unneeded due to track argument for SharedMemory' python_version='3.12'
+    # This SharedMemory instance is intentionally not closed or unlinked.
-        kwargs['track'] = False
+    # Closing will occur naturally in the SharedMemory finalizer.
-    try:
+    # Unlinking is the responsibility of the process which created it.
     shm = SharedMemory(name=os.environ['_ANSIBLE_SSH_ASKPASS_SHM'], **kwargs)
-    except FileNotFoundError:
+
-        # We must be running after the ansible fork is shutting down
+    if not can_track:
-        sys.exit(1)
+        # When track=False is not available, we must unregister explicitly, since it otherwise only occurs during unlink.
+        # This avoids resource tracker noise on stderr during process exit.
+        multiprocessing.resource_tracker.unregister(shm._name, 'shared_memory')
+
     cfg = json.loads(shm.buf.tobytes().rstrip(b'\x00'))
 
-    try:
+    if cfg['prompt'] not in prompt:
-        if cfg['prompt'] not in sys.argv[1]:
+        return False
-            sys.exit(1)
-    except IndexError:
-        sys.exit(1)
 
-    sys.stdout.buffer.write(cfg['password'].encode('utf-8'))
+    # Report the password provided by the SharedMemory instance.
+    # The contents are left untouched after consumption to allow subsequent attempts to succeed.
+    # This can occur when multiple password prompting methods are enabled, such as password and keyboard-interactive, which is the default on macOS.
+    sys.stdout.write(cfg['password'])
     sys.stdout.flush()
-    shm.buf[:] = b'\x00' * shm.size
+    return True
-    shm.close()
-    sys.exit(0)

lib/ansible/plugins/connection/ssh.py

@@ -961,6 +961,13 @@ class Connection(ConnectionBase):
                 b_args = (b"-o", b'ControlPath="%s"' % to_bytes(self.control_path % dict(directory=cpdir), errors='surrogate_or_strict'))
                 self._add_args(b_command, b_args, u"found only ControlPersist; added ControlPath")
 
+        if password_mechanism == "ssh_askpass":
+            self._add_args(
+                b_command,
+                (b"-o", b"NumberOfPasswordPrompts=1"),
+                "Restrict number of password prompts in case incorrect password is provided.",
+            )
+
         # Finally, we add any caller-supplied extras.
         if other_args:
             b_command += [to_bytes(a) for a in other_args]

test/integration/targets/connection_ssh/test_ssh_askpass.yml

@@ -23,7 +23,42 @@
         state: restarted
       when: ansible_facts.system != 'Darwin'
 
-    - command:
+    - name: Test incorrect password
+      command:
+        argv:
+          - ansible
+          - localhost
+          - -m
+          - command
+          - -a
+          - id
+          - -vvv
+          - -e
+          - ansible_pipelining=yes
+          - -e
+          - ansible_connection=ssh
+          - -e
+          - ansible_ssh_password_mechanism=ssh_askpass
+          - -e
+          - ansible_user={{ test_user_name }}
+          - -e
+          - ansible_password=INCORRECT_PASSWORD
+      environment:
+        ANSIBLE_NOCOLOR: "1"
+        ANSIBLE_FORCE_COLOR: "0"
+      register: askpass_out
+      ignore_errors: true
+
+    - assert:
+        that:
+          - askpass_out is failed
+          - askpass_out.stdout is contains('UNREACHABLE')
+          - askpass_out.stdout is contains('Permission denied')
+          - askpass_out.stdout is not contains('Permission denied, please try again.')  # password tried only once
+          - askpass_out.stdout is not contains('Traceback (most recent call last)')
+
+    - name: Test correct password
+      command:
         argv:
           - ansible
           - localhost