How can I replicate Pwntools-esque: p.sendline(payload) functionality with Qiling? (or any similar functionality)

[jt0dd]

In pwntools, where a program is waiting for user input (via fgets or similar) an exploit might look like:

from pwn import *
p = process("./vuln")
nop_sled = asm(shellcraft.nop() * 100)
p.sendline(nop_sled)

But I want to use Qiling, which I thought might look something like:

from pwn import *
from qiling import *
from qiling.const import QL_VERBOSE
from qiling.os.const import STRING

def my_fgets(ql):
    pass

def sandbox(path, rootfs):
# Setup Qiling engine
    ql = Qiling(path, rootfs, verbose=QL_VERBOSE.DISASM)
    # ql.os.set_api('fgets', my_fgets) # I thought maybe this would be involved, but unsure.
    # now emulate the EXE
    return ql

p = sandbox(["vuln"], "qiling/examples/rootfs/x8664_linux")
p.run()
nop_sled = asm(shellcraft.nop() * 100)
p.sendline(nop_sled)

I tried variations of this, reading through the documentation about hijacking, particularly ql.os.set_api but am unsure how this works. Basically I know what I want to do but don't understand the underlying systems enough to do it.

The reason I want to use Qiling instead of a typical process is it seems much easier to observe state with qiling than a debugger. Also Qiling offers a unique ability to apply conditional hooks. I could do things like:

• Hook all jumps with destination addresses that seem unusual (not the start address of any local or imported subroutine, nor loop). This would be an effective analysis option against ROPing and dynamically executing code.
• Hook any subroutines which accept untrusted external input. This would be effective as part of taint analysis.
• If new data gets made dynamically executable, these hooks still apply instead of me having to go set breakpoints in the dynamic code as I would would with debuggers / IDA / Ghidra, etc.

But qiling runs differently, it doesn't pause to await stdin input on API calls like fgets / gets. Or maybe it does and I just don't know how to send it. So I think I need to implement that behavior, and as stated above I read the documentation on how to do something similar, but am struggling to understand how to go further.

[jt0dd]

Maybe the solution is simpler than I make it out to be. I tried tinkering with various examples from the documentation like:

class MyPipe():
    def __init__(self):
        self.buf = b''

    def write(self, s):
        self.buf += s

    def read(self, size):
        if size <= len(self.buf):
            ret = self.buf[: size]
            self.buf = self.buf[size:]
        else:
            ret = self.buf
            self.buf = ''
        return ret

    def fileno(self):
        return 0

    def show(self):
        pass

    def clear(self):
        pass

    def flush(self):
        pass

    def close(self):
        self.outpipe.close()

    def fstat(self):
        return stdin_fstat

def main(input_file, enable_trace=False):
    stdin = MyPipe()
    ql = Qiling(["./x8664_fuzz"], "../rootfs/x8664_linux",
                stdin=stdin,
                stdout=1 if enable_trace else None,
                stderr=1 if enable_trace else None,
                console = True if enable_trace else False)

but I didn't figure it out.

[elicn]

Hi @jt0dd, Depends on whether you use master branch or dev, taking over the standard streams would be cone a bit differently but they essetntially work the same way. Both require you to create a stream object of your own and which you can control (you may find simple ones on extensions.pipe).

The idea is basically feeding the stream's internal buffer with your input for the program to pick up on its own time. You may do so based on certain circumstances (e.g. inside a hook that fired-up when emulation hit a certain address), or unconditionally (e.g. even before Qiling starts to run).

You may find a farily simple example in examples/crackme_x86_windows_auto.py.

[xwings]

Close for now.

We updated the codebase for Qiling and Unicorn since this issue being posted.

Feel free to try the latest version.