Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Action: Get vault secrets #1026

Closed
wants to merge 3 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions Vault/legos/vault_get_secrets/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
[<img align="left" src="https://unskript.com/assets/favicon.png" width="100" height="100" style="padding-right: 5px">]
(https://unskript.com/assets/favicon.png)
<h1>Get Vault secrets</h1>

## Description
Fetches secrets from the specified paths in Vault.

## Lego Details
vault_get_secrets fetches secrets from the given path and mount point


## Lego Input
This Lego takes inputs handle, paths, and mount_point

## Lego Output
Here is a sample output.
<img src="./1.png">

## See it in Action

You can see this Lego in action following this link [unSkript Live](https://us.app.unskript.io)
Empty file.
11 changes: 11 additions & 0 deletions Vault/legos/vault_get_secrets/vault_get_secrets.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
{
"action_title": "Get Vault secrets",
"action_description": "Fetches secrets in vault for specific paths",
"action_type": "LEGO_TYPE_VAULT",
"action_entry_function": "vault_get_secrets",
"action_needs_credential": true,
"action_output_type": "ACTION_OUTPUT_TYPE_DICT",
"action_supports_iteration": true,
"action_supports_poll": true,
"action_categories":["CATEGORY_TYPE_INFO","LEGO_TYPE_VAULT"]
}
64 changes: 64 additions & 0 deletions Vault/legos/vault_get_secrets/vault_get_secrets.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
from typing import Tuple, Optional
import hvac
from pydantic import BaseModel, Field

class InputSchema(BaseModel):
paths: list = Field(
title='List of paths',
description='List of paths')
mount_point: str = Field(
title= 'Mount point of secrets',
description= 'Mount point of secrets'
)

def vault_get_secrets_printer(output):
"""
Prints the secrets fetched from Vault in a readable format.

:param secrets: A dictionary with paths as keys and their corresponding secrets.
"""
if output:
print("Fetched secrets from Vault:")
for path, secret in output.items():
print(f"\nPath: {path}")
for key, value in secret.items():
print(f" - {key}: {value}")
else:
print("No secrets fetched or Vault contains no secrets.")

def vault_get_secrets(handle, paths: list, mount_point:str):
"""
Fetches secrets from the specified paths in Vault.

:type handle: hvac.Client
:param handle: Handle containing the Vault instance.

:type paths: Optional[List[str]]
:param paths: Optional list of paths to fetch the secrets from. Fetches all if None.

:rtype: Dict containing the paths as keys and the fetched secrets as values.
"""
secrets = {}

for path in paths:
# Trim any wildcard or file indicators from the path for the API call
clean_path = path.replace(mount_point, "").rstrip("/*")
try:
# Attempt to list secrets in the path if it's a directory
list_response = handle.secrets.kv.v2.list_secrets(mount_point=mount_point, path=clean_path)
if 'keys' in list_response['data']:
for key in list_response['data']['keys']:
secret_path = f"{clean_path}/{key}".rstrip("/")
read_response = handle.secrets.kv.read_secret_version(path=secret_path, mount_point=mount_point)
secrets[f"{mount_point}{secret_path}"] = read_response['data']['data']
except hvac.exceptions.InvalidPath:
# If the path is not a directory, try to read it directly as a secret
try:
read_response = handle.secrets.kv.read_secret_version(path=clean_path, mount_point=mount_point)
secrets[f"{mount_point}{clean_path}"] = read_response['data']['data']
except Exception as e:
print(f"Error fetching secret from {mount_point}{clean_path}: {e}")
except Exception as e:
print(f"Error processing path {mount_point}{clean_path}: {e}")

return secrets
Loading