How to How We Survived Firewall: What Works

import httpx
import ssl
import logging
import time
import os
import sys
from typing import Optional, Dict, Any

# Configure logging to capture firewall rejection events
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

class FirewallResilientClient:
    """HTTP/2 client optimized for traversal of enterprise firewalls with strict egress rules."""

    def __init__(
        self,
        base_url: str,
        max_retries: int = 3,
        timeout: float = 10.0,
        tls_1_3_only: bool = True
    ):
        self.base_url = base_url
        self.max_retries = max_retries
        self.timeout = timeout
        self.ssl_context = self._create_ssl_context(tls_1_3_only)
        # Use HTTP/2 by default, falls back to HTTP/1.1 if firewall blocks ALPN
        self.client = httpx.Client(
            http2=True,
            timeout=httpx.Timeout(timeout),
            verify=self.ssl_context,
            follow_redirects=True
        )

    def _create_ssl_context(self, tls_1_3_only: bool) -> ssl.SSLContext:
        """Create SSL context that only negotiates TLS 1.3 to avoid firewall DPI stripping older versions."""
        ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        if tls_1_3_only:
            # Explicitly disable TLS 1.2 and below, which are commonly blocked by next-gen firewalls
            ctx.options |= ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 | ssl.OP_NO_TLSv1_2
            # Enforce TLS 1.3 only (requires Python 3.7+)
            ctx.minimum_version = ssl.TLSVersion.TLSv1_3
            ctx.maximum_version = ssl.TLSVersion.TLSv1_3
        return ctx

    def request(
        self,
        method: str,
        endpoint: str,
        headers: Optional[Dict[str, str]] = None,
        data: Optional[Dict[str, Any]] = None
    ) -> httpx.Response:
        """Send request with exponential backoff retry for firewall-related errors."""
        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
        retry_delay = 1.0

        for attempt in range(self.max_retries + 1):
            try:
                logger.info(f"Attempt {attempt + 1} to {method} {url}")
                response = self.client.request(
                    method=method,
                    url=url,
                    headers=headers,
                    json=data
                )
                # Retry on 403 (common firewall block response) or 504 (gateway timeout from DPI)
                if response.status_code in (403, 504) and attempt < self.max_retries:
                    logger.warning(f"Firewall blocked request (status {response.status_code}), retrying...")
                    time.sleep(retry_delay)
                    retry_delay *= 2  # Exponential backoff
                    continue
                response.raise_for_status()
                return response
            except (httpx.ConnectError, httpx.ConnectTimeout, httpx.RemoteProtocolError) as e:
                # These errors almost always indicate firewall interference
                logger.error(f"Firewall connection error: {str(e)}")
                if attempt < self.max_retries:
                    time.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logger.critical(f"Max retries exceeded for {url}")
                    raise
            except Exception as e:
                # Catch-all for unexpected errors, log and re-raise
                logger.error(f"Unexpected error: {str(e)}")
                raise
        raise RuntimeError("Failed to send request after max retries")

    def close(self):
        """Clean up HTTP client resources."""
        self.client.close()

if __name__ == "__main__":
    # Example usage: query a public API behind a firewall
    client = FirewallResilientClient(base_url="https://api.example.com")
    try:
        response = client.request(method="GET", endpoint="/health")
        print(f"Response status: {response.status_code}")
        print(f"Response body: {response.text}")
    except Exception as e:
        print(f"Failed to reach API: {str(e)}", file=sys.stderr)
        sys.exit(1)
    finally:
        client.close()

package main

import (
    "context"
    "fmt"
    "log"
    "net"
    "os"
    "os/signal"
    "syscall"
    "time"

    "golang.zx2c4.com/wireguard/device"
    "golang.zx2c4.com/wireguard/tun"
)

// TunnelConfig holds WireGuard tunnel parameters for firewall traversal
type TunnelConfig struct {
    LocalIP      string        // Internal IP assigned to the tunnel interface
    PublicKey    string        // Public key of the remote WireGuard peer
    Endpoint     string        // Remote endpoint (IP:port, port 443 for firewall bypass)
    AllowedIPs   []string      // CIDR ranges to route through the tunnel
    MTU          int           // MTU to avoid fragmentation (1420 is standard for WireGuard)
    Keepalive    time.Duration // Persistent keepalive to prevent firewall NAT mapping expiration
}

func main() {
    // Example config: tunnel all traffic to 10.0.0.0/8 through a WireGuard peer on port 443
    cfg := TunnelConfig{
        LocalIP:   "10.0.0.2/32",
        PublicKey: "remote-peer-public-key-here",
        Endpoint:  "203.0.113.5:443", // Port 443 is rarely blocked by egress firewalls
        AllowedIPs: []string{"10.0.0.0/8"},
        MTU:       1420,
        Keepalive: 25 * time.Second, // Send keepalive every 25s to keep NAT mapping open
    }

    // Create TUN device (requires root privileges)
    tunDevice, err := tun.CreateTUN("wg-firewall", cfg.MTU)
    if err != nil {
        log.Fatalf("Failed to create TUN device: %v", err)
    }
    defer tunDevice.Close()

    // Get TUN file descriptor for WireGuard device initialization
    tunFD, err := tunDevice.File()
    if err != nil {
        log.Fatalf("Failed to get TUN file descriptor: %v", err)
    }
    defer tunFD.Close()

    // Initialize WireGuard device with the TUN FD
    wgDevice, err := device.NewDevice(tunDevice, cfg.MTU, device.NewLogger(device.LogLevelInfo, "wireguard"))
    if err != nil {
        log.Fatalf("Failed to create WireGuard device: %v", err)
    }
    defer wgDevice.Close()

    // Configure WireGuard device with our tunnel settings
    // Note: In production, use wgctrl-go to configure peers instead of raw UAPI for better reliability
    uapiCfg := fmt.Sprintf(`
private_key=%s
listen_port=0
replace_peers=true
public_key=%s
endpoint=%s
allowed_ip=%s
persistent_keepalive_interval=%d
`,
        "local-private-key-here", // Load from secure secret manager in production
        cfg.PublicKey,
        cfg.Endpoint,
        cfg.AllowedIPs[0],
        int(cfg.Keepalive.Seconds()),
    )

    if err := wgDevice.IpcSet(uapiCfg); err != nil {
        log.Fatalf("Failed to configure WireGuard device: %v", err)
    }

    // Assign local IP to the TUN interface
    if err := configureTUNIP(tunDevice, cfg.LocalIP); err != nil {
        log.Fatalf("Failed to assign IP to TUN interface: %v", err)
    }

    log.Println("WireGuard tunnel established successfully")
    log.Printf("Routing traffic to %v via %s", cfg.AllowedIPs, cfg.Endpoint)

    // Wait for interrupt signal to shut down gracefully
    ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer cancel()

    <-ctx.Done()
    log.Println("Shutting down WireGuard tunnel...")
}

// configureTUNIP assigns an IP address to the TUN interface (Linux-only, extend for other OS)
func configureTUNIP(tun tun.Device, ipCIDR string) error {
    // Note: This is a simplified example, use netlink or iproute2 in production for cross-platform support
    ip, _, err := net.ParseCIDR(ipCIDR)
    if err != nil {
        return fmt.Errorf("invalid CIDR: %w", err)
    }
    // In a real implementation, you would use syscall or netlink to add the IP to the interface
    // This is omitted for brevity but required for the tunnel to function
    log.Printf("Assigned IP %s to TUN interface (simulated for example)", ip.String())
    return nil
}

# Terraform configuration for deploying an stunnel sidecar to bypass firewall TLS inspection
# Requires: Terraform 1.6+, kubernetes provider 2.20+, stunnel 5.72+

terraform {
  required_version = ">= 1.6.0"
  required_providers {
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = ">= 2.20.0"
    }
  }
}

variable "stunnel_image" {
  type        = string
  default     = "stunnel/stunnel:5.72"
  description = "Stunnel image version, pinned to 5.72 for consistent firewall traversal behavior"
}

variable "target_endpoint" {
  type        = string
  description = "Remote endpoint to tunnel to, e.g., 203.0.113.5:443"
}

variable "kubernetes_namespace" {
  type        = string
  default     = "firewall-tunnels"
  description = "Namespace to deploy the stunnel sidecar into"
}

# Create namespace for tunnel resources
resource "kubernetes_namespace" "tunnel_ns" {
  metadata {
    name = var.kubernetes_namespace
    labels = {
      app = "stunnel-sidecar"
      env = "production"
    }
  }
}

# ConfigMap holding stunnel configuration
resource "kubernetes_config_map" "stunnel_config" {
  metadata {
    name      = "stunnel-config"
    namespace = kubernetes_namespace.tunnel_ns.metadata[0].name
  }

  data = {
    "stunnel.conf" = <<-EOT
      setuid = stunnel4
      setgid = stunnel4
      pid = /var/run/stunnel.pid
      # Use TLS 1.3 to avoid firewall DPI stripping older versions
      sslVersion = TLSv1.3
      # Listen on localhost:8080, forward to remote endpoint via TLS
      [tunnel]
      accept = 127.0.0.1:8080
      connect = ${var.target_endpoint}
      # Enable client mode (we are initiating the tunnel from the pod)
      client = yes
      # Verify remote certificate to prevent MITM by firewall
      verify = 2
      CAfile = /etc/stunnel/certs/ca.crt
      # Disable compression to avoid firewall false positives for "suspicious" payloads
      compression = zlib
    EOT
  }
}

# Secret holding CA certificate for remote endpoint verification
resource "kubernetes_secret" "stunnel_ca" {
  metadata {
    name      = "stunnel-ca"
    namespace = kubernetes_namespace.tunnel_ns.metadata[0].name
  }

  data = {
    "ca.crt" = filebase64("${path.module}/certs/ca.crt") # Load CA cert from local file, base64 encoded for K8s
  }

  type = "Opaque"
}

# Deployment for stunnel sidecar, runs alongside application pods
resource "kubernetes_deployment" "stunnel_sidecar" {
  metadata {
    name      = "stunnel-sidecar"
    namespace = kubernetes_namespace.tunnel_ns.metadata[0].name
    labels = {
      app = "stunnel-sidecar"
    }
  }

  spec {
    replicas = 2 # Run 2 replicas for high availability

    selector {
      match_labels = {
        app = "stunnel-sidecar"
      }
    }

    template {
      metadata {
        labels = {
          app = "stunnel-sidecar"
        }
      }

      spec {
        container {
          name  = "stunnel"
          image = var.stunnel_image
          port {
            container_port = 8080
            protocol       = "TCP"
          }
          volume_mount {
            name       = "stunnel-config"
            mount_path = "/etc/stunnel/stunnel.conf"
            sub_path   = "stunnel.conf"
          }
          volume_mount {
            name       = "stunnel-ca"
            mount_path = "/etc/stunnel/certs/ca.crt"
            sub_path   = "ca.crt"
          }
          # Liveness probe to restart stunnel if it crashes due to firewall interference
          liveness_probe {
            tcp_socket {
              port = 8080
            }
            initial_delay_seconds = 10
            period_seconds        = 5
          }
        }

        volume {
          name = "stunnel-config"
          config_map {
            name = kubernetes_config_map.stunnel_config.metadata[0].name
          }
        }

        volume {
          name = "stunnel-ca"
          secret {
            secret_name = kubernetes_secret.stunnel_ca.metadata[0].name
          }
        }
      }
    }
  }
}

output "stunnel_service_endpoint" {
  value       = "127.0.0.1:8080"
  description = "Local endpoint applications can use to route traffic through the stunnel tunnel"
}

# Example go.mod snippet with pinned WireGuard version
module github.com/yourorg/firewall-tunnel

go 1.21

require (
    golang.zx2c4.com/wireguard v0.0.0-20211208183345-5a6f7d1e8b1a // WireGuard 1.0.20211208
    golang.zx2c4.com/wireguard/wgctrl v0.0.0-20211222083345-3a6f7d1e8b1a
)

# Python example enabling TCP keepalives for HTTP client
import socket
import httpx

def create_keepalive_client():
    # Create custom transport with TCP keepalives enabled
    transport = httpx.HTTPTransport(
        socket_options=[
            (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
            (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 20), # 20s idle before keepalive
            (socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10), # 10s between keepalives
            (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3), # 3 failed keepalives before drop
        ]
    )
    return httpx.Client(transport=transport, http2=True)

# Example CI step using containerized pfSense to test firewall traversal
name: Firewall Compatibility Test
on: [push]
jobs:
  test-firewall:
    runs-on: ubuntu-latest
    steps:
      - name: Start pfSense container
        run: docker run -d --name pfsense -p 443:443 pfsense/pfsense:2.7.0
      - name: Run connection tests
        run: go test -v ./... -firewall-endpoint=localhost:443
      - name: Stop pfSense container
        run: docker stop pfsense

firewall-survival-kit/
├── python-client/
│   ├── __init__.py
│   ├── client.py          # Code Example 1: FirewallResilientClient
│   ├── requirements.txt   # httpx>=0.24.0, pytest>=7.4.0
│   └── tests/
│       └── test_client.py
├── go-wireguard/
│   ├── go.mod             # Code Example 2: WireGuard sidecar
│   ├── main.go
│   └── tunnel.go
├── terraform-stunnel/
│   ├── main.tf            # Code Example 3: Stunnel Terraform deployment
│   ├── variables.tf
│   ├── outputs.tf
│   └── certs/
│       └── ca.crt
├── benchmarks/
│   ├── results.csv        # 10k test results from comparison table
│   └── run-benchmarks.sh
└── README.md              # Setup instructions and troubleshooting