fix(secops): standardize ISO datetime parsing and normalize to UTC

server/secops/secops_mcp/tools/curated_rules_management.py

@@ -18,6 +18,7 @@
 from typing import Any, Dict, Optional
 
 from secops_mcp.server import get_chronicle_client, server
+from secops_mcp.utils import parse_iso_datetime
 
 
 logger = logging.getLogger("secops-mcp")
@@ -318,8 +319,8 @@ async def search_curated_detections(
 
         chronicle = get_chronicle_client(project_id, customer_id, region)
 
-        start_dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
-        end_dt = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
+        start_dt = parse_iso_datetime(start_time)
+        end_dt = parse_iso_datetime(end_time)
 
         result = chronicle.search_curated_detections(
             rule_id=rule_id,

server/secops/secops_mcp/tools/log_ingestion.py

@@ -20,6 +20,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 from secops_mcp.server import get_chronicle_client, server
+from secops_mcp.utils import parse_iso_datetime
 
 
 # Configure logging
@@ -119,9 +120,9 @@ async def ingest_raw_log(
         if labels:
             ingestion_params['labels'] = labels
         if log_entry_time:
-            ingestion_params['log_entry_time'] = datetime.fromisoformat(log_entry_time.replace('Z', '+00:00'))
+            ingestion_params['log_entry_time'] = parse_iso_datetime(log_entry_time)
         if collection_time:
-            ingestion_params['collection_time'] = datetime.fromisoformat(collection_time.replace('Z', '+00:00'))
+            ingestion_params['collection_time'] = parse_iso_datetime(collection_time)
 
         # Ingest the log(s)
         result = chronicle.ingest_log(**ingestion_params)

server/secops/secops_mcp/tools/security_rules.py

@@ -18,6 +18,7 @@
 from typing import Any, Dict, Optional
 
 from secops_mcp.server import get_chronicle_client, server
+from secops_mcp.utils import parse_iso_datetime
 
 # Configure logging
 logger = logging.getLogger("secops-mcp")
@@ -969,21 +970,8 @@ async def create_retrohunt(
         chronicle = get_chronicle_client(project_id, customer_id, region)
 
         # Parse time strings to datetime objects if needed
-        if isinstance(start_time, str):
-            start_dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
-        else:
-            start_dt = start_time
-
-        if isinstance(end_time, str):
-            end_dt = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
-        else:
-            end_dt = end_time
-
-        # Ensure timezone awareness
-        if start_dt.tzinfo is None:
-            start_dt = start_dt.replace(tzinfo=timezone.utc)
-        if end_dt.tzinfo is None:
-            end_dt = end_dt.replace(tzinfo=timezone.utc)
+        start_dt = parse_iso_datetime(start_time)
+        end_dt = parse_iso_datetime(end_time)
 
         # Create retrohunt
         retrohunt = chronicle.create_retrohunt(rule_id, start_dt, end_dt)
@@ -1253,21 +1241,8 @@ async def search_rule_alerts(
         chronicle = get_chronicle_client(project_id, customer_id, region)
 
         # Parse time strings to datetime objects
-        if isinstance(start_time, str):
-            start_dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
-        else:
-            start_dt = start_time
-
-        if isinstance(end_time, str):
-            end_dt = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
-        else:
-            end_dt = end_time
-
-        # Ensure timezone awareness
-        if start_dt.tzinfo is None:
-            start_dt = start_dt.replace(tzinfo=timezone.utc)
-        if end_dt.tzinfo is None:
-            end_dt = end_dt.replace(tzinfo=timezone.utc)
+        start_dt = parse_iso_datetime(start_time) if start_time else None
+        end_dt = parse_iso_datetime(end_time) if end_time else None
 
         # Search for rule alerts
         alerts_response = chronicle.search_rule_alerts(

server/secops/secops_mcp/utils.py

@@ -1,44 +1,72 @@
 """Utility functions for SecOps MCP."""
 
 from datetime import datetime, timedelta, timezone
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Union
+
+
+def parse_iso_datetime(time_input: Union[str, datetime]) -> datetime:
+    """Parses an ISO 8601 string or datetime object and returns a UTC timezone-aware datetime.
+
+    Handles trailing 'Z'/'z', explicit timezone offsets (+HH:MM/-HH:MM),
+    and naive ISO strings/datetimes (which default to UTC). Always returns a datetime
+    normalized to UTC (tzinfo=timezone.utc).
+
+    Args:
+        time_input: ISO 8601 formatted date/time string, or an existing datetime object.
+
+    Returns:
+        A timezone-aware datetime object with tzinfo=timezone.utc.
+
+    Raises:
+        ValueError: If time_input is empty, not a string/datetime, or not a valid ISO 8601 string.
+    """
+    if isinstance(time_input, datetime):
+        if time_input.tzinfo is None:
+            time_input = time_input.replace(tzinfo=timezone.utc)
+        return time_input.astimezone(timezone.utc)
+
+    if not isinstance(time_input, str) or not time_input.strip():
+        raise ValueError(f"Invalid datetime input: {time_input!r}")
+
+    cleaned = time_input.strip()
+    if cleaned.endswith(("z", "Z")):
+        cleaned = cleaned[:-1] + "+00:00"
+
+    dt = datetime.fromisoformat(cleaned)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
 
 def parse_time_range(
-    start_time: Optional[str], 
-    end_time: Optional[str], 
-    hours_back: int
+    start_time: Optional[Union[str, datetime]],
+    end_time: Optional[Union[str, datetime]],
+    hours_back: int,
 ) -> Tuple[datetime, datetime]:
     """Parses ISO strings or defaults to hours_back.
-    
+
     Args:
-        start_time: ISO 8601 start time string (e.g. 2023-01-01T00:00:00Z).
-        end_time: ISO 8601 end time string.
+        start_time: ISO 8601 start time string or datetime (e.g. 2023-01-01T00:00:00Z).
+        end_time: ISO 8601 end time string or datetime.
         hours_back: Fallback hours to look back if start_time is not provided.
-        
+
     Returns:
-        Tuple of (start_dt, end_dt) as timezone-aware datetime objects.
-        
+        Tuple of (start_dt, end_dt) as timezone-aware datetime objects in UTC.
+
     Raises:
         ValueError: If the date strings are malformed or start_time is after end_time.
     """
-    # Parse end_time if provided, otherwise default to now
     if end_time:
-        end_dt = datetime.fromisoformat(end_time)
-        if end_dt.tzinfo is None:
-            end_dt = end_dt.replace(tzinfo=timezone.utc)
+        end_dt = parse_iso_datetime(end_time)
     else:
         end_dt = datetime.now(timezone.utc)
 
-    # Parse start_time if provided
     if start_time:
-        start_dt = datetime.fromisoformat(start_time)
-        if start_dt.tzinfo is None:
-            start_dt = start_dt.replace(tzinfo=timezone.utc)
+        start_dt = parse_iso_datetime(start_time)
     else:
-        # Fallback to hours_back from end_dt
         start_dt = end_dt - timedelta(hours=hours_back)
-            
+
     if start_dt > end_dt:
         raise ValueError(f"Start time ({start_dt}) cannot be after end time ({end_dt})")
-        
+
     return start_dt, end_dt

server/secops/tests/test_secops_tools_unit.py

@@ -339,3 +339,59 @@ async def test_test_rule_buffers_end_time_to_start_of_hour(mock_get_client):
     assert "Total Detections: 1" in result
     assert "Rule successfully detected 1 event(s)" in result
 
+
+# =========================================================================
+# Tests for parse_iso_datetime (Issue #291)
+# =========================================================================
+
+def test_parse_iso_datetime_utc():
+    """Test parse_iso_datetime with UTC timestamps ending in Z and z."""
+    from secops_mcp.utils import parse_iso_datetime
+
+    # Uppercase Z
+    dt1 = parse_iso_datetime("2025-01-20T10:00:00Z")
+    assert dt1 == datetime(2025, 1, 20, 10, 0, 0, tzinfo=timezone.utc)
+    assert dt1.tzinfo == timezone.utc
+
+    # Lowercase z
+    dt2 = parse_iso_datetime("2025-01-20T10:00:00z")
+    assert dt2 == datetime(2025, 1, 20, 10, 0, 0, tzinfo=timezone.utc)
+    assert dt2.tzinfo == timezone.utc
+
+
+def test_parse_iso_datetime_timezone_offsets():
+    """Test parse_iso_datetime normalizes non-UTC timezone offsets to UTC."""
+    from secops_mcp.utils import parse_iso_datetime
+
+    # -05:00 offset (12:00 EST = 17:00 UTC)
+    dt_est = parse_iso_datetime("2025-01-20T12:00:00-05:00")
+    assert dt_est == datetime(2025, 1, 20, 17, 0, 0, tzinfo=timezone.utc)
+    assert dt_est.tzinfo == timezone.utc
+    assert dt_est.hour == 17
+
+    # +02:00 offset (12:00 EET = 10:00 UTC)
+    dt_eet = parse_iso_datetime("2025-01-20T12:00:00+02:00")
+    assert dt_eet == datetime(2025, 1, 20, 10, 0, 0, tzinfo=timezone.utc)
+    assert dt_eet.tzinfo == timezone.utc
+    assert dt_eet.hour == 10
+
+
+def test_parse_iso_datetime_naive_defaults_to_utc():
+    """Test parse_iso_datetime defaults naive ISO strings to UTC."""
+    from secops_mcp.utils import parse_iso_datetime
+
+    dt = parse_iso_datetime("2025-01-20T10:00:00")
+    assert dt == datetime(2025, 1, 20, 10, 0, 0, tzinfo=timezone.utc)
+    assert dt.tzinfo == timezone.utc
+
+
+def test_parse_iso_datetime_invalid():
+    """Test parse_iso_datetime raises ValueError on invalid inputs."""
+    from secops_mcp.utils import parse_iso_datetime
+
+    with pytest.raises(ValueError):
+        parse_iso_datetime("invalid-date")
+
+    with pytest.raises(ValueError):
+        parse_iso_datetime("2025-13-45T99:99:99")
+