"""ows-users connector.""" import base64 import json from typing import Literal, Optional, TypedDict from flask import g from owsrequest import request from permissions import config from permissions.constants import constants from permissions.types import AdminIdentity, Auth0UserMetadata, Auth0UserResponse from permissions.utils.brand_utils import default_brand_from_brand # Values taken from graphql-user INVITE_TRACKING_EVENT = 'Email Opened' INVITE_TRACKING_SUBJECT = 'inviteTracking' def get_settings_profile_for_identity(identity_id: str): """Get the SettingsProfile for the user with identity_id. Args: identity_id (str): the id of the user. Returns: response.Response: the data of the model. """ resource_url = f'/users/identity/{identity_id}/profiles' res = request.process( application=config.SERVICE_NAME, environment=config.ENVIRONMENT, method='GET', service_name='ows-users', path=resource_url, ) if res.status_code != 200: return None profiles = res.json().get('items', []) profiles = [p for p in profiles if p.get('profile_type') == 'SettingsProfile'] settings_profile = next(iter(profiles), None) return settings_profile def get_auth0_user_id_by_email(email) -> Optional[str]: """Get auth0_users by email if existed. Args: email (str): the email of the user. Returns: Optional[str]: the auth0 user_id of the auth0 user, if any """ resource_url = f'/users/auth0/email/{email}' result = request.process( application=config.SERVICE_NAME, environment=config.ENVIRONMENT, method='GET', service_name='ows-users', path=resource_url, ) users = result.json().get('result') if not users: return None else: return users[0] def get_auth0_user_organizations( admin: AdminIdentity, auth0_user_id: str, ) -> list[str]: """Get auth0 user organizations by auth0_user_id. Args: admin(AdminIdentity): admin identity. auth0_user_id(str): the auth0_user_id of the user. Returns: list[str]: list with user's organizations. """ resource_url = f'/auth0/{auth0_user_id}/organizations' result = request.get( 'ows-users', resource_url, headers={ 'content-type': 'application/json', 'Orchard-Profile-Id': str(admin.settings_profile.profile_id), 'Orchard-Profile-Type': admin.settings_profile.profile_type, }, ) organizations = result.json() # replace 'orchard' from auth0 to 'theorchard' in the neo4j return list( map( lambda x: x.replace(constants.AUTH0_ORCHARD_ORG_NAME, constants.THEORCHARD_BRAND), organizations, ) ) class UserMetadata(TypedDict): """User metadata to be stored in Auth0.""" orchardIdentityId: str user_types: list[str] username: str vend_contact_id: Optional[int] # i think? can be a str sometimes type: Optional[Literal['alw']] # noqa: A003 inviteTracking: str auth0_user_id: Optional[str] invitation_data: Optional[dict] def create_auth0_org_invitation( email: str, brand: str, admin: AdminIdentity, auth0_application_name: str, user_metadata: UserMetadata, ) -> Optional[dict]: """Create Auth0 org invitation for given user and brand. May raise.""" body = { 'email': email, 'brand': default_brand_from_brand(brand), 'admin_name': admin.name, 'user_metadata': user_metadata, 'auth0_application_name': auth0_application_name, } res = request.process( application=config.SERVICE_NAME, environment=config.ENVIRONMENT, method='POST', service_name='ows-users', path='/auth0/invite/organization-member', json=body, headers={ 'content-type': 'application/json', 'Orchard-Profile-Id': str(admin.settings_profile.profile_id), 'Orchard-Profile-Type': admin.settings_profile.profile_type, 'Orchard-Identity-Id': admin.id, }, ) if not res.ok: res.raise_for_status() return res.json() def create_auth0_org_member( brand: str, auth0_user_id: str, admin: AdminIdentity, ) -> None: """Create Auth0 org member for given user and brand. May raise.""" body = { 'brand': default_brand_from_brand(brand), 'members': [auth0_user_id], } headers = { 'Content-Type': 'application/json', 'Orchard-Profile-Id': str(admin.settings_profile.profile_id), 'Orchard-Profile-Type': admin.settings_profile.profile_type, 'Orchard-Identity-Id': admin.id, } if config.ENVIRONMENT == config.DEV_ENVIRONMENT: headers['authorization'] = g.request_context.authorization res = request.post( service_name='ows-users', path='/auth0/add/organization-members', json=body, headers=headers, ) res.raise_for_status() def create_invite_tracking_string( identity_id: str, email: str, brand: str, auth0_id: Optional[str] ) -> str: """Create invitation tracking string.""" obj = { 'writeKey': config.SEGMENT_WRITE_KEY, 'userId': identity_id, 'event': INVITE_TRACKING_EVENT, 'properties': { 'email': email, 'subject': INVITE_TRACKING_SUBJECT, 'brand': brand, # Use auth0 id if it exists, but if it doesn't we're storing identity id in its place # in neo4j, so let's use it here also 'authId': auth0_id or identity_id, }, } obj_str = json.dumps(obj) return base64.b64encode(str.encode(obj_str)) def set_auth0_vend_contact_id( vend_contact_id: int, auth0_user_id: str, identity_id: str, admin_id: str, admin_profile_id: int, ): """Call ows-users to set a new auth0 vend_contact_id for an identity.""" res = request.process( application=config.SERVICE_NAME, environment=config.ENVIRONMENT, method='PUT', service_name='ows-users', path=f'/users/auth0/{auth0_user_id}/primary', # Yes, this endpoint uses "user_id" to refer to vend_contact id json={'user_id': vend_contact_id, 'identity_id': identity_id}, headers={ 'content-type': 'application/json', 'Orchard-Profile-Id': str(admin_profile_id), 'Orchard-Profile-Type': constants.SETTINGSPROFILE, 'Orchard-Identity-Id': admin_id, }, ) if not res.ok: res.raise_for_status() def get_auth0_user(auth0_user_id: str) -> Auth0UserResponse: """Get auth0 user by auth0_user_id.""" res = request.process( application=config.SERVICE_NAME, environment=config.ENVIRONMENT, method='GET', service_name='ows-users', path=f'/auth0/users/{auth0_user_id}', ) if not res.ok: res.raise_for_status() data = res.json() user_metadata_dict = data.get('user_metadata', {}) user_metadata = Auth0UserMetadata(defaultBrand=user_metadata_dict.get('defaultBrand')) return Auth0UserResponse(user_metadata=user_metadata)