badtests/test.py

#!/usr/bin/env python3
'''test.py - test all badtests.py endpoints with urllib.request, http.client, requests, and aiohttp'''

import http.client
import urllib.request
import sys
import time
import traceback

try:
    import requests as req_lib
except ImportError:
    req_lib = None
    print('[!] requests not installed - pip install requests')

try:
    import aiohttp
    import asyncio
except ImportError:
    aiohttp = None
    print('[!] aiohttp not installed - pip install aiohttp')

HOST = sys.argv[1] if len(sys.argv) > 1 else 'localhost'
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 60666
BASE = f'http://{HOST}:{PORT}'
SEP  = '--------------------------------------------'
TIMEOUT = 5

ENDPOINTS = [
    # (path, title, extra_flags)
    # extra_flags: 'slow' = uses timeout, 'endless' = streaming, 'redir' = follows redirects, 'binary' = don't decode

    # Content-Length Abuses
    ('/cl-word',         'Content-Length: banana',             ''),
    ('/cl-empty',        'Content-Length: (empty)',            ''),
    ('/cl-negative',     'Content-Length: -99999999999999999', ''),
    ('/cl-huge',         'Content-Length: 99999999999999999',  'slow'),
    ('/cl-zero',         'Content-Length: 0 (body sent)',      ''),
    ('/cl-float',        'Content-Length: 3.14159',            ''),
    ('/cl-hex',          'Content-Length: 0xDEAD',             ''),
    ('/cl-null',         'Content-Length: 5\\x00',             ''),
    ('/cl-negative-one', 'Content-Length: -1',                 ''),
    ('/cl-nan',          'Content-Length: NaN',                ''),
    ('/cl-inf',          'Content-Length: Infinity',           ''),
    ('/cl-mismatch',     'Content-Length: 1 (body 30 bytes)', ''),

    # Content-Type Abuses
    ('/ct-garbage',      'Content-Type: 200 random chars',    ''),
    ('/ct-multi',        'Content-Type: 5 MIME types',        ''),
    ('/ct-empty',        'Content-Type: (empty)',              ''),
    ('/ct-null',         'Content-Type: text/\\x00html',      ''),
    ('/ct-emoji',        'Content-Type: emoji',                ''),
    ('/ct-mega',         'Content-Type: 10KB long',            ''),
    ('/ct-crlf',         'Content-Type with CRLF injection',  ''),
    ('/ct-semicolon',    'Content-Type with 50 charsets',     ''),

    # Status Code Abuses
    ('/status-neg',      'Status: -1',                        ''),
    ('/status-zero',     'Status: 0',                         ''),
    ('/status-999',      'Status: 999',                       ''),
    ('/status-huge',     'Status: 99999',                     ''),
    ('/status-word',     'Status: BANANA',                    ''),

    # Header Abuses
    ('/hdr-mega-name',   'Header name: 64KB of As',           ''),
    ('/hdr-mega-val',    'Header value: 1MB of Bs',           ''),
    ('/hdr-1000',        '1000 headers',                      ''),
    ('/hdr-empty-name',  'Header: : empty_name',              ''),
    ('/hdr-no-colon',    'Header without colon',              ''),
    ('/hdr-dup-cl',      'Duplicate Content-Length',           ''),
    ('/hdr-null',        'Header with null bytes',            ''),
    ('/hdr-newline',     'Header with bare \\n',              ''),

    # Body Abuses
    ('/body-endless',    'Infinite body stream',              'slow'),
    ('/body-none',       'Content-Length: 5, no body',        'slow'),
    ('/body-binary',     '4KB binary as text/html',           'binary'),

    # Transfer-Encoding Abuses
    ('/te-invalid',      'Transfer-Encoding: banana',         ''),
    ('/te-double',       'Content-Length + Transfer-Encoding', ''),
    ('/te-bad-chunk',    'Chunked with ZZZ chunk size',       ''),

    # Protocol Abuses
    ('/proto-garbage',       '512 random bytes (no HTTP)',     ''),
    ('/proto-empty',         'Empty response (0 bytes)',       ''),
    ('/proto-half',          'HTTP/9.9 version',               ''),
    ('/proto-no-headers',    'Status line + body, no headers', ''),
    ('/proto-just-newlines', 'Response is just CRLF',          ''),

    # Encoding Abuses
    ('/enc-utf16',       'Body is UTF-16, header says UTF-8', 'binary'),
    ('/enc-gzip-lie',    'Content-Encoding: gzip (plaintext)',''),

    # Redirect Abuses
    ('/redir-self',      '301 redirect loop',                 ''),
    ('/redir-garbage',   '301 garbage Location URL',          ''),
    ('/redir-chain',     '302 infinite unique chain',         ''),

    # Timing Abuses
    ('/slow-headers',    'Slowloris: 1 byte per 0.5s',       'slow'),
    ('/slow-body',       'Body: 1 byte per second (a-z)',     'slow'),

    # HTML/Content Abuses
    ('/html-utf16',           'Full HTML page as UTF-16LE',   ''),
    ('/html-long-title',      'HTML title is 1000 characters',''),
    ('/html-multiline-title', 'HTML title with newlines',     ''),
    ('/html-irc-title',       'Title with IRC injection',     ''),
    ('/html-unicode',         'Random unicode title/body',    ''),
    ('/html-meta-refresh',    'Meta refresh loop (0s)',        ''),

    # Terminal Abuses
    ('/ansi-chaos',      'ANSI escape code nightmare',        'binary'),

    # Network Abuses
    ('/net-gzip-bomb',   'Gzip bomb (~10KB to 10MB)',          'binary'),
    ('/net-event-flood', 'SSE event flood (infinite)',         'slow'),
    ('/net-mime-sniff',  'MIME type confusion',                ''),
    ('/net-cache-poison','Contradicting cache headers',        ''),
    ('/net-hsts-bomb',   'HSTS with insane max-age',           ''),

    # Miscellaneous
    ('/misc-ipv4',       'Shows your IPv4 address',            ''),
    ('/misc-ipv6',       'Shows your IPv6 address',            ''),
]


MAX_BODY = 8192  # max bytes to read from any response body

def truncate(data, maxlen=500):
    '''Truncate binary/long output for display'''
    if isinstance(data, bytes):
        if len(data) > maxlen:
            return repr(data[:maxlen]) + f'... ({len(data)} bytes total)'
        return repr(data)
    s = str(data)
    if len(s) > maxlen:
        return s[:maxlen] + f'... ({len(s)} chars total)'
    return s


def test_urllib(path, flags):
    '''Test with urllib.request'''
    url = f'{BASE}{path}'
    timeout = TIMEOUT if 'slow' in flags else 10
    try:
        r = urllib.request.urlopen(url, timeout=timeout)
        body = r.read(MAX_BODY)
        headers = dict(r.headers)
        print(f'  Status:  {r.status}')
        print(f'  Headers: {truncate(headers, 300)}')
        print(f'  Body:    {truncate(body)}')
    except Exception as e:
        print(f'  ERROR:   {type(e).__name__}: {e}')


def test_http_client(path, flags):
    '''Test with http.client'''
    timeout = TIMEOUT if 'slow' in flags else 10
    try:
        c = http.client.HTTPConnection(HOST, PORT, timeout=timeout)
        c.request('GET', path)
        r = c.getresponse()
        body = r.read(MAX_BODY)
        headers = list(r.getheaders())
        print(f'  Status:  {r.status} {r.reason}')
        print(f'  Headers: {truncate(headers, 300)}')
        print(f'  Body:    {truncate(body)}')
        c.close()
    except Exception as e:
        print(f'  ERROR:   {type(e).__name__}: {e}')


def test_requests(path, flags):
    '''Test with requests library'''
    if req_lib is None:
        print('  SKIPPED: requests not installed')
        return
    url = f'{BASE}{path}'
    timeout = TIMEOUT if 'slow' in flags else 10
    try:
        r = req_lib.get(url, timeout=timeout, allow_redirects=False, stream=True)
        body = r.raw.read(MAX_BODY)
        headers = dict(r.headers)
        print(f'  Status:  {r.status_code}')
        print(f'  Headers: {truncate(headers, 300)}')
        print(f'  Body:    {truncate(body)}')
        r.close()
    except Exception as e:
        print(f'  ERROR:   {type(e).__name__}: {e}')


async def test_aiohttp_single(path, flags):
    '''Test with aiohttp'''
    if aiohttp is None:
        print('  SKIPPED: aiohttp not installed')
        return
    url = f'{BASE}{path}'
    timeout_val = TIMEOUT if 'slow' in flags else 10
    to = aiohttp.ClientTimeout(total=timeout_val)
    try:
        async with aiohttp.ClientSession(timeout=to) as session:
            async with session.get(url, allow_redirects=False) as r:
                body = await r.content.read(MAX_BODY)
                headers = dict(r.headers)
                print(f'  Status:  {r.status}')
                print(f'  Headers: {truncate(headers, 300)}')
                print(f'  Body:    {truncate(body)}')
    except Exception as e:
        print(f'  ERROR:   {type(e).__name__}: {e}')


def test_aiohttp(path, flags):
    '''Wrapper to run aiohttp test synchronously'''
    if aiohttp is None:
        print('  SKIPPED: aiohttp not installed')
        return
    asyncio.run(test_aiohttp_single(path, flags))


def main():
    print(f'Testing badtests.py at {BASE}')
    print(f'Endpoints: {len(ENDPOINTS)}')
    print(f'Libraries: urllib.request, http.client' +
          (', requests' if req_lib else '') +
          (', aiohttp' if aiohttp else ''))
    print()

    for path, title, flags in ENDPOINTS:
        print(SEP)
        print(f'ENDPOINT: {path}')
        print(f'TITLE:    {title}')
        print()

        print('[urllib.request]')
        test_urllib(path, flags)
        print()

        print('[http.client]')
        test_http_client(path, flags)
        print()

        print('[requests]')
        test_requests(path, flags)
        print()

        print('[aiohttp]')
        test_aiohttp(path, flags)
        print()

    print(SEP)
    print('ALL TESTS COMPLETE')
    print(SEP)


if __name__ == '__main__':
    main()