38 min read

 In this article, given by Pradeeban Kathiravelu, author of the book Python Network Programming Cookbook – Second Edition, we will cover the following topics:

  • Forwarding a local port to a remote host
  • Pinging hosts on the network with ICMP
  • Waiting for a remote network service
  • Enumerating interfaces on your machine
  • Finding the IP address for a specific interface on your machine
  • Finding whether an interface is up on your machine
  • Detecting inactive machines on your network
  • Performing a basic IPC using connected sockets (socketpair)
  • Performing IPC using Unix domain sockets
  • Finding out if your Python supports IPv6 sockets
  • Extracting an IPv6 prefix from an IPv6 address
  • Writing an IPv6 echo client/server

(For more resources related to this topic, see here.)

This article extends the use of Python’s socket library with a few third-party libraries. It also discusses some advanced techniques, for example, the asynchronous asyncore module from the Python standard library. This article also touches upon various protocols, ranging from an ICMP ping to an IPv6 client/server.

In this article, a few useful Python third-party modules have been introduced by some example recipes. For example, the network packet capture library, Scapy, is well known among Python network programmers.

A few recipes have been dedicated to explore the IPv6 utilities in Python including an IPv6 client/server. Some other recipes cover Unix domain sockets.

Forwarding a local port to a remote host

Sometimes, you may need to create a local port forwarder that will redirect all traffic from a local port to a particular remote host. This might be useful to enable proxy users to browse a certain site while preventing them from browsing some others.

How to do it…

Let us create a local port forwarding script that will redirect all traffic received at port 8800 to the Google home page (http://www.google.com). We can pass the local and remote host as well as port number to this script. For the sake of simplicity, let’s only specify the local port number as we are aware that the web server runs on port 80.

Listing 3.1 shows a port forwarding example, as follows:

#!/usr/bin/env python
# This program is optimized for Python 2.7.12 and Python 3.5.2.
# It may run on any other version with/without modifications.

import argparse

LOCAL_SERVER_HOST = 'localhost'
REMOTE_SERVER_HOST = 'www.google.com'
BUFSIZE = 4096

import asyncore
import socket

class PortForwarder(asyncore.dispatcher):
    def __init__(self, ip, port, remoteip,remoteport,backlog=5):

    def handle_accept(self):
        conn, addr = self.accept()
        print ("Connected to:",addr)

class Receiver(asyncore.dispatcher):
    def __init__(self,conn):

    def handle_connect(self):

    def handle_read(self):
        read = self.recv(BUFSIZE)
        self.from_remote_buffer += read

    def writable(self):
        return (len(self.to_remote_buffer) > 0)

    def handle_write(self):
        sent = self.send(self.to_remote_buffer)
        self.to_remote_buffer = self.to_remote_buffer[sent:]

    def handle_close(self):
        if self.sender:

class Sender(asyncore.dispatcher):
    def __init__(self, receiver, remoteaddr,remoteport):
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect((remoteaddr, remoteport))

    def handle_connect(self):

    def handle_read(self):
        read = self.recv(BUFSIZE)
        self.receiver.to_remote_buffer += read

    def writable(self):
        return (len(self.receiver.from_remote_buffer) > 0)

    def handle_write(self):
        sent = self.send(self.receiver.from_remote_buffer)
        self.receiver.from_remote_buffer = self.receiver.from_remote_buffer[sent:]

    def handle_close(self):

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Stackless Socket Server Example')
    parser.add_argument('--local-host', action="store", dest="local_host", default=LOCAL_SERVER_HOST)
    parser.add_argument('--local-port', action="store", dest="local_port", type=int, required=True)
    parser.add_argument('--remote-host', action="store", dest="remote_host",  default=REMOTE_SERVER_HOST)
    parser.add_argument('--remote-port', action="store", dest="remote_port", type=int, default=80)
    given_args = parser.parse_args() 
    local_host, remote_host = given_args.local_host, given_args.remote_host
    local_port, remote_port = given_args.local_port, given_args.remote_port

    print ("Starting port forwarding local %s:%s => remote %s:%s" % (local_host, local_port, remote_host, remote_port))
    PortForwarder(local_host, local_port, remote_host, remote_port)

If you run this script, it will show the following output:

$ python 3_1_port_forwarding.py --local-port=8800
Starting port forwarding local localhost:8800 => remote www.google.com:80 

Now, open your browser and visit http://localhost:8800. This will take you to the Google home page and the script will print something similar to the following command:

('Connected to:', ('', 37236))

The following screenshot shows the forwarding a local port to a remote host:

Python Network Programming Cookbook - Second Edition

How it works…

We created a port forwarding class, PortForwarder subclassed, from asyncore.dispatcher, which wraps around the socket object. It provides a few additional helpful functions when certain events occur, for example, when the connection is successful or a client is connected to a server socket. You have the choice of overriding the set of methods defined in this class. In our case, we only override the handle_accept() method.

Two other classes have been derived from asyncore.dispatcher. The receiver class handles the incoming client requests and the sender class takes this receiver instance and processes the sent data to the clients. As you can see, these two classes override the handle_read(), handle_write(), and writeable() methods to facilitate the bi-directional communication between the remote host and local client.

In summary, the PortForwarder class takes the incoming client request in a local socket and passes this to the sender class instance, which in turn uses the receiver class instance to initiate a bi-directional communication with a remote server in the specified port.

Pinging hosts on the network with ICMP

An ICMP ping is the most common type of network scanning you have ever encountered. It is very easy to open a command-line prompt or terminal and type ping www.google.com. How difficult is that from inside a Python program? This recipe shows you an example of a Python ping.

Getting ready

You need the superuser or administrator privilege to run this recipe on your machine.

How to do it…

You can lazily write a Python script that calls the system ping command-line tool, as follows:

import subprocess
import shlex

command_line = "ping -c 1 www.google.com"
args = shlex.split(command_line)
    print ("Google web server is up!")
except subprocess.CalledProcessError:
    print ("Failed to get ping.")

However, in many circumstances, the system’s ping executable may not be available or may be inaccessible. In this case, we need a pure Python script to do that ping. Note that this script needs to be run as a superuser or administrator.

Listing 3.2 shows the ICMP ping, as follows:

#!/usr/bin/env python
# This program is optimized for Python 3.5.2.
# Instructions to make it run with Python 2.7.x is as follows.
# It may run on any other version with/without modifications.

import os 
import argparse 
import socket
import struct
import select
import time

ICMP_ECHO_REQUEST = 8 # Platform specific

class Pinger(object):
    """ Pings to a host -- the Pythonic way"""

    def __init__(self, target_host, count=DEFAULT_COUNT, timeout=DEFAULT_TIMEOUT):
        self.target_host = target_host
        self.count = count
        self.timeout = timeout

    def do_checksum(self, source_string):
        """  Verify the packet integritity """
        sum = 0
        max_count = (len(source_string)/2)*2
        count = 0
        while count > 16)  +  (sum & 0xffff)
         sum = sum + (sum >> 16)
         answer = ~sum
         answer = answer & 0xffff
         answer = answer >> 8 | (answer 

We need a send_ping() method that will send the data of a ping request to the target host. Also, this will call the do_checksum() method for checking the integrity of the ping data, as follows:

def send_ping(self, sock,  ID):
        Send ping to the target host
        target_addr  =  socket.gethostbyname(self.target_host)

        my_checksum = 0

        # Create a dummy heder with a 0 checksum.
        header = struct.pack("bbHHh", ICMP_ECHO_REQUEST, 0, my_checksum, ID, 1)
        bytes_In_double = struct.calcsize("d")
        data = (192 - bytes_In_double) * "Q"
        data = struct.pack("d", time.time()) + bytes(data.encode('utf-8'))

        # Get the checksum on the data and the dummy header.
        my_checksum = self.do_checksum(header + data)
        header = struct.pack(
            "bbHHh", ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), ID, 1
        packet = header + data
        sock.sendto(packet, (target_addr, 1))

Let us define another method called ping_once() that makes a single ping call to the target host. It creates a raw ICMP socket by passing the ICMP protocol to socket(). The exception handling code takes care if the script is not run by a superuser or if any other socket error occurs. Let’s take a look at the following code:

    def ping_once(self):
        Returns the delay (in seconds) or none on timeout.
        icmp = socket.getprotobyname("icmp")
            sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp)
        except socket.error as e:
            if e.errno == 1:
                # Not superuser, so operation not permitted
                e.msg +=  "ICMP messages can only be sent from root user processes"
                raise socket.error(e.msg)
        except Exception as e:
            print ("Exception: %s" %(e))

        my_ID = os.getpid() & 0xFFFF

        self.send_ping(sock, my_ID)
        delay = self.receive_pong(sock, my_ID, self.timeout)
        return delay

The main executive method of this class is ping(). It runs a for loop inside which the ping_once() method is called count times and receives a delay in the ping response in seconds. If no delay is returned, that means the ping has failed. Let’s take a look at the following code:

def ping(self):
        Run the ping process
        for i in range(self.count):
            print ("Ping to %s..." % self.target_host,)
                delay  =  self.ping_once()
            except socket.gaierror as e:
                print ("Ping failed. (socket error: '%s')" % e[1])

            if delay  ==  None:
                print ("Ping failed. (timeout within %ssec.)" % self.timeout)
                delay  =  delay * 1000
                print ("Get pong in %0.4fms" % delay)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Python ping')
    parser.add_argument('--target-host', action="store", dest="target_host", required=True)
    given_args = parser.parse_args()  
    target_host = given_args.target_host
    pinger = Pinger(target_host=target_host)

This script shows the following output. This has been run with the superuser privilege:

$ sudo python 3_2_ping_remote_host.py --target-host=www.google.com  
Ping to www.google.com...
Get pong in 27.0808ms
Ping to www.google.com...
Get pong in 17.3445ms
Ping to www.google.com...
Get pong in 33.3586ms
Ping to www.google.com...
Get pong in 32.3212ms

How it works…

A Pinger class has been constructed to define a few useful methods. The class initializes with a few user-defined or default inputs, which are as follows:

  • target_host: This is the target host to ping
  • count: This is how many times to do the ping
  • timeout: This is the value that determines when to end an unfinished ping operation

The send_ping() method gets the DNS hostname of the target host and creates an ICMP_ECHO_REQUEST packet using the struct module. It is necessary to check the data integrity of the method using the do_checksum() method. It takes the source string and manipulates it to produce a proper checksum. On the receiving end, the receive_pong() method waits for a response until the timeout occurs or receives the response. It captures the ICMP response header and then compares the packet ID and calculates the delay in the request and response cycle.

Waiting for a remote network service

Sometimes, during the recovery of a network service, it might be useful to run a script to check when the server is online again.

How to do it…

We can write a client that will wait for a particular network service forever or for a timeout. In this example, by default, we would like to check when a web server is up in localhost. If you specified some other remote host or port, that information will be used instead.

Listing 3.3 shows waiting for a remote network service, as follows:

#!/usr/bin/env python
# This program is optimized for Python 2.7.12 and Python 3.5.2.
# It may run on any other version with/without modifications.

import argparse
import socket
import errno
from time import time as now


class NetServiceChecker(object):
    """ Wait for a network service to come online"""
    def __init__(self, host, port, timeout=DEFAULT_TIMEOUT):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    def end_wait(self):

    def check(self):
        """ Check the service """
        if self.timeout:
            end_time = now() + self.timeout

        while True:
                if self.timeout:
                    next_timeout = end_time - now()
                    if next_timeout 

If a web server is running on your machine, this script will show the following output:

$ python 3_3_wait_for_remote_service.py 
Waiting for network service localhost:80 ... 
setting socket next timeout 120.0s 
Service is available again!

If you do not have a web server already running in your computer, make sure to install one such as Apache 2 Web Server:

$ sudo apt install apache2

Now, stop the Apache process:

$ sudo /etc/init.d/apache2 stop

It will print the following message while stopping the service.

[ ok ] Stopping apache2 (via systemctl): apache2.service.

Run this script, and start Apache again.

$ sudo /etc/init.d/apache2 start
[ ok ] Starting apache2 (via systemctl): apache2.service.

The output pattern will be different. On my machine, the following output pattern was found:

Exception: [Errno 103] Software caused connection abort
setting socket next timeout 119.0s
Exception: [Errno 111] Connection refused
setting socket next timeout 119.0s
Exception: [Errno 103] Software caused connection abort
setting socket next timeout 119.0s
Exception: [Errno 111] Connection refused
setting socket next timeout 119.0s

And finally when Apache2 is up again, the following log is printed:

Service is available again!

The following screenshot shows the waiting for an active Apache web server process:

Python Network Programming Cookbook - Second Edition

How it works…

The preceding script uses the argparse module to take the user input and process the hostname, port, and timeout, that is how long our script will wait for the desired network service. It launches an instance of the NetServiceChecker class and calls the check() method. This method calculates the final end time of waiting and uses the socket’s settimeout() method to control each round’s end time, that is next_timeout. It then uses the socket’s connect() method to test if the desired network service is available until the socket timeout occurs. This method also catches the socket timeout error and checks the socket timeout against the timeout values given by the user.

Enumerating interfaces on your machine

If you need to list the network interfaces present on your machine, it is not very complicated in Python. There are a couple of third-party libraries out there that can do this job in a few lines. However, let’s see how this is done using a pure socket call.

Getting ready

You need to run this recipe on a Linux box. To get the list of available interfaces, you can execute the following command:

$ /sbin/ifconfig

How to do it…

Listing 3.4 shows how to list the networking interfaces, as follows:

#!/usr/bin/env python
# Python Network Programming Cookbook, Second Edition -- Article - 3
# This program is optimized for Python 2.7.12 and Python 3.5.2.
# It may run on any other version with/without modifications.

import sys
import socket
import fcntl
import struct
import array

SIOCGIFCONF = 0x8912 #from C library sockios.h
STUCT_SIZE_32 = 32
STUCT_SIZE_64 = 40

def list_interfaces():
    interfaces = []
    max_interfaces = DEFAULT_INTERFACES
    is_64bits = sys.maxsize > PLATFORM_32_MAX_NUMBER
    struct_size = STUCT_SIZE_64 if is_64bits else STUCT_SIZE_32
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 
    while True:
        bytes = max_interfaces * struct_size
        interface_names = array.array('B', b'

Subscribe to the weekly Packt Hub newsletter. We'll send you the results of our AI Now Survey, featuring data and insights from across the tech landscape.


Please enter your comment!
Please enter your name here