Dynamic Ansible inventory plugin for ssh config

ssh_config_inventory.py

# Drop in ~/.ansible/plugins/inventory/ssh_config_inventory.py

from __future__ import absolute_import, division, print_function
__metaclass__ = type

import os
import subprocess
from ansible.plugins.inventory import BaseInventoryPlugin
from ansible.errors import AnsibleParserError

DOCUMENTATION = '''
    name: ssh_config_inventory
    plugin_type: inventory
    short_description: Read hosts using `ssh -G`
    description:
      - Parse ~/.ssh/config for Host entries and use `ssh -G` to dynamically build inventory.
    options:
      plugin:
        description: Token that ensures this is a source file for the plugin.
        required: true
        choices: ['ssh_config_inventory']
      config_file:
        description: Path to SSH config file
        type: path
        default: ~/.ssh/config
'''

class InventoryModule(BaseInventoryPlugin):

    NAME = 'ssh_config_inventory'

    def verify_file(self, path):
        """Always verify inventory source file."""
        return True

    def parse(self, inventory, loader, path, cache=True):
        super().parse(inventory, loader, path, cache)

        config_file = self.get_option('config_file') or os.path.expanduser('~/.ssh/config')
        if config_file.startswith('~'):
            config_file = os.path.expanduser(config_file)

        if not os.path.exists(config_file):
            raise AnsibleParserError(f"SSH config file not found: {config_file}")

        host_aliases = self.parse_ssh_config_for_hosts(config_file)

        for alias in host_aliases:
            # Ignore wildcards like Host *
            if '*' in alias or '?' in alias:
                continue

            host_config = self.lookup_ssh_config(alias)

            hostname = host_config.get('hostname', alias)
            user = host_config.get('user', None)
            port = host_config.get('port', None)
            proxyjump = host_config.get('proxyjump', None)

            self.inventory.add_host(alias)
            self.inventory.set_variable(alias, 'ansible_host', hostname)
            if user:
                self.inventory.set_variable(alias, 'ansible_user', user)
            if port:
                self.inventory.set_variable(alias, 'ansible_port', int(port))
            if proxyjump:
                self.inventory.set_variable(alias, 'ansible_ssh_common_args', f"-o ProxyJump={proxyjump}")

    def parse_ssh_config_for_hosts(self, path):
        """Parse SSH config and return list of Host aliases."""
        hosts = []
        with open(path) as f:
            for line in f:
                stripped = line.strip()
                if stripped.lower().startswith('host '):
                    parts = stripped.split()
                    if len(parts) > 1:
                        hosts.extend(parts[1:])
        return hosts

    def lookup_ssh_config(self, hostname):
        """Use `ssh -G` to resolve final SSH config for a given host."""
        try:
            result = subprocess.run(
                ["ssh", "-G", hostname],
                check=True,
                capture_output=True,
                text=True
            )
        except subprocess.CalledProcessError as e:
            raise AnsibleParserError(f"Failed running 'ssh -G {hostname}': {e}")

        config = {}
        for line in result.stdout.splitlines():
            if not line.strip():
                continue
            key, value = line.strip().split(None, 1)
            config[key.lower()] = value
        return config

ssh_config_inventory.yml

---
plugin: ssh_config_inventory