# SPDX-License-Identifier: MIT-0 """ garage_credentials lookup plugin Lookup S3 credentials from a Garage instance running in a Docker container. Examples: # Lookup credentials for a specific key - debug: msg: "{{ lookup('digitalboard.core.garage_credentials', 'nextcloud', host='backend') }}" # Use in vars nextcloud_s3_key: "{{ lookup('digitalboard.core.garage_credentials', 'nextcloud', host='backend')['key_id'] }}" nextcloud_s3_secret: "{{ lookup('digitalboard.core.garage_credentials', 'nextcloud', host='backend')['secret_key'] }}" """ from __future__ import (absolute_import, division, print_function) __metaclass__ = type DOCUMENTATION = r""" name: garage_credentials author: Digital Board Team version_added: "1.0.0" short_description: Lookup S3 credentials from Garage description: - This lookup returns S3 credentials (key_id and secret_key) from a Garage instance. - It executes a docker command on the specified host to retrieve the credentials. options: _terms: description: - The name of the S3 key to lookup required: True host: description: - The inventory hostname where Garage is running type: string required: True container: description: - The name of the Garage container type: string default: garage """ EXAMPLES = r""" - name: Get credentials for nextcloud key debug: msg: "{{ lookup('digitalboard.core.garage_credentials', 'nextcloud', host='backend') }}" - name: Use credentials in configuration set_fact: s3_key_id: "{{ lookup('digitalboard.core.garage_credentials', 'nextcloud', host='backend')['key_id'] }}" s3_secret: "{{ lookup('digitalboard.core.garage_credentials', 'nextcloud', host='backend')['secret_key'] }}" """ RETURN = r""" _value: description: - A dictionary containing key_id and secret_key type: dict elements: dict contains: key_id: description: The S3 access key ID type: string secret_key: description: The S3 secret key type: string """ import re from ansible.errors import AnsibleError, AnsibleParserError from ansible.plugins.lookup import LookupBase from ansible.module_utils.common.text.converters import to_native from ansible.utils.display import Display display = Display() class LookupModule(LookupBase): def run(self, terms, variables=None, **kwargs): """ Lookup S3 credentials from Garage container. Args: terms: List containing the key name as first element variables: Ansible variables dict **kwargs: Additional options (host, container) Returns: List containing a single dict with key_id and secret_key """ if not terms: raise AnsibleError('garage_credentials lookup requires a key name') key_name = terms[0] # Get options host = kwargs.get('host') if not host: raise AnsibleError('garage_credentials lookup requires "host" parameter') container = kwargs.get('container', 'garage') use_cache = kwargs.get('use_cache', True) if not variables: raise AnsibleError('No variables available') display.vvv(f"garage_credentials: Looking up credentials for key '{key_name}' on host '{host}'") # First, try to get from cache if enabled if use_cache: garage_credentials = variables.get('hostvars', {}).get(host, {}).get('garage_s3_credentials', {}) if garage_credentials and key_name in garage_credentials: display.vvv(f"garage_credentials: Found cached credentials for '{key_name}'") return [garage_credentials[key_name]] display.vvv(f"garage_credentials: No cached credentials found, will query Garage directly") # Get connection to the target host try: # Get task and connection from the play context from ansible.plugins.loader import connection_loader from ansible.parsing.dataloader import DataLoader from ansible.vars.manager import VariableManager from ansible.inventory.manager import InventoryManager # We need to create a connection to the target host # Get the play context variables for the target host hostvars = variables.get('hostvars', {}).get(host, {}) # Get ansible connection info ansible_connection = hostvars.get('ansible_connection', 'ssh') ansible_host = hostvars.get('ansible_host', host) ansible_user = hostvars.get('ansible_user', hostvars.get('ansible_ssh_user')) ansible_port = hostvars.get('ansible_port', hostvars.get('ansible_ssh_port')) docker_cmd = f"docker exec {container} /garage key info {key_name} --show-secret" # Create a simple SSH command if using SSH connection if ansible_connection in ['ssh', 'smart']: import subprocess # Build SSH command ssh_cmd = ['ssh'] if ansible_port: ssh_cmd.extend(['-p', str(ansible_port)]) if ansible_user: ssh_cmd.append(f'{ansible_user}@{ansible_host}') else: ssh_cmd.append(ansible_host) ssh_cmd.append(docker_cmd) display.vvv(f"garage_credentials: Executing: {' '.join(ssh_cmd)}") try: result = subprocess.run( ssh_cmd, capture_output=True, text=True, check=True, timeout=30 ) output = result.stdout except subprocess.CalledProcessError as e: raise AnsibleError( f"Failed to execute docker command on {host}: {e.stderr}" ) except subprocess.TimeoutExpired: raise AnsibleError( f"Timeout while querying Garage on {host}" ) else: raise AnsibleError( f"Unsupported connection type '{ansible_connection}'. " f"Only 'ssh' and 'smart' connections are supported." ) # Parse the output to extract key_id and secret_key key_id_match = re.search(r'Key ID:\s+(\S+)', output) secret_key_match = re.search(r'Secret key:\s+(\S+)', output) if not key_id_match or not secret_key_match: raise AnsibleError( f"Could not parse Garage output for key '{key_name}'. " f"Output was: {output}" ) credentials = { 'key_id': key_id_match.group(1), 'secret_key': secret_key_match.group(1) } display.vvv(f"garage_credentials: Successfully retrieved credentials for '{key_name}'") return [credentials] except ImportError as e: raise AnsibleError(f'Failed to import required modules: {to_native(e)}') except Exception as e: raise AnsibleError(f'Error looking up garage credentials: {to_native(e)}')