import smtplib from datetime import datetime from email.mime.image import MIMEImage from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import formatdate from email.utils import make_msgid from onyx.configs.app_configs import EMAIL_CONFIGURED from onyx.configs.app_configs import EMAIL_FROM from onyx.configs.app_configs import SMTP_PASS from onyx.configs.app_configs import SMTP_PORT from onyx.configs.app_configs import SMTP_SERVER from onyx.configs.app_configs import SMTP_USER from onyx.configs.app_configs import WEB_DOMAIN from onyx.configs.constants import AuthType from onyx.configs.constants import ONYX_DEFAULT_APPLICATION_NAME from onyx.configs.constants import ONYX_SLACK_URL from onyx.db.models import User from onyx.server.runtime.onyx_runtime import OnyxRuntime from onyx.utils.file import FileWithMimeType from onyx.utils.url import add_url_params from onyx.utils.variable_functionality import fetch_versioned_implementation from shared_configs.configs import MULTI_TENANT HTML_EMAIL_TEMPLATE = """\ {title}

{heading}

{message}

{cta_block}

""" def build_html_email( application_name: str | None, heading: str, message: str, cta_text: str | None = None, cta_link: str | None = None, ) -> str: slack_fragment = "" if application_name == ONYX_DEFAULT_APPLICATION_NAME: slack_fragment = f'
Have questions? Join our Slack community here.' if cta_text and cta_link: cta_block = f'{cta_text}' else: cta_block = "" return HTML_EMAIL_TEMPLATE.format( application_name=application_name, title=heading, heading=heading, message=message, cta_block=cta_block, slack_fragment=slack_fragment, year=datetime.now().year, ) def send_email( user_email: str, subject: str, html_body: str, text_body: str, mail_from: str = EMAIL_FROM, inline_png: tuple[str, bytes] | None = None, ) -> None: if not EMAIL_CONFIGURED: raise ValueError("Email is not configured.") # Create a multipart/alternative message - this indicates these are alternative versions of the same content msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["To"] = user_email if mail_from: msg["From"] = mail_from msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid(domain="onyx.app") # Add text part first (lowest priority) text_part = MIMEText(text_body, "plain") msg.attach(text_part) if inline_png: # For HTML with images, create a multipart/related container related = MIMEMultipart("related") # Add the HTML part to the related container html_part = MIMEText(html_body, "html") related.attach(html_part) # Add image with proper Content-ID to the related container img = MIMEImage(inline_png[1], _subtype="png") img.add_header("Content-ID", f"<{inline_png[0]}>") img.add_header("Content-Disposition", "inline", filename=inline_png[0]) related.attach(img) # Add the related part to the message (higher priority than text) msg.attach(related) else: # No images, just add HTML directly (higher priority than text) html_part = MIMEText(html_body, "html") msg.attach(html_part) try: with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as s: s.starttls() s.login(SMTP_USER, SMTP_PASS) s.send_message(msg) except Exception as e: raise e def send_subscription_cancellation_email(user_email: str) -> None: """This is templated but isn't meaningful for whitelabeling.""" # Example usage of the reusable HTML try: load_runtime_settings_fn = fetch_versioned_implementation( "onyx.server.enterprise_settings.store", "load_runtime_settings" ) settings = load_runtime_settings_fn() application_name = settings.application_name except ModuleNotFoundError: application_name = ONYX_DEFAULT_APPLICATION_NAME onyx_file = OnyxRuntime.get_emailable_logo() subject = f"Your {application_name} Subscription Has Been Canceled" heading = "Subscription Canceled" message = ( "

We're sorry to see you go.

" "

Your subscription has been canceled and will end on your next billing date.

" "

If you change your mind, you can always come back!

" ) cta_text = "Renew Subscription" cta_link = "https://www.onyx.app/pricing" html_content = build_html_email( application_name, heading, message, cta_text, cta_link, ) text_content = ( "We're sorry to see you go.\n" "Your subscription has been canceled and will end on your next billing date.\n" "If you change your mind, visit https://www.onyx.app/pricing" ) send_email( user_email, subject, html_content, text_content, inline_png=("logo.png", onyx_file.data), ) def send_user_email_invite( user_email: str, current_user: User, auth_type: AuthType ) -> None: onyx_file: FileWithMimeType | None = None try: load_runtime_settings_fn = fetch_versioned_implementation( "onyx.server.enterprise_settings.store", "load_runtime_settings" ) settings = load_runtime_settings_fn() application_name = settings.application_name except ModuleNotFoundError: application_name = ONYX_DEFAULT_APPLICATION_NAME onyx_file = OnyxRuntime.get_emailable_logo() subject = f"Invitation to Join {application_name} Organization" heading = "You've Been Invited!" # the exact action taken by the user, and thus the message, depends on the auth type message = f"

You have been invited by {current_user.email} to join an organization on {application_name}.

" if auth_type == AuthType.CLOUD: message += ( "

To join the organization, please click the button below to set a password " "or login with Google and complete your registration.

" ) elif auth_type == AuthType.BASIC: message += ( "

To join the organization, please click the button below to set a password " "and complete your registration.

" ) elif auth_type == AuthType.GOOGLE_OAUTH: message += ( "

To join the organization, please click the button below to login with Google " "and complete your registration.

" ) elif auth_type == AuthType.OIDC or auth_type == AuthType.SAML: message += ( "

To join the organization, please click the button below to" " complete your registration.

" ) else: raise ValueError(f"Invalid auth type: {auth_type}") cta_text = "Join Organization" cta_link = f"{WEB_DOMAIN}/auth/signup?email={user_email}" html_content = build_html_email( application_name, heading, message, cta_text, cta_link, ) # text content is the fallback for clients that don't support HTML # not as critical, so not having special cases for each auth type text_content = ( f"You have been invited by {current_user.email} to join an organization on {application_name}.\n" "To join the organization, please visit the following link:\n" f"{WEB_DOMAIN}/auth/signup?email={user_email}\n" ) if auth_type == AuthType.CLOUD: text_content += "You'll be asked to set a password or login with Google to complete your registration." send_email( user_email, subject, html_content, text_content, inline_png=("logo.png", onyx_file.data), ) def send_forgot_password_email( user_email: str, token: str, tenant_id: str, mail_from: str = EMAIL_FROM, ) -> None: # Builds a forgot password email with or without fancy HTML try: load_runtime_settings_fn = fetch_versioned_implementation( "onyx.server.enterprise_settings.store", "load_runtime_settings" ) settings = load_runtime_settings_fn() application_name = settings.application_name except ModuleNotFoundError: application_name = ONYX_DEFAULT_APPLICATION_NAME onyx_file = OnyxRuntime.get_emailable_logo() subject = f"Reset Your {application_name} Password" heading = "Reset Your Password" tenant_param = f"&tenant={tenant_id}" if tenant_id and MULTI_TENANT else "" message = "

Please click the button below to reset your password. This link will expire in 24 hours.

" cta_text = "Reset Password" cta_link = f"{WEB_DOMAIN}/auth/reset-password?token={token}{tenant_param}" html_content = build_html_email( application_name, heading, message, cta_text, cta_link, ) text_content = ( f"Please click the following link to reset your password. This link will expire in 24 hours.\n" f"{WEB_DOMAIN}/auth/reset-password?token={token}{tenant_param}" ) send_email( user_email, subject, html_content, text_content, mail_from, inline_png=("logo.png", onyx_file.data), ) def send_user_verification_email( user_email: str, token: str, new_organization: bool = False, mail_from: str = EMAIL_FROM, ) -> None: # Builds a verification email try: load_runtime_settings_fn = fetch_versioned_implementation( "onyx.server.enterprise_settings.store", "load_runtime_settings" ) settings = load_runtime_settings_fn() application_name = settings.application_name except ModuleNotFoundError: application_name = ONYX_DEFAULT_APPLICATION_NAME onyx_file = OnyxRuntime.get_emailable_logo() subject = f"{application_name} Email Verification" link = f"{WEB_DOMAIN}/auth/verify-email?token={token}" if new_organization: link = add_url_params(link, {"first_user": "true"}) message = ( f"

Click the following link to verify your email address:

{link}

" ) html_content = build_html_email( application_name, "Verify Your Email", message, ) text_content = f"Click the following link to verify your email address: {link}" send_email( user_email, subject, html_content, text_content, mail_from, inline_png=("logo.png", onyx_file.data), )