Перейти к содержанию

Vault Client

Модуль vault предоставляет клиент для безопасной работы с HashiCorp Vault - системой для хранения и управления секретами. Основной класс VaultClient инкапсулирует всю логику взаимодействия с Vault API, включая аутентификацию и чтение секретов.

Особенности

  • Поддержка аутентификации с помощью токена
  • Автоматическая обработка KV Secrets Engine v2
  • Полностью асинхронный API
  • Расширенная обработка ошибок с конкретными типами исключений
  • Логирование всех операций

Документация API

VaultClient

Клиент для работы с HashiCorp Vault - системой для безопасного хранения секретов.

Позволяет получать секреты из Vault с использованием токена аутентификации.

Пример использования

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

vault = VaultClient( addr='http://localhost:8200', # Адрес Vault сервера token_file='/path/to/token' # Файл с токеном доступа )

Получение секрета

secret = vault.read_secret('путь/к/секрету')

Примечание
  • Для работы требуется настроенный и запущенный сервер Vault
  • Токен доступа должен иметь права на чтение запрашиваемых секретов
  • По умолчанию используется KV Secrets Engine версии 2
Source code in src/adapters/vault.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class VaultClient:
    """Клиент для работы с HashiCorp Vault - системой для безопасного хранения секретов.

    Позволяет получать секреты из Vault с использованием токена аутентификации.

    Пример использования:
        # Инициализация клиента
        vault = VaultClient(
            addr='http://localhost:8200',  # Адрес Vault сервера
            token_file='/path/to/token'    # Файл с токеном доступа
        )

        # Получение секрета
        secret = vault.read_secret('путь/к/секрету')

    Примечание:
        - Для работы требуется настроенный и запущенный сервер Vault
        - Токен доступа должен иметь права на чтение запрашиваемых секретов
        - По умолчанию используется KV Secrets Engine версии 2
    """

    def __init__(self, addr: str | None = None, token_file: str | None = None) -> None:
        """Инициализирует клиент Vault с параметрами подключения.

        Args:
            addr (str | None): URL сервера Vault. Если не указан, будет прочитан из
            переменной окружения 'VAULT_ADDR'.
            token_file (str | None): Путь к файлу с токеном аутентификации Vault. Если не указан,
            будет прочитан из переменной окружения 'VAULT_TOKEN_FILE'.

        Raises:
            ValueError: Если не указаны параметры addr/token_file и соответствующие
            переменные окружения отсутствуют.

        """
        self._addr = addr or os.environ.get(VAULT_ADDR_ENV_NAME)
        self._token_file = token_file or os.environ.get(VAULT_TOKEN_FILE_ENV_NAME)
        self._client = self._init_client()

    def _init_client(self) -> hvac.Client:
        """Создает и настраивает клиент для работы с Vault.

        Returns:
            Аутентифицированный клиент Vault

        Raises:
            VaultError: Если не удалось инициализировать клиент
            VaultConnectionError: Если не удалось подключиться к серверу
            VaultTokenError: Если возникла проблема с токеном
            VaultAuthenticationError: Если аутентификация не удалась

        """
        if not self._addr or not self._token_file:
            logger.error(
                f'Не заданы переменные окружения {VAULT_ADDR_ENV_NAME} '
                f'и {VAULT_TOKEN_FILE_ENV_NAME}',
            )
            raise VaultError(
                f'Переменные окружения {VAULT_ADDR_ENV_NAME} и '
                f'{VAULT_TOKEN_FILE_ENV_NAME} должны быть заданы',
            )

        try:
            with open(self._token_file) as f:
                token = f.read().strip()

            if not token:
                logger.error('Отсутствие токена при инициализации Vault-клиента')
                raise VaultTokenError('Токен не может быть пустым')

            client = hvac.Client(url=self._addr, token=token)

            if not client.is_authenticated():
                logger.error('Ошибка при аутентификации в Vault-клиенте')
                raise VaultAuthenticationError('Не удалось аутентифицироваться в Vault')

            logger.info('Vault-клиент успешно аутентифицирован')
            return client

        except FileNotFoundError as e:
            logger.exception(f'Файл с токеном для Vault-клиента не найден: {self._token_file}')
            raise VaultTokenError(f'Файл с токеном не найден: {self._token_file}') from e
        except PermissionError as e:
            logger.exception(f'Нет прав на чтение токена Vault-клиентом: {self._token_file}')
            raise VaultPermissionError(f'Нет прав на чтение токена: {self._token_file}') from e
        except VaultError:
            raise
        except Exception as e:
            if 'connection' in str(e).lower():
                logger.exception(f'Не удалось подключиться к Vault: {e}')
                raise VaultConnectionError(f'Не удалось подключиться к Vault: {e}') from e
            logger.exception(f'Ошибка при инициализации клиента Vault: {e}')
            raise VaultError(f'Ошибка при инициализации клиента Vault: {e}') from e

    async def get_secret(self, path: str, key: str | None = None) -> dict[str, Any]:  # type: ignore
        """Получает секрет из Vault по указанному пути.

        Опционально можно передать key - конкретно интересующий ключ по пути

        Пример:
            # Получение секрета
            secret = vault.read_secret('eebook/users/database')
            # Результат: {'username': 'admin', 'password': 's3cr3t'} # pragma: allowlist secret

            # Использование полученных данных
            db_user = secret['username']
            db_pass = secret['password']

        Params:
            path: Путь к секрету в формате 'путь/к/секрету'.
                  Для KV v2 автоматически добавляется префикс 'data/' при необходимости.
            key: Название секрета (пр. db_port)

        Returns:
            Словарь с данными секрета

        Raises:
            VaultSecretNotFoundError: Если секрет не найден по указанному пути
            VaultPermissionError: Если недостаточно прав для доступа к секрету
            VaultError: При других ошибках при чтении секрета

        Примечания:
            - Для KV v2 секреты хранятся в формате 'путь/к/секрету',
              а не в полном пути 'secret/data/путь/к/секрету'
            - Метод автоматически обрабатывает версионирование KV v2

        """
        try:
            secret = self._client.secrets.kv.v2.read_secret_version(path=path)
            data = secret['data']['data']
            logger.debug('Успешно прочитан секрет из Vault')

            if key is not None:
                if key not in data:
                    logger.error(f'Ключ "{key}" не найден в секрете по пути: {path}')
                    raise VaultSecretNotFoundError(
                        f'Ключ "{key}" не найден в секрете по пути: {path}',
                    )
                return data[key]

            return data
        except VaultError:
            raise

        except Exception as e:
            if '404' in str(e) or 'Not found' in str(e):
                logger.exception(
                    f'Не удалось обнаружить секрет в Vault по пути: {path}',
                )
                raise VaultSecretNotFoundError(f'Секрет не найден: {path}') from e
            logger.exception('Ошибка при чтении Vault-секрета')
            raise VaultError(f'Ошибка при чтении секрета: {e}') from e

__init__(addr: str | None = None, token_file: str | None = None) -> None

Инициализирует клиент Vault с параметрами подключения.

Parameters:

Name Type Description Default
addr str | None

URL сервера Vault. Если не указан, будет прочитан из

 None
token_file str | None

Путь к файлу с токеном аутентификации Vault. Если не указан,

 None

Raises:

Type Description
ValueError

Если не указаны параметры addr/token_file и соответствующие
Source code in src/adapters/vault.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(self, addr: str | None = None, token_file: str | None = None) -> None:
    """Инициализирует клиент Vault с параметрами подключения.

    Args:
        addr (str | None): URL сервера Vault. Если не указан, будет прочитан из
        переменной окружения 'VAULT_ADDR'.
        token_file (str | None): Путь к файлу с токеном аутентификации Vault. Если не указан,
        будет прочитан из переменной окружения 'VAULT_TOKEN_FILE'.

    Raises:
        ValueError: Если не указаны параметры addr/token_file и соответствующие
        переменные окружения отсутствуют.

    """
    self._addr = addr or os.environ.get(VAULT_ADDR_ENV_NAME)
    self._token_file = token_file or os.environ.get(VAULT_TOKEN_FILE_ENV_NAME)
    self._client = self._init_client()

get_secret(path: str, key: str | None = None) -> dict[str, Any] async

Получает секрет из Vault по указанному пути.

Опционально можно передать key - конкретно интересующий ключ по пути

Пример

Получение секрета

secret = vault.read_secret('eebook/users/database')

Результат: {'username': 'admin', 'password': 's3cr3t'} # pragma: allowlist secret

Использование полученных данных

db_user = secret['username'] db_pass = secret['password']

Parameters:

Name Type Description Default
path str

Путь к секрету в формате 'путь/к/секрету'. Для KV v2 автоматически добавляется префикс 'data/' при необходимости.

 required
key str | None

Название секрета (пр. db_port)

 None

Returns:

Type Description
dict[str, Any]

Словарь с данными секрета

Raises:

Type Description
VaultSecretNotFoundError

Если секрет не найден по указанному пути
VaultPermissionError

Если недостаточно прав для доступа к секрету
VaultError

При других ошибках при чтении секрета
Примечания
  • Для KV v2 секреты хранятся в формате 'путь/к/секрету', а не в полном пути 'secret/data/путь/к/секрету'
  • Метод автоматически обрабатывает версионирование KV v2
Source code in src/adapters/vault.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def get_secret(self, path: str, key: str | None = None) -> dict[str, Any]:  # type: ignore
    """Получает секрет из Vault по указанному пути.

    Опционально можно передать key - конкретно интересующий ключ по пути

    Пример:
        # Получение секрета
        secret = vault.read_secret('eebook/users/database')
        # Результат: {'username': 'admin', 'password': 's3cr3t'} # pragma: allowlist secret

        # Использование полученных данных
        db_user = secret['username']
        db_pass = secret['password']

    Params:
        path: Путь к секрету в формате 'путь/к/секрету'.
              Для KV v2 автоматически добавляется префикс 'data/' при необходимости.
        key: Название секрета (пр. db_port)

    Returns:
        Словарь с данными секрета

    Raises:
        VaultSecretNotFoundError: Если секрет не найден по указанному пути
        VaultPermissionError: Если недостаточно прав для доступа к секрету
        VaultError: При других ошибках при чтении секрета

    Примечания:
        - Для KV v2 секреты хранятся в формате 'путь/к/секрету',
          а не в полном пути 'secret/data/путь/к/секрету'
        - Метод автоматически обрабатывает версионирование KV v2

    """
    try:
        secret = self._client.secrets.kv.v2.read_secret_version(path=path)
        data = secret['data']['data']
        logger.debug('Успешно прочитан секрет из Vault')

        if key is not None:
            if key not in data:
                logger.error(f'Ключ "{key}" не найден в секрете по пути: {path}')
                raise VaultSecretNotFoundError(
                    f'Ключ "{key}" не найден в секрете по пути: {path}',
                )
            return data[key]

        return data
    except VaultError:
        raise

    except Exception as e:
        if '404' in str(e) or 'Not found' in str(e):
            logger.exception(
                f'Не удалось обнаружить секрет в Vault по пути: {path}',
            )
            raise VaultSecretNotFoundError(f'Секрет не найден: {path}') from e
        logger.exception('Ошибка при чтении Vault-секрета')
        raise VaultError(f'Ошибка при чтении секрета: {e}') from e

Примеры использования

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

from src.adapters.vault import VaultClient

# Инициализация с явным указанием параметров
vault = VaultClient(
    addr='http://localhost:8200',  # Адрес сервера Vault
    token_file='/path/to/vault/token'  # Файл с токеном доступа
)

# Или с использованием переменных окружения
# export VAULT_ADDR='http://localhost:8200'
# export VAULT_TOKEN_FILE='/path/to/vault/token'
vault = VaultClient()

Работа с секретами

# Чтение всего секрета
secret = await vault.get_secret('eebook/database')
# Результат: {'username': 'admin', 'password': 's3cr3t'}

# Чтение конкретного ключа из секрета
db_password = await vault.get_secret('eebook/database', key='password')
# Результат: 's3cr3t'

# Обработка ошибок
try:
    secret = await vault.get_secret('nonexistent/secret')
except VaultSecretNotFoundError:
    print('Секрет не найден')
except VaultPermissionError:
    print('Недостаточно прав для доступа к секрету')
except VaultError as e:
    print(f'Ошибка Vault: {e}')

Рекомендации по использованию

  1. Безопасность
  2. Никогда не храните токены в коде или системе контроля версий
  3. Используйте минимально необходимые права доступа для токенов

  4. Регулярно обновляйте токены доступа

  5. Обработка ошибок

  6. Всегда обрабатывайте специфические исключения Vault
  7. Реализуйте механизм повторных попыток при временных сбоях

  8. Логируйте ошибки для последующего аудита

  9. Производительность

  10. Кэшируйте часто используемые секреты на стороне приложения
  11. Избегайте частых обращений к Vault за одними и теми же данными

  12. Используйте пулы соединений для часто используемых клиентов

  13. Тестирование

  14. Используйте моки для тестирования без реального сервера Vault
  15. Тестируйте различные сценарии ошибок
  16. Проверяйте обработку отсутствующих или невалидных данных

Обработка ошибок

Модуль определяет несколько типов исключений для различных сценариев сбоев:

  • VaultError: Базовый класс для всех исключений Vault
  • VaultConnectionError: Ошибка подключения к серверу Vault
  • VaultTokenError: Ошибка аутентификации или невалидный токен
  • VaultPermissionError: Недостаточно прав для выполнения операции
  • VaultSecretNotFoundError: Запрашиваемый секрет не найден

Переменные окружения

  • VAULT_ADDR: URL сервера Vault (например, 'http://localhost:8200')
  • VAULT_TOKEN_FILE: Путь к файлу, содержащему токен аутентификации