elastoo-team / sitri Goto Github PK

import hvac

from sitri.providers.contrib.vault import VaultKVConfigProvider
from sitri.settings.contrib.vault import VaultKVSettings
from sitri.providers.contrib.system import SystemConfigProvider


configurator = SystemConfigProvider()
ENV = configurator.get("ENV")

is_local_mode = ENV == "local"
local_mode_file_path = configurator.get("LOCAL_MODE_FILE_PATH")


def vault_client_factory() -> hvac.Client:
    """
    vault-provider
    :return: connected vault-client
    """
    client = hvac.Client(url=configurator.get("VAULT_URL"), verify=False)
    jwt_token_file = open('/run/secrets/kubernetes.io/serviceaccount/token')
    jwt = jwt_token_file.read()

    role = str(configurator.get("NAMESPACE")).lower()
    cluster = str(configurator.get("CLUSTERNAME")).lower()

    # Auth
    client.auth_kubernetes(
        role=role,
        jwt=jwt,
        mount_point=f'kubernetes/{cluster}',
    )

    client.is_authenticated()
    return client


provider = VaultKVConfigProvider(
    vault_connector=vault_client_factory,
    mount_point=f"{configurator.get('VAULT_PATH')}/{ENV}",
)


class BaseConfig(VaultKVSettings.VaultKVSettingsConfig):
    provider = provider
    local_mode = is_local_mode
    local_provider_args = {"json_path": local_mode_file_path}

from pydantic import BaseSettings, Field
from src.core.config.database_settings import DBSettings

class AppSettings(BaseSettings):
    db: DBSettings = Field(default_factory=DBSettings)

from typing import Optional
from pydantic import Field
from sitri.settings.contrib.vault import VaultKVSettings

from src.core.config.provider_config import provider, BaseConfig


class DBSettings(VaultKVSettings):
    user: str = Field(...)
    password: str = Field(...)
    host: str = Field(...)
    port: int = Field(...)
    db_name: str = Field(...)
    sqlalchemy_db_url: Optional[str] = None

    def __init__(self, **data):
        super(DBSettings, self).__init__(**data)
        self.sqlalchemy_db_url = f"postgresql://{self.user}:{self.password}@{self.host}:{self.port}/{self.db_name}"

    class Config(BaseConfig):
        provider = provider
        default_secret_path = "db"
        local_mode_path_prefix = "db"

def get(
        self, key: str, mount_point: typing.Optional[str] = None, secret_path: typing.Optional[str] = None, **kwargs
    ) -> typing.Optional[str]:
        request_params = {
            "path": secret_path if secret_path else self._secret_path,
            "mount_point": mount_point if mount_point else self._mount_point,
        }
        # current
        response = self._vault.secrets.kv.v1.read_secret(**request_params)
        # should be
        response = self._vault.secrets.kv.v1.read_secret(**request_params)["data"]

        return response["data"].get(key)