Remove reliance on sshpass and utilize SSH_ASKPASS (#83936)

* Add SSH_ASKPASS as an alternative means to provide ssh with passwords
This commit is contained in:
Matt Martz
2025-03-17 10:07:44 -05:00
committed by GitHub
parent 2e7e5b65e7
commit 3684b4824d
9 changed files with 242 additions and 105 deletions
@@ -1,3 +1,5 @@
needs/ssh
shippable/posix/group3
needs/target/connection
needs/target/setup_test_user
setup/always/setup_passlib_controller # required for setup_test_user
@@ -19,6 +19,7 @@ if command -v sshpass > /dev/null; then
# that the flag gets passed to sshpass.
timeout 5 ansible -m ping \
-e ansible_connection=ssh \
-e ansible_ssh_password_mechanism=sshpass \
-e ansible_sshpass_prompt=notThis: \
-e ansible_password=foo \
-e ansible_user=definitelynotroot \
@@ -34,6 +35,7 @@ if command -v sshpass > /dev/null; then
else
ansible -m ping \
-e ansible_connection=ssh \
-e ansible_ssh_password_mechanism=sshpass \
-e ansible_sshpass_prompt=notThis: \
-e ansible_password=foo \
-e ansible_user=definitelynotroot \
@@ -82,3 +84,5 @@ ANSIBLE_SSH_CONTROL_PATH='/tmp/ssh cp with spaces' ansible -m ping all -e ansibl
# Test that timeout on waiting on become is an unreachable error
ansible-playbook test_unreachable_become_timeout.yml "$@"
ANSIBLE_ROLES_PATH=../ ansible-playbook "$@" -i ../../inventory test_ssh_askpass.yml
@@ -0,0 +1,57 @@
- hosts: all
tasks:
- import_role:
role: setup_test_user
# macos currently allows password auth, and macos/15.3 prevents restarting sshd
- when: ansible_facts.system != 'Darwin'
block:
- find:
paths: /etc/ssh
recurse: true
contains: 'PasswordAuthentication'
register: sshd_confs
- lineinfile:
path: '{{ item }}'
regexp: '^PasswordAuthentication'
line: PasswordAuthentication yes
loop: '{{ sshd_confs.files|default([{"path": "/etc/ssh/sshd_config"}], true)|map(attribute="path") }}'
- service:
name: ssh{{ '' if ansible_facts.os_family == 'Debian' else 'd' }}
state: restarted
when: ansible_facts.system != 'Darwin'
- 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={{ test_user_plaintext_password }}
environment:
ANSIBLE_NOCOLOR: "1"
ANSIBLE_FORCE_COLOR: "0"
register: askpass_out
- debug:
var: askpass_out
- assert:
that:
- '"EXEC ssh " in askpass_out.stdout'
- '"sshpass" not in askpass_out.stdout'
- askpass_out.stdout is search('uid=\d+\(' ~ test_user_name ~ '\)')
@@ -1,6 +1,8 @@
- name: set variables
set_fact:
test_user_group: staff
test_user_groups:
- com.apple.access_ssh
- name: set plaintext password
no_log: yes
-64
View File
@@ -23,7 +23,6 @@ from selectors import SelectorKey, EVENT_READ
import pytest
from ansible.errors import AnsibleAuthenticationFailure
import unittest
from unittest.mock import patch, MagicMock, PropertyMock
from ansible.errors import AnsibleError, AnsibleConnectionFailure, AnsibleFileNotFound
@@ -54,22 +53,6 @@ class TestConnectionBaseClass(unittest.TestCase):
res = conn._connect()
self.assertEqual(conn, res)
ssh.SSHPASS_AVAILABLE = False
self.assertFalse(conn._sshpass_available())
ssh.SSHPASS_AVAILABLE = True
self.assertTrue(conn._sshpass_available())
with patch('subprocess.Popen') as p:
ssh.SSHPASS_AVAILABLE = None
p.return_value = MagicMock()
self.assertTrue(conn._sshpass_available())
ssh.SSHPASS_AVAILABLE = None
p.return_value = None
p.side_effect = OSError()
self.assertFalse(conn._sshpass_available())
conn.close()
self.assertFalse(conn._connected)
@@ -412,29 +395,6 @@ class TestSSHConnectionRun(object):
assert self.conn._send_initial_data.call_count == 1
assert self.conn._send_initial_data.call_args[0][1] == 'this is input data'
def test_with_password(self):
# test with a password set to trigger the sshpass write
self.pc.password = '12345'
self.mock_popen_res.stdout.read.side_effect = [b"some data", b"", b""]
self.mock_popen_res.stderr.read.side_effect = [b""]
self.mock_selector.select.side_effect = [
[(SelectorKey(self.mock_popen_res.stdout, 1001, [EVENT_READ], None), EVENT_READ)],
[(SelectorKey(self.mock_popen_res.stdout, 1001, [EVENT_READ], None), EVENT_READ)],
[(SelectorKey(self.mock_popen_res.stderr, 1002, [EVENT_READ], None), EVENT_READ)],
[(SelectorKey(self.mock_popen_res.stdout, 1001, [EVENT_READ], None), EVENT_READ)],
[]]
self.mock_selector.get_map.side_effect = lambda: True
return_code, b_stdout, b_stderr = self.conn._run(["ssh", "is", "a", "cmd"], "this is more data")
assert return_code == 0
assert b_stdout == b'some data'
assert b_stderr == b''
assert self.mock_selector.register.called is True
assert self.mock_selector.register.call_count == 2
assert self.conn._send_initial_data.called is True
assert self.conn._send_initial_data.call_count == 1
assert self.conn._send_initial_data.call_args[0][1] == 'this is more data'
def _password_with_prompt_examine_output(self, sourice, state, b_chunk, sudoable):
if state == 'awaiting_prompt':
self.conn._flags['become_prompt'] = True
@@ -525,30 +485,6 @@ class TestSSHConnectionRun(object):
@pytest.mark.usefixtures('mock_run_env')
class TestSSHConnectionRetries(object):
def test_incorrect_password(self, monkeypatch):
self.conn.set_option('host_key_checking', False)
self.conn.set_option('reconnection_retries', 5)
self.mock_popen_res.stdout.read.side_effect = [b'']
self.mock_popen_res.stderr.read.side_effect = [b'Permission denied, please try again.\r\n']
type(self.mock_popen_res).returncode = PropertyMock(side_effect=[5] * 4)
self.mock_selector.select.side_effect = [
[(SelectorKey(self.mock_popen_res.stdout, 1001, [EVENT_READ], None), EVENT_READ)],
[(SelectorKey(self.mock_popen_res.stderr, 1002, [EVENT_READ], None), EVENT_READ)],
[],
]
self.mock_selector.get_map.side_effect = lambda: True
self.conn._build_command = MagicMock()
self.conn._build_command.return_value = [b'sshpass', b'-d41', b'ssh', b'-C']
exception_info = pytest.raises(AnsibleAuthenticationFailure, self.conn.exec_command, 'sshpass', 'some data')
assert exception_info.value.message == ('Invalid/incorrect username/password. Skipping remaining 5 retries to prevent account lockout: '
'Permission denied, please try again.')
assert self.mock_popen.call_count == 1
def test_retry_then_success(self, monkeypatch):
self.conn.set_option('host_key_checking', False)
self.conn.set_option('reconnection_retries', 3)