diff --git a/CLI.md b/CLI.md index 2af82c2..3be8b01 100644 --- a/CLI.md +++ b/CLI.md @@ -1507,6 +1507,16 @@ secops rule-exclusion update-deployment \ --archived false \ --detection-exclusion-application '"{\"curatedRules\": [],\"curatedRuleSets\": [],\"rules\": []}' ``` + +Test a rule exclusion before creating or deploying it +```bash +secops rule-exclusion test \ + --type "DETECTION_EXCLUSION" \ + --query '(ip="8.8.8.8")' \ + --time-window 168 \ + --detection-exclusion-application '{"curatedRules":["projects/my-project/locations/us/instances/my-instance/curatedRules/ur_123"]}' +``` + Compute rule exclusion activity for specific exclusion ```bash secops rule-exclusion compute-activity \ @@ -2398,4 +2408,4 @@ secops dashboard-query get --id query-id ## Conclusion -The SecOps CLI provides a powerful way to interact with Google Security Operations products directly from your terminal. For more detailed information about the SDK capabilities, refer to the [main README](README.md). \ No newline at end of file +The SecOps CLI provides a powerful way to interact with Google Security Operations products directly from your terminal. For more detailed information about the SDK capabilities, refer to the [main README](README.md). diff --git a/README.md b/README.md index 52d8183..7656196 100644 --- a/README.md +++ b/README.md @@ -2593,6 +2593,22 @@ chronicle.update_rule_exclusion_deployment( } ) +# Test a rule exclusion before creating or deploying it +end_time = datetime.utcnow() +start_time = end_time - timedelta(days=7) + +test_result = chronicle.test_rule_exclusion( + refinement_type=RuleExclusionType.DETECTION_EXCLUSION, + query='(ip = "8.8.8.8")', + start_time=start_time, + end_time=end_time, + detection_exclusion_application={ + "curatedRules": [ + "projects/my-project/locations/us/instances/my-instance/curatedRules/ur_123" + ] + }, +) + # Compute rule exclusion Activity for provided time period end_time = datetime.utcnow() start_time = end_time - timedelta(days=7) diff --git a/api_module_mapping.md b/api_module_mapping.md index 000bf38..8151e17 100644 --- a/api_module_mapping.md +++ b/api_module_mapping.md @@ -607,7 +607,7 @@ Following shows mapping between SecOps [REST Resource](https://cloud.google.com/ | searchRawLogs | v1alpha | | | | summarizeEntitiesFromQuery | v1alpha | chronicle.entity.summarize_entity | secops entity | | summarizeEntity | v1alpha | chronicle.entity.summarize_entity | | -| testFindingsRefinement | v1alpha | | | +| testFindingsRefinement | v1alpha | chronicle.rule_exclusion.test_rule_exclusion | secops rule-exclusion test | | translateUdmQuery | v1alpha | chronicle.nl_search.translate_nl_to_udm | | | translateYlRule | v1alpha | | | | udmSearch | v1alpha | chronicle.search.search_udm | secops search | @@ -1007,7 +1007,7 @@ Following shows mapping between SecOps [REST Resource](https://cloud.google.com/ | searchRawLogs | v1alpha | chronicle.log_search.search_raw_logs | secops search raw-logs | | summarizeEntitiesFromQuery | v1alpha | chronicle.entity.summarize_entity | secops entity | | summarizeEntity | v1alpha | chronicle.entity.summarize_entity | | -| testFindingsRefinement | v1alpha | | | +| testFindingsRefinement | v1alpha | chronicle.rule_exclusion.test_rule_exclusion | secops rule-exclusion test | | translateUdmQuery | v1alpha | chronicle.nl_search.translate_nl_to_udm | | | translateYlRule | v1alpha | | | | udmSearch | v1alpha | chronicle.search.search_udm | secops search | diff --git a/src/secops/chronicle/__init__.py b/src/secops/chronicle/__init__.py index 610273e..8196803 100644 --- a/src/secops/chronicle/__init__.py +++ b/src/secops/chronicle/__init__.py @@ -180,6 +180,7 @@ get_rule_exclusion_deployment, list_rule_exclusions, patch_rule_exclusion, + test_rule_exclusion, update_rule_exclusion_deployment, ) from secops.chronicle.rule_retrohunt import ( @@ -311,6 +312,7 @@ "list_rule_exclusions", "patch_rule_exclusion", "compute_rule_exclusion_activity", + "test_rule_exclusion", "get_rule_exclusion_deployment", "update_rule_exclusion_deployment", # UDM Mapping diff --git a/src/secops/chronicle/client.py b/src/secops/chronicle/client.py index 45f704c..0c163f8 100644 --- a/src/secops/chronicle/client.py +++ b/src/secops/chronicle/client.py @@ -282,6 +282,9 @@ from secops.chronicle.rule_exclusion import ( patch_rule_exclusion as _patch_rule_exclusion, ) +from secops.chronicle.rule_exclusion import ( + test_rule_exclusion as _test_rule_exclusion, +) from secops.chronicle.rule_exclusion import ( update_rule_exclusion_deployment as _update_rule_exclusion_deployment, ) @@ -4319,6 +4322,46 @@ def compute_rule_exclusion_activity( end_time=end_time, ) + def test_rule_exclusion( + self, + refinement_type: str, + query: str, + start_time: datetime, + end_time: datetime, + detection_exclusion_application: str | dict[str, Any] | None = None, + outcome_filters: str | list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + """Test a rule exclusion without creating or deploying it. + + Args: + refinement_type: The type of the Findings refinement + Must be one of: + - DETECTION_EXCLUSION + - FINDINGS_REFINEMENT_TYPE_UNSPECIFIED + query: The query for the findings refinement. + start_time: Start of the time window to test + end_time: End of the time window to test + detection_exclusion_application: The resources which the detection + exclusion is applied to. Must be either valid JSON or JSON + string. + outcome_filters: Optional outcome filters as a list or JSON string. + + Returns: + Dictionary containing tested findings refinement activity + + Raises: + APIError: If the API request fails + """ + return _test_rule_exclusion( + self, + refinement_type=RuleExclusionType[refinement_type], + query=query, + start_time=start_time, + end_time=end_time, + detection_exclusion_application=detection_exclusion_application, + outcome_filters=outcome_filters, + ) + def get_rule_exclusion_deployment( self, exclusion_id: str ) -> dict[str, Any]: diff --git a/src/secops/chronicle/rule_exclusion.py b/src/secops/chronicle/rule_exclusion.py index c2a13ee..8faf0f8 100644 --- a/src/secops/chronicle/rule_exclusion.py +++ b/src/secops/chronicle/rule_exclusion.py @@ -17,7 +17,7 @@ import json import sys from dataclasses import asdict, dataclass -from datetime import datetime +from datetime import datetime, timezone from typing import Annotated, Any from secops.chronicle.utils.format_utils import ( @@ -176,6 +176,80 @@ def create_rule_exclusion( ) +def _format_timestamp(dt: datetime) -> str: + """Format a datetime for Chronicle API timestamp fields.""" + if dt.tzinfo is not None: + dt = dt.astimezone(timezone.utc).replace(tzinfo=None) + return dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ") + + +def _parse_json_field( + value: str | dict[str, Any] | list[dict[str, Any]] | None, + field_name: str, +) -> dict[str, Any] | list[dict[str, Any]] | None: + """Parse JSON strings while allowing already-parsed values.""" + if value is None or isinstance(value, (dict, list)): + return value + try: + return json.loads(value) + except json.JSONDecodeError as e: + raise ValueError(f"Invalid JSON string for {field_name}: {e}") from e + + +def test_rule_exclusion( + client, + refinement_type: RuleExclusionType, + query: str, + start_time: datetime, + end_time: datetime, + detection_exclusion_application: str | dict[str, Any] | None = None, + outcome_filters: str | list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Test a rule exclusion without creating or deploying it. + + Args: + client: ChronicleClient instance + refinement_type: The type of the Findings refinement + query: The query for the findings refinement + start_time: Start of the time window to test + end_time: End of the time window to test + detection_exclusion_application: Resources the detection exclusion + applies to. Must be a dictionary or valid JSON string. + outcome_filters: Optional outcome filters as a list or JSON string. + + Returns: + Dictionary containing tested findings refinement activity + + Raises: + APIError: If the API request fails + """ + body = remove_none_values( + { + "type": refinement_type, + "query": query, + "outcomeFilters": _parse_json_field( + outcome_filters, "outcome_filters" + ), + "interval": { + "startTime": _format_timestamp(start_time), + "endTime": _format_timestamp(end_time), + }, + "detectionExclusionApplication": _parse_json_field( + detection_exclusion_application, + "detection_exclusion_application", + ), + } + ) + + return chronicle_request( + client, + method="POST", + endpoint_path=":testFindingsRefinement", + json=body, + error_message="Failed to test rule exclusion", + ) + + def patch_rule_exclusion( client, exclusion_id: str, diff --git a/src/secops/cli/commands/rule_exclusion.py b/src/secops/cli/commands/rule_exclusion.py index a634f7b..0b893ff 100644 --- a/src/secops/cli/commands/rule_exclusion.py +++ b/src/secops/cli/commands/rule_exclusion.py @@ -21,6 +21,7 @@ add_time_range_args, ) from secops.cli.utils.formatters import output_formatter +from secops.cli.utils.input_utils import load_json_or_file from secops.cli.utils.time_utils import get_time_range @@ -99,6 +100,38 @@ def setup_rule_exclusion_command(subparsers): add_time_range_args(activity_parser) activity_parser.set_defaults(func=handle_rule_exclusion_activity_command) + # Test rule exclusion command + test_parser = re_subparsers.add_parser( + "test", help="Test a rule exclusion without deploying it" + ) + test_parser.add_argument( + "--type", + dest="refinement_type", + choices=["DETECTION_EXCLUSION", "FINDINGS_REFINEMENT_TYPE_UNSPECIFIED"], + required=True, + help="Rule exclusion refinement type", + ) + test_parser.add_argument( + "--query", required=True, help="Rule exclusion query" + ) + add_time_range_args(test_parser) + test_parser.add_argument( + "--detection-exclusion-application", + "--detection_exclusion_application", + dest="detection_exclusion_application", + help=( + "Detection exclusion application as JSON string or path to a JSON " + "file" + ), + ) + test_parser.add_argument( + "--outcome-filters", + "--outcome_filters", + dest="outcome_filters", + help="Outcome filters as JSON string or path to a JSON file", + ) + test_parser.set_defaults(func=handle_rule_exclusion_test_command) + # Get rule exclusion deployment command get_deployment_parser = re_subparsers.add_parser( "get-deployment", help="Get rule exclusion deployment" @@ -209,6 +242,35 @@ def handle_rule_exclusion_activity_command(args, chronicle): sys.exit(1) +def handle_rule_exclusion_test_command(args, chronicle): + """Handle rule exclusion test command.""" + try: + start_time, end_time = get_time_range(args) + detection_exclusion_application = ( + load_json_or_file(args.detection_exclusion_application) + if args.detection_exclusion_application + else None + ) + outcome_filters = ( + load_json_or_file(args.outcome_filters) + if args.outcome_filters + else None + ) + + result = chronicle.test_rule_exclusion( + refinement_type=args.refinement_type, + query=args.query, + start_time=start_time, + end_time=end_time, + detection_exclusion_application=detection_exclusion_application, + outcome_filters=outcome_filters, + ) + output_formatter(result, args.output) + except Exception as e: # pylint: disable=broad-exception-caught + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + + def handle_rule_exclusion_get_deployment_command(args, chronicle): """Handle rule exclusion get deployment command.""" try: diff --git a/tests/chronicle/test_rule_exclusion.py b/tests/chronicle/test_rule_exclusion.py index 0060b86..132cb56 100644 --- a/tests/chronicle/test_rule_exclusion.py +++ b/tests/chronicle/test_rule_exclusion.py @@ -431,6 +431,99 @@ def test_compute_rule_exclusion_activity_specific( assert result == {"testKey": "testValue"} +# --- test_rule_exclusion Tests --- + + +def test_test_rule_exclusion(chronicle_client, response_mock): + """Test test_rule_exclusion function.""" + chronicle_client.session.request.return_value = response_mock + start_time = datetime(2026, 1, 29, 15, 28, 13, 975619) + end_time = datetime(2026, 4, 29, 15, 28, 13, 975619) + + detection_exclusion_application = { + "curatedRules": [ + "projects/test-project/locations/us/instances/test-customer/" + "curatedRules/ur_123" + ] + } + outcome_filters = [{"field": "principal.ip", "value": "8.8.8.8"}] + + result = rule_exclusion.test_rule_exclusion( + client=chronicle_client, + refinement_type=rule_exclusion.RuleExclusionType.DETECTION_EXCLUSION, + query='ip = "8.8.8.8"', + start_time=start_time, + end_time=end_time, + detection_exclusion_application=detection_exclusion_application, + outcome_filters=outcome_filters, + ) + + expected_body = { + "type": "DETECTION_EXCLUSION", + "query": 'ip = "8.8.8.8"', + "outcomeFilters": outcome_filters, + "interval": { + "startTime": "2026-01-29T15:28:13.975619Z", + "endTime": "2026-04-29T15:28:13.975619Z", + }, + "detectionExclusionApplication": detection_exclusion_application, + } + + chronicle_client.session.request.assert_called_once_with( + method="POST", + url=f"{chronicle_client.base_url}/{chronicle_client.instance_id}" + ":testFindingsRefinement", + params=None, + json=expected_body, + headers=ANY, + timeout=None, + ) + + assert result == {"testKey": "testValue"} + + +def test_test_rule_exclusion_with_json_strings( + chronicle_client, response_mock +): + """Test test_rule_exclusion parses JSON string fields.""" + chronicle_client.session.request.return_value = response_mock + + result = rule_exclusion.test_rule_exclusion( + client=chronicle_client, + refinement_type=rule_exclusion.RuleExclusionType.DETECTION_EXCLUSION, + query='metadata.event_type = "NETWORK_CONNECTION"', + start_time=datetime(2026, 1, 1), + end_time=datetime(2026, 1, 2), + detection_exclusion_application='{"rules": ["rules/ru_123"]}', + outcome_filters='[{"field": "security_result.action"}]', + ) + + request_body = chronicle_client.session.request.call_args.kwargs["json"] + assert request_body["detectionExclusionApplication"] == { + "rules": ["rules/ru_123"] + } + assert request_body["outcomeFilters"] == [ + {"field": "security_result.action"} + ] + assert result == {"testKey": "testValue"} + + +def test_test_rule_exclusion_error(chronicle_client, response_mock): + """Test test_rule_exclusion function with error response.""" + response_mock.status_code = 400 + response_mock.text = "Bad Request" + chronicle_client.session.request.return_value = response_mock + + with pytest.raises(APIError, match="Failed to test rule exclusion"): + rule_exclusion.test_rule_exclusion( + client=chronicle_client, + refinement_type=rule_exclusion.RuleExclusionType.DETECTION_EXCLUSION, + query='ip = "8.8.8.8"', + start_time=datetime(2026, 1, 1), + end_time=datetime(2026, 1, 2), + ) + + # --- get_rule_exclusion_deployment Tests --- diff --git a/tests/cli/test_rule_exclusion.py b/tests/cli/test_rule_exclusion.py new file mode 100644 index 0000000..e2054b0 --- /dev/null +++ b/tests/cli/test_rule_exclusion.py @@ -0,0 +1,52 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Tests for the SecOps CLI rule exclusion commands.""" + +import argparse +from datetime import datetime, timezone +from unittest.mock import Mock, patch + +from secops.cli.commands.rule_exclusion import ( + handle_rule_exclusion_test_command, +) + + +@patch("secops.cli.commands.rule_exclusion.output_formatter") +def test_handle_rule_exclusion_test_command(mock_output_formatter): + """Test the rule-exclusion test command handler.""" + chronicle = Mock() + chronicle.test_rule_exclusion.return_value = {"activity": {"count": 1}} + args = argparse.Namespace( + refinement_type="DETECTION_EXCLUSION", + query='ip = "8.8.8.8"', + start_time="2026-01-29T15:28:13.975619Z", + end_time="2026-04-29T15:28:13.975619Z", + time_window=24, + detection_exclusion_application='{"curatedRules": ["curatedRules/ur_123"]}', + outcome_filters='[{"field": "principal.ip"}]', + output="json", + ) + + handle_rule_exclusion_test_command(args, chronicle) + + chronicle.test_rule_exclusion.assert_called_once_with( + refinement_type="DETECTION_EXCLUSION", + query='ip = "8.8.8.8"', + start_time=datetime(2026, 1, 29, 15, 28, 13, 975619, tzinfo=timezone.utc), + end_time=datetime(2026, 4, 29, 15, 28, 13, 975619, tzinfo=timezone.utc), + detection_exclusion_application={"curatedRules": ["curatedRules/ur_123"]}, + outcome_filters=[{"field": "principal.ip"}], + ) + mock_output_formatter.assert_called_once_with({"activity": {"count": 1}}, "json")