Source code for baseplate.clients.hvac
import datetime
from typing import Any
import hvac
import requests
from baseplate import Span
from baseplate.clients import ContextFactory
from baseplate.lib import config
from baseplate.lib.secrets import SecretsStore
[docs]def hvac_factory_from_config(
app_config: config.RawConfig, secrets_store: SecretsStore, prefix: str = "vault."
) -> "HvacContextFactory":
"""Make an HVAC client factory from a configuration dictionary.
The keys useful to :py:func:`hvac_factory_from_config` should be prefixed,
e.g. ``vault.timeout``. The ``prefix`` argument specifies the prefix used
to filter keys.
Supported keys:
* ``timeout``: How long to wait for calls to Vault.
(:py:func:`~baseplate.lib.config.Timespan`)
:param app_config: The raw application configuration.
:param secrets_store: A configured secrets store from which we can get a
Vault authentication token.
:param prefix: The prefix for configuration keys.
"""
assert prefix.endswith(".")
parser = config.SpecParser(
{"timeout": config.Optional(config.Timespan, default=datetime.timedelta(seconds=1))}
)
options = parser.parse(prefix[:-1], app_config)
return HvacContextFactory(secrets_store, options.timeout)
[docs]class HvacClient(config.Parser):
"""Configure an HVAC client.
This is meant to be used with
:py:meth:`baseplate.Baseplate.configure_context`.
See :py:func:`hvac_factory_from_config` for available configuration settings.
:param secrets: The configured secrets store for this application.
"""
def __init__(self, secrets: SecretsStore):
self.secrets = secrets
def parse(self, key_path: str, raw_config: config.RawConfig) -> "HvacContextFactory":
return hvac_factory_from_config(
raw_config, secrets_store=self.secrets, prefix=f"{key_path}."
)
[docs]class HvacContextFactory(ContextFactory):
"""HVAC client context factory.
This factory will attach a proxy object which acts like an
:py:class:`hvac.Client` to an attribute on the
:py:class:`~baseplate.RequestContext`. All methods that talk to Vault will
be automatically instrumented for tracing and diagnostic metrics.
:param baseplate.lib.secrets.SecretsStore secrets_store: Configured secrets
store from which we can get a Vault authentication token.
:param datetime.timedelta timeout: How long to wait for calls to Vault.
"""
def __init__(self, secrets_store: SecretsStore, timeout: datetime.timedelta):
self.secrets = secrets_store
self.timeout = timeout
self.session = requests.Session()
[docs] def make_object_for_context(self, name: str, span: Span) -> "InstrumentedHvacClient":
vault_url = self.secrets.get_vault_url()
vault_token = self.secrets.get_vault_token()
return InstrumentedHvacClient(
url=vault_url,
token=vault_token,
timeout=self.timeout.total_seconds(),
session=self.session,
context_name=name,
server_span=span,
)
class InstrumentedHvacClient(hvac.Client):
def __init__(
self,
url: str,
token: str,
timeout: float,
session: requests.Session,
context_name: str,
server_span: Span,
):
self.context_name = context_name
self.server_span = server_span
super().__init__(url=url, token=token, timeout=timeout, session=session)
# this ugliness is us undoing the name mangling that __request turns into
# inside python. this feels very dirty.
def _Client__request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
span_name = f"{self.context_name}.request"
with self.server_span.make_child(span_name) as span:
span.set_tag("http.method", method.upper())
span.set_tag("http.url", url)
# pylint: disable=no-member
response = super()._Client__request(method=method, url=url, **kwargs)
# this means we can't get the status code from error responses.
# that's unfortunate, but hvac doesn't make it easy.
span.set_tag("http.status_code", response.status_code)
return response