feature: support for classify logs

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.32.0] - 2026-01-02
+### Added
+- Support for classify log method
+
 ## [0.31.0] - 2025-12-29
 ### Added
 - Support for list featured content rules method

CLI.md

@@ -249,10 +249,15 @@ secops log types --search "windows"
 # Fetch specific page using token
 secops log types --page-size 50 --page-token "next_page_token"
 
-# Search for log types
-secops log types --search "firewall"
+# Classify logs to predict log type:
+secops log classify --log '{"eventType": "user.session.start", "actor": {"alternateId": "user@example.com"}}'
+
+# Classify a log from a file
+secops log classify --log /path/to/log_file.json
 ```
 
+> **Note:** The classify command returns predictions sorted by confidence score. Confidence scores are provided by the API as guidance only and may not always accurately reflect classification certainty. Use scores for relative ranking rather than absolute confidence.
+
 > **Note:** Chronicle uses parsers to process and normalize raw log data into UDM format. If you're ingesting logs for a custom format, you may need to create or configure parsers. See the [Parser Management](#parser-management) section for details on managing parsers.
 
 ### Forwarder Management

README.md

@@ -389,7 +389,20 @@ else:
     print("Invalid log type")
 ```
 
-4. Use custom forwarders:
+4. Classify logs to predict log type:
+```python
+# Classify a raw log to determine its type
+okta_log = '{"eventType": "user.session.start", "actor": {"alternateId": "user@example.com"}}'
+predictions = chronicle.classify_logs(log_data=okta_log)
+
+# Display predictions sorted by confidence score
+for prediction in predictions:
+    print(f"Log Type: {prediction['logType']}, Score: {prediction['score']}")
+```
+
+> **Note:** Confidence scores are provided by the API as guidance only and may not always accurately reflect classification certainty. Use scores for relative ranking rather than absolute confidence.
+
+5. Use custom forwarders:
 ```python
 # Create or get a custom forwarder
 forwarder = chronicle.get_or_create_forwarder(display_name="MyCustomForwarder")

api_module_mapping.md

@@ -311,7 +311,7 @@ Following shows mapping between SecOps [REST Resource](https://cloud.google.com/
 |logProcessingPipelines.list                                                   |v1alpha|chronicle.log_processing_pipelines.list_log_processing_pipelines|secops log-processing list             |
 |logProcessingPipelines.patch                                                  |v1alpha|chronicle.log_processing_pipelines.update_log_processing_pipeline|secops log-processing update           |
 |logProcessingPipelines.testPipeline                                           |v1alpha|chronicle.log_processing_pipelines.test_pipeline            |secops log-processing test             |
-|logs.classify                                                                 |v1alpha|                                                            |                                       |
+|logs.classify                                                                 |v1alpha|chronicle.log_types.classify_logs                           |secops log classify                    |
 | nativeDashboards.addChart                                                      | v1alpha |chronicle.dashboard.add_chart                                                              |secops dashboard add-chart                                         |
 | nativeDashboards.create                                                        | v1alpha |chronicle.dashboard.create_dashboard                                                              |secops dashboard create                                         |
 | nativeDashboards.delete                                                        | v1alpha |chronicle.dashboard.delete_dashboard                                                              |secops dashboard delete                                         |

examples/classify_logs_example.py

@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+
+# 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.
+#
+"""Example demonstrating log type classification with Chronicle."""
+
+import argparse
+import json
+from datetime import datetime, timezone
+
+from secops import SecOpsClient
+from secops.exceptions import APIError
+
+
+def create_sample_okta_log(username: str = "jdoe@example.com") -> str:
+    """Create a sample OKTA log in JSON format.
+
+    Args:
+        username: The username to include in the log.
+
+    Returns:
+        A JSON string representing an OKTA log.
+    """
+    current_time = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    okta_log = {
+        "actor": {"displayName": "Joe Doe", "alternateId": username},
+        "client": {
+            "ipAddress": "192.168.1.100",
+            "userAgent": {"os": "Mac OS X", "browser": "SAFARI"},
+        },
+        "displayMessage": "User login to Okta",
+        "eventType": "user.session.start",
+        "outcome": {"result": "SUCCESS"},
+        "published": current_time,
+    }
+
+    return json.dumps(okta_log)
+
+
+def create_sample_windows_log(username: str = "user123") -> str:
+    """Create a sample Windows XML log.
+
+    Args:
+        username: The username to include in the log.
+
+    Returns:
+        An XML string representing a Windows Event log.
+    """
+    current_time = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    return f"""<Event xmlns='http://schemas.microsoft.com/win/2004/08/events/event'>
+  <System>
+    <Provider Name='Microsoft-Windows-Security-Auditing'
+      Guid='{{54849625-5478-4994-A5BA-3E3B0328C30D}}'/>
+    <EventID>4624</EventID>
+    <Version>1</Version>
+    <Level>0</Level>
+    <Task>12544</Task>
+    <Opcode>0</Opcode>
+    <Keywords>0x8020000000000000</Keywords>
+    <TimeCreated SystemTime='{current_time}'/>
+    <EventRecordID>202117513</EventRecordID>
+    <Correlation/>
+    <Execution ProcessID='656' ThreadID='700'/>
+    <Channel>Security</Channel>
+    <Computer>WIN-SERVER.xyz.net</Computer>
+    <Security/>
+  </System>
+  <EventData>
+    <Data Name='SubjectUserSid'>S-1-0-0</Data>
+    <Data Name='SubjectUserName'>-</Data>
+    <Data Name='TargetUserName'>{username}</Data>
+    <Data Name='WorkstationName'>CLIENT-PC</Data>
+    <Data Name='LogonType'>3</Data>
+  </EventData>
+</Event>"""
+
+
+def create_sample_aws_cloudtrail_log() -> str:
+    """Create a sample AWS CloudTrail log.
+
+    Returns:
+        A JSON string representing an AWS CloudTrail log.
+    """
+    current_time = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    cloudtrail_log = {
+        "eventVersion": "1.05",
+        "userIdentity": {
+            "type": "IAMUser",
+            "principalId": "AIDAI1234EXAMPLE",
+            "arn": "arn:aws:iam::123456789012:user/admin",
+            "accountId": "123456789012",
+            "accessKeyId": "AKIAI1234EXAMPLE",
+            "userName": "admin",
+        },
+        "eventTime": current_time,
+        "eventSource": "s3.amazonaws.com",
+        "eventName": "GetObject",
+        "awsRegion": "us-east-1",
+        "sourceIPAddress": "192.0.2.1",
+        "userAgent": "aws-cli/2.1.0",
+        "requestParameters": {
+            "bucketName": "my-bucket",
+            "key": "example-file.txt",
+        },
+        "responseElements": None,
+        "requestID": "1234567890ABCDEF",
+        "eventID": "abcd1234-5678-90ef-ghij-klmnopqrstuv",
+        "eventType": "AwsApiCall",
+        "recipientAccountId": "123456789012",
+    }
+
+    return json.dumps(cloudtrail_log)
+
+
+def log_classification(chronicle_client):
+    """Raw log classification."""
+    print("\n=== Log Type Classification Example ===\n")
+
+    okta_log = create_sample_okta_log()
+    print(f"Classifying OKTA log...")
+    print(f"Raw log sample: {okta_log[:100]}...\n")
+
+    try:
+        log_type_predictions = chronicle_client.classify_logs(log_data=okta_log)
+
+        if log_type_predictions:
+            print("Predictions (sorted by confidence):")
+            for idx, pred in enumerate(log_type_predictions[:5], 1):
+                log_type = pred.get("logType", "Unknown")
+                score = pred.get("score", 0)
+                print(f"  {idx}. {log_type}: {score:.2%}")
+
+            top_pred = log_type_predictions[0]
+            print(f"\nTop prediction: {top_pred.get('logType')}")
+        else:
+            print("No predictions available")
+
+    except APIError as e:
+        print(f"Error classifying log: {e}")
+    except ValueError as e:
+        print(f"Validation error: {e}")
+
+
+def main():
+    """Run the example."""
+    parser = argparse.ArgumentParser(
+        description="Example of log type classification with Chronicle"
+    )
+    parser.add_argument(
+        "--customer-id",
+        "--customer_id",
+        required=True,
+        help="Chronicle instance ID",
+    )
+    parser.add_argument(
+        "--project-id", "--project_id", required=True, help="GCP project ID"
+    )
+    parser.add_argument("--region", default="us", help="Chronicle API region")
+
+    args = parser.parse_args()
+
+    client = SecOpsClient()
+
+    chronicle = client.chronicle(
+        customer_id=args.customer_id,
+        project_id=args.project_id,
+        region=args.region,
+    )
+
+    log_classification(chronicle)
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "secops"
-version = "0.31.0"
+version = "0.32.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

@@ -85,6 +85,7 @@
     update_log_processing_pipeline,
 )
 from secops.chronicle.log_types import (
+    classify_logs,
     get_all_log_types,
     get_log_type_description,
     is_valid_log_type,
@@ -219,6 +220,7 @@
     "extract_forwarder_id",
     "update_forwarder",
     # Log Types
+    "classify_logs",
     "get_all_log_types",
     "is_valid_log_type",
     "get_log_type_description",

src/secops/chronicle/client.py

@@ -108,6 +108,7 @@
 from secops.chronicle.log_ingest import ingest_udm as _ingest_udm
 from secops.chronicle.log_ingest import list_forwarders as _list_forwarders
 from secops.chronicle.log_ingest import update_forwarder as _update_forwarder
+from secops.chronicle.log_types import classify_logs as _classify_logs
 from secops.chronicle.log_types import get_all_log_types as _get_all_log_types
 from secops.chronicle.log_types import (
     get_log_type_description as _get_log_type_description,
@@ -3077,6 +3078,29 @@ def search_log_types(
             client=self,
         )
 
+    def classify_logs(
+        self,
+        log_data: str,
+    ) -> list[dict[str, Any]]:
+        """Classify a raw log to predict its log type.
+
+        Args:
+            log_data: Raw log string
+
+        Returns:
+            List of possible log types sorted by confidence score.
+
+        Note:
+            Confidence scores are provided by the API as guidance only and
+            may not always accurately reflect classification certainty.
+            Use scores for relative ranking rather than absolute confidence.
+
+        Raises:
+            SecOpsError: If log_data is empty or not a string.
+            APIError: If the API request fails.
+        """
+        return _classify_logs(client=self, log_data=log_data)
+
     def ingest_udm(
         self,
         udm_events: dict[str, Any] | list[dict[str, Any]],

src/secops/chronicle/log_types.py

@@ -20,8 +20,11 @@
 product or vendor.
 """
 
+import base64
 from typing import TYPE_CHECKING, Any
 
+from secops.exceptions import APIError, SecOpsError
+
 if TYPE_CHECKING:
     from secops.chronicle.client import ChronicleClient
 
@@ -238,3 +241,51 @@ def search_log_types(
                 results.append(log_type_data)
 
     return results
+
+
+def classify_logs(
+    client: "ChronicleClient",
+    log_data: str,
+) -> list[dict[str, Any]]:
+    """Classify a raw log to predict its log type.
+
+    Args:
+        client: ChronicleClient instance.
+        log_data: Raw log string.
+
+    Returns:
+        List of possible log types sorted by confidence score.
+        Example:
+            [
+                {"logType": "OKTA", "score": 0.95},
+                {"logType": "ONELOGIN", "score": 0.03}
+            ]
+
+    Note:
+        Confidence scores are provided by the API as guidance only and
+        may not always accurately reflect classification certainty.
+        Use scores for relative ranking rather than absolute confidence.
+
+    Raises:
+        SecOpsError: If log_data is empty or not a string.
+        APIError: If the API request fails.
+    """
+
+    if not log_data:
+        raise SecOpsError("log data cannot be empty")
+
+    if not isinstance(log_data, str):
+        raise SecOpsError("log data must be a string")
+
+    url = f"{client.base_url}/{client.instance_id}/logs:classify"
+
+    encoded_log = base64.b64encode(log_data.encode("utf-8")).decode("utf-8")
+    payload = {"logData": [encoded_log]}
+
+    response = client.session.post(url, json=payload)
+
+    if response.status_code != 200:
+        raise APIError(f"Failed to classify log: {response.text}")
+
+    data = response.json()
+    return data.get("predictions", [])