Code forHow to Build a Custom Netcat with Python Tutorial

netcat.py

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

Join 50,000+ Python Programmers & Enthusiasts like you!

Subscribe

Tags

New Tutorials

Popular Tutorials