add a worker queue to get updates from the main results thread (#79886)

* Create a queue per WorkerProcess to receive intra-task updates
* Update `pause` action to use the worker queue
* Deprecate ConnectionBase()._new_stdin
* Add new `Display`  convenience method `prompt_until` to manage both controller- and worker-sourced prompting without cross-fork stdin sharing, in-worker mechanism to handle request-response over new worker queue.

changelogs/fragments/prompt-via-worker-queue.yml

@@ -0,0 +1,4 @@
+minor_changes:
+  - Get user input for ``pause`` and ``paramiko_ssh`` from the strategy rather than access ``sys.stdin`` in the WorkerProcess.
+deprecated_features:
+  - The ``ConnectionBase()._new_stdin`` attribute is deprecated, use ``display.prompt_until(msg)`` instead.

lib/ansible/errors/__init__.py

@@ -211,6 +211,14 @@ class AnsibleError(Exception):
         return error_message
 
 
+class AnsiblePromptInterrupt(AnsibleError):
+    '''User interrupt'''
+
+
+class AnsiblePromptNoninteractive(AnsibleError):
+    '''Unable to get user input'''
+
+
 class AnsibleAssertionError(AnsibleError, AssertionError):
     '''Invalid assertion'''
     pass

lib/ansible/executor/process/worker.py

@@ -24,8 +24,9 @@ import sys
 import traceback
 
 from jinja2.exceptions import TemplateNotFound
+from multiprocessing.queues import Queue
 
-from ansible.errors import AnsibleConnectionFailure
+from ansible.errors import AnsibleConnectionFailure, AnsibleError
 from ansible.executor.task_executor import TaskExecutor
 from ansible.module_utils._text import to_text
 from ansible.utils.display import Display
@@ -35,6 +36,17 @@ __all__ = ['WorkerProcess']
 
 display = Display()
 
+current_worker = None
+
+
+class WorkerQueue(Queue):
+    """Queue that raises AnsibleError items on get()."""
+    def get(self, *args, **kwargs):
+        result = super(WorkerQueue, self).get(*args, **kwargs)
+        if isinstance(result, AnsibleError):
+            raise result
+        return result
+
 
 class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defined]
     '''
@@ -43,7 +55,7 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
     for reading later.
     '''
 
-    def __init__(self, final_q, task_vars, host, task, play_context, loader, variable_manager, shared_loader_obj):
+    def __init__(self, final_q, task_vars, host, task, play_context, loader, variable_manager, shared_loader_obj, worker_id):
 
         super(WorkerProcess, self).__init__()
         # takes a task queue manager as the sole param:
@@ -60,6 +72,9 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
         # clear var to ensure we only delete files for this child
         self._loader._tempfiles = set()
 
+        self.worker_queue = WorkerQueue(ctx=multiprocessing_context)
+        self.worker_id = worker_id
+
     def _save_stdin(self):
         self._new_stdin = None
         try:
@@ -155,6 +170,9 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
         # Set the queue on Display so calls to Display.display are proxied over the queue
         display.set_queue(self._final_q)
 
+        global current_worker
+        current_worker = self
+
         try:
             # execute the task and build a TaskResult from the result
             display.debug("running TaskExecutor() for %s/%s" % (self._host, self._task))

lib/ansible/executor/task_queue_manager.py

@@ -24,6 +24,7 @@ import sys
 import tempfile
 import threading
 import time
+import typing as t
 import multiprocessing.queues
 
 from ansible import constants as C
@@ -45,6 +46,7 @@ from ansible.utils.display import Display
 from ansible.utils.lock import lock_decorator
 from ansible.utils.multiprocessing import context as multiprocessing_context
 
+from dataclasses import dataclass
 
 __all__ = ['TaskQueueManager']
 
@@ -64,6 +66,16 @@ class DisplaySend:
         self.kwargs = kwargs
 
 
+@dataclass
+class PromptSend:
+    worker_id: int
+    prompt: str
+    private: bool = True
+    seconds: int = None
+    interrupt_input: t.Iterable[bytes] = None
+    complete_input: t.Iterable[bytes] = None
+
+
 class FinalQueue(multiprocessing.queues.Queue):
     def __init__(self, *args, **kwargs):
         kwargs['ctx'] = multiprocessing_context
@@ -91,6 +103,12 @@ class FinalQueue(multiprocessing.queues.Queue):
             block=False
         )
 
+    def send_prompt(self, **kwargs):
+        self.put(
+            PromptSend(**kwargs),
+            block=False
+        )
+
 
 class AnsibleEndPlay(Exception):
     def __init__(self, result):

lib/ansible/modules/pause.py

@@ -15,6 +15,7 @@ description:
   - To pause/wait/sleep per host, use the M(ansible.builtin.wait_for) module.
   - You can use C(ctrl+c) if you wish to advance a pause earlier than it is set to expire or if you need to abort a playbook run entirely.
     To continue early press C(ctrl+c) and then C(c). To abort a playbook press C(ctrl+c) and then C(a).
+  - Prompting for a set amount of time is not supported. Pausing playbook execution is interruptable but does not return user input.
   - The pause module integrates into async/parallelized playbooks without any special considerations (see Rolling Updates).
     When using pauses with the C(serial) playbook parameter (as in rolling updates) you are only prompted once for the current group of hosts.
   - This module is also supported for Windows targets.
@@ -29,10 +30,11 @@ options:
   prompt:
     description:
       - Optional text to use for the prompt message.
+      - User input is only returned if I(seconds=None) and I(minutes=None), otherwise this is just a custom message before playbook execution is paused.
   echo:
     description:
       - Controls whether or not keyboard input is shown when typing.
-      - Has no effect if 'seconds' or 'minutes' is set.
+      - Only has effect if I(seconds=None) and I(minutes=None).
     type: bool
     default: 'yes'
     version_added: 2.5

lib/ansible/plugins/action/pause.py

@@ -18,91 +18,15 @@ from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
 import datetime
-import signal
-import sys
-import termios
 import time
-import tty
 
-from os import (
-    getpgrp,
-    isatty,
-    tcgetpgrp,
-)
-from ansible.errors import AnsibleError
+from ansible.errors import AnsibleError, AnsiblePromptInterrupt, AnsiblePromptNoninteractive
 from ansible.module_utils._text import to_text
 from ansible.plugins.action import ActionBase
 from ansible.utils.display import Display
 
 display = Display()
 
-try:
-    import curses
-    import io
-
-    # Nest the try except since curses.error is not available if curses did not import
-    try:
-        curses.setupterm()
-        HAS_CURSES = True
-    except (curses.error, TypeError, io.UnsupportedOperation):
-        HAS_CURSES = False
-except ImportError:
-    HAS_CURSES = False
-
-MOVE_TO_BOL = b'\r'
-CLEAR_TO_EOL = b'\x1b[K'
-if HAS_CURSES:
-    # curses.tigetstr() returns None in some circumstances
-    MOVE_TO_BOL = curses.tigetstr('cr') or MOVE_TO_BOL
-    CLEAR_TO_EOL = curses.tigetstr('el') or CLEAR_TO_EOL
-
-
-def setraw(fd, when=termios.TCSAFLUSH):
-    """Put terminal into a raw mode.
-
-    Copied from ``tty`` from CPython 3.11.0, and modified to not remove OPOST from OFLAG
-
-    OPOST is kept to prevent an issue with multi line prompts from being corrupted now that display
-    is proxied via the queue from forks. The problem is a race condition, in that we proxy the display
-    over the fork, but before it can be displayed, this plugin will have continued executing, potentially
-    setting stdout and stdin to raw which remove output post processing that commonly converts NL to CRLF
-    """
-    mode = termios.tcgetattr(fd)
-    mode[tty.IFLAG] = mode[tty.IFLAG] & ~(termios.BRKINT | termios.ICRNL | termios.INPCK | termios.ISTRIP | termios.IXON)
-    # mode[tty.OFLAG] = mode[tty.OFLAG] & ~(termios.OPOST)
-    mode[tty.CFLAG] = mode[tty.CFLAG] & ~(termios.CSIZE | termios.PARENB)
-    mode[tty.CFLAG] = mode[tty.CFLAG] | termios.CS8
-    mode[tty.LFLAG] = mode[tty.LFLAG] & ~(termios.ECHO | termios.ICANON | termios.IEXTEN | termios.ISIG)
-    mode[tty.CC][termios.VMIN] = 1
-    mode[tty.CC][termios.VTIME] = 0
-    termios.tcsetattr(fd, when, mode)
-
-
-class AnsibleTimeoutExceeded(Exception):
-    pass
-
-
-def timeout_handler(signum, frame):
-    raise AnsibleTimeoutExceeded
-
-
-def clear_line(stdout):
-    stdout.write(b'\x1b[%s' % MOVE_TO_BOL)
-    stdout.write(b'\x1b[%s' % CLEAR_TO_EOL)
-
-
-def is_interactive(fd=None):
-    if fd is None:
-        return False
-
-    if isatty(fd):
-        # Compare the current process group to the process group associated
-        # with terminal of the given file descriptor to determine if the process
-        # is running in the background.
-        return getpgrp() == tcgetpgrp(fd)
-    else:
-        return False
-
 
 class ActionModule(ActionBase):
     ''' pauses execution for a length or time, or until input is received '''
@@ -168,143 +92,57 @@ class ActionModule(ActionBase):
         result['start'] = to_text(datetime.datetime.now())
         result['user_input'] = b''
 
-        stdin_fd = None
-        old_settings = None
-        try:
-            if seconds is not None:
-                if seconds < 1:
-                    seconds = 1
+        default_input_complete = None
+        if seconds is not None:
+            if seconds < 1:
+                seconds = 1
 
-                # setup the alarm handler
-                signal.signal(signal.SIGALRM, timeout_handler)
-                signal.alarm(seconds)
+            # show the timer and control prompts
+            display.display("Pausing for %d seconds%s" % (seconds, echo_prompt))
 
-                # show the timer and control prompts
-                display.display("Pausing for %d seconds%s" % (seconds, echo_prompt))
+            # show the prompt specified in the task
+            if new_module_args['prompt']:
                 display.display("(ctrl+C then 'C' = continue early, ctrl+C then 'A' = abort)\r")
-
-                # show the prompt specified in the task
-                if new_module_args['prompt']:
-                    display.display(prompt)
-
             else:
-                display.display(prompt)
+                # corner case where enter does not continue, wait for timeout/interrupt only
+                prompt = "(ctrl+C then 'C' = continue early, ctrl+C then 'A' = abort)\r"
 
-            # save the attributes on the existing (duped) stdin so
-            # that we can restore them later after we set raw mode
-            stdin_fd = None
-            stdout_fd = None
+            # don't complete on LF/CR; we expect a timeout/interrupt and ignore user input when a pause duration is specified
+            default_input_complete = tuple()
+
+        # Only echo input if no timeout is specified
+        echo = seconds is None and echo
+
+        user_input = b''
+        try:
+            _user_input = display.prompt_until(prompt, private=not echo, seconds=seconds, complete_input=default_input_complete)
+        except AnsiblePromptInterrupt:
+            user_input = None
+        except AnsiblePromptNoninteractive:
+            if seconds is None:
+                display.warning("Not waiting for response to prompt as stdin is not interactive")
+            else:
+                # wait specified duration
+                time.sleep(seconds)
+        else:
+            if seconds is None:
+                user_input = _user_input
+        # user interrupt
+        if user_input is None:
+            prompt = "Press 'C' to continue the play or 'A' to abort \r"
             try:
-                stdin = self._connection._new_stdin.buffer
-                stdout = sys.stdout.buffer
-                stdin_fd = stdin.fileno()
-                stdout_fd = stdout.fileno()
-            except (ValueError, AttributeError):
-                # ValueError: someone is using a closed file descriptor as stdin
-                # AttributeError: someone is using a null file descriptor as stdin on windoze
-                stdin = None
-            interactive = is_interactive(stdin_fd)
-            if interactive:
-                # grab actual Ctrl+C sequence
-                try:
-                    intr = termios.tcgetattr(stdin_fd)[6][termios.VINTR]
-                except Exception:
-                    # unsupported/not present, use default
-                    intr = b'\x03'  # value for Ctrl+C
+                user_input = display.prompt_until(prompt, private=not echo, interrupt_input=(b'a',), complete_input=(b'c',))
+            except AnsiblePromptInterrupt:
+                raise AnsibleError('user requested abort!')
 
-                # get backspace sequences
-                try:
-                    backspace = termios.tcgetattr(stdin_fd)[6][termios.VERASE]
-                except Exception:
-                    backspace = [b'\x7f', b'\x08']
+        duration = time.time() - start
+        result['stop'] = to_text(datetime.datetime.now())
+        result['delta'] = int(duration)
 
-                old_settings = termios.tcgetattr(stdin_fd)
-                setraw(stdin_fd)
-
-                # Only set stdout to raw mode if it is a TTY. This is needed when redirecting
-                # stdout to a file since a file cannot be set to raw mode.
-                if isatty(stdout_fd):
-                    setraw(stdout_fd)
-
-                # Only echo input if no timeout is specified
-                if not seconds and echo:
-                    new_settings = termios.tcgetattr(stdin_fd)
-                    new_settings[3] = new_settings[3] | termios.ECHO
-                    termios.tcsetattr(stdin_fd, termios.TCSANOW, new_settings)
-
-                # flush the buffer to make sure no previous key presses
-                # are read in below
-                termios.tcflush(stdin, termios.TCIFLUSH)
-
-            while True:
-                if not interactive:
-                    if seconds is None:
-                        display.warning("Not waiting for response to prompt as stdin is not interactive")
-                    if seconds is not None:
-                        # Give the signal handler enough time to timeout
-                        time.sleep(seconds + 1)
-                    break
-
-                try:
-                    key_pressed = stdin.read(1)
-
-                    if key_pressed == intr:  # value for Ctrl+C
-                        clear_line(stdout)
-                        raise KeyboardInterrupt
-
-                    if not seconds:
-                        # read key presses and act accordingly
-                        if key_pressed in (b'\r', b'\n'):
-                            clear_line(stdout)
-                            break
-                        elif key_pressed in backspace:
-                            # delete a character if backspace is pressed
-                            result['user_input'] = result['user_input'][:-1]
-                            clear_line(stdout)
-                            if echo:
-                                stdout.write(result['user_input'])
-                            stdout.flush()
-                        else:
-                            result['user_input'] += key_pressed
-
-                except KeyboardInterrupt:
-                    signal.alarm(0)
-                    display.display("Press 'C' to continue the play or 'A' to abort \r")
-                    if self._c_or_a(stdin):
-                        clear_line(stdout)
-                        break
-
-                    clear_line(stdout)
-
-                    raise AnsibleError('user requested abort!')
-
-        except AnsibleTimeoutExceeded:
-            # this is the exception we expect when the alarm signal
-            # fires, so we simply ignore it to move into the cleanup
-            pass
-        finally:
-            # cleanup and save some information
-            # restore the old settings for the duped stdin stdin_fd
-            if not (None in (stdin_fd, old_settings)) and isatty(stdin_fd):
-                termios.tcsetattr(stdin_fd, termios.TCSADRAIN, old_settings)
-
-            duration = time.time() - start
-            result['stop'] = to_text(datetime.datetime.now())
-            result['delta'] = int(duration)
-
-            if duration_unit == 'minutes':
-                duration = round(duration / 60.0, 2)
-            else:
-                duration = round(duration, 2)
-            result['stdout'] = "Paused for %s %s" % (duration, duration_unit)
-
-        result['user_input'] = to_text(result['user_input'], errors='surrogate_or_strict')
+        if duration_unit == 'minutes':
+            duration = round(duration / 60.0, 2)
+        else:
+            duration = round(duration, 2)
+        result['stdout'] = "Paused for %s %s" % (duration, duration_unit)
+        result['user_input'] = to_text(user_input, errors='surrogate_or_strict')
         return result
-
-    def _c_or_a(self, stdin):
-        while True:
-            key_pressed = stdin.read(1)
-            if key_pressed.lower() == b'a':
-                return False
-            elif key_pressed.lower() == b'c':
-                return True

lib/ansible/plugins/connection/__init__.py

@@ -59,7 +59,7 @@ class ConnectionBase(AnsiblePlugin):
 
     default_user = None
 
-    def __init__(self, play_context, new_stdin, shell=None, *args, **kwargs):
+    def __init__(self, play_context, new_stdin=None, shell=None, *args, **kwargs):
 
         super(ConnectionBase, self).__init__()
 
@@ -67,8 +67,9 @@ class ConnectionBase(AnsiblePlugin):
         if not hasattr(self, '_play_context'):
             # Backwards compat: self._play_context isn't really needed, using set_options/get_option
             self._play_context = play_context
-        if not hasattr(self, '_new_stdin'):
-            self._new_stdin = new_stdin
+        # Delete once the deprecation period is over for WorkerProcess._new_stdin
+        if not hasattr(self, '__new_stdin'):
+            self.__new_stdin = new_stdin
         if not hasattr(self, '_display'):
             # Backwards compat: self._display isn't really needed, just import the global display and use that.
             self._display = display
@@ -88,6 +89,15 @@ class ConnectionBase(AnsiblePlugin):
 
         self.become = None
 
+    @property
+    def _new_stdin(self):
+        display.deprecated(
+            "The connection's stdin object is deprecated. "
+            "Call display.prompt_until(msg) instead.",
+            version='2.19',
+        )
+        return self.__new_stdin
+
     def set_become_plugin(self, plugin):
         self.become = plugin
 
@@ -269,7 +279,7 @@ class NetworkConnectionBase(ConnectionBase):
     # Do not use _remote_is_local in other connections
     _remote_is_local = True
 
-    def __init__(self, play_context, new_stdin, *args, **kwargs):
+    def __init__(self, play_context, new_stdin=None, *args, **kwargs):
         super(NetworkConnectionBase, self).__init__(play_context, new_stdin, *args, **kwargs)
         self._messages = []
         self._conn_closed = False

lib/ansible/plugins/connection/paramiko_ssh.py

@@ -218,10 +218,8 @@ import socket
 import tempfile
 import traceback
 import fcntl
-import sys
 import re
 
-from termios import tcflush, TCIFLUSH
 from ansible.module_utils.compat.version import LooseVersion
 from binascii import hexlify
 
@@ -260,8 +258,7 @@ class MyAddPolicy(object):
     local L{HostKeys} object, and saving it.  This is used by L{SSHClient}.
     """
 
-    def __init__(self, new_stdin, connection):
-        self._new_stdin = new_stdin
+    def __init__(self, connection):
         self.connection = connection
         self._options = connection._options
 
@@ -277,18 +274,10 @@ class MyAddPolicy(object):
                 # to the question anyway
                 raise AnsibleError(AUTHENTICITY_MSG[1:92] % (hostname, ktype, fingerprint))
 
-            self.connection.connection_lock()
-
-            old_stdin = sys.stdin
-            sys.stdin = self._new_stdin
-
-            # clear out any premature input on sys.stdin
-            tcflush(sys.stdin, TCIFLUSH)
-
-            inp = input(AUTHENTICITY_MSG % (hostname, ktype, fingerprint))
-            sys.stdin = old_stdin
-
-            self.connection.connection_unlock()
+            inp = to_text(
+                display.prompt_until(AUTHENTICITY_MSG % (hostname, ktype, fingerprint), private=False),
+                errors='surrogate_or_strict'
+            )
 
             if inp not in ['yes', 'y', '']:
                 raise AnsibleError("host connection rejected by user")
@@ -418,7 +407,7 @@ class Connection(ConnectionBase):
 
         ssh_connect_kwargs = self._parse_proxy_command(port)
 
-        ssh.set_missing_host_key_policy(MyAddPolicy(self._new_stdin, self))
+        ssh.set_missing_host_key_policy(MyAddPolicy(self))
 
         conn_password = self.get_option('password') or self._play_context.password
 

lib/ansible/plugins/strategy/__init__.py

@@ -40,7 +40,7 @@ from ansible.executor import action_write_locks
 from ansible.executor.play_iterator import IteratingStates
 from ansible.executor.process.worker import WorkerProcess
 from ansible.executor.task_result import TaskResult
-from ansible.executor.task_queue_manager import CallbackSend, DisplaySend
+from ansible.executor.task_queue_manager import CallbackSend, DisplaySend, PromptSend
 from ansible.module_utils.six import string_types
 from ansible.module_utils._text import to_text
 from ansible.module_utils.connection import Connection, ConnectionError
@@ -53,6 +53,7 @@ from ansible.plugins import loader as plugin_loader
 from ansible.template import Templar
 from ansible.utils.display import Display
 from ansible.utils.fqcn import add_internal_fqcns
+from ansible.utils.multiprocessing import context as multiprocessing_context
 from ansible.utils.unsafe_proxy import wrap_var
 from ansible.utils.vars import combine_vars, isidentifier
 from ansible.vars.clean import strip_internal_keys, module_response_deepcopy
@@ -126,6 +127,24 @@ def results_thread_main(strategy):
                 strategy.normalize_task_result(result)
                 with strategy._results_lock:
                     strategy._results.append(result)
+            elif isinstance(result, PromptSend):
+                try:
+                    value = display.prompt_until(
+                        result.prompt,
+                        private=result.private,
+                        seconds=result.seconds,
+                        complete_input=result.complete_input,
+                        interrupt_input=result.interrupt_input,
+                    )
+                except AnsibleError as e:
+                    value = e
+                except BaseException as e:
+                    # relay unexpected errors so bugs in display are reported and don't cause workers to hang
+                    try:
+                        raise AnsibleError(f"{e}") from e
+                    except AnsibleError as e:
+                        value = e
+                strategy._workers[result.worker_id].worker_queue.put(value)
             else:
                 display.warning('Received an invalid object (%s) in the result queue: %r' % (type(result), result))
         except (IOError, EOFError):
@@ -242,6 +261,8 @@ class StrategyBase:
         self._results = deque()
         self._results_lock = threading.Condition(threading.Lock())
 
+        self._worker_queues = dict()
+
         # create the result processing thread for reading results in the background
         self._results_thread = threading.Thread(target=results_thread_main, args=(self,))
         self._results_thread.daemon = True
@@ -385,7 +406,10 @@ class StrategyBase:
                         'play_context': play_context
                     }
 
-                    worker_prc = WorkerProcess(self._final_q, task_vars, host, task, play_context, self._loader, self._variable_manager, plugin_loader)
+                    # Pass WorkerProcess its strategy worker number so it can send an identifier along with intra-task requests
+                    worker_prc = WorkerProcess(
+                        self._final_q, task_vars, host, task, play_context, self._loader, self._variable_manager, plugin_loader, self._cur_worker,
+                    )
                     self._workers[self._cur_worker] = worker_prc
                     self._tqm.send_callback('v2_runner_on_start', host, task)
                     worker_prc.start()

lib/ansible/utils/display.py

@@ -18,31 +18,41 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
+try:
+    import curses
+except ImportError:
+    HAS_CURSES = False
+else:
+    # this will be set to False if curses.setupterm() fails
+    HAS_CURSES = True
+
 import ctypes.util
 import fcntl
 import getpass
+import io
 import logging
 import os
 import random
 import subprocess
 import sys
+import termios
 import textwrap
 import threading
 import time
+import tty
+import typing as t
 
+from functools import wraps
 from struct import unpack, pack
-from termios import TIOCGWINSZ
 
 from ansible import constants as C
-from ansible.errors import AnsibleError, AnsibleAssertionError
+from ansible.errors import AnsibleError, AnsibleAssertionError, AnsiblePromptInterrupt, AnsiblePromptNoninteractive
 from ansible.module_utils._text import to_bytes, to_text
 from ansible.module_utils.six import text_type
 from ansible.utils.color import stringc
 from ansible.utils.multiprocessing import context as multiprocessing_context
 from ansible.utils.singleton import Singleton
 from ansible.utils.unsafe_proxy import wrap_var
-from functools import wraps
-
 
 _LIBC = ctypes.cdll.LoadLibrary(ctypes.util.find_library('c'))
 # Set argtypes, to avoid segfault if the wrong type is provided,
@@ -52,6 +62,9 @@ _LIBC.wcswidth.argtypes = (ctypes.c_wchar_p, ctypes.c_int)
 # Max for c_int
 _MAX_INT = 2 ** (ctypes.sizeof(ctypes.c_int) * 8 - 1) - 1
 
+MOVE_TO_BOL = b'\r'
+CLEAR_TO_EOL = b'\x1b[K'
+
 
 def get_text_width(text):
     """Function that utilizes ``wcswidth`` or ``wcwidth`` to determine the
@@ -183,6 +196,62 @@ def _synchronize_textiowrapper(tio, lock):
     buffer.flush = _wrap_with_lock(buffer.flush, lock)
 
 
+def setraw(fd, when=termios.TCSAFLUSH):
+    """Put terminal into a raw mode.
+
+    Copied from ``tty`` from CPython 3.11.0, and modified to not remove OPOST from OFLAG
+
+    OPOST is kept to prevent an issue with multi line prompts from being corrupted now that display
+    is proxied via the queue from forks. The problem is a race condition, in that we proxy the display
+    over the fork, but before it can be displayed, this plugin will have continued executing, potentially
+    setting stdout and stdin to raw which remove output post processing that commonly converts NL to CRLF
+    """
+    mode = termios.tcgetattr(fd)
+    mode[tty.IFLAG] = mode[tty.IFLAG] & ~(termios.BRKINT | termios.ICRNL | termios.INPCK | termios.ISTRIP | termios.IXON)
+    mode[tty.OFLAG] = mode[tty.OFLAG] & ~(termios.OPOST)
+    mode[tty.CFLAG] = mode[tty.CFLAG] & ~(termios.CSIZE | termios.PARENB)
+    mode[tty.CFLAG] = mode[tty.CFLAG] | termios.CS8
+    mode[tty.LFLAG] = mode[tty.LFLAG] & ~(termios.ECHO | termios.ICANON | termios.IEXTEN | termios.ISIG)
+    mode[tty.CC][termios.VMIN] = 1
+    mode[tty.CC][termios.VTIME] = 0
+    termios.tcsetattr(fd, when, mode)
+
+
+def clear_line(stdout):
+    stdout.write(b'\x1b[%s' % MOVE_TO_BOL)
+    stdout.write(b'\x1b[%s' % CLEAR_TO_EOL)
+
+
+def setup_prompt(stdin_fd, stdout_fd, seconds, echo):
+    # type: (int, int, int, bool) -> None
+    setraw(stdin_fd)
+
+    # Only set stdout to raw mode if it is a TTY. This is needed when redirecting
+    # stdout to a file since a file cannot be set to raw mode.
+    if os.isatty(stdout_fd):
+        setraw(stdout_fd)
+
+    if echo:
+        new_settings = termios.tcgetattr(stdin_fd)
+        new_settings[3] = new_settings[3] | termios.ECHO
+        termios.tcsetattr(stdin_fd, termios.TCSANOW, new_settings)
+
+
+def setupterm():
+    # Nest the try except since curses.error is not available if curses did not import
+    try:
+        curses.setupterm()
+    except (curses.error, TypeError, io.UnsupportedOperation):
+        global HAS_CURSES
+        HAS_CURSES = False
+    else:
+        global MOVE_TO_BOL
+        global CLEAR_TO_EOL
+        # curses.tigetstr() returns None in some circumstances
+        MOVE_TO_BOL = curses.tigetstr('cr') or MOVE_TO_BOL
+        CLEAR_TO_EOL = curses.tigetstr('el') or CLEAR_TO_EOL
+
+
 class Display(metaclass=Singleton):
 
     def __init__(self, verbosity=0):
@@ -228,6 +297,8 @@ class Display(metaclass=Singleton):
         except Exception as ex:
             self.warning(f"failed to patch stdout/stderr for fork-safety: {ex}")
 
+        self.setup_curses = False
+
     def set_queue(self, queue):
         """Set the _final_q on Display, so that we know to proxy display over the queue
         instead of directly writing to stdout/stderr from forks
@@ -520,7 +591,139 @@ class Display(metaclass=Singleton):
 
     def _set_column_width(self):
         if os.isatty(1):
-            tty_size = unpack('HHHH', fcntl.ioctl(1, TIOCGWINSZ, pack('HHHH', 0, 0, 0, 0)))[1]
+            tty_size = unpack('HHHH', fcntl.ioctl(1, termios.TIOCGWINSZ, pack('HHHH', 0, 0, 0, 0)))[1]
         else:
             tty_size = 0
         self.columns = max(79, tty_size - 1)
+
+    def prompt_until(self, msg, private=False, seconds=None, interrupt_input=None, complete_input=None):
+        if self._final_q:
+            from ansible.executor.process.worker import current_worker
+            self._final_q.send_prompt(
+                worker_id=current_worker.worker_id, prompt=msg, private=private, seconds=seconds,
+                interrupt_input=interrupt_input, complete_input=complete_input
+            )
+            return current_worker.worker_queue.get()
+
+        if HAS_CURSES and not self.setup_curses:
+            setupterm()
+            self.setup_curses = True
+
+        if (
+            self._stdin_fd is None
+            or not os.isatty(self._stdin_fd)
+            # Compare the current process group to the process group associated
+            # with terminal of the given file descriptor to determine if the process
+            # is running in the background.
+            or os.getpgrp() != os.tcgetpgrp(self._stdin_fd)
+        ):
+            raise AnsiblePromptNoninteractive('stdin is not interactive')
+
+        # When seconds/interrupt_input/complete_input are all None, this does mostly the same thing as input/getpass,
+        # but self.prompt may raise a KeyboardInterrupt, which must be caught in the main thread.
+        # If the main thread handled this, it would also need to send a newline to the tty of any hanging pids.
+        # if seconds is None and interrupt_input is None and complete_input is None:
+        #     try:
+        #         return self.prompt(msg, private=private)
+        #     except KeyboardInterrupt:
+        #         # can't catch in the results_thread_main daemon thread
+        #         raise AnsiblePromptInterrupt('user interrupt')
+
+        self.display(msg)
+        result = b''
+        with self._lock:
+            original_stdin_settings = termios.tcgetattr(self._stdin_fd)
+            try:
+                setup_prompt(self._stdin_fd, self._stdout_fd, seconds, not private)
+
+                # flush the buffer to make sure no previous key presses
+                # are read in below
+                termios.tcflush(self._stdin, termios.TCIFLUSH)
+
+                # read input 1 char at a time until the optional timeout or complete/interrupt condition is met
+                return self._read_non_blocking_stdin(echo=not private, seconds=seconds, interrupt_input=interrupt_input, complete_input=complete_input)
+            finally:
+                # restore the old settings for the duped stdin stdin_fd
+                termios.tcsetattr(self._stdin_fd, termios.TCSADRAIN, original_stdin_settings)
+
+    def _read_non_blocking_stdin(
+        self,
+        echo=False,  # type: bool
+        seconds=None,  # type: int
+        interrupt_input=None,  # type: t.Iterable[bytes]
+        complete_input=None,  # type: t.Iterable[bytes]
+    ):  # type: (...) -> bytes
+
+        if self._final_q:
+            raise NotImplementedError
+
+        if seconds is not None:
+            start = time.time()
+        if interrupt_input is None:
+            try:
+                interrupt = termios.tcgetattr(sys.stdin.buffer.fileno())[6][termios.VINTR]
+            except Exception:
+                interrupt = b'\x03'  # value for Ctrl+C
+
+        try:
+            backspace_sequences = [termios.tcgetattr(self._stdin_fd)[6][termios.VERASE]]
+        except Exception:
+            # unsupported/not present, use default
+            backspace_sequences = [b'\x7f', b'\x08']
+
+        result_string = b''
+        while seconds is None or (time.time() - start < seconds):
+            key_pressed = None
+            try:
+                os.set_blocking(self._stdin_fd, False)
+                while key_pressed is None and (seconds is None or (time.time() - start < seconds)):
+                    key_pressed = self._stdin.read(1)
+            finally:
+                os.set_blocking(self._stdin_fd, True)
+                if key_pressed is None:
+                    key_pressed = b''
+
+            if (interrupt_input is None and key_pressed == interrupt) or (interrupt_input is not None and key_pressed.lower() in interrupt_input):
+                clear_line(self._stdout)
+                raise AnsiblePromptInterrupt('user interrupt')
+            if (complete_input is None and key_pressed in (b'\r', b'\n')) or (complete_input is not None and key_pressed.lower() in complete_input):
+                clear_line(self._stdout)
+                break
+            elif key_pressed in backspace_sequences:
+                clear_line(self._stdout)
+                result_string = result_string[:-1]
+                if echo:
+                    self._stdout.write(result_string)
+                self._stdout.flush()
+            else:
+                result_string += key_pressed
+        return result_string
+
+    @property
+    def _stdin(self):
+        if self._final_q:
+            raise NotImplementedError
+        try:
+            return sys.stdin.buffer
+        except AttributeError:
+            return None
+
+    @property
+    def _stdin_fd(self):
+        try:
+            return self._stdin.fileno()
+        except (ValueError, AttributeError):
+            return None
+
+    @property
+    def _stdout(self):
+        if self._final_q:
+            raise NotImplementedError
+        return sys.stdout.buffer
+
+    @property
+    def _stdout_fd(self):
+        try:
+            return self._stdout.fileno()
+        except (ValueError, AttributeError):
+            return None

test/integration/targets/pause/pause-6.yml

@@ -0,0 +1,25 @@
+- name: Test pause module input isn't captured with a timeout
+  hosts: localhost
+  become: no
+  gather_facts: no
+
+  tasks:
+    - name: pause with the default message
+      pause:
+        seconds: 3
+      register: default_msg_pause
+
+    - name: pause with a custom message
+      pause:
+        prompt: Wait for three seconds
+        seconds: 3
+      register: custom_msg_pause
+
+    - name: Ensure that input was not captured
+      assert:
+        that:
+          - default_msg_pause.user_input == ''
+          - custom_msg_pause.user_input == ''
+
+    - debug:
+        msg: Task after pause

test/integration/targets/pause/test-pause.py

@@ -168,7 +168,9 @@ pause_test = pexpect.spawn(
 pause_test.logfile = log_buffer
 pause_test.expect(r'Pausing for \d+ seconds')
 pause_test.expect(r"\(ctrl\+C then 'C' = continue early, ctrl\+C then 'A' = abort\)")
+pause_test.send('\n')  # test newline does not stop the prompt - waiting for a timeout or ctrl+C
 pause_test.send('\x03')
+pause_test.expect("Press 'C' to continue the play or 'A' to abort")
 pause_test.send('C')
 pause_test.expect('Task after pause')
 pause_test.expect(pexpect.EOF)
@@ -187,6 +189,7 @@ pause_test.logfile = log_buffer
 pause_test.expect(r'Pausing for \d+ seconds')
 pause_test.expect(r"\(ctrl\+C then 'C' = continue early, ctrl\+C then 'A' = abort\)")
 pause_test.send('\x03')
+pause_test.expect("Press 'C' to continue the play or 'A' to abort")
 pause_test.send('A')
 pause_test.expect('user requested abort!')
 pause_test.expect(pexpect.EOF)
@@ -225,6 +228,7 @@ pause_test.expect(r'Pausing for \d+ seconds')
 pause_test.expect(r"\(ctrl\+C then 'C' = continue early, ctrl\+C then 'A' = abort\)")
 pause_test.expect(r"Waiting for two seconds:")
 pause_test.send('\x03')
+pause_test.expect("Press 'C' to continue the play or 'A' to abort")
 pause_test.send('C')
 pause_test.expect('Task after pause')
 pause_test.expect(pexpect.EOF)
@@ -244,6 +248,7 @@ pause_test.expect(r'Pausing for \d+ seconds')
 pause_test.expect(r"\(ctrl\+C then 'C' = continue early, ctrl\+C then 'A' = abort\)")
 pause_test.expect(r"Waiting for two seconds:")
 pause_test.send('\x03')
+pause_test.expect("Press 'C' to continue the play or 'A' to abort")
 pause_test.send('A')
 pause_test.expect('user requested abort!')
 pause_test.expect(pexpect.EOF)
@@ -275,6 +280,24 @@ pause_test.send('\r')
 pause_test.expect(pexpect.EOF)
 pause_test.close()
 
+# Test input is not returned if a timeout is given
+
+playbook = 'pause-6.yml'
+
+pause_test = pexpect.spawn(
+    'ansible-playbook',
+    args=[playbook] + args,
+    timeout=10,
+    env=os.environ
+)
+
+pause_test.logfile = log_buffer
+pause_test.expect(r'Wait for three seconds:')
+pause_test.send('ignored user input')
+pause_test.expect('Task after pause')
+pause_test.expect(pexpect.EOF)
+pause_test.close()
+
 
 # Test that enter presses may not continue the play when a timeout is set.
 

test/units/utils/display/test_curses.py

@@ -20,7 +20,7 @@ if PY2:
 
 
 def test_pause_curses_tigetstr_none(mocker, monkeypatch):
-    monkeypatch.delitem(sys.modules, 'ansible.plugins.action.pause')
+    monkeypatch.delitem(sys.modules, 'ansible.utils.display')
 
     dunder_import = __import__
 
@@ -35,7 +35,9 @@ def test_pause_curses_tigetstr_none(mocker, monkeypatch):
 
     mocker.patch(builtin_import, _import)
 
-    mod = importlib.import_module('ansible.plugins.action.pause')
+    mod = importlib.import_module('ansible.utils.display')
+    if mod.HAS_CURSES:
+        mod.setupterm()
 
     assert mod.HAS_CURSES is True
     assert mod.MOVE_TO_BOL == b'\r'
@@ -43,7 +45,7 @@ def test_pause_curses_tigetstr_none(mocker, monkeypatch):
 
 
 def test_pause_missing_curses(mocker, monkeypatch):
-    monkeypatch.delitem(sys.modules, 'ansible.plugins.action.pause')
+    monkeypatch.delitem(sys.modules, 'ansible.utils.display')
 
     dunder_import = __import__
 
@@ -55,7 +57,9 @@ def test_pause_missing_curses(mocker, monkeypatch):
 
     mocker.patch(builtin_import, _import)
 
-    mod = importlib.import_module('ansible.plugins.action.pause')
+    mod = importlib.import_module('ansible.utils.display')
+    if mod.HAS_CURSES:
+        mod.setupterm()
 
     with pytest.raises(AttributeError):
         mod.curses  # pylint: disable=pointless-statement
@@ -67,7 +71,7 @@ def test_pause_missing_curses(mocker, monkeypatch):
 
 @pytest.mark.parametrize('exc', (curses.error, TypeError, io.UnsupportedOperation))
 def test_pause_curses_setupterm_error(mocker, monkeypatch, exc):
-    monkeypatch.delitem(sys.modules, 'ansible.plugins.action.pause')
+    monkeypatch.delitem(sys.modules, 'ansible.utils.display')
 
     dunder_import = __import__
 
@@ -82,7 +86,9 @@ def test_pause_curses_setupterm_error(mocker, monkeypatch, exc):
 
     mocker.patch(builtin_import, _import)
 
-    mod = importlib.import_module('ansible.plugins.action.pause')
+    mod = importlib.import_module('ansible.utils.display')
+    if mod.HAS_CURSES:
+        mod.setupterm()
 
     assert mod.HAS_CURSES is False
     assert mod.MOVE_TO_BOL == b'\r'