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()