Feature/progress callback (thread safe.)
This time the implementation is cleaner, doesn't have the terminal cleanup code, and is thread safe.
Summary: Added support for real-time progress reporting in Nmap scans via a progress_callback parameter.
Motivation:
- Provide real-time feedback during long-running scans.
- Preserve compatibility with XML parsing.
Usage Example:
def my_progress_callback(progress: str):
print(progress)
nmap = Nmap()
nmap.scan_top_ports(progress_callback=my_progress_callback)
Changes:
- Introduced progress_callback parameter to run_command, scan_command and nmap scan functions.
- Callback receives a string representing information about the progress of the scan.
- Captures progress from Nmap's stdout using the "--stats-every 1s" flag.
- Switched XML output handling to temporary files created with tempfile module since stdout is used for progress information.
- XML output from nmap is still captured from stdout if the progress_callback function is not provided just as before.
Testing: Code was tested with file: test.py
Adding multi-threading should be fine and not cause any issues. In fact, the old code output, errs = sub_proc.communicate(timeout=timeout) uses subprocess communicate function and communicate also spins up two threads to read stdout and stderr concurrently. This is a necessity and is the only way to read stdout and stderr without causing process blocking.
here is a snip-it from the subprocess module code for reference:
def _communicate(self, input, endtime, orig_timeout):
# Start reader threads feeding into a list hanging off of this
# object, unless they've already been started.
if self.stdout and not hasattr(self, "_stdout_buff"):
self._stdout_buff = []
self.stdout_thread = \
threading.Thread(target=self._readerthread,
args=(self.stdout, self._stdout_buff))
self.stdout_thread.daemon = True
self.stdout_thread.start()
if self.stderr and not hasattr(self, "_stderr_buff"):
self._stderr_buff = []
self.stderr_thread = \
threading.Thread(target=self._readerthread,
args=(self.stderr, self._stderr_buff))
self.stderr_thread.daemon = True
self.stderr_thread.start()