"""Profiles to represent individual test hosts or a user-provided inventory file.""" from __future__ import annotations import abc import dataclasses import os import tempfile import time import typing as t from .io import ( write_text_file, ) from .config import ( CommonConfig, EnvironmentConfig, IntegrationConfig, TerminateMode, ) from .host_configs import ( ControllerConfig, ControllerHostConfig, DockerConfig, HostConfig, NetworkInventoryConfig, NetworkRemoteConfig, OriginConfig, PosixConfig, PosixRemoteConfig, PosixSshConfig, PythonConfig, RemoteConfig, VirtualPythonConfig, WindowsInventoryConfig, WindowsRemoteConfig, ) from .core_ci import ( AnsibleCoreCI, SshKey, ) from .util import ( ApplicationError, SubprocessError, cache, display, get_type_map, sanitize_host_name, sorted_versions, ) from .util_common import ( intercept_python, ) from .docker_util import ( docker_exec, docker_rm, get_docker_hostname, ) from .bootstrap import ( BootstrapDocker, BootstrapRemote, ) from .venv import ( get_virtual_python, ) from .ssh import ( SshConnectionDetail, ) from .ansible_util import ( ansible_environment, get_hosts, parse_inventory, ) from .containers import ( CleanupMode, HostType, get_container_database, run_support_container, ) from .connections import ( Connection, DockerConnection, LocalConnection, SshConnection, ) from .become import ( Su, Sudo, ) TControllerHostConfig = t.TypeVar('TControllerHostConfig', bound=ControllerHostConfig) THostConfig = t.TypeVar('THostConfig', bound=HostConfig) TPosixConfig = t.TypeVar('TPosixConfig', bound=PosixConfig) TRemoteConfig = t.TypeVar('TRemoteConfig', bound=RemoteConfig) @dataclasses.dataclass(frozen=True) class Inventory: """Simple representation of an Ansible inventory.""" host_groups: t.Dict[str, t.Dict[str, t.Dict[str, str]]] extra_groups: t.Optional[t.Dict[str, t.List[str]]] = None @staticmethod def create_single_host(name, variables): # type: (str, t.Dict[str, str]) -> Inventory """Return an inventory instance created from the given hostname and variables.""" return Inventory(host_groups=dict(all={name: variables})) def write(self, args, path): # type: (CommonConfig, str) -> None """Write the given inventory to the specified path on disk.""" # NOTE: Switching the inventory generation to write JSON would be nice, but is currently not possible due to the use of hard-coded inventory filenames. # The name `inventory` works for the POSIX integration tests, but `inventory.winrm` and `inventory.networking` will only parse in INI format. # If tests are updated to use the `INVENTORY_PATH` environment variable, then this could be changed. # Also, some tests detect the test type by inspecting the suffix on the inventory filename, which would break if it were changed. inventory_text = '' for group, hosts in self.host_groups.items(): inventory_text += f'[{group}]\n' for host, variables in hosts.items(): kvp = ' '.join(f'{key}="{value}"' for key, value in variables.items()) inventory_text += f'{host} {kvp}\n' inventory_text += '\n' for group, children in (self.extra_groups or {}).items(): inventory_text += f'[{group}]\n' for child in children: inventory_text += f'{child}\n' inventory_text += '\n' inventory_text = inventory_text.strip() if not args.explain: write_text_file(path, inventory_text) display.info(f'>>> Inventory\n{inventory_text}', verbosity=3) class HostProfile(t.Generic[THostConfig], metaclass=abc.ABCMeta): """Base class for host profiles.""" def __init__(self, *, args, # type: EnvironmentConfig config, # type: THostConfig targets, # type: t.Optional[t.List[HostConfig]] ): # type: (...) -> None self.args = args self.config = config self.controller = bool(targets) self.targets = targets or [] self.state = {} # type: t.Dict[str, t.Any] """State that must be persisted across delegation.""" self.cache = {} # type: t.Dict[str, t.Any] """Cache that must not be persisted across delegation.""" def provision(self): # type: () -> None """Provision the host before delegation.""" def setup(self): # type: () -> None """Perform out-of-band setup before delegation.""" def deprovision(self): # type: () -> None """Deprovision the host after delegation has completed.""" def wait(self): # type: () -> None """Wait for the instance to be ready. Executed before delegation for the controller and after delegation for targets.""" def configure(self): # type: () -> None """Perform in-band configuration. Executed before delegation for the controller and after delegation for targets.""" def __getstate__(self): return {key: value for key, value in self.__dict__.items() if key not in ('args', 'cache')} def __setstate__(self, state): self.__dict__.update(state) # args will be populated after the instances are restored self.cache = {} class PosixProfile(HostProfile[TPosixConfig], metaclass=abc.ABCMeta): """Base class for POSIX host profiles.""" @property def python(self): # type: () -> PythonConfig """ The Python to use for this profile. If it is a virtual python, it will be created the first time it is requested. """ python = self.state.get('python') if not python: python = self.config.python if isinstance(python, VirtualPythonConfig): python = VirtualPythonConfig( version=python.version, system_site_packages=python.system_site_packages, path=os.path.join(get_virtual_python(self.args, python), 'bin', 'python'), ) self.state['python'] = python return python class ControllerHostProfile(PosixProfile[TControllerHostConfig], metaclass=abc.ABCMeta): """Base class for profiles usable as a controller.""" @abc.abstractmethod def get_origin_controller_connection(self): # type: () -> Connection """Return a connection for accessing the host as a controller from the origin.""" @abc.abstractmethod def get_working_directory(self): # type: () -> str """Return the working directory for the host.""" class SshTargetHostProfile(HostProfile[THostConfig], metaclass=abc.ABCMeta): """Base class for profiles offering SSH connectivity.""" @abc.abstractmethod def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" class RemoteProfile(SshTargetHostProfile[TRemoteConfig], metaclass=abc.ABCMeta): """Base class for remote instance profiles.""" @property def core_ci_state(self): # type: () -> t.Optional[t.Dict[str, str]] """The saved Ansible Core CI state.""" return self.state.get('core_ci') @core_ci_state.setter def core_ci_state(self, value): # type: (t.Dict[str, str]) -> None """The saved Ansible Core CI state.""" self.state['core_ci'] = value def provision(self): # type: () -> None """Provision the host before delegation.""" self.core_ci = self.create_core_ci(load=True) self.core_ci.start() self.core_ci_state = self.core_ci.save() def deprovision(self): # type: () -> None """Deprovision the host after delegation has completed.""" if self.args.remote_terminate == TerminateMode.ALWAYS or (self.args.remote_terminate == TerminateMode.SUCCESS and self.args.success): self.delete_instance() @property def core_ci(self): # type: () -> t.Optional[AnsibleCoreCI] """Return the cached AnsibleCoreCI instance, if any, otherwise None.""" return self.cache.get('core_ci') @core_ci.setter def core_ci(self, value): # type: (AnsibleCoreCI) -> None """Cache the given AnsibleCoreCI instance.""" self.cache['core_ci'] = value def get_instance(self): # type: () -> t.Optional[AnsibleCoreCI] """Return the current AnsibleCoreCI instance, loading it if not already loaded.""" if not self.core_ci and self.core_ci_state: self.core_ci = self.create_core_ci(load=False) self.core_ci.load(self.core_ci_state) return self.core_ci def delete_instance(self): """Delete the AnsibleCoreCI VM instance.""" core_ci = self.get_instance() if not core_ci: return # instance has not been provisioned core_ci.stop() def wait_for_instance(self): # type: () -> AnsibleCoreCI """Wait for an AnsibleCoreCI VM instance to become ready.""" core_ci = self.get_instance() core_ci.wait() return core_ci def create_core_ci(self, load): # type: (bool) -> AnsibleCoreCI """Create and return an AnsibleCoreCI instance.""" return AnsibleCoreCI( args=self.args, platform=self.config.platform, version=self.config.version, provider=self.config.provider, suffix='controller' if self.controller else 'target', load=load, ) class ControllerProfile(SshTargetHostProfile[ControllerConfig], PosixProfile[ControllerConfig]): """Host profile for the controller as a target.""" def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" settings = SshConnectionDetail( name='localhost', host='localhost', port=None, user='root', identity_file=SshKey(self.args).key, python_interpreter=self.args.controller_python.path, ) return [SshConnection(self.args, settings)] class DockerProfile(ControllerHostProfile[DockerConfig], SshTargetHostProfile[DockerConfig]): """Host profile for a docker instance.""" @property def container_name(self): # type: () -> t.Optional[str] """Return the stored container name, if any, otherwise None.""" return self.state.get('container_name') @container_name.setter def container_name(self, value): # type: (str) -> None """Store the given container name.""" self.state['container_name'] = value def provision(self): # type: () -> None """Provision the host before delegation.""" container = run_support_container( args=self.args, context='__test_hosts__', image=self.config.image, name=f'ansible-test-{"controller" if self.controller else "target"}-{self.args.session_name}', ports=[22], publish_ports=not self.controller, # connections to the controller over SSH are not required options=self.get_docker_run_options(), cleanup=CleanupMode.NO, ) if not container: return self.container_name = container.name def setup(self): # type: () -> None """Perform out-of-band setup before delegation.""" bootstrapper = BootstrapDocker( controller=self.controller, python_versions=[self.python.version], ssh_key=SshKey(self.args), ) setup_sh = bootstrapper.get_script() shell = setup_sh.splitlines()[0][2:] docker_exec(self.args, self.container_name, [shell], data=setup_sh) def deprovision(self): # type: () -> None """Deprovision the host after delegation has completed.""" if not self.container_name: return # provision was never called or did not succeed, so there is no container to remove if self.args.docker_terminate == TerminateMode.ALWAYS or (self.args.docker_terminate == TerminateMode.SUCCESS and self.args.success): docker_rm(self.args, self.container_name) def wait(self): # type: () -> None """Wait for the instance to be ready. Executed before delegation for the controller and after delegation for targets.""" if not self.controller: con = self.get_controller_target_connections()[0] for dummy in range(1, 60): try: con.run(['id'], capture=True) except SubprocessError as ex: if 'Permission denied' in ex.message: raise time.sleep(1) else: return def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" containers = get_container_database(self.args) access = containers.data[HostType.control]['__test_hosts__'][self.container_name] host = access.host_ip port = dict(access.port_map())[22] settings = SshConnectionDetail( name=self.config.name, user='root', host=host, port=port, identity_file=SshKey(self.args).key, python_interpreter=self.python.path, ) return [SshConnection(self.args, settings)] def get_origin_controller_connection(self): # type: () -> DockerConnection """Return a connection for accessing the host as a controller from the origin.""" return DockerConnection(self.args, self.container_name) def get_working_directory(self): # type: () -> str """Return the working directory for the host.""" return '/root' def get_docker_run_options(self): # type: () -> t.List[str] """Return a list of options needed to run the container.""" options = [ '--volume', '/sys/fs/cgroup:/sys/fs/cgroup:ro', f'--privileged={str(self.config.privileged).lower()}', ] if self.config.memory: options.extend([ f'--memory={self.config.memory}', f'--memory-swap={self.config.memory}', ]) if self.config.seccomp != 'default': options.extend(['--security-opt', f'seccomp={self.config.seccomp}']) docker_socket = '/var/run/docker.sock' if get_docker_hostname() != 'localhost' or os.path.exists(docker_socket): options.extend(['--volume', f'{docker_socket}:{docker_socket}']) return options class NetworkInventoryProfile(HostProfile[NetworkInventoryConfig]): """Host profile for a network inventory.""" class NetworkRemoteProfile(RemoteProfile[NetworkRemoteConfig]): """Host profile for a network remote instance.""" def wait(self): # type: () -> None """Wait for the instance to be ready. Executed before delegation for the controller and after delegation for targets.""" self.wait_until_ready() def get_inventory_variables(self): """Return inventory variables for accessing this host.""" core_ci = self.wait_for_instance() connection = core_ci.connection variables = dict( ansible_connection=self.config.connection, ansible_pipelining='yes', ansible_host=connection.hostname, ansible_port=connection.port, ansible_user=connection.username, ansible_ssh_private_key_file=core_ci.ssh_key.key, ansible_network_os=f'{self.config.collection}.{self.config.platform}' if self.config.collection else self.config.platform, ) return variables def wait_until_ready(self): # type: () -> None """Wait for the host to respond to an Ansible module request.""" core_ci = self.wait_for_instance() if not isinstance(self.args, IntegrationConfig): return # skip extended checks unless we're running integration tests inventory = Inventory.create_single_host(sanitize_host_name(self.config.name), self.get_inventory_variables()) env = ansible_environment(self.args) module_name = f'{self.config.collection + "." if self.config.collection else ""}{self.config.platform}_command' with tempfile.NamedTemporaryFile() as inventory_file: inventory.write(self.args, inventory_file.name) cmd = ['ansible', '-m', module_name, '-a', 'commands=?', '-i', inventory_file.name, 'all'] for dummy in range(1, 90): try: intercept_python(self.args, self.args.controller_python, cmd, env) except SubprocessError: time.sleep(10) else: return raise ApplicationError(f'Timeout waiting for {self.config.name} instance {core_ci.instance_id}.') def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" core_ci = self.wait_for_instance() settings = SshConnectionDetail( name=core_ci.name, host=core_ci.connection.hostname, port=core_ci.connection.port, user=core_ci.connection.username, identity_file=core_ci.ssh_key.key, ) return [SshConnection(self.args, settings)] class OriginProfile(ControllerHostProfile[OriginConfig]): """Host profile for origin.""" def get_origin_controller_connection(self): # type: () -> LocalConnection """Return a connection for accessing the host as a controller from the origin.""" return LocalConnection(self.args) def get_working_directory(self): # type: () -> str """Return the working directory for the host.""" return os.getcwd() class PosixRemoteProfile(ControllerHostProfile[PosixRemoteConfig], RemoteProfile[PosixRemoteConfig]): """Host profile for a POSIX remote instance.""" def wait(self): # type: () -> None """Wait for the instance to be ready. Executed before delegation for the controller and after delegation for targets.""" self.wait_until_ready() def configure(self): # type: () -> None """Perform in-band configuration. Executed before delegation for the controller and after delegation for targets.""" # a target uses a single python version, but a controller may include additional versions for targets running on the controller python_versions = [self.python.version] + [target.python.version for target in self.targets if isinstance(target, ControllerConfig)] python_versions = sorted_versions(list(set(python_versions))) core_ci = self.wait_for_instance() pwd = self.wait_until_ready() display.info(f'Remote working directory: {pwd}', verbosity=1) bootstrapper = BootstrapRemote( controller=self.controller, platform=self.config.platform, platform_version=self.config.version, python_versions=python_versions, ssh_key=core_ci.ssh_key, ) setup_sh = bootstrapper.get_script() shell = setup_sh.splitlines()[0][2:] ssh = self.get_origin_controller_connection() ssh.run([shell], data=setup_sh) def get_ssh_connection(self): # type: () -> SshConnection """Return an SSH connection for accessing the host.""" core_ci = self.wait_for_instance() settings = SshConnectionDetail( name=core_ci.name, user=core_ci.connection.username, host=core_ci.connection.hostname, port=core_ci.connection.port, identity_file=core_ci.ssh_key.key, python_interpreter=self.python.path, ) if settings.user == 'root': become = None elif self.config.platform == 'freebsd': become = Su() elif self.config.platform == 'macos': become = Sudo() elif self.config.platform == 'rhel': become = Sudo() else: raise NotImplementedError(f'Become support has not been implemented for platform "{self.config.platform}" and user "{settings.user}" is not root.') return SshConnection(self.args, settings, become) def wait_until_ready(self): # type: () -> str """Wait for instance to respond to SSH, returning the current working directory once connected.""" core_ci = self.wait_for_instance() for dummy in range(1, 90): try: return self.get_working_directory() except SubprocessError as ex: if 'Permission denied' in ex.message: raise time.sleep(10) raise ApplicationError(f'Timeout waiting for {self.config.name} instance {core_ci.instance_id}.') def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" return [self.get_ssh_connection()] def get_origin_controller_connection(self): # type: () -> SshConnection """Return a connection for accessing the host as a controller from the origin.""" return self.get_ssh_connection() def get_working_directory(self): # type: () -> str """Return the working directory for the host.""" if not self.pwd: ssh = self.get_origin_controller_connection() stdout = ssh.run(['pwd'], capture=True)[0] if self.args.explain: return '/pwd' pwd = stdout.strip().splitlines()[-1] if not pwd.startswith('/'): raise Exception(f'Unexpected current working directory "{pwd}" from "pwd" command output:\n{stdout.strip()}') self.pwd = pwd return self.pwd @property def pwd(self): # type: () -> t.Optional[str] """Return the cached pwd, if any, otherwise None.""" return self.cache.get('pwd') @pwd.setter def pwd(self, value): # type: (str) -> None """Cache the given pwd.""" self.cache['pwd'] = value class PosixSshProfile(SshTargetHostProfile[PosixSshConfig], PosixProfile[PosixSshConfig]): """Host profile for a POSIX SSH instance.""" def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" settings = SshConnectionDetail( name='target', user=self.config.user, host=self.config.host, port=self.config.port, identity_file=SshKey(self.args).key, python_interpreter=self.python.path, ) return [SshConnection(self.args, settings)] class WindowsInventoryProfile(SshTargetHostProfile[WindowsInventoryConfig]): """Host profile for a Windows inventory.""" def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" inventory = parse_inventory(self.args, self.config.path) hosts = get_hosts(inventory, 'windows') identity_file = SshKey(self.args).key settings = [SshConnectionDetail( name=name, host=config['ansible_host'], port=22, user=config['ansible_user'], identity_file=identity_file, shell_type='powershell', ) for name, config in hosts.items()] if settings: details = '\n'.join(f'{ssh.name} {ssh.user}@{ssh.host}:{ssh.port}' for ssh in settings) display.info(f'Generated SSH connection details from inventory:\n{details}', verbosity=1) return [SshConnection(self.args, setting) for setting in settings] class WindowsRemoteProfile(RemoteProfile[WindowsRemoteConfig]): """Host profile for a Windows remote instance.""" def wait(self): # type: () -> None """Wait for the instance to be ready. Executed before delegation for the controller and after delegation for targets.""" self.wait_until_ready() def get_inventory_variables(self): """Return inventory variables for accessing this host.""" core_ci = self.wait_for_instance() connection = core_ci.connection variables = dict( ansible_connection='winrm', ansible_pipelining='yes', ansible_winrm_server_cert_validation='ignore', ansible_host=connection.hostname, ansible_port=connection.port, ansible_user=connection.username, ansible_password=connection.password, ansible_ssh_private_key_file=core_ci.ssh_key.key, ) # HACK: force 2016 to use NTLM + HTTP message encryption if self.config.version == '2016': variables.update( ansible_winrm_transport='ntlm', ansible_winrm_scheme='http', ansible_port='5985', ) return variables def wait_until_ready(self): # type: () -> None """Wait for the host to respond to an Ansible module request.""" core_ci = self.wait_for_instance() if not isinstance(self.args, IntegrationConfig): return # skip extended checks unless we're running integration tests inventory = Inventory.create_single_host(sanitize_host_name(self.config.name), self.get_inventory_variables()) env = ansible_environment(self.args) module_name = 'ansible.windows.win_ping' with tempfile.NamedTemporaryFile() as inventory_file: inventory.write(self.args, inventory_file.name) cmd = ['ansible', '-m', module_name, '-i', inventory_file.name, 'all'] for dummy in range(1, 120): try: intercept_python(self.args, self.args.controller_python, cmd, env) except SubprocessError: time.sleep(10) else: return raise ApplicationError(f'Timeout waiting for {self.config.name} instance {core_ci.instance_id}.') def get_controller_target_connections(self): # type: () -> t.List[SshConnection] """Return SSH connection(s) for accessing the host as a target from the controller.""" core_ci = self.wait_for_instance() settings = SshConnectionDetail( name=core_ci.name, host=core_ci.connection.hostname, port=22, user=core_ci.connection.username, identity_file=core_ci.ssh_key.key, shell_type='powershell', ) return [SshConnection(self.args, settings)] @cache def get_config_profile_type_map(): # type: () -> t.Dict[t.Type[HostConfig], t.Type[HostProfile]] """Create and return a mapping of HostConfig types to HostProfile types.""" return get_type_map(HostProfile, HostConfig) def create_host_profile( args, # type: EnvironmentConfig config, # type: HostConfig controller, # type: bool ): # type: (...) -> HostProfile """Create and return a host profile from the given host configuration.""" profile_type = get_config_profile_type_map()[type(config)] profile = profile_type(args=args, config=config, targets=args.targets if controller else None) return profile