Removing user host name and other path related data captured during t…

…elemetry for security and PII protection purpose. As its not required.

README.md

@@ -493,6 +493,8 @@ We believe that great quality comes from opinions about a product. So we have in
 
 3. **Open Core and Open Source**: Hercules is built on an **open-core model**, combining the spirit of open source with the support and expertise of a commercial company, **TestZeus**. By providing Hercules as open source (licensed under AGPL v3), TestZeus is committed to empowering the testing community with a robust, adaptable tool that's freely accessible and modifiable. Open source offers transparency, trust, and collaborative potential, allowing anyone to dive into the code, contribute, and shape the project's direction.
 
+4. **Telemetry** : All great products are built on good feedback. We have setup telemetry so that we can take feedback, without disturbing the user. Telemetry is enabled by default, but we also believe strongly in the values of "Trust" and "Transparency" so it can be turned off by the users.
+
 ---
 
 ## 🪙 Token Usage

testzeus_hercules/telemetry.py

@@ -8,16 +8,18 @@
 from typing import Any, Dict, Optional
 
 import sentry_sdk
+from sentry_sdk.types import Event, Hint
+from sentry_sdk.scrubber import EventScrubber, DEFAULT_DENYLIST, DEFAULT_PII_DENYLIST
 from pydantic import BaseModel
 
 DSN = "https://d14d2ee82f26a3585b2a892fab7fffaa@o4508256143540224.ingest.us.sentry.io/4508256153042944"
 
 # Telemetry flag, default is enabled (1) unless set to "0" in the environment variable
 ENABLE_TELEMETRY = os.getenv("ENABLE_TELEMETRY", "1") == "1"
 
-# Initialize Sentry only if telemetry is enabled
-if ENABLE_TELEMETRY:
-    sentry_sdk.init(dsn=DSN, max_breadcrumbs=0, send_default_pii=True)
+# custom denylist
+denylist = DEFAULT_DENYLIST + ["sys.argv", "argv", "server_name"]
+pii_denylist = DEFAULT_PII_DENYLIST + ["sys.argv", "argv", "server_name"]
 
 
 class EventType(Enum):
@@ -33,10 +35,51 @@ class EventType(Enum):
 
 class EventData(BaseModel):
     detail: Optional[str] = None
-    additional_data: Optional[Dict[str, Any]] = None  # For any extra details specific to certain events
+    additional_data: Optional[Dict[str, Any]] = (
+        None  # For any extra details specific to certain events
+    )
+
+
+def my_before_send(event: Event, hint: Hint) -> Event | None:
+    # Filter out all ZeroDivisionError events.
+    # Note that the exception type is available in the hint,
+    # but we should handle the case where the exception info
+    # is missing.
+    if hint.get("exc_info", [None])[0] == ZeroDivisionError:
+        return None
+
+    # We can set extra data on the event's "extra" field.
+    event["extra"]["session_summary"] = build_final_message()
+    if "contexts" in event:
+        contexts = event["contexts"]
+        # Check if sys.argv exists and redact secrets
+        if "argv" in contexts:
+            contexts.pop("argv")
 
+    # We have modified the event as desired, so return the event.
+    # The SDK will then send the returned event to Sentry.
+    return event
+
+
+# Initialize Sentry only if telemetry is enabled
+if ENABLE_TELEMETRY:
 
-def get_installation_id(file_path="installation_id.txt"):
+    sentry_sdk.init(
+        dsn=DSN,
+        before_send=my_before_send,
+        max_breadcrumbs=0,
+        send_default_pii=False,
+        send_client_reports=False,
+        server_name=None,
+        event_scrubber=EventScrubber(
+            denylist=denylist, pii_denylist=pii_denylist, recursive=True
+        ),
+    )
+    sentry_sdk.set_extra("sys.argv", None)
+    sentry_sdk.set_user(None)
+
+
+def get_installation_id(file_path="installation_id.txt") -> str:
     """Generate or load a unique installation ID."""
     if os.path.exists(file_path):
         with open(file_path, "r") as file:
@@ -59,7 +102,7 @@ def get_installation_id(file_path="installation_id.txt"):
 }
 
 
-def add_event(event_type: EventType, event_data: EventData):
+def add_event(event_type: EventType, event_data: EventData) -> None:
     """
     Adds an event to the event collector in the appropriate event_type bucket,
     only if telemetry is enabled.
@@ -80,7 +123,7 @@ def add_event(event_type: EventType, event_data: EventData):
     event_collector["buckets"][event_type.value]["event_count"] += 1
 
 
-async def send_message_to_sentry():
+async def send_message_to_sentry() -> None:
     """
     Sends the final message to Sentry asynchronously, only if telemetry is enabled.
     """
@@ -89,25 +132,28 @@ async def send_message_to_sentry():
     try:
         message = build_final_message()
         with sentry_sdk.push_scope() as scope:
-            scope.set_context("session_summary", message)
+            scope.set_extra("session_summary", message)
             sentry_sdk.capture_message("Program execution summary")
     except Exception as e:
         print(f"Error sending message to Sentry: {e}")
 
 
-def build_final_message():
+def build_final_message() -> Dict[str, Any]:
     """
     Builds the final message from collected events, organized by event_type buckets.
     """
     message = {
         "installation_id": event_collector["installation_id"],
         "session_start": event_collector["start_time"],
-        "buckets": {event_type_s: events for event_type_s, events in event_collector["buckets"].items()},
+        "buckets": {
+            event_type_s: events
+            for event_type_s, events in event_collector["buckets"].items()
+        },
     }
     return message
 
 
-def register_shutdown():
+def register_shutdown() -> None:
     """
     Register a shutdown handler to run send_message_to_sentry on exit,
     only if telemetry is enabled.