Just built a script to auto-create risks from cloud config drift tools.

Alright, listen up. The GRC team keeps banging on about "real-time risk visibility" from our cloud environments, but their manual process for turning CSPM findings into ServiceNow risk records was slower than a frozen pipe. They'd get a weekly spreadsheet from the cloud security team, manually triage, and then click through the ServiceNow UI. By the time a risk was logged, the resource had been compromised, fixed, or deleted three times over.

So I did what any sane engineer would do: built a scrappy pipeline that sucks in findings from tools like Wiz, Prisma Cloud, or even plain old AWS Config, and auto-creates Risk records in ServiceNow GRC. It's not pretty, but it works and it's event-driven. The key is mapping the severity and context from the drift tool to ServiceNow's risk scoring model. You'll need to tweak the mappings for your own risk matrix.

Here's the core of the ingestion script. It sits in a Lambda (or a container in your pipeline du jour) triggered by the CSPM tool's webhook or an SQS queue. The script uses the ServiceNow REST API and handles the field mapping.

```python
import json
import os
import requests
from typing import Dict, Any

SERVICE_NOW_INSTANCE = os.environ['SNOW_INSTANCE']
SERVICE_NOW_USER = os.environ['SNOW_USER']
SERVICE_NOW_PASS = os.environ['SNOW_PASS']

# Mappings from CSPM severity to ServiceNow GRC Risk Impact/Risk Probability
SEVERITY_MAPPING = {
"CRITICAL": {"impact": "1", "probability": "1"},
"HIGH": {"impact": "2", "probability": "2"},
"MEDIUM": {"impact": "3", "probability": "3"},
"LOW": {"impact": "4", "probability": "4"},
}

def cspm_finding_to_snow_risk(finding: Dict[str, Any]) -> Dict[str, Any]:
"""Transform a generic CSPM finding into a ServiceNow Risk payload."""
snow_risk = {
"short_description": f"{finding.get('provider')} - {finding.get('finding_type')} on {finding.get('resource_id')}",
"description": finding.get('description', ''),
"u_cloud_resource_id": finding.get('resource_id'),
"u_cspm_finding_id": finding.get('finding_id'),
"u_remediation_guidance": finding.get('remediation', ''),
"u_impact": SEVERITY_MAPPING.get(finding.get('severity', 'LOW'), {}).get('impact', '4'),
"u_probability": SEVERITY_MAPPING.get(finding.get('severity', 'LOW'), {}).get('probability', '4'),
"u_risk_owner": finding.get('owner_group', 'Cloud Security'),
}
return {k: v for k, v in snow_risk.items() if v is not None}

def create_snow_risk(risk_payload: Dict[str, Any]) -> str:
"""POST the risk payload to ServiceNow GRC's Risk table."""
url = f"https://{SERVICE_NOW_INSTANCE}/api/now/table/u_risk_register"
headers = {"Content-Type": "application/json", "Accept": "application/json"}
response = requests.post(url, auth=(SERVICE_NOW_USER, SERVICE_NOW_PASS),
headers=headers, data=json.dumps(risk_payload))
response.raise_for_status()
return response.json().get('result', {}).get('sys_id')

def lambda_handler(event, context):
for record in event.get('Records', []):
# Parse your CSPM tool's event format here
raw_finding = json.loads(record['body'])
normalized_finding = {
"provider": raw_finding.get('cloudProvider'),
"finding_type": raw_finding.get('type'),
"resource_id": raw_finding.get('resourceId'),
"severity": raw_finding.get('severity'),
"description": raw_finding.get('detail', {}).get('description'),
"finding_id": raw_finding.get('id'),
"remediation": raw_finding.get('remediation'),
}
risk_payload = cspm_finding_to_snow_risk(normalized_finding)
risk_sys_id = create_snow_risk(risk_payload)
print(f"Created Risk record: {risk_sys_id}")
```

**Pitfalls and considerations:**

* **Field Mapping:** This is the real grunt work. You'll need to map to your specific ServiceNow GRC Risk table's custom fields (`u_impact`, `u_probability`, etc.). The example above uses a generic structure.
* **Duplication:** You need a de-duplication strategy. I check for an existing open risk with the same `u_cspm_finding_id` before creating.
* **State Synchronization:** When the CSPM tool marks a finding as resolved, you should probably close or update the corresponding risk. This script doesn't handle that; you'd need a separate sync process.
* **Ownership:** Auto-assigning risk owners is a political minefield. We default to a central Cloud Security group, but your mileage will vary.
* **Cost:** This is cheap to run, but remember you're hitting ServiceNow API limits. Batch if you have high volume.

It's not a silver bullet, but it shaved two weeks off their "risk identification" timeline. Now if only we could automate the endless compliance audit evidence collection... but that's a horror story for another day.

-- old salt