NO-SNOW: ignore some reqs and masks secrets

Author: sfc-gh-fpawlowski

ci/github/save_mitm_requests.py

@@ -2,6 +2,15 @@
 import json
 from datetime import datetime
 
+from snowflake.connector.secret_detector import SecretDetector
+
+# Domains to ignore (pip/installation traffic)
+IGNORE_DOMAINS = {
+    "pypi.org",
+    "files.pythonhosted.org",
+    "example.com",  # Test domain from setup
+}
+
 # Open CSV file for writing requests
 f = open("test_requests.csv", "w", newline="", encoding="utf-8")
 writer = csv.writer(f)
@@ -29,6 +38,11 @@
 def response(flow):
     """Called when a response is received"""
     try:
+        # Skip if domain should be ignored
+        host = flow.request.pretty_host.lower()
+        if any(ignored_domain in host for ignored_domain in IGNORE_DOMAINS):
+            return
+
         # Calculate duration
         duration_ms = (
             int((flow.response.timestamp_end - flow.request.timestamp_start) * 1000)
@@ -40,16 +54,22 @@ def response(flow):
         request_size = len(flow.request.content) if flow.request.content else 0
         response_size = len(flow.response.content) if flow.response.content else 0
 
-        # Convert headers to JSON strings (easier to parse later)
-        request_headers = json.dumps(dict(flow.request.headers))
-        response_headers = json.dumps(dict(flow.response.headers))
+        # Convert headers to JSON strings and mask secrets
+        request_headers_dict = dict(flow.request.headers)
+        response_headers_dict = dict(flow.response.headers)
 
-        # Extract key info
+        request_headers = SecretDetector.mask_secrets(
+            json.dumps(request_headers_dict)
+        ).masked_text
+        response_headers = SecretDetector.mask_secrets(
+            json.dumps(response_headers_dict)
+        ).masked_text
+
+        # Extract key info and mask sensitive data
         timestamp = datetime.now().isoformat()
         method = flow.request.method
-        url = flow.request.pretty_url
-        host = flow.request.pretty_host
-        path = flow.request.path
+        url = SecretDetector.mask_secrets(flow.request.pretty_url).masked_text
+        path = SecretDetector.mask_secrets(flow.request.path).masked_text
         status_code = flow.response.status_code
         reason = flow.response.reason
         content_type = flow.response.headers.get("content-type", "")
@@ -76,25 +96,31 @@ def response(flow):
         f.flush()  # Ensure it's written immediately
 
     except Exception as e:
-        # Write error row
-        writer.writerow(
-            [
-                datetime.now().isoformat(),
-                "ERROR",
-                str(e),
-                "",
-                "",
-                "",
-                "",
-                "",
-                "",
-                "",
-                "",
-                "",
-                "",
-            ]
-        )
-        f.flush()
+        # Write error row (only for non-ignored domains)
+        if "host" in locals():
+            host_check = locals()["host"]
+        else:
+            host_check = getattr(flow.request, "pretty_host", "").lower()
+
+        if not any(ignored_domain in host_check for ignored_domain in IGNORE_DOMAINS):
+            writer.writerow(
+                [
+                    datetime.now().isoformat(),
+                    "ERROR",
+                    SecretDetector.mask_secrets(str(e)).masked_text,
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                ]
+            )
+            f.flush()
 
 
 def done():