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 "" insert your subnet. You can add multiple optional arguments like --os for os-detection or --log for logging.

  1. It sends ARP request to the entire network to capture hosts who are up.
  2. For each IP in subnet that is up.
  3. It gets the ports from etc/services/ and send an TCP or UDP scan depending on the port type.


  1. UDP scanner
  2. Logging
  3. Getting the ports from /etc/services/ as suggested by a comment
  4. Os - Detection (although scapy os detect is a bit broken/outdated)


import argparse
from textwrap import dedent
import re
from collections import namedtuple
    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
            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',
                                        python subnet_map.py "" -s
                                        python subnet_map.py "" --timeout 10
                                        python subnet_map.py " -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
    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:

        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__':


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?

  • \$\begingroup\$ You are on your seventh edit of this question, where most have a minor change, and a caption like 'I can't copy and paste'. Please stop. \$\endgroup\$
    – Peilonrayz
    Oct 19, 2017 at 14:47
  • \$\begingroup\$ It looks a bit gamy, and is a little annoying if you're going to answer. But it's fine, just try not to do it again, :) \$\endgroup\$
    – Peilonrayz
    Oct 20, 2017 at 8:37


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.