Added support for asynchronous calls in class NmapAsync to those who want to perform asynchronous scanning.

Author: nmmapper

nmap3/nmap3.py

@@ -19,25 +19,18 @@
 #
 #
 
-import csv
-import io
 import os
-import re
 import shlex
 import subprocess
 import sys
 import simplejson as json
 import argparse
+import asyncio
 from xml.etree import ElementTree as ET
 from xml.etree.ElementTree import ParseError
-from nmapparser import NmapCommandParser
-from utils import get_nmap_path, user_is_root
-from exceptions import NmapNotInstalledError, NmapXMLParserError, NmapExecutionError
-
-#from nmap3.nmapparser import NmapCommandParser
-#from nmap3.utils import get_nmap_path, user_is_root
-#from nmap3.exceptions import NmapNotInstalledError, NmapXMLParserError, NmapExecutionError
-
+from nmap3.nmapparser import NmapCommandParser
+from nmap3.utils import get_nmap_path, user_is_root
+from nmap3.exceptions import NmapNotInstalledError, NmapXMLParserError, NmapExecutionError
 import xml
 
 __author__ = 'Wangolo Joel ([email protected])'
@@ -225,6 +218,7 @@ def nmap_os_detection(self, target, arg="-O", args=None):  # requires root
         nmap -oX - nmmapper.com -O
         NOTE: Requires root
         """
+        print("Performing some scannins")
         xml_root = self.scan_command(target=target, arg=arg, args=args)
         results = self.parser.os_identifier_parser(xml_root)
         return results
@@ -272,7 +266,7 @@ def run_command(self, cmd, timeout=None):
                 return output.decode('utf8').strip()
         else:
             raise NmapNotInstalledError()
-
+            
     def get_xml_et(self, command_output):
         """
         @ return xml ET
@@ -283,7 +277,6 @@ def get_xml_et(self, command_output):
         except ParseError:
             raise NmapXMLParserError()
 
-
 class NmapScanTechniques(Nmap):
     """
     Extends Nmap to include nmap commands
@@ -339,27 +332,26 @@ def tpl(i):
 
         return xml_root
 
+    @user_is_root
     def nmap_fin_scan(self, target, args=None):
         """
         Perform scan using nmap's fin scan
 
         @cmd nmap -sF 192.168.178.1
 
         """
-        self.require_root()
-
         xml_root = self.scan_command(self.fin_scan, target=target, args=args)
         results = self.parser.filter_top_ports(xml_root)
         return results
-
+    
+    @user_is_root
     def nmap_syn_scan(self, target, args=None):
         """
         Perform syn scan on this given
         target
 
         @cmd nmap -sS 192.168.178.1
         """
-        self.require_root()
         xml_root = self.scan_command(self.sync_scan, target=target, args=args)
         results = self.parser.filter_top_ports(xml_root)
         return results
@@ -375,14 +367,14 @@ def nmap_tcp_scan(self, target, args=None):
         xml_root = self.scan_command(self.tcp_connt, target=target, args=args)
         results = self.parser.filter_top_ports(xml_root)
         return results
-
+    
+    @user_is_root
     def nmap_udp_scan(self, target, args=None):
         """
         Scan target using the nmap tcp connect
 
         @cmd nmap -sU 192.168.178.1
         """
-        self.require_root()
 
         if (args):
             assert (isinstance(args, str)), "Expected string got {0} instead".format(type(args))
@@ -410,7 +402,6 @@ def nmap_idle_scan(self, target, args=None):
         results = self.parser.filter_top_ports(xml_root)
         return results
 
-
 class NmapHostDiscovery(Nmap):
     """
     This object will perform host discovery
@@ -501,12 +492,95 @@ def nmap_disable_dns(self, target, args=None):
         results = self.parser.filter_top_ports(xml_root)
         return results
 
+class NmapAsync(Nmap):
+    def __init__(self, path=None):
+        super(NmapAsync, self).__init__(path=path)
+        self.stdout = asyncio.subprocess.PIPE
+        self.stderr = asyncio.subprocess.PIPE
+        
+    async def run_command(self, cmd, timeout=None):        
+        if (os.path.exists(self.nmaptool)):            
+            process = await asyncio.create_subprocess_shell(cmd,stdout=self.stdout,stderr=self.stderr)
+            
+            try:
+                data, stderr = await process.communicate()
+            except Exception as e:
+                raise (e)
+            else:
+                if 0 != process.returncode:
+                    raise NmapExecutionError('Error during command: "' + ' '.join(cmd) + '"\n\n' + errs.decode('utf8'))
+
+                # Response is bytes so decode the output and return
+                return data.decode('utf8').strip()
+        else:
+            raise NmapNotInstalledError()
+    
+    async def scan_command(self, target, arg, args=None, timeout=None):
+        self.target == target
+
+        command_args = "{target}  {default}".format(target=target, default=arg)
+        scancommand = self.default_command() + command_args
+        if (args):
+            scancommand += " {0}".format(args)
+
+        output = await self.run_command(scancommand, timeout=timeout)
+        xml_root = self.get_xml_et(output)
 
+        return xml_root
+        
+    async def scan_top_ports(self, target, default=10, args=None, timeout=None):
+        top_port_args = " {target} --top-ports {default}".format(target=target, default=default)
+        command = self.default_command() + top_port_args
+        if (args):
+            command += " {0}".format(args)
+
+        output = await self.run_command(command, timeout=timeout)
+        if not output:
+            raise ValueError("Unable to perform requested command")
+
+        self.top_ports = self.parser.filter_top_ports(self.get_xml_et(output))
+        return self.top_ports
+    
+    async def nmap_dns_brute_script(self, target, dns_brute="--script dns-brute.nse", timeout=None):
+        self.target = target
+
+        dns_brute_args = "{target}  {default}".format(target=target, default=dns_brute)
+        dns_brute_command = self.default_command() + dns_brute_args
+
+        # Run the command and get the output
+        output = await self.run_command(dns_brute_command, timeout=timeout)
+        subdomains = self.parser.filter_subdomains(self.get_xml_et(output))
+        return subdomains
+
+    async def nmap_version_detection(self, target, arg="-sV", args=None, timeout=None):
+        xml_root = await self.scan_command(target=target, arg=arg, timeout=timeout)
+        services = self.parser.filter_top_ports(xml_root)
+        return services
+
+    async def nmap_stealth_scan(self, target, arg="-Pn -sZ", args=None):
+        xml_root = await self.scan_command(target=target, arg=arg, args=args)
+        self.top_ports = self.parser.filter_top_ports(xml_root)
+        return self.top_ports
+
+    async def nmap_os_detection(self, target, arg="-O", args=None):  # requires root
+        xml_root = await self.scan_command(target=target, arg=arg, args=args)
+        results = self.parser.os_identifier_parser(xml_root)
+        return results
+
+    async def nmap_subnet_scan(self, target, arg="-p-", args=None):  # requires root
+        xml_root = await self.scan_command(target=target, arg=arg, args=args)
+        results = self.parser.filter_top_ports(xml_root)
+        return results
+
+    async def nmap_list_scan(self, target, arg="-sL", args=None):  # requires root
+        xml_root = await self.scan_command(target=target, arg=arg, args=args)
+        results = self.parser.filter_top_ports(xml_root)
+        return results
+        
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(prog="Python3 nmap")
     parser.add_argument('-d', '--d', help='Help', required=True)
     args = parser.parse_args()
 
-    nmap = Nmap()
-    result = nmap.scan_top_ports(target='127.0.0.1')
-    print(json.dumps(result, indent=4, sort_keys=True))
+    nmap = NmapAsync()
+    asyncio.run(nmap.nmap_version_detection(target='127.0.0.1'))

nmap3/utils.py

@@ -24,12 +24,12 @@
 import re
 import os
 import ctypes
+import functools
 
 __author__ = 'Wangolo Joel ([email protected])'
 __version__ = '1.4.9'
 __last_modification__ = '2019/12/11'
 
-
 def get_nmap_path():
     """
     Returns the location path where nmap is installed
@@ -54,7 +54,6 @@ def get_nmap_path():
         else:
             return output.decode('utf8').strip()
 
-
 def get_nmap_version():
     nmap = get_nmap_path()
     cmd = nmap + " --version"
@@ -82,3 +81,17 @@ def wrapper(*args, **kwargs):
         else:
             return {"error":True, "msg":"You must be root/administrator to continue!"}
     return wrapper 
+
+def nmap_is_installed_async():
+    def wrapper(func):
+        @functools.wraps(func)
+        async def wrapped(*args, **kwargs):
+            nmap_path = get_nmap_path()
+                
+            if(os.path.exists(nmap_path)):
+                return await func(*args, **kwargs)
+            else:
+                print({"error":True, "msg":"Nmap has not been install on this system yet!"})
+                return {"error":True, "msg":"Nmap has not been install on this system yet!"}
+        return wrapped
+    return  wrapper