Firewall Rule Order — Misordered DENY Led to 45-Min Outage

# Simulating a basic stateless packet-filtering firewall in Python.
# In production, this logic lives in kernel space (iptables, nftables, Windows Firewall).
# Here we model it in userspace so you can SEE the decision process.

from dataclasses import dataclass
from typing import List

@dataclass
class NetworkPacket:
    source_ip: str
    destination_ip: str
    destination_port: int
    protocol: str  # 'TCP', 'UDP', 'ICMP'

@dataclass
class FirewallRule:
    """
    A single firewall rule. Rules are evaluated top-to-bottom.
    The first matching rule wins — just like real iptables chains.
    """
    source_ip: str        # '*' means any
    destination_port: int # -1 means any
    protocol: str         # '*' means any
    action: str           # 'ALLOW' or 'DENY'

class PacketFilterFirewall:
    def __init__(self, default_policy: str = 'DENY'):
        """
        Default policy is 'DENY' — this is the secure default.
        'Allow all, deny specific' is the WRONG approach (allowlist vs denylist).
        """
        self.rules: List[FirewallRule] = []
        self.default_policy = default_policy

    def add_rule(self, rule: FirewallRule):
        # Rules are ordered — first match wins, so order matters enormously
        self.rules.append(rule)

    def inspect(self, packet: NetworkPacket) -> str:
        for rule in self.rules:
            # Check each condition — '*' is a wildcard that matches anything
            ip_match = (rule.source_ip == '*' or rule.source_ip == packet.source_ip)
            port_match = (rule.destination_port == -1 or rule.destination_port == packet.destination_port)
            proto_match = (rule.protocol == '*' or rule.protocol == packet.protocol)

            if ip_match and port_match and proto_match:
                # Found the first matching rule — enforce it immediately
                return rule.action

        # No rule matched — fall back to the default policy
        return self.default_policy

# --- Setting up the firewall ---
firewall = PacketFilterFirewall(default_policy='DENY')

# Rule 1: Allow all HTTP traffic from anywhere
firewall.add_rule(FirewallRule(source_ip='*', destination_port=80, protocol='TCP', action='ALLOW'))

# Rule 2: Allow HTTPS from anywhere
firewall.add_rule(FirewallRule(source_ip='*', destination_port=443, protocol='TCP', action='ALLOW'))

# Rule 3: Allow SSH only from the trusted admin IP
firewall.add_rule(FirewallRule(source_ip='10.0.0.5', destination_port=22, protocol='TCP', action='ALLOW'))

# Rule 4: Explicitly block a known malicious IP on any port
firewall.add_rule(FirewallRule(source_ip='185.220.101.9', destination_port=-1, protocol='*', action='DENY'))

# --- Testing packets against the firewall ---
test_packets = [
    NetworkPacket('203.0.113.42', '192.168.1.1', 80,  'TCP'),  # Regular web request
    NetworkPacket('203.0.113.42', '192.168.1.1', 443, 'TCP'),  # HTTPS request
    NetworkPacket('198.51.100.7', '192.168.1.1', 22,  'TCP'),  # SSH from unknown IP
    NetworkPacket('10.0.0.5',     '192.168.1.1', 22,  'TCP'),  # SSH from trusted admin
    NetworkPacket('185.220.101.9','192.168.1.1', 443, 'TCP'),  # Known bad actor on HTTPS
    NetworkPacket('203.0.113.42', '192.168.1.1', 8080,'TCP'),  # Non-standard port — no rule
]

print(f"{'Source IP':<20} {'Port':<6} {'Result'}")
print('-' * 40)
for pkt in test_packets:
    result = firewall.inspect(pkt)
    print(f"{pkt.source_ip:<20} {pkt.destination_port:<6} {result}")

# This demo simulates the REQUEST FLOW through both a forward and reverse proxy.
# We use Python's http.server and threading to run real local HTTP servers.
# Run this script and watch the printed logs to see who talks to whom.

import threading
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.request import urlopen, Request
from urllib.error import URLError

# ─────────────────────────────────────────────────
# 1. THE ORIGIN SERVER — represents your backend API
# ─────────────────────────────────────────────────
class OriginServerHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Log who is connecting — in a reverse proxy setup, this will be
        # the proxy's IP, NOT the original client's IP
        print(f"[ORIGIN SERVER] Received request from: {self.client_address[0]}")
        print(f"[ORIGIN SERVER] Path requested: {self.path}")

        response_body = b"Hello from the Origin Server! Path: " + self.path.encode()
        self.send_response(200)
        self.send_header('Content-Type', 'text/plain')
        self.send_header('Content-Length', len(response_body))
        self.end_headers()
        self.wfile.write(response_body)

    # Suppress default request logging to keep output clean
    def log_message(self, format, *args): pass

# ─────────────────────────────────────────────────
# 2. THE REVERSE PROXY — sits in front of origin
# ─────────────────────────────────────────────────
class ReverseProxyHandler(BaseHTTPRequestHandler):
    ORIGIN_SERVER_URL = 'http://127.0.0.1:9001'
    BLOCKED_PATHS = ['/admin', '/internal']

    def do_GET(self):
        print(f"\n[REVERSE PROXY] Client {self.client_address[0]} wants: {self.path}")

        # Block sensitive internal paths — client never knows these exist
        if self.path in self.BLOCKED_PATHS:
            print(f"[REVERSE PROXY] BLOCKING sensitive path: {self.path}")
            self.send_response(403)
            self.end_headers()
            self.wfile.write(b"403 Forbidden")
            return

        # Forward the request to the origin server on the client's behalf
        # The origin server will see 127.0.0.1, not the real client IP
        target_url = self.ORIGIN_SERVER_URL + self.path
        print(f"[REVERSE PROXY] Forwarding to origin: {target_url}")

        try:
            with urlopen(Request(target_url), timeout=5) as origin_response:
                body = origin_response.read()

            # Pass the origin's response back to the original client
            self.send_response(200)
            self.send_header('Content-Type', 'text/plain')
            self.send_header('X-Served-By', 'TheCodeForge-ReverseProxy')  # Custom header
            self.end_headers()
            self.wfile.write(body)
            print(f"[REVERSE PROXY] Successfully relayed {len(body)} bytes back to client")

        except URLError as network_error:
            print(f"[REVERSE PROXY] Origin server unreachable: {network_error}")
            self.send_response(502)  # 502 Bad Gateway — classic reverse proxy error
            self.end_headers()
            self.wfile.write(b"502 Bad Gateway")

    def log_message(self, format, *args): pass

# ─────────────────────────────────────────────────
# 3. SPIN UP BOTH SERVERS IN BACKGROUND THREADS
# ─────────────────────────────────────────────────
def start_server(handler_class, port):
    server = HTTPServer(('127.0.0.1', port), handler_class)
    server.serve_forever()

origin_thread = threading.Thread(target=start_server, args=(OriginServerHandler, 9001), daemon=True)
proxy_thread  = threading.Thread(target=start_server, args=(ReverseProxyHandler,  9000), daemon=True)

origin_thread.start()
proxy_thread.start()
time.sleep(0.5)  # Give servers a moment to bind their ports

# ─────────────────────────────────────────────────
# 4. SIMULATE CLIENT REQUESTS (client talks to PROXY only)
# ─────────────────────────────────────────────────
print("=" * 55)
print("CLIENT: Requesting /api/users through the reverse proxy")
print("=" * 55)
with urlopen('http://127.0.0.1:9000/api/users', timeout=5) as resp:
    print(f"CLIENT received: {resp.read().decode()}")

time.sleep(0.2)

print("\n" + "=" * 55)
print("CLIENT: Attempting to access /admin (should be blocked)")
print("=" * 55)
try:
    urlopen('http://127.0.0.1:9000/admin', timeout=5)
except Exception as e:
    print(f"CLIENT received error (expected): HTTP 403")

time.sleep(0.2)
print("\nDemo complete.")

# Production-grade Nginx config that acts as both a reverse proxy AND
# an application-layer access control layer.
# This sits in front of a Node.js app running on port 3000 internally.

# --- Rate limiting zone: track clients by IP, 10MB memory, max 30 req/min ---
limit_req_zone $binary_remote_addr zone=api_rate_limit:10m rate=30r/m;

server {
    listen 443 ssl http2;
    server_name api.yourcorp.com;

    # TLS termination happens HERE — backend never handles raw TLS
    ssl_certificate     /etc/ssl/certs/yourcorp.crt;
    ssl_certificate_key /etc/ssl/private/yourcorp.key;
    ssl_protocols       TLSv1.2 TLSv1.3;  # Reject weak TLS 1.0 and 1.1

    # ── Block internal/admin paths from public internet ──────────────────────
    location ~ ^/(internal|metrics|health/debug) {
        # Only allow requests from the internal VPC CIDR range
        allow 10.0.0.0/8;
        deny  all;  # Everyone else gets 403 — firewall at the HTTP layer
    }

    # ── Public API — apply rate limiting ─────────────────────────────────────
    location /api/ {
        # Apply rate limit — burst of 10 requests allowed before throttling
        limit_req zone=api_rate_limit burst=10 nodelay;

        # THE CORE PROXY ACTION: forward to backend, hide backend's identity
        proxy_pass http://127.0.0.1:3000;

        # Tell the backend the REAL client IP (not Nginx's loopback address)
        proxy_set_header X-Forwarded-For $remote_addr;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Strip headers the backend might leak to clients
        proxy_hide_header X-Powered-By;  # Don't expose 'Express' or 'Node'
        proxy_hide_header Server;        # Don't expose backend server version

        # Timeout settings — don't let slow backends hold connections forever
        proxy_connect_timeout 5s;
        proxy_read_timeout    30s;
    }

    # ── Static assets — serve directly, never hit the backend ────────────────
    location /static/ {
        root /var/www/yourcorp;
        expires 30d;       # Cache for 30 days — no need to hit backend
        add_header Cache-Control "public, immutable";
    }

    # ── Redirect all HTTP to HTTPS at the server level ───────────────────────
    # (The firewall allows port 80 inbound only for this redirect)
}

server {
    listen 80;
    server_name api.yourcorp.com;
    return 301 https://$host$request_uri;  # Permanent redirect to HTTPS
}

# Terraform example showing Security Group vs NACL rules for a web tier.
# The Security Group allows HTTP/S inbound; the NACL additionally controls ephemeral ports.

resource "aws_security_group" "web_sg" {
  name_prefix = "web-tier-sg-"
  description = "Allow HTTP and HTTPS from internet"
  vpc_id      = aws_vpc.main.id

  ingress {
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
    description = "HTTPS from anywhere"
  }

  ingress {
    from_port   = 80
    to_port     = 80
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
    description = "HTTP redirect from anywhere"
  }

  # SSH only from corporate CIDR (bastion or VPN)
  ingress {
    from_port   = 22
    to_port     = 22
    protocol    = "tcp"
    cidr_blocks = ["203.0.113.0/24"]  # Replace with your corp IP range
    description = "SSH from corporate network"
  }

  # Security Group is stateful — no need for explicit egress rule for replies
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_network_acl" "public_subnet_nacl" {
  vpc_id = aws_vpc.main.id
  subnet_ids = [aws_subnet.public.id]

  # NACL is STATELESS — must allow both inbound AND outbound ephemeral responses
  ingress {
    rule_no    = 100
    from_port  = 80
    to_port    = 80
    protocol   = "tcp"
    cidr_block = "0.0.0.0/0"
    action     = "allow"
  }

  ingress {
    rule_no    = 110
    from_port  = 443
    to_port    = 443
    protocol   = "tcp"
    cidr_block = "0.0.0.0/0"
    action     = "allow"
  }

  # CRITICAL: allow ephemeral inbound responses from internet (stateless requires this)
  ingress {
    rule_no    = 120
    from_port  = 1024
    to_port    = 65535
    protocol   = "tcp"
    cidr_block = "0.0.0.0/0"
    action     = "allow"
  }

  # Outbound: allow HTTP/S to internet, and ephemeral for responses to clients
  egress {
    rule_no    = 100
    from_port  = 80
    to_port    = 80
    protocol   = "tcp"
    cidr_block = "0.0.0.0/0"
    action     = "allow"
  }

  egress {
    rule_no    = 110
    from_port  = 443
    to_port    = 443
    protocol   = "tcp"
    cidr_block = "0.0.0.0/0"
    action     = "allow"
  }

  # Outbound ephemeral responses to clients
  egress {
    rule_no    = 120
    from_port  = 1024
    to_port    = 65535
    protocol   = "tcp"
    cidr_block = "0.0.0.0/0"
    action     = "allow"
  }

  # Deny all other traffic (implicit but best practice to have explicit deny)
  ingress {
    rule_no    = 200
    from_port  = 0
    to_port    = 0
    protocol   = "-1"
    cidr_block = "0.0.0.0/0"
    action     = "deny"
  }

  egress {
    rule_no    = 200
    from_port  = 0
    to_port    = 0
    protocol   = "-1"
    cidr_block = "0.0.0.0/0"
    action     = "deny"
  }
}

# Nginx reverse proxy configuration with authentication, caching, and logging.

# Cache zone: 1GB memory, lasts 60 minutes, not accessed for 10 minutes = purge
proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=my_cache:1g max_size=10g inactive=10m use_temp_path=off;

server {
    listen 443 ssl http2;

    # ── Authentication: validate JWT from header before proxying ──────────
    # (Simplified — in production use auth_request subrequest or modules)
    location /api/ {
        # Custom auth check: if header missing, return 401
        if ($http_authorization = "") {
            return 401;
        }

        # Cache configuration — vary on Authorization header for private responses
        proxy_cache my_cache;
        proxy_cache_key "$scheme$request_method$host$request_uri$http_authorization";
        proxy_cache_valid 200 1h;       # Cache 200 OK for 1 hour
        proxy_cache_valid 404 1m;       # Cache 404 for 1 minute (avoids stampede)
        proxy_cache_use_stale error timeout updating;  # Serve stale on backend failure

        proxy_pass http://backend:3000;
        proxy_set_header X-Forwarded-For $remote_addr;
        proxy_set_header Host $host;
    }

    # ── Caching static assets aggressively ─────────────────────────────────
    location /static/ {
        proxy_cache my_cache;
        proxy_cache_valid 200 30d;      # Cache images for 30 days
        add_header Cache-Control "public, immutable";
        expires 30d;
        root /var/www/static;
    }

    # ── Logging with buffering to avoid disk I/O bottleneck ────────────────
    access_log /var/log/nginx/api_access.log buffer=32k flush=5s;
    error_log  /var/log/nginx/api_error.log warn;

    # JSON formatted access log for easier parsing
    log_format json escape=json
        '{'
            '"time":"$time_iso8601",'
            '"remote_addr":"$remote_addr",'
            '"request":"$request",'
            '"status":$status,'
            '"body_bytes_sent":$body_bytes_sent,'
            '"request_time":$request_time,'
            '"http_x_forwarded_for":"$http_x_forwarded_for"'
        '}';
    access_log /var/log/nginx/api_json.log json buffer=32k flush=5s;
}

// io.thecodeforge — cs-fundamentals tutorial

import subprocess
import json

# Simulate inspecting a stateful firewall's connection table (e.g., iptables conntrack)
# In production, you'd parse /proc/net/nf_conntrack or use conntrack -L

def inspect_state_table():
    """
    Check for half-open connections that indicate state table exhaustion.
    Common symptom: clients can connect but requests timeout randomly.
    """
    try:
        # Replace with real conntrack command if available
        result = subprocess.run(
            ["conntrack", "-L", "--output", "json"],
            capture_output=True,
            text=True,
            timeout=5
        )
        if result.returncode != 0:
            print("Error: conntrack not found or permissions denied")
            return

        connections = json.loads(result.stdout)
        total_connections = len(connections)
        half_open = [
            conn for conn in connections
            if conn.get("tcp-state") in ["SYN_SENT", "SYN_RECV"]
        ]

        print(f"Total tracked connections: {total_connections}")
        print(f"Half-open (potentially abusive): {len(half_open)}")
        print(f"State table utilization: {(total_connections / 65536) * 100:.1f}% (max 65536)")

        if len(half_open) > 1000:
            print("WARNING: Possible SYN flood or connection leak detected")

    except FileNotFoundError:
        print("Firewall state table inspection requires root or conntrack module")

if __name__ == "__main__":
    inspect_state_table()

# io.thecodeforge — cs-fundamentals tutorial
# App-level gateway: enforce header whitelist

import socket, ssl

def filter_request(headers: dict) -> bool:
    allowed_hosts = {"api.thecodeforge.io", "cdn.thecodeforge.io"}
    allowed_methods = {"GET", "POST"}
    if headers.get("Host") not in allowed_hosts:
        return False
    if headers.get(":method") not in allowed_methods:
        return False
    return True

# Inside proxy handler:
def handle_client(client_sock):
    data = client_sock.recv(8192)
    headers = parse_http_request(data)  # assumes helper
    if not filter_request(headers):
        client_sock.send(b"HTTP/1.1 403 Forbidden\r\n\r\n")
        client_sock.close()
        return
    # forward to upstream
    upstream = socket.create_connection((headers['Host'], 443))
    tls_up = ssl.wrap_socket(upstream)
    tls_up.send(data)
    client_sock.send(tls_up.recv(8192))

// io.thecodeforge — cs-fundamentals tutorial

import socket
import ssl

def reject_weak_tls(proxy_socket):
    context = ssl.create_default_context()
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    try:
        secure_sock = context.wrap_socket(
            proxy_socket, server_side=True
        )
        cipher = secure_sock.cipher()
        # cipher[0] = name, cipher[1] = protocol
        if 'TLSv1.0' in cipher[1] or 'TLSv1.1' in cipher[1]:
            secure_sock.close()
            raise ConnectionRefusedError(
                "TLS 1.0/1.1 rejected by security policy"
            )
        return secure_sock
    except ssl.SSLError as e:
        print(f"Blocked connection: {e}")
        proxy_socket.close()
        return None

// io.thecodeforge — cs-fundamentals tutorial

class FirewallRule:
    def evaluate(self, packet):
        return (
            packet['source_ip'] in ALLOWED_SOURCES
            and packet['dest_port'] == 443
        )

class ReverseProxy:
    def handle(self, request):
        if not self._authenticate(request.headers):
            return 401, "Unauthorized"
        if '/internal/' in request.path:
            return 403, "Forbidden"
        return self._origin.fetch(request)

def main_pipeline(packet, request):
    if not FirewallRule().evaluate(packet):
        return drop(packet)
    status, msg = ReverseProxy().handle(request)
    return status, msg

# Example output
print(main_pipeline(packet_test, request_test))

// io.thecodeforge — cs-fundamentals tutorial

# Simulates why blocking at firewall alone misses app-layer attacks
import re

class Firewall:
    def __init__(self):
        self.blocked_ips = {"10.0.0.1"}

    def allow_packet(self, ip, port):
        return ip not in self.blocked_ips and port == 443

class ReverseProxy:
    def inspect_request(self, path, body):
        # Catches SQL injection a firewall would miss
        if re.search(r"['\"].*OR.*1=1", body, re.I):
            return "BLOCKED: SQL injection"
        return f"Allowed: {path}"

# Attack from trusted IP with malicious payload
fw = Firewall()
if fw.allow_packet("192.168.1.5", 443):
    result = ReverseProxy().inspect_request("/login", "username=' OR 1=1--")
    print(result)  # BLOCKED: SQL injection

// io.thecodeforge — cs-fundamentals tutorial

# Shows how default outbound allow creates hidden risk
class SecurityGroup:
    def __init__(self, inbound_default="deny", outbound_default="allow"):
        self.inbound_default = inbound_default
        self.outbound_default = outbound_default
        self.rules = []

    def add_inbound_rule(self, proto, port, cidr):
        self.rules.append(("in", proto, port, cidr))

    def check_outbound(self, dest_ip, port):
        # Default allow means any external IP is reachable
        return self.outbound_default == "allow"

sg = SecurityGroup()
sg.add_inbound_rule("tcp", 443, "0.0.0.0/0")
print("Outbound to malicious C2:", sg.check_outbound("5.5.5.5", 4444))  # True