Coverage for osm_nbi/auth.py: 0%

414 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-04-10 20:04 +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 project_desc["git_name"] = "osm_admin" 

236 pid = self.backend.create_project(project_desc) 

237 self.logger.info( 

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

239 ) 

240 return pid 

241 

242 def create_admin_user(self, project_id): 

243 """ 

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

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

246 """ 

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

248 users = self.backend.get_user_list() 

249 if users: 

250 return None 

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

252 now = time() 

253 user_desc = { 

254 "username": "admin", 

255 "password": "admin", 

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

257 } 

258 if project_id: 

259 pid = project_id 

260 else: 

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

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

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

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

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

266 if pid and roles: 

267 user_desc["project_role_mappings"] = [ 

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

269 ] 

270 uid = self.backend.create_user(user_desc) 

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

272 return uid 

273 

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

275 """ 

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

277 and insert the predefined mappings between roles and permissions. 

278 

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

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

281 """ 

282 

283 records = self.backend.get_role_list() 

284 

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

286 if not records or not next( 

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

288 ): 

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

290 roles_to_operations_yaml = yaml.safe_load(stream) 

291 

292 role_names = [] 

293 for role_with_operations in roles_to_operations_yaml["roles"]: 

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

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

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

297 else: 

298 raise AuthException( 

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

300 role_with_operations["name"], self.roles_to_operations_file 

301 ) 

302 ) 

303 

304 if not role_with_operations["permissions"]: 

305 continue 

306 

307 for permission, is_allowed in role_with_operations[ 

308 "permissions" 

309 ].items(): 

310 if not isinstance(is_allowed, bool): 

311 raise AuthException( 

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

313 permission, 

314 role_with_operations["name"], 

315 self.roles_to_operations_file, 

316 ) 

317 ) 

318 

319 # TODO check permission is ok 

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

321 raise AuthException( 

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

323 permission, 

324 role_with_operations["name"], 

325 self.roles_to_operations_file, 

326 ) 

327 ) 

328 

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

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

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

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

333 

334 now = time() 

335 role_with_operations["_admin"] = { 

336 "created": now, 

337 "modified": now, 

338 } 

339 

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

341 try: 

342 self.backend.create_role(role_with_operations) 

343 self.logger.info( 

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

345 ) 

346 except (AuthException, AuthconnException) as e: 

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

348 raise 

349 self.logger.error( 

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

351 role_with_operations["name"], e 

352 ) 

353 ) 

354 

355 # Create admin project&user if required 

356 pid = self.create_admin_project() 

357 user_id = self.create_admin_user(pid) 

358 

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

360 if not user_id: 

361 try: 

362 users = self.backend.get_user_list() 

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

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

365 user_with_system_admin = False 

366 user_admin_id = None 

367 for user in users: 

368 if not user_admin_id: 

369 user_admin_id = user["_id"] 

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

371 user_admin_id = user["_id"] 

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

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

374 user_with_system_admin = True 

375 break 

376 if user_with_system_admin: 

377 break 

378 if not user_with_system_admin: 

379 self.backend.update_user( 

380 { 

381 "_id": user_admin_id, 

382 "add_project_role_mappings": [ 

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

384 ], 

385 } 

386 ) 

387 self.logger.info( 

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

389 user_admin_id 

390 ) 

391 ) 

392 except Exception as e: 

393 self.logger.error( 

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

395 type(e).__name__, e 

396 ) 

397 ) 

398 

399 self.load_operation_to_allowed_roles() 

400 

401 def load_operation_to_allowed_roles(self): 

402 """ 

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

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

405 :return: None 

406 """ 

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

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

409 records = self.backend.get_role_list() 

410 

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

412 for record in records: 

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

414 continue 

415 record_permissions = { 

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

417 for oper in self.role_permissions 

418 } 

419 operations_joined = [ 

420 (oper, value) 

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

422 if oper not in ignore_fields 

423 ] 

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

425 

426 for oper in operations_joined: 

427 match = list( 

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

429 ) 

430 

431 for m in match: 

432 record_permissions[m] = oper[1] 

433 

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

435 

436 for allowed_op in allowed_operations: 

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

438 

439 self.operation_to_allowed_roles = permissions 

440 

441 def authorize( 

442 self, role_permission=None, query_string_operations=None, item_id=None 

443 ): 

444 token = None 

445 user_passwd64 = None 

446 try: 

447 # 1. Get token Authorization bearer 

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

449 if auth: 

450 auth_list = auth.split(" ") 

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

452 token = auth_list[-1] 

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

454 user_passwd64 = auth_list[-1] 

455 if not token: 

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

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

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

459 "Authorization" 

460 ) 

461 if token == "logout": 

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

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

464 "auth.allow_basic_authentication" 

465 ): 

466 # 3. Get new token from user password 

467 user = None 

468 passwd = None 

469 try: 

470 user_passwd = standard_b64decode(user_passwd64).decode() 

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

472 except Exception: 

473 pass 

474 outdata = self.new_token( 

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

476 ) 

477 token = outdata["_id"] 

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

479 

480 if not token: 

481 raise AuthException( 

482 "Needed a token or Authorization http header", 

483 http_code=HTTPStatus.UNAUTHORIZED, 

484 ) 

485 

486 # try to get from cache first 

487 now = time() 

488 token_info = self.tokens_cache.get(token) 

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

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

491 self.tokens_cache.pop(token, None) 

492 token_info = None 

493 

494 # get from database if not in cache 

495 if not token_info: 

496 token_info = self.backend.validate_token(token) 

497 # Clear cache if token limit reached 

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

499 self.tokens_cache.clear() 

500 self.tokens_cache[token] = token_info 

501 # TODO add to token info remote host, port 

502 

503 if role_permission: 

504 RBAC_auth = self.check_permissions( 

505 token_info, 

506 cherrypy.request.method, 

507 role_permission, 

508 query_string_operations, 

509 item_id, 

510 ) 

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

512 if RBAC_auth: 

513 cef_event( 

514 self.cef_logger, 

515 { 

516 "name": "System Access", 

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

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

519 token_info.get("project_name") 

520 ), 

521 }, 

522 ) 

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

524 token_info["allow_show_user_project_role"] = RBAC_auth 

525 

526 return token_info 

527 except AuthException as e: 

528 if not isinstance(e, AuthExceptionUnauthorized): 

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

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

531 cherrypy.response.headers[ 

532 "WWW-Authenticate" 

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

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

535 return { 

536 "id": "testing-token", 

537 "_id": "testing-token", 

538 "project_id": self.test_project_id, 

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

540 "roles": [self.system_admin_role_id], 

541 "admin": True, 

542 "allow_show_user_project_role": True, 

543 } 

544 raise 

545 

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

547 if indata.get("email_id"): 

548 return self.backend.send_email(indata) 

549 else: 

550 if indata.get("otp"): 

551 otp_validation = self.backend.validate_otp(indata) 

552 if otp_validation.get("password_change"): 

553 new_token_info = self.backend.authenticate( 

554 credentials=indata, 

555 token_info=token_info, 

556 ) 

557 new_token_info["otp"] = "valid" 

558 else: 

559 otp_validation["otp"] = "invalid" 

560 return otp_validation 

561 else: 

562 new_token_info = self.backend.authenticate( 

563 credentials=indata, 

564 token_info=token_info, 

565 ) 

566 

567 new_token_info["remote_port"] = remote.port 

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

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

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

571 new_token_info["admin"] = ( 

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

573 ) 

574 # TODO put admin in RBAC 

575 

576 if remote.name: 

577 new_token_info["remote_host"] = remote.name 

578 elif remote.ip: 

579 new_token_info["remote_host"] = remote.ip 

580 

581 # TODO call self._internal_tokens_prune(now) ? 

582 return deepcopy(new_token_info) 

583 

584 def get_token_list(self, token_info): 

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

586 return self._internal_get_token_list(token_info) 

587 else: 

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

589 return [ 

590 deepcopy(token) 

591 for token in self.tokens_cache.values() 

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

593 ] 

594 

595 def get_token(self, token_info, token): 

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

597 return self._internal_get_token(token_info, token) 

598 else: 

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

600 token_value = self.tokens_cache.get(token) 

601 if not token_value: 

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

603 if ( 

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

605 and not token_info["admin"] 

606 ): 

607 raise AuthException( 

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

609 ) 

610 return token_value 

611 

612 def del_token(self, token): 

613 try: 

614 self.backend.revoke_token(token) 

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

616 self.remove_token_from_cache(token) 

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

618 except KeyError: 

619 raise AuthException( 

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

621 ) 

622 

623 def check_permissions( 

624 self, 

625 token_info, 

626 method, 

627 role_permission=None, 

628 query_string_operations=None, 

629 item_id=None, 

630 ): 

631 """ 

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

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

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

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

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

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

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

639 assigned role allows this query string for this method 

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

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

642 :raises: AuthExceptionUnauthorized if access denied 

643 """ 

644 self.load_operation_to_allowed_roles() 

645 

646 roles_required = self.operation_to_allowed_roles[role_permission] 

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

648 

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

650 token_info["admin"] = False 

651 for role in roles_allowed: 

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

653 token_info["admin"] = True 

654 break 

655 

656 if "anonymous" in roles_required: 

657 return True 

658 operation_allowed = False 

659 for role in roles_allowed: 

660 if role in roles_required: 

661 operation_allowed = True 

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

663 if not query_string_operations: 

664 return True 

665 for query_string_operation in query_string_operations: 

666 if ( 

667 role 

668 not in self.operation_to_allowed_roles[query_string_operation] 

669 ): 

670 break 

671 else: 

672 return True 

673 

674 # Bug 853 - Final Solution 

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

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

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

678 if ( 

679 role_permission 

680 in [ 

681 "projects:get", 

682 "projects:id:get", 

683 "roles:get", 

684 "roles:id:get", 

685 "users:get", 

686 ] 

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

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

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

690 return False 

691 

692 if not operation_allowed: 

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

694 else: 

695 raise AuthExceptionUnauthorized( 

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

697 ) 

698 

699 def get_user_list(self): 

700 return self.backend.get_user_list() 

701 

702 def _normalize_url(self, url, method): 

703 # DEPRECATED !!! 

704 # Removing query strings 

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

706 normalized_url_splitted = normalized_url.split("/") 

707 parameters = {} 

708 

709 filtered_keys = [ 

710 key 

711 for key in self.resources_to_operations_mapping.keys() 

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

713 ] 

714 

715 for idx, path_part in enumerate(normalized_url_splitted): 

716 tmp_keys = [] 

717 for tmp_key in filtered_keys: 

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

719 if idx >= len(splitted): 

720 continue 

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

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

723 tmp_keys.append(tmp_key) 

724 continue 

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

726 normalized_url_splitted 

727 ) != len(splitted): 

728 continue 

729 else: 

730 tmp_keys.append(tmp_key) 

731 elif splitted[idx] == path_part: 

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

733 normalized_url_splitted 

734 ) != len(splitted): 

735 continue 

736 else: 

737 tmp_keys.append(tmp_key) 

738 filtered_keys = tmp_keys 

739 if ( 

740 len(filtered_keys) == 1 

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

742 ): 

743 break 

744 

745 if len(filtered_keys) == 0: 

746 raise AuthException( 

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

748 url 

749 ) 

750 ) 

751 elif len(filtered_keys) > 1: 

752 raise AuthException( 

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

754 url 

755 ) 

756 ) 

757 

758 filtered_key = filtered_keys[0] 

759 

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

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

762 if path_part == "<artifactPath>": 

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

764 normalized_url_splitted[idx:] 

765 ) 

766 else: 

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

768 

769 return filtered_key, parameters 

770 

771 def _internal_get_token_list(self, token_info): 

772 now = time() 

773 token_list = self.db.get_list( 

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

775 ) 

776 return token_list 

777 

778 def _internal_get_token(self, token_info, token_id): 

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

780 if not token_value: 

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

782 if ( 

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

784 and not token_info["admin"] 

785 ): 

786 raise AuthException( 

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

788 ) 

789 return token_value 

790 

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

792 now = now or time() 

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

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

795 self.next_db_prune_time = self.periodin_db_pruning + now 

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

797 

798 def remove_token_from_cache(self, token=None): 

799 if token: 

800 self.tokens_cache.pop(token, None) 

801 else: 

802 self.tokens_cache.clear() 

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

804 

805 def check_password_expiry(self, outdata): 

806 """ 

807 This method will check for password expiry of the user 

808 :param outdata: user token information 

809 """ 

810 user_list = None 

811 present_time = time() 

812 user = outdata["username"] 

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

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

815 if user_list: 

816 user_content = user_list[0] 

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

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

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

820 password_expire_time = user_content["_admin"][ 

821 "password_expire_time" 

822 ] 

823 else: 

824 password_expire_time = present_time 

825 uid = user_content["_id"] 

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

827 if not present_time < password_expire_time: 

828 return True 

829 else: 

830 pass