File: user.py

package info (click to toggle)
python-softlayer 5.8.5-2
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 3,888 kB
  • sloc: python: 41,631; makefile: 289; sh: 43
file content (275 lines) | stat: -rw-r--r-- 11,188 bytes parent folder | download
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
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
"""
    SoftLayer.user
    ~~~~~~~~~~~~~
    User Manager/helpers

    :license: MIT, see LICENSE for more details.
"""
import datetime
import logging
from operator import itemgetter

from SoftLayer import exceptions
from SoftLayer import utils

LOGGER = logging.getLogger(__name__)


class UserManager(utils.IdentifierMixin, object):
    """Manages Users.

    See: https://softlayer.github.io/reference/datatypes/SoftLayer_User_Customer/

    Example::

       # Initialize the Manager.
       import SoftLayer
       client = SoftLayer.create_client_from_env()
       mgr = SoftLayer.UserManager(client)

    :param SoftLayer.API.BaseClient client: the client instance

    """

    def __init__(self, client):
        self.client = client
        self.user_service = self.client['SoftLayer_User_Customer']
        self.account_service = self.client['SoftLayer_Account']
        self.resolvers = [self._get_id_from_username]
        self.all_permissions = None

    def list_users(self, objectmask=None, objectfilter=None):
        """Lists all users on an account

        :param string objectmask: Used to overwrite the default objectmask.
        :param dictionary objectfilter: If you want to use an objectfilter.
        :returns: A list of dictionaries that describe each user

        Example::
            result = mgr.list_users()
        """

        if objectmask is None:
            objectmask = """mask[id, username, displayName, userStatus[name], hardwareCount, virtualGuestCount,
                                 email, roles]"""

        return self.account_service.getUsers(mask=objectmask, filter=objectfilter)

    def get_user(self, user_id, objectmask=None):
        """Calls SoftLayer_User_Customer::getObject

        :param int user_id: Id of the user
        :param string objectmask: default is 'mask[userStatus[name], parent[id, username]]'
        :returns: A user object.
        """
        if objectmask is None:
            objectmask = "mask[userStatus[name], parent[id, username]]"
        return self.user_service.getObject(id=user_id, mask=objectmask)

    def get_current_user(self, objectmask=None):
        """Calls SoftLayer_Account::getCurrentUser"""

        if objectmask is None:
            objectmask = "mask[userStatus[name], parent[id, username]]"
        return self.account_service.getCurrentUser(mask=objectmask)

    def get_all_permissions(self):
        """Calls SoftLayer_User_CustomerPermissions_Permission::getAllObjects

        Stores the result in self.all_permissions
        :returns: A list of dictionaries that contains all valid permissions
        """
        if self.all_permissions is None:
            permissions = self.client.call('User_Customer_CustomerPermission_Permission', 'getAllObjects')
            self.all_permissions = sorted(permissions, key=itemgetter('keyName'))
        return self.all_permissions

    def add_permissions(self, user_id, permissions):
        """Enables a list of permissions for a user

        :param int id: user id to set
        :param list permissions: List of permissions keynames to enable
        :returns: True on success, Exception otherwise

        Example::
            add_permissions(123, ['BANDWIDTH_MANAGE'])
        """
        pretty_permissions = self.format_permission_object(permissions)
        LOGGER.warning("Adding the following permissions to %s: %s", user_id, pretty_permissions)
        return self.user_service.addBulkPortalPermission(pretty_permissions, id=user_id)

    def remove_permissions(self, user_id, permissions):
        """Disables a list of permissions for a user

        :param int id: user id to set
        :param list permissions: List of permissions keynames to disable
        :returns: True on success, Exception otherwise

        Example::
            remove_permissions(123, ['BANDWIDTH_MANAGE'])
        """
        pretty_permissions = self.format_permission_object(permissions)
        LOGGER.warning("Removing the following permissions to %s: %s", user_id, pretty_permissions)
        return self.user_service.removeBulkPortalPermission(pretty_permissions, id=user_id)

    def permissions_from_user(self, user_id, from_user_id):
        """Sets user_id's permission to be the same as from_user_id's

        Any permissions from_user_id has will be added to user_id.
        Any permissions from_user_id doesn't have will be removed from user_id.

        :param int user_id: The user to change permissions.
        :param int from_user_id: The use to base permissions from.
        :returns: True on success, Exception otherwise.
        """

        from_permissions = self.get_user_permissions(from_user_id)
        self.add_permissions(user_id, from_permissions)
        all_permissions = self.get_all_permissions()
        remove_permissions = []

        for permission in all_permissions:
            # If permission does not exist for from_user_id add it to the list to be removed
            if _keyname_search(from_permissions, permission['keyName']):
                continue
            remove_permissions.append({'keyName': permission['keyName']})

        self.remove_permissions(user_id, remove_permissions)
        return True

    def get_user_permissions(self, user_id):
        """Returns a sorted list of a users permissions"""
        permissions = self.user_service.getPermissions(id=user_id)
        return sorted(permissions, key=itemgetter('keyName'))

    def get_logins(self, user_id, start_date=None):
        """Gets the login history for a user, default start_date is 30 days ago

        :param int id: User id to get
        :param string start_date: "%m/%d/%Y %H:%M:%s" formatted string.
        :returns: list https://softlayer.github.io/reference/datatypes/SoftLayer_User_Customer_Access_Authentication/
        Example::
            get_logins(123, '04/08/2018 0:0:0')
        """

        if start_date is None:
            date_object = datetime.datetime.today() - datetime.timedelta(days=30)
            start_date = date_object.strftime("%m/%d/%Y 0:0:0")

        date_filter = {
            'loginAttempts': {
                'createDate': {
                    'operation': 'greaterThanDate',
                    'options': [{'name': 'date', 'value': [start_date]}]
                }
            }
        }
        login_log = self.user_service.getLoginAttempts(id=user_id, filter=date_filter)
        return login_log

    def get_events(self, user_id, start_date=None):
        """Gets the event log for a specific user, default start_date is 30 days ago

        :param int id: User id to view
        :param string start_date: "%Y-%m-%dT%H:%M:%s.0000-06:00" is the full formatted string.
                                  The Timezone part has to be HH:MM, notice the : there.
        :returns: https://softlayer.github.io/reference/datatypes/SoftLayer_Event_Log/
        """

        if start_date is None:
            date_object = datetime.datetime.today() - datetime.timedelta(days=30)
            start_date = date_object.strftime("%Y-%m-%dT00:00:00")

        object_filter = {
            'userId': {
                'operation': user_id
            },
            'eventCreateDate': {
                'operation': 'greaterThanDate',
                'options': [{'name': 'date', 'value': [start_date]}]
            }
        }

        events = self.client.call('Event_Log', 'getAllObjects', filter=object_filter)
        if events is None:
            events = [{'eventName': 'No Events Found'}]
        return events

    def _get_id_from_username(self, username):
        """Looks up a username's id

        :param string username: Username to lookup
        :returns: The id that matches username.
        """
        _mask = "mask[id, username]"
        _filter = {'users': {'username': utils.query_filter(username)}}
        user = self.list_users(_mask, _filter)
        if len(user) == 1:
            return [user[0]['id']]
        elif len(user) > 1:
            raise exceptions.SoftLayerError("Multiple users found with the name: %s" % username)
        else:
            raise exceptions.SoftLayerError("Unable to find user id for %s" % username)

    def format_permission_object(self, permissions):
        """Formats a list of permission key names into something the SLAPI will respect.

        :param list permissions: A list of SLAPI permissions keyNames.
                                 keyName of ALL will return all permissions.
        :returns: list of dictionaries that can be sent to the api to add or remove permissions
        :throws SoftLayerError: If any permission is invalid this exception will be thrown.
        """
        pretty_permissions = []
        available_permissions = self.get_all_permissions()
        # pp(available_permissions)
        for permission in permissions:
            # Handle data retrieved directly from the API
            if isinstance(permission, dict):
                permission = permission['keyName']
            permission = permission.upper()
            if permission == 'ALL':
                return available_permissions
            # Search through available_permissions to make sure what the user entered was valid
            if _keyname_search(available_permissions, permission):
                pretty_permissions.append({'keyName': permission})
            else:
                raise exceptions.SoftLayerError("'%s' is not a valid permission" % permission)
        return pretty_permissions

    def create_user(self, user_object, password):
        """Blindly sends user_object to SoftLayer_User_Customer::createObject

        :param dictionary user_object: https://softlayer.github.io/reference/datatypes/SoftLayer_User_Customer/
        """
        LOGGER.warning("Creating User %s", user_object['username'])

        try:
            return self.user_service.createObject(user_object, password, None)
        except exceptions.SoftLayerAPIError as ex:
            if ex.faultCode == "SoftLayer_Exception_User_Customer_DelegateIamIdInvitationToPaas":
                raise exceptions.SoftLayerError("Your request for a new user was received, but it needs to be "
                                                "processed by the Platform Services API first. Barring any errors on "
                                                "the Platform Services side, your new user should be created shortly.")
            raise

    def edit_user(self, user_id, user_object):
        """Blindly sends user_object to SoftLayer_User_Customer::editObject

        :param int user_id: User to edit
        :param dictionary user_object: https://softlayer.github.io/reference/datatypes/SoftLayer_User_Customer/
        """
        return self.user_service.editObject(user_object, id=user_id)

    def add_api_authentication_key(self, user_id):
        """Calls SoftLayer_User_Customer::addApiAuthenticationKey

        :param int user_id: User to add API key to
        """
        return self.user_service.addApiAuthenticationKey(id=user_id)


def _keyname_search(haystack, needle):
    for item in haystack:
        if item.get('keyName') == needle:
            return True
    return False