Read about our upcoming Code of Conduct on this issue

user.py 12.2 KB
Newer Older
1
import attr
2
import logging
3
import requests
4
5
import time

6
from gitlab import Gitlab  # python-gitlab
7
8
9
10
11
12
13
from .selenium import (
    webdriver_wait_get,
)
from .session import (
    login_as_root,
    sign_in_page_login,
)
14

15
logger = logging.getLogger(__name__)
16
17
18
19
20
21
22
23
24
25


def default_password(username):
    return username + '.' * (8 - len(username))


@attr.s
class User(object):
    name = attr.ib()
    id = attr.ib()
Georges Racinet's avatar
Georges Racinet committed
26
27
    heptapod = attr.ib(repr=False)
    password = attr.ib(default=None, repr=False)
28
    token = attr.ib(default=None, repr=False)
29
    _webdriver = attr.ib(default=None, repr=False)
30
    _python_gitlab_client = attr.ib(default=None, repr=False)
31
32

    @classmethod
Georges Racinet's avatar
Georges Racinet committed
33
    def create(cls, heptapod, name, password=None, fullname=None, **kw):
34
35
36
        if password is None:
            # requirements on password length are also enforced in the API
            password = default_password(name)
37
38
        if fullname is None:
            fullname = name
39
40
41

        resp = requests.post(
            '/'.join((heptapod.api_url, 'users')),
42
            data=dict(name=fullname,
43
44
45
46
47
                      username=name,
                      password=password,
                      reset_password=False,
                      email=name + '@heptapod.example.net',
                      skip_confirmation=True,
Georges Racinet's avatar
Georges Racinet committed
48
                      **kw
49
50
51
52
                      ),
            headers=heptapod.root_token_headers,
        )
        assert resp.status_code < 300, "User %r creation failed" % name
53
        user = cls(name=name, heptapod=heptapod, password=password,
54
                   id=resp.json()['id'])
55
56
        user.store_in_heptapod()
        return user
57
58

    @classmethod
59
60
    def init_root(cls, heptapod, password):
        """Singled out because we have to use hardcoded values for root."""
61
62
63
        user = cls(id=1, heptapod=heptapod, name='root', password=password)
        user.store_in_heptapod()
        return user
64

65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
    @classmethod
    def ensure(cls, heptapod, name, password=None, fullname=None):
        """Retrieve user or create it.

        :returns: `User` instance
        """
        if password is None:
            password = default_password(password)

        user = heptapod.users.get(name)
        if user is None:
            user = User.search(heptapod, name)
            if user is None:
                user = User.create(heptapod, name,
                                   password=password, fullname=fullname)
            user.store_in_heptapod()

        user.password = password
        user.ensure_private_token()
        return user

86
    @classmethod
87
    def search(cls, heptapod, name):
88
89
90
91
92
93
        """Lookup user in API and returns instance.

        The returned User instance doesn't know its password. Either it
        has an expected value, or it won't be suitable for webdriver
        authentication (API auth is usually done with access tokens).
        """
94
95
96
97
98
99
100
101
102
103
        resp = requests.get(
            '/'.join((heptapod.api_url, 'users')),
            params=dict(username=name),
            headers=heptapod.root_token_headers,
        )
        found = resp.json()
        if resp.status_code == 404 or not found:
            return

        assert len(found) == 1
104
105
106
107
        user = cls(name=name, id=found[0]['id'], heptapod=heptapod)
        return user

    def store_in_heptapod(self):
108
        self.heptapod.users[self.name] = self
109

110
111
112
    def login(self):
        """Perform the webdriver login sequence if not already done.

113
        Assumes :attr:`_webdriver` not to be None
114
        """
115
        heptapod, driver, name = self.heptapod, self._webdriver, self.name
116
117
118
119
120
121
122
123
124
125
126
        if name == 'root':
            login_as_root(driver, heptapod, self.password)
        else:
            start = time.time()
            webdriver_wait_get(heptapod, driver, relative_uri='/users/sign_in')
            sign_in_page_login(driver, heptapod, self.name,
                               password=self.password)
            logger.info("Signed in user %s in %.2f seconds",
                        self.name, time.time() - start)
        return driver

127
128
129
    @property
    def webdriver(self):
        """Selenium webdriver, authentified as self.
130
        """
131
132
133
134
135
136
137
138
139
140
        driver = self._webdriver
        if driver is not None:
            return driver

        logger.info("Initializing a signed-in webdriver for user %r",
                    self.name)

        # guaranteeing driver to be available for teardown as soon as created
        self._webdriver = self.heptapod.new_webdriver()
        return self.login()
141

142
143
144
145
146
147
148
149
150
151
    def close_webdriver(self):
        """Close current web driver if there is one."""
        driver = self._webdriver
        if driver is None:
            return

        logger.info("Closing webdriver for user %r", self.name)
        driver.close()
        self._webdriver = None

Georges Racinet's avatar
Georges Racinet committed
152
153
154
155
156
157
158
159
160
161
162
163
164
    def api_request(self, method, subpath=None, **data):
        """Perform a request on user API as root"""
        hepta = self.heptapod
        segments = [hepta.api_url, 'users', str(self.id)]
        if subpath is not None:
            segments.append(subpath)

        return requests.request(method, '/'.join(segments),
                                headers=hepta.root_token_headers)

    def api_post(self, subpath=None, **data):
        return self.api_request('POST', subpath=subpath, **data)

165
    def delete(self, hard=True):
166
167
        self.close_webdriver()
        # TODO also remove from Heptapod instance
168
169
170
171
172
173
174
175
        hepta = self.heptapod
        resp = requests.delete(
            '/'.join((hepta.api_url, 'users', str(self.id))),
            data=dict(id=self.id, hard_delete=hard),
            headers=hepta.root_token_headers,
        )

        assert resp.status_code == 204, "Failed to delete %r" % self
176

Georges Racinet's avatar
Georges Racinet committed
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
    def edit(self, **data):
        resp = self.api_request('PUT', **data)
        assert resp.status_code == 200, "Failed to edit %r" % self

    def block(self):
        resp = self.api_post(subpath='block')
        assert resp.status_code == 201, "Failed to block %r" % self

    def unblock(self):
        resp = self.api_post(subpath='unblock')
        assert resp.status_code == 201, "Failed to unblock %r" % self

    def deactivate(self):
        resp = self.api_post(subpath='deactivate')
        assert resp.status_code == 201, "Failed to deactivate %r" % self

    def activate(self):
        resp = self.api_post(subpath='activate')
        assert resp.status_code == 201, "Failed to activate %r" % self

197
    def ensure_ssh_pub_key(self, pubkey, title='heptapod-tests'):
198
199
200
201
202
        """Upload the SSH public key if needed.

        Does not require any special privilege if a token for the user
        is already known
        """
203
        hepta = self.heptapod
204
205
206
207
208
209
210
211
212
        if self.token:
            keys_api_url = '/'.join((
                hepta.api_url, 'user', 'keys'))
            headers = self.token_headers()
        else:
            keys_api_url = '/'.join((
                hepta.api_url, 'users', str(self.id), 'keys'))
            headers = hepta.root_token_headers

213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
        ls_resp = requests.get(keys_api_url, headers=headers)
        assert ls_resp.status_code == 200

        # check existence based on title. This is quite lame (we should
        # check fingerprint), but that's good enough for these first tests.
        for key_info in ls_resp.json():
            if key_info['title'] == title:
                return

        create_resp = requests.post(keys_api_url,
                                    data=dict(id=self.id,
                                              title=title,
                                              key=pubkey),
                                    headers=headers,
                                    )
        assert create_resp.status_code == 201
        assert create_resp.json()['title'] == title
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
    def delete_ssh_keys(self):
        """Delete all SSH keys of current user.

        Does not require any special privilege.
        """
        hepta = self.heptapod

        if self.token:
            keys_api_url = '/'.join((
                hepta.api_url, 'user', 'keys'))
            headers = self.token_headers()
        else:
            keys_api_url = '/'.join((
                hepta.api_url, 'users', str(self.id), 'keys'))
            headers = hepta.root_token_headers

        ls_resp = requests.get(keys_api_url, headers=headers)
        assert ls_resp.status_code == 200
        # check existence based on title. This is quite lame (we should
        # check fingerprint), but that's good enough for these first tests.
        for key_info in ls_resp.json():
            resp = requests.delete('%s/%d' % (keys_api_url, key_info['id']),
                                   headers=headers)
            assert resp.status_code == 204

256
257
258
259
260
261
262
263
264
265
266
267
268
    @property
    def ssh_command(self):
        """Return a SSH command to perform operations as the given user.
        """
        # IdentitiesOnly makes sure no other SSH key than the one
        # specified with -i are going to be used. Otherwise, current user
        # SSH keys could be attempted before the correct one, leading
        # to either too much auth failures, or use of the wrong key if
        # one happens to be also known by Heptapod
        return ' '.join(('ssh',
                         '-o', 'IdentitiesOnly=yes',
                         '-o', 'NoHostAuthenticationForLocalhost=yes ',
                         '-o', 'StrictHostKeyChecking=no',
269
                         '-i', self.ssh['priv']))
270

Georges Racinet's avatar
Georges Racinet committed
271
272
273
274
275
    def token_headers(self):
        """Return a *new* headers dict suitable for authenticated API acces.
        """
        return {'Private-Token': self.token}

276
277
278
279
280
281
282
283
284
    @property
    def basic_auth_url(self):
        url = self.heptapod.parsed_url
        return "{scheme}://{auth}@{netloc}".format(
            scheme=url.scheme,
            netloc=url.netloc,
            auth=':'.join((self.name, self.password)),
        )

Georges Racinet's avatar
Georges Racinet committed
285
286
287
288
289
290
291
292
293
294
    def api_get(self):
        """Perform GET on users API for self, using :attr:`token`.

        :return: response or `None` if :attr:`token` is None
        """
        token = self.token
        if token is None:
            return None

        return requests.get(self.heptapod.api_url + '/users/%d' % self.id,
295
296
297
298
299
300
301
302
303
304
305
306
307
308
                            headers=self.token_headers())

    def api_owned_groups(self):
        """List the groups owned by this user.

        TODO it lists actually the first page, which is expected to
        be good enough for the foreseeable future.
        """
        resp = requests.get(self.heptapod.api_url + '/groups',
                            params=dict(owned=True),
                            headers=self.token_headers(),
                            )
        assert resp.status_code < 400
        return resp.json()
Georges Racinet's avatar
Georges Racinet committed
309

310
311
312
313
314
    def ensure_private_token(self):
        """Generate a private token for already logged in user if needed.

        The generation involves obtaining a signed-in webdriver for the user.
        It is therefore costly.
315
        """
316
317
318
319
320
        resp = self.api_get()
        if resp is not None and resp.status_code == 200:
            logger.info("%r: reusing existing access token.", self)
            return

321
322
        heptapod = self.heptapod
        start = time.time()
323
        webdriver = heptapod.get_user_webdriver(self.name)
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
        webdriver.get(heptapod.url + '/profile/personal_access_tokens')
        name_elt = webdriver.find_element_by_xpath(
            '//input[@name="personal_access_token[name]"]')
        name_elt.send_keys("heptapod-tests")

        for scope in webdriver.find_elements_by_xpath(
                '//input[@type="checkbox"]'):
            scope.click()
        submit = webdriver.find_element_by_xpath('//input[@type="submit"]')
        submit.click()

        elem = webdriver.find_element_by_id('created-personal-access-token')
        token = elem.get_attribute('value')
        self.token = token

        logger.info("%r: access token generated in %.2f seconds",
                    self, time.time() - start)
341

342
343
344
345
346
347
348
349
350
351
352
353
    @property
    def python_gitlab_client(self):
        """Return the API client from the python-gitlab library

        See https://python-gitlab.readthedocs.io
        """
        cl = self._python_gitlab_client
        if cl is None:
            cl = self._python_gitlab_client = Gitlab(self.heptapod.url,
                                                     private_token=self.token)
        return cl

354
355
356
357
358
359
    def __enter__(self):
        return self

    def __exit__(self, *exc_args):
        self.delete()
        return False