Chronicle API: Ingestion Methods

This looks like a mix of legacy backstory api’s and new Chronicle api’s. Adding a sample python script I have used for the new Chronicle Ingestion API if this is helpful. If you believe the api’s are incorrect, please reach out to your Google SecOps CE or account team to investigate this more. 

 

import os
import json
import base64
import requests
from google.oauth2 import service_account
from google.auth.transport.requests import Request

# =====================================================================
# CONFIGURATION
# Replace the placeholder values below with your environment details.
# =====================================================================

# Path to your Google Cloud Service Account JSON key
KEY_FILE_PATH = "path/to/your/service_account_key.json"

# Google Cloud and SecOps Environment Details
PROJECT_ID = "YOUR_PROJECT_ID"
INSTANCE_ID = "YOUR_INSTANCE_ID"
REGION = "YOUR_REGION"  # e.g., "us", "europe", "asia"

# Log Details
LOG_TYPE = "YOUR_LOG_TYPE"  # e.g., "WINEVTLOG", "NIX_SYSTEM", "AIDE"
SAMPLE_LOG_TEXT = "Jan 01 12:00:00Z generic-host test-process[123]: This is a sample log entry."

# =====================================================================

def get_access_token(key_file_path):
    """
    Authenticates using the provided Service Account JSON key and 
    returns a short-lived OAuth 2.0 access token.
    """
    try:
        credentials = service_account.Credentials.from_service_account_file(
            key_file_path,
            scopes=['https://www.googleapis.com/auth/cloud-platform']
        )
        credentials.refresh(Request())
        return credentials.token
    except Exception as e:
        print(f"❌ Failed to obtain access token: {e}")
        raise

def ingest_log_to_secops():
    """
    Constructs the payload and sends the log to the Google SecOps Ingestion API.
    """
    try:
        # 1. Authenticate
        print("Authenticating service account...")
        access_token = get_access_token(KEY_FILE_PATH)

        # 2. Construct Endpoint (Regional v1beta)
        base_url = f"https://{REGION}-chronicle.googleapis.com/v1beta"
        parent_path = f"projects/{PROJECT_ID}/locations/{REGION}/instances/{INSTANCE_ID}/logTypes/{LOG_TYPE}"
        endpoint = f"{base_url}/{parent_path}/logs:import"

        # 3. Encode Log Text to Base64
        # Note: The v1beta 'data' field strictly requires base64 encoded strings.
        log_bytes = SAMPLE_LOG_TEXT.encode("utf-8")
        base64_log = base64.b64encode(log_bytes).decode("utf-8")

        # 4. Prepare Payload
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }

        payload = {
            "inlineSource": {
                "logs": [
                    {
                        "data": base64_log
                        # Optional: Add a specific timestamp if the log lacks one
                        # "logEntryTime": "2024-01-01T12:00:00Z" 
                    }
                ]
            }
        }

        # 5. Send Request
        print(f"Sending log to SecOps region: {REGION.upper()}...")
        response = requests.post(endpoint, headers=headers, json=payload)

        # 6. Output Results
        if response.status_code == 200:
            print("\n✅ SUCCESS: Log ingested successfully.")
            print("API Response:", json.dumps(response.json(), indent=2))
            print(f"\nNext Step: Go to your SecOps Search and query: metadata.log_type = \"{LOG_TYPE}\"")
        else:
            print(f"\n❌ ERROR: Ingestion failed (HTTP Status: {response.status_code})")
            print("Response Body:", response.text)

    except Exception as e:
        print(f"\n❌ EXCEPTION: {str(e)}")

if __name__ == "__main__":
    ingest_log_to_secops()