Merge branch 'main' into identity-attack-cross-platform

Author: jmorascalyr

Backend/api/app/services/scenario_service.py

@@ -149,6 +149,15 @@ def __init__(self):
                     {"name": "Data Exfiltration", "generators": ["microsoft_365_collaboration"], "duration": 10}
                 ]
             },
+            "tor_user": {
+                "id": "tor_user",
+                "name": "Tor User",
+                "description": "Palo Alto firewall logs showing Tor usage followed by Okta authentication for the same user. Triggers Tor-usage detections with cross-source pivoting.",
+                "phases": [
+                    {"name": "Tor Browsing", "generators": ["paloalto_firewall"], "duration": 5},
+                    {"name": "Okta Login", "generators": ["okta_authentication"], "duration": 5}
+                ]
+            },
             "hr_phishing_pdf_c2": {
                 "id": "hr_phishing_pdf_c2",
                 "name": "HR Phishing PDF -> PowerShell -> Scheduled Task -> C2",
@@ -422,6 +431,26 @@ async def _execute_correlation_scenario(
     async def _execute_scenario(self, execution_id: str, scenario: Dict[str, Any]):
         """Execute scenario in background"""
         try:
+            # Special handling: invoke sender for Tor User scenario
+            if scenario.get("id") == "tor_user":
+                base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+                scenarios_dir = os.path.join(base_dir, "scenarios")
+                if scenarios_dir not in sys.path:
+                    sys.path.insert(0, scenarios_dir)
+                try:
+                    sender = importlib.import_module("tor_user_sender")
+                    workers_env = os.getenv("S1_HEC_WORKERS", "4")
+                    workers = int(workers_env) if workers_env.isdigit() else 4
+                    await asyncio.to_thread(sender.send_tor_user, workers)
+                    self.running_scenarios[execution_id]["status"] = "completed"
+                    self.running_scenarios[execution_id]["completed_at"] = datetime.utcnow().isoformat()
+                    return
+                except Exception as e:
+                    logger.error(f"Failed to run Tor User scenario sender: {e}")
+                    self.running_scenarios[execution_id]["status"] = "failed"
+                    self.running_scenarios[execution_id]["error"] = str(e)
+                    return
+
             # Special handling: invoke sender for HR phishing scenario to generate and send real events
             if scenario.get("id") == "hr_phishing_pdf_c2":
                 # Compute repo root and add scenarios path for import

Backend/event_generators/shared/hec_sender.py

@@ -608,7 +608,9 @@ def _post(url, headers=None, data=None, json=None, timeout=10):
     'verify': DEFAULT_VERIFY_TLS,
     'tls_low': DEFAULT_TLS_LOW,
     'auth_scheme': None,
-    'session': None
+    'session': None,
+    'raw_fallback': False,
+    'event_gzip_supported': None,
 }
 
 # Batch mode controls
@@ -788,32 +790,102 @@ def _send_batch(lines: list, is_json: bool, product: str):
     POST = _CONNECTION_CACHE['session']
     headers_auth = {**HEADERS}
     headers_auth["Authorization"] = f"{_CONNECTION_CACHE['auth_scheme']} {HEC_TOKEN}"
-    # Both endpoints use text/plain with gzip for batched events
+
+    def _raise_with_batch_diag(resp_obj, target_url: str):
+        try:
+            resp_obj.raise_for_status()
+        except Exception:
+            if _VERBOSITY in ('info', 'verbose', 'debug'):
+                body_preview = (resp_obj.text or "")[:500]
+                print(f"[BATCH] HTTP {resp_obj.status_code} from {target_url}", flush=True)
+                if body_preview:
+                    print(f"[BATCH] Error body: {body_preview}", flush=True)
+                sys.stdout.flush()
+            raise
+
+    def _post_event_line_with_fallback(line_str: str, target_url: str):
+        line_bytes = line_str.encode('utf-8')
+        gzip_supported = _CONNECTION_CACHE.get('event_gzip_supported')
+
+        # Try gzip first unless we've already learned this endpoint rejects it.
+        if gzip_supported is not False:
+            gz_line = gzip.compress(line_bytes, compresslevel=1)
+            gz_headers = {**headers_auth, "Content-Type": "application/json", "Content-Encoding": "gzip"}
+            gz_resp = POST(target_url, headers=gz_headers, data=gz_line, timeout=30)
+            try:
+                gz_resp.raise_for_status()
+                _CONNECTION_CACHE['event_gzip_supported'] = True
+                return
+            except Exception:
+                # Fall back to plain JSON for content/parse/media-type failures.
+                status = getattr(gz_resp, "status_code", None)
+                if status not in (400, 415, 422):
+                    _raise_with_batch_diag(gz_resp, target_url)
+                if _CONNECTION_CACHE.get('event_gzip_supported') is not False and _VERBOSITY in ('info', 'verbose', 'debug'):
+                    print("[BATCH] /event gzip rejected, falling back to plain JSON", flush=True)
+                    sys.stdout.flush()
+
+        plain_headers = {**headers_auth, "Content-Type": "application/json"}
+        plain_resp = POST(target_url, headers=plain_headers, data=line_bytes, timeout=30)
+        _raise_with_batch_diag(plain_resp, target_url)
+        _CONNECTION_CACHE['event_gzip_supported'] = False
     body = "\n".join(lines).encode('utf-8')
-    
-    # Use fast compression (level 1) for high throughput - trades compression ratio for speed
-    # Level 1 is ~10x faster than default level 9, with only ~10% larger output
+
+    # Use fast compression (level 1) for high throughput on /raw batches.
+    # For /event batches, send plain JSON to maximize compatibility with
+    # pipeline front-ends that reject gzip-encoded /event payloads.
     gz = gzip.compress(body, compresslevel=1)
-    headers = {**headers_auth, "Content-Type": "text/plain", "Content-Encoding": "gzip"}
+
+    if _VERBOSITY in ('info', 'verbose', 'debug'):
+        print(f"[BATCH] Flushing {len(lines)} events ({len(body)} bytes raw)", flush=True)
+        sys.stdout.flush()
 
     if is_json:
-        # JSON products to /event endpoint
+        # JSON products to /event endpoint.
+        # Send each queued event line individually (not NDJSON body).
+        # Try gzip first; if rejected, auto-fallback to plain JSON.
         url = _CONNECTION_CACHE['event_base']
+        for line in lines:
+            _post_event_line_with_fallback(line, url)
+        if _VERBOSITY == 'debug':
+            print(f"[BATCH] Sent {len(lines)} /event items successfully", flush=True)
+            sys.stdout.flush()
+        return
+    elif not _CONNECTION_CACHE['raw_fallback']:
+        # Raw products: try /raw first
+        raw_url = f"{_CONNECTION_CACHE['raw_base']}?{_build_qs(product)}"
+        raw_headers = {**headers_auth, "Content-Type": "text/plain", "Content-Encoding": "gzip"}
+        try:
+            resp = POST(raw_url, headers=raw_headers, data=gz, timeout=30)
+            resp.raise_for_status()
+            if _VERBOSITY == 'debug':
+                print(f"[BATCH] Response: {resp.status_code} - {resp.text[:200] if resp.text else 'OK'}", flush=True)
+                sys.stdout.flush()
+            return
+        except Exception as raw_err:
+            if _VERBOSITY in ('info', 'verbose', 'debug'):
+                print(f"[BATCH] /raw failed ({type(raw_err).__name__}: {raw_err}), switching to /event with _raw wrapper", flush=True)
+                sys.stdout.flush()
+            _CONNECTION_CACHE['raw_fallback'] = True
+            # Fall through to _raw JSON wrapper below
+            event_lines = [json.dumps(_envelope({"_raw": l}, product, {}, None), separators=(",", ":")) for l in lines]
+            url = _CONNECTION_CACHE['event_base']
+            for line in event_lines:
+                _post_event_line_with_fallback(line, url)
+            if _VERBOSITY == 'debug':
+                print(f"[BATCH] Sent {len(event_lines)} /event+_raw items successfully", flush=True)
+                sys.stdout.flush()
+            return
     else:
-        # Raw/syslog products to /raw endpoint
-        url = f"{_CONNECTION_CACHE['raw_base']}?{_build_qs(product)}"
-    
-    # Show batch flush in info mode and above (not debug only)
-    if _VERBOSITY in ('info', 'verbose', 'debug'):
-        print(f"[BATCH] Flushing {len(lines)} events ({len(gz)} bytes compressed)", flush=True)
-        sys.stdout.flush()
-    
-    resp = POST(url, headers=headers, data=gz, timeout=30)
-    resp.raise_for_status()
-    
-    if _VERBOSITY == 'debug':
-        print(f"[BATCH] Response: {resp.status_code} - {resp.text[:200] if resp.text else 'OK'}", flush=True)
-        sys.stdout.flush()
+        # raw_fallback already cached — go straight to /event with _raw wrapper
+        event_lines = [json.dumps(_envelope({"_raw": l}, product, {}, None), separators=(",", ":")) for l in lines]
+        url = _CONNECTION_CACHE['event_base']
+        for line in event_lines:
+            _post_event_line_with_fallback(line, url)
+        if _VERBOSITY == 'debug':
+            print(f"[BATCH] Sent {len(event_lines)} /event+_raw items successfully", flush=True)
+            sys.stdout.flush()
+        return
 
 SOURCETYPE_MAP_OVERRIDES = {
     # ===== FIXED PARSER MAPPINGS (Based on actual parser directory names) =====
@@ -1196,15 +1268,16 @@ def send_one(line, product: str, attr_fields: dict, event_time: float | None = N
             env_raw = base + "/raw"
     bases = []
     if env_event and env_raw:
+        # Explicit URL configured — use only that; don't fall back to other regions
+        # (tokens are region-specific and cross-region fallback would also fail)
         bases.append((env_event, env_raw))
-    bases.extend([
-        ("https://ingest.us1.sentinelone.net/services/collector/event",
-         "https://ingest.us1.sentinelone.net/services/collector/raw"),
-        ("https://ingest.usea1.sentinelone.net/services/collector/event",
-         "https://ingest.usea1.sentinelone.net/services/collector/raw"),
-        ("https://ingest.sentinelone.net/services/collector/event",
-         "https://ingest.sentinelone.net/services/collector/raw"),
-    ])
+    else:
+        bases.extend([
+            ("https://ingest.us1.sentinelone.net/services/collector/event",
+             "https://ingest.us1.sentinelone.net/services/collector/raw"),
+            ("https://ingest.usea1.sentinelone.net/services/collector/event",
+             "https://ingest.usea1.sentinelone.net/services/collector/raw"),
+        ])
 
     # Try verification/TLS combinations (secure → low TLS → insecure as last resort)
     combos = [
@@ -1251,6 +1324,12 @@ def send_one(line, product: str, attr_fields: dict, event_time: float | None = N
                 payload = _envelope(line, product, attr_fields, event_time)
                 headers = {**headers_auth, "Content-Type": "application/json"}
                 resp = POST(url, headers=headers, json=payload, timeout=10)
+            elif _CONNECTION_CACHE['raw_fallback']:
+                url = _CONNECTION_CACHE['event_base']
+                raw_str = line if isinstance(line, str) else json.dumps(line, separators=(",", ":"))
+                payload = _envelope({"_raw": raw_str}, product, attr_fields, event_time)
+                headers = {**headers_auth, "Content-Type": "application/json"}
+                resp = POST(url, headers=headers, json=payload, timeout=10)
             else:
                 url = f"{_CONNECTION_CACHE['raw_base']}?{_build_qs(product)}"
                 payload = line
@@ -1323,6 +1402,43 @@ def send_one(line, product: str, attr_fields: dict, event_time: float | None = N
                     # On SSL/connection errors, continue to next combo/base
                     continue
 
+    # For raw products: retry all bases using /event with _raw JSON wrapper as fallback
+    if product not in JSON_PRODUCTS:
+        for event_base, raw_base in bases:
+            for verify, tls_low in combos:
+                POST = _make_poster(verify=verify, tls_low=tls_low)
+                for scheme in auth_schemes:
+                    headers_auth = {**HEADERS}
+                    headers_auth["Authorization"] = f"{scheme} {HEC_TOKEN}"
+                    raw_str = line if isinstance(line, str) else json.dumps(line, separators=(",", ":"))
+                    try:
+                        url = event_base
+                        payload = _envelope({"_raw": raw_str}, product, attr_fields, event_time)
+                        headers = {**headers_auth, "Content-Type": "application/json"}
+                        if DEBUG:
+                            print(f"[DEBUG] Sending to {url} (_raw fallback)")
+                        resp = POST(url, headers=headers, json=payload, timeout=10)
+                        if resp.status_code in (401, 403) and scheme == auth_schemes[0]:
+                            continue
+                        resp.raise_for_status()
+                        _CONNECTION_CACHE['configured'] = True
+                        _CONNECTION_CACHE['event_base'] = event_base
+                        _CONNECTION_CACHE['raw_base'] = raw_base
+                        _CONNECTION_CACHE['verify'] = verify
+                        _CONNECTION_CACHE['tls_low'] = tls_low
+                        _CONNECTION_CACHE['auth_scheme'] = scheme
+                        _CONNECTION_CACHE['session'] = POST
+                        _CONNECTION_CACHE['raw_fallback'] = True
+                        if _VERBOSITY in ('info', 'verbose', 'debug'):
+                            print(f"[INFO] /raw not supported, using /event+_raw wrapper for {product}", flush=True)
+                        try:
+                            return resp.json()
+                        except ValueError:
+                            return {"status": "OK", "code": resp.status_code}
+                    except Exception as e:
+                        last_error = e
+                        continue
+
     # If all attempts failed, raise last error for visibility
     if last_error:
         raise last_error