Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion CLI.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 \
Expand Down Expand Up @@ -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).
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).
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions api_module_mapping.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down Expand Up @@ -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 |
Expand Down
2 changes: 2 additions & 0 deletions src/secops/chronicle/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
Expand Down
43 changes: 43 additions & 0 deletions src/secops/chronicle/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -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]:
Expand Down
76 changes: 75 additions & 1 deletion src/secops/chronicle/rule_exclusion.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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,
Expand Down
62 changes: 62 additions & 0 deletions src/secops/cli/commands/rule_exclusion.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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:
Expand Down
Loading