import time import typing from datetime import date from datetime import datetime import flask from sqlalchemy import or_ from atlas_um import pgdb, consts from atlas_um.analytics import user_events from atlas_um.audit import admin_audit from atlas_um.caching import strategies as caching_strategies from atlas_um.consts import SystemEvents, Products, AccountStateSource from atlas_um.extensions import auth0, redis, cache, notifications, usm from atlas_um.helpers import either from atlas_um.helpers.auth0 import get_sub_by_auth0_user from atlas_um.helpers.services import BaseLogicService from atlas_um.logs import logger from atlas_um.pgdb import ( DNAAccount, DNAAccountExternalState, ClaimValue, ResourceGroup, ClaimName, ) from atlas_um.pgdb.associations import dna_account_claim_table from atlas_um.pgdb.dna_account import DNAAccountStatuses from atlas_um.settings import Settings from atlas_um.tokens.bearer_tokens import DNABearerToken __all__ = [ "DisableDNAAccountClaimsService", "SyncDNAAccountWithAuth0Service", "UpdateDNAAccountService", "UpdateDNAAccountClaimsService", "CreateDNAAccountService", "SuspendDNAAccountService", "SyncDNAAccountWithUSM", "ImportDNAAccountFromAuth0Service", "SendDNAInvitationService", "SyncInternalUserDomainsWithUSM", "SendDNAAccessExpirationNotificationService", "SyncDNAAccountExternalStates", ] class CreateDNAAccountService(BaseLogicService): """Creates DNA account and all related entities.""" @user_events.track( SystemEvents.creating_dna_account, dna_account_id=lambda result: result.value.sub, ) @admin_audit.track(SystemEvents.creating_dna_account) @cache.invalidated_with(caching_strategies.DNAAccountsCache) def process(self, **kwargs): dna_account = pgdb.DNAAccount() dna_account.given_name = kwargs.get("given_name") dna_account.family_name = kwargs.get("family_name") dna_account.email = kwargs.get("email") dna_account.sub = dna_account.generate_sub(kwargs.get("email")) dna_account.business_unit = kwargs.get("business_unit") dna_account.job_category = kwargs.get("job_category") dna_account.personnel_type = kwargs.get("personnel_type") dna_account.is_sony_employee = kwargs.get("is_sony_employee", False) dna_account.is_vip = kwargs.get("is_vip", False) dna_account.expiration_date = kwargs.get("expiration_date") dna_account.supervisor_email = kwargs.get("supervisor_email") dna_account.supervisor_name = kwargs.get("supervisor_name") dna_account.job_title = kwargs.get("job_title", "") dna_account.location = kwargs.get("location") dna_account.preferred_username = kwargs.get("preferred_username") dna_account.tags = kwargs.get("tags") dna_account.no_mfa = kwargs.get("no_mfa", False) self.session.add(dna_account) return either.Right(dna_account) class UpdateDNAAccountService(BaseLogicService): """Updates DNA account related information on all related entities.""" @user_events.track( SystemEvents.updating_dna_account, dna_account_id=lambda result: result.value.sub, ) @admin_audit.track(SystemEvents.updating_dna_account) @cache.invalidated_with(caching_strategies.DNAAccountsCache) def process(self, dna_account: pgdb.DNAAccount, **kwargs): expiration_date = kwargs.get("expiration_date") if dna_account.status == DNAAccountStatuses.SUSPENDED and ( expiration_date is None or expiration_date > date.today() ): flask.flash("User reactivated successfully.", "positive") dna_account.business_unit = kwargs.get("business_unit") dna_account.job_category = kwargs.get("job_category") dna_account.personnel_type = kwargs.get("personnel_type") dna_account.is_sony_employee = kwargs.get("is_sony_employee", False) dna_account.is_vip = kwargs.get("is_vip", False) dna_account.supervisor_email = kwargs.get("supervisor_email") dna_account.supervisor_name = kwargs.get("supervisor_name") dna_account.expiration_date = expiration_date dna_account.job_title = kwargs.get("job_title", "") dna_account.location = kwargs.get("location") dna_account.preferred_username = kwargs.get("preferred_username") dna_account.tags = kwargs.get("tags") or [] dna_account.no_mfa = kwargs.get("no_mfa", False) if not dna_account.usm_account: dna_account.given_name = kwargs.get("given_name") dna_account.family_name = kwargs.get("family_name") if dna_account.email.lower() != kwargs.get("email", "").lower(): dna_account.invitation_is_sent = False dna_account.email = kwargs.get("email") dna_account.sub = dna_account.generate_sub(kwargs.get("email")) return either.Right(dna_account) class UpdateDNAAccountClaimsService(BaseLogicService): """Updates claims for DNA account.""" @user_events.track( SystemEvents.updating_dna_account_claims, dna_account_id=lambda result: result.value.sub, ) @admin_audit.track(SystemEvents.updating_dna_account_claims) @cache.invalidated_with(caching_strategies.DNAAccountsCache) def process( self, dna_account: pgdb.DNAAccount, claim_values, resource_group ): cleanup_subq = pgdb.pgdb.select(dna_account_claim_table.c.id).where( dna_account_claim_table.c.dna_account_id == dna_account.id, dna_account_claim_table.c.claim_value_id == ClaimValue.id, or_( ClaimValue.claim_name_id == ClaimName.id, dna_account_claim_table.c.claim_name_id == ClaimName.id, ), ClaimName.resource_group_id == resource_group.id, ) pgdb.pgdb.session.execute( pgdb.pgdb.delete(dna_account_claim_table).where( dna_account_claim_table.c.id.in_(cleanup_subq) ) ) claim_names_mapping = self._get_claim_names_mapping( resource_group, claim_values ) if claim_values: pgdb.pgdb.session.execute( pgdb.pgdb.insert(dna_account_claim_table), [ { "dna_account_id": dna_account.id, "claim_value_id": cv.id, "claim_name_id": claim_names_mapping.get(cv).id, } for cv in claim_values ], ) pgdb.pgdb.session.execute( pgdb.pgdb.update(dna_account_claim_table) .where( dna_account_claim_table.c.dna_account_id == dna_account.id, dna_account_claim_table.c.claim_value_id.in_( [cv.id for cv in claim_values] ), ) .values(is_disabled=False) ) # set token length token = DNABearerToken(dna_account) dna_account.token_length = token.token_length return either.Right(dna_account) def _get_claim_names_mapping(self, resource_group, claim_values): mapping = {} for cv in claim_values: if cv.resource_group == resource_group: mapping[cv] = cv.claim_name else: mapping[cv] = ClaimName.query.filter( ClaimName.claim_values_source == cv.claim_name, ClaimName.resource_group == resource_group, ).first() return mapping class DisableDNAAccountClaimsService(BaseLogicService): """Disables claims for DNA account.""" @user_events.track( SystemEvents.disabling_dna_account_claims, dna_account_id=lambda result: result.value.sub, ) @admin_audit.track(SystemEvents.disabling_dna_account_claims) @cache.invalidated_with(caching_strategies.DNAAccountsCache) def process( self, dna_account: pgdb.DNAAccount, claim_values, resource_group ): update_subq = pgdb.pgdb.select(dna_account_claim_table.c.id).where( dna_account_claim_table.c.dna_account_id == dna_account.id, dna_account_claim_table.c.claim_value_id == ClaimValue.id, or_( ClaimValue.claim_name_id == ClaimName.id, dna_account_claim_table.c.claim_name_id == ClaimName.id, ), ClaimName.resource_group_id == resource_group.id, dna_account_claim_table.c.claim_value_id.in_( [cv.id for cv in claim_values] ), ) pgdb.pgdb.session.execute( pgdb.pgdb.update(dna_account_claim_table) .where(dna_account_claim_table.c.id.in_(update_subq)) .values(is_disabled=True) ) globally_disabled_values = ( pgdb.pgdb.select(pgdb.ClaimValue.id) .join( pgdb.ClaimName, pgdb.ClaimName.id == pgdb.ClaimValue.claim_name_id, ) .join( pgdb.ResourceGroup, pgdb.ResourceGroup.id == pgdb.ClaimName.resource_group_id, ) .where( or_( pgdb.ClaimValue.is_deleted == True, # noqa pgdb.ClaimName.is_deleted == True, # noqa pgdb.ResourceGroup.is_deleted == True, # noqa ) ) ) pgdb.pgdb.session.execute( pgdb.pgdb.update(dna_account_claim_table) .where( dna_account_claim_table.c.dna_account_id == dna_account.id, dna_account_claim_table.c.claim_value_id.in_( globally_disabled_values ), ) .values(is_disabled=True) ) return either.Right(dna_account) class SuspendDNAAccountService(BaseLogicService): """Suspends DNA account by updating expiration date.""" @user_events.track( SystemEvents.suspending_dna_account, dna_account_id=lambda result: result.value.sub, ) @admin_audit.track(SystemEvents.suspending_dna_account) @cache.invalidated_with(caching_strategies.DNAAccountsCache) def process(self, dna_account: pgdb.DNAAccount): dna_account.expiration_date = date.today() return either.Right(dna_account) class SyncInternalUserDomainsWithUSM(BaseLogicService): def process(self): result = usm.get_internal_domains() if result.is_left: msg = "Error syncing internal domains with USM" logger.error(msg) return either.Left(msg) internal_domains = result.value if internal_domains: pgdb.InternalUserDomain.query.delete() logger.info("InternalUserDomain cleaned") else: msg = "Domains not received from USM" logger.warning(msg) return either.Left(msg) internal_domain_objs = [ pgdb.InternalUserDomain(domain=domain) for domain in set(d.lower() for d in internal_domains) ] self.session.bulk_save_objects(internal_domain_objs) logger.bind(count_domains=len(internal_domain_objs)).info( "InternalUserDomain saved domains from USM" ) return either.Right(internal_domain_objs) class SyncDNAAccountWithUSM(BaseLogicService): """ Updates all fields in DNAAccount and and related instances with actual data from USM. """ @admin_audit.track(SystemEvents.syncing_dna_account_with_usm) def process(self, dna_account: pgdb.DNAAccount): if dna_account.is_external: result = usm.lookup_user(dna_account.email) else: result = usm.get_user(dna_account.email) if result.is_left: return either.Left("Error syncing account with USM") usm_data = result.value if dna_account.is_external: dna_account.given_name = usm_data.get("firstName") dna_account.family_name = usm_data.get("lastName") else: dna_account.given_name = usm_data.get("givenName") dna_account.family_name = usm_data.get("surname") dna_account.location = usm_data.get("country") or "" dna_account.job_title = usm_data.get("jobTitle") or "" return either.Right(dna_account) class SendDNAInvitationService(BaseLogicService): """Sending DNA invitation.""" @admin_audit.track(SystemEvents.sending_dna_invitation) def process( self, dna_account: pgdb.DNAAccount, admin_dna_account: pgdb.DNAAccount, resource_group: typing.Optional[pgdb.ResourceGroup] = None, ): if dna_account.is_external: self._send_external_user_invitation( dna_account, admin_dna_account, resource_group ) return either.Right(dna_account) def _send_external_user_invitation( self, dna_account, admin_dna_account, resource_group ): result = usm.save_user( dna_account.email, dna_account.given_name, dna_account.family_name, admin_dna_account.email, ) if result.is_left: msg = "Error saving external user in USM" logger.bind(dna_account=dna_account).error(msg) return either.Left(msg) result = usm.reset_password(dna_account.email, admin_dna_account.email) if result.is_left: msg = "Error resetting password in USM" logger.bind(dna_account=dna_account).error(msg) return either.Left(msg) link = result.value body = flask.render_template( "notifications/dna_invitation.html", account=dna_account, link=link, resource_group=resource_group, ) result = notifications.send_email( "Sony Music Product Design & Engineering Invitation", dna_account.email, body, ) if result.is_left: msg = "Error sending DNA invitation email" logger.bind(dna_account=dna_account).error(msg) return either.Left(msg) dna_account.invitation_is_sent = True class SendDNAAccessExpirationNotificationService(BaseLogicService): """Sending access expiration notification.""" @admin_audit.track(SystemEvents.access_expiration_notification) def process( self, dna_account: pgdb.DNAAccount, ): body = flask.render_template( "notifications/account_expiration.html", account=dna_account, ) result = notifications.send_email( "Sony Music Product Design & Engineering", dna_account.email, body, ) if result.is_left: msg = "Error sending DNA access expiration notification" logger.bind(dna_account=dna_account).error(msg) return either.Left(msg) send_at = datetime.now() key = ( f"{consts.ACCOUNTS_EXPIRATION_NOTIFY_REDIS_KEY}-" f"{dna_account.expiration_date.strftime('%y_%m_%d')}-" f"{dna_account.id}" ) redis.setex( key, 60 * 60 * 24 * Settings.ACCOUNTS_EXPIRATION_NOTIFY_DAYS, send_at.strftime("%d/%m/%y"), ) logger.bind(dna_account=dna_account, send_at=send_at).info( "User notified about the access expiration", ) return either.Right(dna_account) class ImportDNAAccountFromAuth0Service(BaseLogicService): """ Imports DNAAccount and related claims from Auth0. If account with sub or email is already exists, then only updates it`s claims. """ MIGRATION_SETTINGS = { # The Decibel product is already integrated, but let`s keep for # the time being Products.DECIBEL: { "roles_mapping": { "MctUser": "user", "DecibelAdmin": "superuser", "DecibelUser": "user", "DecibelSuperUser": "superuser", }, "claims_mapping": { "app_metadata.mct.user_labels": "labels", }, }, Products.RTI: { "roles_mapping": {"RtiUser": "user", "RtiSuperUser": "superuser"}, "claims_mapping": { "app_metadata.ama.artistPermissions": "artists", }, }, Products.APOLLO: { "roles_mapping": { "ApolloUser": "user", "ApolloSuperuser": "superuser", }, "claims_mapping": { "user_metadata.apollo.homeMarketID": "home_market", "app_metadata.marketPermissions": "market_permissions", }, }, } @classmethod def import_all_by_product( cls, product, filter_ids=None, batch_from=None, batch_to=None ): settings = cls.MIGRATION_SETTINGS.get(product) if not settings: logger.bind(product=product).error("No settings for product") return for external_role, local_role in settings.get( "roles_mapping", {} ).items(): users_ids = auth0.list_users_ids( external_role, batch_from, batch_to ) if filter_ids: users_ids = [ user_id for user_id in users_ids if user_id in filter_ids ] for user_id in users_ids: cls.execute(product, local_role, user_id) time.sleep(0.1) @admin_audit.track(SystemEvents.importing_dna_account_from_auth0) @cache.invalidated_with(caching_strategies.DNAAccountsCache) def process(self, product, role, auth0_user_id): user = auth0.get_user(auth0_user_id) dna_account = self._get_or_create_dna_account(user) self._assign_claims_for_dna_account(user, dna_account, product, role) return either.Right(dna_account) def _get_or_create_dna_account(self, user): sub = get_sub_by_auth0_user(user) dna_account = DNAAccount.query.filter( or_(DNAAccount.sub.ilike(sub), DNAAccount.email.ilike(user.email)) ).first() if dna_account is None: dna_account = DNAAccount() dna_account.sub = sub # blocked accounts should be migrated, but adjusted to new logic if user.blocked: dna_account.expiration_date = datetime.today() dna_account.preferred_username = user.nickname dna_account.email = user.email ( dna_account.given_name, dna_account.family_name, ) = self._get_first_last_name(user) self.session.add(dna_account) return dna_account def _assign_claims_for_dna_account(self, user, dna_account, product, role): log = logger.bind( user=user, dna_account=dna_account, product=product, role=role ) resource_group = ResourceGroup.query.by_external_id(product).first() claim_values = [] claim_value = ClaimValue.query.by_value_external_ids( product, "role", role ).first() if claim_value is None: log.bind(product=product, role=role).error("Claim value not found") else: claim_values.append(claim_value) claims_mapping = self.MIGRATION_SETTINGS.get(product, {}).get( "claims_mapping", {} ) claims_ids_suffixes = self.MIGRATION_SETTINGS.get(product, {}).get( "claims_ids_suffixes", {} ) for path, claim_name_id in claims_mapping.items(): external_value = self._get_user_value_by_path(user, path) if external_value is None: log.bind(path=path).error("External value not found for path") continue id_suffix = claims_ids_suffixes.get(claim_name_id, "") if isinstance(external_value, list): if ( product == "rti" and external_value and isinstance(external_value[0], dict) ): external_value = [ item.get("artistID") for item in external_value ] elif ( product == "apollo" and external_value and isinstance(external_value[0], dict) ): external_value = [ f'{item.get("marketID")}_{item.get("role")}' for item in external_value ] for item in set(external_value): claim_value = ClaimValue.query.by_value_external_ids( product, claim_name_id, f"{item}{id_suffix}" ).first() if claim_value is None: log.bind( claim_name_id=claim_name_id, claim_value_id=item, ).error("Claim value not found") continue claim_values.append(claim_value) else: claim_value = ClaimValue.query.by_value_external_ids( product, claim_name_id, f"{external_value}{id_suffix}" ).first() if claim_value is None: log.bind( claim_name_id=claim_name_id, claim_value_id=external_value, ).error("Claim value not found") continue claim_values.append(claim_value) UpdateDNAAccountClaimsService.execute( dna_account=dna_account, claim_values=claim_values, resource_group=resource_group, ) def _get_first_last_name(self, user): first_last_name = user.name.split(" ") if len(first_last_name) == 2: first_name = first_last_name[0] last_name = first_last_name[1] else: first_name = user.name if user.name else user.email last_name = "-" return first_name, last_name def _get_user_value_by_path(self, user, path): val = user.api_result for bit in path.split("."): val = val.get(bit) if not isinstance(val, dict): break return val class SyncDNAAccountWithAuth0Service(BaseLogicService): @admin_audit.track(SystemEvents.syncing_dna_account_with_auth0) def process(self, dna_account: pgdb.DNAAccount): auth0_user = auth0.get_user_by_email(dna_account.email) if auth0_user: dna_account.sub = get_sub_by_auth0_user(auth0_user) return either.Right(dna_account) class SyncDNAAccountExternalStates(BaseLogicService): @classmethod def sync_all(cls): for dna_account in DNAAccount.query.all(): cls.execute(dna_account=dna_account) time.sleep(0.1) def process( self, dna_account: DNAAccount ) -> typing.Optional[either.Either]: states = [] usm_state = self._sync_with_usm(dna_account) if usm_state: states.append(usm_state) return either.Right(states) def _sync_with_usm(self, dna_account: DNAAccount): """ If the account is reachable via the USM API, we are assuming that it is active. If we have some previous state but account is unreachable, we are assuming that it is disabled. If we do not have any state, but the account is on one of the internal domains , we are assuming that the account is disabled. In all other cases the state in unknown, so we are not creating any state object. """ state = DNAAccountExternalState.query.filter( DNAAccountExternalState.dna_account_id == dna_account.id, DNAAccountExternalState.source == AccountStateSource.USM.value, ).first() result = usm.get_user(dna_account.email) if result.is_left: result = usm.lookup_user(dna_account.email) if result.is_left: if state and state.is_active: state.is_active = False state.last_login = dna_account.last_login elif not state and not dna_account.is_external: state = DNAAccountExternalState() self.session.add(state) state.dna_account = dna_account state.is_active = False state.source = AccountStateSource.USM.value state.last_login = dna_account.last_login else: if state: state.is_active = True state.last_login = dna_account.last_login else: state = DNAAccountExternalState() self.session.add(state) state.dna_account = dna_account state.is_active = True state.source = AccountStateSource.USM.value state.last_login = dna_account.last_login logger.bind( sub=dna_account.sub, email=dna_account.email, is_active=state.is_active if state else None, ).info("Updating external state value from USM for DNA account") return state