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