Maximum runtime for a single target

.github/workflows/ci.yml

@@ -28,7 +28,7 @@ jobs:
         python3 dirsearch.py -w ./tests/static/wordlist.txt -l ./tests/static/targets.txt --subdirs /,admin/ --exclude-extensions conf -q -L -f -i 200 --user-agent a --log tmp_log.log
         python3 dirsearch.py -w ./tests/static/wordlist.txt --nmap-report ./tests/static/nmap.xml --max-rate 2 -H K:V --random-agent --overwrite-extensions --no-color
         python3 dirsearch.py -w ./tests/static/wordlist.txt --raw ./tests/static/raw.txt --prefixes . --suffixes ~ --skip-on-status 404 -m POST -d test=1 --crawl --min-response-size 9
-        echo https://self-signed.badssl.com | python3 dirsearch.py -w ./tests/static/wordlist.txt --stdin --max-time 9 --auth u:p --auth-type basic --scheme http
+        echo https://self-signed.badssl.com | python3 dirsearch.py -w ./tests/static/wordlist.txt --stdin --max-time 8 --auth u:p --auth-type basic --scheme http --target-max-time 9
 
     - name: Unit Test
       run: python3 testing.py

CHANGELOG.md

@@ -6,8 +6,9 @@
 - Support variables in file path and SQL table name for saving results
 - Support non-default network interface
 - Load targets from a Nmap XML report
-- Added --async option to enable asynchronous mode (use coroutines instead of threads)
+- Added option to enable asynchronous mode (use coroutines instead of threads)
 - Added option to disable CLI output entirely
+- Maximum runtime per target
 - Added OpenAI and Ollama REST API endpoints to the dictionary
 
 ## [0.4.3] - October 2nd, 2022

config.ini

@@ -13,6 +13,7 @@ max-recursion-depth = 0
 exclude-subdirs = %%ff/,.;/,..;/,;/,./,../,%%2e/,%%2e%%2e/
 random-user-agents = False
 max-time = 0
+target-max-time = 0
 exit-on-error = False
 skip-on-status = 429
 #subdirs = /,api/

lib/connection/requester.py

@@ -240,7 +240,7 @@ def request(self, path: str, proxy: str | None = None) -> Response:
                         err_msg = f"Error with the proxy: {proxy}"
                     else:
                         err_msg = "Error with the system proxy"
-                    # Prevent from re-using it in the future
+                    # Prevent from reusing it in the future
                     if proxy in options["proxies"] and len(options["proxies"]) > 1:
                         options["proxies"].remove(proxy)
                 elif "InvalidURL" in str(e):

lib/controller/controller.py

@@ -259,6 +259,8 @@ def run(self) -> None:
                 interface.error("Failed to delete old session file, remove it to free some space")
 
     def start(self) -> None:
+        start_time = time.time()
+
         while self.directories:
             try:
                 gc.collect()
@@ -276,10 +278,10 @@ def start(self) -> None:
                     # use a future to get exceptions from handle_pause
                     # https://stackoverflow.com/a/64230941
                     self.pause_future = self.loop.create_future()
-                    self.loop.run_until_complete(self._start_coroutines())
+                    self.loop.run_until_complete(self.start_coroutines(start_time))
                 else:
                     self.fuzzer.start()
-                    self.process()
+                    self.process(start_time)
 
             except (KeyboardInterrupt, asyncio.CancelledError):
                 pass
@@ -291,26 +293,52 @@ def start(self) -> None:
                 self.jobs_processed += 1
                 self.old_session = False
 
-    async def _start_coroutines(self) -> None:
+    async def start_coroutines(self, start_time: float) -> None:
         task = self.loop.create_task(self.fuzzer.start())
+        timeout = min(
+            t for t in [
+                options["max_time"] - (time.time() - self.start_time),
+                options["target_max_time"] - (time.time() - start_time),
+            ] if t > 0
+        ) if options["max_time"] or options["target_max_time"] else None
 
         try:
             await asyncio.wait_for(
                 asyncio.wait(
                     [self.pause_future, task],
                     return_when=asyncio.FIRST_COMPLETED,
                 ),
-                timeout=options["max_time"] if options["max_time"] > 0 else None,
+                timeout=timeout,
             )
         except asyncio.TimeoutError:
-            raise SkipTargetInterrupt("Runtime exceeded the maximum set by the user")
+            if time.time() - self.start_time > options["max_time"] > 0:
+                raise QuitInterrupt("Runtime exceeded the maximum set by the user")
+
+            raise SkipTargetInterrupt("Runtime for target exceeded the maximum set by the user")
 
         if self.pause_future.done():
             task.cancel()
             await self.pause_future  # propagate the exception, if raised
 
         await task  # propagate the exception, if raised
 
+    def process(self, start_time: float) -> None:
+        while True:
+            while not self.fuzzer.is_finished():
+                now = time.time()
+                if now - self.start_time > options["max_time"] > 0:
+                    raise QuitInterrupt(
+                        "Runtime exceeded the maximum set by the user"
+                    )
+                if now - start_time > options["target_max_time"] > 0:
+                    raise SkipTargetInterrupt(
+                        "Runtime for target exceeded the maximum set by the user"
+                    )
+
+                time.sleep(0.5)
+
+            break
+
     def set_target(self, url: str) -> None:
         # If no scheme specified, unset it first
         if "://" not in url:
@@ -501,20 +529,6 @@ def handle_pause(self) -> None:
                 else:
                     raise skipexc
 
-    def is_timed_out(self) -> bool:
-        return time.time() - self.start_time > options["max_time"] > 0
-
-    def process(self) -> None:
-        while True:
-            while not self.fuzzer.is_finished():
-                if self.is_timed_out():
-                    raise SkipTargetInterrupt(
-                        "Runtime exceeded the maximum set by the user"
-                    )
-                time.sleep(0.5)
-
-            break
-
     def add_directory(self, path: str) -> None:
         """Add directory to the recursion queue"""
 

lib/core/data.py

@@ -59,7 +59,8 @@
     "skip_on_status": set(),
     "minimum_response_size": 0,
     "maximum_response_size": 0,
-    "maxtime": 0,
+    "max_time": 0,
+    "target_max_time": 0,
     "http_method": "GET",
     "data": None,
     "data_file": None,

lib/core/options.py

@@ -311,6 +311,9 @@ def merge_config(opt: Values) -> Values:
         "general", "skip-on-status", ""
     )
     opt.max_time = opt.max_time or config.safe_getint("general", "max-time")
+    opt.target_max_time = opt.target_max_time or config.safe_getint(
+        "general", "target-max-time"
+    )
     opt.exit_on_error = opt.exit_on_error or config.safe_getboolean(
         "general", "exit-on-error"
     )

lib/parse/cmdline.py

@@ -307,6 +307,14 @@ def parse_arguments() -> Values:
         metavar="SECONDS",
         help="Maximum runtime for the scan",
     )
+    general.add_option(
+        "--target-max-time",
+        action="store",
+        type="int",
+        dest="target_max_time",
+        metavar="SECONDS",
+        help="Maximum runtime for a target",
+    )
     general.add_option(
         "--exit-on-error",
         action="store_true",