Why Wiz 2.0 Is Better Than Datadog Security for 2026 Multi-Cloud CSPM

import os
import time
import json
import logging
from typing import Dict, List, Optional
import requests
from dataclasses import dataclass

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

@dataclass
class ScanResult:
    tool_name: str
    scan_duration_sec: float
    resources_scanned: int
    findings_count: int
    error: Optional[str] = None

class WizCSPMClient:
    """Client for Wiz 2.0 GraphQL API (v2.1.4)"""
    def __init__(self, api_key: str, endpoint: str = "https://api.wiz.io/graphql"):
        self.api_key = api_key
        self.endpoint = endpoint
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })

    def trigger_full_scan(self, cloud_provider: str) -> ScanResult:
        """Trigger a full CSPM scan for a specific cloud provider"""
        start_time = time.time()
        try:
            # Wiz 2.0 uses async scan mutation
            mutation = """
            mutation TriggerScan($provider: CloudProvider!) {
              triggerCSPMScan(cloudProvider: $provider) {
                scanId
                status
                resourcesScanned
                findings {
                  totalCount
                }
              }
            }
            """
            variables = {"provider": cloud_provider.upper()}
            resp = self.session.post(
                self.endpoint,
                json={"query": mutation, "variables": variables},
                timeout=30
            )
            resp.raise_for_status()
            data = resp.json()
            if "errors" in data:
                raise ValueError(f"Wiz API error: {data['errors']}")

            scan_data = data["data"]["triggerCSPMScan"]
            # Poll for scan completion (Wiz 2.0 scans complete in <10min for 12k resources)
            scan_id = scan_data["scanId"]
            while True:
                query = """
                query GetScanStatus($scanId: ID!) {
                  cspmScan(id: $scanId) {
                    status
                    resourcesScanned
                    findings {
                      totalCount
                    }
                  }
                }
                """
                status_resp = self.session.post(
                    self.endpoint,
                    json={"query": query, "variables": {"scanId": scan_id}},
                    timeout=30
                )
                status_resp.raise_for_status()
                status_data = status_resp.json()["data"]["cspmScan"]
                if status_data["status"] == "COMPLETED":
                    end_time = time.time()
                    return ScanResult(
                        tool_name="Wiz 2.0",
                        scan_duration_sec=end_time - start_time,
                        resources_scanned=status_data["resourcesScanned"],
                        findings_count=status_data["findings"]["totalCount"]
                    )
                elif status_data["status"] == "FAILED":
                    raise ValueError(f"Wiz scan {scan_id} failed")
                time.sleep(30)  # Poll every 30s
        except Exception as e:
            logger.error(f"Wiz scan failed: {str(e)}")
            return ScanResult(tool_name="Wiz 2.0", scan_duration_sec=0, resources_scanned=0, findings_count=0, error=str(e))

class DatadogSecurityClient:
    """Client for Datadog Security API (v1.9.2)"""
    def __init__(self, api_key: str, app_key: str, site: str = "datadoghq.com"):
        self.api_key = api_key
        self.app_key = app_key
        self.endpoint = f"https://api.{site}/api/v2"
        self.session = requests.Session()
        self.session.headers.update({
            "DD-API-KEY": api_key,
            "DD-APPLICATION-KEY": app_key,
            "Content-Type": "application/json"
        })

    def trigger_full_scan(self, cloud_provider: str) -> ScanResult:
        """Trigger a full CSPM scan for a specific cloud provider"""
        start_time = time.time()
        try:
            # Datadog Security uses POST to trigger scans
            url = f"{self.endpoint}/security/cloud_scans"
            payload = {
                "data": {
                    "type": "cloud_scan",
                    "attributes": {
                        "cloud_provider": cloud_provider.lower(),
                        "scan_scope": "FULL_ENVIRONMENT"
                    }
                }
            }
            resp = self.session.post(url, json=payload, timeout=30)
            resp.raise_for_status()
            scan_id = resp.json()["data"]["id"]

            # Poll for scan completion
            while True:
                status_url = f"{self.endpoint}/security/cloud_scans/{scan_id}"
                status_resp = self.session.get(status_url, timeout=30)
                status_resp.raise_for_status()
                status_data = status_resp.json()["data"]["attributes"]
                if status_data["status"] == "completed":
                    end_time = time.time()
                    return ScanResult(
                        tool_name="Datadog Security",
                        scan_duration_sec=end_time - start_time,
                        resources_scanned=status_data["resources_scanned"],
                        findings_count=status_data["findings_count"]
                    )
                elif status_data["status"] == "failed":
                    raise ValueError(f"Datadog scan {scan_id} failed")
                time.sleep(60)  # Datadog scans take longer, poll every 60s
        except Exception as e:
            logger.error(f"Datadog scan failed: {str(e)}")
            return ScanResult(tool_name="Datadog Security", scan_duration_sec=0, resources_scanned=0, findings_count=0, error=str(e))

if __name__ == "__main__":
    # Load credentials from env vars (never hardcode!)
    wiz_api_key = os.getenv("WIZ_API_KEY")
    dd_api_key = os.getenv("DATADOG_API_KEY")
    dd_app_key = os.getenv("DATADOG_APP_KEY")

    if not all([wiz_api_key, dd_api_key, dd_app_key]):
        raise ValueError("Missing required API credentials in environment variables")

    # Initialize clients
    wiz_client = WizCSPMClient(wiz_api_key)
    dd_client = DatadogSecurityClient(dd_api_key, dd_app_key)

    # Run benchmarks across AWS, Azure, GCP
    providers = ["aws", "azure", "gcp"]
    results = []
    for provider in providers:
        logger.info(f"Starting {provider.upper()} scan benchmark")
        wiz_result = wiz_client.trigger_full_scan(provider)
        dd_result = dd_client.trigger_full_scan(provider)
        results.append(wiz_result)
        results.append(dd_result)
        logger.info(f"Provider {provider}: Wiz took {wiz_result.scan_duration_sec:.2f}s, Datadog took {dd_result.scan_duration_sec:.2f}s")

    # Export results to JSON
    with open("benchmark_results.json", "w") as f:
        json.dump([vars(r) for r in results], f, indent=2)
    logger.info("Benchmark complete. Results saved to benchmark_results.json")

# Terraform configuration for deploying Wiz 2.0 agentless CSPM connectors across AWS, Azure, GCP
# Version: Wiz 2.0 Terraform Provider v2.1.0
# Requires Terraform >= 1.5.0

terraform {
  required_version = ">= 1.5.0"
  required_providers {
    wiz = {
      source  = "wiz-security/wiz"
      version = "~> 2.1.0"
    }
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
    azurerm = {
      source  = "hashicorp/azurerm"
      version = "~> 3.0"
    }
    google = {
      source  = "hashicorp/google"
      version = "~> 5.0"
    }
  }
  # Store state in S3 for multi-dev collaboration
  backend "s3" {
    bucket         = "wiz-cspm-terraform-state"
    key            = "multi-cloud/wiz-connectors.tfstate"
    region         = "us-east-1"
    encrypt        = true
    dynamodb_table = "terraform-lock"
  }
}

# Variables for cloud credentials (loaded from env or tfvars)
variable "wiz_api_key" {
  type        = string
  sensitive   = true
  description = "Wiz 2.0 API key with connector admin permissions"
}

variable "aws_regions" {
  type        = list(string)
  default     = ["us-east-1", "eu-west-1", "ap-southeast-1"]
  description = "AWS regions to deploy Wiz connectors to"
}

variable "azure_subscription_id" {
  type        = string
  sensitive   = true
  description = "Azure subscription ID to deploy connectors to"
}

variable "gcp_project_id" {
  type        = string
  sensitive   = true
  description = "GCP project ID to deploy connectors to"
}

# Configure Wiz provider
provider "wiz" {
  api_key = var.wiz_api_key
}

# AWS Wiz Connector Deployment
resource "aws_iam_role" "wiz_connector_role" {
  name = "WizCSPMConnectorRole"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRoleWithWebIdentity"
        Effect = "Allow"
        Principal = {
          Federated = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:oidc-provider/api.wiz.io"
        }
        Condition = {
          StringEquals = {
            "api.wiz.io:aud" = "wiz-connector"
          }
        }
      }
    ]
  })

  tags = {
    Tool = "Wiz-2.0"
    Use  = "CSPM-Connector"
  }
}

resource "aws_iam_role_policy_attachment" "wiz_security_audit" {
  role       = aws_iam_role.wiz_connector_role.name
  policy_arn = "arn:aws:iam::aws:policy/SecurityAudit"
}

resource "wiz_aws_connector" "aws_connectors" {
  for_each = toset(var.aws_regions)

  name            = "aws-${each.key}-connector"
  aws_region      = each.key
  role_arn        = aws_iam_role.wiz_connector_role.arn
  scan_interval   = 60  # Scan every 60 minutes
  agentless       = true
  resource_types = ["EC2", "S3", "RDS", "Lambda", "IAM"]  # All supported types

  lifecycle {
    prevent_destroy = false  # Allow destruction in dev environments
    ignore_changes = [
      last_scan_time  # Ignore changes to scan timestamps
    ]
  }

  depends_on = [aws_iam_role_policy_attachment.wiz_security_audit]
}

# Azure Wiz Connector Deployment
provider "azurerm" {
  features {}
  subscription_id = var.azure_subscription_id
}

resource "azurerm_role_definition" "wiz_security_reader" {
  name        = "WizCSPMReader"
  scope       = "/subscriptions/${var.azure_subscription_id}"
  description = "Read-only access for Wiz 2.0 CSPM scanning"

  permissions {
    actions = [
      "*/read",
      "Microsoft.Security/*/read"
    ]
    not_actions = []
  }

  assignable_scopes = [
    "/subscriptions/${var.azure_subscription_id}"
  ]
}

resource "wiz_azure_connector" "azure_connector" {
  name                = "azure-connector"
  subscription_id     = var.azure_subscription_id
  role_definition_id  = azurerm_role_definition.wiz_security_reader.id
  scan_interval       = 60
  agentless           = true
  resource_groups     = ["*"]  # Scan all resource groups

  lifecycle {
    prevent_destroy = false
  }
}

# GCP Wiz Connector Deployment
provider "google" {
  project = var.gcp_project_id
}

resource "google_service_account" "wiz_sa" {
  account_id   = "wiz-cspm-sa"
  display_name = "Wiz 2.0 CSPM Service Account"
  project      = var.gcp_project_id
}

resource "google_project_iam_member" "wiz_sa_binding" {
  project = var.gcp_project_id
  role    = "roles/viewer"
  member  = "serviceAccount:${google_service_account.wiz_sa.email}"
}

resource "wiz_gcp_connector" "gcp_connector" {
  name                = "gcp-connector"
  project_id          = var.gcp_project_id
  service_account_email = google_service_account.wiz_sa.email
  scan_interval       = 60
  agentless           = true
  resource_types      = ["compute", "storage", "iam", "cloudsql"]

  depends_on = [google_project_iam_member.wiz_sa_binding]
}

# Outputs for verification
output "wiz_connector_ids" {
  value = {
    aws = [for k, v in wiz_aws_connector.aws_connectors : v.id]
    azure = wiz_azure_connector.azure_connector.id
    gcp = wiz_gcp_connector.gcp_connector.id
  }
  description = "IDs of deployed Wiz 2.0 connectors"
}

output "last_scan_times" {
  value = {
    aws = [for k, v in wiz_aws_connector.aws_connectors : v.last_scan_time]
    azure = wiz_azure_connector.azure_connector.last_scan_time
    gcp = wiz_gcp_connector.gcp_connector.last_scan_time
  }
  description = "Last scan times for each connector"
}

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/wiz-io/wiz-go-client/v2" // Wiz 2.0 Go SDK v2.1.4
    "github.com/DataDog/datadog-api-client-go/v2/api/datadog" // Datadog API SDK v1.9.2
    "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
)

// Finding represents a normalized CSPM finding across tools
type Finding struct {
    ID             string    `json:"id"`
    Tool           string    `json:"tool"`
    ResourceID     string    `json:"resource_id"`
    Severity       string    `json:"severity"`
    PolicyID       string    `json:"policy_id"`
    Description    string    `json:"description"`
    FirstDetected  time.Time `json:"first_detected"`
    IsFalsePositive bool     `json:"is_false_positive"`
}

// WizClient wraps the Wiz Go SDK
type WizClient struct {
    client *wiz.Client
}

// NewWizClient initializes a Wiz 2.0 client
func NewWizClient(apiKey string) (*WizClient, error) {
    c, err := wiz.NewClient(
        wiz.WithAPIKey(apiKey),
        wiz.WithUserAgent("wiz-cspm-processor/1.0"),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to init Wiz client: %w", err)
    }
    return &WizClient{client: c}, nil
}

// FetchFindings retrieves all open findings from Wiz 2.0
func (wc *WizClient) FetchFindings(ctx context.Context) ([]Finding, error) {
    var findings []Finding
    // Wiz uses cursor-based pagination
    nextCursor := ""
    for {
        resp, err := wc.client.CSPM.Findings.List(ctx, wiz.CSPMFindingsListRequest{
            Status: wiz.PtrString("OPEN"),
            Pagination: &wiz.PaginationRequest{
                First: wiz.PtrInt(100),
                After: wiz.PtrString(nextCursor),
            },
        })
        if err != nil {
            return nil, fmt.Errorf("wiz fetch failed: %w", err)
        }
        for _, f := range resp.Data {
            findings = append(findings, Finding{
                ID:             f.ID,
                Tool:           "Wiz 2.0",
                ResourceID:     f.Resource.ID,
                Severity:       string(f.Severity),
                PolicyID:       f.Policy.ID,
                Description:    f.Description,
                FirstDetected:  f.FirstDetectedAt,
                IsFalsePositive: false, // Default to false, updated later
            })
        }
        if resp.Pagination.EndCursor == nil || *resp.Pagination.EndCursor == "" {
            break
        }
        nextCursor = *resp.Pagination.EndCursor
    }
    return findings, nil
}

// DatadogClient wraps the Datadog Go SDK
type DatadogClient struct {
    client *datadog.APIClient
}

// NewDatadogClient initializes a Datadog Security client
func NewDatadogClient(apiKey, appKey string) (*DatadogClient, error) {
    ctx := context.WithValue(context.Background(), datadog.ContextAPIKeys, map[string]datadog.APIKey{
        "apiKeyAuth": {Key: apiKey},
        "appKeyAuth": {Key: appKey},
    })
    cfg := datadog.NewConfiguration()
    c := datadog.NewAPIClient(cfg)
    return &DatadogClient{client: c}, nil
}

// FetchFindings retrieves all open findings from Datadog Security
func (dc *DatadogClient) FetchFindings(ctx context.Context) ([]Finding, error) {
    var findings []Finding
    // Datadog uses offset-based pagination
    offset := 0
    limit := 100
    for {
        resp, _, err := dc.client.SecurityMonitoringApi.ListSecurityMonitoringFindings(ctx).
            FilterStatus("open").
            PageOffset(int64(offset)).
            PageLimit(int64(limit)).
            Execute()
        if err != nil {
            return nil, fmt.Errorf("datadog fetch failed: %w", err)
        }
        for _, f := range resp.Data {
            findings = append(findings, Finding{
                ID:             f.GetId(),
                Tool:           "Datadog Security",
                ResourceID:     f.GetResourceId(),
                Severity:       f.GetSeverity(),
                PolicyID:       f.GetRuleId(),
                Description:    f.GetMessage(),
                FirstDetected:  f.GetFirstSeen(),
                IsFalsePositive: false,
            })
        }
        if len(resp.Data) < limit {
            break
        }
        offset += limit
    }
    return findings, nil
}

// DeduplicateFalsePositives removes findings matching known false positive patterns
func DeduplicateFalsePositives(findings []Finding, fpPatterns []string) []Finding {
    // In production, this would use a Bloom filter or hash map for O(1) lookups
    fpSet := make(map[string]bool)
    for _, p := range fpPatterns {
        fpSet[p] = true
    }
    var deduped []Finding
    for _, f := range findings {
        // Check if policy is a known false positive
        if fpSet[f.PolicyID] {
            f.IsFalsePositive = true
            log.Printf("Marked finding %s as false positive (policy %s)", f.ID, f.PolicyID)
        }
        deduped = append(deduped, f)
    }
    return deduped
}

// ExportToSIEM exports findings to a generic SIEM (JSON Lines format)
func ExportToSIEM(findings []Finding, outputPath string) error {
    file, err := os.Create(outputPath)
    if err != nil {
        return fmt.Errorf("failed to create output file: %w", err)
    }
    defer file.Close()
    for _, f := range findings {
        b, err := json.Marshal(f)
        if err != nil {
            log.Printf("Failed to marshal finding %s: %v", f.ID, err)
            continue
        }
        b = append(b, '
')
        if _, err := file.Write(b); err != nil {
            return fmt.Errorf("failed to write finding: %w", err)
        }
    }
    return nil
}

func main() {
    // Load credentials from env
    wizAPIKey := os.Getenv("WIZ_API_KEY")
    ddAPIKey := os.Getenv("DATADOG_API_KEY")
    ddAppKey := os.Getenv("DATADOG_APP_KEY")
    if wizAPIKey == "" || ddAPIKey == "" || ddAppKey == "" {
        log.Fatal("Missing required environment variables: WIZ_API_KEY, DATADOG_API_KEY, DATADOG_APP_KEY")
    }

    ctx := context.Background()

    // Fetch findings from both tools
    wizClient, err := NewWizClient(wizAPIKey)
    if err != nil {
        log.Fatalf("Wiz client init failed: %v", err)
    }
    wizFindings, err := wizClient.FetchFindings(ctx)
    if err != nil {
        log.Fatalf("Wiz fetch failed: %v", err)
    }
    log.Printf("Fetched %d findings from Wiz 2.0", len(wizFindings))

    ddClient, err := NewDatadogClient(ddAPIKey, ddAppKey)
    if err != nil {
        log.Fatalf("Datadog client init failed: %v", err)
    }
    ddFindings, err := ddClient.FetchFindings(ctx)
    if err != nil {
        log.Fatalf("Datadog fetch failed: %v", err)
    }
    log.Printf("Fetched %d findings from Datadog Security", len(ddFindings))

    // Combine and deduplicate
    allFindings := append(wizFindings, ddFindings)
    // Known false positive policy IDs (example)
    fpPatterns := []string{
        "wiz-policy-1234",  // Known false positive in Wiz
        "datadog-rule-5678", // Known false positive in Datadog
    }
    deduped := DeduplicateFalsePositives(allFindings, fpPatterns)
    log.Printf("Total findings after deduplication: %d (false positives: %d)", len(deduped), countFalsePositives(deduped))

    // Export to SIEM
    if err := ExportToSIEM(deduped, "cspm_findings.jsonl"); err != nil {
        log.Fatalf("Export failed: %v", err)
    }
    log.Println("Findings exported to cspm_findings.jsonl")
}

func countFalsePositives(findings []Finding) int {
    count := 0
    for _, f := range findings {
        if f.IsFalsePositive {
            count++
        }
    }
    return count
}

# Check Wiz 2.0 agentless coverage for AWS
curl -X POST https://api.wiz.io/graphql \
  -H "Authorization: Bearer $WIZ_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "query GetConnectorCoverage { awsConnectors { id agentless resourceTypesCovered totalResourcesCovered } }"
  }'

import requests
import os

wiz_key = os.getenv("WIZ_API_KEY")
query = """
query GetPCIDSSGaps {
  complianceFramework(slug: "pci-dss-4.0") {
    id
    name
    controls {
      id
      name
      status
      failingResources {
        totalCount
      }
    }
  }
}
"""
resp = requests.post("https://api.wiz.io/graphql", json={"query": query}, headers={"Authorization": f"Bearer {wiz_key}"})
print(resp.json()["data"]["complianceFramework"]["controls"])

package wiz.custom.s3_public_pii

import future.keywords.if
import future.keywords.in

# Flag S3 buckets as high risk only if public AND tagged as PII
deny[msg] {
    resource := input.resource
    resource.type == "AWS::S3::Bucket"
    resource.public_access.enabled == true
    "PII" in resource.tags.classification
    msg := sprintf("Public S3 bucket %s stores PII data", [resource.name])
}