1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
#!/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"
class RabbitMQ(VaultApiBase):
"""RabbitMQ Secrets Engine (API).
Reference: https://www.vaultproject.io/api/secret/rabbitmq/index.html
"""
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,
)
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,
)
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,
)
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,
)
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,
)
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,
)
|
