ansible-collections/infisical.vault
1
0
Fork 0
mirror of https://github.com/Infisical/ansible-collection.git synced 2025-08-09 15:44:25 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: use infisicalsdk

Browse source
This commit is contained in:
Alex Fomenko 2025-01-15 12:50:14 +03:00
commit aa6de06525
1 changed files with 15 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

42
plugins/lookup/read_secrets.py
View file

@ -3,7 +3,7 @@ from ansible.plugins.lookup import LookupBase
HAS_INFISICAL = False
try:
from infisical_client import InfisicalClient, ClientSettings, GetSecretOptions, ListSecretsOptions
from infisical_sdk import InfisicalSDKClient
HAS_INFISICAL = True
except ImportError as e:
HAS_INFISICAL = False
@ -77,7 +77,7 @@ class LookupModule(LookupBase):
self.set_options(var_options=variables, direct=kwargs)
if not HAS_INFISICAL:
raise AnsibleError("Please pip install infisical-python to use the infisical_vault lookup module.")
raise AnsibleError("Please pip install infisicalsdk to use the infisical_vault lookup module.")
machine_identity_client_id = self.get_option("universal_auth_client_id")
machine_identity_client_secret = self.get_option("universal_auth_client_secret")
@ -87,18 +87,10 @@ class LookupModule(LookupBase):
if not machine_identity_client_id or not machine_identity_client_secret:
raise AnsibleError("Please provide the universal_auth_client_id and universal_auth_client_secret")
# Create the client settings
settings = ClientSettings(
client_id=machine_identity_client_id,
client_secret=machine_identity_client_secret,
site_url=url
)
# Initialize the Infisical client
client = InfisicalClient(settings=settings)
client = InfisicalSDKClient(host=url)
client.auth.universal_auth.login(client_id=machine_identity_client_id,
client_secret=machine_identity_client_secret)
secretName = kwargs.get('secret_name')
envSlug = kwargs.get('env_slug')
@ -112,31 +104,27 @@ class LookupModule(LookupBase):
def get_single_secret(self, client, project_id, secret_name, environment, path):
try:
options = GetSecretOptions(
environment=environment,
project_id=project_id,
secret = client.secrets.get_secret_by_name(
secret_name=secret_name,
path=path,
type="shared"
project_id=project_id,
environment_slug=environment,
secret_path=path,
)
secret = client.getSecret(options=options)
return [{"value": secret.secret_value, "key": secret.secret_key}]
return [[{"value": secret.secretValue, "key": secret.secretKey}]]
except Exception as e:
print(e)
raise AnsibleError(f"Error fetching single secret {e}")
def get_all_secrets(self, client, project_id, environment="dev", path="/"):
try:
options = ListSecretsOptions(
environment=environment,
secrets = client.secrets.list_secrets(
project_id=project_id,
path=path,
environment_slug=environment,
secret_path=path,
tag_filters=tags
)
secrets = client.listSecrets(options=options)
return [{"value": s.secret_value, "key": s.secret_key} for s in secrets]
return [[{"value": s.secretValue, "key": s.secretKey} for s in secrets.secrets]]
except Exception as e:
raise AnsibleError(f"Error fetching all secrets {e}")