etc.git


etc

~ scripts and random things
git clone https://github.com/djnnvx/etc.git
log | files | refs | archive

twft (6054B)

      1 #!/bin/env python3
      2 
      3 import argparse
      4 import os
      5 import json
      6 from typing import Dict, List, Optional, Iterator
      7 import concurrent.futures
      8 
      9 import urllib.request
     10 from urllib.parse import urlencode
     11 
     12 FUZZ_KEY = 'FUZZME'
     13 
     14 class Request:
     15     method: str
     16     route: str
     17     http_version: str  # NOTE(djnn): parsed but unsupported rn (idc tbh)
     18 
     19     body: Optional[Dict[str, str]]
     20     request_params: Optional[Dict[str, str]]
     21     headers: Dict[str, str]
     22 
     23 
     24 
     25 def read_file_batches(file_path: str, batch_size: int = 10) -> Iterator[List[str]]:
     26     """Reads a text file in batches of lines."""
     27     with open(file_path, 'r') as file:
     28         lines = []
     29 
     30         for _ in range(batch_size):
     31             try:
     32                 line = next(file)
     33                 lines.append(line)
     34 
     35             # End of file reached, return the last batch if it's not full
     36             except StopIteration:
     37                 if lines:
     38                     yield lines
     39 
     40                 break
     41         else:
     42             yield lines
     43 
     44 
     45 def parse_request_file(filepath: str) -> Optional[Request]:
     46     """parse the request file"""
     47 
     48     try:
     49         result = Request()
     50 
     51         with open(filepath, 'r') as file:
     52             # we could read line by line here, but we want to keep the body unified
     53             contents = file.read()
     54             headers_and_body = contents.split('\n\n')
     55 
     56             # assigning body if there is one
     57             assert len(headers_and_body) < 3
     58             result.body = json.loads(headers_and_body[1]) if len(headers_and_body) == 2 else None
     59 
     60             # splitting headers into lines
     61             headers = headers_and_body[0].split('\n')
     62             assert len(headers) != 0
     63 
     64             # parsing the first line (eg GET /?test=123 HTTP/1.1)
     65             first_line_split = headers[0].split(' ')
     66             assert len(first_line_split) == 3
     67 
     68             result.method = first_line_split[0]
     69             result.route = first_line_split[1]
     70             result.http_version = first_line_split[2]
     71 
     72             # split route after ?, if any items -> put them into the request params
     73             split_route = result.route.split('?')
     74             assert len(split_route) < 3
     75 
     76             # we have query params!
     77             if len(split_route) == 2:
     78                 result.route = split_route[0]
     79                 result.request_params = {}
     80 
     81                 for param in split_route[1].split('&'):
     82                     keyval = param.split('=')
     83                     assert len(keyval) == 2
     84                     result.request_params[keyval[0]] = keyval[1]
     85 
     86 
     87             # parsing all headers :)
     88             for hdr_idx in range(1, len(headers)):
     89                 header = headers[hdr_idx]
     90                 keyval = header.split(': ')
     91 
     92                 assert len(keyval) == 2
     93                 result.headers[keyval[0]] = keyval[1]
     94 
     95             return result
     96 
     97     except (FileNotFoundError, PermissionError, AssertionError):
     98         return None
     99 
    100 
    101 def run_request(replace_with: str, filter_by_size: int, use_https: bool, request: Request) -> None:
    102     """Function to be executed by each worker."""
    103 
    104     local_route = request.route.replace(FUZZ_KEY, replace_with)
    105     local_headers = {
    106         key.replace(FUZZ_KEY, replace_with): val.replace(FUZZ_KEY, replace_with)
    107              for key, val in request.headers.items()
    108     }
    109     local_params = {
    110      key.replace(FUZZ_KEY, replace_with): val.replace(FUZZ_KEY, replace_with)
    111              for key, val in request.request_params.items()
    112     } if request.request_params else {}
    113 
    114 
    115     scheme = 'https' if use_https else 'https'
    116     full_uri = f'{scheme}://{local_headers["Host"]}{local_route}'
    117     if local_params != {}:
    118         full_uri = f'{full_uri}?{urlencode(local_params)}'
    119 
    120     req = urllib.request.Request(full_uri, method=request.method)
    121     for key, val in local_headers.items():
    122         req.add_header(key, val)
    123 
    124     with urllib.request.urlopen(req) as f:
    125         contents = f.read().decode('utf-8')
    126         if len(contents) != filter_by_size:
    127 
    128             print(f'[+] Found interesting request: {full_uri}')
    129             print('----====[ REQUEST HEADERS ]====----')
    130             for key, val in local_headers.items():
    131                 print(f'\t{key}: {val}')
    132             print(f'\n\nStatus: {f.status} (len: {len(contents)})\n')
    133             print('----====[ RESPONSE ]====----')
    134             print(f'{contents}\n----------------------------\n')
    135 
    136 
    137 def main() -> None:
    138 
    139     # set up argument parser and command-line options
    140     parser = argparse.ArgumentParser(description="tiny web fuzzing tool. Copy your request from burp, replace the items you want to fuzz with FUZZME and that's it!")
    141     parser.add_argument("-s", "--https", action="store_true", help="force https usage", default=False)
    142     parser.add_argument("-w", "--wordlist", help="wordlist filepath", default="common.txt")
    143     parser.add_argument("-fs", "--filter-size", help="filter by size", type=int, default=0)
    144     parser.add_argument("request", help="filepath containing the HTTP request definition")
    145 
    146     parser.parse_args()
    147 
    148     if not parser.request or not parser.wordlist:   # type: ignore
    149         print("[!] error: missing request or wordlist parameter")
    150         print("Please run --help option!")
    151         os._exit(1)
    152 
    153     request_items = parse_request_file(parser.request)  # type: ignore
    154     if not request_items:
    155         print('[!] error: could not read from file.')
    156         os._exit(1)
    157 
    158     filter_size = int(parser.filter_size)  # type: ignore
    159     use_https = bool(parser.https if parser.https else False)  # type: ignore
    160     batch_iterator = read_file_batches(parser.wordlist)  # type: ignore
    161     with concurrent.futures.ThreadPoolExecutor() as executor:
    162         while True:
    163             try:
    164                 batch = next(batch_iterator)
    165                 if not batch:
    166                     break
    167 
    168                 futures = [executor.submit(run_request, line, filter_size, use_https, request_items) for line in batch]
    169                 concurrent.futures.wait(futures)
    170 
    171             except StopIteration:
    172                 break
    173 
    174 
    175 if __name__ == '__main__':
    176     main()