Firewall How We Survived: A Data-Backed Analysis

#!/usr/bin/env python3
"""
Firewall Rule Validator v1.2
Validates nftables rule sets for syntax errors, shadowed rules, and redundant entries.
Requires: pyroute2==0.7.3, nftables>=1.0.7
"""
import sys
import json
from pyroute2 import nftables
from pyroute2.netlink.exceptions import NetlinkError
from typing import List, Dict, Tuple

class FirewallRuleValidator:
    def __init__(self, rule_set_path: str):
        self.rule_set_path = rule_set_path
        self.nft = nftables.Nftables()
        self.errors: List[str] = []
        self.warnings: List[str] = []
        self.valid_rules: List[Dict] = []

    def load_rule_set(self) -> bool:
        """Load rule set from JSON file exported via nft -j list ruleset"""
        try:
            with open(self.rule_set_path, 'r') as f:
                self.raw_ruleset = json.load(f)
            return True
        except FileNotFoundError:
            self.errors.append(f"Rule set file {self.rule_set_path} not found")
            return False
        except json.JSONDecodeError as e:
            self.errors.append(f"Invalid JSON in rule set: {str(e)}")
            return False

    def validate_syntax(self) -> None:
        """Check each rule for nftables syntax validity via dry-run"""
        for table in self.raw_ruleset.get('nftables', []):
            if table.get('table') is None:
                continue
            table_name = table['table']['name']
            for chain in table.get('chain', []):
                chain_name = chain['chain']['name']
                for rule in chain.get('rule', []):
                    rule_str = json.dumps(rule)
                    # Dry-run rule addition to check syntax
                    try:
                        self.nft.cmd(f"add rule {table_name} {chain_name} {rule_str} -d")
                    except NetlinkError as e:
                        self.errors.append(
                            f"Syntax error in {table_name}/{chain_name}: {str(e)}"
                        )
                    else:
                        self.valid_rules.append(rule)

    def check_shadowed_rules(self) -> None:
        """Detect rules that are never hit because a higher-priority rule matches first"""
        # Group rules by table/chain, sort by priority (lower = higher priority)
        rule_groups = {}
        for rule in self.valid_rules:
            key = f"{rule['table']}/{rule['chain']}"
            if key not in rule_groups:
                rule_groups[key] = []
            rule_groups[key].append(rule)

        for group_key, rules in rule_groups.items():
            # Sort by priority ascending (higher priority first)
            sorted_rules = sorted(rules, key=lambda x: x.get('priority', 0))
            for i in range(len(sorted_rules)):
                for j in range(i+1, len(sorted_rules)):
                    if self._rules_overlap(sorted_rules[i], sorted_rules[j]):
                        self.warnings.append(
                            f"Shadowed rule detected: {group_key} rule {j} shadowed by rule {i}"
                        )

    def _rules_overlap(self, rule_a: Dict, rule_b: Dict) -> bool:
        """Check if two rules match the same traffic (simplified for demo)"""
        # Compare match expressions (src ip, dst port, protocol)
        matches_a = rule_a.get('match', [])
        matches_b = rule_b.get('match', [])
        # Simplified overlap check: same protocol and dst port
        proto_a = next((m['right'] for m in matches_a if m['left']['payload']['field'] == 'proto'), None)
        proto_b = next((m['right'] for m in matches_b if m['left']['payload']['field'] == 'proto'), None)
        if proto_a != proto_b:
            return False
        dport_a = next((m['right'] for m in matches_a if m['left']['payload']['field'] == 'dport'), None)
        dport_b = next((m['right'] for m in matches_b if m['left']['payload']['field'] == 'dport'), None)
        return dport_a == dport_b

    def generate_report(self) -> Dict:
        """Return validation results"""
        return {
            "errors": self.errors,
            "warnings": self.warnings,
            "valid_rule_count": len(self.valid_rules),
            "total_rule_count": len(self.valid_rules) + len(self.errors)
        }

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} ")
        sys.exit(1)

    validator = FirewallRuleValidator(sys.argv[1])
    if not validator.load_rule_set():
        print("Failed to load rule set")
        sys.exit(1)

    validator.validate_syntax()
    validator.check_shadowed_rules()
    report = validator.generate_report()

    print(json.dumps(report, indent=2))
    if report["errors"]:
        sys.exit(1)

// Firewall Log Metrics Exporter v0.9.1
// Parses nftables log output, exports request counts, blocked ratios to Prometheus.
// Requires: go 1.22+, github.com/prometheus/client_golang v1.19.0
package main

import (
    "bufio"
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "regexp"
    "strconv"
    "strings"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// Metric definitions
var (
    totalRequests = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "firewall_total_requests",
            Help: "Total firewall requests by action (accept/drop) and protocol",
        },
        []string{"action", "protocol"},
    )
    blockedRatio = prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "firewall_blocked_ratio",
            Help: "Ratio of dropped requests to total requests over 5m window",
        },
    )
    ruleHitCount = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "firewall_rule_hit_count",
            Help: "Hit count per firewall rule ID",
        },
        []string{"rule_id"},
    )
)

// nftables log line regex (matches log prefix "FW: " with fields)
var logRegex = regexp.MustCompile(
    `FW: action=(?Paccept|drop) proto=(?P\w+) src=(?P\S+) dst=(?P\S+) dport=(?P\d+) rule_id=(?P\d+)`,
)

type logParser struct {
    ctx    context.Context
    cancel context.CancelFunc
}

func newLogParser() *logParser {
    ctx, cancel := context.WithCancel(context.Background())
    return &logParser{ctx: ctx, cancel: cancel}
}

func (p *logParser) parseLogFile(path string) error {
    file, err := os.Open(path)
    if err != nil {
        return fmt.Errorf("failed to open log file: %w", err)
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    // Increase scanner buffer for long log lines
    scanner.Buffer(make([]byte, 1024*1024), 1024*1024)

    // Sliding window for blocked ratio calculation
    window := make([]string, 0, 300) // 5m window at 1 log/s
    windowStart := time.Now()

    for scanner.Scan() {
        select {
        case <-p.ctx.Done():
            return nil
        default:
            line := scanner.Text()
            matches := logRegex.FindStringSubmatch(line)
            if matches == nil {
                continue
            }

            // Extract named groups
            action := matches[logRegex.SubexpIndex("action")]
            proto := matches[logRegex.SubexpIndex("proto")]
            ruleID := matches[logRegex.SubexpIndex("rule_id")]

            // Update metrics
            totalRequests.WithLabelValues(action, proto).Inc()
            ruleHitCount.WithLabelValues(ruleID).Inc()

            // Update sliding window for blocked ratio
            now := time.Now()
            if now.Sub(windowStart) > 5*time.Minute {
                // Reset window
                window = window[:0]
                windowStart = now
            }
            window = append(window, action)

            // Calculate blocked ratio every 10 seconds
            if len(window) > 0 && now.Second()%10 == 0 {
                dropped := 0
                for _, a := range window {
                    if a == "drop" {
                        dropped++
                    }
                }
                blockedRatio.Set(float64(dropped) / float64(len(window)))
            }
        }
    }

    if err := scanner.Err(); err != nil {
        return fmt.Errorf("scanner error: %w", err)
    }
    return nil
}

func main() {
    // Register Prometheus metrics
    prometheus.MustRegister(totalRequests)
    prometheus.MustRegister(blockedRatio)
    prometheus.MustRegister(ruleHitCount)

    // Start log parser
    parser := newLogParser()
    go func() {
        logPath := os.Getenv("FIREWALL_LOG_PATH")
        if logPath == "" {
            logPath = "/var/log/nftables.log"
        }
        if err := parser.parseLogFile(logPath); err != nil {
            log.Fatalf("Log parser failed: %v", err)
        }
    }()

    // Start Prometheus HTTP server
    http.Handle("/metrics", promhttp.Handler())
    addr := ":9091"
    log.Printf("Metrics server listening on %s", addr)
    if err := http.ListenAndServe(addr, nil); err != nil {
        log.Fatalf("HTTP server failed: %v", err)
    }
}

# AWS Network Firewall Deployment v2.1
# Modular, testable firewall config with cost estimation and rule validation.
# Requires: terraform >= 1.7.0, aws provider ~> 5.0

terraform {
  required_version = ">= 1.7.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
  # Store state in S3 for team collaboration
  backend "s3" {
    bucket         = "my-org-terraform-state"
    key            = "firewall/network-firewall.tfstate"
    region         = "us-east-1"
    encrypt        = true
    dynamodb_table = "terraform-lock"
  }
}

provider "aws" {
  region = var.aws_region
}

variable "aws_region" {
  type        = string
  default     = "us-east-1"
  description = "AWS region to deploy firewall in"
}

variable "vpc_id" {
  type        = string
  description = "ID of the VPC to attach firewall to"
  validation {
    condition     = length(var.vpc_id) > 0
    error_message = "VPC ID must not be empty."
  }
}

variable "allowed_ingress_ports" {
  type        = list(number)
  default     = [80, 443, 22]
  description = "Ports allowed for ingress traffic"
  validation {
    condition     = alltrue([for p in var.allowed_ingress_ports : p >= 1 && p <= 65535])
    error_message = "All ports must be between 1 and 65535."
  }
}

variable "blocked_cidrs" {
  type        = list(string)
  default     = ["10.0.0.0/8", "172.16.0.0/12"]
  description = "CIDR blocks to block ingress from"
  validation {
    condition     = alltrue([for cidr in var.blocked_cidrs : can(cidrhost(cidr, 0))])
    error_message = "All blocked CIDRs must be valid IPv4 CIDR blocks."
  }
}

# Fetch VPC data to get subnet IDs
data "aws_vpc" "selected" {
  id = var.vpc_id
}

data "aws_subnets" "public" {
  filter {
    name   = "vpc-id"
    values = [var.vpc_id]
  }
  tags = {
    Type = "Public"
  }
}

# IAM role for Network Firewall to publish logs to CloudWatch
resource "aws_iam_role" "firewall_logs" {
  name = "network-firewall-logs-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "network-firewall.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "firewall_logs" {
  name = "network-firewall-logs-policy"
  role = aws_iam_role.firewall_logs.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = [
          "logs:CreateLogGroup",
          "logs:CreateLogStream",
          "logs:PutLogEvents"
        ]
        Effect   = "Allow"
        Resource = "arn:aws:logs:*:*:log-group:/aws/network-firewall/*"
      }
    ]
  })
}

# Network Firewall policy with stateless and stateful rule groups
resource "aws_networkfirewall_firewall_policy" "main" {
  name = "main-firewall-policy"

  firewall_policy {
    # Stateless rule group for high-volume, simple filtering
    stateless_default_actions = ["aws:forward_to_sfe"]
    stateless_fragment_default_actions = ["aws:forward_to_sfe"]

    # Stateful rule group for application-layer filtering
    stateful_default_actions = ["aws:drop"]
    stateful_engine_options {
      rule_order = "STRICT_ORDER"
    }

    # Reference stateful rule groups
    stateful_rule_group_reference {
      resource_arn = aws_networkfirewall_rule_group.stateful.arn
    }
  }
}

# Stateful rule group: allow inbound on allowed ports, block blocked CIDRs
resource "aws_networkfirewall_rule_group" "stateful" {
  capacity = 1000
  name     = "stateful-ingress-rules"
  type     = "STATEFUL"
  rule_group {
    rules_source {
      rules_string = templatefile("${path.module}/stateful_rules.tpl", {
        allowed_ports = var.allowed_ingress_ports
        blocked_cidrs = var.blocked_cidrs
      })
    }
  }
}

# Network Firewall instance
resource "aws_networkfirewall_firewall" "main" {
  name                = "main-network-firewall"
  firewall_policy_arn = aws_networkfirewall_firewall_policy.main.arn
  vpc_id              = var.vpc_id
  subnet_mapping {
    subnet_id = data.aws_subnets.public.ids[0]
  }

  logging_configuration {
    log_destination_config {
      log_destination = {
        logGroup = aws_cloudwatch_log_group.firewall_logs.name
      }
      log_destination_type = "CloudWatchLogs"
      log_type             = "ALERT"
    }
  }

  tags = {
    Environment = "production"
    ManagedBy   = "terraform"
  }
}

# CloudWatch Log Group for firewall alerts
resource "aws_cloudwatch_log_group" "firewall_logs" {
  name              = "/aws/network-firewall/alerts"
  retention_in_days = 30
}

# Cost estimation output
output "estimated_monthly_cost_usd" {
  value = (
    # Network Firewall base cost: $0.25 per hour per endpoint
    (730 * 0.25) +
    # Rule group cost: $0.01 per rule per month (assuming 100 rules)
    (100 * 0.01) +
    # Log storage cost: $0.03 per GB, assume 10GB/month
    (10 * 0.03)
  )
  description = "Estimated monthly cost of the firewall deployment"
}

output "firewall_endpoint_id" {
  value       = aws_networkfirewall_firewall.main.firewall_id
  description = "ID of the deployed Network Firewall"
}

# CI validation step for nftables rules
validate_firewall:
  stage: test
  image: debian:12
  script:
    - apt-get update && apt-get install -y nftables python3-pip
    - pip install pyroute2==0.7.3
    - nft -d -f ./nftables.conf # Dry-run syntax check
    - python3 ./validate_rules.py ./nftables.conf # Check shadowed/redundant rules
  only:
    - merge_requests
    - main

# Ory Oathkeeper rule for admin access (replaces static IP allowlist)
id: "admin-access"
version: "v2.0.0"
match:
  url: "https://admin.example.com/*"
  methods: ["GET", "POST"]
  # No IP-based matching here
authenticators:
  - handler: jwt
    config:
      jwks_urls: ["https://sso.example.com/.well-known/jwks.json"]
      required_claims:
        roles: ["admin"]
authorizer:
  handler: allow
mutators:
  - handler: noop

# Prometheus alert for anomalous firewall blocked ratio
groups:
- name: firewall
  rules:
  - alert: HighFirewallBlockedRatio
    expr: firewall_blocked_ratio > 0.2
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "Firewall blocked ratio is {{ $value | printf '%.2f' }} (threshold 0.2)"
      description: "Blocked ratio for firewall is above 20% for 1 minute. Check for DDoS or misconfiguration."