X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FNBI.git;a=blobdiff_plain;f=osm_nbi%2Fauth.py;h=dc3d386ff03599dd8fc53dc9deb884fec23708b7;hp=fde7455a74c5a3aecc069718bd3ef8415a82947b;hb=45bd94c2a096f53a5692f438aa5148c3d42631fa;hpb=1546f2a46d99a4741b23857e6ceb4b813223e297 diff --git a/osm_nbi/auth.py b/osm_nbi/auth.py index fde7455..dc3d386 100644 --- a/osm_nbi/auth.py +++ b/osm_nbi/auth.py @@ -39,14 +39,14 @@ from http import HTTPStatus from time import time from os import path -from authconn import AuthException, AuthExceptionUnauthorized -from authconn_keystone import AuthconnKeystone -from authconn_internal import AuthconnInternal # Comment out for testing&debugging, uncomment when ready +from osm_nbi.authconn import AuthException, AuthExceptionUnauthorized +from osm_nbi.authconn_keystone import AuthconnKeystone +from osm_nbi.authconn_internal import AuthconnInternal # Comment out for testing&debugging, uncomment when ready from osm_common import dbmongo from osm_common import dbmemory from osm_common.dbbase import DbException +from osm_nbi.validation import is_valid_uuid from itertools import chain - from uuid import uuid4 @@ -214,7 +214,7 @@ class Authenticator: # Loading permissions to MongoDB if there is not any permission. if not records or (len(records) == 1 and records[0]["name"] == "admin"): with open(self.roles_to_operations_file, "r") as stream: - roles_to_operations_yaml = yaml.load(stream) + roles_to_operations_yaml = yaml.load(stream, Loader=yaml.Loader) role_names = [] for role_with_operations in roles_to_operations_yaml["roles"]: @@ -257,13 +257,31 @@ class Authenticator: # Create admin project&user if required pid = self.create_admin_project() - self.create_admin_user(pid) + user_id = self.create_admin_user(pid) - # self.backend.update_user({"_id": "admin", - # "add_project_role_mappings": {"project": "admin", "role": "system_admin"}}) - if self.config["authentication"]["backend"] == "keystone": + # try to assign system_admin role to user admin if not any user has this role + if not user_id: try: - self.backend.assign_role_to_user("admin", "admin", "system_admin") + users = self.backend.get_user_list() + roles = self.backend.get_role_list({"name": "system_admin"}) + role_id = roles[0]["_id"] + user_with_system_admin = False + user_admin_id = None + for user in users: + if not user_admin_id: + user_admin_id = user["_id"] + if user["username"] == "admin": + user_admin_id = user["_id"] + for prm in user.get("project_role_mappings", ()): + if prm["role"] == role_id: + user_with_system_admin = True + break + if user_with_system_admin: + break + if not user_with_system_admin: + self.backend.update_user({"_id": user_admin_id, + "add_project_role_mappings": [{"project": pid, "role": role_id}]}) + self.logger.info("Added role system admin to user='{}' project=admin".format(user_admin_id)) except Exception: pass @@ -302,7 +320,7 @@ class Authenticator: self.operation_to_allowed_roles = permissions - def authorize(self, role_permission=None, query_string_operations=None): + def authorize(self, role_permission=None, query_string_operations=None, item_id=None): token = None user_passwd64 = None try: @@ -340,14 +358,22 @@ class Authenticator: # TODO add to token info remote host, port if role_permission: - self.check_permissions(token_info, cherrypy.request.method, role_permission, - query_string_operations) + RBAC_auth = self.check_permissions(token_info, cherrypy.request.method, role_permission, + query_string_operations, item_id) + token_info["allow_show_user_project_role"] = RBAC_auth + return token_info except AuthException as e: if not isinstance(e, AuthExceptionUnauthorized): if cherrypy.session.get('Authorization'): del cherrypy.session['Authorization'] cherrypy.response.headers["WWW-Authenticate"] = 'Bearer realm="{}"'.format(e) + elif self.config.get("user_not_authorized"): + # TODO provide user_id, roles id (not name), project_id + return {"id": "fake-token-id-for-test", + "project_id": self.config.get("project_not_authorized", "admin"), + "username": self.config["user_not_authorized"], + "roles": ["system_admin"]} raise def new_token(self, token_info, indata, remote): @@ -403,7 +429,7 @@ class Authenticator: except KeyError: raise AuthException("Token '{}' not found".format(token), http_code=HTTPStatus.NOT_FOUND) - def check_permissions(self, token_info, method, role_permission=None, query_string_operations=None): + def check_permissions(self, token_info, method, role_permission=None, query_string_operations=None, item_id=None): """ Checks that operation has permissions to be done, base on the assigned roles to this user project :param token_info: Dictionary that contains "roles" with a list of assigned roles. @@ -413,7 +439,9 @@ class Authenticator: :param role_permission: role permission name of the operation required :param query_string_operations: list of possible admin query strings provided by user. It is checked that the assigned role allows this query string for this method - :return: None if granted, exception if not allowed + :param item_id: item identifier if included in the URL, None otherwise + :return: True if access granted by permission rules, False if access granted by default rules (Bug 853) + :raises: AuthExceptionUnauthorized if access denied """ roles_required = self.operation_to_allowed_roles[role_permission] @@ -427,19 +455,29 @@ class Authenticator: break if "anonymous" in roles_required: - return + return True operation_allowed = False for role in roles_allowed: if role in roles_required: operation_allowed = True # if query_string operations, check if this role allows it if not query_string_operations: - return + return True for query_string_operation in query_string_operations: if role not in self.operation_to_allowed_roles[query_string_operation]: break else: - return + return True + + # Bug 853 - Final Solution + # User/Project/Role whole listings are filtered elsewhere + # uid, pid, rid = ("user_id", "project_id", "id") if is_valid_uuid(id) else ("username", "project_name", "name") + uid = "user_id" if is_valid_uuid(item_id) else "username" + if (role_permission in ["projects:get", "projects:id:get", "roles:get", "roles:id:get", "users:get"]) \ + or (role_permission == "users:id:get" and item_id == token_info[uid]): + # or (role_permission == "projects:id:get" and item_id == token_info[pid]) \ + # or (role_permission == "roles:id:get" and item_id in [role[rid] for role in token_info["roles"]]): + return False if not operation_allowed: raise AuthExceptionUnauthorized("Access denied: lack of permissions.")