Benchmark: Vanta 2.0 vs Drata 2.0 vs Compliancy 1.0 for SOC 2 Automation

import os
import time
import logging
from typing import List, Dict, Any
import boto3
from vanta_sdk import VantaClient, VantaError, EvidenceType
from kubernetes import client, config

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.FileHandler(\"vanta_eks_evidence.log\"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Retry decorator for transient AWS/K8s API errors
def retry_on_error(max_retries: int = 3, delay: int = 2):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logger.warning(f\"Attempt {attempt + 1} failed: {str(e)}\")
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay * (2 ** attempt))
            return None
        return wrapper
    return decorator

class EKSSecurityEvidenceCollector:
    def __init__(self, vanta_api_key: str, cluster_name: str, region: str = \"us-east-1\"):
        self.vanta_client = VantaClient(api_key=vanta_api_key)
        self.cluster_name = cluster_name
        self.region = region
        # Initialize AWS and K8s clients
        self.eks_client = boto3.client(\"eks\", region_name=region)
        self.iam_client = boto3.client(\"iam\", region_name=region)
        # Load in-cluster K8s config (runs inside EKS pod) or local config
        try:
            config.load_incluster_config()
        except:
            config.load_kube_config()
        self.k8s_client = client.ApiClient()
        self.rbac_api = client.RbacAuthorizationV1Api(self.k8s_client)
        logger.info(f\"Initialized collector for EKS cluster {cluster_name}\")

    @retry_on_error(max_retries=3)
    def get_pod_security_policies(self) -> List[Dict[str, Any]]:
        \\\"\\\"\\\"Collect Pod Security Policies for CC6.1 compliance\\\"\\\"\\\"
        try:
            psps = self.rbac_api.list_pod_security_policy()
            return [{\"name\": psp.metadata.name, \"spec\": psp.spec.to_dict()} for psp in psps.items]
        except Exception as e:
            logger.error(f\"Failed to list PSPs: {str(e)}\")
            raise

    @retry_on_error(max_retries=3)
    def submit_evidence_to_vanta(self, evidence: List[Dict[str, Any]]) -> str:
        \\\"\\\"\\\"Submit collected evidence to Vanta 2.0 for SOC 2 control CC6.1\\\"\\\"\\\"
        try:
            response = self.vanta_client.evidence.submit(
                control_id=\"CC6.1\",
                evidence_type=EvidenceType.KUBERNETES_CONFIG,
                resource_id=f\"eks:{self.cluster_name}:psp\",
                payload={\"pod_security_policies\": evidence},
                metadata={\"collection_time\": time.time(), \"collector_version\": \"1.0.0\"}
            )
            logger.info(f\"Submitted evidence to Vanta: {response.evidence_id}\")
            return response.evidence_id
        except VantaError as e:
            logger.error(f\"Vanta API error: {str(e)}\")
            raise
        except Exception as e:
            logger.error(f\"Unexpected error submitting evidence: {str(e)}\")
            raise

if __name__ == \"__main__\":
    # Load environment variables (never hardcode credentials)
    vanta_api_key = os.getenv(\"VANTA_API_KEY\")
    cluster_name = os.getenv(\"EKS_CLUSTER_NAME\", \"prod-saas-cluster\")
    region = os.getenv(\"AWS_REGION\", \"us-east-1\")

    if not vanta_api_key:
        logger.error(\"Missing VANTA_API_KEY environment variable\")
        exit(1)

    collector = EKSSecurityEvidenceCollector(
        vanta_api_key=vanta_api_key,
        cluster_name=cluster_name,
        region=region
    )

    try:
        logger.info(\"Collecting Pod Security Policies from EKS...\")
        psps = collector.get_pod_security_policies()
        logger.info(f\"Collected {len(psps)} PSPs\")
        evidence_id = collector.submit_evidence_to_vanta(psps)
        print(f\"Successfully submitted evidence: {evidence_id}\")
    except Exception as e:
        logger.error(f\"Evidence collection failed: {str(e)}\")
        exit(1)

import os
import time
import json
import logging
from typing import List, Dict, Any
from datetime import datetime, timedelta
from google.cloud import logging as gcp_logging
from google.oauth2 import service_account
from drata_sdk import DrataClient, DrataError, EvidenceFormat

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.FileHandler(\"drata_gcp_evidence.log\"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class GCPCloudSQLAccessCollector:
    def __init__(self, drata_api_key: str, project_id: str, instance_id: str, credentials_path: str = None):
        self.drata_client = DrataClient(api_key=drata_api_key)
        self.project_id = project_id
        self.instance_id = instance_id
        # Initialize GCP logging client with explicit credentials if provided
        if credentials_path:
            credentials = service_account.Credentials.from_service_account_file(
                credentials_path,
                scopes=[\"https://www.googleapis.com/auth/logging.read\"]
            )
            self.gcp_client = gcp_logging.Client(credentials=credentials, project=project_id)
        else:
            self.gcp_client = gcp_logging.Client(project=project_id)
        logger.info(f\"Initialized collector for GCP Cloud SQL instance {instance_id} in project {project_id}\")

    def _build_log_filter(self, start_time: datetime, end_time: datetime) -> str:
        \\\"\\\"\\\"Construct GCP log filter for Cloud SQL access events\\\"\\\"\\\"
        return f\"\"\"
        resource.type=\"cloudsql_database\"
        resource.labels.database_id=\"{self.project_id}:{self.instance_id}\"
        logName=\"projects/{self.project_id}/logs/cloudsql.googleapis.com%2Fmysql-general.log\"
        timestamp>=\"{start_time.isoformat()}Z\"
        timestamp<=\"{end_time.isoformat()}Z\"
        \"\"\"

    def collect_access_logs(self, hours_ago: int = 24) -> List[Dict[str, Any]]:
        \\\"\\\"\\\"Collect Cloud SQL access logs for CC2.1 (access control) compliance\\\"\\\"\\\"
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=hours_ago)
        log_filter = self._build_log_filter(start_time, end_time)
        logger.info(f\"Collecting logs from {start_time} to {end_time}\")

        logs = []
        try:
            for entry in self.gcp_client.list_entries(filter_=log_filter):
                log_payload = {
                    \"timestamp\": entry.timestamp.isoformat(),
                    \"severity\": entry.severity,
                    \"log_name\": entry.log_name,
                    \"resource\": entry.resource.to_dict(),
                    \"payload\": entry.payload if isinstance(entry.payload, dict) else json.loads(entry.payload)
                }
                logs.append(log_payload)
            logger.info(f\"Collected {len(logs)} log entries\")
            return logs
        except Exception as e:
            logger.error(f\"Failed to collect GCP logs: {str(e)}\")
            raise

    def submit_evidence_to_drata(self, logs: List[Dict[str, Any]]) -> str:
        \\\"\\\"\\\"Submit evidence to Drata 2.0 for SOC 2 control CC2.1\\\"\\\"\\\"
        try:
            # Drata requires evidence in JSON Lines format for log-based controls
            evidence_payload = \"\\n\".join([json.dumps(log) for log in logs])
            response = self.drata_client.evidence.submit(
                control_id=\"CC2.1\",
                evidence_format=EvidenceFormat.JSON_LINES,
                resource_id=f\"gcp:cloudsql:{self.instance_id}\",
                payload=evidence_payload,
                metadata={
                    \"collection_start\": start_time.isoformat(),
                    \"collection_end\": end_time.isoformat(),
                    \"log_count\": len(logs)
                }
            )
            logger.info(f\"Submitted evidence to Drata: {response.evidence_id}\")
            return response.evidence_id
        except DrataError as e:
            logger.error(f\"Drata API error: {str(e)}\")
            raise
        except Exception as e:
            logger.error(f\"Unexpected error submitting evidence: {str(e)}\")
            raise

if __name__ == \"__main__\":
    drata_api_key = os.getenv(\"DRATA_API_KEY\")
    project_id = os.getenv(\"GCP_PROJECT_ID\")
    instance_id = os.getenv(\"CLOUDSQL_INSTANCE_ID\")
    credentials_path = os.getenv(\"GCP_CREDENTIALS_PATH\")

    if not all([drata_api_key, project_id, instance_id]):
        logger.error(\"Missing required environment variables: DRATA_API_KEY, GCP_PROJECT_ID, CLOUDSQL_INSTANCE_ID\")
        exit(1)

    collector = GCPCloudSQLAccessCollector(
        drata_api_key=drata_api_key,
        project_id=project_id,
        instance_id=instance_id,
        credentials_path=credentials_path
    )

    try:
        logs = collector.collect_access_logs(hours_ago=24)
        if logs:
            evidence_id = collector.submit_evidence_to_drata(logs)
            print(f\"Successfully submitted {len(logs)} log entries: {evidence_id}\")
        else:
            logger.info(\"No access logs found in the last 24 hours\")
    except Exception as e:
        logger.error(f\"Evidence collection failed: {str(e)}\")
        exit(1)

import os
import time
import logging
from typing import List, Dict, Any
from datetime import datetime, timedelta
from azure.identity import DefaultAzureCredential
from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.mgmt.monitor import MonitorManagementClient
from compliancy_sdk import CompliancyClient, CompliancyError, EvidenceType

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.FileHandler(\"compliancy_azure_evidence.log\"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class AzureCosmosThroughputCollector:
    def __init__(self, compliancy_api_key: str, subscription_id: str, resource_group: str, account_name: str):
        self.compliancy_client = CompliancyClient(api_key=compliancy_api_key)
        self.subscription_id = subscription_id
        self.resource_group = resource_group
        self.account_name = account_name
        # Initialize Azure clients with DefaultAzureCredential (supports MSI, CLI, etc.)
        credential = DefaultAzureCredential()
        self.cosmos_client = CosmosDBManagementClient(credential, subscription_id)
        self.monitor_client = MonitorManagementClient(credential, subscription_id)
        logger.info(f\"Initialized collector for Cosmos DB account {account_name} in {resource_group}\")

    def get_throughput_metrics(self, hours_ago: int = 24) -> List[Dict[str, Any]]:
        \\\"\\\"\\\"Collect Cosmos DB throughput metrics for CC7.1 (system availability) compliance\\\"\\\"\\\"
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=hours_ago)
        metrics = []

        try:
            # List all databases in the Cosmos DB account
            databases = self.cosmos_client.sql_resources.list_sql_databases(
                resource_group, self.account_name
            )
            for db in databases:
                db_name = db.name
                # List containers in the database
                containers = self.cosmos_client.sql_resources.list_sql_containers(
                    resource_group, self.account_name, db_name
                )
                for container in containers:
                    container_name = container.name
                    # Get throughput metric for the container
                    metric_data = self.monitor_client.metrics.list(
                        resource_uri=f\"/subscriptions/{self.subscription_id}/resourceGroups/{self.resource_group}/providers/Microsoft.DocumentDB/databaseAccounts/{self.account_name}/sqlDatabases/{db_name}/containers/{container_name}\",
                        timespan=f\"{start_time}/{end_time}\",
                        interval=\"PT1H\",
                        metricnames=\"TotalRequests\",
                        aggregation=\"Total\"
                    )
                    for point in metric_data.value[0].timeseries[0].data:
                        metrics.append({
                            \"database\": db_name,
                            \"container\": container_name,
                            \"timestamp\": point.time_stamp.isoformat(),
                            \"total_requests\": point.total,
                            \"unit\": \"count\"
                        })
            logger.info(f\"Collected {len(metrics)} throughput data points\")
            return metrics
        except Exception as e:
            logger.error(f\"Failed to collect Azure metrics: {str(e)}\")
            raise

    def submit_evidence_to_compliancy(self, metrics: List[Dict[str, Any]]) -> str:
        \\\"\\\"\\\"Submit evidence to Compliancy 1.0 for SOC 2 control CC7.1\\\"\\\"\\\"
        try:
            response = self.compliancy_client.evidence.submit(
                control_id=\"CC7.1\",
                evidence_type=EvidenceType.AZURE_METRIC,
                resource_id=f\"azure:cosmosdb:{self.account_name}\",
                payload=metrics,
                metadata={
                    \"collection_start\": start_time.isoformat(),
                    \"collection_end\": end_time.isoformat(),
                    \"metric_count\": len(metrics)
                }
            )
            logger.info(f\"Submitted evidence to Compliancy: {response.evidence_id}\")
            return response.evidence_id
        except CompliancyError as e:
            logger.error(f\"Compliancy API error: {str(e)}\")
            raise
        except Exception as e:
            logger.error(f\"Unexpected error submitting evidence: {str(e)}\")
            raise

if __name__ == \"__main__\":
    compliancy_api_key = os.getenv(\"COMPLIANCY_API_KEY\")
    subscription_id = os.getenv(\"AZURE_SUBSCRIPTION_ID\")
    resource_group = os.getenv(\"AZURE_RESOURCE_GROUP\")
    account_name = os.getenv(\"COSMOS_ACCOUNT_NAME\")

    if not all([compliancy_api_key, subscription_id, resource_group, account_name]):
        logger.error(\"Missing required environment variables: COMPLIANCY_API_KEY, AZURE_SUBSCRIPTION_ID, AZURE_RESOURCE_GROUP, COSMOS_ACCOUNT_NAME\")
        exit(1)

    collector = AzureCosmosThroughputCollector(
        compliancy_api_key=compliancy_api_key,
        subscription_id=subscription_id,
        resource_group=resource_group,
        account_name=account_name
    )

    try:
        metrics = collector.get_throughput_metrics(hours_ago=24)
        if metrics:
            evidence_id = collector.submit_evidence_to_compliancy(metrics)
            print(f\"Successfully submitted {len(metrics)} metric points: {evidence_id}\")
        else:
            logger.info(\"No throughput metrics found in the last 24 hours\")
    except Exception as e:
        logger.error(f\"Evidence collection failed: {str(e)}\")
        exit(1)

import hashlib

def generate_evidence_hash(payload: List[Dict[str, Any]]) -> str:
    \\\"\\\"\\\"Generate deterministic SHA-256 hash of evidence payload\\\"\\\"\\\"
    payload_str = json.dumps(payload, sort_keys=True).encode(\"utf-8\")
    return hashlib.sha256(payload_str).hexdigest()

# In submit_evidence_to_vanta method:
evidence_hash = generate_evidence_hash(evidence)
response = self.vanta_client.evidence.submit(
    control_id=\"CC6.1\",
    evidence_type=EvidenceType.KUBERNETES_CONFIG,
    resource_id=f\"eks:{self.cluster_name}:psp:{evidence_hash[:8]}\",  # Truncate for readability
    payload={\"pod_security_policies\": evidence},
    metadata={\"collection_time\": time.time(), \"collector_version\": \"1.0.0\", \"payload_hash\": evidence_hash}
)

from flask import Flask, request, jsonify
import hmac
import hashlib
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

DRATA_WEBHOOK_SECRET = os.getenv(\"DRATA_WEBHOOK_SECRET\")

@app.route(\"/drata-webhook\", methods=[\"POST\"])
def handle_drata_webhook():
    # Verify webhook signature to prevent spoofing
    signature = request.headers.get(\"X-Drata-Signature\")
    if not signature:
        logger.error(\"Missing webhook signature\")
        return jsonify({\"error\": \"Unauthorized\"}), 401

    payload = request.get_data()
    expected_signature = hmac.new(
        DRATA_WEBHOOK_SECRET.encode(\"utf-8\"),
        payload,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected_signature):
        logger.error(\"Invalid webhook signature\")
        return jsonify({\"error\": \"Unauthorized\"}), 401

    event = request.json
    logger.info(f\"Received Drata event: {event['event_type']}\")

    if event[\"event_type\"] == \"control.failed\":
        logger.error(f\"Control {event['control_id']} failed: {event['details']}\")
        # Trigger PagerDuty/Opsgenie alert here
    elif event[\"event_type\"] == \"evidence.submitted\":
        logger.info(f\"Evidence {event['evidence_id']} submitted successfully\")

    return jsonify({\"status\": \"success\"}), 200

if __name__ == \"__main__\":
    app.run(port=5000)

# Compliancy 1.0 has no native GCP SDK support, so you must use their generic REST API
import requests

COMPLIANCY_API_BASE = \"https://api.compliancy.io/v1\"

def submit_gcp_evidence_compliancy(api_key: str, control_id: str, evidence: List[Dict[str, Any]]) -> str:
    \\\"\\\"\\\"Submit GCP evidence to Compliancy 1.0 via generic REST API\\\"\\\"\\\"
    headers = {
        \"Authorization\": f\"Bearer {api_key}\",
        \"Content-Type\": \"application/json\"
    }
    payload = {
        \"control_id\": control_id,
        \"evidence_type\": \"GCP_LOG\",
        \"resource_id\": \"gcp:cloudsql:my-instance\",
        \"payload\": evidence,
        \"metadata\": {\"collector_version\": \"1.0.0\"}
    }
    response = requests.post(
        f\"{COMPLIANCY_API_BASE}/evidence\",
        headers=headers,
        json=payload
    )
    if response.status_code != 201:
        raise Exception(f\"Compliancy API error: {response.text}\")
    return response.json()[\"evidence_id\"]