Merge pull request #204 from simonebruzzechesse/main

New fetch_parser_candidates for parsers

Author: mihirvala08

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.42.0] - 2026-04-15
+### Added
+- `fetch_parser_candidates()` method to retrieve parser candidates for a given log type
+- CLI command `secops parser fetch-candidates` for fetching parser candidates for given type
+
 ## [0.41.0] - 2026-04-09
 ### Added
 - Comprehensive SOAR integration management capabilities

CLI.md

@@ -593,6 +593,12 @@ secops parser list --log-type "OKTA" --page-size 50 --filter "state=ACTIVE"
 secops parser get --log-type "WINDOWS" --id "pa_12345"
 ```
 
+#### Fetch parser candidates:
+
+```bash
+secops parser fetch-candidates --log-type "WINDOWS_DHCP" --parser-action "PARSER_ACTION_OPT_IN_TO_PREVIEW"
+```
+
 #### Create a new parser:
 
 ```bash

README.md

@@ -1739,6 +1739,12 @@ print(f"Parser content: {parser.get('text')}")
 chronicle.activate_parser(log_type=log_type, id=parser_id)
 chronicle.deactivate_parser(log_type=log_type, id=parser_id)
 
+# Fetch parser candidates (unactivated prebuilt parsers)
+candidates = chronicle.fetch_parser_candidates(
+    log_type=log_type,
+    parser_action="PARSER_ACTION_OPT_IN_TO_PREVIEW"
+)
+
 # Copy an existing parser as a starting point
 copied_parser = chronicle.copy_parser(log_type=log_type, id="pa_existing_parser")
 

api_module_mapping.md

@@ -553,6 +553,7 @@ Following shows mapping between SecOps [REST Resource](https://cloud.google.com/
 | logTypes.parsers.create                                                          | v1alpha   | chronicle.parser.create_parser                                                                                               | secops parser create                                         |
 | logTypes.parsers.deactivate                                                      | v1alpha   | chronicle.parser.deactivate_parser                                                                                           | secops parser deactivate                                     |
 | logTypes.parsers.delete                                                          | v1alpha   | chronicle.parser.delete_parser                                                                                               | secops parser delete                                         |
+| logTypes.parsers.fetchParserCandidates                                           | v1alpha   | chronicle.parser.fetch_parser_candidates                                                                                     | secops parser fetch-candidates                               |
 | logTypes.parsers.get                                                             | v1alpha   | chronicle.parser.get_parser                                                                                                  | secops parser get                                            |
 | logTypes.parsers.list                                                            | v1alpha   | chronicle.parser.list_parsers                                                                                                | secops parser list                                           |
 | logTypes.parsers.validationReports.get                                           | v1alpha   |                                                                                                                              |                                                              |

examples/example.py

@@ -78,8 +78,8 @@ def example_udm_search(chronicle):
 
 
 def example_udm_search_view(chronicle):
-    """Example 14: UDM Search View."""
-    print("\n=== Example 14: UDM Search View ===")
+    """Example 15: UDM Search View."""
+    print("\n=== Example 15: UDM Search View ===")
     start_time, end_time = get_time_range()
 
     try:
@@ -1413,9 +1413,48 @@ def example_parser_workflow(chronicle):
         print(f"\nUnexpected error: {e}")
 
 
+def example_fetch_parser_candidates(chronicle):
+    """Example 13: Fetch Parser Candidates for a log type."""
+    print("\n=== Example 13: Fetch Parser Candidates ===")
+
+    log_type = "OKTA"
+    parser_action = "PARSER_ACTION_OPT_IN_TO_PREVIEW"
+
+    try:
+        print(
+            f"\nFetching parser candidates for log type '{log_type}' "
+            f"with action '{parser_action}'..."
+        )
+        candidates = chronicle.fetch_parser_candidates(
+            log_type=log_type,
+            parser_action=parser_action,
+        )
+
+        if not candidates:
+            print(f"No parser candidates found for log type '{log_type}'.")
+            return
+
+        print(f"Found {len(candidates)} parser candidate(s):")
+        for candidate in candidates:
+            name = candidate.get("name", "N/A")
+            state = candidate.get("state", "N/A")
+            parser_id = name.split("/")[-1]
+            print(f"  - ID: {parser_id}, State: {state}")
+
+    except APIError as e:
+        print(f"\nAPI Error: {e}")
+        print("\nTroubleshooting tips:")
+        print(
+            "- Ensure the log type supports prebuilt parser candidates"
+        )
+        print("- Check if you have the required permissions")
+    except ValueError as e:
+        print(f"\nInvalid input: {e}")
+
+
 def example_rule_test(chronicle):
-    """Example 13: Test a detection rule against historical data."""
-    print("\n=== Example 13: Test a Detection Rule Against Historical Data ===")
+    """Example 14: Test a detection rule against historical data."""
+    print("\n=== Example 14: Test a Detection Rule Against Historical Data ===")
 
     # Define time range for testing - use a recent time period (last 7 days)
     end_time = datetime.now(timezone.utc) - timedelta(minutes=15)
@@ -1491,8 +1530,9 @@ def example_rule_test(chronicle):
     "10": example_udm_ingestion,
     "11": example_gemini,
     "12": example_parser_workflow,
-    "13": example_rule_test,
-    "14": example_udm_search_view,
+    "13": example_fetch_parser_candidates,
+    "14": example_rule_test,
+    "15": example_udm_search_view,
 }
 
 
@@ -1507,7 +1547,7 @@ def main():
     parser.add_argument(
         "--example",
         "-e",
-        help="Example number to run (1-14). If not specified, runs all examples.",
+        help="Example number to run (1-15). If not specified, runs all examples.",
     )
 
     args = parser.parse_args()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "secops"
-version = "0.41.0"
+version = "0.42.0"
 description = "Python SDK for wrapping the Google SecOps API for common use cases"
 readme = "README.md"
 requires-python = ">=3.10"

src/secops/chronicle/__init__.py

@@ -137,6 +137,7 @@
     ListBasis,
     MonthlyScheduleDetails,
     OneTimeScheduleDetails,
+    ParserAction,
     PrevalenceData,
     PythonVersion,
     ScheduleType,
@@ -151,6 +152,7 @@
     WidgetMetadata,
 )
 from secops.chronicle.nl_search import translate_nl_to_udm
+from secops.chronicle.parser import fetch_parser_candidates
 from secops.chronicle.reference_list import (
     ReferenceListSyntaxType,
     ReferenceListView,
@@ -243,6 +245,9 @@
     "search_raw_logs",
     # Natural Language Search
     "translate_nl_to_udm",
+    # Parser
+    "fetch_parser_candidates",
+    "ParserAction",
     # Entity
     "import_entities",
     "summarize_entity",

src/secops/chronicle/client.py

@@ -174,16 +174,17 @@
 )
 from secops.chronicle.models import (
     APIVersion,
+    AlertState,
     CaseCloseReason,
     CaseList,
     CasePriority,
     DashboardChart,
     DashboardQuery,
     EntitySummary,
     InputInterval,
-    TileType,
-    AlertState,
     ListBasis,
+    ParserAction,
+    TileType,
 )
 from secops.chronicle.nl_search import nl_search as _nl_search
 from secops.chronicle.nl_search import translate_nl_to_udm
@@ -195,6 +196,9 @@
 from secops.chronicle.parser import create_parser as _create_parser
 from secops.chronicle.parser import deactivate_parser as _deactivate_parser
 from secops.chronicle.parser import delete_parser as _delete_parser
+from secops.chronicle.parser import (
+    fetch_parser_candidates as _fetch_parser_candidates,
+)
 from secops.chronicle.parser import get_parser as _get_parser
 from secops.chronicle.parser import list_parsers as _list_parsers
 from secops.chronicle.parser import run_parser as _run_parser
@@ -2774,6 +2778,35 @@ def get_parser(
         """
         return _get_parser(self, log_type=log_type, id=id)
 
+    def fetch_parser_candidates(
+        self,
+        log_type: str,
+        parser_action: ParserAction | str,
+    ) -> list[Any]:
+        """Retrieves prebuilt parser candidates.
+
+        Args:
+            log_type: Log type of the parser
+            parser_action: Action to perform on the parser candidates. Can be
+                a ParserAction enum value or a string. Valid values:
+                - ParserAction.PARSER_ACTION_UNSPECIFIED
+                - ParserAction.PARSER_ACTION_OPT_IN_TO_PREVIEW
+                - ParserAction.PARSER_ACTION_OPT_OUT_OF_PREVIEW
+                - ParserAction.CLONE_PREBUILT
+
+        Returns:
+            List of candidate parsers
+
+        Raises:
+            ValueError: If parser_action is an invalid string value
+            APIError: If the API request fails
+        """
+        return _fetch_parser_candidates(
+            self,
+            log_type=log_type,
+            parser_action=parser_action,
+        )
+
     def list_parsers(
         self,
         log_type: str = "-",

src/secops/chronicle/models.py

@@ -1150,3 +1150,18 @@ class APIVersion(StrEnum):
     V1 = "v1"
     V1BETA = "v1beta"
     V1ALPHA = "v1alpha"
+
+
+class ParserAction(StrEnum):
+    """Actions that can be performed on parser candidates.
+
+    See:
+        https://cloud.google.com/chronicle/docs/reference/rest/v1beta/
+        projects.locations.instances.logTypes.parsers/
+        fetchParserCandidates#ParserAction
+    """
+
+    PARSER_ACTION_UNSPECIFIED = "PARSER_ACTION_UNSPECIFIED"
+    PARSER_ACTION_OPT_IN_TO_PREVIEW = "PARSER_ACTION_OPT_IN_TO_PREVIEW"
+    PARSER_ACTION_OPT_OUT_OF_PREVIEW = "PARSER_ACTION_OPT_OUT_OF_PREVIEW"
+    CLONE_PREBUILT = "CLONE_PREBUILT"

src/secops/chronicle/parser.py

@@ -19,14 +19,15 @@
 import logging
 from typing import Any
 
-from secops.chronicle.models import APIVersion
+from secops.chronicle.models import APIVersion, ParserAction
 from secops.chronicle.utils.format_utils import remove_none_values
 from secops.chronicle.utils.request_utils import (
     chronicle_paginated_request,
     chronicle_request,
 )
 from secops.exceptions import APIError, SecOpsError
 
+
 # Constants for size limits
 MAX_LOG_SIZE = 10 * 1024 * 1024  # 10MB per log
 MAX_LOGS = 1000  # Maximum number of logs to process
@@ -235,6 +236,54 @@ def get_parser(
     )
 
 
+def fetch_parser_candidates(
+    client: "ChronicleClient",
+    log_type: str,
+    parser_action: ParserAction | str,
+) -> list[Any]:
+    """Retrieves prebuilt parser candidates.
+
+    Args:
+        client: ChronicleClient instance
+        log_type: Log type of the parser
+        parser_action: Action to perform on the parser candidates. Can be a
+            ParserAction enum value or a string. Valid values:
+            - ParserAction.PARSER_ACTION_UNSPECIFIED
+            - ParserAction.PARSER_ACTION_OPT_IN_TO_PREVIEW
+            - ParserAction.PARSER_ACTION_OPT_OUT_OF_PREVIEW
+            - ParserAction.CLONE_PREBUILT
+
+    Returns:
+        List of candidate parsers
+
+    Raises:
+        ValueError: If log_type is empty or parser_action is an invalid string
+        APIError: If the API request fails
+    """
+    if not log_type:
+        raise ValueError("log_type cannot be empty")
+    if isinstance(parser_action, str) and not isinstance(
+        parser_action, ParserAction
+    ):
+        try:
+            parser_action = ParserAction(parser_action)
+        except ValueError as e:
+            valid = ", ".join(m.value for m in ParserAction)
+            raise ValueError(
+                f'Invalid parser_action: "{parser_action}". '
+                f"Valid values: {valid}"
+            ) from e
+
+    data = chronicle_request(
+        client,
+        method="GET",
+        endpoint_path=f"logTypes/{log_type}/parsers:fetchParserCandidates",
+        params={"parserAction": parser_action},
+        error_message="Failed to fetch parser candidates",
+    )
+    return data.get("candidates", [])
+
+
 def list_parsers(
     client: "ChronicleClient",
     log_type: str = "-",