Gestión de secretos mediante Python SDK

Qué es el Python SDK

El Python SDK es una biblioteca para la interacción programática con Passwork. Está diseñado para automatización avanzada donde las capacidades de CLI no son suficientes.

Funcionalidad	Descripción
Lectura de secretos	Obtener valores de campos de contraseña, campos personalizados y archivos adjuntos por ID o búsqueda
Modificación de secretos	Actualizar valores de campos de contraseña, campos personalizados, etiquetas y descripciones
Creación de secretos	Añadir registros programáticamente a carpetas específicas
Gestión de estructura	Trabajar con carpetas y bóvedas, mover registros

Cuándo usar SDK vs CLI

Escenario	Recomendación
Pipeline de CI/CD, script de despliegue	passwork-cli — más simple, no requiere código
Rotación de contraseñas con lógica personalizada	Python SDK — más flexibilidad, mejor manejo de errores
Migración desde otro sistema	Python SDK — transformar datos según sea necesario
Verificación de integridad de secretos	Python SDK — implementar validación compleja
Obtener un solo secreto en bash	passwork-cli get — solo un comando

Instalación

Instale con pip desde PyPI o GitHub:

# PyPI
pip install passwork-python

# Or from GitHub via SSH
pip install git+ssh://[email protected]:passwork-me/passwork-python.git

# Or from GitHub via HTTPS
pip install git+https://github.com/passwork-me/passwork-python.git

Consulte también: Conector Python.

Inicialización del cliente

from passwork_client import PassworkClient

client = PassworkClient(host="https://passwork.example.com")
client.set_tokens("your_access_token", "your_refresh_token")  # refresh_token is optional
client.set_master_key("your_master_key")  # if client-side encryption is enabled

tip

Evite codificar tokens en su código. Utilice variables de entorno en su lugar:

import os
from passwork_client import PassworkClient

client = PassworkClient(host=os.environ["PASSWORK_HOST"])
client.set_tokens(
    os.environ["PASSWORK_ACCESS_TOKEN"],
    os.environ.get("PASSWORK_REFRESH_TOKEN"),
)
if os.environ.get("PASSWORK_MASTER_KEY"):
    client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])

Lectura de secretos

Por ID de registro

item = client.get_item("<item-id>")

# Standard fields
print(item["login"])
print(item["password"])

# Custom fields (customs)
for custom in item.get("customs", []):
    if custom["name"] == "DB_HOST":
        db_host = custom["value"]
    if custom["name"] == "API_KEY":
        api_key = custom["value"]

Por carpeta

# Fetch all records from a folder
items = client.search_and_decrypt(folder_ids=["<folder-id>"])

for item in items:
    print(f"{item['name']}: {item['login']}")

Por cadena de búsqueda

# Find records matching a query
items = client.search_and_decrypt(query="app-db prod")

for item in items:
    print(f"{item['id']}: {item['name']}")

Modificación de secretos

Actualización de un registro existente

# Load the record
item = client.get_item("<item-id>")

# Prepare update payload
updated_data = {
    "vaultId": item["vaultId"],
    "password": "new-strong-password",
    "customs": [
        {"name": "API_KEY", "value": "new-api-key", "type": "password"},
        # keep remaining customs that should be preserved
        *[c for c in item.get("customs", []) if c["name"] != "API_KEY"],
    ],
}

# Persist changes
client.update_item(item["id"], updated_data)

Creación de un nuevo registro

item_id = client.create_item({
    "vaultId": "<vault-id>",
    "folderId": "<folder-id>",  # optional
    "name": "Production DB",
    "login": "app_user",
    "password": "secure-password",
    "customs": [
        {"name": "DB_HOST", "value": "db.example.com", "type": "text"},
        {"name": "DB_PORT", "value": "5432", "type": "text"},
    ],
    "tags": ["prod", "database"],
})

Operaciones masivas

El SDK permite automatización como:

• mover registros entre carpetas;
• añadir o eliminar etiquetas de forma masiva;
• actualizar campos basándose en una plantilla;
• migrar la jerarquía secrets/*.

# Add a tag to all records in a folder
items = client.search_and_decrypt(folder_ids=["<folder-id>"])

for item in items:
    tags = list(item.get("tags") or [])
    if "legacy" not in tags:
        tags.append("legacy")
        client.update_item(item["id"], {
            "vaultId": item["vaultId"],
            "tags": tags,
        })

Ejemplos prácticos

Rotación de una contraseña de base de datos

Ejemplo completo con manejo de errores para la rotación de contraseñas de PostgreSQL:

import os
import secrets
import psycopg2
from passwork_client import PassworkClient

def get_custom(item: dict, name: str, default=None):
    for custom in item.get("customs", []):
        if custom["name"] == name:
            return custom["value"]
    return default

def rotate_db_password(item_id: str, db_role: str):
    """Rotate the password for a PostgreSQL role."""

    client = PassworkClient(host=os.environ["PASSWORK_HOST"])
    client.set_tokens(
        os.environ["PASSWORK_ACCESS_TOKEN"],
        os.environ.get("PASSWORK_REFRESH_TOKEN"),
    )
    if os.environ.get("PASSWORK_MASTER_KEY"):
        client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])

    # Create a fresh password
    new_password = secrets.token_urlsafe(32)

    # Retrieve admin credentials for the DB connection
    admin_item = client.get_item(os.environ["DB_ADMIN_ITEM_ID"])

    try:
        # Apply the new password in PostgreSQL
        conn = psycopg2.connect(
            host=get_custom(admin_item, "DB_HOST"),
            dbname="postgres",
            user=admin_item["login"],
            password=admin_item["password"],
        )
        conn.autocommit = True

        with conn.cursor() as cur:
            # Use a parameterized query for safety
            cur.execute(
                "ALTER ROLE %s WITH PASSWORD %s",
                (db_role, new_password)
            )

        conn.close()

        # Store the updated password in Passwork
        item = client.get_item(item_id)
        client.update_item(item_id, {
            "vaultId": item["vaultId"],
            "password": new_password,
        })

        print(f"Password rotated for {db_role}")
        return True

    except Exception as e:
        print(f"Rotation failed: {e}")
        return False

# Usage
rotate_db_password(
    item_id="<item-id>",
    db_role="app_user"
)

Verificación de validez de secretos

Script que verifica periódicamente que las credenciales almacenadas siguen funcionando:

import os
import psycopg2
from passwork_client import PassworkClient

def get_custom(item: dict, name: str, default=None):
    for custom in item.get("customs", []):
        if custom["name"] == name:
            return custom["value"]
    return default

def check_db_credentials(folder_id: str):
    """Verify that database credentials in Passwork are valid."""

    client = PassworkClient(host=os.environ["PASSWORK_HOST"])
    client.set_tokens(
        os.environ["PASSWORK_ACCESS_TOKEN"],
        os.environ.get("PASSWORK_REFRESH_TOKEN"),
    )
    if os.environ.get("PASSWORK_MASTER_KEY"):
        client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])

    items = client.search_and_decrypt(folder_ids=[folder_id])
    broken = []

    for item in items:
        if "database" not in (item.get("tags") or []):
            continue

        try:
            conn = psycopg2.connect(
                host=get_custom(item, "DB_HOST"),
                dbname=get_custom(item, "DB_NAME", "postgres"),
                user=item["login"],
                password=item["password"],
                connect_timeout=5,
            )
            conn.close()
            print(f"✅ {item['name']}")

        except Exception as e:
            print(f"❌ {item['name']}: {e}")
            broken.append(item)

            # Flag the record for review
            tags = list(item.get("tags") or [])
            if "needs-review" not in tags:
                tags.append("needs-review")
                client.update_item(item["id"], {
                    "vaultId": item["vaultId"],
                    "tags": tags,
                })

    return broken

# Usage
broken = check_db_credentials(folder_id="<folder-id>")
if broken:
    print(f"\n{len(broken)} secrets need review")

Importación de secretos desde archivos

Migre secretos desde archivos .env a Passwork:

import os
from pathlib import Path
from passwork_client import PassworkClient

def migrate_env_file(env_path: str, vault_id: str, folder_id: str | None, tags: list):
    """Import secrets from a .env file into Passwork."""

    client = PassworkClient(host=os.environ["PASSWORK_HOST"])
    client.set_tokens(
        os.environ["PASSWORK_ACCESS_TOKEN"],
        os.environ.get("PASSWORK_REFRESH_TOKEN"),
    )
    if os.environ.get("PASSWORK_MASTER_KEY"):
        client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])

    env_file = Path(env_path)
    secrets_dict = {}

    # Parse the .env file
    for line in env_file.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" in line:
            key, value = line.split("=", 1)
            secrets_dict[key.strip()] = value.strip().strip('"\'')

    # Store in Passwork
    item_data = {
        "vaultId": vault_id,
        "name": env_file.stem,
        "customs": [
            {"name": key, "value": value, "type": "text"}
            for key, value in secrets_dict.items()
        ],
        "tags": tags,
    }
    if folder_id:
        item_data["folderId"] = folder_id

    client.create_item(item_data)

    print(f"Imported {len(secrets_dict)} secrets from {env_path}")

# Usage
migrate_env_file(
    env_path="./legacy/.env.production",
    vault_id="<vault-id>",
    folder_id="<folder-id>",
    tags=["prod", "migrated"],
)