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 tasks.mgr.dashboard.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         self.create_user('admin2', 'admin2', ['administrator'])
  34         self._post("/api/auth", {'username': 'admin2', 'password': 'admin2'})
  35         self.assertStatus(201)
  36         data = self.jsonBody()
  37         self._validate_jwt_token(data['token'], "admin2", data['permissions'])
  38         self.delete_user('admin2')
  39
  40     def test_login_valid(self):
  41         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
  42         self.assertStatus(201)
  43         data = self.jsonBody()
  44         self.assertSchema(data, JObj(sub_elems={
  45             'token': JLeaf(str),
  46             'username': JLeaf(str),
  47             'permissions': JObj(sub_elems={}, allow_unknown=True),
  48             'sso': JLeaf(bool),
  49             'pwdExpirationDate': JLeaf(int, none=True),
  50             'pwdUpdateRequired': JLeaf(bool)
  51         }, allow_unknown=False))
  52         self._validate_jwt_token(data['token'], "admin", data['permissions'])
  53
  54     def test_login_invalid(self):
  55         self._post("/api/auth", {'username': 'admin', 'password': 'inval'})
  56         self.assertStatus(400)
  57         self.assertJsonBody({
  58             "component": "auth",
  59             "code": "invalid_credentials",
  60             "detail": "Invalid credentials"
  61         })
  62
  63     def test_login_without_password(self):
  64         self.create_user('admin2', '', ['administrator'])
  65         self._post("/api/auth", {'username': 'admin2', 'password': ''})
  66         self.assertStatus(400)
  67         self.assertJsonBody({
  68             "component": "auth",
  69             "code": "invalid_credentials",
  70             "detail": "Invalid credentials"
  71         })
  72         self.delete_user('admin2')
  73
  74     def test_logout(self):
  75         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
  76         self.assertStatus(201)
  77         data = self.jsonBody()
  78         self._validate_jwt_token(data['token'], "admin", data['permissions'])
  79         self.set_jwt_token(data['token'])
  80         self._post("/api/auth/logout")
  81         self.assertStatus(200)
  82         self.assertJsonBody({
  83             "redirect_url": "#/login"
  84         })
  85         self._get("/api/host")
  86         self.assertStatus(401)
  87         self.set_jwt_token(None)
  88
  89     def test_token_ttl(self):
  90         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '5'])
  91         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
  92         self.assertStatus(201)
  93         self.set_jwt_token(self.jsonBody()['token'])
  94         self._get("/api/host")
  95         self.assertStatus(200)
  96         time.sleep(6)
  97         self._get("/api/host")
  98         self.assertStatus(401)
  99         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '28800'])
 100         self.set_jwt_token(None)
 101
 102     def test_remove_from_blacklist(self):
 103         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '5'])
 104         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 105         self.assertStatus(201)
 106         self.set_jwt_token(self.jsonBody()['token'])
 107         # the following call adds the token to the blacklist
 108         self._post("/api/auth/logout")
 109         self.assertStatus(200)
 110         self._get("/api/host")
 111         self.assertStatus(401)
 112         time.sleep(6)
 113         self._ceph_cmd(['dashboard', 'set-jwt-token-ttl', '28800'])
 114         self.set_jwt_token(None)
 115         self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
 116         self.assertStatus(201)
 117         self.set_jwt_token(self.jsonBody()['token'])
 118         # the following call removes expired tokens from the blacklist
 119         self._post("/api/auth/logout")
 120         self.assertStatus(200)
 121
 122     def test_unauthorized(self):
 123         self._get("/api/host")
 124         self.assertStatus(401)
 125
 126     def test_invalidate_token_by_admin(self):
 127         self._get("/api/host")
 128         self.assertStatus(401)
 129         self.create_user('user', 'user', ['read-only'])
 130         time.sleep(1)
 131         self._post("/api/auth", {'username': 'user', 'password': 'user'})
 132         self.assertStatus(201)
 133         self.set_jwt_token(self.jsonBody()['token'])
 134         self._get("/api/host")
 135         self.assertStatus(200)
 136         time.sleep(1)
 137         self._ceph_cmd(['dashboard', 'ac-user-set-password', '--force-password',
 138                         'user', 'user2'])
 139         time.sleep(1)
 140         self._get("/api/host")
 141         self.assertStatus(401)
 142         self.set_jwt_token(None)
 143         self._post("/api/auth", {'username': 'user', 'password': 'user2'})
 144         self.assertStatus(201)
 145         self.set_jwt_token(self.jsonBody()['token'])
 146         self._get("/api/host")
 147         self.assertStatus(200)
 148         self.delete_user("user")
 149
 150     def test_check_token(self):
 151         self.login("admin", "admin")
 152         self._post("/api/auth/check", {"token": self.jsonBody()["token"]})
 153         self.assertStatus(200)
 154         data = self.jsonBody()
 155         self.assertSchema(data, JObj(sub_elems={
 156             "username": JLeaf(str),
 157             "permissions": JObj(sub_elems={}, allow_unknown=True),
 158             "sso": JLeaf(bool),
 159             "pwdUpdateRequired": JLeaf(bool)
 160         }, allow_unknown=False))
 161         self.logout()
 162
 163     def test_check_wo_token(self):
 164         self.login("admin", "admin")
 165         self._post("/api/auth/check", {"token": ""})
 166         self.assertStatus(200)
 167         data = self.jsonBody()
 168         self.assertSchema(data, JObj(sub_elems={
 169             "login_url": JLeaf(str)
 170         }, allow_unknown=False))
 171         self.logout()