Code for How to Build a Custom Netcat with Python Tutorial

View on Github

netcat.py

import sys, socket, getopt, threading, subprocess, signal, time 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 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 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 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 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()