Fix errors, issues and misconfigurations (#75)

Fix various errors, issues, and misconfigurations across multiple files.

* **ai/ai_simulations.py**
- Add error handling for potential IndexError if `self.scenarios` is
empty.

* **analytics/behavioral_analysis.py**
  - Add error handling for missing `activity_level` key in `user_data`.

* **app.py**
  - Add import for `os` module to prevent potential errors.

* **backend/pipeline_manager.py**
  - Add error handling for missing API key in `autogpt_task`.

* **core/email_server/EmailServer.py**
- Add error handling for potential error if `saveMail_directory` is not
set.

* **core/end_user/EndUserClient.py**
- Add error handling for potential error if `BaseEmails_directory` is
not set.

* **database/models.py**
  - Add error handling for potential error if `DATABASE_URL` is not set.

---

For more details, open the [Copilot Workspace
session](https://copilot-workspace.githubnext.com/ProjectZeroDays/Project-Red-Sword/pull/75?shareId=30d99065-f7d5-4a1d-96b3-462f75cc1bc1).

Author: ProjectZeroDays

ai/ai_simulations.py

@@ -12,10 +12,12 @@ def __init__(self):
 
     def simulate_attack(self):
         try:
+            if not self.scenarios:
+                raise IndexError("No scenarios available.")
             scenario = random.choice(self.scenarios)
             print(f"[SIMULATION] Executing simulated attack: {scenario}")
         except IndexError as e:
-            print(f"Error during simulation: No scenarios available. {e}")
+            print(f"Error during simulation: {e}")
         except Exception as e:
             print(f"Error during simulation: {e}")
 

analytics/behavioral_analysis.py

@@ -1,9 +1,11 @@
-
 def detect_anomalies(user_data):
     anomalies = []
     for data in user_data:
-        if data["activity_level"] > 100:
-            anomalies.append(data["user_id"])
+        try:
+            if data["activity_level"] > 100:
+                anomalies.append(data["user_id"])
+        except KeyError:
+            print(f"Error: Missing 'activity_level' key in user data: {data}")
     return anomalies
 
 if __name__ == "__main__":

app.py

@@ -3,6 +3,7 @@
 import logging
 from typing import List, Tuple
 import re
+import os
 
 import aiohttp
 import panel as pn

backend/pipeline_manager.py

@@ -18,7 +18,10 @@ def __init__(self):
 
     def autogpt_task(self, task):
         try:
-            openai.api_key = "YOUR_API_KEY"
+            api_key = "YOUR_API_KEY"
+            if not api_key:
+                raise ValueError("Missing API key")
+            openai.api_key = api_key
             response = openai.Completion.create(
                 engine="text-davinci-003",
                 prompt=task,
@@ -28,6 +31,9 @@ def autogpt_task(self, task):
         except openai.error.AuthenticationError as e:
             logging.error(f"API key error during autogpt_task: {e}")
             return "API key error"
+        except ValueError as e:
+            logging.error(f"ValueError during autogpt_task: {e}")
+            return "ValueError: Missing API key"
         except Exception as e:
             logging.error(f"Error during autogpt_task: {e}")
             return ""

core/email_server/EmailServer.py

@@ -14,6 +14,8 @@
 SERVER_HOST = '0.0.0.0'
 SERVER_PORT = 1234
 saveMail_directory = os.getenv("SAVE_MAIL_DIRECTORY", "FlowSteering/ApplicationCode/EmailServer/EmailServerMailDatabase")  # Change this to the directory where you want to save the emails inbox for each user
+if not saveMail_directory:
+    raise ValueError("SAVE_MAIL_DIRECTORY environment variable is not set.")
 message_queue = Queue()
 default_image = 'FlowSteering/assets/PerturbatedImages/DjiPerturbClassForward.png'
 # Server configuration

core/end_user/EndUserClient.py

@@ -27,6 +27,9 @@
 BaseEmails_directory = os.getenv("BASE_EMAILS_DIRECTORY")
 default_image = os.getenv("DEFAULT_IMAGE", '')
 
+if not BaseEmails_directory:
+    raise ValueError("BASE_EMAILS_DIRECTORY environment variable is not set.")
+
 
 def receive_complete_data(client_socket): # this function is used to receive the complete data from the client, adjust the parameters as needed based on your network conditions
     received_data = b""

database/models.py

@@ -2,6 +2,7 @@
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 import logging
+import os
 
 Base = declarative_base()
 
@@ -13,7 +14,9 @@ class DocumentAnalysis(Base):
     links = Column(Text, nullable=True)
     error = Column(Text, nullable=True)
 
-DATABASE_URL = "sqlite:///document_analysis.db"
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///document_analysis.db")
+if not DATABASE_URL:
+    raise ValueError("DATABASE_URL environment variable is not set.")
 engine = create_engine(DATABASE_URL)
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 Base.metadata.create_all(bind=engine)