lock.diff

diff --git a/lib/ansible/plugins/connections/ssh.py b/lib/ansible/plugins/connections/ssh.py
index f7b17c5..5a6a732 100644
--- a/lib/ansible/plugins/connections/ssh.py
+++ b/lib/ansible/plugins/connections/ssh.py
@@ -360,6 +360,14 @@ class Connection(ConnectionBase):
         display_cmd = map(pipes.quote, cmd[:-1]) + [cmd[-1]]
         self._display.vvv('SSH: EXEC {0}'.format(' '.join(display_cmd)), host=self.host)
 
+        # We begin by acquiring the connection lock, so that if ssh issues a
+        # host key verification prompt, output from other connections doesn't
+        # obscure it. The lock can be released once we know that the host key
+        # has been accepted (either because of known_hosts or the prompt).
+
+        self.connection_lock()
+        self._lock = True
+
         # Start the given command. If we don't need to pipeline data, we can try
         # to use a pseudo-tty (ssh will have been invoked with -tt). If we are
         # pipelining data, or can't create a pty, we fall back to using plain
@@ -425,6 +433,7 @@ class Connection(ConnectionBase):
         tmp_stdout = tmp_stderr = ''
 
         self._flags = dict(
+            can_unlock=False,
             become_prompt=False, become_success=False,
             become_error=False, become_nopasswd_error=False
         )
@@ -460,11 +469,12 @@ class Connection(ConnectionBase):
                 tmp_stderr += chunk
                 #self._display.debug("stderr chunk (state=%s):\n>>>%s<<<\n" % (state, chunk))
 
-            # We examine the output line-by-line until we have negotiated any
-            # privilege escalation prompt and subsequent success/error message.
-            # Afterwards, we can accumulate output without looking at it.
+            # We examine the output line-by-line until we have released the
+            # connection lock and negotiated any privilege escalation prompt and
+            # subsequent success or error message. Afterwards, we can accumulate
+            # output without looking at it.
 
-            if state <= states.index('ready_to_send'):
+            if self._lock or state <= states.index('ready_to_send'):
                 if tmp_stdout:
                     output, unprocessed = self._examine_output('stdout', states[state], tmp_stdout, sudoable)
                     stdout += output
@@ -479,6 +489,16 @@ class Connection(ConnectionBase):
                 stderr += tmp_stderr
                 tmp_stdout = tmp_stderr = ''
 
+            # We're always looking for the earliest opportunity to release the
+            # connection lock based on ssh host key messages (but if we see an
+            # escalation prompt or confirmation, that means the connection has
+            # been successfully established too).
+
+            if self._lock:
+                if self._flags['can_unlock'] or self._flags['become_prompt'] or self._flags['become_success']:
+                    self.connection_unlock()
+                    self._lock = False
+
             # If we see a privilege escalation prompt, we send the password.
 
             if states[state] == 'awaiting_prompt' and self._flags['become_prompt']:
@@ -547,6 +567,15 @@ class Connection(ConnectionBase):
         # completely (see also issue #848)
         stdin.close()
 
+        # One last check to make sure we release the connection lock. We should
+        # have been able to release it much earlier based on ssh messages or an
+        # escalation prompt. If we reach this, it probably means that ssh debug
+        # messages changed and we didn't request escalation.
+
+        if self._lock:
+            self._display.warning('Released connection lock only at process exit')
+            self.connection_unlock()
+
         controlpersisterror = 'Bad configuration option: ControlPersist' in stderr or 'unknown configuration option: ControlPersist' in stderr
 
         if C.HOST_KEY_CHECKING:
@@ -580,6 +609,18 @@ class Connection(ConnectionBase):
         self._flags['become_error'] = False
         self._flags['become_nopasswd_error'] = False
 
+        # The most reliable indication that ssh will not prompt for host key
+        # confirmation is a "Host ... is known and matches the ... host key"
+        # debug1 message, followed by the "Permanently added ..." warning on
+        # first-time connections.
+        #
+        # For multiplexed (ControlPersist) connections, the best we can do
+        # is to look for 'mux_client_request_session: entering'
+
+        key_known = "debug1: Host '\S+' is known and matches the \S+ host key"
+        key_added = "Warning: Permanently added '\S+' (\S+) to the list of known hosts"
+        mux_session = 'mux_client_request_session: entering'
+
         output = []
         for l in chunk.splitlines(True):
             suppress_output = False
@@ -599,6 +640,8 @@ class Connection(ConnectionBase):
             elif sudoable and self.check_missing_password(l):
                 self._display.debug("become_nopasswd_error: (source=%s, state=%s): '%s'" % (source, state, l.rstrip('\r\n')))
                 self._flags['become_nopasswd_error'] = True
+            else:
+                self._examine_ssh_line(l)
 
             if not suppress_output:
                 output.append(l)
@@ -615,6 +658,32 @@ class Connection(ConnectionBase):
 
         return ''.join(output), remainder
 
+    def _examine_ssh_line(self, line):
+        self._flags['can_unlock'] = False
+
+        # Do nothing if we've acquired and released the lock already.
+        if self._lock is False:
+            return
+
+        # The most reliable indication that ssh will not prompt for host key
+        # confirmation is a "Host ... is known and matches the ... host key"
+        # debug1 message, followed by the "Permanently added ..." warning on
+        # first-time connections. Connection failures and key verification
+        # failures also qualify.
+        #
+        # For multiplexed (ControlPersist) connections, the best we can do
+        # is to look for 'mux_client_request_session: entering'
+
+        key_known = "debug1: Host '\S+' is known and matches the \S+ host \S+"
+        key_added = "Warning: Permanently added '\S+' (\S+) to the list of known hosts"
+
+        if re.match(key_known, line) or \
+            re.match(key_added, line) or \
+            'Host key verification failed' in line or \
+            'mux_client_request_session: entering' in line or \
+            re.match('ssh: connect to host \S+ port \S+: ', line):
+            self._flags['can_unlock'] = True
+
     # Utility functions
 
     def _terminate_process(self, p):