12857916376

Committed 19 Jan 2025 10:59PM UTC coverage: 72.708% (+0.9%) from 71.843%

Build # 12857916376

Build Type

push

github

Committed by

savon-noir

Commit Message

fix: gh actions fix bllint issue

Run Details

1745 of 2400 relevant lines covered (72.71%)

2.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.13

/libnmap/process.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import platform
import shlex
import string
import subprocess
import warnings
from threading import Thread
from xml.dom import pulldom

try:
    import pwd
except ImportError:
    pass


__all__ = ["NmapProcess"]


class NmapTask(object):
    """
    NmapTask is a internal class used by process. Each time nmap
    starts a new task during the scan, a new class will be instantiated.
    Classes examples are: "Ping Scan", "NSE script", "DNS Resolve",..
    To each class an estimated time to complete is assigned and updated
    at least every second within the NmapProcess.
    A property NmapProcess.current_task points to the running task at
    time T and a dictionnary NmapProcess.tasks with "task name" as key
    is built during scan execution
    """

    def __init__(self, name, starttime=0, extrainfo=""):
        self.name = name
        self.etc = 0
        self.progress = 0
        self.percent = 0
        self.remaining = 0
        self.status = "started"
        self.starttime = starttime
        self.endtime = 0
        self.extrainfo = extrainfo
        self.updated = 0


class NmapProcess(Thread):
    """
    NmapProcess is a class which wraps around the nmap executable.

    Consequently, in order to run an NmapProcess, nmap should be installed
    on the host running the script. By default NmapProcess will produce
    the output of the nmap scan in the nmap XML format. This could be then
    parsed out via the NmapParser class from libnmap.parser module.
    """

    def __init__(
        self,
        targets="127.0.0.1",
        options="-sT",
        event_callback=None,
        safe_mode=True,
        fqp=None,
    ):
        """
        Constructor of NmapProcess class.

        :param targets: hosts to be scanned. Could be a string of hosts \
        separated with a coma or a python list of hosts/ip.
        :type targets: string or list

        :param options: list of nmap options to be applied to scan. \
        These options are all documented in nmap's man pages.

        :param event_callback: callable function which will be ran \
        each time nmap process outputs data. This function will receive \
        two parameters:

            1. the nmap process object
            2. the data produced by nmap process. See readme for examples.

        :param safe_mode: parameter to protect unsafe options like -oN, -oG, \
        -iL, -oA,...

        :param fqp: full qualified path, if None, nmap will be searched \
        in the PATH

        :return: NmapProcess object

        """
        Thread.__init__(self)
        unsafe_opts = set(
            [
                "-oG",
                "-oN",
                "-iL",
                "-oA",
                "-oS",
                "-oX",
                "--iflist",
                "--resume",
                "--stylesheet",
                "--datadir",
            ]
        )
        # more reliable than just using os.name() (cygwin)
        self.__is_windows = platform.system() == "Windows"
        if fqp:
            if os.path.isfile(fqp) and os.access(fqp, os.X_OK):
                self.__nmap_binary = fqp
            else:
                raise EnvironmentError(1, "wrong path or not executable", fqp)
        else:
            nmap_binary_name = "nmap"
            self.__nmap_binary = self._whereis(nmap_binary_name)
        self.__nmap_fixed_options = "-oX - -vvv --stats-every 1s"

        if self.__nmap_binary is None:
            em = "nmap is not installed or could not be found in system path"
            raise EnvironmentError(1, em)

        if isinstance(targets, str):
            self.__nmap_targets = targets.replace(" ", "").split(",")
        elif isinstance(targets, list):
            self.__nmap_targets = targets
        else:
            raise Exception(
                "Supplied target list should be either a string or a list"
            )

        for target in self.__nmap_targets:
            self.__validate_target(target)

        self._nmap_options = set(options.split())
        if safe_mode and not self._nmap_options.isdisjoint(unsafe_opts):
            raise Exception(
                "unsafe options activated while safe_mode is set True"
            )
        self.__nmap_dynamic_options = options
        self.__sudo_run = ""
        self.__nmap_command_line = self.get_command_line()

        if event_callback and callable(event_callback):
            self.__nmap_event_callback = event_callback
        else:
            self.__nmap_event_callback = None
        (
            self.DONE,
            self.READY,
            self.RUNNING,
            self.CANCELLED,
            self.FAILED,
        ) = range(5)
        self._run_init()

    def _run_init(self):
        self.__nmap_command_line = self.get_command_line()
        # API usable in callback function
        self.__nmap_proc = None
        self.__nmap_rc = 0
        self.__state = self.RUNNING
        self.__starttime = 0
        self.__endtime = 0
        self.__version = ""
        self.__elapsed = ""
        self.__summary = ""
        self.__stdout = ""
        self.__stderr = ""
        self.__current_task = ""
        self.__nmap_tasks = {}

    def _whereis(self, program):
        """
        Protected method enabling the object to find the full path of a binary
        from its PATH environment variable.

        :param program: name of a binary for which the full path needs to
        be discovered.

        :return: the full path to the binary.

        :todo: add a default path list in case PATH is empty.
        """
        split_char = ";" if self.__is_windows else ":"
        program = program + ".exe" if self.__is_windows else program
        for path in os.environ.get("PATH", "").split(split_char):
            _file_path = os.path.join(path, program)
            if os.path.exists(_file_path) and not os.path.isdir(_file_path):
                return os.path.join(path, program)
        return None

    def get_command_line(self):
        """
        Public method returning the reconstructed command line ran via the lib

        :return: the full nmap command line to run
        :rtype: string
        """
        return "{0} {1} {2} {3} {4}".format(
            self.__sudo_run,
            self.__nmap_binary,
            self.__nmap_fixed_options,
            self.__nmap_dynamic_options,
            " ".join(self.__nmap_targets),
        )

    def _ensure_user_exists(self, username=""):
        try:
            pwd.getpwnam(username).pw_uid
        except KeyError as eobj:
            _exmsg = (
                "Username {0} does not exists. Please supply"
                " a valid username: {1}".format(username, eobj)
            )
            raise EnvironmentError(_exmsg)

    def sudo_run(self, run_as="root"):
        """
        Public method enabling the library's user to run the scan with
        privileges via sudo. The sudo configuration should be set manually
        on the local system otherwise sudo will prompt for a password.
        This method alters the command line by prefixing the sudo command to
        nmap and will then call self.run()

        :param run_as: user name to which the lib needs to sudo to run the scan

        :return: return code from nmap execution
        """
        sudo_user = run_as.split().pop()
        self._ensure_user_exists(sudo_user)

        sudo_path = self._whereis("sudo")
        if sudo_path is None:
            raise EnvironmentError(
                2,
                "sudo is not installed or "
                "could not be found in system path: "
                "cannot run nmap with sudo",
            )

        self.__sudo_run = "{0} -u {1}".format(sudo_path, sudo_user)
        rc = self.run()
        self.__sudo_run = ""

        return rc

    def sudo_run_background(self, run_as="root"):
        """
        Public method enabling the library's user to run in background a
        nmap scan with privileges via sudo.
        The sudo configuration should be set manually on the local system
        otherwise sudo will prompt for a password.
        This method alters the command line by prefixing the sudo command to
        nmap and will then call self.run()

        :param run_as: user name to which the lib needs to sudo to run the scan

        :return: return code from nmap execution
        """
        sudo_user = run_as.split().pop()
        self._ensure_user_exists(sudo_user)

        sudo_path = self._whereis("sudo")
        if sudo_path is None:
            raise EnvironmentError(
                2,
                "sudo is not installed or "
                "could not be found in system path: "
                "cannot run nmap with sudo",
            )

        self.__sudo_run = "{0} -u {1}".format(sudo_path, sudo_user)
        super(NmapProcess, self).start()

    def run(self):
        """
        Public method which is usually called right after the constructor
        of NmapProcess. This method starts the nmap executable's subprocess.
        It will also bind a Process that will read from subprocess' stdout
        and stderr and push the lines read in a python queue for futher
        processing. This processing is waken-up each time data is pushed
        from the nmap binary into the stdout reading routine. Processing
        could be performed by a user-provided callback. The whole
        NmapProcess object could be accessible asynchroneously.

        return: return code from nmap execution
        """
        self._run_init()
        _tmp_cmdline = (
            self.__build_windows_cmdline()
            if self.__is_windows
            else shlex.split(self.__nmap_command_line)
        )
        try:
            self.__nmap_proc = subprocess.Popen(
                args=_tmp_cmdline,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                universal_newlines=True,
                bufsize=0,
            )
            self.__state = self.RUNNING
        except OSError as emsg:
            self.__state = self.FAILED
            raise EnvironmentError(
                1,
                "nmap is not installed or could "
                "not be found in system path: {0}".format(emsg),
            )

        while self.__nmap_proc.poll() is None:
            self.__process_nmap_proc_stdout()
        self.__process_nmap_proc_stdout()
        self.__stderr += self.__nmap_proc.stderr.read()

        self.__nmap_rc = self.__nmap_proc.poll()
        if self.rc is None:
            self.__state = self.CANCELLED
        elif self.rc == 0:
            self.__state = self.DONE
            if self.current_task:
                self.__nmap_tasks[self.current_task.name].progress = 100
        else:
            self.__state = self.FAILED
        # Call the callback one last time to signal the new state
        if self.__nmap_event_callback:
            self.__nmap_event_callback(self)
        return self.rc

    def run_background(self):
        """
        run nmap scan in background as a thread.
        For privileged scans, consider NmapProcess.sudo_run_background()
        """
        self.__state = self.RUNNING
        super(NmapProcess, self).start()

    def is_running(self):
        """
        Checks if nmap is still running.

        :return: True if nmap is still running
        """
        return self.state == self.RUNNING

    def has_terminated(self):
        """
        Checks if nmap has terminated. Could have failed or succeeded

        :return: True if nmap process is not running anymore.
        """
        return (
            self.state == self.DONE
            or self.state == self.FAILED
            or self.state == self.CANCELLED
        )

    def has_failed(self):
        """
        Checks if nmap has failed.

        :return: True if nmap process errored.
        """
        return self.state == self.FAILED

    def is_successful(self):
        """
        Checks if nmap terminated successfully.

        :return: True if nmap terminated successfully.
        """
        return self.state == self.DONE

    def stop(self):
        """
        Send KILL -15 to the nmap subprocess and gently ask the threads to
        stop.
        """
        self.__state = self.CANCELLED
        if self.__nmap_proc.poll() is None:
            self.__nmap_proc.kill()

    def __process_nmap_proc_stdout(self):
        for streamline in iter(self.__nmap_proc.stdout.readline, ""):
            self.__stdout += streamline
            evnt = self.__process_event(streamline)
            if self.__nmap_event_callback and evnt:
                self.__nmap_event_callback(self)

    def __process_event(self, eventdata):
        """
        Private method called while nmap process is running. It enables the
        library to handle specific data/events produced by nmap process.
        So far, the following events are supported:

        1. task progress: updates estimated time to completion and percentage
           done while scan is running. Could be used in combination with a
           callback function which could then handle this data while scan is
           running.
        2. nmap run: header of the scan. Usually displayed when nmap is started
        3. finished: when nmap scan ends.

        :return: True is event is known.

        :todo: handle parsing directly via NmapParser.parse()
        """
        rval = False
        try:
            edomdoc = pulldom.parseString(eventdata)
            for xlmnt, xmlnode in edomdoc:
                if xlmnt is not None and xlmnt == pulldom.START_ELEMENT:
                    if (
                        xmlnode.nodeName == "taskbegin"
                        and xmlnode.attributes.keys()
                    ):
                        xt = xmlnode.attributes
                        taskname = xt["task"].value
                        starttime = xt["time"].value
                        xinfo = ""
                        if "extrainfo" in xt.keys():
                            xinfo = xt["extrainfo"].value
                        newtask = NmapTask(taskname, starttime, xinfo)
                        self.__nmap_tasks[newtask.name] = newtask
                        self.__current_task = newtask.name
                        rval = True
                    elif (
                        xmlnode.nodeName == "taskend"
                        and xmlnode.attributes.keys()
                    ):
                        xt = xmlnode.attributes
                        tname = xt["task"].value
                        xinfo = ""
                        self.__nmap_tasks[tname].endtime = xt["time"].value
                        if "extrainfo" in xt.keys():
                            xinfo = xt["extrainfo"].value
                        self.__nmap_tasks[tname].extrainfo = xinfo
                        self.__nmap_tasks[tname].status = "ended"
                        rval = True
                    elif (
                        xmlnode.nodeName == "taskprogress"
                        and xmlnode.attributes.keys()
                    ):
                        xt = xmlnode.attributes
                        tname = xt["task"].value
                        percent = xt["percent"].value
                        etc = xt["etc"].value
                        remaining = xt["remaining"].value
                        updated = xt["time"].value
                        self.__nmap_tasks[tname].percent = percent
                        self.__nmap_tasks[tname].progress = percent
                        self.__nmap_tasks[tname].etc = etc
                        self.__nmap_tasks[tname].remaining = remaining
                        self.__nmap_tasks[tname].updated = updated
                        rval = True
                    elif (
                        xmlnode.nodeName == "nmaprun"
                        and xmlnode.attributes.keys()
                    ):
                        self.__starttime = xmlnode.attributes["start"].value
                        self.__version = xmlnode.attributes["version"].value
                        rval = True
                    elif (
                        xmlnode.nodeName == "finished"
                        and xmlnode.attributes.keys()
                    ):
                        self.__endtime = xmlnode.attributes["time"].value
                        self.__elapsed = xmlnode.attributes["elapsed"].value
                        self.__summary = xmlnode.attributes["summary"].value
                        rval = True
        except Exception:
            pass
        return rval

    def __build_windows_cmdline(self):
        cmdline = []
        cmdline.append(self.__nmap_binary)
        if self.__nmap_fixed_options:
            cmdline += self.__nmap_fixed_options.split()
        if self.__nmap_dynamic_options:
            cmdline += self.__nmap_dynamic_options.split()
        if self.__nmap_targets:
            cmdline += self.__nmap_targets  # already a list
        return cmdline

    @staticmethod
    def __validate_target(target):
        """
        Check if a provided target is valid. This function was created
        in order to address CVE-2022-30284

        See https://nmap.org/book/man-target-specification.html for all the
        ways targets can be specified

        This function verifies the following:

        - matches the user specified target against a list of allowed chars
        - check if dashes are used at the start or at the end of target

        FQDN can contain dashes anywhere except at the beginning or end
        This check also fixes/prevents CVE-2022-30284, which depends on being
        able to pass options such as --script as a target

        :return: False if target contains forbidden characters
        """
        allowed_characters = frozenset(
            string.ascii_letters + string.digits + "-.:/% "
        )
        if not set(target).issubset(allowed_characters):
            raise Exception(
                "Target '{}' contains invalid characters".format(target)
            )
        elif target.startswith("-") or target.endswith("-"):
            raise Exception(
                "Target '{}' cannot begin or end with a dash ('-')".format(
                    target
                )
            )
        return True

    @property
    def command(self):
        """
        return the constructed nmap command or empty string if not
        constructed yet.

        :return: string
        """
        return self.__nmap_command_line or ""

    @property
    def targets(self):
        """
        Provides the list of targets to scan

        :return: list of string
        """
        return self.__nmap_targets

    @property
    def options(self):
        """
        Provides the list of options for that scan

        :return: list of string (nmap options)
        """
        return self._nmap_options

    @property
    def state(self):
        """
        Accessor for nmap execution state. Possible states are:

        - self.READY
        - self.RUNNING
        - self.FAILED
        - self.CANCELLED
        - self.DONE

        :return: integer (from above documented enum)
        """
        return self.__state

    @property
    def starttime(self):
        """
        Accessor for time when scan started

        :return: string. Unix timestamp
        """
        return self.__starttime

    @property
    def endtime(self):
        """
        Accessor for time when scan ended

        :return: string. Unix timestamp
        """
        warnings.warn(
            "data collected from finished events are deprecated."
            "Use NmapParser.parse()",
            DeprecationWarning,
        )
        return self.__endtime

    @property
    def elapsed(self):
        """
        Accessor returning for how long the scan ran (in seconds)

        :return: string
        """
        warnings.warn(
            "data collected from finished events are deprecated."
            "Use NmapParser.parse()",
            DeprecationWarning,
        )
        return self.__elapsed

    @property
    def summary(self):
        """
        Accessor returning a short summary of the scan's results

        :return: string
        """
        warnings.warn(
            "data collected from finished events are deprecated."
            "Use NmapParser.parse()",
            DeprecationWarning,
        )
        return self.__summary

    @property
    def tasks(self):
        """
        Accessor returning for the list of tasks ran during nmap scan

        :return: dict of NmapTask object
        """
        return self.__nmap_tasks

    @property
    def version(self):
        """
        Accessor for nmap binary version number

        :return: version number of nmap binary
        :rtype: string
        """
        return self.__version

    @property
    def current_task(self):
        """
        Accessor for the current NmapTask beeing run

        :return: NmapTask or None if no task started yet
        """
        rval = None
        if len(self.__current_task):
            rval = self.tasks[self.__current_task]
        return rval

    @property
    def etc(self):
        """
        Accessor for estimated time to completion

        :return:  estimated time to completion
        """
        rval = 0
        if self.current_task:
            rval = self.current_task.etc
        return rval

    @property
    def progress(self):
        """
        Accessor for progress status in percentage

        :return: percentage of job processed.
        """
        rval = 0
        if self.current_task:
            rval = self.current_task.progress
        return rval

    @property
    def rc(self):
        """
        Accessor for nmap execution's return code

        :return: nmap execution's return code
        """
        return self.__nmap_rc

    @property
    def stdout(self):
        """
        Accessor for nmap standart output

        :return: output from nmap scan in XML
        :rtype: string
        """
        return self.__stdout

    @property
    def stderr(self):
        """
        Accessor for nmap standart error

        :return: output from nmap when errors occured.
        :rtype: string
        """
        return self.__stderr


def main():
    def mycallback(nmapscan=None):
        if nmapscan.is_running() and nmapscan.current_task:
            ntask = nmapscan.current_task
            print(
                "Task {0} ({1}): ETC: {2} DONE: {3}%".format(
                    ntask.name, ntask.status, ntask.etc, ntask.progress
                )
            )

    nm = NmapProcess(
        "scanme.nmap.org", options="-A", event_callback=mycallback
    )
    rc = nm.run()
    if rc == 0:
        print(
            "Scan started at {0} nmap version: {1}".format(
                nm.starttime, nm.version
            )
        )
        print("state: {0} (rc: {1})".format(nm.state, nm.rc))
        print("results size: {0}".format(len(nm.stdout)))
        print("Scan ended {0}: {1}".format(nm.endtime, nm.summary))
    else:
        print("state: {0} (rc: {1})".format(nm.state, nm.rc))
        print("Error: {stderr}".format(stderr=nm.stderr))
        print("Result: {0}".format(nm.stdout))


if __name__ == "__main__":
    main()

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	import os	4✔
5	import platform	4✔
6	import shlex	4✔
7	import string	4✔
8	import subprocess	4✔
9	import warnings	4✔
10	from threading import Thread	4✔
11	from xml.dom import pulldom	4✔
12
13	try:	4✔
14	import pwd	4✔
15	except ImportError:	×
16	pass	×
17
18
19	__all__ = ["NmapProcess"]	4✔
20
21
22	class NmapTask(object):	4✔
23	"""
24	NmapTask is a internal class used by process. Each time nmap
25	starts a new task during the scan, a new class will be instantiated.
26	Classes examples are: "Ping Scan", "NSE script", "DNS Resolve",..
27	To each class an estimated time to complete is assigned and updated
28	at least every second within the NmapProcess.
29	A property NmapProcess.current_task points to the running task at
30	time T and a dictionnary NmapProcess.tasks with "task name" as key
31	is built during scan execution
32	"""
33
34	def __init__(self, name, starttime=0, extrainfo=""):	4✔
35	self.name = name	4✔
36	self.etc = 0	4✔
37	self.progress = 0	4✔
38	self.percent = 0	4✔
39	self.remaining = 0	4✔
40	self.status = "started"	4✔
41	self.starttime = starttime	4✔
42	self.endtime = 0	4✔
43	self.extrainfo = extrainfo	4✔
44	self.updated = 0	4✔
45
46
47	class NmapProcess(Thread):	4✔
48	"""
49	NmapProcess is a class which wraps around the nmap executable.
50
51	Consequently, in order to run an NmapProcess, nmap should be installed
52	on the host running the script. By default NmapProcess will produce
53	the output of the nmap scan in the nmap XML format. This could be then
54	parsed out via the NmapParser class from libnmap.parser module.
55	"""
56
57	def __init__(	4✔
58	self,
59	targets="127.0.0.1",
60	options="-sT",
61	event_callback=None,
62	safe_mode=True,
63	fqp=None,
64	):
65	"""
66	Constructor of NmapProcess class.
67
68	:param targets: hosts to be scanned. Could be a string of hosts \
69	separated with a coma or a python list of hosts/ip.
70	:type targets: string or list
71
72	:param options: list of nmap options to be applied to scan. \
73	These options are all documented in nmap's man pages.
74
75	:param event_callback: callable function which will be ran \
76	each time nmap process outputs data. This function will receive \
77	two parameters:
78
79	1. the nmap process object
80	2. the data produced by nmap process. See readme for examples.
81
82	:param safe_mode: parameter to protect unsafe options like -oN, -oG, \
83	-iL, -oA,...
84
85	:param fqp: full qualified path, if None, nmap will be searched \
86	in the PATH
87
88	:return: NmapProcess object
89
90	"""
91	Thread.__init__(self)	4✔
92	unsafe_opts = set(	4✔
93	[
94	"-oG",
95	"-oN",
96	"-iL",
97	"-oA",
98	"-oS",
99	"-oX",
100	"--iflist",
101	"--resume",
102	"--stylesheet",
103	"--datadir",
104	]
105	)
106	# more reliable than just using os.name() (cygwin)
107	self.__is_windows = platform.system() == "Windows"	4✔
108	if fqp:	4✔
109	if os.path.isfile(fqp) and os.access(fqp, os.X_OK):	4✔
110	self.__nmap_binary = fqp	×
111	else:
112	raise EnvironmentError(1, "wrong path or not executable", fqp)	4✔
113	else:
114	nmap_binary_name = "nmap"	4✔
115	self.__nmap_binary = self._whereis(nmap_binary_name)	4✔
116	self.__nmap_fixed_options = "-oX - -vvv --stats-every 1s"	4✔
117
118	if self.__nmap_binary is None:	4✔
119	em = "nmap is not installed or could not be found in system path"	4✔
120	raise EnvironmentError(1, em)	4✔
121
122	if isinstance(targets, str):	4✔
123	self.__nmap_targets = targets.replace(" ", "").split(",")	4✔
124	elif isinstance(targets, list):	4✔
125	self.__nmap_targets = targets	4✔
126	else:
127	raise Exception(	4✔
128	"Supplied target list should be either a string or a list"
129	)
130
131	for target in self.__nmap_targets:	4✔
132	self.__validate_target(target)	4✔
133
134	self._nmap_options = set(options.split())	4✔
135	if safe_mode and not self._nmap_options.isdisjoint(unsafe_opts):	4✔
136	raise Exception(	4✔
137	"unsafe options activated while safe_mode is set True"
138	)
139	self.__nmap_dynamic_options = options	4✔
140	self.__sudo_run = ""	4✔
141	self.__nmap_command_line = self.get_command_line()	4✔
142
143	if event_callback and callable(event_callback):	4✔
144	self.__nmap_event_callback = event_callback	×
145	else:
146	self.__nmap_event_callback = None	4✔
147	(	4✔
148	self.DONE,
149	self.READY,
150	self.RUNNING,
151	self.CANCELLED,
152	self.FAILED,
153	) = range(5)
154	self._run_init()	4✔
155
156	def _run_init(self):	4✔
157	self.__nmap_command_line = self.get_command_line()	4✔
158	# API usable in callback function
159	self.__nmap_proc = None	4✔
160	self.__nmap_rc = 0	4✔
161	self.__state = self.RUNNING	4✔
162	self.__starttime = 0	4✔
163	self.__endtime = 0	4✔
164	self.__version = ""	4✔
165	self.__elapsed = ""	4✔
166	self.__summary = ""	4✔
167	self.__stdout = ""	4✔
168	self.__stderr = ""	4✔
169	self.__current_task = ""	4✔
170	self.__nmap_tasks = {}	4✔
171
172	def _whereis(self, program):	4✔
173	"""
174	Protected method enabling the object to find the full path of a binary
175	from its PATH environment variable.
176
177	:param program: name of a binary for which the full path needs to
178	be discovered.
179
180	:return: the full path to the binary.
181
182	:todo: add a default path list in case PATH is empty.
183	"""
184	split_char = ";" if self.__is_windows else ":"	4✔
185	program = program + ".exe" if self.__is_windows else program	4✔
186	for path in os.environ.get("PATH", "").split(split_char):	4✔
187	_file_path = os.path.join(path, program)	4✔
188	if os.path.exists(_file_path) and not os.path.isdir(_file_path):	4✔
189	return os.path.join(path, program)	4✔
190	return None	4✔
191
192	def get_command_line(self):	4✔
193	"""
194	Public method returning the reconstructed command line ran via the lib
195
196	:return: the full nmap command line to run
197	:rtype: string
198	"""
199	return "{0} {1} {2} {3} {4}".format(	4✔
200	self.__sudo_run,
201	self.__nmap_binary,
202	self.__nmap_fixed_options,
203	self.__nmap_dynamic_options,
204	" ".join(self.__nmap_targets),
205	)
206
207	def _ensure_user_exists(self, username=""):	4✔
208	try:	4✔
209	pwd.getpwnam(username).pw_uid	4✔
210	except KeyError as eobj:	4✔
211	_exmsg = (	4✔
212	"Username {0} does not exists. Please supply"
213	" a valid username: {1}".format(username, eobj)
214	)
215	raise EnvironmentError(_exmsg)	4✔
216
217	def sudo_run(self, run_as="root"):	4✔
218	"""
219	Public method enabling the library's user to run the scan with
220	privileges via sudo. The sudo configuration should be set manually
221	on the local system otherwise sudo will prompt for a password.
222	This method alters the command line by prefixing the sudo command to
223	nmap and will then call self.run()
224
225	:param run_as: user name to which the lib needs to sudo to run the scan
226
227	:return: return code from nmap execution
228	"""
229	sudo_user = run_as.split().pop()	4✔
230	self._ensure_user_exists(sudo_user)	4✔
231
232	sudo_path = self._whereis("sudo")	×
233	if sudo_path is None:	×
234	raise EnvironmentError(	×
235	2,
236	"sudo is not installed or "
237	"could not be found in system path: "
238	"cannot run nmap with sudo",
239	)
240
241	self.__sudo_run = "{0} -u {1}".format(sudo_path, sudo_user)	×
242	rc = self.run()	×
243	self.__sudo_run = ""	×
244
245	return rc	×
246
247	def sudo_run_background(self, run_as="root"):	4✔
248	"""
249	Public method enabling the library's user to run in background a
250	nmap scan with privileges via sudo.
251	The sudo configuration should be set manually on the local system
252	otherwise sudo will prompt for a password.
253	This method alters the command line by prefixing the sudo command to
254	nmap and will then call self.run()
255
256	:param run_as: user name to which the lib needs to sudo to run the scan
257
258	:return: return code from nmap execution
259	"""
260	sudo_user = run_as.split().pop()	4✔
261	self._ensure_user_exists(sudo_user)	4✔
262
263	sudo_path = self._whereis("sudo")	×
264	if sudo_path is None:	×
265	raise EnvironmentError(	×
266	2,
267	"sudo is not installed or "
268	"could not be found in system path: "
269	"cannot run nmap with sudo",
270	)
271
272	self.__sudo_run = "{0} -u {1}".format(sudo_path, sudo_user)	×
273	super(NmapProcess, self).start()	×
274
275	def run(self):	4✔
276	"""
277	Public method which is usually called right after the constructor
278	of NmapProcess. This method starts the nmap executable's subprocess.
279	It will also bind a Process that will read from subprocess' stdout
280	and stderr and push the lines read in a python queue for futher
281	processing. This processing is waken-up each time data is pushed
282	from the nmap binary into the stdout reading routine. Processing
283	could be performed by a user-provided callback. The whole
284	NmapProcess object could be accessible asynchroneously.
285
286	return: return code from nmap execution
287	"""
288	self._run_init()	4✔
289	_tmp_cmdline = (	4✔
290	self.__build_windows_cmdline()
291	if self.__is_windows
292	else shlex.split(self.__nmap_command_line)
293	)
294	try:	4✔
295	self.__nmap_proc = subprocess.Popen(	4✔
296	args=_tmp_cmdline,
297	stdout=subprocess.PIPE,
298	stderr=subprocess.PIPE,
299	universal_newlines=True,
300	bufsize=0,
301	)
302	self.__state = self.RUNNING	4✔
303	except OSError as emsg:	×
304	self.__state = self.FAILED	×
305	raise EnvironmentError(	×
306	1,
307	"nmap is not installed or could "
308	"not be found in system path: {0}".format(emsg),
309	)
310
311	while self.__nmap_proc.poll() is None:	4✔
312	self.__process_nmap_proc_stdout()	4✔
313	self.__process_nmap_proc_stdout()	4✔
314	self.__stderr += self.__nmap_proc.stderr.read()	4✔
315
316	self.__nmap_rc = self.__nmap_proc.poll()	4✔
317	if self.rc is None:	4✔
318	self.__state = self.CANCELLED	×
319	elif self.rc == 0:	4✔
320	self.__state = self.DONE	4✔
321	if self.current_task:	4✔
322	self.__nmap_tasks[self.current_task.name].progress = 100	4✔
323	else:
324	self.__state = self.FAILED	×
325	# Call the callback one last time to signal the new state
326	if self.__nmap_event_callback:	4✔
327	self.__nmap_event_callback(self)	×
328	return self.rc	4✔
329
330	def run_background(self):	4✔
331	"""
332	run nmap scan in background as a thread.
333	For privileged scans, consider NmapProcess.sudo_run_background()
334	"""
335	self.__state = self.RUNNING	×
336	super(NmapProcess, self).start()	×
337
338	def is_running(self):	4✔
339	"""
340	Checks if nmap is still running.
341
342	:return: True if nmap is still running
343	"""
344	return self.state == self.RUNNING	×
345
346	def has_terminated(self):	4✔
347	"""
348	Checks if nmap has terminated. Could have failed or succeeded
349
350	:return: True if nmap process is not running anymore.
351	"""
352	return (	×
353	self.state == self.DONE
354	or self.state == self.FAILED
355	or self.state == self.CANCELLED
356	)
357
358	def has_failed(self):	4✔
359	"""
360	Checks if nmap has failed.
361
362	:return: True if nmap process errored.
363	"""
364	return self.state == self.FAILED	×
365
366	def is_successful(self):	4✔
367	"""
368	Checks if nmap terminated successfully.
369
370	:return: True if nmap terminated successfully.
371	"""
372	return self.state == self.DONE	×
373
374	def stop(self):	4✔
375	"""
376	Send KILL -15 to the nmap subprocess and gently ask the threads to
377	stop.
378	"""
379	self.__state = self.CANCELLED	×
380	if self.__nmap_proc.poll() is None:	×
381	self.__nmap_proc.kill()	×
382
383	def __process_nmap_proc_stdout(self):	4✔
384	for streamline in iter(self.__nmap_proc.stdout.readline, ""):	4✔
385	self.__stdout += streamline	4✔
386	evnt = self.__process_event(streamline)	4✔
387	if self.__nmap_event_callback and evnt:	4✔
388	self.__nmap_event_callback(self)	×
389
390	def __process_event(self, eventdata):	4✔
391	"""
392	Private method called while nmap process is running. It enables the
393	library to handle specific data/events produced by nmap process.
394	So far, the following events are supported:
395
396	1. task progress: updates estimated time to completion and percentage
397	done while scan is running. Could be used in combination with a
398	callback function which could then handle this data while scan is
399	running.
400	2. nmap run: header of the scan. Usually displayed when nmap is started
401	3. finished: when nmap scan ends.
402
403	:return: True is event is known.
404
405	:todo: handle parsing directly via NmapParser.parse()
406	"""
407	rval = False	4✔
408	try:	4✔
409	edomdoc = pulldom.parseString(eventdata)	4✔
410	for xlmnt, xmlnode in edomdoc:	4✔
411	if xlmnt is not None and xlmnt == pulldom.START_ELEMENT:	4✔
412	if (	4✔
413	xmlnode.nodeName == "taskbegin"
414	and xmlnode.attributes.keys()
415	):
416	xt = xmlnode.attributes	4✔
417	taskname = xt["task"].value	4✔
418	starttime = xt["time"].value	4✔
419	xinfo = ""	4✔
420	if "extrainfo" in xt.keys():	4✔
421	xinfo = xt["extrainfo"].value	×
422	newtask = NmapTask(taskname, starttime, xinfo)	4✔
423	self.__nmap_tasks[newtask.name] = newtask	4✔
424	self.__current_task = newtask.name	4✔
425	rval = True	4✔
426	elif (	4✔
427	xmlnode.nodeName == "taskend"
428	and xmlnode.attributes.keys()
429	):
430	xt = xmlnode.attributes	4✔
431	tname = xt["task"].value	4✔
432	xinfo = ""	4✔
433	self.__nmap_tasks[tname].endtime = xt["time"].value	4✔
434	if "extrainfo" in xt.keys():	4✔
435	xinfo = xt["extrainfo"].value	4✔
436	self.__nmap_tasks[tname].extrainfo = xinfo	4✔
437	self.__nmap_tasks[tname].status = "ended"	4✔
438	rval = True	4✔
439	elif (	4✔
440	xmlnode.nodeName == "taskprogress"
441	and xmlnode.attributes.keys()
442	):
443	xt = xmlnode.attributes	×
444	tname = xt["task"].value	×
445	percent = xt["percent"].value	×
446	etc = xt["etc"].value	×
447	remaining = xt["remaining"].value	×
448	updated = xt["time"].value	×
449	self.__nmap_tasks[tname].percent = percent	×
450	self.__nmap_tasks[tname].progress = percent	×
451	self.__nmap_tasks[tname].etc = etc	×
452	self.__nmap_tasks[tname].remaining = remaining	×
453	self.__nmap_tasks[tname].updated = updated	×
454	rval = True	×
455	elif (	4✔
456	xmlnode.nodeName == "nmaprun"
457	and xmlnode.attributes.keys()
458	):
459	self.__starttime = xmlnode.attributes["start"].value	4✔
460	self.__version = xmlnode.attributes["version"].value	4✔
461	rval = True	4✔
462	elif (	4✔
463	xmlnode.nodeName == "finished"
464	and xmlnode.attributes.keys()
465	):
466	self.__endtime = xmlnode.attributes["time"].value	4✔
467	self.__elapsed = xmlnode.attributes["elapsed"].value	4✔
468	self.__summary = xmlnode.attributes["summary"].value	4✔
469	rval = True	4✔
470	except Exception:	4✔
471	pass	4✔
472	return rval	4✔
473
474	def __build_windows_cmdline(self):	4✔
475	cmdline = []	×
476	cmdline.append(self.__nmap_binary)	×
477	if self.__nmap_fixed_options:	×
478	cmdline += self.__nmap_fixed_options.split()	×
479	if self.__nmap_dynamic_options:	×
480	cmdline += self.__nmap_dynamic_options.split()	×
481	if self.__nmap_targets:	×
482	cmdline += self.__nmap_targets # already a list	×
483	return cmdline	×
484
485	@staticmethod	4✔
486	def __validate_target(target):	3✔
487	"""
488	Check if a provided target is valid. This function was created
489	in order to address CVE-2022-30284
490
491	See https://nmap.org/book/man-target-specification.html for all the
492	ways targets can be specified
493
494	This function verifies the following:
495
496	- matches the user specified target against a list of allowed chars
497	- check if dashes are used at the start or at the end of target
498
499	FQDN can contain dashes anywhere except at the beginning or end
500	This check also fixes/prevents CVE-2022-30284, which depends on being
501	able to pass options such as --script as a target
502
503	:return: False if target contains forbidden characters
504	"""
505	allowed_characters = frozenset(	4✔
506	string.ascii_letters + string.digits + "-.:/% "
507	)
508	if not set(target).issubset(allowed_characters):	4✔
509	raise Exception(	4✔
510	"Target '{}' contains invalid characters".format(target)
511	)
512	elif target.startswith("-") or target.endswith("-"):	4✔
513	raise Exception(	4✔
514	"Target '{}' cannot begin or end with a dash ('-')".format(
515	target
516	)
517	)
518	return True	4✔
519
520	@property	4✔
521	def command(self):	3✔
522	"""
523	return the constructed nmap command or empty string if not
524	constructed yet.
525
526	:return: string
527	"""
528	return self.__nmap_command_line or ""	×
529
530	@property	4✔
531	def targets(self):	3✔
532	"""
533	Provides the list of targets to scan
534
535	:return: list of string
536	"""
537	return self.__nmap_targets	4✔
538
539	@property	4✔
540	def options(self):	3✔
541	"""
542	Provides the list of options for that scan
543
544	:return: list of string (nmap options)
545	"""
546	return self._nmap_options	×
547
548	@property	4✔
549	def state(self):	3✔
550	"""
551	Accessor for nmap execution state. Possible states are:
552
553	- self.READY
554	- self.RUNNING
555	- self.FAILED
556	- self.CANCELLED
557	- self.DONE
558
559	:return: integer (from above documented enum)
560	"""
561	return self.__state	×
562
563	@property	4✔
564	def starttime(self):	3✔
565	"""
566	Accessor for time when scan started
567
568	:return: string. Unix timestamp
569	"""
570	return self.__starttime	×
571
572	@property	4✔
573	def endtime(self):	3✔
574	"""
575	Accessor for time when scan ended
576
577	:return: string. Unix timestamp
578	"""
579	warnings.warn(	×
580	"data collected from finished events are deprecated."
581	"Use NmapParser.parse()",
582	DeprecationWarning,
583	)
584	return self.__endtime	×
585
586	@property	4✔
587	def elapsed(self):	3✔
588	"""
589	Accessor returning for how long the scan ran (in seconds)
590
591	:return: string
592	"""
593	warnings.warn(	×
594	"data collected from finished events are deprecated."
595	"Use NmapParser.parse()",
596	DeprecationWarning,
597	)
598	return self.__elapsed	×
599
600	@property	4✔
601	def summary(self):	3✔
602	"""
603	Accessor returning a short summary of the scan's results
604
605	:return: string
606	"""
607	warnings.warn(	×
608	"data collected from finished events are deprecated."
609	"Use NmapParser.parse()",
610	DeprecationWarning,
611	)
612	return self.__summary	×
613
614	@property	4✔
615	def tasks(self):	3✔
616	"""
617	Accessor returning for the list of tasks ran during nmap scan
618
619	:return: dict of NmapTask object
620	"""
621	return self.__nmap_tasks	4✔
622
623	@property	4✔
624	def version(self):	3✔
625	"""
626	Accessor for nmap binary version number
627
628	:return: version number of nmap binary
629	:rtype: string
630	"""
631	return self.__version	×
632
633	@property	4✔
634	def current_task(self):	3✔
635	"""
636	Accessor for the current NmapTask beeing run
637
638	:return: NmapTask or None if no task started yet
639	"""
640	rval = None	4✔
641	if len(self.__current_task):	4✔
642	rval = self.tasks[self.__current_task]	4✔
643	return rval	4✔
644
645	@property	4✔
646	def etc(self):	3✔
647	"""
648	Accessor for estimated time to completion
649
650	:return: estimated time to completion
651	"""
652	rval = 0	×
653	if self.current_task:	×
654	rval = self.current_task.etc	×
655	return rval	×
656
657	@property	4✔
658	def progress(self):	3✔
659	"""
660	Accessor for progress status in percentage
661
662	:return: percentage of job processed.
663	"""
664	rval = 0	×
665	if self.current_task:	×
666	rval = self.current_task.progress	×
667	return rval	×
668
669	@property	4✔
670	def rc(self):	3✔
671	"""
672	Accessor for nmap execution's return code
673
674	:return: nmap execution's return code
675	"""
676	return self.__nmap_rc	4✔
677
678	@property	4✔
679	def stdout(self):	3✔
680	"""
681	Accessor for nmap standart output
682
683	:return: output from nmap scan in XML
684	:rtype: string
685	"""
686	return self.__stdout	4✔
687
688	@property	4✔
689	def stderr(self):	3✔
690	"""
691	Accessor for nmap standart error
692
693	:return: output from nmap when errors occured.
694	:rtype: string
695	"""
696	return self.__stderr	×
697
698
699	def main():	4✔
700	def mycallback(nmapscan=None):	×
701	if nmapscan.is_running() and nmapscan.current_task:	×
702	ntask = nmapscan.current_task	×
703	print(	×
704	"Task {0} ({1}): ETC: {2} DONE: {3}%".format(
705	ntask.name, ntask.status, ntask.etc, ntask.progress
706	)
707	)
708
709	nm = NmapProcess(	×
710	"scanme.nmap.org", options="-A", event_callback=mycallback
711	)
712	rc = nm.run()	×
713	if rc == 0:	×
714	print(	×
715	"Scan started at {0} nmap version: {1}".format(
716	nm.starttime, nm.version
717	)
718	)
719	print("state: {0} (rc: {1})".format(nm.state, nm.rc))	×
720	print("results size: {0}".format(len(nm.stdout)))	×
721	print("Scan ended {0}: {1}".format(nm.endtime, nm.summary))	×
722	else:
723	print("state: {0} (rc: {1})".format(nm.state, nm.rc))	×
724	print("Error: {stderr}".format(stderr=nm.stderr))	×
725	print("Result: {0}".format(nm.stdout))	×
726
727
728	if __name__ == "__main__":	4✔
729	main()	×

savon-noir / python-libnmap / 12857916376

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous