# Copyright (c) 2014, Mathieu GAUTHIER-LAFAYE # Copyright (c) 2016, Matt Harris # Copyright (c) 2020, Robert Kaussow # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) """Dynamic inventory plugin for Proxmox VE.""" from __future__ import absolute_import, division, print_function __metaclass__ = type DOCUMENTATION = """ --- name: proxmox short_description: Proxmox VE inventory source version_added: 1.1.0 description: - Get inventory hosts from the proxmox service. - "Uses a configuration file as an inventory source, it must end in C(.proxmox.yml) or C(.proxmox.yaml) and has a C(plugin: xoxys.general.proxmox) entry." extends_documentation_fragment: - inventory_cache options: plugin: description: The name of this plugin, it should always be set to C(xoxys.general.proxmox) for this plugin to recognize it as it's own. required: yes choices: ["xoxys.general.proxmox"] api_host: description: - Specify the target host of the Proxmox VE cluster. type: str required: true env: - name: PROXMOX_SERVER api_user: description: - Specify the user to authenticate with. type: str required: true env: - name: PROXMOX_USER api_password: description: - Specify the password to authenticate with. type: str env: - name: PROXMOX_PASSWORD api_token_id: description: - Specify the token ID. type: str env: - name: PROXMOX_TOKEN_ID api_token_secret: description: - Specify the token secret. type: str env: - name: PROXMOX_TOKEN_SECRET verify_ssl: description: - If C(false), SSL certificates will not be validated. - This should only be used on personally controlled sites using self-signed certificates. type: bool default: True auth_timeout: description: Proxmox VE authentication timeout. type: int default: 5 exclude_vmid: description: VMID's to exclude from inventory. type: list default: [] elements: str exclude_state: description: VM states to exclude from inventory. type: list default: [] elements: str group: description: Group to place all hosts into. type: string default: proxmox want_facts: description: Toggle, if C(true) the plugin will retrieve host facts from the server type: boolean default: True requirements: - "proxmoxer" """ # noqa EXAMPLES = """ # proxmox.yml plugin: xoxys.general.proxmox api_user: root@pam api_password: secret api_host: helldorado """ import json import re import socket from collections import defaultdict from ansible.errors import AnsibleError from ansible.module_utils._text import to_native from ansible.module_utils.parsing.convert_bool import boolean from ansible.module_utils.six import iteritems from ansible.plugins.inventory import BaseInventoryPlugin from ansible_collections.xoxys.general.plugins.module_utils.version import LooseVersion try: from proxmoxer import ProxmoxAPI HAS_PROXMOXER = True except ImportError: HAS_PROXMOXER = False try: from requests.packages import urllib3 HAS_URLLIB3 = True except ImportError: try: import urllib3 HAS_URLLIB3 = True except ImportError: HAS_URLLIB3 = False class InventoryModule(BaseInventoryPlugin): """Provide Proxmox VE inventory.""" NAME = "xoxys.general.proxmox" def _proxmox_auth(self): auth_args = {"user": self.get_option("api_user")} if not (self.get_option("api_token_id") and self.get_option("api_token_secret")): auth_args["password"] = self.get_option("api_password") else: auth_args["token_name"] = self.get_option("api_token_id") auth_args["token_value"] = self.get_option("api_token_secret") verify_ssl = boolean(self.get_option("verify_ssl"), strict=False) if not verify_ssl and HAS_URLLIB3: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.client = ProxmoxAPI( self.get_option("api_host"), verify_ssl=verify_ssl, timeout=self.get_option("auth_timeout"), **auth_args, ) def _get_version(self): return LooseVersion(self.client.version.get()["version"]) def _get_major(self): return LooseVersion(self.client.version.get()["release"]) def _get_names(self, pve_list, pve_type): if pve_type == "node": return [node["node"] for node in pve_list] if pve_type == "pool": return [pool["poolid"] for pool in pve_list] return [] def _get_variables(self, pve_list, pve_type): variables = {} if pve_type in ["qemu", "container"]: for vm in pve_list: nested = {} for key, value in iteritems(vm): nested["proxmox_" + key] = value variables[vm["name"]] = nested return variables def _get_ip_address(self, pve_type, pve_node, vmid): def validate(address): try: # IP address validation if socket.inet_aton(address) and address != "127.0.0.1": return address except OSError: return False address = False networks = False if pve_type == "qemu": # If qemu agent is enabled, try to gather the IP address try: if self.client.nodes(pve_node).get(pve_type, vmid, "agent", "info") is not None: networks = self.client.nodes(pve_node).get( "qemu", vmid, "agent", "network-get-interfaces" )["result"] except Exception: # noqa pass if networks and isinstance(networks, list): for network in networks: for ip_address in network.get("ip-addresses", []): address = validate(ip_address["ip-address"]) else: try: config = self.client.nodes(pve_node).get(pve_type, vmid, "config") address = re.search(r"ip=(\d*\.\d*\.\d*\.\d*)", config["net0"]).group(1) except Exception: # noqa pass return address def _exclude(self, pve_list): filtered = [] for item in pve_list: obj = defaultdict(dict, item) if obj["template"] == 1: continue if obj["status"] in self.get_option("exclude_state"): continue if obj["vmid"] in self.get_option("exclude_vmid"): continue filtered.append(item.copy()) return filtered def _propagate(self): for node in self._get_names(self.client.nodes.get(), "node"): try: qemu_list = self._exclude(self.client.nodes(node).qemu.get()) container_list = self._exclude(self.client.nodes(node).lxc.get()) except Exception as e: # noqa raise AnsibleError(f"Proxmoxer API error: {to_native(e)}") from e # Merge QEMU and Containers lists from this node instances = self._get_variables(qemu_list, "qemu").copy() instances.update(self._get_variables(container_list, "container")) for host in instances: vmid = instances[host]["proxmox_vmid"] try: pve_type = instances[host]["proxmox_type"] except KeyError: pve_type = "qemu" try: description = self.client.nodes(node).get(pve_type, vmid, "config")[ "description" ] except KeyError: description = None except Exception as e: # noqa raise AnsibleError(f"Proxmoxer API error: {to_native(e)}") from e try: metadata = json.loads(description) except TypeError: metadata = {} except ValueError: metadata = {"notes": description} # Add hosts to default group self.inventory.add_group(group=self.get_option("group")) self.inventory.add_host(group=self.get_option("group"), host=host) # Group hosts by status self.inventory.add_group(group=instances[host]["proxmox_status"]) self.inventory.add_host(group=instances[host]["proxmox_status"], host=host) if "groups" in metadata: for group in metadata["groups"]: self.inventory.add_group(group=group) self.inventory.add_host(group=group, host=host) if self.get_option("want_facts"): for attr in instances[host]: if attr not in ["proxmox_template"]: self.inventory.set_variable(host, attr, instances[host][attr]) address = self._get_ip_address(pve_type, node, vmid) if address: self.inventory.set_variable(host, "ansible_host", address) for pool in self._get_names(self.client.pools.get(), "pool"): try: pool_list = self._exclude(self.client.pool(pool).get()["members"]) except Exception as e: # noqa raise AnsibleError(f"Proxmoxer API error: {to_native(e)}") from e members = [ member["name"] for member in pool_list if (member["type"] == "qemu" or member["type"] == "lxc") ] for member in members: self.inventory.add_host(group=pool, host=member) def verify_file(self, path): """Verify the Proxmox VE configuration file.""" if super().verify_file(path): endings = ("proxmox.yaml", "proxmox.yml") if any(path.endswith(ending) for ending in endings): return True return False def parse(self, inventory, loader, path, cache=True): # noqa """Dynamically parse the Proxmox VE cloud inventory.""" if not HAS_PROXMOXER: raise AnsibleError( "The Proxmox VE dynamic inventory plugin requires proxmoxer: " "https://pypi.org/project/proxmoxer/" ) super().parse(inventory, loader, path) self._read_config_data(path) self._proxmox_auth() self._propagate()