Skip to content

Conversation

westnt
Copy link

@westnt westnt commented Aug 28, 2025

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 added 26 commits August 24, 2025 21:31
…eference to self.xml_path. implemented io from xml_path file
…out is used for status and xml is written to disk.
@westnt
Copy link
Author

westnt commented Aug 28, 2025

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant