How to Build a Custom Netcat with Python

Learn how to create a custom Python-based Netcat tool for network debugging, file transfers, and remote command execution. Perfect for networking and cybersecurity enthusiasts, this guide provides hands-on code and practical examples.
  · 14 min read · Updated nov 2024 · Ethical Hacking

Step up your coding game with AI-powered Code Explainer. Get insights like never before!

Introduction

Netcat, often called the "Swiss Army Knife" of networking tools, is an essential tool for network debugging, port scanning, data transfer, and basic pentesting. This tutorial explores a custom Python implementation that provides similar functionality. Given its versatility (Netcat), it could be a powerful asset if an attacker gains access to a system. 

With netcat, you can read and write data across the network, enabling you to execute remote commands, transfer files, or even open a remote shell. There have been situations where servers lacked the netcat tool but had Python installed. In such cases, creating a simple network client and server in Python can be invaluable, allowing you to push files or set up a listener for command-line access. If you've exploited a vulnerability through a web application, dropping a Python callback for secondary access can provide a secure alternative without risking the use of trojans or backdoors. This is a perfect exercise to sharpen your networking skills.

Our custom netcat implementation will offer several powerful capabilities:

  • Creating client and server connections
  • Executing remote commands
  • File uploading
  • Interactive command shell
  • Graceful connection handling and error management

Now given these capabilities, you may be wondering how what we are about to build is different from a reverse shell. It's a valid thought. However, you should know that netcat is more versatile than a reverse shell. As a matter of fact, netcat can also be used as a reverse shell amongst other things - as we'll see.

Requirements

For this demonstration, I'll be making use of Kali Linux and Python 3. Netcat comes pre-installed with Kali. You can get a glimpse of the program using the following command:

$ nc -h

Implementation

Now, let's start building ours! Open up a single Python file and name it netcat.py

We'll start by importing the necessary modules. No need to install anything as they're all part of Python's standard library.

import sys, socket, getopt, threading, subprocess, signal, time
  • sys – For system-specific parameters and functions
  • socket – For low-level networking interface
  • getopt – For command-line option parsing. Similar to argparse, it's API is designed for users of the getopt() function in C. Always a good idea to try different methods.
  • threading – For creating and managing threads
  • subprocess – For spawning new processes and interacting with them
  • signal – For Handling asynchronous events (signals)
  • time – For Time-related functions

Next, we create a class for our netcat tool. This will include all the necessary methods:

class NetCat:
    def __init__(self, target, port):
        self.listen = False
        self.command = False
        self.upload = False
        self.execute = ""
        self.target = target
        self.upload_destination = ""
        self.port = port
        self.running = True
        self.threads = []

    def signal_handler(self, signum, frame):
        print('\n[*] User requested an interrupt. Exiting gracefully.')
        self.running = False
        time.sleep(0.5)
        sys.exit(0)

    def run_command(self, cmd):
        cmd = cmd.rstrip()
        try:
            output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True)
        except subprocess.CalledProcessError as e:
            output = e.output
        except Exception as e:
            output = str(e).encode()
        return output

The NetCat() class defines a network communication tool that can handle signals, run system commands, and manage various networking functions like listening, uploading, and executing commands remotely.

    def handle_client(self, client_socket):
        try:
            if len(self.upload_destination):
                file_buffer = ""
                while self.running:
                    try:
                        data = client_socket.recv(1024)
                        if not data:
                            break
                        else:
                            file_buffer += data.decode('utf-8')
                    except (ConnectionResetError, BrokenPipeError) as e:
                        print(f"[!] Connection error during upload: {str(e)}")
                        break
                    except Exception as e:
                        print(f"[!] Error receiving data: {str(e)}")
                        break

                try:
                    with open(self.upload_destination, "wb") as file_descriptor:
                        file_descriptor.write(file_buffer.encode('utf-8'))
                    try:
                        client_socket.send(
                            f"Successfully saved file to {self.upload_destination}\r\n".encode('utf-8'))
                    except (BrokenPipeError, ConnectionResetError):
                        print("[!] Couldn't send success message - connection lost")
                except OSError as e:
                    print(f"[!] File operation failed: {str(e)}")
                    try:
                        client_socket.send(
                            f"Failed to save file to {self.upload_destination}\r\n".encode('utf-8'))
                    except (BrokenPipeError, ConnectionResetError):
                        print("[!] Couldn't send error message - connection lost")

            if len(self.execute) and self.running:
                try:
                    output = self.run_command(self.execute)
                    client_socket.send(output)
                except (BrokenPipeError, ConnectionResetError):
                    print("[!] Couldn't send command output - connection lost")
                except Exception as e:
                    print(f"[!] Error executing command: {str(e)}")

            if self.command:
                while self.running:
                    try:
                        # Send prompt
                        client_socket.send(b"<Target:#> ")
                        
                        # Receive command
                        cmd_buffer = b''
                        while b"\n" not in cmd_buffer and self.running:
                            try:
                                data = client_socket.recv(1024)
                                if not data:
                                    raise ConnectionResetError("No data received")
                                cmd_buffer += data
                            except socket.timeout:
                                continue
                            except (ConnectionResetError, BrokenPipeError):
                                raise
                        
                        if not self.running:
                            break

                        # Execute command and send response
                        try:
                            cmd = cmd_buffer.decode().strip()
                            if cmd.lower() in ['exit', 'quit']:
                                print("[*] User requested exit")
                                break
                                
                            output = self.run_command(cmd)
                            if output:
                                client_socket.send(output + b"\n")
                            else:
                                client_socket.send(b"Command completed without output\n")
                                
                        except (BrokenPipeError, ConnectionResetError):
                            print("[!] Connection lost while sending response")
                            break
                        except Exception as e:
                            error_msg = f"Error executing command: {str(e)}\n"
                            try:
                                client_socket.send(error_msg.encode())
                            except:
                                break

                    except ConnectionResetError:
                        print("[!] Connection reset by peer")
                        break
                    except BrokenPipeError:
                        print("[!] Broken pipe - connection lost")
                        break
                    except Exception as e:
                        print(f"[!] Error in command loop: {str(e)}")
                        break

        except Exception as e:
            print(f"[!] Exception in handle_client: {str(e)}")
        finally:
            try:
                client_socket.close()
                print("[*] Client connection closed")
            except:
                pass

The handle_client() method in the NetCat class manages communication with a connected client. It performs the following tasks:

  1. File Uploading: If a file destination is specified (upload_destination), it listens for data from the client in chunks, storing the data in a buffer. Once the entire file is received, it attempts to save the file locally at the specified location. If successful, it sends a success message to the client; otherwise, it sends an error message if file operations fail.
  2. Command Execution: If a command (execute) is specified, it runs the command using the run_command() method and sends the output back to the client.
  3. Interactive Command Shell: If the command flag is set, it enters a loop where it continuously prompts the client for commands. It reads the command input, executes it, and sends the output back to the client. If the command is "exit" or "quit," the connection is terminated. If any errors occur while receiving or sending data, appropriate error messages are printed and sent to the client.
  4. Error Handling: Throughout the process, the method catches exceptions for common connection errors (like BrokenPipeError or ConnectionResetError), and prints appropriate messages if errors occur during data reception, file saving, or command execution.
  5. Graceful Termination: At the end of the communication, the method ensures that the client socket is closed properly.

Next, the server loop:

    def server_loop(self):
        server = None
        try:
            if not len(self.target):
                self.target = "0.0.0.0"
            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            server.bind((self.target, self.port))
            server.listen(5)
            print(f"[*] Listening on {self.target}:{self.port}")
            server.settimeout(1.0)
            while self.running:
                try:
                    client_socket, addr = server.accept()
                    print(f"[*] Accepted connection from {addr[0]}:{addr[1]}")
                    
                    client_thread = threading.Thread(
                        target=self.handle_client,
                        args=(client_socket,)
                    )
                    client_thread.daemon = True
                    self.threads.append(client_thread)
                    client_thread.start()
                except socket.timeout:
                    continue
                except Exception as e:
                    if self.running:
                        print(f"[!] Exception in server_loop: {str(e)}")
                    break
        except Exception as e:
            print(f"[!] Failed to create server: {str(e)}")
        finally:
            if server:
                try:
                    server.close()
                    print("[*] Server socket closed")
                except:
                    pass

            for thread in self.threads:
                try:
                    thread.join(timeout=1.0)
                except threading.ThreadError:
                    pass

The server_loop() method sets up a TCP server that listens for client connections on the specified target and port. It accepts incoming connections, spawns a new thread to handle each client with the handle_client method, and manages exceptions. The server continues running until stopped, then shuts down properly by closing the server socket and waiting for all threads to finish.

    def client_sender(self, buffer):
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            print(f"[*] Connecting to {self.target}:{self.port}")
            client.connect((self.target, self.port))
            if len(buffer):
                try:
                    client.send(buffer.encode('utf-8'))
                except (BrokenPipeError, ConnectionResetError):
                    print("[!] Failed to send initial buffer - connection lost")
                    return
            while self.running:
                try:
                    # Receive response from server
                    recv_len = 1
                    response = b''
                    while recv_len:
                        data = client.recv(4096)
                        recv_len = len(data)
                        response += data
                        if recv_len < 4096:
                            break
                    if response:
                        print(response.decode('utf-8'), end='')
                    # Get next command
                    buffer = input()
                    if not self.running:
                        break
                    if buffer.lower() in ['exit', 'quit']:
                        break
                    buffer += "\n"
                    try:
                        client.send(buffer.encode('utf-8'))
                    except (BrokenPipeError, ConnectionResetError):
                        print("\n[!] Failed to send data - connection lost")
                        break
                except ConnectionResetError:
                    print("\n[!] Connection reset by peer")
                    break
                except BrokenPipeError:
                    print("\n[!] Broken pipe - connection lost")
                    break
                except EOFError:
                    print("\n[!] EOF detected - exiting")
                    break
                except Exception as e:
                    print(f"\n[!] Exception in client loop: {str(e)}")
                    break
        except socket.error as exc:
            print("\n[!] Exception! Exiting.")
            print(f"[!] Caught exception socket.error: {exc}")
        finally:
            print("[*] Closing connection")
            try:
                client.close()
            except:
                pass

The client_sender() method establishes a TCP connection to a target server, sends an initial buffer (if provided), and enters a loop where it continuously sends user input to the server and prints the server's responses. It handles exceptions for connection issues and ensures that the connection is closed properly when the loop ends or an error occurs.

def main():
    if len(sys.argv[1:]) == 0:
        print("Custom Netcat")
        print("\nSYNOPSIS")
        print("    netcat.py [OPTIONS...]\n")
        print("OPTIONS")
        print("    -l, --listen              Start server in listening mode on specified host:port")
        print("    -e, --execute=<file>      Execute specified file upon connection establishment")
        print("    -c, --command             Initialize an interactive command shell session")
        print("    -u, --upload=<path>       Upload file to specified destination path on connection")
        print("    -t, --target=<host>       Specify target hostname or IP address")
        print("    -p, --port=<port>         Specify target port number")
        print()
        sys.exit(0)

    try:
        opts, args = getopt.getopt(sys.argv[1:], "hle:t:p:cu:",
                                   ["help", "listen", "execute", "target",
                                    "port", "command", "upload"])
        
        for o, a in opts:
            if o in ("-h", "--help"):
                main()
            elif o in ("-l", "--listen"):
                toolkit.listen = True
            elif o in ("-e", "--execute"):
                toolkit.execute = a
            elif o in ("-c", "--command"):
                toolkit.command = True
            elif o in ("-u", "--upload"):
                toolkit.upload_destination = a
            elif o in ("-t", "--target"):
                toolkit.target = a
            elif o in ("-p", "--port"):
                toolkit.port = int(a)
            else:
                assert False, "Unhandled Option"

    except getopt.GetoptError as err:
        print(str(err))
        main()

    signal.signal(signal.SIGINT, toolkit.signal_handler)
    signal.signal(signal.SIGTERM, toolkit.signal_handler)

    try:
        if not toolkit.listen and len(toolkit.target) and toolkit.port > 0:
            buffer = sys.stdin.read()
            toolkit.client_sender(buffer)

        if toolkit.listen:
            toolkit.server_loop()
    except KeyboardInterrupt:
        print("\n[*] User requested shutdown")
    except Exception as e:
        print(f"\n[!] Unexpected error: {str(e)}")
    finally:
        toolkit.running = False
        print("[*] Shutdown complete")
        sys.exit(0)

if __name__ == "__main__":
    toolkit = NetCat("", 0)
    main()

The main() function parses command-line arguments, sets up options for running the NetCat tool in different modes, and starts either a client or server based on the provided flags. Here's a breakdown of its steps:

  1. Argument Parsing:
    • If no arguments are provided, it displays usage information for the tool.
    • It then uses getopt to parse command-line options, such as -l for listening mode, -e for executing a file, -c  for command shell mode, -u for uploading a file, -t for specifying the target, and -p for the port.
  2. Signal Handling:
    • It sets up signal handlers for graceful shutdown when the user sends an interrupt or termination signal (SIGINT or SIGTERM).
  3. Running the Client or Server:
    • If --listen is not set and a valid target and port are provided, it reads data from stdin and sends it as a buffer to the target server using client_sender.
    • If --listen is enabled, it runs the server in listening mode using server_loop.
  4. Again, Graceful Shutdown:
    • The program handles KeyboardInterrupt for a user-initiated shutdown and any unexpected errors with a catch-all exception handler. 
    • It ensures the NetCat tool shuts down gracefully by setting toolkit.running to False and exiting cleanly.

Running Our Code

Note: I am going to be running my code from the same Kali machine. However, if you have a Unix-based host machine, I strongly advise you to use it. This is because of the commands we're going to run. In our code, we have a line output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True)

Because we set shell=True, the consistency may not be the same on Windows. 

On Unix/Linux: Commands like ls, ps, cat are built into the shell, while on Windows, commands like dir, type, echo are built into cmd.exe. This creates inconsistency between platforms.

This means that to test this program on Windows, you would need to set shell=False and specify Windows commands. However, that's beyond the scope of this program. 

As a penetration tester, you'll mostly be operating unix-based systems anyway. 

You can check out our reverse shell tutorial to see how to specify custom OS commands. 

Don't worry, my demonstration isn't going to confuse you as I'll be running the program from two different directories within Kali, but the same concepts apply in a real world scenario. 

So, open up a terminal, navigate to the directory where you have your code saved. Copy your Python file to another directory. One of the scripts will serve as the listener while the other will serve as the client side. 

First Test Case

Command for the (Listener). This is for testing the system commands:

$ python3 netcat.py -t 192.168.17.128 -p 8080 -l -c

Result:

The Listener is listening. Next, we run the client side. Notice that the directories will be different but the same code.

Result:

Notice that the Pwd command gives us the directory of where we ran the listener. You can use "quit", "exit" or Ctrl + C to exit the program.

Second Test Case

Now that we have seen an example with basic system commands, let's explore how we can execute files on the target system (using the -e flag). For this test case, we'll try to view the content of the /etc/passwd.

/etc/passwd is a system file on Unix-like operating systems that stores essential information about all user accounts on the system, such as usernames, user IDs, group IDs, home directories, and default shells.

To do this, run the following:

Command(Listener):

$ python3 netcat.py -t 192.168.17.128 -p 8080 -l -e "cat /etc/passwd"

 Command (Client):

$ python3 netcat.py -t 192.168.17.128 -p 8080

As always, you run the listener before the client side.

Result:

As you can see, we were able to obtain the contents of the /etc/passwd file.

Third Test Case

For the third and final test case, we'll combine our program with the actual Netcat program. We'll again try to view the contents of the /etc/passwd however this time, instead of using our program as the client, we'll use Netcat. 

Command (Listener) will be same as before:

$ python3 netcat.py -t 192.168.17.128 -p 8080 -l -e "cat /etc/passwd"

Command (Client):

$ nc 192.168.17.128 8080

Result:

We have the same result as before even though we used Netcat this time. I'm sure now we can see how important this tutorial is. 

Conclusion

We have seen various use cases and the importance of netcat in Networking and Cybersecurity. Feel free to test the other features that were not demonstrated in this tutorial (like File Upload). You can also modify it and add other features you want. That's the point of building ours!

Check out similar tutorials:

I hope you enjoyed this one. Till next time!

Why juggle between languages when you can convert? Check out our Code Converter. Try it out today!

View Full Code Assist My Coding
Sharing is caring!



Read Also



Comment panel

    Got a coding query or need some guidance before you comment? Check out this Python Code Assistant for expert advice and handy tips. It's like having a coding tutor right in your fingertips!