How to Automate SOC2 and GDPR Compliance Scans with ComplianceWeave

---
title: "I Automated My SOC2 Audit Prep in an Afternoon (And You Can Too)"
published: false
tags: [security, python, devops, compliance]
---

# I Automated My SOC2 Audit Prep in an Afternoon (And You Can Too)

Let me paint you a picture.

It's 11 PM. Your audit is in three weeks. You're cross-referencing a spreadsheet with 847 rows against cloud console screenshots you took last Tuesday, hoping nothing changed since then. Your coffee is cold. Your eyes hurt. Your auditor just emailed asking for "just a few more controls."

I've been there. Most of us have.

This tutorial is about escaping that reality using **ComplianceWeave** — a continuous compliance monitoring tool that covers SOC2, GDPR, HIPAA, and ISO 27001. By the end, you'll have a Python script that scans your infrastructure, pulls audit-ready reports, and even kicks off remediation — all without touching a single spreadsheet.

Let's build something useful.

---

## Prerequisites

- Python 3.9+
- A ComplianceWeave account and API key
- `requests` and `python-dotenv` installed (`pip install requests python-dotenv`)
- An infrastructure worth auditing (relatable)

Store your API key safely:

bash

.env

COMPLIANCEWEAVE_API_KEY=your_api_key_here
COMPLIANCEWEAVE_BASE_URL=https://api.complianceweave.io/v1

---

## Step 1: Build Your Client Foundation

Before touching any endpoints, let's write a reusable client. Future-you will appreciate this when you're wiring this into a CI/CD pipeline at 9 AM instead of a spreadsheet at 11 PM.

python

compliance_client.py

import os
import time
import requests
from dotenv import load_dotenv

load_dotenv()

class ComplianceWeaveClient:
def init(self):
self.api_key = os.getenv("COMPLIANCEWEAVE_API_KEY")
self.base_url = os.getenv("COMPLIANCEWEAVE_BASE_URL")

    if not self.api_key:
        raise EnvironmentError(
            "COMPLIANCEWEAVE_API_KEY not set. "
            "Check your .env file before your auditor does."
        )

    self.session = requests.Session()
    self.session.headers.update({
        "Authorization": f"Bearer {self.api_key}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    })

def _request(self, method, endpoint, **kwargs):
    url = f"{self.base_url}{endpoint}"

    try:
        response = self.session.request(method, url, timeout=30, **kwargs)
        response.raise_for_status()
        return response.json()

    except requests.exceptions.Timeout:
        raise RuntimeError(f"Request to {endpoint} timed out. Your infrastructure might be large — try again.")

    except requests.exceptions.HTTPError as e:
        status = e.response.status_code
        if status == 401:
            raise PermissionError("Invalid API key. Rotate it in your ComplianceWeave dashboard.")
        elif status == 429:
            # Respect rate limits — be a good API citizen
            retry_after = int(e.response.headers.get("Retry-After", 60))
            print(f"Rate limited. Waiting {retry_after}s...")
            time.sleep(retry_after)
            return self._request(method, endpoint, **kwargs)
        else:
            raise RuntimeError(f"API error {status}: {e.response.text}")

    except requests.exceptions.ConnectionError:
        raise RuntimeError("Can't reach ComplianceWeave API. Check your network or their status page.")

This client handles the three failure modes that will actually bite you: timeouts on large infrastructure scans, expired API keys, and rate limits during bulk operations.

---

## Step 2: Trigger Your First Compliance Scan

Here's where the magic starts. `POST /compliance/scan` kicks off a scan against your chosen frameworks. We'll target SOC2 and GDPR — the two that tend to generate the most auditor paperwork.

python

In compliance_client.py, add this method to ComplianceWeaveClient:

def trigger_scan(self, frameworks: list[str], scope: dict = None) -> dict:
    """
    Trigger a compliance scan.

    frameworks: List from ["SOC2", "GDPR", "HIPAA", "ISO27001"]
    scope: Optional dict to target specific resources
    """
    valid_frameworks = {"SOC2", "GDPR", "HIPAA", "ISO27001"}
    invalid = set(frameworks) - valid_frameworks
    if invalid:
        raise ValueError(f"Unknown frameworks: {invalid}. Valid options: {valid_frameworks}")

    payload = {
        "frameworks": frameworks,
        "scope": scope or {"include": "all"}
    }

    print(f"🔍 Starting scan for: {', '.join(frameworks)}")
    result = self._request("POST", "/compliance/scan", json=payload)

    scan_id = result.get("scan_id")
    print(f"✅ Scan initiated. ID: {scan_id}")
    print(f"   Estimated completion: {result.get('estimated_duration', 'unknown')}")

    return result

--- Run it ---

if name == "main":
client = ComplianceWeaveClient()

scan = client.trigger_scan(
    frameworks=["SOC2", "GDPR"],
    scope={"regions": ["us-east-1", "eu-west-1"]}
)
print(f"\nScan response: {scan}")

**Expected output:**

json
🔍 Starting scan for: SOC2, GDPR
✅ Scan initiated. ID: scan_a3f92b1c
Estimated completion: 8-12 minutes

Scan response: {
"scan_id": "scan_a3f92b1c",
"status": "running",
"frameworks": ["SOC2", "GDPR"],
"estimated_duration": "8-12 minutes",
"created_at": "2024-11-14T14:23:01Z"
}

---

## Step 3: Fetch Your Audit-Ready Reports

Once the scan completes, `GET /compliance/reports` hands you structured evidence. Let's add polling logic — because nobody wants to manually refresh an endpoint.

python

Add to ComplianceWeaveClient:

def get_reports(self, scan_id: str, poll: bool = True, poll_interval: int = 30) -> dict:
    """
    Retrieve compliance reports. Optionally polls until scan completes.
    """
    endpoint = f"/compliance/reports?scan_id={scan_id}"

    if not poll:
        return self._request("GET", endpoint)

    print(f"⏳ Waiting for scan {scan_id} to complete...")
    attempts = 0
    max_attempts = 40  # ~20 minutes max wait

    while attempts < max_attempts:
        result = self._request("GET", endpoint)
        status = result.get("status")

        if status == "completed":
            findings = result.get("findings", [])
            summary = result.get("summary", {})

            print(f"\n📊 Scan Complete!")
            print(f"   Total controls checked: {summary.get('total_controls', 0)}")
            print(f"   ✅ Passing: {summary.get('passing', 0)}")
            print(f"   ❌ Failing: {summary.get('failing', 0)}")
            print(f"   ⚠️  Warnings: {summary.get('warnings', 0)}")

            # Surface the critical failures immediately
            critical = [f for f in findings if f.get("severity") == "critical"]
            if critical:
                print(f"\n🚨 Critical issues requiring immediate attention:")
                for issue in critical[:5]:  # Top 5
                    print(f"   - [{issue['framework']}] {issue['control_id']}: {issue['description']}")

            return result

        elif status == "failed":
            raise RuntimeError(f"Scan failed: {result.get('error', 'Unknown error')}")

        attempts += 1
        print(f"   Still scanning... ({attempts * poll_interval}s elapsed)", end="\r")
        time.sleep(poll_interval)

    raise TimeoutError("Scan didn't complete within expected window. Check the dashboard.")

--- Run it ---

report = client.get_reports(scan["scan_id"], poll=True)

**Expected output:**

plaintext
⏳ Waiting for scan scan_a3f92b1c to complete...
Still scanning... (270s elapsed)

📊 Scan Complete!
Total controls checked: 214
✅ Passing: 189
❌ Failing: 18
⚠️ Warnings: 7

🚨 Critical issues requiring immediate attention:

• [SOC2] CC6.1: Encryption at rest not enabled on 3 RDS instances
• [GDPR] Art.32: Data transfer logging disabled in eu-west-1

---

## Step 4: Automate Remediation for Low-Hanging Fruit

This is the part that makes auditors nervous and engineers happy. `POST /compliance/remediate` can fix certain findings automatically — think misconfigured settings, missing tags, disabled logging.

python

Add to ComplianceWeaveClient:

def remediate(self, finding_ids: list[str], dry_run: bool = True) -> dict:
    """
    Attempt automated remediation.

    Always dry_run=True first. Always.
    """
    if not finding_ids:
        print("No findings to remediate.")
        return {}

    payload = {
        "finding_ids": finding_ids,
        "dry_run": dry_run
    }

    mode = "DRY RUN" if dry_run else "LIVE"
    print(f"🔧 Running remediation [{mode}] on {len(finding_ids)} findings...")

    result = self._request("POST", "/compliance/remediate", json=payload)

    for action in result.get("actions", []):
        status_icon = "✅" if action["status"] == "success" else "❌"
        print(f"   {status_icon} {action['finding_id']}: {action['action_taken']}")
        if action.get("requires_manual_review"):
            print(f"      ⚠️  Manual review required: {action['reason']}")

    return result

--- Full workflow ---

if name == "main":
client = ComplianceWeaveClient()

# 1. Scan
scan = client.trigger_scan(frameworks=["SOC2", "GDPR"])

# 2. Get report
report = client.get_reports(scan["scan_id"])

# 3. Find auto-remediable issues (low severity only — don't auto-fix critical)
auto_fix_candidates = [
    f["finding_id"] 
    for f in report.get("findings", [])
    if f.get("auto_remediable") and f.get("severity") in ("low", "medium")
]

# 4. Dry run first, then flip to dry_run=False when you're confident
if auto_fix_candidates:
    client.remediate(auto_fix_candidates, dry_run=True)

**Expected output:**

plaintext
🔧 Running remediation [DRY RUN] on 11 findings...
✅ find_b291: Would enable CloudTrail logging in us-east-1
✅ find_c847: Would apply required data classification tags to S3 buckets
❌ find_d103: Cannot auto-remediate — requires IAM policy approval
⚠️ Manual review required: Privilege escalation risk

---

## Putting It All Together

Here's your complete audit automation script — the one you'll actually commit to your repo:

python

audit_runner.py

from compliance_client import ComplianceWeaveClient
import json
from datetime import datetime

def run_audit(frameworks=None, output_file=None):
frameworks = frameworks or ["SOC2", "GDPR", "HIPAA", "ISO27001"]
client = ComplianceWeaveClient()

scan = client.trigger_scan(frameworks=frameworks)
report = client.get_reports(scan["scan_id"])

# Save full report for your auditor
if output_file:
    with open(output_file, "w") as f:
        json.dump(report, f, indent=2)
    print(f"\n📁 Full report saved to {output_file}")

# Auto-remediate safe findings
candidates = [
    f["finding_id"] for f in report.get("findings", [])
    if f.get("auto_remediable") and f.get("severity") == "low"
]

if candidates:
    client.remediate(candidates, dry_run=False)

return report

if name == "main":
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
run_audit(output_file=f"audit_report_{timestamp}.json")

Schedule this with cron or a GitHub Actions workflow, and you've just transformed compliance from a quarterly panic into background infrastructure.

---

## What You Built

In ~150 lines of Python, you now have:

- **Continuous scanning** across four major compliance frameworks
- **Structured, auditor-ready reports** pulled programmatically
- **Automated remediation** for safe, low-risk findings
- **Proper error handling** for the production realities of timeouts, rate limits, and auth failures

The next time an auditor emails asking for evidence, you send them a JSON file and a timestamp. That's the goal.

ComplianceWeave's API is designed to slot into existing workflows — CI/CD pipelines, scheduled jobs, Slack alerting, whatever fits your stack. The hard part isn't the code. It was always the manual evidence collection. Now that's handled.

Go get some sleep.