python3-nmap icon indicating copy to clipboard operation
python3-nmap copied to clipboard

Feature/progress callback (thread safe.)

Open westnt opened this issue 7 months ago • 1 comments

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

westnt avatar Aug 28 '25 20:08 westnt

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

westnt avatar Aug 28 '25 20:08 westnt