"""Logic for auth0 related functions.""" import datetime import json import random import re from typing import Optional from auth0.v3 import Auth0Error from auth0.v3.authentication import Database, GetToken, revoke_token from auth0.v3.management import Auth0 from auth0.v3.management.guardian import Guardian from auth0.v3.management.users import Users from auth0.v3.management.users_by_email import UsersByEmail import boto3 from flask import g from owsresponse import response from users import config, constants from users.connectors import redis from users.connectors.sentry import sentry_client from users.logic import profiles, user_info as user_info_logic from users.models import identities, user_info def gen_password(): """Generate a random password that will pass Auth0. Must be at least 10 characters including at least 3 of the following 4 types of characters: a lower-case letter an upper-case letter a number a special character (such as !@#$%^&*) Not more than 2 identical characters in a row (such as 111 is not allowed). Params: None Returns: A random 13 character string """ alphalower = 'abcdefghijklmnopqrstuvwxyz' alphaupper = 'ABCDEFGHIZKLMNOPQRSTUVWXYZ' nums = '1234567890' special = '!@#$%^&*' password = [] for _ in range(random.randint(12, 15)): password.append(random.choice(alphalower)) for _ in range(random.randint(2, 3)): password.append(random.choice(alphaupper)) for _ in range(random.randint(2, 3)): password.append(random.choice(nums)) for _ in range(random.randint(2, 3)): password.append(random.choice(special)) random.shuffle(password) password = ''.join(password) if re.search(r'(\w)\1+', password): password = re.sub(r'(\w)\1+', r'\1', password) return password def get_auth0_management_handle(): """Create an Auth0 Management Handle. Params: None Returns: An Auth0 handler object attached to auth0 configured environment """ get_token = GetToken(config.AUTH0_DOMAIN) token = get_token.client_credentials( config.AUTH0_MACHINE_CLIENT_ID, config.AUTH0_MACHINE_CLIENT_SECRET, config.AUTH0_URL ) access_token = token['access_token'] return Auth0(config.AUTH0_DOMAIN, access_token) def send_password_reset( email, connection=config.AUTH0_CONNECTION, client_id=config.AUTH0_MACHINE_CLIENT_ID ): """Send a password reset notification. This is used in the process of the creation of a new user as well as password recovery. Params: email (str): The email belonging to the user connection (str): The name of the Auth0 connection defaulting to the environments connection client_id (str): The client_id the email comes from Returns: A success message directly from Auth0 """ auth0_db = Database(config.AUTH0_DOMAIN) result = auth0_db.change_password(client_id, email, connection) return result def resend_verify_email(user_id): """Trigger Verify email. Params: user_id (str): Auth0 user id. Returns: A success message directly from Auth0 """ auth0 = get_auth0_management_handle() try: result = auth0.jobs.send_verification_email({'user_id': user_id}) return response.Response(result) except Auth0Error as errors: g.ows.log.error(str(errors)) return response.create_fatal_response(errors.message) def bulk_password_reset(user_ids): """Bulk password reset to auth0 user_ids. Password reset call needs email address, but we have auth0 ids. So make a list call and get email address for all auth0 ids, and then call send password reset. No bulk password reset call so it is in-loop. Params: user_ids (list): Auth0 user ids. Returns: A success message directly from Auth0 """ auth0 = get_auth0_management_handle() try: result = auth0.users.list( q='user_id:({})'.format(' OR '.join(user_ids)), search_engine='v3', fields=['email'] ) if not result or not result.get('users'): return response.Response({'message': 'No users found with these ids.'}, status=204) for user_details in result.get('users'): user_details['result'] = send_password_reset(user_details['email']) return response.Response(result) except Auth0Error as errors: g.ows.log.error(str(errors)) return response.create_fatal_response(errors.message) def create_user(data, audit_user=None): """Create a new Auth0 User for any connection. By default this method will send both an email verification email and a password reset email. Params: data (dict): The dictionary of data required to create a user audit_user (str): Identity id of the user for audit. Example of Data: { 'user_id': '1234', # optional 'name': 'larmstead@theorchard.com', # defaults to email 'email': 'larmstead@theorchard.com', # required 'password': '1234abcD$', # will be generated if not sent 'user_metadata': { 'username': 'larmstead@theorchard.com', 'vend_contact_id': '1234', 'type': 'alw' # can also be OA }, 'app_metadata': {}, 'email_verified': False, 'verify_email': False, 'reset_password': True, 'connection': 'art-relations' # required } Returns: flask.Response containing the new user object from Auth0 """ if 'email' not in data: return response.Response({'message': 'The email key is required.'}, status=400) auth0 = get_auth0_management_handle() if audit_user: if 'user_metadata' not in data: data['user_metadata'] = {} data['user_metadata']['audit_user'] = audit_user userData = { 'password': gen_password(), 'connection': config.AUTH0_CONNECTION, 'name': data['email'], 'verify_email': False, 'email_verified': True, 'reset_password': True, } userData.update(data) reset_password, email_client_id = _extract_email_info_from_create_data(userData) try: user = auth0.users.create(userData) if reset_password: send_password_reset(userData['email'], userData['connection'], email_client_id) return response.Response(user, status=201) except Auth0Error as err: if err.status_code == 409: return response.create_error_response( constants.ERROR_CODE_ALREADY_EXISTS, 'User already exists with this email.' ) raise err def read_user(user_id): """Read user information from Auth0. Params: user_id (str): The id of the user Returns: flask.Response containg the user object from Auth0 """ auth0 = get_auth0_management_handle() try: user = auth0.users.get(user_id) return response.Response(user, status=200) except Exception: return response.Response({'message': 'User does not exist.'}, status=404) def update_user(user_id, data): """Update an Auth0 User. Refer to the user creation commentary for information on data type Additionally the field "blocked": true controls the Auth0 block Must pass the Auth0 "connection" field in the data object it is defauled to "art-relations" Params: data (dict): The dictionary of data to update the user with Returns: flask.Response containing the updated user object from Auth0 """ auth0 = get_auth0_management_handle() userData = {'connection': config.AUTH0_CONNECTION} userData.update(data) if 'blocked' in userData and userData['blocked']: dt = datetime.datetime.now().isoformat() try: userData['user_metadata']['blocked_at'] = dt except Exception: userData['user_metadata'] = {'blocked_at': dt} elif 'blocked' in userData and not userData['blocked']: try: userData['user_metadata']['blocked_at'] = None except Exception: userData['user_metadata'] = {'blocked_at': None} if 'email' in userData: # Set `email_verified` to True if email was changed userData['email_verified'] = True try: user = auth0.users.update(user_id, userData) orchard_identity_id = user.get('user_metadata', {}).get('orchardIdentityId') if ('email' in userData or 'name' in userData) and orchard_identity_id: try: neo4j_update = { 'email': userData.get('email', user['email']), 'name': user.get('name', user['name']), } vend_contact_details = {} if 'first_name' in user['user_metadata'] and 'last_name' in user['user_metadata']: first_name = user['user_metadata'].get('first_name') last_name = user['user_metadata'].get('last_name') neo4j_update.update(first_name=first_name, last_name=last_name) vend_contact_details.setdefault('contact', {}).setdefault( 'first_name', first_name ) vend_contact_details.setdefault('contact', {}).setdefault( 'last_name', last_name ) raw_auth0_id = user_id.split('auth0|')[1] primary_user = user_info.fetch_primary_for_auth0_user(raw_auth0_id) if primary_user.status == 200: user_id = primary_user.message['user_id'].split(':')[1] primary_user_vendor_id = primary_user.message['account']['vendor_id'] vend_contact_details.setdefault('contact', {}).setdefault( 'email', neo4j_update['email'] ) vend_contact_details.setdefault( 'login', f'{primary_user_vendor_id}_{neo4j_update["email"]}' ) user_info_logic.update_vend_contact_user(user_id, vend_contact_details) neo4j_update_result = identities.update_identity(orchard_identity_id, neo4j_update) if not neo4j_update_result: raise Exception(neo4j_update_result.errors) except Exception: # ideally this should not happen, but it will in qa where # auth0 and neo4j ids are not in sync. # Log it so we can monitor it. sentry_client.capture_exception() return response.Response(user, status=200) except Exception as e: return response.Response({'message': e.message}, status=e.status_code) def delete_user(user_id): """Delete an Auth0 User. Params: user_id (str): The id of the user to remove Returns: flask.Response contaning an empty dictionary and a status of 204 """ auth0 = get_auth0_management_handle() auth0.users.delete(user_id) return response.Response({}, status=204) def get_auth0_users(email): """Get auth0 users that match this email. Params: email (str): Email address. Returns: flask.Response: containing user metadata. """ get_token = GetToken(config.AUTH0_DOMAIN) token = get_token.client_credentials( config.AUTH0_MACHINE_CLIENT_ID, config.AUTH0_MACHINE_CLIENT_SECRET, config.AUTH0_URL ) user_obj = UsersByEmail(config.AUTH0_DOMAIN, token['access_token']) fields = ('user_id', 'email_verified', 'identities') # this returns users for all connections under that tenant. result = user_obj.search_users_by_email(email, fields) users_in_connection = [ each_user['user_id'] for each_user in result if [ db['connection'] for db in each_user['identities'] if db['connection'] == config.AUTH0_CONNECTION ] ] if not users_in_connection: g.ows.log.warning(constants.WARNING_MESSAGE_EMPTY_RESPONSE.format('get_auth0_users')) return response.Response({'result': users_in_connection}) def get_auth0_users_paginated(page=0, per_page=100, q=None): """Get a page of auth0 users.. Params: page: Zero-indexed page to fetch per_page: Number of items per page (max 100) q: Lucene syntax query for filtering users by attributes Returns: flask.Response: contains the page of users. """ get_token = GetToken(config.AUTH0_DOMAIN) token = get_token.client_credentials( config.AUTH0_MACHINE_CLIENT_ID, config.AUTH0_MACHINE_CLIENT_SECRET, config.AUTH0_URL ) users_obj = Users(config.AUTH0_DOMAIN, token['access_token']) try: result = users_obj.list(page=page, per_page=per_page, q=q, search_engine='v3') return response.Response(result['users']) except Auth0Error as exc: sentry_client.capture_exception() return response.create_error_response( code=exc.error_code, message=exc.message, status=exc.status_code ) def get_bulk_auth0_users_paginated(auth0_user_ids): """Get a page of auth0 users using auth0 search. Params: auth0_user_ids: list of auth0 user_ids to search Returns: flask.Response: contains the page of users. """ user_ids = ['user_id:auth0|{}'.format(_id) for _id in auth0_user_ids] query = ' OR '.join(user_ids) result = get_auth0_users_paginated(q=query) if not result.message: g.ows.log.warning( constants.WARNING_MESSAGE_EMPTY_RESPONSE.format('get_bulk_paginated_auth0_users') ) return result def send_sqs_message(user_id, email, auth0_id, labels=[]): """Send notification for Auth0 single sign on. Args: user_id (str): alw:Vend contact id for now. email (str): auth0 user email. auth0_id (str): Auth0 user id. labels (list): List of label names. Returns: flask.Response: Sqs queue request and response. """ sqs_msg = { 'user_ids': [user_id], 'users_info': [ { 'user_id': user_id, 'auth0_id': auth0_id, 'email': email, } ], 'feed_name': constants.SSO_NOTIFICATION_FEED_NAME, 'feed_id': 'sso_{}'.format(auth0_id), 'template': constants.SSO_NOTIFICATION_TEMPLATE, 'payload': {'label_names': labels}, } client = boto3.client('sqs', region_name=config.AWS_REGION) result = client.send_message( QueueUrl=config.DAEMON_NOTIFICATIONS_SQS_URL, MessageBody=json.dumps(sqs_msg), ) return response.Response({'request': sqs_msg, 'response': result}) def get_guardian_enrollments_for_user(auth0_user_id: str) -> response.Response: """Get a list of devices enrolled for guardian mfa for a user. Args: auth0_user_id (str): Auth0 user id to get devices for. Returns: flask.Response: Message contains list of devices. """ get_token = GetToken(config.AUTH0_DOMAIN) token = get_token.client_credentials( config.AUTH0_MACHINE_CLIENT_ID, config.AUTH0_MACHINE_CLIENT_SECRET, config.AUTH0_URL ) users_obj = Users(config.AUTH0_DOMAIN, token['access_token']) try: result = users_obj.get_guardian_enrollments(auth0_user_id) except Auth0Error as e: g.log.error( 'Auth0 error getting guardian enrollments for user', resources={'error': e.message, 'auth0_user_id': auth0_user_id}, ) return response.Response(message=e.message, status=e.status_code) return response.Response(result) def reset_mfa_devices_for_user(auth0_user_id): """Reset all mfa devices for a user. Args: auth0_user_id (str): Auth0 user id to get devices for. Returns: flask.Response: Message contains list of devices. """ get_token = GetToken(config.AUTH0_DOMAIN) token = get_token.client_credentials( config.AUTH0_MACHINE_CLIENT_ID, config.AUTH0_MACHINE_CLIENT_SECRET, config.AUTH0_URL ) guardian_obj = Guardian(config.AUTH0_DOMAIN, token['access_token']) devices = get_guardian_enrollments_for_user(auth0_user_id) if not devices: return devices result = [] for device in devices.message: guardian_obj.delete_enrollment(device['id']) result.append(device['id']) return response.Response({'deleted_devices': result}) def _extract_email_info_from_create_data(data): reset_password = data['reset_password'] del data['reset_password'] email_client_id = config.AUTH0_MACHINE_CLIENT_ID if 'email_client_id' in data: email_client_id = data['email_client_id'] del data['email_client_id'] return [reset_password, email_client_id] def revoke_refresh_token(refresh_token): """Revoke a user's refresh token. Args: refresh_token (str): Auth0 refresh token. Returns: flask.Response: Message success if the request was successful. """ token = revoke_token.RevokeToken(config.AUTH0_DOMAIN) token.revoke_refresh_token(config.AUTH0_ORCHARDGO_NATIVE_CLIENT_ID, refresh_token) return response.Response({'revoke_token': 'success'}) def _get_application_by_name(application_name): """Get application by name. Adding a layer of redis cache so we dont keep calling auth0 each time. """ non_spa_apps = ['workstation-login', 'songwhip-login'] app_type = 'regular_web' if application_name in non_spa_apps else 'spa' cache_key = f'auth0_applications_{app_type}' applications = json.loads(redis.client.get(cache_key) or '{}') if not applications or len(applications) < 1: auth0 = get_auth0_management_handle() params = {'name': application_name, 'app_type': app_type} applications = auth0.clients.all(fields=['client_id', 'name'], extra_params=params) if not applications or len(applications) < 1: return response.create_fatal_response(f'Invalid Auth0 application {application_name}') redis.client.set(cache_key, json.dumps(applications), ex=config.REDIS_CACHE_TTL) for app in applications: if app['name'] == application_name: return app return None def _get_connection_by_name(connection_name): """Get connection by name. Adding a layer of redis cache so we dont keep calling auth0 each time. List of DB/ SSO connections wont change often. """ cache_key = 'auth0_spa_connections' connections = json.loads(redis.client.get(cache_key) or '{}') if not connections or len(connections) < 1: auth0 = get_auth0_management_handle() connections = auth0.connections.all() if not connections or len(connections) < 1: return response.create_fatal_response('Invalid Auth0 connection configured') redis.client.set(cache_key, json.dumps(connections), ex=config.REDIS_CACHE_TTL) for each in connections: if each['name'] == connection_name: return each return None def get_connection_id_from_name(connection_name: str) -> Optional[str]: """ Get the connection id from the connection name. Args: connection_name: The name of the connection. Returns: The connection id if it exists, None otherwise. """ conn_name_to_get_id = connection_name or config.AUTH0_CONNECTION connection = _get_connection_by_name(conn_name_to_get_id) if not connection: return None connection_id = connection.get('id') return connection_id def create_organization_invitation(data, admin_identity_id): """Create an invitation to an organization. Args: data (dict): Attributes for the invitation to create. See: https://auth0.com/docs/api/management/v2#!/Organizations/post_invitations """ auth0 = get_auth0_management_handle() brand = ( data.get('brand').lower() if data.get('brand').lower() in constants.ORG_TYPES else constants.AUTH0_ORCHARD_ORG_NAME ) application_name = data.get('auth0_application_name', config.AUTH0_INSIGHTS_APP_NAME) # Unless admin_name is explicitly passed, get inviter name from admin's identity admin_name = data.get('admin_name') if not admin_name: admin_identity = identities.get_identity(admin_identity_id).message first_name = admin_identity.get('first_name', None) last_name = admin_identity.get('last_name', None) admin_name = ( f'{first_name} {last_name}' if (first_name and last_name) else admin_identity.get('name') ) try: organization = auth0.organizations.get_organization_by_name(brand) if not organization: return response.create_fatal_response(f'Invalid brand name: {brand}') connection_name = user_info_logic.get_auth0_connection_name_from_email(data.get('email')) connection_id = get_connection_id_from_name(connection_name) if not connection_id: return response.create_fatal_response(f'Invalid Auth0 connection: {connection_name}') application = _get_application_by_name(application_name) if not application: return response.create_fatal_response(f'Invalid Auth0 application: {application_name}') application_id = application.get('client_id') invitee_email = data.get('email') identity_res = identities.get_identity_by_email(invitee_email) identity = identity_res and identity_res.message or {} user_metadata = data.get('user_metadata', {}) user_metadata['localization'] = identity.get('localization', 'en') user_metadata['number_format'] = identity.get('number_format', 'us') app_access_res = profiles.get_applications_for_identity_tx(identity.get('id')) app_access = app_access_res and app_access_res.message or {'items': []} user_metadata['app_access'] = app_access.get('items') # lambda-gda-account-creation sets this to false user_metadata['use_new_unified_template'] = user_metadata.get( 'use_new_unified_template', True ) inv_data = { 'inviter': { 'name': admin_name, }, 'invitee': { 'email': invitee_email, }, 'client_id': application_id, 'connection_id': connection_id, 'user_metadata': user_metadata, 'ttl_sec': config.AUTH0_ORG_INVITE_TTL, 'send_invitation_email': True, } g.log.info( 'Creating an org invitation', resources={ 'org_name': organization['name'], 'inviter': admin_name, 'client_id': application_id, 'admin_identity_id': admin_identity_id, 'user_metadata': user_metadata, 'inviter_identity_id': admin_identity_id, 'invitee': invitee_email, }, ) result = auth0.organizations.create_organization_invitation(organization['id'], inv_data) if result: try: identity = identities.update_identity_organization_invitation( email=data.get('email'), invitation_id=result['id'], organization_id=organization['id'], organization_name=organization['name'], admin_identity_id=admin_identity_id, ) if identity: result['identity'] = identity except Exception as e: raise Exception(f'error setting user invitation_id: {str(e)}') from e return response.Response(result) except Auth0Error as errors: g.ows.log.error(str(errors)) return response.create_fatal_response(errors.message) def create_organization_members(data, admin_identity_id): """Add members to an organization. Args: admin_identity_id (str): Identity id of the admin who is adding the user. data (dict): A list of member ids and brand. See: https://auth0.com/docs/api/management/v2#!/Organizations/post_members Sample data: { "brand": "orchard", "members": ["auth0|60c8bad6583e49996c4ad085", "google-apps|someuser@theorchard.com"] } """ auth0 = get_auth0_management_handle() brand = ( data.get('brand') if data.get('brand') != constants.ORCHARD_BRAND else constants.AUTH0_ORCHARD_ORG_NAME ) members = data.get('members') try: organization = auth0.organizations.get_organization_by_name(brand) if not organization: return response.create_fatal_response(f'Organization/brand:{brand} does not exist.') if not members: return response.create_fatal_response( 'Members list cannot be empty. Atleast 1 auth0 user id required.' ) user_data = {'members': members} members_data = [ { 'email': read_user(auth0_user_id).message.get('email'), 'auth0_user_id': auth0_user_id, } for auth0_user_id in members ] g.log.info( 'Adding members to org', resources={ 'org_name': organization['name'], 'org_id': organization['id'], 'members': members_data, 'admin_identity_id': admin_identity_id, }, ) result = auth0.organizations.create_organization_members(organization['id'], user_data) return response.Response(result) except Auth0Error as errors: g.ows.log.error(str(errors)) return response.create_fatal_response(errors.message) def get_organization(org_name: str): """Get organization by name. Args: org_name (str): The name of the organization. """ auth0 = get_auth0_management_handle() try: result = auth0.organizations.get_organization_by_name(org_name) if not result: return response.create_not_found_response( f'Organization/brand:{org_name} does not exist.' ) return response.Response(result) except Auth0Error as errors: if errors.status_code == 404: return response.create_not_found_response(errors.message) g.ows.log.error(str(errors)) return response.create_fatal_response(errors.message) def list_user_organizations(auth0_user_id): """List the organizations that the given auth0 user is in. Args: auth0_user_id (str): The auth0 user's id. """ auth0 = get_auth0_management_handle() try: user_organizations = auth0.users.list_organizations(auth0_user_id) return response.Response([org.get('name') for org in user_organizations['organizations']]) except Auth0Error as errors: if errors.status_code == 404 and errors.message == constants.AUTH0_USER_NOT_FOUND_MESSAGE: return response.Response([]) g.ows.log.error(str(errors)) return response.create_fatal_response(errors.message)