Merge pull request #1 from MrMatch246/claude/nifty-brattain-ce0887

add --import-nmap flag to skip port scanning from existing XML

autorecon/config.py

@@ -77,5 +77,6 @@ config = {
 	'max_plugin_target_instances': None,
 	'max_plugin_global_instances': None,
 	'accessible': False,
-	'verbose': 0
+	'verbose': 0,
+	'imported_nmap_services': None
 }

autorecon/main.py

@@ -1,6 +1,7 @@
 #!/usr/bin/python3
 
 import argparse, asyncio, importlib.util, inspect, ipaddress, math, os, re, select, shutil, signal, socket, sys, termios, time, traceback, tty
+import xml.etree.ElementTree as ET
 from datetime import datetime
 
 try:
@@ -38,6 +39,67 @@ def needs_update(src, dst):
 		return True
 	return latest_mtime(src) > latest_mtime(dst)
 
+def parse_nmap_xml(xml_file):
+	"""Parse an nmap XML file and return {ip: [(protocol, port, service_name, secure), ...]}."""
+	result = {}
+	try:
+		tree = ET.parse(xml_file)
+		root = tree.getroot()
+	except ET.ParseError as exc:
+		fail('Error parsing nmap XML file "' + xml_file + '": ' + str(exc))
+		return result
+
+	for host in root.findall('host'):
+		status = host.find('status')
+		if status is not None and status.get('state') != 'up':
+			continue
+
+		address = None
+		for addr_elem in host.findall('address'):
+			addrtype = addr_elem.get('addrtype', '')
+			if addrtype in ('ipv4', 'ipv6'):
+				address = addr_elem.get('addr')
+				break
+
+		if address is None:
+			hostnames = host.find('hostnames')
+			if hostnames is not None:
+				for hn in hostnames.findall('hostname'):
+					address = hn.get('name')
+					break
+
+		if address is None:
+			continue
+
+		services = []
+		ports_elem = host.find('ports')
+		if ports_elem is None:
+			continue
+
+		for port_elem in ports_elem.findall('port'):
+			state = port_elem.find('state')
+			if state is None or state.get('state') != 'open':
+				continue
+
+			protocol = port_elem.get('protocol', 'tcp').lower()
+			port_num = int(port_elem.get('portid', 0))
+
+			svc_elem = port_elem.find('service')
+			if svc_elem is not None:
+				name = svc_elem.get('name', 'unknown')
+				tunnel = svc_elem.get('tunnel', '')
+				secure = tunnel in ('ssl', 'tls') or 'ssl' in name or 'tls' in name
+			else:
+				name = 'unknown'
+				secure = False
+
+			services.append((protocol, port_num, name, secure))
+
+		if services:
+			result[address] = services
+
+	return result
+
 # ----------------------- CONFIG DIR -----------------------
 if not os.path.exists(config['config_dir']):
 	shutil.rmtree(config['config_dir'], ignore_errors=True, onerror=None)
@@ -493,7 +555,7 @@ async def service_scan(plugin, service):
 async def generate_report(plugin, targets):
 	semaphore = autorecon.service_scan_semaphore
 
-	if not config['force_services']:
+	if not config['force_services'] and config.get('imported_nmap_services') is None:
 		semaphore = await get_semaphore(autorecon)
 
 	async with semaphore:
@@ -571,6 +633,49 @@ async def scan_target(target):
 			heartbeat.cancel()
 			autorecon.errors = True
 			return
+	elif config.get('imported_nmap_services') is not None:
+		imported_services_data = config['imported_nmap_services'].get(target.ip) or config['imported_nmap_services'].get(target.address)
+		if imported_services_data:
+			for (protocol, port_num, name, secure) in imported_services_data:
+				if config['proxychains'] and protocol == 'udp':
+					warn('Service ' + protocol + '/' + str(port_num) + '/' + name + ' uses UDP and --proxychains is enabled. Skipping.')
+					continue
+				svc = Service(protocol, port_num, name, secure)
+				svc.target = target
+				services.append(svc)
+			if services:
+				pending.add(asyncio.create_task(asyncio.sleep(0)))
+			else:
+				warn('No usable services for ' + target.address + ' in imported nmap data (all skipped).')
+				heartbeat.cancel()
+				return
+		else:
+			warn('Target ' + target.address + ' (' + target.ip + ') was not found in the imported nmap XML. Falling back to port scanning.')
+			for plugin in target.autorecon.plugin_types['port']:
+				if config['proxychains'] and plugin.type == 'udp':
+					continue
+				processed_marker = os.path.join(scandir, '.port_scans', f".{plugin.slug}")
+				if os.path.exists(processed_marker):
+					info(f"Port Plugin {plugin.name} ({plugin.slug}) has already been run against {target.address}. Skipping.")
+					continue
+				if config['port_scans'] and plugin.slug in config['port_scans']:
+					matching_tags = True
+					excluded_tags = False
+				else:
+					plugin_tag_set = set(plugin.tags)
+					matching_tags = False
+					for tag_group in target.autorecon.tags:
+						if set(tag_group).issubset(plugin_tag_set):
+							matching_tags = True
+							break
+					excluded_tags = False
+					for tag_group in target.autorecon.excluded_tags:
+						if set(tag_group).issubset(plugin_tag_set):
+							excluded_tags = True
+							break
+				if matching_tags and not excluded_tags:
+					target.scans['ports'][plugin.slug] = {'plugin': plugin, 'commands': []}
+					pending.add(asyncio.create_task(port_scan(plugin, target)))
 	else:
 		for plugin in target.autorecon.plugin_types['port']:
 			if config['proxychains'] and plugin.type == 'udp':
@@ -621,7 +726,7 @@ async def scan_target(target):
 				timed_out = True
 				break
 
-		if not config['force_services']:
+		if not config['force_services'] and config.get('imported_nmap_services') is None:
 			# Extract Services
 			services = []
 
@@ -950,6 +1055,7 @@ async def run():
 	parser.add_argument('--disable-keyboard-control', action='store_true', help='Disables keyboard control ([s]tatus, Up, Down) if you are in SSH or Docker.')
 	parser.add_argument('--ignore-plugin-checks', action='store_true', help='Ignores errors from plugin check functions that would otherwise prevent AutoRecon from running. Default: %(default)s')
 	parser.add_argument('--force-services', action='store', nargs='+', metavar='SERVICE', help='A space separated list of services in the following style: tcp/80/http tcp/443/https/secure')
+	parser.add_argument('--import-nmap', action='store', type=str, metavar='XML_FILE', help='Import hosts and open ports from an existing nmap XML file and run service scans without port scanning. Targets found in the XML are added automatically unless targets are specified explicitly.')
 	parser.add_argument('-mpti', '--max-plugin-target-instances', action='store', nargs='+', metavar='PLUGIN:NUMBER', help='A space separated list of plugin slugs with the max number of instances (per target) in the following style: nmap-http:2 dirbuster:1. Default: %(default)s')
 	parser.add_argument('-mpgi', '--max-plugin-global-instances', action='store', nargs='+', metavar='PLUGIN:NUMBER', help='A space separated list of plugin slugs with the max number of global instances in the following style: nmap-http:2 dirbuster:1. Default: %(default)s')
 	parser.add_argument('--accessible', action='store_true', help='Attempts to make AutoRecon output more accessible to screenreaders. Default: %(default)s')
@@ -1385,7 +1491,7 @@ async def run():
 		errors = True
 
 	if not errors:
-		if config['force_services']:
+		if config['force_services'] or config.get('imported_nmap_services') is not None:
 			autorecon.service_scan_semaphore = asyncio.Semaphore(config['max_scans'])
 		else:
 			autorecon.port_scan_semaphore = asyncio.Semaphore(config['max_port_scans'])
@@ -1443,6 +1549,21 @@ async def run():
 			error('The target file ' + args.target_file + ' could not be read.')
 			sys.exit(1)
 
+	if args.import_nmap:
+		if not os.path.isfile(args.import_nmap):
+			error('The nmap XML file "' + args.import_nmap + '" was not found.')
+			sys.exit(1)
+		imported = parse_nmap_xml(args.import_nmap)
+		if not imported:
+			error('No hosts with open ports found in nmap XML file "' + args.import_nmap + '".')
+			sys.exit(1)
+		config['imported_nmap_services'] = imported
+		info('Imported ' + str(len(imported)) + ' host(s) from "' + args.import_nmap + '".')
+		# When no explicit targets given, use all hosts from the XML.
+		if not raw_targets:
+			for address in imported:
+				raw_targets.append(address)
+
 	unresolvable_targets = False
 	for target in raw_targets:
 		try:
@@ -1542,7 +1663,7 @@ async def run():
 		error('A total of ' + str(len(autorecon.pending_targets)) + ' targets would be scanned. If this is correct, re-run with the --disable-sanity-checks option to suppress this check.')
 		errors = True
 
-	if not config['force_services']:
+	if not config['force_services'] and config.get('imported_nmap_services') is None:
 		port_scan_plugin_count = 0
 		for plugin in autorecon.plugin_types['port']:
 			if config['port_scans'] and plugin.slug in config['port_scans']:
@@ -1621,7 +1742,7 @@ async def run():
 		for targ in autorecon.scanning_targets:
 			for process_list in targ.running_tasks.values():
 				# If we're not scanning ports, count ServiceScans instead.
-				if config['force_services']:
+				if config['force_services'] or config.get('imported_nmap_services') is not None:
 					if issubclass(process_list['plugin'].__class__, ServiceScan): # TODO should we really count ServiceScans? Test...
 						port_scan_task_count += 1
 				else: