user.py 12.5 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
    heptapod = attr.ib(repr=False)
Georges Racinet's avatar
Georges Racinet committed
27
    email = attr.ib(default=None, repr=False)
Georges Racinet's avatar
Georges Racinet committed
28
    password = attr.ib(default=None, repr=False)
29
    token = attr.ib(default=None, repr=False)
30
31
32
33
34
35
36
37

    vcs_token_only = attr.ib(default=False, repr=False)
    """Are VCS HTTP operations to be made only with the access token?

    If two-factor authentication is activated, or if the user can only log
    through SSO, only private tokens can enable VCS HTTP push and pull.
    """

38
    _webdriver = attr.ib(default=None, repr=False)
39
    _python_gitlab_client = attr.ib(default=None, repr=False)
40
41

    @classmethod
Georges Racinet's avatar
Georges Racinet committed
42
43
44
45
    def create(cls, heptapod, name,
               password=None,
               email=None,
               fullname=None, **kw):
46
47
48
        if password is None:
            # requirements on password length are also enforced in the API
            password = default_password(name)
49
50
        if fullname is None:
            fullname = name
Georges Racinet's avatar
Georges Racinet committed
51
52
        if email is None:
            email = name + '@heptapod.example.net',
53
54
55

        resp = requests.post(
            '/'.join((heptapod.api_url, 'users')),
56
            data=dict(name=fullname,
57
58
59
                      username=name,
                      password=password,
                      reset_password=False,
Georges Racinet's avatar
Georges Racinet committed
60
                      email=email,
61
                      skip_confirmation=True,
Georges Racinet's avatar
Georges Racinet committed
62
                      **kw
63
64
65
66
                      ),
            headers=heptapod.root_token_headers,
        )
        assert resp.status_code < 300, "User %r creation failed" % name
67
        user = cls(name=name, heptapod=heptapod, password=password,
Georges Racinet's avatar
Georges Racinet committed
68
                   email=email,
69
                   id=resp.json()['id'])
70
71
        user.store_in_heptapod()
        return user
72
73

    @classmethod
74
75
    def init_root(cls, heptapod, password):
        """Singled out because we have to use hardcoded values for root."""
76
77
78
        user = cls(id=1, heptapod=heptapod, name='root', password=password)
        user.store_in_heptapod()
        return user
79

80
81
82
83
84
85
86
87
88
89
90
    @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:
91
            user = cls.search(heptapod, name)
92
            if user is None:
93
94
                user = cls.create(heptapod, name,
                                  password=password, fullname=fullname)
95
96
97
98
99
100
            user.store_in_heptapod()

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

101
    @classmethod
102
    def search(cls, heptapod, name):
103
104
105
106
107
108
        """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).
        """
109
110
111
112
113
114
115
116
117
118
        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
Georges Racinet's avatar
Georges Racinet committed
119
120
121
122
123
        details = found[0]
        return cls(name=name,
                   id=details['id'],
                   email=details['email'],
                   heptapod=heptapod)
124
125

    def store_in_heptapod(self):
126
        self.heptapod.users[self.name] = self
127

128
129
130
    @property
    def webdriver(self):
        """Selenium webdriver, authentified as self.
131
        """
132
133
134
135
136
137
138
139
        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
140
141
142
143
144
145
146
147
148
149
150
151
152
        driver = self._webdriver = self.heptapod.new_webdriver()

        heptapod = self.heptapod
        if self.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
153

154
155
156
157
158
159
160
161
162
163
    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
164
165
166
167
168
169
170
171
172
173
174
175
176
    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)

177
    def delete(self, hard=True):
178
179
        self.close_webdriver()
        # TODO also remove from Heptapod instance
180
181
182
183
184
185
186
187
        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
188

Georges Racinet's avatar
Georges Racinet committed
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
    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

209
    def ensure_ssh_pub_key(self, pubkey, title='heptapod-tests'):
210
211
212
213
214
        """Upload the SSH public key if needed.

        Does not require any special privilege if a token for the user
        is already known
        """
215
        hepta = self.heptapod
216
217
218
219
220
221
222
223
224
        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

225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
        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
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
    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

268
269
270
271
272
273
274
275
276
277
278
279
280
    @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',
281
                         '-i', self.ssh['priv']))
282

Georges Racinet's avatar
Georges Racinet committed
283
284
285
286
287
    def token_headers(self):
        """Return a *new* headers dict suitable for authenticated API acces.
        """
        return {'Private-Token': self.token}

288
289
290
291
292
293
294
295
296
    @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
297
298
299
300
301
302
303
304
305
306
    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,
307
308
309
310
311
312
313
314
315
316
317
318
319
320
                            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
321

322
323
324
325
326
    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.
327
        """
328
329
330
331
332
        resp = self.api_get()
        if resp is not None and resp.status_code == 200:
            logger.info("%r: reusing existing access token.", self)
            return

333
334
        heptapod = self.heptapod
        start = time.time()
335
        webdriver = self.webdriver
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
        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)
353

354
355
356
357
358
359
360
361
362
363
364
365
    @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

366
367
368
369
370
371
    def __enter__(self):
        return self

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