Source code for hvac.api.secrets_engines.rabbitmq
#!/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 configure(self, connection_uri="", username="", password="", verify_connection=True,
mount_point=DEFAULT_MOUNT_POINT):
"""Configure shared information for the rabbitmq secrets engine.
Supported methods:
POST: /{mount_point}/config/connection. Produces: 204 (empty body)
:param connection_uri: Specifies the RabbitMQ connection URI.
:type connection_uri: str | unicode
:param username: Specifies the RabbitMQ management administrator username.
:type username: str | unicode
:password: Specifies the RabbitMQ management administrator password.
:type password: str | unicode
:verify_connection: Specifies whether to verify connection URI, username, and password.
:type verify_connection: bool
: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
"""
params = {
"connection_uri": connection_uri,
"verify_connection": verify_connection,
"username": username,
"password": password
}
api_path = utils.format_url('/v1/{mount_point}/config/connection', mount_point=mount_point)
return self._adapter.post(
url=api_path,
json=params,
)
[docs] def configure_lease(self, ttl, max_ttl, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint configures the lease settings for generated credentials.
:param ttl: Specifies the lease ttl provided in seconds.
:type ttl: int
:param max_ttl: Specifies the maximum ttl provided in seconds.
:type max_ttl: int
: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/{}/config/lease", mount_point)
params = {
"ttl": ttl,
"max_ttl": max_ttl,
}
return self._adapter.post(
url=api_path,
json=params,
)
[docs] def create_role(self, name, tags="", vhosts="", 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 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,
}
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,
).json()
[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,
).json()