A rough sketch of an ansible connection plugin that uses aws ec2-instance-connect to publish ssh keys and then proxies through SSM. Not suitable for production use, it does not deduplicate AWS api calls and will exhaust your rate limit.

aws_ec2ic.py

# Released under the same license as found in https://github.com/ansible-collections/community.aws (GPL V2.0+ as of this writing)

from __future__ import (absolute_import, division, print_function)
from paramiko.rsakey import RSAKey
from distutils.version import LooseVersion

from ansible.module_utils.compat.paramiko import PARAMIKO_IMPORT_ERR, paramiko
from ansible.utils.display import Display
from ansible.module_utils.basic import missing_required_lib
from ansible.plugins.connection.paramiko_ssh import MyAddPolicy, Connection as ParamikoSshConnection
from ansible.module_utils._text import to_native, to_text
from ansible.errors import AnsibleConnectionFailure, AnsibleAuthenticationFailure, AnsibleError, AnsibleFileNotFound
import sys
import time
import json
import os
import shlex
from collections import OrderedDict
from datetime import datetime
from functools import lru_cache
__metaclass__ = type

import base64

display = Display()

DOCUMENTATION = """
    author:
    - Dan Fallon <Daniel@Fallon.io>
    connection: aws_ec2ic
    short_description: execute via Paramiko SSH, push key via ec2-instance-connect 
    description:
        - This connection plugin connects to a remote host via SSH, but uses 
          ec2-instance-connect to push a randomly generated ssh key to the host.
        - See the paramiko_ssh connection plugin for more details. 
    requirements:
    - Currently requires python 3+
    - The remote EC2 instance must have ec2-instance-connect configured.
    - To use session-manager as a jumphost, the control machine must have the aws session manager plugin installed
    - To use session-manager as a jumphost, the remote machine must be running the AWS Systems Manager Agent (SSM Agent)
    options:
      access_key_id:
        description: The STS access key to use for aws api calls to ec2-instance-connect, ec2, and ssm.
        env:
          - name: AWS_ACCESS_KEY_ID
        vars:
          - name: ansible_aws_access_key_id
          - name: ansible_aws_ec2ic_access_key_id
        version_added: 1.3.0
      secret_access_key:
        description: The STS secret key to use for aws api calls to ec2-instance-connect, ec2, and ssm.
        env:
          - name: AWS_ACCESS_KEY_ID
        vars:
          - name: ansible_aws_secret_access_key
          - name: ansible_aws_ec2ic_secret_access_key
        version_added: 1.3.0
      session_token:
        description: The STS session token to use for aws api calls to ec2-instance-connect, ec2, and ssm.
        env:
          - name: AWS_SESSION_TOKEN
        vars:
          - name: ansible_aws_session_token
          - name: ansible_aws_ec2ic_session_token
        version_added: 1.3.0
      region:
        description: The aws region to use for aws api calls to ec2-instance-connect, ec2, and ssm.
        env:
        - name: AWS_DEFAULT_REGION
        vars:
          - name: ansible_aws_region
          - name: ansible_aws_ec2ic_region
        default: 'us-east-1'
      remote_addr:
        description:
            - Address of the remote target
        default: inventory_hostname
        vars:
            - name: ansible_host
            - name: ansible_ssh_host
            - name: ansible_paramiko_host
            - name: ansible_ec2_instance_id
      instance_id:
        description:
            - The instance id of the remote target
        vars:
            - name: ansible_aws_instance_id
      availability_zone:
        description:
            - The availability zone of the remote target
        vars:
            - name: availability_zone
            - name: ansible_aws_availability_zone
            - name: ansible_aws_ec2ic_availability_zone
      remote_user:
        description:
            - User to login/authenticate as
            - User to push public key for ec2-instance-connect
            - Can be set from the CLI via the C(--user) or C(-u) options.
        vars:
            - name: ansible_user
            - name: ansible_ssh_user
            - name: ansible_paramiko_user
        env:
            - name: ANSIBLE_REMOTE_USER
            - name: ANSIBLE_PARAMIKO_REMOTE_USER
        ini:
            - section: defaults
              key: remote_user
            - section: paramiko_connection
              key: remote_user
            - section: ec2ic_connection
              key: remote_user
      session_manager_plugin:
        description: This defines the location of the session-manager-plugin binary.
        vars:
          - name: ansible_aws_ssm_plugin
        default: '/usr/local/bin/session-manager-plugin'
      host_key_auto_add:
        description: 'TODO: write it'
        env: [{name: ANSIBLE_PARAMIKO_HOST_KEY_AUTO_ADD}]
        ini:
          - {key: host_key_auto_add, section: paramiko_connection}
        type: boolean
      proxy_command:
        default: ''
        description:
            - Proxy information for running the connection via a jumphost
            - Also this plugin will scan 'ssh_args', 'ssh_extra_args' and 'ssh_common_args' from the 'ssh' plugin settings for proxy information if set.
        env: [{name: ANSIBLE_PARAMIKO_PROXY_COMMAND}]
        ini:
          - {key: proxy_command, section: paramiko_connection}
      pty:
        default: True
        description: 'TODO: write it'
        env:
          - name: ANSIBLE_PARAMIKO_PTY
        ini:
          - section: paramiko_connection
            key: pty
        type: boolean
      record_host_keys:
        default: True
        description: 'TODO: write it'
        env: [{name: ANSIBLE_PARAMIKO_RECORD_HOST_KEYS}]
        ini:
          - section: paramiko_connection
            key: record_host_keys
        type: boolean
      host_key_checking:
        description: 'Set this to "False" if you want to avoid host key checking by the underlying tools Ansible uses to connect to the host'
        type: boolean
        default: True
        env:
          - name: ANSIBLE_HOST_KEY_CHECKING
          - name: ANSIBLE_SSH_HOST_KEY_CHECKING
            version_added: '2.5'
          - name: ANSIBLE_PARAMIKO_HOST_KEY_CHECKING
            version_added: '2.5'
        ini:
          - section: defaults
            key: host_key_checking
          - section: paramiko_connection
            key: host_key_checking
            version_added: '2.5'
        vars:
          - name: ansible_host_key_checking
            version_added: '2.5'
          - name: ansible_ssh_host_key_checking
            version_added: '2.5'
          - name: ansible_paramiko_host_key_checking
            version_added: '2.5'
      use_persistent_connections:
        description: 'Toggles the use of persistence for connections'
        type: boolean
        default: False
        env:
          - name: ANSIBLE_USE_PERSISTENT_CONNECTIONS
        ini:
          - section: defaults
            key: use_persistent_connections
# TODO:
# timeout=self._play_context.timeout,
"""

if sys.version_info[0] < 3:
    raise Exception("Must be using Python 3")

try:
    import boto3
    HAS_BOTO_3 = True
except ImportError as e:
    HAS_BOTO_3_ERROR = str(e)
    HAS_BOTO_3 = False


@lru_cache(maxsize=128)
def get_boto_client(service,
                    aws_access_key_id=None, aws_secret_access_key=None,
                    aws_session_token=None, region_name=None):
    return boto3.client(service,
                 aws_access_key_id=aws_access_key_id,
                 aws_secret_access_key=aws_secret_access_key,
                 aws_session_token=aws_session_token,
                 region_name=region_name)


class InstanceAZCache(OrderedDict):
    def __init__(self, maxsize=1024):
        self.maxsize = maxsize
        super().__init__()

    def __getitem__(self, key):
        value = super().__getitem__(key)
        self.move_to_end(key)
        return value

    def __setitem__(self, key, value):
        if key in self:
            self.move_to_end(key)
        super().__setitem__(key, value)
        if len(self) > self.maxsize:
            oldest = next(iter(self))
            del self[oldest]

    def lookup(self, instance_id, client=None, default=None):
        # Check cache, we assume that instance ids will never change AZ
        if instance_id in self:
            return self[instance_id]

        if default:
            self[instance_id] = default
        elif client:
            params = {"InstanceIds": [instance_id]}
            display.vvvv(f"EC2 describe_instances({params})")
            response = client.describe_instances(**params)
            self[instance_id] = response['Reservations'][0]['Instances'][0]['Placement']['AvailabilityZone']
        
        return self[instance_id]

EC2_INSTANCE_AZ_CACHE = InstanceAZCache()

class PublicKeyTTLCache():
    ttl = 0
    cache = {}

    def __init__(self, ttl=30):
        self.ttl = ttl

    def set_available(self, instance_id, remote_user, public_key):
        key = str((instance_id, remote_user, public_key))
        display.vvvvv(f"TTLCache Writing key ({key})")
        self.cache[key] = time.time() + self.ttl
        display.vvvv(f"{self.cache}")
        key = str((instance_id, remote_user, public_key))
        display.vvvv(f"{self.cache[key]}")
        display.vvvv(f"ISAVAIL id: {hex(id(self))}")
        display.vvvv(f"ISAVAIL pid: {os.getpid()}")


    def is_available(self, instance_id, remote_user, public_key):
        key = str((instance_id, remote_user, public_key))
        display.vvvv(f"ISAVAIL {self.cache}")
        display.vvvv(f"ISAVAIL {self.cache.get(key, None)}")
        display.vvvv(f"ISAVAIL id: {hex(id(self))}")
        display.vvvv(f"ISAVAIL pid: {os.getpid()}")

        display.vvvvv(f"TTLCache Reading key ({key})")
        available = False
        value = self.cache.get(
            key, None)
        display.vvvvv(f"TTL VAl Lookup {value}")
        
        available = value and value > time.time()
        
        # if not available:
        #     self.cache.pop(key, None)

        return available

EC2IC_KEY_PUSHED_CACHE = PublicKeyTTLCache()

display.vvv("EC2INSTANCECONECT Generating new private key")
EC2IC_GENERATED_PRIVATE_KEY = RSAKey.generate(4096)
EC2IC_GENERATED_PUBLIC_KEY = "{0} {1} {2}".format(
    EC2IC_GENERATED_PRIVATE_KEY.get_name(),
    EC2IC_GENERATED_PRIVATE_KEY.get_base64(),
    f"Temporary Ansible Executor {datetime.now().isoformat()}"
)

class Connection(ParamikoSshConnection):
    ''' SSH based connections with key setup via ec2-instance-connect'''

    transport = 'aws_ec2instanceconnect'
    supports_persistence = True
    def __init__(self, *args, **kwargs):
        if sys.version_info [0] < 3:
            raise AnsibleError("")
        if not HAS_BOTO_3:
            raise AnsibleError('{0}: {1}'.format(missing_required_lib("boto3"), HAS_BOTO_3_ERROR))
        super(Connection, self).__init__(*args, **kwargs)
        self.set_option('look_for_keys', False)


    def _push_ssh_public_key(self):
        '''pushes public key to host via ec2-instance-connect'''
        instance_id = self.get_option('instance_id')
        remote_user = self._play_context.remote_user or self.get_option('remote_user')

        if EC2IC_KEY_PUSHED_CACHE.is_available(
            instance_id, remote_user, EC2IC_GENERATED_PUBLIC_KEY):
            display.vvvv(
                f"EC2IC Reusing generated public key for instance ({instance_id})")
            return

        availability_zone = EC2_INSTANCE_AZ_CACHE.lookup(
            instance_id, client = self._get_boto_client('ec2'),
            default = self.get_option('availability_zone'))
        try:
            client = self._get_boto_client('ec2-instance-connect')
            params = {
                "InstanceId": instance_id,
                "InstanceOSUser": remote_user,
                "SSHPublicKey": EC2IC_GENERATED_PUBLIC_KEY,
                "AvailabilityZone": availability_zone
            }
            display.vvv(
                f"EC2IC Pushing generated public key to instance ({instance_id})")
            display.vvvv(
                f"EC2IC api call: send_ssh_public_key({params})")
            client.send_ssh_public_key(**params)
            EC2IC_KEY_PUSHED_CACHE.set_available(
                instance_id, remote_user, EC2IC_GENERATED_PUBLIC_KEY)
        except Exception as e:
            raise AnsibleConnectionFailure(
                f"EC2IC Failed to push ssh key to ec2 instance ({instance_id}): {e}")

    def _parse_proxy_command(self, port=None):
        '''creates the appropriate proxy command, ignoring user proxy command'''
        profile_name = ''
        region_name = self.get_option('region')
        instance_id = self.get_option('instance_id')
        ssm_parameters = dict()

        if port is not None:
            ssm_parameters["portNumber"] = (str(port),)

        executable = self.get_option('session_manager_plugin')
        try:
            client = self._get_boto_client('ssm')
            response = client.start_session(
                Target=instance_id, DocumentName="AWS-StartSSHSession", Parameters=ssm_parameters)
            self._session_id = response['SessionId']
        except Exception as e:
            raise AnsibleError(f"SSM Unable to create proxy session for instance ({instance_id})")
        proxy_command = shlex.join([
            executable,
            json.dumps(response),
            region_name,
            "StartSession",
            profile_name,
            json.dumps({"Target": instance_id}),
            client.meta.endpoint_url
        ])

        return {'sock': paramiko.ProxyCommand(proxy_command)}

    def _connect_uncached(self):
        ''' activates the connection object '''

        if paramiko is None:
            raise AnsibleError("paramiko is not installed: %s" %
                               to_native(PARAMIKO_IMPORT_ERR))

        port = self._play_context.port or 22
        display.vvv("ESTABLISH PARAMIKO SSH CONNECTION FOR USER: %s on PORT %s TO %s" % (self._play_context.remote_user, port, self._play_context.remote_addr),
                    host=self._play_context.remote_addr)

        ssh = paramiko.SSHClient()

        # override paramiko's default logger name
        if self._log_channel is not None:
            ssh.set_log_channel(self._log_channel)

        self.keyfile = os.path.expanduser("~/.ssh/known_hosts")

        if self.get_option('host_key_checking'):
            for ssh_known_hosts in ("/etc/ssh/ssh_known_hosts", "/etc/openssh/ssh_known_hosts"):
                try:
                    # TODO: check if we need to look at several possible locations, possible for loop
                    ssh.load_system_host_keys(ssh_known_hosts)
                    break
                except IOError:
                    pass  # file was not found, but not required to function
            ssh.load_system_host_keys()

        ssh_connect_kwargs = self._parse_proxy_command(port)

        ssh.set_missing_host_key_policy(MyAddPolicy(self._new_stdin, self))

        try:
            # paramiko 2.2 introduced auth_timeout parameter
            if LooseVersion(paramiko.__version__) >= LooseVersion('2.2.0'):
                ssh_connect_kwargs['auth_timeout'] = self._play_context.timeout
            self._push_ssh_public_key()
            ssh.connect(
                self._play_context.remote_addr.lower(),
                username=self._play_context.remote_user,
                pkey=EC2IC_GENERATED_PRIVATE_KEY,
                timeout=self._play_context.timeout,
                port=port,
                **ssh_connect_kwargs
            )
        except paramiko.ssh_exception.BadHostKeyException as e:
            raise AnsibleConnectionFailure(
                'host key mismatch for %s' % e.hostname)
        except paramiko.ssh_exception.AuthenticationException as e:
            msg = 'Failed to authenticate: {0}'.format(to_text(e))
            raise AnsibleAuthenticationFailure(msg)
        except Exception as e:
            msg = to_text(e)
            if u"PID check failed" in msg:
                raise AnsibleError(
                    "paramiko version issue, please upgrade paramiko on the machine running ansible")
            elif u"Private key file is encrypted" in msg:
                msg = 'ssh %s@%s:%s : %s\nTo connect as a different user, use -u <username>.' % (
                    self._play_context.remote_user, self._play_context.remote_addr, port, msg)
                raise AnsibleConnectionFailure(msg)
            else:
                raise AnsibleConnectionFailure(msg)

        return ssh

    def _get_boto_client(self, service):
        ''' Gets a boto3 client based on the STS token '''

        aws_access_key_id = self.get_option('access_key_id')
        aws_secret_access_key = self.get_option('secret_access_key')
        aws_session_token = self.get_option('session_token')
        region_name = self.get_option('region')
        return get_boto_client(service,
                               aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key,
                               aws_session_token=aws_session_token, region_name=region_name)