SecOps API Wrapper SDK for Python

New version released https://pypi.org/project/secops/0.1.2/

Features for Rule Management: 

The SDK provides comprehensive support for managing Chronicle detection rules:

Create new detection rules using YARA-L 2.0 syntax:

rule_text = """
rule simple_network_rule {
    meta:
        description = "Example rule to detect network connections"
        author = "SecOps SDK Example"
        severity = "Medium"
        priority = "Medium"
        yara_version = "YL2.0"
        rule_version = "1.0"
    events:
        $e.metadata.event_type = "NETWORK_CONNECTION"
        $e.principal.hostname != ""
    condition:
        $e
}
"""

# Create the rule
rule = chronicle.create_rule(rule_text)
rule_id = rule.get("name", "").split("/")[-1]
print(f"Rule ID: {rule_id}")

Retrieve, list, update, enable/disable, and delete rules:

# List all rules
rules = chronicle.list_rules()
for rule in rules.get("rules", []):
    rule_id = rule.get("name", "").split("/")[-1]
    enabled = rule.get("deployment", {}).get("enabled", False)
    print(f"Rule ID: {rule_id}, Enabled: {enabled}")

# Get specific rule
rule = chronicle.get_rule(rule_id)
print(f"Rule content: {rule.get('text')}")

# Update rule
updated_rule = chronicle.update_rule(rule_id, updated_rule_text)

# Enable/disable rule
deployment = chronicle.enable_rule(rule_id, enabled=True)  # Enable
deployment = chronicle.enable_rule(rule_id, enabled=False) # Disable

# Delete rule
chronicle.delete_rule(rule_id)

Run rules against historical data to find past matches:

from datetime import datetime, timedelta, timezone

# Set time range for retrohunt
end_time = datetime.now(timezone.utc)
start_time = end_time - timedelta(days=7)  # Search past 7 days

# Create retrohunt
retrohunt = chronicle.create_retrohunt(rule_id, start_time, end_time)
operation_id = retrohunt.get("name", "").split("/")[-1]

# Check retrohunt status
retrohunt_status = chronicle.get_retrohunt(rule_id, operation_id)
is_complete = retrohunt_status.get("metadata", {}).get("done", False)

Monitor rule detections and execution errors:

# List detections for a rule
detections = chronicle.list_detections(rule_id)
for detection in detections.get("detections", []):
    detection_id = detection.get("id", "")
    event_time = detection.get("eventTime", "")
    alerting = detection.get("alertState", "") == "ALERTING"
    print(f"Detection: {detection_id}, Time: {event_time}, Alerting: {alerting}")

# List execution errors for a rule
errors = chronicle.list_errors(rule_id)
for error in errors.get("ruleExecutionErrors", []):
    error_message = error.get("error_message", "")
    create_time = error.get("create_time", "")
    print(f"Error: {error_message}, Time: {create_time}")

Search for alerts generated by rules:

# Set time range for alert search
end_time = datetime.now(timezone.utc)
start_time = end_time - timedelta(days=7)  # Search past 7 days

# Search for rule alerts
alerts_response = chronicle.search_rule_alerts(
    start_time=start_time,
    end_time=end_time,
    page_size=10
)

# The API returns a nested structure where alerts are grouped by rule
# Extract and process all alerts from this structure
all_alerts = []
too_many_alerts = alerts_response.get('tooManyAlerts', False)

# Process the nested response structure - alerts are grouped by rule
for rule_alert in alerts_response.get('ruleAlerts', []):
    # Extract rule metadata
    rule_metadata = rule_alert.get('ruleMetadata', {})
    rule_id = rule_metadata.get('properties', {}).get('ruleId', 'Unknown')
    rule_name = rule_metadata.get('properties', {}).get('name', 'Unknown')
    
    # Get alerts for this rule
    rule_alerts = rule_alert.get('alerts', [])
    
    # Process each alert
    for alert in rule_alerts:
        # Extract important fields
        alert_id = alert.get("id", "")
        detection_time = alert.get("detectionTimestamp", "")
        commit_time = alert.get("commitTimestamp", "")
        alerting_type = alert.get("alertingType", "")
        
        print(f"Alert ID: {alert_id}")
        print(f"Rule ID: {rule_id}")
        print(f"Rule Name: {rule_name}")
        print(f"Detection Time: {detection_time}")
        
        # Extract events from the alert
        if 'resultEvents' in alert:
            for var_name, event_data in alert.get('resultEvents', {}).items():
                if 'eventSamples' in event_data:
                    for sample in event_data.get('eventSamples', []):
                        if 'event' in sample:
                            event = sample['event']
                            # Process event data
                            event_type = event.get('metadata', {}).get('eventType', 'Unknown')
                            print(f"Event Type: {event_type}")

If tooManyAlerts is True in the response, consider narrowing your search criteria using a smaller time window or more specific filters.

Manage curated rule sets:

# Define deployments for rule sets
deployments = [
    {
        "category_id": "category-uuid",
        "rule_set_id": "ruleset-uuid",
        "precision": "broad",
        "enabled": True,
        "alerting": False
    }
]

# Update rule set deployments
chronicle.batch_update_curated_rule_set_deployments(deployments)

Validate a YARA-L2 rule before creating or updating it:

# Example rule
rule_text = """
rule test_rule {
    meta:
        description = "Test rule for validation"
        author = "Test Author"
        severity = "Low"
        yara_version = "YL2.0"
        rule_version = "1.0"
    events:
        $e.metadata.event_type = "NETWORK_CONNECTION"
    condition:
        $e
}
"""

# Validate the rule
result = chronicle.validate_rule(rule_text)

if result.success:
    print("Rule is valid")
else:
    print(f"Rule is invalid: {result.message}")
    if result.position:
        print(f"Error at line {result.position['startLine']}, column {result.position['startColumn']}")

Looks very interesting @raybrian - I'm about to start testing the use of the get_stats method which I can see results in a json output. 

I noticed you had a separate action that returns the results of a udm query in csv format - are we able to run stats searches and have them returned in csv format?

I did do a test and I'm getting an error when running the get_stats method:

 <p>The requested URL <code>/v1alpha/projects/<project_id>/locations/<region>/instances/<customer_id>/legacy:legacyFetchUdmSearchView</code> was not found on this server.  <ins>That’s all we know.</ins>

 I have of course removed project_id, customer_id and region values from the url. 

The one thing I'm unsure of is region - is there a list of potential region values (the wrapper uses "us" as an example)?

Awesome. We would love your feedback. The CSV Format is it's own API for UDM search that doesn't support stats search today. It would be really easy to wrap the stats query to get you what you're looking for: 

It depends on which region your instance is running in, the choices are:

US

EUROPE

ASIA_SOUTHEAST1

EUROPE_WEST2

AUSTRALIA_SOUTHEAST1

ME_WEST1

EUROPE_WEST6

EUROPE_WEST3

ME_CENTRAL2

ASIA_SOUTH1

ASIA_NORTHEAST1

NORTHAMERICA_NORTHEAST2

EUROPE_WEST12

ME_CENTRAL1

SOUTHAMERICA_EAST1

EUROPE_WEST9

New update for 0.1.3. Now you can pass a natural language search and get UDM events back

New update for 0.1.4. Now supporting log ingestion, forwarder management, and log type discovery.
https://github.com/google/secops-wrapper?tab=readme-ov-file#log-ingestion

@raybrian I have performed a test stats search using a service account that was created to integrate Google Sec Ops SIEM To SOAR. I'm seeing an IAM permissions error being returned:

status": "PERMISSION_DENIED", "details": [ { "@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "IAM_PERMISSION_DENIED", "domain": "chronicle.googleapis.com", "metadata": { "resource": "projects/xxxxxx", "permission": "chronicle.legacies.legacyFetchUdmSearchView"

Are there specific permissions that must be provided to the service account in order for the various actions to work?

The service account needs permission to read Chronicle API in the "BYOP" project associated with your account. Some tenants might not have migrated to the new API, or you could be using a service account associated with the older APIs. If you DM me your chronicle tenant URL and region, I can look at the details for you.

Thanks @raybrian  - I dont seem to have the ability to DM which I've created a separate post about. Once that is resolved, I'll reach out, thanks!

Thanks so much for this, @raybrian. The wrapper and your examples are very useful.

Just wanted to call out that if a service account needs to be created to use the wrapper I think the minimal role required for the service account is Chronicle API Viewer. @raybrian please correct me if I'm wrong.

Version 0.1.5 released with UDM Event Ingestion

result = chronicle.ingest_udm(udm_events=network_event)

New update

New update today includes access to specialized Gemini AI functions to get security insights, generate detection rules, explain security concepts, and more.

https://github.com/google/secops-wrapper?tab=readme-ov-file#gemini-ai

@raybrian I just read in this blog that that the UDM Entity Search is coming into preview: UDM Entity Search. In this post, I’ll explore the new… | by Chris Martin (@thatsiemguy) | Apr, 2025 | Medium

I know the SecOps SDK wrapper support stats UDM Events searches, but does it support stats UDM Entity Searches?

It uses the same API as entity search, so it should work. Want to test and let us know? Otherwise I can try it later this week. (Super slammed with RSA this week)

hello! very much interested in using the sdk to automatically download cases. i'm having issues with example 6 in the given examples of the sdk. are you seeingthe same issue? thank you.

thanks for your report. I fixed this in version 0.4.1. Please update

it works! thank you @raybrian 

a few more questions: for cases, seems that we can only pull given a list of alerts. are we able to pull created cases given a start and end time? lastly, is there a way to output cases as JSON? thank you.

Are you using SOAR? Most of the case management is on the SOAR side

Thanks for the direction! I'll have to look into SOAR (and find the Google doc).

​@raybrian Do you know if you can specify a log_type using the ingest_udm? I am not finding any documentation. The events are just showing up with a UDM as the log_type.

The example for ingest_log a log_type is defined.

labeled_result = chronicle.ingest_log(
    log_type="OKTA",
    log_message=json.dumps(okta_log),
    labels={"environment": "production", "app": "web-portal", "team": "security"}

The example for ingest_udm does not have a log_type definded

result = chronicle.ingest_udm(udm_events=[network_event, process_event])

Does the log_type need to be in the UDM payload?

{
  "log_type": "Custom",
  "events": []
}

labeled_result = chronicle.ingest_log(
    log_type="OKTA",
    log_message=json.dumps(okta_log),
    labels={"environment": "production", "app": "web-portal", "team": "security"}

The example for ingest_udm does not have a log_type definded

result = chronicle.ingest_udm(udm_events=[network_event, process_event])

Does the log_type need to be in the UDM payload?

{
  "log_type": "Custom",
  "events": []
}

You can specify log_type for each event using `metadata`. 

The example for passing log_type in UDM:​​​​​​

events = [
    {
        "metadata": {
            "log_type": "Custom",
            "event_type": "NETWORK_CONNECTION",
            "product_name": "My Security Product"
        },
        "principal": {"ip": "192.168.1.100"},
        "target": {"ip": "10.0.0.1"}
    },
    {
        "metadata": {
            "log_type": "Custom",
            "event_type": "PROCESS_LAUNCH",
            "product_name": "My Security Product"
        },
        "principal": {
            "hostname": "workstation1",
            "process": {"command_line": "./malware.exe"}
        }
    }
]

result = chronicle.ingest_udm(events)

More detail on fields available for event can be found here.

I think since UDM events are processed entity, there is no option to specify log type to associate with all events (similar to log ingestion) as each event could be associated with different log types.