From 19fe2ffea2df13481bfa71e88ecd66429d462bcc Mon Sep 17 00:00:00 2001 From: Saifeddine ALOUI Date: Sat, 9 Mar 2024 22:40:08 +0100 Subject: [PATCH] Added security measures to sanitize paths in extension endpoints --- .../endpoints/lollms_extensions_infos.py | 19 ++-- lollms/server/endpoints/lollms_xtts.py | 18 +++- lollms/utilities/outlook.py | 87 +++++++++++++++++++ 3 files changed, 116 insertions(+), 8 deletions(-) create mode 100644 lollms/utilities/outlook.py diff --git a/lollms/server/endpoints/lollms_extensions_infos.py b/lollms/server/endpoints/lollms_extensions_infos.py index db6ba7c..8cc2f4e 100644 --- a/lollms/server/endpoints/lollms_extensions_infos.py +++ b/lollms/server/endpoints/lollms_extensions_infos.py @@ -19,6 +19,7 @@ from pathlib import Path from typing import List import psutil import yaml +from lollms.security import sanitize_path # --------------------- Parameter Classes ------------------------------- class ExtensionInstallInfos(BaseModel): @@ -128,6 +129,8 @@ def install_extension(data: ExtensionInstallInfos): except Exception as ex: lollmsElfServer.error(ex) return + else: + data.name = sanitize_path(data.name) try: extension_path = lollmsElfServer.lollms_paths.extensions_zoo_path / data.name ASCIIColors.info(f"- Reinstalling extension {data.name}...") @@ -151,6 +154,8 @@ def reinstall_extension(data: ExtensionInstallInfos): except Exception as ex: lollmsElfServer.error(ex) return + else: + data.name = sanitize_path(data.name) try: extension_path = lollmsElfServer.lollms_paths.extensions_zoo_path / data.name ASCIIColors.info(f"- Reinstalling extension {data.name}...") @@ -180,8 +185,8 @@ def reinstall_extension(data: ExtensionInstallInfos): @router.post("/mount_extension") def mount_extension(data:ExtensionMountingInfos): print("- Mounting extension") - category = data.category - name = data.folder + category = sanitize_path(data.category) + name = sanitize_path(data.folder) package_path = f"{category}/{name}" package_full_path = lollmsElfServer.lollms_paths.extensions_zoo_path/package_path @@ -206,8 +211,8 @@ def mount_extension(data:ExtensionMountingInfos): @router.post("/remount_extension") def remount_extension(data:ExtensionMountingInfos): print("- Remounting extension") - category = data.category - name = data.folder + category = sanitize_path(data.category) + name = sanitize_path(data.folder) package_path = f"{category}/{name}" package_full_path = lollmsElfServer.lollms_paths.extensions_zoo_path/package_path @@ -246,9 +251,9 @@ def remount_extension(data:ExtensionMountingInfos): @router.post("/unmount_extension") def unmount_extension(data:ExtensionMountingInfos): print("- Unmounting extension ...") - category = data.category - name = data.folder - language = data.get('language',None) + category = sanitize_path(data.category) + name = sanitize_path(data.folder) + language = sanitize_path(data.get('language',None)) try: personality_id = f"{category}/{name}" if language is None else f"{category}/{name}:{language}" index = lollmsElfServer.config["personalities"].index(personality_id) diff --git a/lollms/server/endpoints/lollms_xtts.py b/lollms/server/endpoints/lollms_xtts.py index ae9eb5c..b17c4f8 100644 --- a/lollms/server/endpoints/lollms_xtts.py +++ b/lollms/server/endpoints/lollms_xtts.py @@ -150,4 +150,20 @@ def install_xtts(): return {"status":True} except Exception as ex: lollmsElfServer.HideBlockingMessage() - return {"status":False, 'error':str(ex)} \ No newline at end of file + return {"status":False, 'error':str(ex)} + +@router.get("/start_xtts") +def start_xtts(): + try: + lollmsElfServer.ShowBlockingMessage("Starting xTTS api server\nPlease stand by") + from lollms.services.xtts.lollms_xtts import LollmsXTTS + if lollmsElfServer.tts is None: + lollmsElfServer.tts = LollmsXTTS( + lollmsElfServer, + voice_samples_path=Path(__file__).parent/"voices", + xtts_base_url= lollmsElfServer.config.xtts_base_url + ) + lollmsElfServer.HideBlockingMessage() + except Exception as ex: + lollmsElfServer.HideBlockingMessage() + return {"url": None, "error":f"{ex}"} diff --git a/lollms/utilities/outlook.py b/lollms/utilities/outlook.py new file mode 100644 index 0000000..68738ad --- /dev/null +++ b/lollms/utilities/outlook.py @@ -0,0 +1,87 @@ +import win32com.client +from win32com.client import Dispatch + +class OutlookUtilities: + def __init__(self): + try: + self.outlook = Dispatch("Outlook.Application").GetNamespace("MAPI") + except Exception as e: + print(f"Failed to initialize Outlook: {e}") + self.outlook = None + + def open_outlook(self): + # This function doesn't need to do anything in code + # Opening Outlook is handled by the Dispatch call in __init__ + pass + + def get_email_addresses(self): + """Returns a list of email addresses from the user's inbox.""" + if not self.outlook: + print("Outlook is not initialized.") + return [] + try: + inbox = self.outlook.GetDefaultFolder(6) # 6 refers to the inbox + emails = inbox.Items + email_addresses = set() + for email in emails: + email_addresses.add(email.SenderEmailAddress) + return list(email_addresses) + except Exception as e: + print(f"Failed to get email addresses: {e}") + return [] + + def load_mails_from_address(self, address): + """Loads all mails received from a specific email address. + + Args: + address (str): The email address to filter mails from. + + Returns: + list: A list of email objects received from the specified address. + """ + if not self.outlook: + print("Outlook is not initialized.") + return [] + try: + inbox = self.outlook.GetDefaultFolder(6) + emails = inbox.Items + filtered_emails = [] + for email in emails: + if email.SenderEmailAddress == address: + filtered_emails.append(email) + return filtered_emails + except Exception as e: + print(f"Failed to load mails from {address}: {e}") + return [] + + def send_reply(self, original_mail, reply_message): + """Sends a reply to an email. + + Args: + original_mail: The original email object to reply to. + reply_message (str): The message content for the reply. + """ + if not self.outlook: + print("Outlook is not initialized.") + return + try: + reply = original_mail.ReplyAll() + reply.Body = reply_message + reply.Send() + except Exception as e: + print(f"Failed to send reply: {e}") + +# Example usage: +if __name__ == "__main__": + outlook_utils = OutlookUtilities() + outlook_utils.open_outlook() # Not necessary to call, but here for clarity + email_addresses = outlook_utils.get_email_addresses() + print("Email addresses found:", email_addresses) + + specific_address = 'example@example.com' + emails_from_address = outlook_utils.load_mails_from_address(specific_address) + print(f"Loaded {len(emails_from_address)} emails from {specific_address}") + + if emails_from_address: + outlook_utils.send_reply(emails_from_address[0], "This is a test reply.") + print("Reply sent.")