# coding=utf-8
"""
HTTP Client Library Adapters
"""
from abc import ABCMeta, abstractmethod
import requests
import requests.exceptions
from hvac import utils
DEFAULT_BASE_URI = 'http://localhost:8200'
[docs]class Adapter(object):
"""Abstract base class used when constructing adapters for use with the Client class."""
__metaclass__ = ABCMeta
[docs] def __init__(self, base_uri=DEFAULT_BASE_URI, token=None, cert=None, verify=True, timeout=30, proxies=None,
allow_redirects=True, session=None, namespace=None):
"""Create a new request adapter instance.
:param base_uri: Base URL for the Vault instance being addressed.
:type base_uri: str
:param token: Authentication token to include in requests sent to Vault.
:type token: str
:param cert: Certificates for use in requests sent to the Vault instance. This should be a tuple with the
certificate and then key.
:type cert: tuple
:param verify: Either a boolean to indicate whether TLS verification should be performed when sending requests to Vault,
or a string pointing at the CA bundle to use for verification. See http://docs.python-requests.org/en/master/user/advanced/#ssl-cert-verification.
:type verify: Union[bool,str]
:param timeout: The timeout value for requests sent to Vault.
:type timeout: int
:param proxies: Proxies to use when preforming requests.
See: http://docs.python-requests.org/en/master/user/advanced/#proxies
:type proxies: dict
:param allow_redirects: Whether to follow redirects when sending requests to Vault.
:type allow_redirects: bool
:param session: Optional session object to use when performing request.
:type session: request.Session
:param namespace: Optional Vault Namespace.
:type namespace: str
"""
if not session:
session = requests.Session()
self.base_uri = base_uri
self.token = token
self.namespace = namespace
self.session = session
self.allow_redirects = allow_redirects
self._kwargs = {
'cert': cert,
'verify': verify,
'timeout': timeout,
'proxies': proxies,
}
[docs] @staticmethod
def urljoin(*args):
"""Joins given arguments into a url. Trailing and leading slashes are stripped for each argument.
:param args: Multiple parts of a URL to be combined into one string.
:type args: str | unicode
:return: Full URL combining all provided arguments
:rtype: str | unicode
"""
return '/'.join(map(lambda x: str(x).strip('/'), args))
[docs] def close(self):
"""Close the underlying Requests session.
"""
self.session.close()
[docs] def get(self, url, **kwargs):
"""Performs a GET request.
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
return self.request('get', url, **kwargs)
[docs] def post(self, url, **kwargs):
"""Performs a POST request.
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
return self.request('post', url, **kwargs)
[docs] def put(self, url, **kwargs):
"""Performs a PUT request.
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
return self.request('put', url, **kwargs)
[docs] def delete(self, url, **kwargs):
"""Performs a DELETE request.
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
return self.request('delete', url, **kwargs)
[docs] def list(self, url, **kwargs):
"""Performs a LIST request.
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
return self.request('list', url, **kwargs)
[docs] def head(self, url, **kwargs):
"""Performs a HEAD request.
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
return self.request('head', url, **kwargs)
[docs] def login(self, url, use_token=True, **kwargs):
"""Perform a login request.
Associated request is typically to a path prefixed with "/v1/auth") and optionally stores the client token sent
in the resulting Vault response for use by the :py:meth:`hvac.adapters.Adapter` instance under the _adapater
Client attribute.
:param url: Path to send the authentication request to.
:type url: str | unicode
:param use_token: if True, uses the token in the response received from the auth request to set the "token"
attribute on the the :py:meth:`hvac.adapters.Adapter` instance under the _adapater Client attribute.
:type use_token: bool
:param kwargs: Additional keyword arguments to include in the params sent with the request.
:type kwargs: dict
:return: The response of the auth request.
:rtype: requests.Response
"""
response = self.post(url, **kwargs).json()
if use_token:
self.token = response['auth']['client_token']
return response
[docs] @utils.deprecated_method(
to_be_removed_in_version='0.9.0',
new_method=login,
)
def auth(self, url, use_token=True, **kwargs):
return self.login(
url=url,
use_token=use_token,
**kwargs
)
[docs] @abstractmethod
def request(self, method, url, headers=None, raise_exception=True, **kwargs):
"""Main method for routing HTTP requests to the configured Vault base_uri. Intended to be implement by subclasses.
:param method: HTTP method to use with the request. E.g., GET, POST, etc.
:type method: str
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param headers: Additional headers to include with the request.
:type headers: dict
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:param raise_exception: If True, raise an exception via utils.raise_for_error(). Set this parameter to False to
bypass this functionality.
:type raise_exception: bool
:return: The response of the request.
:rtype: requests.Response
"""
raise NotImplementedError
[docs]class Request(Adapter):
"""The Request adapter class"""
[docs] def request(self, method, url, headers=None, raise_exception=True, **kwargs):
"""Main method for routing HTTP requests to the configured Vault base_uri.
:param method: HTTP method to use with the request. E.g., GET, POST, etc.
:type method: str
:param url: Partial URL path to send the request to. This will be joined to the end of the instance's base_uri
attribute.
:type url: str | unicode
:param headers: Additional headers to include with the request.
:type headers: dict
:param raise_exception: If True, raise an exception via utils.raise_for_error(). Set this parameter to False to
bypass this functionality.
:type raise_exception: bool
:param kwargs: Additional keyword arguments to include in the requests call.
:type kwargs: dict
:return: The response of the request.
:rtype: requests.Response
"""
if '//' 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.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)
response = self.session.request(
method=method,
url=url,
headers=headers,
allow_redirects=self.allow_redirects,
**_kwargs
)
if raise_exception and 400 <= response.status_code < 600:
text = errors = None
if response.headers.get('Content-Type') == 'application/json':
errors = response.json().get('errors')
if errors is None:
text = response.text
utils.raise_for_error(response.status_code, text, errors=errors)
return response