feat(modules/ngsiem): add LogScale/CQL query execution for NG-SIEM

.gitignore

@@ -6,6 +6,9 @@ __pycache__/
 # C extensions
 *.so
 
+# macOS system files
+.DS_Store
+
 # Distribution / packaging
 .Python
 build/
@@ -204,6 +207,10 @@ CLAUDE.md
 .clinerules/
 memory-bank/
 
+# MCP Configuration
+#  MCP configuration files may contain sensitive credentials and local settings
+.mcp.json
+
 # E2E artifacts
 static_test_report.html
 test_results.json

README.md

@@ -25,6 +25,7 @@
   - [Identity Protection Module](#identity-protection-module)
   - [Incidents Module](#incidents-module)
   - [Intel Module](#intel-module)
+  - [NG-SIEM Module](#ng-siem-module)
   - [Sensor Usage Module](#sensor-usage-module)
   - [Serverless Module](#serverless-module)
   - [Spotlight Module](#spotlight-module)
@@ -86,6 +87,7 @@ The Falcon MCP Server supports different modules, each requiring specific API sc
 | **Identity Protection** | `Identity Protection Entities:read`<br>`Identity Protection Timeline:read`<br>`Identity Protection Detections:read`<br>`Identity Protection Assessment:read`<br>`Identity Protection GraphQL:write` | Comprehensive entity investigation and identity protection analysis |
 | **Incidents** | `Incidents:read` | Analyze security incidents and coordinated activities |
 | **Intel** | `Actors (Falcon Intelligence):read`<br>`Indicators (Falcon Intelligence):read`<br>`Reports (Falcon Intelligence):read` | Research threat actors, IOCs, and intelligence reports |
+| **NG-SIEM** | `NGSIEM:read`<br>`NGSIEM:write` | Execute LogScale/CQL queries against CrowdStrike NG-SIEM for advanced threat hunting and security analysis |
 | **Sensor Usage** | `Sensor Usage:read` | Access and analyze sensor usage data |
 | **Serverless** | `Falcon Container Image:read` | Search for vulnerabilities in serverless functions across cloud service providers |
 | **Spotlight** | `Vulnerabilities:read` | Manage and analyze vulnerability data and security assessments |
@@ -225,6 +227,23 @@ Provides tools for accessing and analyzing CrowdStrike Intelligence:
 
 **Use Cases**: Threat intelligence research, adversary tracking, IOC analysis, threat landscape assessment
 
+### NG-SIEM Module
+
+**API Scopes Required**: `NGSIEM:read`, `NGSIEM:write`
+
+Provides tools for executing LogScale/CQL queries against CrowdStrike NG-SIEM:
+
+- `falcon_execute_ngsiem_query`: Execute LogScale/CQL queries against CrowdStrike NG-SIEM
+
+**Key Features**:
+- **LogScale Query Execution**: Execute LogScale/CQL queries with JSON output
+- **Time Range Support**: Relative formats ("15m", "24h", "7d", "2w") and absolute timestamp ranges
+- **Fixed Timeout**: 3-minute timeout with automatic polling for query completion
+
+**Important**: Always start queries with an event type filter like `#event_simpleName=ProcessRollup2 | tail(3)` for better performance and relevant results.
+
+**Use Cases**: Threat hunting, log analysis, security investigations
+
 ### Sensor Usage Module
 
 **API Scopes Required**: `Sensor Usage:read`

falcon_mcp/client.py

@@ -8,7 +8,7 @@
 import platform
 import sys
 from importlib.metadata import PackageNotFoundError, version
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, cast
 
 # Import the APIHarnessV2 from FalconPy
 from falconpy import APIHarnessV2
@@ -68,17 +68,17 @@ def authenticate(self) -> bool:
         Returns:
             bool: True if authentication was successful
         """
-        return self.client.login()
+        return cast(bool, self.client.login())
 
     def is_authenticated(self) -> bool:
         """Check if the client is authenticated.
 
         Returns:
             bool: True if the client is authenticated
         """
-        return self.client.token_valid
+        return cast(bool, self.client.token_valid)
 
-    def command(self, operation: str, **kwargs) -> Dict[str, Any]:
+    def command(self, operation: str, **kwargs: Any) -> Dict[str, Any]:
         """Execute a Falcon API command.
 
         Args:
@@ -88,7 +88,7 @@ def command(self, operation: str, **kwargs) -> Dict[str, Any]:
         Returns:
             Dict[str, Any]: The API response
         """
-        return self.client.command(operation, **kwargs)
+        return cast(Dict[str, Any], self.client.command(operation, **kwargs))
 
     def get_user_agent(self) -> str:
         """Get RFC-compliant user agent string for API requests.
@@ -131,7 +131,7 @@ def get_headers(self) -> Dict[str, str]:
         Returns:
             Dict[str, str]: Authentication headers including the bearer token
         """
-        return self.client.auth_headers
+        return cast(Dict[str, str], self.client.auth_headers)
 
 
 def get_version() -> str:
@@ -171,7 +171,7 @@ def get_version() -> str:
                         version_str,
                         pyproject_path,
                     )
-                    return version_str
+                    return cast(str, version_str)
             current_path = current_path.parent
 
         logger.debug("pyproject.toml not found in current or parent directories")

falcon_mcp/common/api_scopes.py

@@ -51,6 +51,10 @@
     "GetSensorUsageWeekly": ["Sensor Usage:read"],
     # Serverless operations
     "GetCombinedVulnerabilitiesSARIF": ["Falcon Container Image:read"],
+    # NGSIEM operations
+    "StartSearchV1": ["ngsiem:write"],
+    "GetSearchStatusV1": ["ngsiem:read"],
+    "StopSearchV1": ["ngsiem:write"],
     # Add more mappings as needed
 }
 

falcon_mcp/common/errors.py

@@ -73,7 +73,7 @@ def _format_error_response(
     Returns:
         Dict[str, Any]: Formatted error response
     """
-    response = {"error": message}
+    response: Dict[str, Any] = {"error": message}
 
     # Add details if provided
     if details:
@@ -116,6 +116,16 @@ def handle_api_response(
     """
     status_code = response.get("status_code")
 
+    # If no status_code is present, assume success (200) for responses with data
+    # This handles cases like NGSIEM query results that don't include status_code
+    if status_code is None:
+        # Check if response contains actual data (successful response)
+        if any(key in response for key in ["results", "search_id", "body", "resources"]):
+            status_code = 200
+        else:
+            # Empty response without status_code - treat as error
+            status_code = 500
+
     if status_code != 200:
         # Get a more descriptive error message based on status code
         status_message = ERROR_CODE_DESCRIPTIONS.get(
@@ -135,10 +145,15 @@ def handle_api_response(
             f"{error_message}: {status_message}", details=response, operation=operation
         )
 
-    # Extract resources from the response body
+    # Handle NGSIEM query responses - they have different structure than standard API responses
+    if ("search_id" in response and "results" in response) or (operation and "NG-SIEM Query" in operation):
+        return response
+
+    # Extract resources from the response body for standard API responses
     resources = response.get("body", {}).get("resources", [])
 
     if not resources and default_result is not None:
         return default_result
 
+    # For standard API responses, return resources list/dict as expected
     return resources