summaryrefslogtreecommitdiff
path: root/test/lib/ansible_test/_internal/ssh.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/lib/ansible_test/_internal/ssh.py')
-rw-r--r--test/lib/ansible_test/_internal/ssh.py51
1 files changed, 46 insertions, 5 deletions
diff --git a/test/lib/ansible_test/_internal/ssh.py b/test/lib/ansible_test/_internal/ssh.py
index a5b40c8bbb..fd01ff253f 100644
--- a/test/lib/ansible_test/_internal/ssh.py
+++ b/test/lib/ansible_test/_internal/ssh.py
@@ -2,6 +2,7 @@
from __future__ import annotations
import dataclasses
+import itertools
import json
import os
import random
@@ -38,10 +39,40 @@ class SshConnectionDetail:
identity_file: str
python_interpreter: t.Optional[str] = None
shell_type: t.Optional[str] = None
+ enable_rsa_sha1: bool = False
def __post_init__(self):
self.name = sanitize_host_name(self.name)
+ @property
+ def options(self) -> dict[str, str]:
+ """OpenSSH config options, which can be passed to the `ssh` CLI with the `-o` argument."""
+ options: dict[str, str] = {}
+
+ if self.enable_rsa_sha1:
+ # Newer OpenSSH clients connecting to older SSH servers must explicitly enable ssh-rsa support.
+ # OpenSSH 8.8, released on 2021-09-26, deprecated using RSA with the SHA-1 hash algorithm (ssh-rsa).
+ # OpenSSH 7.2, released on 2016-02-29, added support for using RSA with SHA-256/512 hash algorithms.
+ # See: https://www.openssh.com/txt/release-8.8
+ algorithms = '+ssh-rsa' # append the algorithm to the default list, requires OpenSSH 7.0 or later
+
+ options.update(dict(
+ # Host key signature algorithms that the client wants to use.
+ # Available options can be found with `ssh -Q HostKeyAlgorithms` or `ssh -Q key` on older clients.
+ # This option was updated in OpenSSH 7.0, released on 2015-08-11, to support the "+" prefix.
+ # See: https://www.openssh.com/txt/release-7.0
+ HostKeyAlgorithms=algorithms,
+ # Signature algorithms that will be used for public key authentication.
+ # Available options can be found with `ssh -Q PubkeyAcceptedAlgorithms` or `ssh -Q key` on older clients.
+ # This option was added in OpenSSH 7.0, released on 2015-08-11.
+ # See: https://www.openssh.com/txt/release-7.0
+ # This option is an alias for PubkeyAcceptedAlgorithms, which was added in OpenSSH 8.5.
+ # See: https://www.openssh.com/txt/release-8.5
+ PubkeyAcceptedKeyTypes=algorithms,
+ ))
+
+ return options
+
class SshProcess:
"""Wrapper around an SSH process."""
@@ -141,7 +172,7 @@ def create_ssh_command(
if ssh.user:
cmd.extend(['-l', ssh.user]) # user to log in as on the remote machine
- ssh_options = dict(
+ ssh_options: dict[str, t.Union[int, str]] = dict(
BatchMode='yes',
ExitOnForwardFailure='yes',
LogLevel='ERROR',
@@ -153,9 +184,7 @@ def create_ssh_command(
ssh_options.update(options or {})
- for key, value in sorted(ssh_options.items()):
- cmd.extend(['-o', '='.join([key, str(value)])])
-
+ cmd.extend(ssh_options_to_list(ssh_options))
cmd.extend(cli_args or [])
cmd.append(ssh.host)
@@ -165,6 +194,18 @@ def create_ssh_command(
return cmd
+def ssh_options_to_list(options: t.Union[dict[str, t.Union[int, str]], dict[str, str]]) -> list[str]:
+ """Format a dictionary of SSH options as a list suitable for passing to the `ssh` command."""
+ return list(itertools.chain.from_iterable(
+ ('-o', f'{key}={value}') for key, value in sorted(options.items())
+ ))
+
+
+def ssh_options_to_str(options: t.Union[dict[str, t.Union[int, str]], dict[str, str]]) -> str:
+ """Format a dictionary of SSH options as a string suitable for passing as `ansible_ssh_extra_args` in inventory."""
+ return shlex.join(ssh_options_to_list(options))
+
+
def run_ssh_command(
args: EnvironmentConfig,
ssh: SshConnectionDetail,
@@ -245,7 +286,7 @@ def generate_ssh_inventory(ssh_connections: list[SshConnectionDetail]) -> str:
ansible_pipelining='yes',
ansible_python_interpreter=ssh.python_interpreter,
ansible_shell_type=ssh.shell_type,
- ansible_ssh_extra_args='-o UserKnownHostsFile=/dev/null', # avoid changing the test environment
+ ansible_ssh_extra_args=ssh_options_to_str(dict(UserKnownHostsFile='/dev/null', **ssh.options)), # avoid changing the test environment
ansible_ssh_host_key_checking='no',
))) for ssh in ssh_connections),
),