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

* 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

(cherry picked from commit 5e28e282a5)
This commit is contained in:
Jordan Borean 2018-05-24 07:17:21 +10:00 committed by GitHub
parent 2cb9f7f5b9
commit a63b6d363b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 188 additions and 12 deletions

View file

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

View file

@ -285,33 +285,60 @@ class Connection(ConnectionBase):
# 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)

View file

@ -13,6 +13,8 @@ import pytest
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 @@ class TestConnectionWinRM(object):
% (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)