"""Orchard Admin (OA) Users. Unlike workstation users, the OA users are not responsible for a particular vendor or subaccount. The dataset Grass cares about is just the user id. """ from ddtrace import tracer from sqlalchemy import Column, Enum, Integer from grass import config from grass.connectors import microservices, mysql from grass.consts import resource, user from grass.utils import response # Deactivated users are considered “Gone” (which corresponds to 410). DEACTIVATED_USER_RESPONSE = response.create_error_response( dict(code='deactivated_user', message='The user is not active anymore.'), status=410 ) class OrchardAdminUser(mysql.BaseModel): """Orchard Admin User. Representation of a Orchard Administrator User. For now, users only contain the `user_id` method. """ __tablename__ = 'orchadmin_users' id = Column(Integer, primary_key=True) # noqa active = Column(Enum('Y', 'N')) orchard_identity_uuid = None roles = [] @property def user_id(self): """Return the user id. The user id is a unique identifier across our different systems. It is composed of the client name (here OA), and the unique id for this user. Returns: str: the user id. """ return f'{user.UserGroups.OA}:{self.id}' @property def user_group(self): """Return the user group. Return: str: the user id. """ return user.UserGroups.OA @property def account_id(arg): """Orchard Admin Users do not have an account id. Returns: None: no account type is associated to a Orchard Admin. """ return @property def account_type(arg): """Orchard Admin Users do not have an account type. Returns: None: no account id is associated to a Orchard Admin. """ return def set_roles(self, roles): """Sets roles attribute for OA user. Converts role names to lowercase. Note that we use OA resources (art_relations.orchadmin_permissions instead of art_relations.orchadmin_roles, as OA permissioning is resource based but we want to keep Roles header name consistent. See oa_user.get_roles_for_user() method for exact query. Args: roles (list): list of OA 'roles' in text format ex. ['publishing', 'quicksearch', 'instantgrat'] """ self.roles = [] for role in roles: self.roles.append(str.lower(role)) def set_identity_uuid(self, identity_uuid): self.orchard_identity_uuid = identity_uuid def set_identity_id(self, identity_id): self.orchard_identity_id = identity_id def set_user_profiles(self, user_id): if user_id.startswith('oa'): self.profile_type = 'OrchAdminProfile' self.profile_id = user_id.replace('oa:', '') @tracer.wrap() def get_oa_user_by_id(user_id): """Get a Orchard Admin User by its id. This method gets the Orchard Admin user and also verifies that the entry in the database matches an “active” Orchard Admin User. If the user is not active, we return a 410 (Gone) response. Args: user_id (str): the user's id. Returns: Response: the user's information. """ user_id = int(user_id.replace('oa:', '')) session = mysql.session() user = ( session.query(OrchardAdminUser).filter(OrchardAdminUser.id == user_id).first() ) session.close() if not user: return response.create_not_found_response() if not is_active(user): return DEACTIVATED_USER_RESPONSE return response.Response(user) @tracer.wrap() def is_active(user): """Verify if a user is active. Args: user (dict, OrchardAdminUser): the user. Returns: bool: if the user is active or not. """ if isinstance(user, OrchardAdminUser): return user.active == 'Y' return user.get('active') == 'Y' @tracer.wrap() def create_oa_user(is_active=False, **data): """Create OA user based on a set of data. Grass should not be able to create OA users. This method is only a helper used in the tests to populate user information. Note: “active” is not a boolean flag in the table. If you want to use a boolean, make sure the value “active” is not present in data, and set the right value on the “is_active” param. It will automatically fill the details for “active”. Args: data (dict): the data related to the user. Returns: Response: the user information and the status of the request. If an error has happened when commiting the information, a 500 is returned. """ assert config.environment == config.TEST_ENVIRONMENT, ( 'You can only create OA users in a test environment. This method ' 'is a helper method for fixtures.' ) if 'active' not in data: data.update(active='Y') if not is_active: data.update(active='N') user = OrchardAdminUser(**data) session = mysql.session() try: session.add(user) session.commit() except: session.rollback() return response.create_fatal_response(dict(error='Unable to create the user')) session.close() return response.Response(user) @tracer.wrap() def get_resource_privileges_for_user(user_id, resource_name): """Get resource privileges for user. Args: user_id (int): the user's id. resource_name (str): the name of the resource. Returns: Response: details of the resource. """ user_id = user_id.replace('oa:', '') connection = mysql.engine.raw_connection() results = [] try: cursor = connection.cursor() cursor.callproc( resource.DB_PRIVILEGES_OA_PROCEDURE_NAME, [user_id, resource_name] ) results = [resource.Resource(*data) for data in cursor.fetchall()] cursor.close() finally: connection.close() return results @tracer.wrap() def get_roles_for_user(user_id): """Get resource permissions for OA user. Note that we use OA resources instead of roles them, as OA permissioning is resource based. Args: user_id (str): oa:. Returns: Response: message payload is a list of all role_ids for user. """ user_id = user_id.replace('oa:', '') roles = dict(role_ids=[], role_names=[]) session = mysql.session() try: sql = """SELECT DISTINCT oar.orchadmin_resource_id, resource FROM orchadmin_user_roles oaur INNER JOIN orchadmin_role_permissions oarp ON oaur.orchadmin_role_id = oarp.orchadmin_role_id AND allow = 'Y' INNER JOIN orchadmin_permissions oap ON oarp.orchadmin_permission_id = oap.orchadmin_permission_id JOIN orchadmin_resources oar ON oap.orchadmin_resource_id = oar.orchadmin_resource_id WHERE oaur.orchadmin_user_id = :user_id""" roles_result = session.execute(sql, {'user_id': user_id}) all_roles = roles_result.fetchall() if all_roles: for role in all_roles: roles['role_ids'].append(role[0]) roles['role_names'].append(role[1]) finally: session.close() return response.Response(message=roles) @tracer.wrap() def get_user_identity(user_id): """Get identity info from ows-users. Args: user_id (int): OA user id. Returns: Response: identity object """ user_id = user_id.replace('oa:', '') result = microservices.request( 'GET', 'ows-users', f'/profile/profile_id/{user_id}/' f'profile_type/OrchAdminProfile/identity', ) if result.status_code == 200: return response.Response(result.json()) else: try: errors = result.json() except ValueError: errors = result.text return response.create_error_response(errors=errors, status=result.status_code) if config.environment == config.TEST_ENVIRONMENT: OrchardAdminUser.__table__.create(mysql.engine)