#!/usr/bin/env python
"""RabbitMQ vault secrets backend module."""
from hvac import utils
from hvac.api.vault_api_base import VaultApiBase
DEFAULT_MOUNT_POINT = 'rabbitmq'
[docs]class RabbitMQ(VaultApiBase):
"""RabbitMQ Secrets Engine (API).
Reference: https://www.vaultproject.io/api/secret/rabbitmq/index.html
"""
[docs] def create_role(self, name, tags="", vhosts="", vhost_topics="", mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint creates or updates the role definition.
:param name: Specifies the name of the role to create.
:type name: str | unicode
:param tags: Specifies a comma-separated RabbitMQ management tags.
:type tags: str | unicode
:param vhosts: pecifies a map of virtual hosts to permissions.
:type vhosts: str | unicode
:param vhost_topics: Specifies a map of virtual hosts and exchanges to topic permissions.
:type vhost_topics: str | unicode
:param mount_point: Specifies the place where the secrets engine will be accessible (default: rabbitmq).
:type mount_point: str | unicode
:return: The JSON response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url("/v1/{}/roles/{}", mount_point, name)
params = {
"tags": tags,
"vhosts": vhosts,
"vhost_topics": vhost_topics
}
return self._adapter.post(
url=api_path,
json=params,
)
[docs] def read_role(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint queries the role definition.
:param name: Specifies the name of the role to read.
:type name: str | unicode
:param mount_point: Specifies the place where the secrets engine will be accessible (default: rabbitmq).
:type mount_point: str | unicode
:return: The JSON response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url("/v1/{}/roles/{}", mount_point, name)
return self._adapter.get(
url=api_path,
)
[docs] def delete_role(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint deletes the role definition.
Even if the role does not exist, this endpoint will still return a successful response.
:param name: Specifies the name of the role to delete.
:type name: str | unicode
:param mount_point: Specifies the place where the secrets engine will be accessible (default: rabbitmq).
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url("/v1/{}/roles/{}", mount_point, name)
return self._adapter.delete(
url=api_path,
)
[docs] def generate_credentials(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint generates a new set of dynamic credentials based on the named role.
:param name: Specifies the name of the role to create credentials against.
:type name: str | unicode
:param mount_point: Specifies the place where the secrets engine will be accessible (default: rabbitmq).
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url("/v1/{}/creds/{}", mount_point, name)
return self._adapter.get(
url=api_path,
)