Coverage for osm_nbi/auth.py: 0%

405 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2024-06-30 10:14 +0000

1# -*- coding: utf-8 -*- 

2 

3# Copyright 2018 Whitestack, LLC 

4# Copyright 2018 Telefonica S.A. 

5# 

6# Licensed under the Apache License, Version 2.0 (the "License"); you may 

7# not use this file except in compliance with the License. You may obtain 

8# a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, software 

13# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

14# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

15# License for the specific language governing permissions and limitations 

16# under the License. 

17# 

18# For those usages not covered by the Apache License, Version 2.0 please 

19# contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com 

20## 

21 

22 

23""" 

24Authenticator is responsible for authenticating the users, 

25create the tokens unscoped and scoped, retrieve the role 

26list inside the projects that they are inserted 

27""" 

28 

29__author__ = "Eduardo Sousa <esousa@whitestack.com>; Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>" 

30__date__ = "$27-jul-2018 23:59:59$" 

31 

32import cherrypy 

33import logging 

34import yaml 

35from base64 import standard_b64decode 

36from copy import deepcopy 

37 

38# from functools import reduce 

39from http import HTTPStatus 

40from time import time 

41from os import path 

42 

43from osm_nbi.authconn import AuthException, AuthconnException, AuthExceptionUnauthorized 

44from osm_nbi.authconn_keystone import AuthconnKeystone 

45from osm_nbi.authconn_internal import AuthconnInternal 

46from osm_nbi.authconn_tacacs import AuthconnTacacs 

47from osm_nbi.utils import cef_event, cef_event_builder 

48from osm_common import dbmemory, dbmongo, msglocal, msgkafka 

49from osm_common.dbbase import DbException 

50from osm_nbi.validation import is_valid_uuid 

51from itertools import chain 

52from uuid import uuid4 

53 

54 

55class Authenticator: 

56 """ 

57 This class should hold all the mechanisms for User Authentication and 

58 Authorization. Initially it should support Openstack Keystone as a 

59 backend through a plugin model where more backends can be added and a 

60 RBAC model to manage permissions on operations. 

61 This class must be threading safe 

62 """ 

63 

64 periodin_db_pruning = ( 

65 60 * 30 

66 ) # for the internal backend only. every 30 minutes expired tokens will be pruned 

67 token_limit = 500 # when reached, the token cache will be cleared 

68 

69 def __init__(self, valid_methods, valid_query_string): 

70 """ 

71 Authenticator initializer. Setup the initial state of the object, 

72 while it waits for the config dictionary and database initialization. 

73 """ 

74 self.backend = None 

75 self.config = None 

76 self.db = None 

77 self.msg = None 

78 self.tokens_cache = dict() 

79 self.next_db_prune_time = ( 

80 0 # time when next cleaning of expired tokens must be done 

81 ) 

82 self.roles_to_operations_file = None 

83 # self.roles_to_operations_table = None 

84 self.resources_to_operations_mapping = {} 

85 self.operation_to_allowed_roles = {} 

86 self.logger = logging.getLogger("nbi.authenticator") 

87 self.role_permissions = [] 

88 self.valid_methods = valid_methods 

89 self.valid_query_string = valid_query_string 

90 self.system_admin_role_id = None # system_role id 

91 self.test_project_id = None # test_project_id 

92 self.cef_logger = None 

93 

94 def start(self, config): 

95 """ 

96 Method to configure the Authenticator object. This method should be called 

97 after object creation. It is responsible by initializing the selected backend, 

98 as well as the initialization of the database connection. 

99 

100 :param config: dictionary containing the relevant parameters for this object. 

101 """ 

102 self.config = config 

103 self.cef_logger = cef_event_builder(config["authentication"]) 

104 

105 try: 

106 if not self.db: 

107 if config["database"]["driver"] == "mongo": 

108 self.db = dbmongo.DbMongo() 

109 self.db.db_connect(config["database"]) 

110 elif config["database"]["driver"] == "memory": 

111 self.db = dbmemory.DbMemory() 

112 self.db.db_connect(config["database"]) 

113 else: 

114 raise AuthException( 

115 "Invalid configuration param '{}' at '[database]':'driver'".format( 

116 config["database"]["driver"] 

117 ) 

118 ) 

119 if not self.msg: 

120 if config["message"]["driver"] == "local": 

121 self.msg = msglocal.MsgLocal() 

122 self.msg.connect(config["message"]) 

123 elif config["message"]["driver"] == "kafka": 

124 self.msg = msgkafka.MsgKafka() 

125 self.msg.connect(config["message"]) 

126 else: 

127 raise AuthException( 

128 "Invalid configuration param '{}' at '[message]':'driver'".format( 

129 config["message"]["driver"] 

130 ) 

131 ) 

132 if not self.backend: 

133 if config["authentication"]["backend"] == "keystone": 

134 self.backend = AuthconnKeystone( 

135 self.config["authentication"], self.db, self.role_permissions 

136 ) 

137 elif config["authentication"]["backend"] == "internal": 

138 self.backend = AuthconnInternal( 

139 self.config["authentication"], self.db, self.role_permissions 

140 ) 

141 self._internal_tokens_prune("tokens") 

142 elif config["authentication"]["backend"] == "tacacs": 

143 self.backend = AuthconnTacacs( 

144 self.config["authentication"], self.db, self.role_permissions 

145 ) 

146 self._internal_tokens_prune("tokens_tacacs") 

147 else: 

148 raise AuthException( 

149 "Unknown authentication backend: {}".format( 

150 config["authentication"]["backend"] 

151 ) 

152 ) 

153 

154 if not self.roles_to_operations_file: 

155 if "roles_to_operations" in config["rbac"]: 

156 self.roles_to_operations_file = config["rbac"][ 

157 "roles_to_operations" 

158 ] 

159 else: 

160 possible_paths = ( 

161 __file__[: __file__.rfind("auth.py")] 

162 + "roles_to_operations.yml", 

163 "./roles_to_operations.yml", 

164 ) 

165 for config_file in possible_paths: 

166 if path.isfile(config_file): 

167 self.roles_to_operations_file = config_file 

168 break 

169 if not self.roles_to_operations_file: 

170 raise AuthException( 

171 "Invalid permission configuration: roles_to_operations file missing" 

172 ) 

173 

174 # load role_permissions 

175 def load_role_permissions(method_dict): 

176 for k in method_dict: 

177 if k == "ROLE_PERMISSION": 

178 for method in chain( 

179 method_dict.get("METHODS", ()), method_dict.get("TODO", ()) 

180 ): 

181 permission = method_dict["ROLE_PERMISSION"] + method.lower() 

182 if permission not in self.role_permissions: 

183 self.role_permissions.append(permission) 

184 elif k in ("TODO", "METHODS"): 

185 continue 

186 elif method_dict[k]: 

187 load_role_permissions(method_dict[k]) 

188 

189 load_role_permissions(self.valid_methods) 

190 for query_string in self.valid_query_string: 

191 for method in ("get", "put", "patch", "post", "delete"): 

192 permission = query_string.lower() + ":" + method 

193 if permission not in self.role_permissions: 

194 self.role_permissions.append(permission) 

195 

196 # get ids of role system_admin and test project 

197 role_system_admin = self.db.get_one( 

198 "roles", {"name": "system_admin"}, fail_on_empty=False 

199 ) 

200 if role_system_admin: 

201 self.system_admin_role_id = role_system_admin["_id"] 

202 test_project_name = self.config["authentication"].get( 

203 "project_not_authorized", "admin" 

204 ) 

205 test_project = self.db.get_one( 

206 "projects", {"name": test_project_name}, fail_on_empty=False 

207 ) 

208 if test_project: 

209 self.test_project_id = test_project["_id"] 

210 

211 except Exception as e: 

212 raise AuthException(str(e)) 

213 

214 def stop(self): 

215 try: 

216 if self.db: 

217 self.db.db_disconnect() 

218 except DbException as e: 

219 raise AuthException(str(e), http_code=e.http_code) 

220 

221 def create_admin_project(self): 

222 """ 

223 Creates a new project 'admin' into database if it doesn't exist. Useful for initialization. 

224 :return: _id identity of the 'admin' project 

225 """ 

226 

227 # projects = self.db.get_one("projects", fail_on_empty=False, fail_on_more=False) 

228 project_desc = {"name": "admin"} 

229 projects = self.backend.get_project_list(project_desc) 

230 if projects: 

231 return projects[0]["_id"] 

232 now = time() 

233 project_desc["_id"] = str(uuid4()) 

234 project_desc["_admin"] = {"created": now, "modified": now} 

235 pid = self.backend.create_project(project_desc) 

236 self.logger.info( 

237 "Project '{}' created at database".format(project_desc["name"]) 

238 ) 

239 return pid 

240 

241 def create_admin_user(self, project_id): 

242 """ 

243 Creates a new user admin/admin into database if database is empty. Useful for initialization 

244 :return: _id identity of the inserted data, or None 

245 """ 

246 # users = self.db.get_one("users", fail_on_empty=False, fail_on_more=False) 

247 users = self.backend.get_user_list() 

248 if users: 

249 return None 

250 # user_desc = {"username": "admin", "password": "admin", "projects": [project_id]} 

251 now = time() 

252 user_desc = { 

253 "username": "admin", 

254 "password": "admin", 

255 "_admin": {"created": now, "modified": now, "user_status": "always-active"}, 

256 } 

257 if project_id: 

258 pid = project_id 

259 else: 

260 # proj = self.db.get_one("projects", {"name": "admin"}, fail_on_empty=False, fail_on_more=False) 

261 proj = self.backend.get_project_list({"name": "admin"}) 

262 pid = proj[0]["_id"] if proj else None 

263 # role = self.db.get_one("roles", {"name": "system_admin"}, fail_on_empty=False, fail_on_more=False) 

264 roles = self.backend.get_role_list({"name": "system_admin"}) 

265 if pid and roles: 

266 user_desc["project_role_mappings"] = [ 

267 {"project": pid, "role": roles[0]["_id"]} 

268 ] 

269 uid = self.backend.create_user(user_desc) 

270 self.logger.info("User '{}' created at database".format(user_desc["username"])) 

271 return uid 

272 

273 def init_db(self, target_version="1.0"): 

274 """ 

275 Check if the database has been initialized, with at least one user. If not, create the required tables 

276 and insert the predefined mappings between roles and permissions. 

277 

278 :param target_version: schema version that should be present in the database. 

279 :return: None if OK, exception if error or version is different. 

280 """ 

281 

282 records = self.backend.get_role_list() 

283 

284 # Loading permissions to AUTH. At lease system_admin must be present. 

285 if not records or not next( 

286 (r for r in records if r["name"] == "system_admin"), None 

287 ): 

288 with open(self.roles_to_operations_file, "r") as stream: 

289 roles_to_operations_yaml = yaml.safe_load(stream) 

290 

291 role_names = [] 

292 for role_with_operations in roles_to_operations_yaml["roles"]: 

293 # Verifying if role already exists. If it does, raise exception 

294 if role_with_operations["name"] not in role_names: 

295 role_names.append(role_with_operations["name"]) 

296 else: 

297 raise AuthException( 

298 "Duplicated role name '{}' at file '{}''".format( 

299 role_with_operations["name"], self.roles_to_operations_file 

300 ) 

301 ) 

302 

303 if not role_with_operations["permissions"]: 

304 continue 

305 

306 for permission, is_allowed in role_with_operations[ 

307 "permissions" 

308 ].items(): 

309 if not isinstance(is_allowed, bool): 

310 raise AuthException( 

311 "Invalid value for permission '{}' at role '{}'; at file '{}'".format( 

312 permission, 

313 role_with_operations["name"], 

314 self.roles_to_operations_file, 

315 ) 

316 ) 

317 

318 # TODO check permission is ok 

319 if permission[-1] == ":": 

320 raise AuthException( 

321 "Invalid permission '{}' terminated in ':' for role '{}'; at file {}".format( 

322 permission, 

323 role_with_operations["name"], 

324 self.roles_to_operations_file, 

325 ) 

326 ) 

327 

328 if "default" not in role_with_operations["permissions"]: 

329 role_with_operations["permissions"]["default"] = False 

330 if "admin" not in role_with_operations["permissions"]: 

331 role_with_operations["permissions"]["admin"] = False 

332 

333 now = time() 

334 role_with_operations["_admin"] = { 

335 "created": now, 

336 "modified": now, 

337 } 

338 

339 # self.db.create(self.roles_to_operations_table, role_with_operations) 

340 try: 

341 self.backend.create_role(role_with_operations) 

342 self.logger.info( 

343 "Role '{}' created".format(role_with_operations["name"]) 

344 ) 

345 except (AuthException, AuthconnException) as e: 

346 if role_with_operations["name"] == "system_admin": 

347 raise 

348 self.logger.error( 

349 "Role '{}' cannot be created: {}".format( 

350 role_with_operations["name"], e 

351 ) 

352 ) 

353 

354 # Create admin project&user if required 

355 pid = self.create_admin_project() 

356 user_id = self.create_admin_user(pid) 

357 

358 # try to assign system_admin role to user admin if not any user has this role 

359 if not user_id: 

360 try: 

361 users = self.backend.get_user_list() 

362 roles = self.backend.get_role_list({"name": "system_admin"}) 

363 role_id = roles[0]["_id"] 

364 user_with_system_admin = False 

365 user_admin_id = None 

366 for user in users: 

367 if not user_admin_id: 

368 user_admin_id = user["_id"] 

369 if user["username"] == "admin": 

370 user_admin_id = user["_id"] 

371 for prm in user.get("project_role_mappings", ()): 

372 if prm["role"] == role_id: 

373 user_with_system_admin = True 

374 break 

375 if user_with_system_admin: 

376 break 

377 if not user_with_system_admin: 

378 self.backend.update_user( 

379 { 

380 "_id": user_admin_id, 

381 "add_project_role_mappings": [ 

382 {"project": pid, "role": role_id} 

383 ], 

384 } 

385 ) 

386 self.logger.info( 

387 "Added role system admin to user='{}' project=admin".format( 

388 user_admin_id 

389 ) 

390 ) 

391 except Exception as e: 

392 self.logger.error( 

393 "Error in Authorization DataBase initialization: {}: {}".format( 

394 type(e).__name__, e 

395 ) 

396 ) 

397 

398 self.load_operation_to_allowed_roles() 

399 

400 def load_operation_to_allowed_roles(self): 

401 """ 

402 Fills the internal self.operation_to_allowed_roles based on database role content and self.role_permissions 

403 It works in a shadow copy and replace at the end to allow other threads working with the old copy 

404 :return: None 

405 """ 

406 permissions = {oper: [] for oper in self.role_permissions} 

407 # records = self.db.get_list(self.roles_to_operations_table) 

408 records = self.backend.get_role_list() 

409 

410 ignore_fields = ["_id", "_admin", "name", "default"] 

411 for record in records: 

412 if not record.get("permissions"): 

413 continue 

414 record_permissions = { 

415 oper: record["permissions"].get("default", False) 

416 for oper in self.role_permissions 

417 } 

418 operations_joined = [ 

419 (oper, value) 

420 for oper, value in record["permissions"].items() 

421 if oper not in ignore_fields 

422 ] 

423 operations_joined.sort(key=lambda x: x[0].count(":")) 

424 

425 for oper in operations_joined: 

426 match = list( 

427 filter(lambda x: x.find(oper[0]) == 0, record_permissions.keys()) 

428 ) 

429 

430 for m in match: 

431 record_permissions[m] = oper[1] 

432 

433 allowed_operations = [k for k, v in record_permissions.items() if v is True] 

434 

435 for allowed_op in allowed_operations: 

436 permissions[allowed_op].append(record["name"]) 

437 

438 self.operation_to_allowed_roles = permissions 

439 

440 def authorize( 

441 self, role_permission=None, query_string_operations=None, item_id=None 

442 ): 

443 token = None 

444 user_passwd64 = None 

445 try: 

446 # 1. Get token Authorization bearer 

447 auth = cherrypy.request.headers.get("Authorization") 

448 if auth: 

449 auth_list = auth.split(" ") 

450 if auth_list[0].lower() == "bearer": 

451 token = auth_list[-1] 

452 elif auth_list[0].lower() == "basic": 

453 user_passwd64 = auth_list[-1] 

454 if not token: 

455 if cherrypy.session.get("Authorization"): # pylint: disable=E1101 

456 # 2. Try using session before request a new token. If not, basic authentication will generate 

457 token = cherrypy.session.get( # pylint: disable=E1101 

458 "Authorization" 

459 ) 

460 if token == "logout": 

461 token = None # force Unauthorized response to insert user password again 

462 elif user_passwd64 and cherrypy.request.config.get( 

463 "auth.allow_basic_authentication" 

464 ): 

465 # 3. Get new token from user password 

466 user = None 

467 passwd = None 

468 try: 

469 user_passwd = standard_b64decode(user_passwd64).decode() 

470 user, _, passwd = user_passwd.partition(":") 

471 except Exception: 

472 pass 

473 outdata = self.new_token( 

474 None, {"username": user, "password": passwd}, None 

475 ) 

476 token = outdata["_id"] 

477 cherrypy.session["Authorization"] = token # pylint: disable=E1101 

478 

479 if not token: 

480 raise AuthException( 

481 "Needed a token or Authorization http header", 

482 http_code=HTTPStatus.UNAUTHORIZED, 

483 ) 

484 

485 # try to get from cache first 

486 now = time() 

487 token_info = self.tokens_cache.get(token) 

488 if token_info and token_info["expires"] < now: 

489 # delete token. MUST be done with care, as another thread maybe already delete it. Do not use del 

490 self.tokens_cache.pop(token, None) 

491 token_info = None 

492 

493 # get from database if not in cache 

494 if not token_info: 

495 token_info = self.backend.validate_token(token) 

496 # Clear cache if token limit reached 

497 if len(self.tokens_cache) > self.token_limit: 

498 self.tokens_cache.clear() 

499 self.tokens_cache[token] = token_info 

500 # TODO add to token info remote host, port 

501 

502 if role_permission: 

503 RBAC_auth = self.check_permissions( 

504 token_info, 

505 cherrypy.request.method, 

506 role_permission, 

507 query_string_operations, 

508 item_id, 

509 ) 

510 self.logger.info("RBAC_auth: {}".format(RBAC_auth)) 

511 if RBAC_auth: 

512 cef_event( 

513 self.cef_logger, 

514 { 

515 "name": "System Access", 

516 "sourceUserName": token_info.get("username"), 

517 "message": "Accessing account with system privileges, Project={}".format( 

518 token_info.get("project_name") 

519 ), 

520 }, 

521 ) 

522 self.logger.info("{}".format(self.cef_logger)) 

523 token_info["allow_show_user_project_role"] = RBAC_auth 

524 

525 return token_info 

526 except AuthException as e: 

527 if not isinstance(e, AuthExceptionUnauthorized): 

528 if cherrypy.session.get("Authorization"): # pylint: disable=E1101 

529 del cherrypy.session["Authorization"] # pylint: disable=E1101 

530 cherrypy.response.headers[ 

531 "WWW-Authenticate" 

532 ] = 'Bearer realm="{}"'.format(e) 

533 if self.config["authentication"].get("user_not_authorized"): 

534 return { 

535 "id": "testing-token", 

536 "_id": "testing-token", 

537 "project_id": self.test_project_id, 

538 "username": self.config["authentication"]["user_not_authorized"], 

539 "roles": [self.system_admin_role_id], 

540 "admin": True, 

541 "allow_show_user_project_role": True, 

542 } 

543 raise 

544 

545 def new_token(self, token_info, indata, remote): 

546 new_token_info = self.backend.authenticate( 

547 credentials=indata, 

548 token_info=token_info, 

549 ) 

550 

551 new_token_info["remote_port"] = remote.port 

552 if not new_token_info.get("expires"): 

553 new_token_info["expires"] = time() + 3600 

554 if not new_token_info.get("admin"): 

555 new_token_info["admin"] = ( 

556 True if new_token_info.get("project_name") == "admin" else False 

557 ) 

558 # TODO put admin in RBAC 

559 

560 if remote.name: 

561 new_token_info["remote_host"] = remote.name 

562 elif remote.ip: 

563 new_token_info["remote_host"] = remote.ip 

564 

565 # TODO call self._internal_tokens_prune(now) ? 

566 return deepcopy(new_token_info) 

567 

568 def get_token_list(self, token_info): 

569 if self.config["authentication"]["backend"] == "internal": 

570 return self._internal_get_token_list(token_info) 

571 else: 

572 # TODO: check if this can be avoided. Backend may provide enough information 

573 return [ 

574 deepcopy(token) 

575 for token in self.tokens_cache.values() 

576 if token["username"] == token_info["username"] 

577 ] 

578 

579 def get_token(self, token_info, token): 

580 if self.config["authentication"]["backend"] == "internal": 

581 return self._internal_get_token(token_info, token) 

582 else: 

583 # TODO: check if this can be avoided. Backend may provide enough information 

584 token_value = self.tokens_cache.get(token) 

585 if not token_value: 

586 raise AuthException("token not found", http_code=HTTPStatus.NOT_FOUND) 

587 if ( 

588 token_value["username"] != token_info["username"] 

589 and not token_info["admin"] 

590 ): 

591 raise AuthException( 

592 "needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED 

593 ) 

594 return token_value 

595 

596 def del_token(self, token): 

597 try: 

598 self.backend.revoke_token(token) 

599 # self.tokens_cache.pop(token, None) 

600 self.remove_token_from_cache(token) 

601 return "token '{}' deleted".format(token) 

602 except KeyError: 

603 raise AuthException( 

604 "Token '{}' not found".format(token), http_code=HTTPStatus.NOT_FOUND 

605 ) 

606 

607 def check_permissions( 

608 self, 

609 token_info, 

610 method, 

611 role_permission=None, 

612 query_string_operations=None, 

613 item_id=None, 

614 ): 

615 """ 

616 Checks that operation has permissions to be done, base on the assigned roles to this user project 

617 :param token_info: Dictionary that contains "roles" with a list of assigned roles. 

618 This method fills the token_info["admin"] with True or False based on assigned tokens, if any allows admin 

619 This will be used among others to hide or not the _admin content of topics 

620 :param method: GET,PUT, POST, ... 

621 :param role_permission: role permission name of the operation required 

622 :param query_string_operations: list of possible admin query strings provided by user. It is checked that the 

623 assigned role allows this query string for this method 

624 :param item_id: item identifier if included in the URL, None otherwise 

625 :return: True if access granted by permission rules, False if access granted by default rules (Bug 853) 

626 :raises: AuthExceptionUnauthorized if access denied 

627 """ 

628 self.load_operation_to_allowed_roles() 

629 

630 roles_required = self.operation_to_allowed_roles[role_permission] 

631 roles_allowed = [role["name"] for role in token_info["roles"]] 

632 

633 # fills token_info["admin"] if some roles allows it 

634 token_info["admin"] = False 

635 for role in roles_allowed: 

636 if role in self.operation_to_allowed_roles["admin:" + method.lower()]: 

637 token_info["admin"] = True 

638 break 

639 

640 if "anonymous" in roles_required: 

641 return True 

642 operation_allowed = False 

643 for role in roles_allowed: 

644 if role in roles_required: 

645 operation_allowed = True 

646 # if query_string operations, check if this role allows it 

647 if not query_string_operations: 

648 return True 

649 for query_string_operation in query_string_operations: 

650 if ( 

651 role 

652 not in self.operation_to_allowed_roles[query_string_operation] 

653 ): 

654 break 

655 else: 

656 return True 

657 

658 # Bug 853 - Final Solution 

659 # User/Project/Role whole listings are filtered elsewhere 

660 # uid, pid, rid = ("user_id", "project_id", "id") if is_valid_uuid(id) else ("username", "project_name", "name") 

661 uid = "user_id" if is_valid_uuid(item_id) else "username" 

662 if ( 

663 role_permission 

664 in [ 

665 "projects:get", 

666 "projects:id:get", 

667 "roles:get", 

668 "roles:id:get", 

669 "users:get", 

670 ] 

671 ) or (role_permission == "users:id:get" and item_id == token_info[uid]): 

672 # or (role_permission == "projects:id:get" and item_id == token_info[pid]) \ 

673 # or (role_permission == "roles:id:get" and item_id in [role[rid] for role in token_info["roles"]]): 

674 return False 

675 

676 if not operation_allowed: 

677 raise AuthExceptionUnauthorized("Access denied: lack of permissions.") 

678 else: 

679 raise AuthExceptionUnauthorized( 

680 "Access denied: You have not permissions to use these admin query string" 

681 ) 

682 

683 def get_user_list(self): 

684 return self.backend.get_user_list() 

685 

686 def _normalize_url(self, url, method): 

687 # DEPRECATED !!! 

688 # Removing query strings 

689 normalized_url = url if "?" not in url else url[: url.find("?")] 

690 normalized_url_splitted = normalized_url.split("/") 

691 parameters = {} 

692 

693 filtered_keys = [ 

694 key 

695 for key in self.resources_to_operations_mapping.keys() 

696 if method in key.split()[0] 

697 ] 

698 

699 for idx, path_part in enumerate(normalized_url_splitted): 

700 tmp_keys = [] 

701 for tmp_key in filtered_keys: 

702 splitted = tmp_key.split()[1].split("/") 

703 if idx >= len(splitted): 

704 continue 

705 elif "<" in splitted[idx] and ">" in splitted[idx]: 

706 if splitted[idx] == "<artifactPath>": 

707 tmp_keys.append(tmp_key) 

708 continue 

709 elif idx == len(normalized_url_splitted) - 1 and len( 

710 normalized_url_splitted 

711 ) != len(splitted): 

712 continue 

713 else: 

714 tmp_keys.append(tmp_key) 

715 elif splitted[idx] == path_part: 

716 if idx == len(normalized_url_splitted) - 1 and len( 

717 normalized_url_splitted 

718 ) != len(splitted): 

719 continue 

720 else: 

721 tmp_keys.append(tmp_key) 

722 filtered_keys = tmp_keys 

723 if ( 

724 len(filtered_keys) == 1 

725 and filtered_keys[0].split("/")[-1] == "<artifactPath>" 

726 ): 

727 break 

728 

729 if len(filtered_keys) == 0: 

730 raise AuthException( 

731 "Cannot make an authorization decision. URL not found. URL: {0}".format( 

732 url 

733 ) 

734 ) 

735 elif len(filtered_keys) > 1: 

736 raise AuthException( 

737 "Cannot make an authorization decision. Multiple URLs found. URL: {0}".format( 

738 url 

739 ) 

740 ) 

741 

742 filtered_key = filtered_keys[0] 

743 

744 for idx, path_part in enumerate(filtered_key.split()[1].split("/")): 

745 if "<" in path_part and ">" in path_part: 

746 if path_part == "<artifactPath>": 

747 parameters[path_part[1:-1]] = "/".join( 

748 normalized_url_splitted[idx:] 

749 ) 

750 else: 

751 parameters[path_part[1:-1]] = normalized_url_splitted[idx] 

752 

753 return filtered_key, parameters 

754 

755 def _internal_get_token_list(self, token_info): 

756 now = time() 

757 token_list = self.db.get_list( 

758 "tokens", {"username": token_info["username"], "expires.gt": now} 

759 ) 

760 return token_list 

761 

762 def _internal_get_token(self, token_info, token_id): 

763 token_value = self.db.get_one("tokens", {"_id": token_id}, fail_on_empty=False) 

764 if not token_value: 

765 raise AuthException("token not found", http_code=HTTPStatus.NOT_FOUND) 

766 if ( 

767 token_value["username"] != token_info["username"] 

768 and not token_info["admin"] 

769 ): 

770 raise AuthException( 

771 "needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED 

772 ) 

773 return token_value 

774 

775 def _internal_tokens_prune(self, token_collection, now=None): 

776 now = now or time() 

777 if not self.next_db_prune_time or self.next_db_prune_time >= now: 

778 self.db.del_list(token_collection, {"expires.lt": now}) 

779 self.next_db_prune_time = self.periodin_db_pruning + now 

780 # self.tokens_cache.clear() # not required any more 

781 

782 def remove_token_from_cache(self, token=None): 

783 if token: 

784 self.tokens_cache.pop(token, None) 

785 else: 

786 self.tokens_cache.clear() 

787 self.msg.write("admin", "revoke_token", {"_id": token} if token else None) 

788 

789 def check_password_expiry(self, outdata): 

790 """ 

791 This method will check for password expiry of the user 

792 :param outdata: user token information 

793 """ 

794 user_list = None 

795 present_time = time() 

796 user = outdata["username"] 

797 if self.config["authentication"].get("user_management"): 

798 user_list = self.db.get_list("users", {"username": user}) 

799 if user_list: 

800 user_content = user_list[0] 

801 if not user_content.get("username") == "admin": 

802 user_content["_admin"]["modified"] = present_time 

803 if user_content.get("_admin").get("password_expire_time"): 

804 password_expire_time = user_content["_admin"][ 

805 "password_expire_time" 

806 ] 

807 else: 

808 password_expire_time = present_time 

809 uid = user_content["_id"] 

810 self.db.set_one("users", {"_id": uid}, user_content) 

811 if not present_time < password_expire_time: 

812 return True 

813 else: 

814 pass