ceph/qa/tasks/mgr/dashboard/test_auth.py

   1 # -*- coding: utf-8 -*-
   2
   3 from __future__ import absolute_import
   4
   5 import time
   6
   7 import jwt
   8 from teuthology.orchestra.run import \
   9     CommandFailedError  # pylint: disable=import-error
  10
  11 from .helper import DashboardTestCase, JObj, JLeaf
  12
  13
  14 class AuthTest(DashboardTestCase):
  15
  16     AUTO_AUTHENTICATE = False
  17
  18     def setUp(self):
  19         super(AuthTest, self).setUp()
  20         self.reset_session()
  21
  22     def _validate_jwt_token(self, token, username, permissions):
  23         payload = jwt.decode(token, options={'verify_signature': False})
  24         self.assertIn('username', payload)
  25         self.assertEqual(payload['username'], username)
  26
  27         for scope, perms in permissions.items():
  28             self.assertIsNotNone(scope)
  29             self.assertIn('read', perms)
  30             self.assertIn('update', perms)
  31             self.assertIn('create', perms)
  32             self.assertIn('delete', perms)
  33
  34     def test_login_without_password(self):
  35         with self.assertRaises(CommandFailedError):
  36             self.create_user('admin2', '', ['administrator'], force_password=True)
  37
  38     def test_a_set_login_credentials(self):
  39         # test with Authorization header
  40         self.create_user('admin2', 'admin2', ['administrator'])
  41         self._post("/api/auth", {'username': 'admin2', 'password': 'admin2'})
  42         self.assertStatus(201)
  43         data = self.jsonBody()
  44         self._validate_jwt_token(data['token'], "admin2", data['permissions'])
  45         self.delete_user('admin2')
  46
  47         # test with Cookies set
  48         self.create_user('admin2', 'admin2', ['administrator'])
  49         self._post("/api/auth", {'username': 'admin2', 'password': 'admin2'}, set_cookies=True)
  50         self.assertStatus(201)
  51         data = self.jsonBody()
  52         self._validate_jwt_token(data['token'], "admin2", data['permissions'])
  53         self.delete_user('admin2')
  54
  55     def test_login_valid(self):
  56         # test with Authorization header
  57         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
  58         self.assertStatus(201)
  59         data = self.jsonBody()
  60         self.assertSchema(data, JObj(sub_elems={
  61             'token': JLeaf(str),
  62             'username': JLeaf(str),
  63             'permissions': JObj(sub_elems={}, allow_unknown=True),
  64             'sso': JLeaf(bool),
  65             'pwdExpirationDate': JLeaf(int, none=True),
  66             'pwdUpdateRequired': JLeaf(bool)
  67         }, allow_unknown=False))
  68         self._validate_jwt_token(data['token'], "admin", data['permissions'])
  69
  70         # test with Cookies set
  71         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
  72         self.assertStatus(201)
  73         data = self.jsonBody()
  74         self.assertSchema(data, JObj(sub_elems={
  75             'token': JLeaf(str),
  76             'username': JLeaf(str),
  77             'permissions': JObj(sub_elems={}, allow_unknown=True),
  78             'sso': JLeaf(bool),
  79             'pwdExpirationDate': JLeaf(int, none=True),
  80             'pwdUpdateRequired': JLeaf(bool)
  81         }, allow_unknown=False))
  82         self._validate_jwt_token(data['token'], "admin", data['permissions'])
  83
  84     def test_login_invalid(self):
  85         # test with Authorization header
  86         self._post("/api/auth", {'username': 'admin', 'password': 'inval'})
  87         self.assertStatus(400)
  88         self.assertJsonBody({
  89             "component": "auth",
  90             "code": "invalid_credentials",
  91             "detail": "Invalid credentials"
  92         })
  93
  94         # test with Cookies set
  95         self._post("/api/auth", {'username': 'admin', 'password': 'inval'}, set_cookies=True)
  96         self.assertStatus(400)
  97         self.assertJsonBody({
  98             "component": "auth",
  99             "code": "invalid_credentials",
 100             "detail": "Invalid credentials"
 101         })
 102
 103     def test_lockout_user(self):
 104         # test with Authorization header
 105         self._ceph_cmd(['dashboard', 'set-account-lockout-attempts', '3'])
 106         for _ in range(3):
 107             self._post("/api/auth", {'username': 'admin', 'password': 'inval'})
 108         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 109         self.assertStatus(400)
 110         self.assertJsonBody({
 111             "component": "auth",
 112             "code": "invalid_credentials",
 113             "detail": "Invalid credentials"
 114         })
 115         self._ceph_cmd(['dashboard', 'ac-user-enable', 'admin'])
 116         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 117         self.assertStatus(201)
 118         data = self.jsonBody()
 119         self.assertSchema(data, JObj(sub_elems={
 120             'token': JLeaf(str),
 121             'username': JLeaf(str),
 122             'permissions': JObj(sub_elems={}, allow_unknown=True),
 123             'sso': JLeaf(bool),
 124             'pwdExpirationDate': JLeaf(int, none=True),
 125             'pwdUpdateRequired': JLeaf(bool)
 126         }, allow_unknown=False))
 127         self._validate_jwt_token(data['token'], "admin", data['permissions'])
 128
 129         # test with Cookies set
 130         self._ceph_cmd(['dashboard', 'set-account-lockout-attempts', '3'])
 131         for _ in range(3):
 132             self._post("/api/auth", {'username': 'admin', 'password': 'inval'}, set_cookies=True)
 133         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
 134         self.assertStatus(400)
 135         self.assertJsonBody({
 136             "component": "auth",
 137             "code": "invalid_credentials",
 138             "detail": "Invalid credentials"
 139         })
 140         self._ceph_cmd(['dashboard', 'ac-user-enable', 'admin'])
 141         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
 142         self.assertStatus(201)
 143         data = self.jsonBody()
 144         self.assertSchema(data, JObj(sub_elems={
 145             'token': JLeaf(str),
 146             'username': JLeaf(str),
 147             'permissions': JObj(sub_elems={}, allow_unknown=True),
 148             'sso': JLeaf(bool),
 149             'pwdExpirationDate': JLeaf(int, none=True),
 150             'pwdUpdateRequired': JLeaf(bool)
 151         }, allow_unknown=False))
 152         self._validate_jwt_token(data['token'], "admin", data['permissions'])
 153
 154     def test_logout(self):
 155         # test with Authorization header
 156         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 157         self.assertStatus(201)
 158         data = self.jsonBody()
 159         self._validate_jwt_token(data['token'], "admin", data['permissions'])
 160         self.set_jwt_token(data['token'])
 161         self._post("/api/auth/logout")
 162         self.assertStatus(200)
 163         self.assertJsonBody({
 164             "redirect_url": "#/login"
 165         })
 166         self._get("/api/host")
 167         self.assertStatus(401)
 168         self.set_jwt_token(None)
 169
 170         # test with Cookies set
 171         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
 172         self.assertStatus(201)
 173         data = self.jsonBody()
 174         self._validate_jwt_token(data['token'], "admin", data['permissions'])
 175         self.set_jwt_token(data['token'])
 176         self._post("/api/auth/logout", set_cookies=True)
 177         self.assertStatus(200)
 178         self.assertJsonBody({
 179             "redirect_url": "#/login"
 180         })
 181         self._get("/api/host", set_cookies=True)
 182         self.assertStatus(401)
 183         self.set_jwt_token(None)
 184
 185     def test_token_ttl(self):
 186         # test with Authorization header
 187         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '5'])
 188         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 189         self.assertStatus(201)
 190         self.set_jwt_token(self.jsonBody()['token'])
 191         self._get("/api/host")
 192         self.assertStatus(200)
 193         time.sleep(6)
 194         self._get("/api/host")
 195         self.assertStatus(401)
 196         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '28800'])
 197         self.set_jwt_token(None)
 198
 199         # test with Cookies set
 200         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '5'])
 201         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
 202         self.assertStatus(201)
 203         self.set_jwt_token(self.jsonBody()['token'])
 204         self._get("/api/host", set_cookies=True)
 205         self.assertStatus(200)
 206         time.sleep(6)
 207         self._get("/api/host", set_cookies=True)
 208         self.assertStatus(401)
 209         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '28800'])
 210         self.set_jwt_token(None)
 211
 212     def test_remove_from_blacklist(self):
 213         # test with Authorization header
 214         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '5'])
 215         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 216         self.assertStatus(201)
 217         self.set_jwt_token(self.jsonBody()['token'])
 218         # the following call adds the token to the blacklist
 219         self._post("/api/auth/logout")
 220         self.assertStatus(200)
 221         self._get("/api/host")
 222         self.assertStatus(401)
 223         time.sleep(6)
 224         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '28800'])
 225         self.set_jwt_token(None)
 226         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 227         self.assertStatus(201)
 228         self.set_jwt_token(self.jsonBody()['token'])
 229         # the following call removes expired tokens from the blacklist
 230         self._post("/api/auth/logout")
 231         self.assertStatus(200)
 232
 233         # test with Cookies set
 234         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '5'])
 235         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
 236         self.assertStatus(201)
 237         self.set_jwt_token(self.jsonBody()['token'])
 238         # the following call adds the token to the blocklist
 239         self._post("/api/auth/logout", set_cookies=True)
 240         self.assertStatus(200)
 241         self._get("/api/host", set_cookies=True)
 242         self.assertStatus(401)
 243         time.sleep(6)
 244         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '28800'])
 245         self.set_jwt_token(None)
 246         self._post("/api/auth", {'username': 'admin', 'password': 'admin'}, set_cookies=True)
 247         self.assertStatus(201)
 248         self.set_jwt_token(self.jsonBody()['token'])
 249         # the following call removes expired tokens from the blocklist
 250         self._post("/api/auth/logout", set_cookies=True)
 251         self.assertStatus(200)
 252
 253     def test_unauthorized(self):
 254         # test with Authorization header
 255         self._get("/api/host")
 256         self.assertStatus(401)
 257
 258         # test with Cookies set
 259         self._get("/api/host", set_cookies=True)
 260         self.assertStatus(401)
 261
 262     def test_invalidate_token_by_admin(self):
 263         # test with Authorization header
 264         self._get("/api/host")
 265         self.assertStatus(401)
 266         self.create_user('user', 'user', ['read-only'])
 267         time.sleep(1)
 268         self._post("/api/auth", {'username': 'user', 'password': 'user'})
 269         self.assertStatus(201)
 270         self.set_jwt_token(self.jsonBody()['token'])
 271         self._get("/api/host")
 272         self.assertStatus(200)
 273         time.sleep(1)
 274         self._ceph_cmd_with_secret(['dashboard', 'ac-user-set-password', '--force-password',
 275                                     'user'],
 276                                    'user2')
 277         time.sleep(1)
 278         self._get("/api/host")
 279         self.assertStatus(401)
 280         self.set_jwt_token(None)
 281         self._post("/api/auth", {'username': 'user', 'password': 'user2'})
 282         self.assertStatus(201)
 283         self.set_jwt_token(self.jsonBody()['token'])
 284         self._get("/api/host")
 285         self.assertStatus(200)
 286         self.delete_user("user")
 287
 288         # test with Cookies set
 289         self._get("/api/host", set_cookies=True)
 290         self.assertStatus(401)
 291         self.create_user('user', 'user', ['read-only'])
 292         time.sleep(1)
 293         self._post("/api/auth", {'username': 'user', 'password': 'user'}, set_cookies=True)
 294         self.assertStatus(201)
 295         self.set_jwt_token(self.jsonBody()['token'])
 296         self._get("/api/host", set_cookies=True)
 297         self.assertStatus(200)
 298         time.sleep(1)
 299         self._ceph_cmd_with_secret(['dashboard', 'ac-user-set-password', '--force-password',
 300                                     'user'],
 301                                    'user2')
 302         time.sleep(1)
 303         self._get("/api/host", set_cookies=True)
 304         self.assertStatus(401)
 305         self.set_jwt_token(None)
 306         self._post("/api/auth", {'username': 'user', 'password': 'user2'}, set_cookies=True)
 307         self.assertStatus(201)
 308         self.set_jwt_token(self.jsonBody()['token'])
 309         self._get("/api/host", set_cookies=True)
 310         self.assertStatus(200)
 311         self.delete_user("user")
 312
 313     def test_check_token(self):
 314         # test with Authorization header
 315         self.login("admin", "admin")
 316         self._post("/api/auth/check", {"token": self.jsonBody()["token"]})
 317         self.assertStatus(200)
 318         data = self.jsonBody()
 319         self.assertSchema(data, JObj(sub_elems={
 320             "username": JLeaf(str),
 321             "permissions": JObj(sub_elems={}, allow_unknown=True),
 322             "sso": JLeaf(bool),
 323             "pwdUpdateRequired": JLeaf(bool)
 324         }, allow_unknown=False))
 325         self.logout()
 326
 327         # test with Cookies set
 328         self.login("admin", "admin", set_cookies=True)
 329         self._post("/api/auth/check", {"token": self.jsonBody()["token"]}, set_cookies=True)
 330         self.assertStatus(200)
 331         data = self.jsonBody()
 332         self.assertSchema(data, JObj(sub_elems={
 333             "username": JLeaf(str),
 334             "permissions": JObj(sub_elems={}, allow_unknown=True),
 335             "sso": JLeaf(bool),
 336             "pwdUpdateRequired": JLeaf(bool)
 337         }, allow_unknown=False))
 338         self.logout(set_cookies=True)
 339
 340     def test_check_wo_token(self):
 341         # test with Authorization header
 342         self.login("admin", "admin")
 343         self._post("/api/auth/check", {"token": ""})
 344         self.assertStatus(200)
 345         data = self.jsonBody()
 346         self.assertSchema(data, JObj(sub_elems={
 347             "login_url": JLeaf(str)
 348         }, allow_unknown=False))
 349         self.logout()
 350
 351         # test with Cookies set
 352         self.login("admin", "admin", set_cookies=True)
 353         self._post("/api/auth/check", {"token": ""}, set_cookies=True)
 354         self.assertStatus(200)
 355         data = self.jsonBody()
 356         self.assertSchema(data, JObj(sub_elems={
 357             "login_url": JLeaf(str)
 358         }, allow_unknown=False))
 359         self.logout(set_cookies=True)