diff --git a/fake_cups_server.py b/fake_cups_server.py new file mode 100644 index 0000000..03e3c4b --- /dev/null +++ b/fake_cups_server.py @@ -0,0 +1,82 @@ +import socket +import threading + +# Fake CUPS Server Configuration +FAKE_CUPS_VERSION = "CUPS/2.4.10-2+b1" +FAKE_PRINTER_NAME = "HP Color LaserJet MFP M478" +FAKE_PRINTER_STATE = "3" # (3 = Idle, 4 = Processing, 5 = Stopped) +FAKE_PRINTER_JOBS = [] + +# Function to parse incoming IPP requests +def parse_ipp_request(data): + if b"GET /" in data or b"HEAD /" in data: # Web-based request + return f"HTTP/1.1 200 OK\r\nServer: {FAKE_CUPS_VERSION}\r\nContent-Type: text/html\r\n\r\n{FAKE_PRINTER_NAME}" + + if b"POST /" in data: # IPP print request + return handle_ipp_request(data) + + return f"HTTP/1.1 400 Bad Request\r\nServer: {FAKE_CUPS_VERSION}\r\n\r\n" + +# Function to handle IPP protocol requests +def handle_ipp_request(data): + if b"operation-id=0x0002" in data: # Get-Printers request + return f"""HTTP/1.1 200 OK +Server: {FAKE_CUPS_VERSION} +Content-Type: application/ipp +Content-Length: 200 + +\x02\x00\x00\x00\x00\x01\x01\x00\x00\x00\x00\x00 +@attribute charset utf-8 +@attribute naturalLanguage en +@printer-name {FAKE_PRINTER_NAME} +@printer-state {FAKE_PRINTER_STATE} +""" + + if b"operation-id=0x000B" in data: # Print-Job request + FAKE_PRINTER_JOBS.append("Job Received") + return f"""HTTP/1.1 200 OK +Server: {FAKE_CUPS_VERSION} +Content-Type: application/ipp +Content-Length: 120 + +\x02\x00\x00\x00\x00\x01\x01\x00\x00\x00\x00\x00 +@job-id {len(FAKE_PRINTER_JOBS)} +@job-state 3 +""" + + return f"HTTP/1.1 400 Bad Request\r\nServer: {FAKE_CUPS_VERSION}\r\n\r\n" + +# Function to handle incoming CUPS/IPP connections +def handle_cups_client(client_socket): + print(f"[+] Connection on port 631 (IPP)") + + try: + data = client_socket.recv(1024) + if not data: + return # No data received + + response = parse_ipp_request(data) + + client_socket.sendall(response.encode()) + + except Exception as e: + print(f"[-] Error handling IPP request: {e}") + + finally: + client_socket.close() + +# Function to start the fake CUPS IPP service +def start_fake_cups_service(): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(("0.0.0.0", 631)) + server_socket.listen(5) + print("[*] Fake CUPS IPP service running on port 631") + + while True: + client_socket, addr = server_socket.accept() + threading.Thread(target=handle_cups_client, args=(client_socket,)).start() + +# Ensure script can be run standalone OR imported +if __name__ == "__main__": + start_fake_cups_service() diff --git a/fake_http_server.py b/fake_http_server.py new file mode 100644 index 0000000..0f86c33 --- /dev/null +++ b/fake_http_server.py @@ -0,0 +1,79 @@ +import socket +import threading +import os + +# Server Configuration +HTTP_PORT = 80 +AUTH_REALM = "HP Color LaserJet Pro MFP M478" +PRINTER_NAME = "HP Color LaserJet Pro MFP M478" +SERVER_BANNER = "uhttpd/1.0.0" +FAVICON_PATH = os.path.join(os.path.dirname(__file__), "favicon.ico") + +# Function to generate HTTP 401 Unauthorized response (always prompts) +def http_401_unauthorized(): + return f"""HTTP/1.1 401 Unauthorized +Server: {SERVER_BANNER} +WWW-Authenticate: Basic realm="{AUTH_REALM}" +Content-Type: text/html +Content-Length: 124 + +{PRINTER_NAME} +

401 Unauthorized

Authentication required.

+""".replace("\n", "\r\n").encode() + +# Function to serve favicon.ico +def serve_favicon(): + if os.path.exists(FAVICON_PATH): + with open(FAVICON_PATH, "rb") as f: + favicon_data = f.read() + return f"""HTTP/1.1 200 OK +Server: {SERVER_BANNER} +Content-Type: image/x-icon +Content-Length: {len(favicon_data)} + +""".replace("\n", "\r\n").encode() + favicon_data + else: + return f"""HTTP/1.1 404 Not Found +Server: {SERVER_BANNER} +Content-Length: 90 + +{PRINTER_NAME} +

404 Not Found

+""".replace("\n", "\r\n").encode() + +# Function to handle incoming HTTP requests +def handle_http_client(client_socket): + try: + data = client_socket.recv(1024).decode(errors="ignore") + if not data: + return + + # Serve favicon.ico + if "GET /favicon.ico" in data: + client_socket.sendall(serve_favicon()) + return + + # Respond with HTTP 401 Unauthorized, always prompting for authentication + client_socket.sendall(http_401_unauthorized()) + + except Exception as e: + print(f"[-] HTTP Error: {e}") + + finally: + client_socket.close() + +# Function to start the fake HTTP auth server +def start_fake_http_auth(): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(("0.0.0.0", HTTP_PORT)) + server_socket.listen(5) + print(f"[*] Fake HTTP server running on port {HTTP_PORT}") + + while True: + client_socket, addr = server_socket.accept() + threading.Thread(target=handle_http_client, args=(client_socket,)).start() + +# Ensure script can be run standalone OR imported +if __name__ == "__main__": + start_fake_http_auth() diff --git a/fake_lpd_server.py b/fake_lpd_server.py new file mode 100644 index 0000000..27ee0dc --- /dev/null +++ b/fake_lpd_server.py @@ -0,0 +1,63 @@ +import socket +import threading + +# Fake printer queue state +FAKE_PRINTER_NAME = "HP Color LaserJet Pro MFP M478" +FAKE_JOB_ID = "001" # Simulated print job ID +FAKE_PRINTER_STATUS = "Ready" + +# Function to handle incoming LPD connections +def handle_lpd_client(client_socket): + print(f"[+] Connection on port 515 (LPD)") + + try: + data = client_socket.recv(1024) + + if not data: + return # No data received, close connection + + command = data[0] # LPD command is the first byte + + if command == 0x02: # "\x02" - Receive job + response = b"\x00" # Acknowledge job request + print("[*] Received LPD job request, sending ACK") + + elif command == 0x05: # "\x05" - Request queue status + response = f"Printer: {FAKE_PRINTER_NAME}\nQueue: Empty\nStatus: {FAKE_PRINTER_STATUS}\r\n".encode() + print("[*] Sent LPD queue status") + + elif command == 0x03: # "\x03" - Receive control file + response = b"\x00" # Acknowledge control file receipt + print("[*] Received LPD control file, sending ACK") + + elif command == 0x04: # "\x04" - Receive print data file + response = b"\x00" # Acknowledge data file receipt + print("[*] Received LPD print data, sending ACK") + + else: + response = b"\x00" # Default acknowledgment + print(f"[*] Received unknown LPD command {hex(command)}, sending generic ACK") + + client_socket.sendall(response) + + except Exception as e: + print(f"[-] Error handling LPD request: {e}") + + finally: + client_socket.close() + +# Function to start the fake LPD service +def start_fake_lpd_service(): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(("0.0.0.0", 515)) + server_socket.listen(5) + print("[*] Fake HP LPD service running on port 515") + + while True: + client_socket, addr = server_socket.accept() + threading.Thread(target=handle_lpd_client, args=(client_socket,)).start() + +# Ensure script can be run standalone OR imported +if __name__ == "__main__": + start_fake_lpd_service() diff --git a/fake_pjl_server.py b/fake_pjl_server.py new file mode 100644 index 0000000..7d7ad80 --- /dev/null +++ b/fake_pjl_server.py @@ -0,0 +1,118 @@ +import socket +import threading +import uuid + +# Function to get local IP address +def get_local_ip(): + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + return s.getsockname()[0] + except Exception: + return "192.168.1.15" # Fallback IP if detection fails + +# Function to get local MAC address +def get_mac_address(): + try: + mac = uuid.getnode() + mac_address = ":".join(f"{(mac >> i) & 0xFF:02X}" for i in range(40, -1, -8)) + return mac_address.replace(":", "").upper() # Remove colons for printer-style format + except Exception: + return "0025B3EDFFD0" # Fallback MAC + +# Get dynamic IP and MAC for response +LOCAL_IP = get_local_ip() +LOCAL_MAC = get_mac_address() + +# PJL Responses +def get_pjl_response(command): + responses = { + "@PJL INFO ID": "HP LaserJet 4\r\n", + "@PJL INFO STATUS": "CODE=10000 READY\r\n@PJL OK\r\n", + "@PJL INFO CONFIG": """@PJL INFO CONFIG +DefaultPaper = A4 +PrintResolution = 600 +Duplex = OFF +@PJL OK\r\n""", + "@PJL INFO VARIABLES": """@PJL INFO VARIABLES +DEFAULT PAPER=A4 +DEFAULT RESOLUTION=600 +DEFAULT COPIES=1 +@PJL OK\r\n""", + "@PJL INFO MEMORY": "TOTAL=8388608 AVAILABLE=4993912\r\n@PJL OK\r\n", + "@PJL INFO FILESYS": """@PJL INFO FILESYS +Filesystem=RAMDISK +Free=4993912 +Total=8388608 +@PJL OK\r\n""", + "@PJL USTATUS": "USTATUS OFF\r\n@PJL OK\r\n", + "@PJL USTATUS TIMED": "USTATUS TIMED=OFF INTERVAL=0\r\n@PJL OK\r\n", + "@PJL USTATUS PAGE": "USTATUS PAGE=ON\r\n@PJL OK\r\n", + "@PJL USTATUS DEVICE": "USTATUS DEVICE=ON\r\n@PJL OK\r\n", + "@PJL DEFAULT PAPER": "DEFAULT PAPER=A4\r\n@PJL OK\r\n", + "@PJL DEFAULT RESOLUTION": "DEFAULT RESOLUTION=600\r\n@PJL OK\r\n", + "@PJL RESET": "\r\n" + } + + if "@PJL INFO PRODINFO" in command: + return f"""@PJL INFO PRODINFO +ProductName = HP Color LaserJet Pro MFP M478 +FormatterNumber = Q910CHL +PrinterNumber = Q1234A +ProductSerialNumber = VNB4G64636 +ServiceID = 20127 +FirmwareDateCode = 20241211 +MaxPrintResolution = 600 +ControllerNumber = Q910CHL +DeviceDescription = HP Color LaserJet Pro MFP M478 +DeviceLang = ZJS PJL ACL HTTP +TotalMemory = 8388608 +AvailableMemory = 4993912 +Personality = 7 +EngFWVer = 15 +IPAddress = {LOCAL_IP} +HWAddress = {LOCAL_MAC} +""" + + return responses.get(command.strip(), None) # Strip for cleaner comparisons + +# Function to handle incoming PJL connections +def handle_pjl_client(client_socket): + print(f"[+] Connection on port 9100 (JetDirect)") + + buffer = "" + + while True: + try: + data = client_socket.recv(1024).decode(errors="ignore") + if not data: + break # Connection closed + + buffer += data + if "\n" in buffer: # Ensure we received a full command + response = get_pjl_response(buffer.strip()) + buffer = "" # Clear the buffer after processing + + if response: + client_socket.sendall(response.encode()) + except Exception as e: + print(f"[-] Error handling PJL request: {e}") + break + + client_socket.close() + +# Function to start the fake JetDirect service +def start_fake_pjl_service(): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(("0.0.0.0", 9100)) + server_socket.listen(5) + print("[*] Fake HP JetDirect service running on port 9100") + + while True: + client_socket, addr = server_socket.accept() + threading.Thread(target=handle_pjl_client, args=(client_socket,)).start() + +# Ensure it can be run standalone OR imported +if __name__ == "__main__": + start_fake_pjl_service()