Before we get started, have you tried our new Python Code Assistant? It's like having an expert coder at your fingertips. Check it out!
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:
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.
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
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 functionssocket
– For low-level networking interfacegetopt
– 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 threadssubprocess
– For spawning new processes and interacting with themsignal
– For Handling asynchronous events (signals)time
– For Time-related functionsNext, 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:
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.run_command()
method and sends the output back to the client.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.BrokenPipeError
or ConnectionResetError
), and prints appropriate messages if errors occur during data reception, file saving, or command execution.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:
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.SIGINT
or SIGTERM
).--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
.--listen
is enabled, it runs the server in listening mode using server_loop
.KeyboardInterrupt
for a user-initiated shutdown and any unexpected errors with a catch-all exception handler. NetCat
tool shuts down gracefully by setting toolkit.running
to False
and exiting cleanly.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.
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.
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.
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.
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!
Just finished the article? Why not take your Python skills a notch higher with our Python Code Assistant? Check it out!
View Full Code Auto-Generate My Code
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!