Управление секретами через Python SDK

Что такое Python SDK

Python SDK — библиотека для программной работы с Пассворком. Подходит для сложной автоматизации, где возможностей CLI недостаточно.

Возможность	Описание
Чтение секретов	Получение значений поля пароля, полей и вложений по ID или поиску
Изменение секретов	Обновление значения поля пароля, полей, тегов и описаний
Создание секретов	Программное создание записей в нужных папках
Управление структурой	Работа с папками, сейфами, перемещение записей

Когда использовать SDK, а когда CLI

Сценарий	Рекомендация
CI/CD-пайплайн, деплой-скрипт	passwork-cli — проще, не требует кода
Ротация паролей с логикой	Python SDK — гибкость, обработка ошибок
Миграция из другой системы	Python SDK — преобразование данных
Проверка целостности секретов	Python SDK — сложная логика валидации
Получение одного секрета в bash	passwork-cli get — одна команда

Установка

Ставится через pip. PyPI — основной вариант, gitVerse — если нужны исходники или конкретная ветка:

# PyPI
pip install pw-connector

# SSH
pip install git+ssh://git@gitverse.ru:2222/passwork-ru/passwork-python.git

# HTTPS
pip install git+https://gitverse.ru/passwork-ru/passwork-python.git

Подробнее: Python-коннектор.

Инициализация клиента

from passwork_client import PassworkClient

client = PassworkClient(host="https://passwork.example.com")
client.set_tokens("your_access_token", "your_refresh_token")  # refresh_token опционален
client.set_master_key("your_master_key")  # если включено клиентское шифрование

подсказка

Не храните токены в коде. Используйте переменные окружения:

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

Чтение секретов

По ID записи

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

# Стандартные поля
print(item["login"])
print(item["password"])

# Пользовательские поля (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"]

По папке

# Получить все записи из папки
items = client.search_and_decrypt(folder_ids=["<folder-id>"])

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

Поиск по строке

# Найти записи по части названия
items = client.search_and_decrypt(query="app-db prod")

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

Изменение секретов

Обновление существующей записи

# Получить запись
item = client.get_item("<item-id>")

# Подготовить данные для обновления
updated_data = {
    "vaultId": item["vaultId"],
    "password": "new-strong-password",
    "customs": [
        {"name": "API_KEY", "value": "new-api-key", "type": "password"},
        # остальные customs, которые нужно сохранить
        *[c for c in item.get("customs", []) if c["name"] != "API_KEY"],
    ],
}

# Сохранить
client.update_item(item["id"], updated_data)

Создание новой записи

item_id = client.create_item({
    "vaultId": "<vault-id>",
    "folderId": "<folder-id>",  # опционально
    "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"],
})

Массовые операции

SDK позволяет автоматизировать:

• перенос записей между папками;
• массовое добавление/удаление тегов;
• обновление полей по шаблону;
• миграцию структуры secrets/*.

# Пример: добавить тег ко всем записям в папке
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,
        })

Практические сценарии

Ротация пароля БД

Полный пример ротации пароля 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):
    """Ротация пароля для роли PostgreSQL."""

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

    # Генерируем новый пароль
    new_password = secrets.token_urlsafe(32)

    # Получаем текущие credentials для подключения к БД
    admin_item = client.get_item(os.environ["DB_ADMIN_ITEM_ID"])

    try:
        # Обновляем пароль в 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:
            # Используем параметризованный запрос для безопасности
            cur.execute(
                "ALTER ROLE %s WITH PASSWORD %s",
                (db_role, new_password)
            )

        conn.close()

        # Обновляем секрет в Пассворке
        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

# Запуск
rotate_db_password(
    item_id="<item-id>",
    db_role="app_user"
)

Проверка целостности секретов

Скрипт для периодической проверки, что секреты актуальны:

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):
    """Проверка, что credentials в Пассворке рабочие."""

    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)

            # Помечаем запись тегом
            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

# Запуск
broken = check_db_credentials(folder_id="<folder-id>")
if broken:
    print(f"\n{len(broken)} secrets need review")

Миграция из файлов

Импорт секретов из .env файлов в Пассворк:

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):
    """Импорт секретов из .env файла в Пассворк."""

    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 = {}

    # Парсим .env файл
    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('"\'')

    # Создаём запись в Пассворке
    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}")

# Пример использования
migrate_env_file(
    env_path="./legacy/.env.production",
    vault_id="<vault-id>",
    folder_id="<folder-id>",
    tags=["prod", "migrated"],
)

Версия для LLM