stakost
25d910cef7
All checks were successful
Build and Deploy MikroTik Bot / build-and-deploy (push) Successful in 26s
84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
import os
|
||
import requests
|
||
import json
|
||
from typing import Dict, Optional
|
||
|
||
|
||
class VaultClient:
|
||
"""Клиент для работы с HashiCorp Vault через AppRole аутентификацию"""
|
||
|
||
def __init__(self):
|
||
self.vault_addr = os.environ.get('VAULT_ADDR', 'http://localhost:8200')
|
||
self.role_id = os.environ.get('VAULT_ROLE_ID')
|
||
self.secret_id = os.environ.get('VAULT_SECRET_ID')
|
||
self.secret_path = os.environ.get('VAULT_SECRET_PATH', 'secret/data/mikrotik-bot')
|
||
self.token = None
|
||
|
||
def authenticate(self) -> bool:
|
||
"""Аутентификация через AppRole"""
|
||
if not self.role_id or not self.secret_id:
|
||
print("❌ VAULT_ROLE_ID или VAULT_SECRET_ID не установлены")
|
||
return False
|
||
|
||
try:
|
||
auth_url = f"{self.vault_addr}/v1/auth/approle/login"
|
||
auth_data = {
|
||
"role_id": self.role_id,
|
||
"secret_id": self.secret_id
|
||
}
|
||
|
||
response = requests.post(auth_url, json=auth_data)
|
||
response.raise_for_status()
|
||
|
||
auth_result = response.json()
|
||
self.token = auth_result['auth']['client_token']
|
||
print("✅ Vault аутентификация успешна")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ Ошибка аутентификации Vault: {e}")
|
||
return False
|
||
|
||
def get_secrets(self) -> Optional[Dict[str, str]]:
|
||
"""Получение секретов из Vault"""
|
||
if not self.token and not self.authenticate():
|
||
return None
|
||
|
||
try:
|
||
headers = {"X-Vault-Token": self.token}
|
||
secret_url = f"{self.vault_addr}/v1/{self.secret_path}"
|
||
|
||
response = requests.get(secret_url, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
secret_data = response.json()
|
||
secrets = secret_data['data']['data']
|
||
|
||
print("✅ Секреты успешно получены из Vault")
|
||
return secrets
|
||
|
||
except Exception as e:
|
||
print(f"❌ Ошибка получения секретов: {e}")
|
||
return None
|
||
|
||
def get_config(self) -> Dict[str, str]:
|
||
"""Получение конфигурации с fallback на environment variables"""
|
||
# Сначала пытаемся получить из Vault
|
||
secrets = self.get_secrets()
|
||
|
||
if secrets:
|
||
return {
|
||
'BOT_TOKEN': secrets.get('bot_token'),
|
||
'ROUTER_HOST': secrets.get('router_host'),
|
||
'ROUTER_USER': secrets.get('router_user'),
|
||
'ROUTER_PASSWORD': secrets.get('router_password'),
|
||
}
|
||
|
||
# Fallback на environment variables
|
||
print("⚠️ Используются environment variables вместо Vault")
|
||
return {
|
||
'BOT_TOKEN': os.environ.get('BOT_TOKEN'),
|
||
'ROUTER_HOST': os.environ.get('ROUTER_HOST'),
|
||
'ROUTER_USER': os.environ.get('ROUTER_USER'),
|
||
'ROUTER_PASSWORD': os.environ.get('ROUTER_PASSWORD'),
|
||
} |