File: policies.py

package info (click to toggle)
python-hvac 2.3.0-5
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 2,800 kB
  • sloc: python: 29,360; makefile: 42; sh: 14
file content (236 lines) | stat: -rw-r--r-- 8,182 bytes parent folder | download | duplicates (2)
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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
import json

from hvac import utils
from hvac.api.system_backend.system_backend_mixin import SystemBackendMixin


class Policies(SystemBackendMixin):
    def list_acl_policies(self):
        """List all configured acl policies.

        Supported methods:
            GET: /sys/policies/acl. Produces: 200 application/json

        :return: The JSON response of the request.
        :rtype: dict
        """
        api_path = "/v1/sys/policies/acl"
        return self._adapter.list(
            url=api_path,
        )

    def read_acl_policy(self, name):
        """Retrieve the policy body for the named acl policy.

        Supported methods:
            GET: /sys/policies/acl/{name}. Produces: 200 application/json

        :param name: The name of the acl policy to retrieve.
        :type name: str | unicode
        :return: The response of the request
        :rtype: dict
        """
        api_path = utils.format_url("/v1/sys/policies/acl/{name}", name=name)
        return self._adapter.get(
            url=api_path,
        )

    def create_or_update_acl_policy(self, name, policy, pretty_print=True):
        """Add a new or update an existing acl policy.

        Once a policy is updated, it takes effect immediately to all associated users.

        Supported methods:
            PUT: /sys/policies/acl/{name}. Produces: 204 (empty body)

        :param name: Specifies the name of the policy to create.
        :type name: str | unicode
        :param policy: Specifies the policy to create or update.
        :type policy: str | unicode | dict
        :param pretty_print: If True, and provided a dict for the policy argument, send the policy JSON to Vault with
            "pretty" formatting.
        :type pretty_print: bool
        :return: The response of the request.
        :rtype: requests.Response
        """
        if isinstance(policy, dict):
            if pretty_print:
                policy = json.dumps(policy, indent=4, sort_keys=True)
            else:
                policy = json.dumps(policy)
        params = {
            "policy": policy,
        }
        api_path = utils.format_url(f"/v1/sys/policies/acl/{name}", name=name)
        return self._adapter.put(
            url=api_path,
            json=params,
        )

    def delete_acl_policy(self, name):
        """Delete the acl policy with the given name.

        This will immediately affect all users associated with this policy.

        Supported methods:
            DELETE: /sys/policies/acl/{name}. Produces: 204 (empty body)

        :param name: Specifies the name of the policy to delete.
        :type name: str | unicode
        :return: The response of the request.
        :rtype: requests.Response
        """
        api_path = utils.format_url("/v1/sys/policies/acl/{name}", name=name)
        return self._adapter.delete(
            url=api_path,
        )

    def list_rgp_policies(self):
        """List all configured rgp policies.

        Supported methods:
            GET: /sys/policies/rgp. Produces: 200 application/json

        :return: The JSON response of the request.
        :rtype: dict
        """
        api_path = "/v1/sys/policies/rgp"
        return self._adapter.list(
            url=api_path,
        )

    def read_rgp_policy(self, name):
        """Retrieve the policy body for the named rgp policy.

        Supported methods:
            GET: /sys/policies/rgp/{name}. Produces: 200 application/json

        :param name: The name of the rgp policy to retrieve.
        :type name: str | unicode
        :return: The response of the request
        :rtype: dict
        """
        api_path = utils.format_url("/v1/sys/policies/rgp/{name}", name=name)
        return self._adapter.get(
            url=api_path,
        )

    def create_or_update_rgp_policy(self, name, policy, enforcement_level):
        """Add a new or update an existing rgp policy.

        Once a policy is updated, it takes effect immediately to all associated users.

        Supported methods:
            PUT: /sys/policies/rgp/{name}. Produces: 204 (empty body)

        :param name: Specifies the name of the policy to create.
        :type name: str | unicode
        :param policy: Specifies the policy to create or update.
        :type policy: str | unicode
        :param enforcement_level: Specifies the enforcement level to use. This must be one of advisory, soft-mandatory, or hard-mandatory
        :type enforcement_level: str | unicode
        :return: The response of the request.
        :rtype: requests.Response
        """
        params = {"policy": policy, "enforcement_level": enforcement_level}
        api_path = utils.format_url(f"/v1/sys/policies/rgp/{name}", name=name)
        return self._adapter.put(
            url=api_path,
            json=params,
        )

    def delete_rgp_policy(self, name):
        """Delete the rgp policy with the given name.

        This will immediately affect all users associated with this policy.

        Supported methods:
            DELETE: /sys/policies/rgp/{name}. Produces: 204 (empty body)

        :param name: Specifies the name of the policy to delete.
        :type name: str | unicode
        :return: The response of the request.
        :rtype: requests.Response
        """
        api_path = utils.format_url("/v1/sys/policies/rgp/{name}", name=name)
        return self._adapter.delete(
            url=api_path,
        )

    def list_egp_policies(self):
        """List all configured egp policies.

        Supported methods:
            GET: /sys/policies/egp. Produces: 200 application/json

        :return: The JSON response of the request.
        :rtype: dict
        """
        api_path = "/v1/sys/policies/egp"
        return self._adapter.list(
            url=api_path,
        )

    def read_egp_policy(self, name):
        """Retrieve the policy body for the named egp policy.

        Supported methods:
            GET: /sys/policies/egp/{name}. Produces: 200 application/json

        :param name: The name of the egp policy to retrieve.
        :type name: str | unicode
        :return: The response of the request
        :rtype: dict
        """
        api_path = utils.format_url("/v1/sys/policies/egp/{name}", name=name)
        return self._adapter.get(
            url=api_path,
        )

    def create_or_update_egp_policy(self, name, policy, enforcement_level, paths):
        """Add a new or update an existing egp policy.

        Once a policy is updated, it takes effect immediately to all associated users.

        Supported methods:
            PUT: /sys/policies/egp/{name}. Produces: 204 (empty body)

        :param name: Specifies the name of the policy to create.
        :type name: str | unicode
        :param policy: Specifies the policy to create or update.
        :type policy: str | unicode
        :param enforcement_level: Specifies the enforcement level to use. This must be one of advisory, soft-mandatory, or hard-mandatory
        :type enforcement_level: str | unicode
        :param paths: Specifies the paths on which this EGP should be applied.
        :type paths: list
        :return: The response of the request.
        :rtype: requests.Response
        """
        params = {
            "policy": policy,
            "enforcement_level": enforcement_level,
            "paths": paths,
        }
        api_path = utils.format_url(f"/v1/sys/policies/egp/{name}", name=name)
        return self._adapter.put(
            url=api_path,
            json=params,
        )

    def delete_egp_policy(self, name):
        """Delete the egp policy with the given name.

        This will immediately affect all users associated with this policy.

        Supported methods:
            DELETE: /sys/policies/egp/{name}. Produces: 204 (empty body)

        :param name: Specifies the name of the policy to delete.
        :type name: str | unicode
        :return: The response of the request.
        :rtype: requests.Response
        """
        api_path = utils.format_url("/v1/sys/policies/egp/{name}", name=name)
        return self._adapter.delete(
            url=api_path,
        )