Source code for gcip2.vault

 1import getpass
 2import sys
 3from os import getenv as os_getenv
 4
 5import hvac
 6import injector
 7
 8from gcip2.logging import logger as LOGGER
 9from gcip2.project_config import ProjectConfig
10from gcip2.project_config.vault_config import VaultConfig
11from gcip2.vault.token_store import TokenStore
12
13

[docs]
14@injector.inject
15class Vault:
16    def __init__(self, vault_config: VaultConfig, _token_store: TokenStore):
17        self.vault_config = vault_config
18        self._client = hvac.Client(url=self.vault_config.url)
19        LOGGER.debug(f"initialized client for vault instance: {self.vault_config.url}")
20
21        self._tokens = _token_store
22        self.auth()
23

[docs]
24    def auth(self) -> None:
25        LOGGER.debug(f"authorization with approle {self.vault_config.app_role_id_env_var}")
26
27        role_id = os_getenv(str(self.vault_config.app_role_id_env_var))
28        secret_id = os_getenv(str(self.vault_config.app_role_secret_id_env_var))
29
30        if role_id and secret_id:
31            self._client.auth.approle.login(role_id=role_id, secret_id=secret_id)  # type: ignore[attr-defined]
32            return
33        self.auth_keyring()

34

[docs]
35    def auth_keyring(self) -> None:
36        token = self._tokens.load(self.vault_config.url)
37        if token:
38            self._client.token = token
39            if self._client.is_authenticated():
40                LOGGER.debug("token from keyring is valid")
41                return
42            LOGGER.debug("token from keyring is expired")
43
44        token = self.auth_interactive()
45        self._client.token = token
46        if not self._client.is_authenticated():
47            raise ValueError("could not authenticate with the obtained token")
48        self._tokens.save(self.vault_config.url, token)

49

[docs]
50    def auth_interactive(self) -> str:
51        if not sys.stdin.isatty():
52            raise ValueError(
53                "cannot authenticate: no approle credentials in environment "
54                f"({self.vault_config.app_role_id_env_var}, "
55                f"{self.vault_config.app_role_secret_id_env_var}) and stdin is not a terminal"
56            )
57        LOGGER.info(f"authenticating to {self.vault_config.url}")
58        username = input("LDAP username: ")
59        password = getpass.getpass("LDAP password: ")
60        return self._client.auth.ldap.login(username=username, password=password)["auth"]["client_token"]

61

[docs]
62    def fetch(self) -> dict[str, str]:
63        LOGGER.debug(f"fetching secret from {self.vault_config.mount_point}:{self.vault_config.path}")
64        return self._client.secrets.kv.read_secret_version(  # type: ignore[attr-defined]
65            mount_point=self.vault_config.mount_point,
66            path=self.vault_config.path,
67            raise_on_deleted_version=True,
68        )["data"]["data"]


69
70

[docs]
71class SecretsHandler:
72    @injector.inject
73    def __init__(self, config: ProjectConfig, _token_store: TokenStore) -> None:
74        self._config = config
75        self._tokens = _token_store
76

[docs]
77    def normalize_secret(self, secret: str) -> VaultConfig:
78        if isinstance(secret, VaultConfig):
79            return secret
80        vault_config = self._config.extra.secrets.vault.get(secret)
81        if not vault_config:
82            raise ValueError(f"missing secret '{secret}'")
83        return vault_config

84

[docs]
85    def fetch(self, secret: str):
86        vault_config = self.normalize_secret(secret)
87        LOGGER.info(f"fetching secret: '{secret}'")
88        client = Vault(vault_config, self._tokens)
89        return client.fetch()