winrm: add better exception handling for krb5 auth with pexpect (ansible#39930)

* winrm: add better exception handling for krb5 auth with pexpect

* Added changelog fragment

* Added exception handler in case kinit path isn't valid, added test cases

* fixed for Python 2 compatibility

Author: jborean93

changelogs/fragments/winrm_kinit_error-fix.yaml

@@ -0,0 +1,2 @@
+bugfixes:
+- winrm - Add better error handling when the kinit process fails

lib/ansible/plugins/connection/winrm.py

@@ -285,33 +285,60 @@ def _kerb_auth(self, principal, password):
         # doing so. Unfortunately it is not available on the built in Python
         # so we can only use it if someone has installed it
         if HAS_PEXPECT:
-            kinit_cmdline = " ".join(kinit_cmdline)
+            proc_mechanism = "pexpect"
+            command = kinit_cmdline.pop(0)
             password = to_text(password, encoding='utf-8',
                                errors='surrogate_or_strict')
 
             display.vvvv("calling kinit with pexpect for principal %s"
                          % principal)
-            events = {
-                ".*:": password + "\n"
-            }
-            # technically this is the stdout but to match subprocess we will
-            # call it stderr
-            stderr, rc = pexpect.run(kinit_cmdline, withexitstatus=True, events=events, env=krb5env, timeout=60)
+            try:
+                child = pexpect.spawn(command, kinit_cmdline, timeout=60,
+                                      env=krb5env)
+            except pexpect.ExceptionPexpect as err:
+                err_msg = "Kerberos auth failure when calling kinit cmd " \
+                          "'%s': %s" % (command, to_native(err))
+                raise AnsibleConnectionFailure(err_msg)
+
+            try:
+                child.expect(".*:")
+                child.sendline(password)
+            except OSError as err:
+                # child exited before the pass was sent, Ansible will raise
+                # error based on the rc below, just display the error here
+                display.vvvv("kinit with pexpect raised OSError: %s"
+                             % to_native(err))
+
+            # technically this is the stdout + stderr but to match the
+            # subprocess error checking behaviour, we will call it stderr
+            stderr = child.read()
+            child.wait()
+            rc = child.exitstatus
         else:
+            proc_mechanism = "subprocess"
             password = to_bytes(password, encoding='utf-8',
                                 errors='surrogate_or_strict')
 
             display.vvvv("calling kinit with subprocess for principal %s"
                          % principal)
-            p = subprocess.Popen(kinit_cmdline, stdin=subprocess.PIPE,
-                                 stdout=subprocess.PIPE,
-                                 stderr=subprocess.PIPE,
-                                 env=krb5env)
+            try:
+                p = subprocess.Popen(kinit_cmdline, stdin=subprocess.PIPE,
+                                     stdout=subprocess.PIPE,
+                                     stderr=subprocess.PIPE,
+                                     env=krb5env)
+
+            except OSError as err:
+                err_msg = "Kerberos auth failure when calling kinit cmd " \
+                          "'%s': %s" % (self._kinit_cmd, to_native(err))
+                raise AnsibleConnectionFailure(err_msg)
+
             stdout, stderr = p.communicate(password + b'\n')
             rc = p.returncode != 0
 
         if rc != 0:
-            raise AnsibleConnectionFailure("Kerberos auth failure: %s" % to_native(stderr.strip()))
+            err_msg = "Kerberos auth failure for principal %s with %s: %s" \
+                      % (principal, proc_mechanism, to_native(stderr.strip()))
+            raise AnsibleConnectionFailure(err_msg)
 
         display.vvvvv("kinit succeeded for principal %s" % principal)
 

test/units/plugins/connection/test_winrm.py

@@ -13,6 +13,8 @@
 from io import StringIO
 
 from ansible.compat.tests.mock import patch, MagicMock
+from ansible.errors import AnsibleConnectionFailure
+from ansible.module_utils._text import to_bytes
 from ansible.playbook.play_context import PlayContext
 from ansible.plugins.loader import connection_loader
 
@@ -219,3 +221,148 @@ def test_set_options(self, play, options, direct, expected, kerb):
                 % (attr, actual, expected)
 
         module_patcher.stop()
+
+
+class TestWinRMKerbAuth(object):
+
+    DATA = (
+        # default
+        ({"_extras": {}}, (["kinit", "user@domain"],), False),
+        ({"_extras": {}}, ("kinit", ["user@domain"],), True),
+
+        # override kinit path from options
+        ({"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'},
+         (["kinit2", "user@domain"],), False),
+        ({"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'},
+         ("kinit2", ["user@domain"],), True),
+
+        # we expect the -f flag when delegation is set
+        ({"_extras": {'ansible_winrm_kerberos_delegation': True}},
+         (["kinit", "-f", "user@domain"],), False),
+        ({"_extras": {'ansible_winrm_kerberos_delegation': True}},
+         ("kinit", ["-f", "user@domain"],), True),
+    )
+
+    # pylint bug: https://github.com/PyCQA/pylint/issues/511
+    # pylint: disable=undefined-variable
+    @pytest.mark.parametrize('options, expected, pexpect',
+                             ((o, e, p) for o, e, p in DATA))
+    def test_kinit_success(self, options, expected, pexpect):
+        def mock_popen_communicate(input=None, timeout=None):
+            return b"", b""
+
+        mock_pexpect = None
+        if pexpect:
+            mock_pexpect = MagicMock()
+            mock_pexpect.spawn.return_value.exitstatus = 0
+
+        mock_subprocess = MagicMock()
+        mock_subprocess.Popen.return_value.communicate = mock_popen_communicate
+        mock_subprocess.Popen.return_value.returncode = 0
+
+        modules = {
+            'pexpect': mock_pexpect,
+            'subprocess': mock_subprocess,
+        }
+
+        with patch.dict(sys.modules, modules):
+            pc = PlayContext()
+            new_stdin = StringIO()
+
+            connection_loader._module_cache = {}
+            conn = connection_loader.get('winrm', pc, new_stdin)
+            conn.set_options(var_options=options)
+
+            conn._kerb_auth("user@domain", "pass")
+            if pexpect:
+                assert len(mock_pexpect.method_calls) == 1
+                assert mock_pexpect.method_calls[0][1] == expected
+                actual_env = mock_pexpect.method_calls[0][2]['env']
+            else:
+                assert len(mock_subprocess.method_calls) == 1
+                assert mock_subprocess.method_calls[0][1] == expected
+                actual_env = mock_subprocess.method_calls[0][2]['env']
+
+            assert list(actual_env.keys()) == ['KRB5CCNAME']
+            assert actual_env['KRB5CCNAME'].startswith("FILE:/")
+
+    # pylint bug: https://github.com/PyCQA/pylint/issues/511
+    # pylint: disable=undefined-variable
+    @pytest.mark.parametrize('use_pexpect', (False, True),)
+    def test_kinit_with_missing_executable(self, use_pexpect):
+        expected_err = "[Errno 2] No such file or directory: " \
+                       "'/fake/kinit': '/fake/kinit'"
+        mock_subprocess = MagicMock()
+        mock_subprocess.Popen = MagicMock(side_effect=OSError(expected_err))
+
+        mock_pexpect = None
+        if use_pexpect:
+            expected_err = "The command was not found or was not " \
+                           "executable: /fake/kinit"
+
+            mock_pexpect = MagicMock()
+            mock_pexpect.ExceptionPexpect = Exception
+            mock_pexpect.spawn = MagicMock(side_effect=Exception(expected_err))
+
+        modules = {
+            'pexpect': mock_pexpect,
+            'subprocess': mock_subprocess,
+        }
+
+        with patch.dict(sys.modules, modules):
+            pc = PlayContext()
+            new_stdin = StringIO()
+
+            connection_loader._module_cache = {}
+            conn = connection_loader.get('winrm', pc, new_stdin)
+            options = {"_extras": {}, "ansible_winrm_kinit_cmd": "/fake/kinit"}
+            conn.set_options(var_options=options)
+
+            with pytest.raises(AnsibleConnectionFailure) as err:
+                conn._kerb_auth("user@domain", "pass")
+            assert str(err.value) == "Kerberos auth failure when calling " \
+                                     "kinit cmd '/fake/kinit': %s" % expected_err
+
+    # pylint bug: https://github.com/PyCQA/pylint/issues/511
+    # pylint: disable=undefined-variable
+    @pytest.mark.parametrize('use_pexpect', (False, True),)
+    def test_kinit_error(self, use_pexpect):
+        mechanism = "subprocess"
+        expected_err = "kinit: krb5_parse_name: " \
+                       "Configuration file does not specify default realm"
+
+        def mock_popen_communicate(input=None, timeout=None):
+            return b"", to_bytes(expected_err)
+
+        mock_subprocess = MagicMock()
+        mock_subprocess.Popen.return_value.communicate = mock_popen_communicate
+        mock_subprocess.Popen.return_value.returncode = 1
+
+        mock_pexpect = None
+        if use_pexpect:
+            mechanism = "pexpect"
+            expected_err = "Configuration file does not specify default realm"
+
+            mock_pexpect = MagicMock()
+            mock_pexpect.spawn.return_value.expect = MagicMock(side_effect=OSError)
+            mock_pexpect.spawn.return_value.read.return_value = to_bytes(expected_err)
+            mock_pexpect.spawn.return_value.exitstatus = 1
+
+        modules = {
+            'pexpect': mock_pexpect,
+            'subprocess': mock_subprocess,
+        }
+
+        with patch.dict(sys.modules, modules):
+            pc = PlayContext()
+            new_stdin = StringIO()
+
+            connection_loader._module_cache = {}
+            conn = connection_loader.get('winrm', pc, new_stdin)
+            conn.set_options(var_options={"_extras": {}})
+
+            with pytest.raises(AnsibleConnectionFailure) as err:
+                conn._kerb_auth("invaliduser", "pass")
+
+            assert str(err.value) == "Kerberos auth failure for principal " \
+                                     "invaliduser with %s: %s" % (mechanism, expected_err)