"""Logic for updating default brand after access changes.""" import requests from flask import g from permissions.connectors import neo4j as neo4j_connector from permissions.constants import constants from permissions.models import ( auth0 as auth0_model, identity as identity_model, owsusers as owsusers_model, tenant as tenant_model, ) from permissions.types import Auth0UserMetadata, Identity from permissions.utils.brand_utils import default_brand_from_brand def get_auth0_user_metadata(identity: Identity) -> Auth0UserMetadata | None: """Fetch Auth0 user metadata, returning None if user doesn't exist (404) or on error.""" try: response = owsusers_model.get_auth0_user(identity.auth0_user_id) return response.user_metadata except requests.exceptions.HTTPError as e: if e.response.status_code == 404: g.log.info( 'Auth0 user does not exist yet', resources={ 'identity_id': identity.id, 'auth0_user_id': identity.auth0_user_id, }, ) return None def get_current_brands(identity_id: str, company_brand: str | None = None) -> list[str]: """ Get list of brands the identity currently has. Args: identity_id: The identity to check company_brand: If provided, use this as the only brand (optimization for reactivation) """ if company_brand is not None: return [company_brand] return tenant_model.get_brands_for_identity(identity_id) def needs_brand_update(default_brand: str | None, current_brands: list[str]) -> bool: """Check if the default brand is no longer in the user's current brands.""" if default_brand is None: return False return default_brand not in current_brands def update_auth0_default_brand(auth0_user_id: str, brand: str | None) -> bool: """Update Auth0 user's default brand. Returns True on success.""" try: auth0_model.update_user_metadata({auth0_user_id: {'defaultBrand': brand}}) return True except Exception as e: g.log.error( 'Error updating Auth0 default brand', resources={'auth0_user_id': auth0_user_id, 'error': str(e)}, ) return False def update_neo4j_default_brand( identity_id: str, brand: str | None, admin_identity_id: str, ) -> None: """Update Neo4j identity's default brand. Raises on failure.""" try: with neo4j_connector.db_session(access_mode=constants.NEO4j_WRITE_ACCESS) as neo4j_session: identity_model.update_identity_default_brand( session=neo4j_session, identity_id=identity_id, default_brand=brand, admin_identity_id=admin_identity_id, ) except Exception as e: g.log.error( 'Error updating Neo4j default brand', resources={'identity_id': identity_id, 'error': str(e)}, ) raise def update_default_brand_if_needed( identity: Identity, admin_identity_id: str, company_brand: str | None = None, ) -> None: """ Ensure default brand is valid after an access change. Args: identity: The identity whose brand may need updating admin_identity_id: The admin making the change company_brand: If provided, use this as the only brand (optimization for reactivation) """ current_brands = get_current_brands(identity.id, company_brand) # auth0 brands are a reduced set of the allowed identity brands auth0_brands = [default_brand_from_brand(b) for b in current_brands] auth0_user_metadata = get_auth0_user_metadata(identity) new_brand = current_brands[0] if current_brands else None # Check Auth0 if auth0_user_metadata: if needs_brand_update(auth0_user_metadata.defaultBrand, auth0_brands): update_auth0_default_brand(identity.auth0_user_id, new_brand) # Check Neo4j if needs_brand_update(identity.default_brand, current_brands): update_neo4j_default_brand(identity.id, new_brand, admin_identity_id)