Geheimnisse über das Python SDK verwalten

Was ist das Python SDK

Das Python SDK ist eine Bibliothek für die programmatische Interaktion mit Passwork. Es ist für erweiterte Automatisierung konzipiert, bei der die CLI-Funktionen nicht ausreichen.

Funktion	Beschreibung
Geheimnisse lesen	Passwortfeldwerte, benutzerdefinierte Felder und Anhänge per ID oder Suche abrufen
Geheimnisse ändern	Passwortfeldwerte, benutzerdefinierte Felder, Tags und Beschreibungen aktualisieren
Geheimnisse erstellen	Einträge programmatisch in bestimmten Ordnern erstellen
Struktur verwalten	Mit Ordnern und Tresoren arbeiten, Einträge verschieben

Wann SDK vs. CLI verwenden

Szenario	Empfehlung
CI/CD-Pipeline, Deploy-Skript	passwork-cli — einfacher, kein Code erforderlich
Passwortrotation mit benutzerdefinierter Logik	Python SDK — mehr Flexibilität, bessere Fehlerbehandlung
Migration von einem anderen System	Python SDK — Daten nach Bedarf transformieren
Prüfung der Geheimnis-Integrität	Python SDK — komplexe Validierung implementieren
Einzelnes Geheimnis in Bash abrufen	passwork-cli get — nur ein Befehl

Installation

Installation mit pip von PyPI oder 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

Siehe auch: Python-Connector.

Client initialisieren

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

tipp

Vermeiden Sie die Hardcodierung von Token in Ihrem Code. Verwenden Sie stattdessen Umgebungsvariablen:

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"])

Geheimnisse lesen

Per Eintrags-ID

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"]

Per Ordner

# 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']}")

Per Suchbegriff

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

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

Geheimnisse ändern

Bestehenden Eintrag aktualisieren

# 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)

Neuen Eintrag erstellen

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"],
})

Massenoperationen

Das SDK ermöglicht Automatisierung wie:

• Einträge zwischen Ordnern verschieben;
• Tags in großen Mengen hinzufügen oder entfernen;
• Felder basierend auf einer Vorlage aktualisieren;
• die secrets/*-Hierarchie migrieren.

# 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,
        })

Praktische Beispiele

Rotation eines Datenbankpassworts

Vollständiges Beispiel mit Fehlerbehandlung für PostgreSQL-Passwortrotation:

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"
)

Prüfung der Geheimnisgültigkeit

Skript, das gespeicherte Anmeldedaten regelmäßig auf Funktionsfähigkeit überprüft:

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")

Geheimnisse aus Dateien importieren

Geheimnisse aus .env-Dateien in Passwork migrieren:

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"],
)