#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Database methods module."""
from hvac import utils
from hvac.api.vault_api_base import VaultApiBase
DEFAULT_MOUNT_POINT = "database"
[docs]class Database(VaultApiBase):
"""Database Secrets Engine (API).
Reference: https://www.vaultproject.io/api/secret/databases/index.html
"""
[docs] def rotate_root_credentials(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint is used to rotate the root superuser credentials stored for the database connection.
This user must have permissions to update its own password.
:param name: Specifies the name of the connection to rotate.
:type name: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/rotate-root/{name}', mount_point=mount_point, name=name)
return self._adapter.post(
url=api_path,
)
[docs] def read_connection(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint returns the configuration settings for a connection.
:param name: Specifies the name of the connection to read.
:type name: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/config/{name}', mount_point=mount_point, name=name)
return self._adapter.get(
url=api_path,
)
[docs] def list_connections(self, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint returns a list of available connections.
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/config', mount_point=mount_point)
return self._adapter.list(
url=api_path,
)
[docs] def delete_connection(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint deletes a connection.
:param name: Specifies the name of the connection to delete.
:type name: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/config/{name}', mount_point=mount_point, name=name)
return self._adapter.delete(
url=api_path,
)
[docs] def reset_connection(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint closes a connection and it's underlying plugin and
restarts it with the configuration stored in the barrier.
:param name: Specifies the name of the connection to reset.
:type name: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/reset/{name}', mount_point=mount_point, name=name)
return self._adapter.post(
url=api_path,
)
[docs] def create_role(self, name, db_name, creation_statements, default_ttl=None, max_ttl=None,
revocation_statements=None, rollback_statements=None, renew_statements=None,
mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint creates or updates a role definition.
:param name: Specifies the database role to manage.
:type name: str | unicode
:param db_name: The name of the database connection to use for this role.
:type db_name: str | unicode
:param creation_statements: Specifies the database statements executed to create and configure a user.
:type creation_statements: list
:param default_ttl: Specifies the TTL for the leases associated with this role.
:type default_ttl: int
:param max_ttl: Specifies the maximum TTL for the leases associated with this role.
:type max_ttl: int
:param revocation_statements: Specifies the database statements to be executed to revoke a user.
:type revocation_statements: list
:param rollback_statements: Specifies the database statements to be executed to rollback
a create operation in the event of an error.
:type rollback_statements: list
:param renew_statements: Specifies the database statements to be executed to renew a user.
:type renew_statements: list
:param mount_point: The "path" the method/backend was mounted on.
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
params = {
"db_name": db_name,
"creation_statements": creation_statements,
}
params.update(
utils.remove_nones({
"default_ttl": default_ttl,
"max_ttl": max_ttl,
"revocation_statements": revocation_statements,
"rollback_statements": rollback_statements,
"renew_statements": renew_statements,
})
)
api_path = utils.format_url('/v1/{mount_point}/roles/{name}', mount_point=mount_point, name=name)
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: The "path" the method/backend was mounted on.
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/roles/{name}', mount_point=mount_point, name=name)
return self._adapter.get(
url=api_path,
)
[docs] def list_roles(self, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint returns a list of available roles.
:param mount_point: The "path" the method/backend was mounted on.
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/roles', mount_point=mount_point)
return self._adapter.list(
url=api_path,
)
[docs] def delete_role(self, name, mount_point=DEFAULT_MOUNT_POINT):
"""This endpoint deletes the role definition.
:param name: Specifies the name of the role to delete.
:type name: str | unicode
:param mount_point: The "path" the method/backend was mounted on.
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/roles/{name}', mount_point=mount_point, name=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: The "path" the method/backend was mounted on.
:type mount_point: str | unicode
:return: The response of the request.
:rtype: requests.Response
"""
api_path = utils.format_url('/v1/{mount_point}/creds/{name}', mount_point=mount_point, name=name)
return self._adapter.get(
url=api_path,
)