File: userpass.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 (167 lines) | stat: -rw-r--r-- 5,582 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
#!/usr/bin/env python
"""USERPASS methods module."""
from hvac import utils
from hvac.api.vault_api_base import VaultApiBase

DEFAULT_MOUNT_POINT = "userpass"


class Userpass(VaultApiBase):
    """USERPASS Auth Method (API).
    Reference: https://www.vaultproject.io/api/auth/userpass/index.html
    """

    def create_or_update_user(
        self,
        username,
        password=None,
        policies=None,
        mount_point=DEFAULT_MOUNT_POINT,
        **kwargs,
    ):
        """
        Create/update user in userpass.

        Supported methods:
            POST: /auth/{mount_point}/users/{username}. Produces: 204 (empty body)

        :param username: The username for the user.
        :type username: str | unicode
        :param password: The password for the user. Only required when creating the user.
        :type password: str | unicode
        :param policies: The list of policies to be set on username created.
        :type policies: str | unicode
        :param mount_point: The "path" the method/backend was mounted on.
        :type mount_point: str | unicode
        :param kwargs: Additional arguments to pass along with the corresponding request to Vault.
        :type kwargs: dict
        """
        params = utils.remove_nones(
            {
                "password": password,
                "policies": policies,
            }
        )
        params.update(kwargs)

        api_path = "/v1/auth/{mount_point}/users/{username}".format(
            mount_point=mount_point, username=username
        )
        return self._adapter.post(
            url=api_path,
            json=params,
        )

    def list_user(self, mount_point=DEFAULT_MOUNT_POINT):
        """
        List existing users that have been created in the auth method

        Supported methods:
            LIST: /auth/{mount_point}/users. Produces: 200 application/json

        :param mount_point: The "path" the method/backend was mounted on.
        :type mount_point: str | unicode
        :return: The JSON response of the list_groups request.
        :rtype: dict
        """
        api_path = f"/v1/auth/{mount_point}/users"
        return self._adapter.list(
            url=api_path,
        )

    def read_user(self, username, mount_point=DEFAULT_MOUNT_POINT):
        """
        Read user in the auth method.

        Supported methods:
            GET: /auth/{mount_point}/users/{username}. Produces: 200 application/json

        :param username: The username for the user.
        :type name: str | unicode
        :param mount_point: The "path" the method/backend was mounted on.
        :type mount_point: str | unicode
        :return: The JSON response of the read_group request.
        :rtype: dict
        """
        api_path = "/v1/auth/{mount_point}/users/{username}".format(
            mount_point=mount_point, username=username
        )
        return self._adapter.get(
            url=api_path,
        )

    def delete_user(self, username, mount_point=DEFAULT_MOUNT_POINT):
        """
        Delete user in the auth method.

        Supported methods:
            GET: /auth/{mount_point}/users/{username}. Produces: 200 application/json

        :param username: The username for the user.
        :type name: str | unicode
        :param mount_point: The "path" the method/backend was mounted on.
        :type mount_point: str | unicode
        :return: The JSON response of the read_group request.
        :rtype: dict
        """
        api_path = "/v1/auth/{mount_point}/users/{username}".format(
            mount_point=mount_point, username=username
        )
        return self._adapter.delete(
            url=api_path,
        )

    def update_password_on_user(
        self, username, password, mount_point=DEFAULT_MOUNT_POINT
    ):
        """
        update password for the user in userpass.

        Supported methods:
            POST: /auth/{mount_point}/users/{username}/password. Produces: 204 (empty body)

        :param username: The username for the user.
        :type username: str | unicode
        :param password: The password for the user. Only required when creating the user.
        :type password: str | unicode
        :param mount_point: The "path" the method/backend was mounted on.
        :type mount_point: str | unicode
        """
        params = {
            "password": password,
        }
        api_path = "/v1/auth/{mount_point}/users/{username}/password".format(
            mount_point=mount_point, username=username
        )
        return self._adapter.post(
            url=api_path,
            json=params,
        )

    def login(
        self, username, password, use_token=True, mount_point=DEFAULT_MOUNT_POINT
    ):
        """
        Log in with USERPASS credentials.

        Supported methods:
            POST: /auth/{mount_point}/login/{username}. Produces: 200 application/json

        :param username: The username for the user.
        :type username: str | unicode
        :param password: The password for the user. Only required when creating the user.
        :type password: str | unicode
        :param mount_point: The "path" the method/backend was mounted on.
        :type mount_point: str | unicode
        """
        params = {
            "password": password,
        }
        api_path = "/v1/auth/{mount_point}/login/{username}".format(
            mount_point=mount_point, username=username
        )
        return self._adapter.login(
            url=api_path,
            use_token=use_token,
            json=params,
        )