This a follow-up question of Python scapy - TCP port scan of subnet
I've decided to make an all powerful subnet mapper in Scapy to better understand how tools like nmap work.
You can call it with python subnet_map.py "192.168.1.0/24" insert your subnet. You can add multiple optional arguments like --os for os-detection or --log for logging.
- It sends ARP request to the entire network to capture hosts who are up.
- For each IP in subnet that is up.
- It gets the ports from
etc/services/and send anTCPorUDPscan depending on the port type.
Changes:
- UDP scanner
- Logging
- Getting the ports from
/etc/services/as suggested by a comment - Os - Detection (although
scapy os detectis a bit broken/outdated)
Code
import argparse
from textwrap import dedent
import re
from collections import namedtuple
try:
from urllib.request import urlopen
except ImportError:
from urllib2 import urlopen
from scapy.all import *
class ScannerStatus(object):
OPEN = "Open"
CLOSED = "Closed"
FILTERED = "Filtered"
OPENFILTERED = "Open|Filtered"
class Scanner(object):
def __init__(self, timeout):
self.timeout = timeout
def arp_ping(self, subnet):
"""ARP Pings entire subnet returns found in subnet."""
conf.verb = 0
answered, unanswered = srp(Ether(dst="ff:ff:ff:ff:ff:ff")/ARP(pdst=subnet), timeout=self.timeout, verbose=False, inter=0.1)
return [rcv.sprintf(r"%Ether.src% - %ARP.psrc%") for snd, rcv in answered]
def _tcp_default(self, dst_ip, dst_port, src_port):
"""Default TCP Scan."""
default_scan = sr1(IP(dst=dst_ip)/TCP(sport=src_port, dport=dst_port, flags="S"), timeout=self.timeout)
if default_scan is not None:
if default_scan.getlayer(TCP).flags == 0x12:
send_rst = sr(IP(dst=dst_ip)/TCP(sport=src_port, dport=dst_port, flags="AR"), timeout=self.timeout)
return ScannerStatus.OPEN
return ScannerStatus.CLOSED
def _tcp_stealth(self, dst_ip, dst_port, src_port):
"""Stealthy TCP Scan"""
stealth_scan = sr1(IP(dst=dst_ip)/TCP(sport=src_port, dport=dst_port, flags="S"), timeout=self.timeout)
if stealth_scan is not None:
if stealth_scan.getlayer(TCP).flags == 0x12:
send_rst = sr(IP(dst=dst_ip)/TCP(sport=src_port, dport=dst_port, flags="R"), timeout=self.timeout)
return ScannerStatus.OPEN
elif stealth_scan.getlayer(TCP).flags == 0x14:
return ScannerStatus.CLOSED
return ScannerStatus.FILTERED
def tcp(self, dst_ip, dst_port, stealth=False):
"""Scan TCP port for availability
arg steath: stealthy or default scan
arg dst_port: targets destination port
arg dst_ip: targets ip address
return: ScannerStatus"""
src_port = RandShort()
fn = self._tcp_stealth if stealth else self._tcp_default
return fn(dst_ip, dst_port, src_port)
def udp(self, dst_ip, dst_port):
"""Scan UDP port for availability
arg dst_port: targets destination port
arg dst_ip: targets ip address
return: ScannerStatus"""
udp_scan = sr1(IP(dst=dst_ip)/UDP(dport=dst_port),timeout=self.timeout)
if udp_scan is not None:
if udp_scan.haslayer(UDP):
return ScannerStatus.OPEN
elif int(udp_scan.getlayer(ICMP).type) == 3 and int(udp_scan.getlayer(ICMP).code) == 3:
return ScannerStatus.CLOSED
else:
retrans = [sr1(IP(dst=dst_ip)/UDP(dport=dst_port),timeout=self.timeout) for count in range(3)]
for item in retrans:
if item is not None:
return ScannerStatus.OPENFILTERED
return ScannerStatus.FILTERED
def get_ports():
"""Get ports from etc/services/
yields: a namedtuple(name, port, type)"""
Port = namedtuple('Port', ['name', 'port', 'type'])
with open('/etc/services') as ports_file:
lines = ports_file.readlines()
for line in lines:
if not line.startswith('#') and line.rstrip():
_name, _port, _type = re.split('[\s/]+', line.split('#', 1)[0])[:3]
yield Port(_name, _port, _type)
def parse_arguments():
"""Arguments parser.
returns: arg.stealth <bool>
arg.subnet <string>
arg.timeout <int>
arg.log <string> filename
arg.os <bool>"""
parser = argparse.ArgumentParser(usage='%(prog)s [options] <subnet>',
description='port scanning tool @Ludisposed',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=dedent('''\
Examples:
python subnet_map.py "192.168.1.0/24" -s
python subnet_map.py "192.168.1.0/24" --timeout 10
python subnet_map.py "192.168.1.0/24 -l -s"'''))
parser.add_argument('-s', '--stealth', default=False, action="store_true", help='Stealthy TCP scan')
parser.add_argument('--timeout', type=int, default=2, help='Timeout parameter of scans')
parser.add_argument('-l', '--log', type=str, default='', help="Log the data to a file")
parser.add_argument('subnet', type=str, help='subnet in from of [ip]/[bitmask]')
parser.add_argument('-O', '--os', default=False, action="store_true", help="Make fingerprint and try scan OS")
return parser.parse_args()
def main():
def save_log(data):
if args.log:
with open(args.log, 'a+') as f:
f.write(data + '\n')
args = parse_arguments()
scanner = Scanner(args.timeout)
scan_type = 'stealth' if args.stealth else 'default'
# Os - detection setup
load_module("nmap")
if not os.path.isfile('nmap-os-fingerprints'):
open('nmap-os-fingerprints', 'wb').write(urlopen('https://raw.githubusercontent.com/nmap/nmap/9efe1892/nmap-os-fingerprints').read())
conf.nmap_base = 'nmap-os-fingerprints'
network = scanner.arp_ping(args.subnet)
for connection in network:
mac, ip = connection.split(' - ')
print('\n[!] Trying port scan of current connection with mac={} and ip={}\n'.format(mac, ip))
for port in get_ports():
data = ''
if port.type == 'tcp':
status = scanner.tcp(ip, int(port.port), args.stealth)
if status == 'Open':
data = '[*] TCP {} scan: dest_ip={} port={}, service={}, status={}' \
.format(scan_type, ip, port.port, port.name, status)
if port.type == 'udp':
status = scanner.udp(ip, int(port.port))
if 'Open' in status:
data = '[*] UDP scan: dest_ip={} port={}, service{}, status={}' \
.format(ip, port.port, port.name, status)
if data:
save_log(data)
print(data)
if args.os:
data = re.findall(r"\[\'(.*)\'\]", str(nmap_fp(ip)))[0]
if data:
save_log("\nOS detected: {}".format(data))
print("\nOS detected: {}".format(data))
if __name__ == '__main__':
main()
Questions
It is still slow as hell! When I compare this with nmap subnet I'm amazed at how fast nmap is. Is there any way to speed things up a bit?
