How We Survived Firewall vs Antivirus: Which Wins?

#!/usr/bin/env python3
"""
Firewall Throughput & Latency Benchmark Tool
Version: 1.2.0
Test Environment:
- Hardware: Intel Xeon Gold 6338 (32 cores), 128GB DDR4, 2x 10Gbps Intel X710 NICs
- OS: Ubuntu 22.04 LTS, kernel 5.15.0-91-generic
- Firewall Under Test: pfSense 2.7.0 (stateful), Palo Alto PA-5400 (NGFW) 11.1.2
- Dependencies: iperf3 3.12, scapy 2.5.0, pandas 2.1.4
"""

import subprocess
import time
import json
import pandas as pd
from typing import Dict, List, Optional
import logging

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

class FirewallBenchmarker:
    def __init__(self, target_ip: str, iperf_port: int = 5201, test_duration: int = 60):
        self.target_ip = target_ip
        self.iperf_port = iperf_port
        self.test_duration = test_duration
        self.results: List[Dict] = []

    def run_iperf_test(self, protocol: str = "tcp", bandwidth: str = "10G") -> Optional[Dict]:
        """Run iperf3 test and return parsed results with error handling"""
        cmd = [
            "iperf3", "-c", self.target_ip,
            "-p", str(self.iperf_port),
            "-t", str(self.test_duration),
            "-J",  # JSON output
            "-b", bandwidth,
            "-P", "4"  # 4 parallel streams
        ]
        if protocol == "udp":
            cmd.append("-u")

        try:
            logger.info(f"Starting {protocol.upper()} iperf test to {self.target_ip} for {self.test_duration}s")
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.test_duration + 10  # Buffer for startup
            )
            if result.returncode != 0:
                logger.error(f"iperf3 failed with return code {result.returncode}: {result.stderr}")
                return None

            iperf_data = json.loads(result.stdout)
            # Extract relevant metrics
            return {
                "protocol": protocol,
                "bandwidth_gbps": iperf_data["end"]["sum_received"]["bits_per_second"] / 1e9,
                "latency_ms": iperf_data["end"]["sum_received"].get("latency_ms", 0),
                "retransmits": iperf_data["end"]["sum_received"].get("retransmits", 0),
                "timestamp": time.time()
            }
        except subprocess.TimeoutExpired:
            logger.error("iperf3 test timed out")
            return None
        except json.JSONDecodeError:
            logger.error("Failed to parse iperf3 JSON output")
            return None
        except Exception as e:
            logger.error(f"Unexpected error in iperf test: {str(e)}")
            return None

    def run_latency_test(self, num_packets: int = 1000, packet_size: int = 64) -> Optional[float]:
        """Measure ICMP latency across firewall using scapy"""
        try:
            from scapy.all import IP, ICMP, sr1
            latencies = []
            for _ in range(num_packets):
                start = time.time()
                pkt = IP(dst=self.target_ip) / ICMP()
                reply = sr1(pkt, timeout=2, verbose=0)
                if reply:
                    latencies.append((time.time() - start) * 1000)  # ms
            return sum(latencies) / len(latencies) if latencies else None
        except ImportError:
            logger.warning("scapy not installed, skipping latency test")
            return None
        except Exception as e:
            logger.error(f"Latency test failed: {str(e)}")
            return None

    def save_results(self, output_path: str = "firewall_results.csv"):
        """Save benchmark results to CSV"""
        if not self.results:
            logger.warning("No results to save")
            return
        df = pd.DataFrame(self.results)
        df.to_csv(output_path, index=False)
        logger.info(f"Saved {len(self.results)} results to {output_path}")

if __name__ == "__main__":
    # Configuration for test run
    FIREWALL_IP = "10.0.1.1"  # Firewall internal interface
    TEST_DURATION = 60  # seconds per test
    ITERATIONS = 3  # Number of test runs per protocol

    benchmarker = FirewallBenchmarker(FIREWALL_IP, test_duration=TEST_DURATION)
    for protocol in ["tcp", "udp"]:
        for _ in range(ITERATIONS):
            # Run throughput test
            throughput_result = benchmarker.run_iperf_test(protocol=protocol)
            if throughput_result:
                # Run latency test
                latency = benchmarker.run_latency_test()
                throughput_result["median_latency_ms"] = latency
                benchmarker.results.append(throughput_result)
                logger.info(f"Collected result: {throughput_result}")
            time.sleep(5)  # Cooldown between tests

    benchmarker.save_results()
    logger.info("Benchmark run completed")

#!/usr/bin/env python3
"""
Antivirus Performance & Detection Benchmark Tool
Version: 1.1.0
Test Environment:
- Hardware: Intel Xeon Gold 6338 (32 cores), 128GB DDR4, 2x 10Gbps Intel X710 NICs
- OS: Windows Server 2022 Datacenter, Build 20348.1668
- Antivirus Under Test: Symantec Endpoint 14.3 (Legacy), CrowdStrike Falcon 6.42 (NGAV)
- Dependencies: EICAR test files, psutil 5.9.8, pandas 2.1.4
"""

import os
import time
import subprocess
import psutil
import pandas as pd
from typing import Dict, List, Optional
import logging
import hashlib

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

# EICAR test string (standard malware test signature)
EICAR_STRING = r"X5O!P%@AP[4\PZX54(P^)7CC)o$!1E" + chr(92) + "3CM7]K:>!H*#D" + chr(92) + ";8P'!*F'0+!3{}@$!1E!*F'0+!3{}@$!1E"

class AVBenchmarker:
    def __init__(self, test_dir: str = "./av_test_files", scan_interval: int = 10):
        self.test_dir = test_dir
        self.scan_interval = scan_interval
        self.results: List[Dict] = []
        os.makedirs(test_dir, exist_ok=True)

    def generate_test_files(self, num_files: int = 100, file_size_mb: int = 10) -> List[str]:
        """Generate test files: mix of clean and EICAR-infected files"""
        file_paths = []
        try:
            for i in range(num_files):
                file_path = os.path.join(self.test_dir, f"test_file_{i}.exe")
                with open(file_path, "wb") as f:
                    if i % 5 == 0:  # 20% infected files
                        f.write(EICAR_STRING.encode())
                    else:
                        # Write random clean data
                        f.write(os.urandom(file_size_mb * 1024 * 1024))
                file_paths.append(file_path)
            logger.info(f"Generated {num_files} test files in {self.test_dir}")
            return file_paths
        except Exception as e:
            logger.error(f"Failed to generate test files: {str(e)}")
            return []

    def get_av_cpu_overhead(self, av_process_names: List[str] = ["Symantec.exe", "CrowdStrike.exe"]) -> Optional[float]:
        """Calculate AV CPU overhead by summing process CPU usage"""
        try:
            total_cpu = 0.0
            for proc in psutil.process_iter(["name", "cpu_percent"]):
                if proc.info["name"] in av_process_names:
                    total_cpu += proc.info["cpu_percent"]
            return total_cpu
        except Exception as e:
            logger.error(f"Failed to get CPU overhead: {str(e)}")
            return None

    def run_detection_test(self, av_scan_cmd: str = "C:\\Program Files\\Symantec\\Scan.exe") -> Optional[Dict]:
        """Run antivirus scan and count detected threats"""
        try:
            # Start CPU measurement
            psutil.cpu_percent(interval=None)  # Reset CPU counter
            start_time = time.time()
            start_cpu = psutil.cpu_percent(interval=1)

            # Run scan
            result = subprocess.run(
                [av_scan_cmd, self.test_dir, "/s"],  # /s for silent scan
                capture_output=True,
                text=True,
                timeout=300  # 5 minute timeout
            )

            end_time = time.time()
            end_cpu = psutil.cpu_percent(interval=1)
            duration = end_time - start_time

            # Parse scan output for detected threats
            detected = 0
            for line in result.stdout.splitlines():
                if "threat" in line.lower() or "malware" in line.lower():
                    detected += 1

            return {
                "scan_duration_s": duration,
                "detected_threats": detected,
                "avg_cpu_percent": (start_cpu + end_cpu) / 2,
                "timestamp": time.time()
            }
        except subprocess.TimeoutExpired:
            logger.error("Antivirus scan timed out")
            return None
        except FileNotFoundError:
            logger.error(f"Antivirus scan command not found: {av_scan_cmd}")
            return None
        except Exception as e:
            logger.error(f"Detection test failed: {str(e)}")
            return None

    def save_results(self, output_path: str = "av_results.csv"):
        """Save benchmark results to CSV"""
        if not self.results:
            logger.warning("No results to save")
            return
        df = pd.DataFrame(self.results)
        df.to_csv(output_path, index=False)
        logger.info(f"Saved {len(self.results)} results to {output_path}")

if __name__ == "__main__":
    # Configuration
    AV_SCAN_CMD = "C:\\Program Files\\CrowdStrike\\Falcon\\falcon.exe"  # Update for target AV
    AV_PROCESS_NAMES = ["FalconWorker.exe"]  # Update for target AV
    NUM_TEST_FILES = 100
    ITERATIONS = 3

    benchmarker = AVBenchmarker()
    test_files = benchmarker.generate_test_files(num_files=NUM_TEST_FILES)

    for i in range(ITERATIONS):
        logger.info(f"Starting iteration {i+1}/{ITERATIONS}")
        # Measure idle CPU overhead
        idle_cpu = benchmarker.get_av_cpu_overhead(av_process_names=AV_PROCESS_NAMES)
        # Run detection test
        detection_result = benchmarker.run_detection_test(av_scan_cmd=AV_SCAN_CMD)
        if detection_result:
            detection_result["idle_cpu_percent"] = idle_cpu
            benchmarker.results.append(detection_result)
        time.sleep(10)  # Cooldown between tests

    benchmarker.save_results()
    logger.info("Antivirus benchmark completed")

# Combined Firewall + NGAV Stack Deployment on AWS
# Version: 1.0.0
# Test Environment:
# - AWS Region: us-east-1
# - Instance Types: m6i.4xlarge (firewall), m6i.2xlarge (workloads)
# - Tools: AWS Network Firewall 2024.03, CrowdStrike Falcon 6.42
# - Terraform 1.7.0, AWS Provider 5.31.0

terraform {
  required_version = ">= 1.7.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.31.0"
    }
    crowdstrike = {
      source  = "CrowdStrike/crowdstrike"
      version = "~> 0.2.0"
    }
  }
  # Store state in S3 for team collaboration
  backend "s3" {
    bucket = "firewall-av-bench-terraform-state"
    key    = "global/s3/terraform.tfstate"
    region = "us-east-1"
  }
}

provider "aws" {
  region = var.aws_region
}

provider "crowdstrike" {
  # Configure via CROWDSTRIKE_CLIENT_ID and CROWDSTRIKE_CLIENT_SECRET env vars
}

# Variables
variable "aws_region" {
  type        = string
  default     = "us-east-1"
  description = "AWS region to deploy resources"
}

variable "vpc_cidr" {
  type        = string
  default     = "10.0.0.0/16"
  description = "CIDR block for the VPC"
}

variable "crowdstrike_customer_id" {
  type        = string
  description = "CrowdStrike customer ID for agent deployment"
}

# VPC Configuration
resource "aws_vpc" "main" {
  cidr_block           = var.vpc_cidr
  enable_dns_support   = true
  enable_dns_hostnames = true

  tags = {
    Name        = "firewall-av-bench-vpc"
    Environment = "benchmark"
  }
}

# Public Subnet for Firewall Management
resource "aws_subnet" "public" {
  vpc_id                  = aws_vpc.main.id
  cidr_block              = "10.0.1.0/24"
  map_public_ip_on_launch = true
  availability_zone       = "${var.aws_region}a"

  tags = {
    Name = "public-subnet"
  }
}

# Private Subnet for Workloads
resource "aws_subnet" "private" {
  vpc_id            = aws_vpc.main.id
  cidr_block        = "10.0.2.0/24"
  availability_zone = "${var.aws_region}a"

  tags = {
    Name = "private-subnet"
  }
}

# AWS Network Firewall (Stateful + NGFW Capabilities)
resource "aws_networkfirewall_firewall" "main" {
  name                = "benchmark-network-firewall"
  firewall_policy_arn = aws_networkfirewall_firewall_policy.main.arn
  vpc_id              = aws_vpc.main.id
  subnet_mapping {
    subnet_id = aws_subnet.public.id
  }

  tags = {
    Name = "benchmark-network-firewall"
  }
}

# Firewall Policy with Stateful and Stateless Rules
resource "aws_networkfirewall_firewall_policy" "main" {
  name = "benchmark-firewall-policy"

  firewall_policy {
    stateless_default_actions          = ["aws:forward_to_sfe"]
    stateless_fragment_default_actions = ["aws:forward_to_sfe"]

    # Stateless rule to allow all traffic (for benchmarking baseline)
    stateless_rule_group_reference {
      resource_arn = aws_networkfirewall_rule_group.stateless.arn
      priority     = 1
    }

    # Stateful rule group for IPS signatures
    stateful_rule_group_reference {
      resource_arn = aws_networkfirewall_rule_group.stateful.arn
    }
  }
}

# Stateless Rule Group (Allow All)
resource "aws_networkfirewall_rule_group" "stateless" {
  capacity = 100
  name     = "allow-all-stateless"
  type     = "STATELESS"

  rule_group {
    rules_source {
      stateless_rules_and_custom_actions {
        stateless_rule {
          priority = 1
          rule_definition {
            actions = ["aws:pass"]
            match_attributes {
              source_addresses {
                address_definition = "0.0.0.0/0"
              }
              destination_addresses {
                address_definition = "0.0.0.0/0"
              }
              protocols = [6, 17]  # TCP, UDP
            }
          }
        }
      }
    }
  }
}

# Stateful Rule Group (Suricata IPS Signatures)
resource "aws_networkfirewall_rule_group" "stateful" {
  capacity = 1000
  name     = "suricata-stateful"
  type     = "STATEFUL"

  rule_group {
    rules_source {
      rules_string = file("${path.module}/suricata_rules.rules")
    }
  }
}

# CrowdStrike Falcon NGAV Deployment on Workload Instances
resource "aws_instance" "workload" {
  count                  = 3
  ami                    = "ami-0c7217cdde317cfec"  # Ubuntu 22.04 LTS us-east-1
  instance_type          = "m6i.2xlarge"
  subnet_id              = aws_subnet.private.id
  vpc_security_group_ids = [aws_security_group.workload.id]

  # Install CrowdStrike Falcon agent on boot
  user_data = <<-EOF
    #!/bin/bash
    curl -X POST -H "Content-Type: application/json" \
      -d '{"customer_id": "${var.crowdstrike_customer_id}"}' \
      https://api.crowdstrike.com/sensors/v3/installers/linux
    chmod +x falcon-linux-installer.sh
    ./falcon-linux-installer.sh -t "benchmark-workload"
    systemctl enable falcon-sensor
    systemctl start falcon-sensor
  EOF

  tags = {
    Name = "benchmark-workload-${count.index}"
  }
}

# Security Group for Workloads (Allow inbound from firewall)
resource "aws_security_group" "workload" {
  name   = "workload-sg"
  vpc_id = aws_vpc.main.id

  ingress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = [aws_networkfirewall_firewall.main.firewall_status[0].sync_states[0].attachment[0].endpoint_id]
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

# Output Firewall Endpoint for Benchmark Tools
output "network_firewall_endpoint" {
  value = aws_networkfirewall_firewall.main.firewall_status[0].sync_states[0].attachment[0].endpoint_id
}

output "workload_private_ips" {
  value = aws_instance.workload[*].private_ip
}

# Short snippet to measure AV CPU overhead
import psutil
def get_av_cpu(process_names=["FalconWorker.exe"]):
    return sum(p.cpu_percent() for p in psutil.process_iter() if p.name() in process_names)
print(f"AV CPU Overhead: {get_av_cpu()}%")

# Terraform snippet to link CrowdStrike alerts to AWS Network Firewall
resource "aws_networkfirewall_rule_group" "crowdstrike_blocks" {
  name     = "crowdstrike-blocklist"
  type     = "STATEFUL"
  capacity = 1000
  rule_group {
    rules_source {
      dynamic "address_set" {
        for_each = crowdstrike_ioc.ip_addresses  # Pulled from CrowdStrike API
        content {
          address_definition = address_set.value
        }
      }
    }
  }
}

# Suricata rule to detect ransomware behavior (encrypts files + network comms)
alert tcp any any -> any any (msg:"Ransomware Behavior: Bulk File Encryption + C2"; \
flow:established; content:"encryption_key"; depth:100; \
content:".encrypted"; content:"/pay/"; distance:0; \
threshold:type limit, track by_src, count 5, seconds 10; sid:1000001; rev:1;)