diff --git a/server/secops/secops_mcp/tools/curated_rules_management.py b/server/secops/secops_mcp/tools/curated_rules_management.py index 62da88f2..ea27489e 100644 --- a/server/secops/secops_mcp/tools/curated_rules_management.py +++ b/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, diff --git a/server/secops/secops_mcp/tools/log_ingestion.py b/server/secops/secops_mcp/tools/log_ingestion.py index 6fbc27ef..b371fb87 100644 --- a/server/secops/secops_mcp/tools/log_ingestion.py +++ b/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) diff --git a/server/secops/secops_mcp/tools/security_rules.py b/server/secops/secops_mcp/tools/security_rules.py index ff55eedc..119014bb 100644 --- a/server/secops/secops_mcp/tools/security_rules.py +++ b/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( diff --git a/server/secops/secops_mcp/utils.py b/server/secops/secops_mcp/utils.py index 4a2a1eb7..ab3b1144 100644 --- a/server/secops/secops_mcp/utils.py +++ b/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 diff --git a/server/secops/tests/test_secops_tools_unit.py b/server/secops/tests/test_secops_tools_unit.py index 77703560..190da07a 100644 --- a/server/secops/tests/test_secops_tools_unit.py +++ b/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") +