infisical.vault/plugins/lookup/read_secrets.py

from ansible.errors import AnsibleError
from ansible.plugins.lookup import LookupBase

HAS_INFISICAL = False
INFISICAL_VERSION = None


try:
    from infisical_sdk import InfisicalSDKClient
    HAS_INFISICAL = True

except ImportError as e:
    HAS_INFISICAL = False


if HAS_INFISICAL:
    try:
        from importlib.metadata import version
        INFISICAL_VERSION = version('infisicalsdk')  # Note: package name might differ
    except ImportError:
        # Fallback for Python < 3.8
        import pkg_resources
        INFISICAL_VERSION = pkg_resources.get_distribution('infisicalsdk').version
    except Exception:
        INFISICAL_VERSION = "unknown"


DOCUMENTATION = r"""
name: read_secrets
author:
  - Infisical Inc.

short_description: Look up secrets stored in Infisical
description:
  - Retrieve secrets from Infisical, granted the caller has the right permissions to access the secret.
  - Secrets can be located either by their name for individual secret loopups or by environment/folder path to return all secrets within the given scope.

options:

  auth_method:
    description: The method to use to authenticate with Infisical
    required: False
    type: string
    version_added: 1.1.3
    default: universal_auth
    choices:
      - universal_auth
      - oidc_auth
    env:
      - name: INFISICAL_AUTH_METHOD
  universal_auth_client_id:
    description: The Machine Identity Client ID used to authenticate
    env:
      - name: UNIVERSAL_AUTH_MACHINE_IDENTITY_CLIENT_ID
      - name: INFISICAL_UNIVERSAL_AUTH_CLIENT_ID
    required: False
    type: string
    version_added: 1.0.0
  universal_auth_client_secret:
    description: The Machine Identity Client Secret used to authenticate
    env:
      - name: UNIVERSAL_AUTH_MACHINE_IDENTITY_CLIENT_SECRET
      - name: INFISICAL_UNIVERSAL_AUTH_CLIENT_SECRET
    required: False
    type: string
    version_added: 1.0.0
  url:
    description: Point to your self hosted instance of Infisical
    default: "https://app.infisical.com"
    env:
      - name: INFISICAL_URL
    required: False
    type: string
    version_added: 1.0.0
  path:
    description: "The folder path where the requested secret resides. For example: /services/backend"
    required: True
    type: string
    version_added: 1.0.0
  env_slug:
    description: "Used to select from which environment (environment slug) secrets should be fetched from. Environment slug is the short name of a given environment"
    required: True
    type: string
    version_added: 1.0.0
  project_id:
    description: "The ID of the project where the secrets are stored"
    required: True
    type: string
    version_added: 1.0.0
  secret_name:
    description: The name of the secret that should be fetched. The name should be exactly as it appears in Infisical.
    required: False
    type: string
    version_added: 1.0.0
  as_dict:
    description: "Return the listed secrets as a dictionary within a list instead of a list of key-value pairs (defaults to False). When True, returns [{'SECRET_KEY': 'secret_value', ...}] instead of [{'key': 'SECRET_KEY', 'value': 'secret_value'}, ...]. This only applies when reading all secrets within a scope, not when reading a single secret by name."
    required: False
    type: bool
    version_added: 1.0.0
  identity_id:
    description: The identity ID of the user that should be authenticated
    env:
      - name: INFISICAL_MACHINE_IDENTITY_ID
    required: False
    type: string
    version_added: 1.1.3
  jwt:
    description: The JWT of the user that should be authenticated
    required: False
    type: string
    version_added: 1.1.3
    env:
      - name: INFISICAL_JWT
      - name: INFISICAL_OIDC_AUTH_JWT
"""

EXAMPLES = r"""
vars:
  read_all_secrets_within_scope: "{{ lookup('infisical_vault', universal_auth_client_id='<>', universal_auth_client_secret='<>', project_id='<>', path='/', env_slug='dev', url='https://spotify.infisical.com') }}"
  # [{ "key": "HOST", "value": "google.com" }, { "key": "SMTP", "value": "gmail.smtp.edu" }]

  read_all_secrets_as_dict: "{{ lookup('infisical_vault', universal_auth_client_id='<>', universal_auth_client_secret='<>', project_id='<>', path='/', env_slug='dev', as_dict=True, url='https://spotify.infisical.com') }}"
  # {"HOST": "google.com", "SMTP": "gmail.smtp.edu"}

  read_secret_by_name_within_scope: "{{ lookup('infisical_vault', universal_auth_client_id='<>', universal_auth_client_secret='<>', project_id='<>', path='/', env_slug='dev', secret_name='HOST', url='https://spotify.infisical.com') }}"
  # [{ "key": "HOST", "value": "google.com" }]
"""



def parse_version_tuple(version_string):
    if version_string == "unknown":
        return (0, 0, 0)  # assume very old version
    
    try:
        parts = version_string.split('.')
        # haandle missing parts (example: "1.2" becomes "1.2.0")
        while len(parts) < 3:
            parts.append('0')
        
        return tuple(int(part) for part in parts[:3])  # only take first 3 parts
    except (ValueError, AttributeError):
        return (0, 0, 0)


def check_minimum_version(current_version, minimum_version):
    """Check if current version meets minimum requirement."""
    current_tuple = parse_version_tuple(current_version)
    minimum_tuple = parse_version_tuple(minimum_version)
    return current_tuple >= minimum_tuple


class LookupModule(LookupBase):

    def get_sdk_client(self):
      url = self.get_option("url")
      client = InfisicalSDKClient(host=url)

      method = self.get_option("auth_method")

      if method == "universal_auth":

        machine_identity_client_id = self.get_option("universal_auth_client_id")
        machine_identity_client_secret = self.get_option("universal_auth_client_secret")

        if not machine_identity_client_id or not machine_identity_client_secret:
            raise AnsibleError("universal_auth_client_id or universal_auth_client_secret is not set. Please set them to use universal auth.")

        client.auth.universal_auth.login(
            machine_identity_client_id,
            machine_identity_client_secret
        )

      elif method == "oidc_auth":

        # make sure the infisicalsdk version is at least 1.0.10
        if not check_minimum_version(INFISICAL_VERSION, "1.0.10"):
            raise AnsibleError("Please upgrade the infisicalsdk to at least 1.0.10 to use oidc auth.")

        identity_id = self.get_option("identity_id")
        jwt = self.get_option("jwt")

        if not identity_id or not jwt:
            raise AnsibleError("identity_id or jwt is not set. Please set them to use oidc auth.")

        client.auth.oidc_auth.login(
            identity_id,
            jwt
        )
      else:
        raise AnsibleError(f"Invalid auth method. Please use universal_auth or oidc_auth. You provided {method}")

      return client



    def run(self, terms, variables=None, **kwargs):

        self.set_options(var_options=variables, direct=kwargs)
        if not HAS_INFISICAL:
            raise AnsibleError("Please pip install infisicalsdk to use the infisical_vault lookup module.")

        client = self.get_sdk_client()

        secretName = kwargs.get('secret_name')
        asDict = kwargs.get('as_dict')
        envSlug = kwargs.get('env_slug')
        path = kwargs.get('path')
        project_id = kwargs.get('project_id')

        if secretName:
            return self.get_single_secret(
                client,
                project_id,
                secretName,
                envSlug,
                path,
            )
        else:
            return self.get_all_secrets(client, project_id, envSlug, path, asDict)

    def get_single_secret(
            self,
            client,
            project_id,
            secret_name,
            environment,
            path
    ):
        try:
            secret = client.secrets.get_secret_by_name(
                secret_name=secret_name,
                project_id=project_id,
                environment_slug=environment,
                secret_path=path
            )

            return [{"value": secret.secretValue, "key": secret.secretKey}]
        except Exception as e:
            raise AnsibleError(f"Error fetching single secret {e}")

    def get_all_secrets(self, client, project_id, environment="dev", path="/", asDict=False):
        try:
            secrets = client.secrets.list_secrets(
                project_id=project_id,
                environment_slug=environment,
                secret_path=path
            )

            if asDict:
                return [{s.secretKey: s.secretValue for s in secrets.secrets}]
            else:
                return [{"value": s.secretValue, "key": s.secretKey} for s in secrets.secrets]
        except Exception as e:
            raise AnsibleError(f"Error fetching all secrets {e}")