Test TCP ports with Python and Scapy

In Stop using Telnet to test ports, I explored several alternative commands and scripts to test TCP connectivity. These commands and scripts range from basic tests to more sophisticated checks, but they are limited to the features provided by supporting tools like Netcat.

There is another option when you want exceptional control and flexibility for your TCP port checks: Do it yourself. Programming languages like Python offer socket programming APIs and access to sophisticated frameworks like Scapy to accomplish just that.

[ Cheat sheet: Get a list of Linux utilities and commands for managing servers and networks. ]

Get started with a TCP port check

Start with a simple TCP port check in Python:

#!/usr/bin/env python3
"""
VERY simple port TCP port check
https://docs.python.org/3/library/socket.html
Author: Jose Vicente Nunez <@josevnz@fosstodon.org>
"""
import socket
from pathlib import Path
from typing import Dict, List
from argparse import ArgumentParser


def load_machines_port(the_data_file: Path) -> Dict[str, List[int]]:
    port_data = {}
    with open(the_data_file, 'r') as d_scan:
        for line in d_scan:
            host, ports = line.split()
            port_data[host] = [int(p) for p in ports.split(',')]
    return port_data


def test_port(address: str, dest_port: int) -> bool:
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            if sock.connect_ex((address, dest_port)) == 0:
                return True
        return False
    except (OSError, ValueError):
        return False


if __name__ == "__main__":
    PARSER = ArgumentParser(description=__doc__)
    PARSER.add_argument("scan_file", type=Path, help="Scan file with list of hosts and ports")
    ARGS = PARSER.parse_args()
    data = load_machines_port(ARGS.scan_file)
    for machine in data:
        for port in data[machine]:
            try:
                results = test_port(machine, port)
            except (OSError, ValueError):
                results = False
            if results:
                print(f"{machine}:{port}: OK")
            else:
                print(f"{machine}:{port}: ERROR")

This application opens the socket and assumes that any error means the port is closed.

Give it a try:

$ ./tcp_port_scan.py port_scan.csv
google.com:80: OK
amazon.com:80: OK
raspberrypi:22: OK
raspberrypi:9090: OK
raspberrypi:8086: OK
raspberrypi:21: ERROR
dmaf5:22: OK
dmaf5:80: ERROR

It works as expected. But what if you could use a framework that allows you to skip all the boilerplate while doing more complex things?

Meet Scapy

Scapy describes itself as "a Python program that enables the user to send, sniff and dissect, and forge network packets." Using this capability, you can build tools that can probe, scan, test, or discover networks.

Most Linux distributions have a package for Scapy. On Fedora, install it like this:

$ sudo dnf install -y python3-scapy.noarch

Scapy requires elevated privileges to run. If you decide to use pip, you may do the following:

sudo -i
python3 -m venv /usr/local/scapy
. /usr/local/scapy/bin/activate
pip install --upgrade pip
pip install wheel
pip install scapy

Just remember to activate your virtual environment before calling Scapy if you install it that way. You can use Scapy as a library or as an interactive shell. Next, I'll show you a few applications.

Try a simple interactive TCP port scanner

In the interactive mode, you call the Scapy terminal as root, as it requires elevated privileges.

For that, you will add layers. First, add an IP network layer:

IP(dst="raspberrypi.home")

Then add TCP ports:

TCP(dport=[22,3000,8086]

Next, send the packets and capture answered and unanswered results:

(ans, notanws) = sr(*)

Then analyze the answered results, filtering only open ports:

ans.summary(lfilter = lambda s,r: r.sprintf("%TCP.flags%") == "SA",prn=lambda s,r: r.sprintf("%TCP.sport% is open"))

Here's what you'll get:

$ sudo scapy3 -H
>>> (ans, notanws) = sr(IP(dst="raspberrypi.home")/TCP(dport=[22,3000,8086]))
Begin emission:
Finished sending 3 packets.

Received 5 packets, got 3 answers, remaining 0 packets

>>> ans.summary(lfilter = lambda s,r: r.sprintf("%TCP.flags%") == "SA",prn=lambda s,r: r.sprintf("%TCP.sport% is open"))
ssh is open
hbci is open
d_s_n is open

Not bad for just two lines of code, compared to 46 from the first Python script.

Next. you'll create an automated port scanner, using what you learned before.

[ Download now: A system administrator's guide to IT automation. ]

Create a Scapy-flavored custom port check

The interactive shell is nice when you are exploring and experimenting to find the best way to tackle a problem. But once you come up with a solution, you can make it a script:

#!/usr/bin/env -S sudo python3
"""
VERY simple port TCP port check, using Scapy
* https://scapy.readthedocs.io/en/latest/usage.html
* https://scapy.readthedocs.io/en/latest/api/scapy.html
* https://0xbharath.github.io/art-of-packet-crafting-with-scapy/scapy/sending_recieving/index.html
* Please check out the original script: https://thepacketgeek.com/scapy/building-network-tools/part-10/
Author: Jose Vicente Nunez <@josevnz@fosstodon.org>
"""
import os
import sys
import traceback
from enum import IntEnum
from pathlib import Path
from random import randint
from typing import Dict, List
from argparse import ArgumentParser
from scapy.layers.inet import IP, TCP, ICMP
from scapy.packet import Packet
from scapy.sendrecv import sr1, sr

NON_PRIVILEGED_LOW_PORT = 1025
NON_PRIVILEGED_HIGH_PORT = 65534
ICMP_DESTINATION_UNREACHABLE = 3


class TcpFlags(IntEnum):
    """
    https://www.wireshark.org/docs/wsug_html_chunked/ChAdvTCPAnalysis.html
    """
    SYNC_ACK = 0x12
    RST_PSH = 0x14


class IcmpCodes(IntEnum):
    """
    ICMP codes, to decide
    https://www.ibm.com/docs/en/qsip/7.4?topic=applications-icmp-type-code-ids
    """
    Host_is_unreachable = 1
    Protocol_is_unreachable = 2
    Port_is_unreachable = 3
    Communication_with_destination_network_is_administratively_prohibited = 9
    Communication_with_destination_host_is_administratively_prohibited = 10
    Communication_is_administratively_prohibited = 13


FILTERED_CODES = [x.value for x in IcmpCodes]


class RESPONSES(IntEnum):
    """
    Customized responses for our port check
    """
    FILTERED = 0
    CLOSED = 1
    OPEN = 2
    ERROR = 3


def load_machines_port(the_data_file: Path) -> Dict[str, List[int]]:
    port_data = {}
    with open(the_data_file, 'r') as d_scan:
        for line in d_scan:
            host, ports = line.split()
            port_data[host] = [int(p) for p in ports.split(',')]
    return port_data


def test_port(
        address: str,
        dest_ports: int,
        verbose: bool = False
) -> RESPONSES:
    """
    Test the address + port combination
    :param address:  Host to check
    :param dest_ports: Ports to check
    :return: Answer and Unanswered packets (filtered)
    """
    src_port = randint(NON_PRIVILEGED_LOW_PORT, NON_PRIVILEGED_HIGH_PORT)
    ip = IP(dst=address)
    ports = TCP(sport=src_port, dport=dest_ports, flags="S")
    reset_tcp = TCP(sport=src_port, dport=dest_ports, flags="S")
    packet: Packet = ip / ports
    verb_level = 0
    if verbose:
        verb_level = 99
        packet.show()
    try:
        answered = sr1(
            packet,
            verbose=verb_level,
            retry=1,
            timeout=1,
            threaded=True
        )
        if not answered:
            return RESPONSES.FILTERED
        elif answered.haslayer(TCP):
            if answered.getlayer(TCP).flags == TcpFlags.SYNC_ACK:
                rst_packet = ip / reset_tcp
                sr(rst_packet, timeout=1, verbose=verb_level)
                return RESPONSES.OPEN
            elif answered.getlayer(TCP).flags == TcpFlags.RST_PSH:
                return RESPONSES.CLOSED
        elif answered.haslayer(ICMP):
            icmp_type = answered.getlayer(ICMP).type
            icmp_code = int(answered.getlayer(ICMP).code)
            if icmp_type == ICMP_DESTINATION_UNREACHABLE and icmp_code in FILTERED_CODES:
                return RESPONSES.FILTERED
    except TypeError:
        traceback.print_exc(file=sys.stdout)
        return RESPONSES.ERROR


if __name__ == "__main__":
    if os.getuid() != 0:
        raise EnvironmentError(f"Sorry, you need to be root to run this program!")
    PARSER = ArgumentParser(description=__doc__)
    PARSER.add_argument("--verbose", action="store_true", help="Toggle verbose mode on/ off")
    PARSER.add_argument("scan_file", type=Path, help="Scan file with list of hosts and ports")
    ARGS = PARSER.parse_args()
    data = load_machines_port(ARGS.scan_file)
    for machine in data:
        m_ports = data[machine]
        for dest_port in m_ports:
            ans = test_port(address=machine, dest_ports=dest_port, verbose=ARGS.verbose)
            print(f"{ans.name} -> {machine}:{dest_port}")

This script is more complex than the first, which uses Python alone, but it offers a more detailed explanation of the analyzed ports. You can run it like this: ./tcp_port_scan_scapy.py port_scan.csv:

$ ./tcp_port_scan_scapy.py port_scan.csv
OPEN -> google.com:80
OPEN -> amazon.com:80
OPEN -> raspberrypi:22
OPEN -> raspberrypi:9090
OPEN -> raspberrypi:8086
CLOSED -> raspberrypi:21
FILTERED -> dmaf5:22
FILTERED -> dmaf5:80

The results for my system show one connection closed and two of them possibly filtered.

The real power of Scapy is the level of customization you now have from a familiar language like Python. The shell mode is particularly important as you can troubleshoot network problems easily while doing some exploration work.

What to learn next

Developing a TCP port scanner using a programming language like Python provides a level of flexibility and customization that is hard to achieve with scripting alone. By adding a specialized library like Scapy, you can perform even more complex network packet manipulation. Read this tutorial for Scapy, and you'll be amazed at what you can do.

[ Network getting out of control? Check out Network automation for everyone, a complimentary book from Red Hat. ]