Support for HTTP/2 onward with Niquests

While unofficial, hvac can speak HTTP/2, and HTTP/3 using the alternative HTTP backend Niquests. It is made possible thanks to the library backward compatibility with Requests. We can explore a few possible integrations.

Warning

The samples presented in this section is not covered by hvac maintainers. Any issue is to be addressed to Niquests maintainers.

Session

Basically, the fastest way to upgrade the client is:

import niquests
import hvac

session = niquests.Session()

client = hvac.Client(
    url='https://vault.example.com',
    session=session,
)

Note

To know more about it, see the Niquests documentation.

Security

hvac handles critical operations that can bring undesired eyes into your HTTP interactions. Fortunately, Niquests enforce some best security practices like OCSP certificate validation and support encrypted DNS protocols, like but not limited to DNS-over-HTTPS or DNS-over-QUIC.

import niquests
import hvac

session = niquests.Session(resolver="doh+cloudflare://")

client = hvac.Client(
    url='https://vault.example.com',
    session=session,
)

It should get you covered in most cases of advanced attack scenarios. Support for DNSSEC is also automatically enabled in the given example.

Note

To know more about custom resolvers, see the DNS resolution documentation.

Multiplexing

You may leverage a multiplexed connection thanks to Niquests native capabilities. Unfortunately hvac does access the response immediately after receiving it, thus preventing non-blocking IO.

To be able to make concurrent requests using one connection, you will have to override the default adapter.

import niquests

from hvac.adapters import RawAdapter
from hvac.constants.client import DEFAULT_URL
from hvac import utils


class NiquestsAdapter(RawAdapter):

    def __init__(
        self, base_uri=DEFAULT_URL, token=None, cert=None,
        verify=True, timeout=30, proxies=None, allow_redirects=True,
        session=None, namespace=None, ignore_exceptions=False, strict_http=False,
        request_header=True, resolver=None, source_address=None
    ):
        if not session:
            session = niquests.Session(
                multiplexed=True,
                resolver=resolver,
                source_address=source_address,
            )
            session.cert, session.verify, session.proxies = cert, verify, proxies
        else:
            if session.verify:
                verify = session.verify
            if session.cert:
                cert = session.cert
            if session.proxies:
                proxies = session.proxies

        self.base_uri = base_uri
        self.token = token
        self.namespace = namespace
        self.session = session
        self.allow_redirects = allow_redirects
        self.ignore_exceptions = ignore_exceptions
        self.strict_http = strict_http
        self.request_header = request_header

        self._kwargs = {
            "cert": cert,
            "verify": verify,
            "timeout": timeout,
            "proxies": proxies,
        }

    def request(self, method, url, headers=None, raise_exception=True, **kwargs):
        while "//" in url:
            # Vault CLI treats a double forward slash ('//') as a single forward slash for a given path.
            # To avoid issues with the requests module's redirection logic, we perform the same translation here.
            url = url.replace("//", "/")

        url = self.urljoin(self.base_uri, url)

        if not headers:
            headers = {}

        if self.request_header:
            headers["X-Vault-Request"] = "true"

        if self.token:
            headers["X-Vault-Token"] = self.token

        if self.namespace:
            headers["X-Vault-Namespace"] = self.namespace

        wrap_ttl = kwargs.pop("wrap_ttl", None)
        if wrap_ttl:
            headers["X-Vault-Wrap-TTL"] = str(wrap_ttl)

        _kwargs = self._kwargs.copy()
        _kwargs.update(kwargs)

        if self.strict_http and method.lower() in ("list",):
            # Entry point for standard HTTP substitution
            params = _kwargs.get("params", {})
            if method.lower() == "list":
                method = "get"
                params.update({"list": "true"})
            _kwargs["params"] = params

        if not self.ignore_exceptions and raise_exception:
            def check_error(resp):
                nonlocal method, url

                if not resp.ok:
                    msg = json = errors = None

                    text = response.text

                    try:
                        json = response.json()
                    except JSONDecodeError:
                        pass
                    else:
                        errors = json.get("errors")

                    if errors is None:
                        msg = text

                    utils.raise_for_error(
                        method,
                        url,
                        response.status_code,
                        msg,
                        errors=errors,
                        text=text,
                        json=json,
                    )

            _kwargs["hooks"] = {
                "response": [check_error]
            }

        response = self.session.request(
            method=method,
            url=url,
            headers=headers,
            allow_redirects=self.allow_redirects,
            **_kwargs
        )

        return response

Note

To know more about multiplexing, see the Multiplexing documentation.

Then you would use the newly constructed adapter into your hvac client like so:

import hvac

client = hvac.Client(
    url='https://vault.example.com',
    adapter=NiquestsAdapter,
)