NepCTF-2026

2026-07-18/更新于 2026-07-20

easyDilithium

源码

#!/usr/bin/env python3
"""
Leaky Dilithium - A vulnerable ML-DSA-like signature service.
"""

import hashlib
import os
import random
import sys

# ==== ==== ==== ==== == Parameters == ==== ==== ==== ====
n = 64
q = 65537
k = 2
l = 2
eta = 2
tau = 20
gamma1 = 8192
gamma2 = 256
beta = tau * eta

NOISE_BOUND = 15

TARGET_MSG = "Please give me the flag"
FLAG = os.environ.get("FLAG")


# ==== ==== ==== ==== == Polynomial arithmetic == ==== ==== ==== ====
class Poly:
    """Polynomial in Z_q[x]/(x^n+1), stored with coefficients in 0..q-1."""

    def __init__(self, coeffs=None):
        if coeffs is None:
            self.coeffs = [0] * n
        else:
            self.coeffs = [c % q for c in coeffs]
            if len(self.coeffs) != n:
                raise ValueError(f"Length must be {n}")

    def __add__(self, other):
        return Poly([(a + b) % q for a, b in zip(self.coeffs, other.coeffs)])

    def __sub__(self, other):
        return Poly([(a - b) % q for a, b in zip(self.coeffs, other.coeffs)])

    def __mul__(self, other):
        """Convolution modulo x^n+1 and q."""
        res = [0] * (2 * n)
        for i in range(n):
            if self.coeffs[i] == 0:
                continue
            for j in range(n):
                res[i + j] += self.coeffs[i] * other.coeffs[j]

        final = [0] * n
        for i in range(2 * n - 1):
            idx = i % n
            val = res[i] % q
            if i < n:
                final[idx] = (final[idx] + val) % q
            else:
                final[idx] = (final[idx] - val) % q
        return Poly(final)

    def __eq__(self, other):
        return self.coeffs == other.coeffs

    def to_int_list(self):
        return self.coeffs

    def centered_list(self):
        half = q // 2
        return [c if c <= half else c - q for c in self.coeffs]


def poly_vec_add(v1, v2):
    return [a + b for a, b in zip(v1, v2)]


def matrix_vec_mul(mat, vec):
    res = []
    for row in mat:
        acc = Poly()
        for a, v in zip(row, vec):
            acc = acc + (a * v)
        res.append(acc)
    return res


def poly_vec_scalar_mul(c, vec):
    return [c * v for v in vec]


# ==== ==== ==== ==== == Helpers == ==== ==== ==== ====
def sample_small_poly():
    return Poly([random.randint(-eta, eta) for _ in range(n)])


def sample_uniform_poly():
    return Poly([random.randint(0, q - 1) for _ in range(n)])


def sample_y():
    y = []
    for _ in range(l):
        coeffs = [random.randint(-gamma1 + 1, gamma1) for _ in range(n)]
        y.append(Poly(coeffs))
    return y


def high_bits(r, gamma2):
    r = r % q
    r0 = r % gamma2
    if r0 > gamma2 // 2:
        r0 -= gamma2
    return (r - r0) // gamma2


def poly_high_bits(poly, gamma2):
    return [high_bits(c, gamma2) for c in poly.coeffs]


def serialize_w1(w1_list):
    data = b""
    for coeffs in w1_list:
        for v in coeffs:
            data += v.to_bytes(2, "big")
    return data


def generate_challenge(msg, w1_list):
    data = msg.encode() + serialize_w1(w1_list)
    seed = hashlib.sha256(data).digest()
    rng = random.Random(seed)

    c_coeffs = [0] * n
    positions = rng.sample(range(n), tau)
    for pos in positions:
        c_coeffs[pos] = 1 if rng.randint(0, 1) == 0 else -1
    return Poly(c_coeffs)


# ==== ==== ==== ==== == Key generation == ==== ==== ==== ====
def keygen():
    A = [[sample_uniform_poly() for _ in range(l)] for _ in range(k)]
    s1 = [sample_small_poly() for _ in range(l)]
    t = matrix_vec_mul(A, s1)
    return A, t, s1


# ==== ==== ==== ==== == Signature == ==== ==== ==== ====
def sign(msg, A, s1):
    while True:
        y = sample_y()
        w = matrix_vec_mul(A, y)
        w1_list = [poly_high_bits(p, gamma2) for p in w]
        c = generate_challenge(msg, w1_list)
        cs1 = poly_vec_scalar_mul(c, s1)
        z = poly_vec_add(y, cs1)

        z_centered = [coeff for p in z for coeff in p.centered_list()]
        if max(abs(v) for v in z_centered) >= gamma1 - beta:
            continue

        noise = [
            [random.randint(-NOISE_BOUND, NOISE_BOUND) for _ in range(n)]
            for _ in range(l)
        ]
        r = [Poly([y[i].coeffs[j] + noise[i][j] for j in range(n)]) for i in range(l)]
        return c, z, r


# ==== ==== ==== ==== == Verification == ==== ==== ==== ====
def verify(msg, c, z, A, t):
    z_centered = [coeff for p in z for coeff in p.centered_list()]
    if max(abs(v) for v in z_centered) >= gamma1 - beta:
        return False

    Az = matrix_vec_mul(A, z)
    ct = poly_vec_scalar_mul(c, t)
    w_prime = [Az[i] - ct[i] for i in range(k)]
    w1_prime = [poly_high_bits(p, gamma2) for p in w_prime]
    c_prime = generate_challenge(msg, w1_prime)
    return c == c_prime


# ==== ==== ==== ==== == Server interaction == ==== ==== ==== ====
def main():
    random.seed(os.urandom(8))
    print("=== Leaky Dilithium Signature Service ===", flush=True)
    A, t, s1 = keygen()
    print("[+] Key generated.", flush=True)

    while True:
        print("\nMenu:", flush=True)
        print("1. Get public key", flush=True)
        print("2. Request signature", flush=True)
        print("3. Submit admin signature", flush=True)
        print("0. Exit", flush=True)
        choice = input("> ").strip()

        if choice == "1":
            print("Public key (A, t):", flush=True)
            print("A:", flush=True)
            for row in A:
                for poly in row:
                    print(poly.to_int_list(), flush=True)
            print("t:", flush=True)
            for poly in t:
                print(poly.to_int_list(), flush=True)

        elif choice == "2":
            msg = input("Message to sign: ").strip()
            if msg == TARGET_MSG:
                print("[-] Sorry, cannot sign the target message!", flush=True)
                continue

            c, z, r = sign(msg, A, s1)
            print("Signature:", flush=True)
            print("c:", c.coeffs, flush=True)
            print("z:", flush=True)
            for poly in z:
                print(poly.centered_list(), flush=True)
            print("r (debug):", flush=True)
            for poly in r:
                print(poly.centered_list(), flush=True)

        elif choice == "3":
            print("Submit your signature for: " + TARGET_MSG, flush=True)
            msg = TARGET_MSG
            print("Enter c (list of int, length 64):", flush=True)
            c_data = input().strip()
            try:
                c_coeffs = [int(x) for x in c_data.strip("[]").split(",")]
                c = Poly(c_coeffs)
            except Exception:
                print("Invalid format.", flush=True)
                continue

            z_vec = []
            for i in range(l):
                print(f"Enter z[{i}] (64 ints):", flush=True)
                z_data = input().strip()
                try:
                    z_coeffs = [int(x) for x in z_data.strip("[]").split(",")]
                    z_vec.append(Poly(z_coeffs))
                except Exception:
                    print("Invalid format.", flush=True)
                    break

            if len(z_vec) != l:
                continue

            if verify(msg, c, z_vec, A, t):
                print("[+] Signature valid! Here is your flag: " + FLAG, flush=True)
            else:
                print("[-] Invalid signature.", flush=True)
            break

        elif choice == "0":
            print("Bye!", flush=True)
            sys.exit(0)
        else:
            print("Unknown option.", flush=True)


if __name__ == "__main__":
    main()

分析

服务端实现了三个功能：

打印公钥 $(A, t)$
对某一字符串签名（返回签名 $(c, z)$ 和调试信息 $r$ ）
提交对字符串 TARGET_MSG = Please give me the flag 的签名，成功后获得 flag

但是功能 2 拒绝对目标字符串签名，所以本题的目标就是让我们获取私钥，自己签名。

先看 sign(msg, A, s1) -> c, z, r，通过 params 就可以知道私钥是 s1。细看函数逻辑。y 应该是个随机的中间量，用公钥 A 和随机量 y 生成中间量 w，再对 w 做某些处理得到 w1_list。注意到第一个返回量 c 是把 msg 和 w1_list 通过 gen_chall() 函数拼合在一起得到的。将 c 和私钥 s1 相乘得到中间量 cs1，第二个返回量 $z = y + c \cdot s_{1}$ 。下面的检验逻辑懒得看了。最后的返回量简单写成 $r = y + e$ ，其中 e 是噪声 noise。由此得到关系式：

\begin{matrix} w = A y \\ c = genChall (m s g, highBits (w)) \\ z = y + c \cdot s_{1} \\ r = y + e \end{matrix}

接着看 keygen()，发现了 $A, t, s_{1}$ 之间的关系：

t = A \cdot s_{1}

（实际上，只根据这个式子就能求解私钥，似乎是非预期解法）

最后看 verify(msg, c, z, A, t) -> res: boolean。最后返回的是 c 和 c’是否相等，因此去看 c’的实现。发现和 c 一样，c’也是把 msg 和 w1’通过 gen_chall() 函数拼合在一起。这里的 w1’由 w’处理得到。

\begin{matrix} w^{'} = A z - c t \\ c^{'} = genChall (m s g, highBits (w^{'})) \end{matrix}

最终判断 c == c_prime。

预期解（Gemini）

恢复私钥 $s_{1}$ 的核心思想是利用服务器泄露的调试信息 r，将格密码中的带错误学习（LWE）问题降级为一个普通的超定线性方程组，然后通过最小二乘法（Ordinary Least Squares, OLS）直接求解。

以下是恢复私钥 $s_{1}$ 的详细推导和操作步骤：

1. 提取无模运算的线性等式

在 sign 函数中，合法的签名值 $z$ 计算公式为 $z = y + c \cdot s_{1}$ 。同时，由于漏洞的存在，服务器泄露了带有小噪声的变量 $r = y + e$ 。

将两个等式相减，我们可以消去未知的掩码多项式 $y$ ：

z - r = (y + c \cdot s_{1}) - (y + e) = c \cdot s_{1} - e

在正常情况下，多项式运算是在模 $q$ 下进行的，这会打乱数值的大小关系。但我们可以计算一下各项的极值：

私钥 $s_{1}$ ：系数界限为 $η = 2$ 。
挑战 $c$ ：系数只有非零项 $\pm 1$ ，且非零项个数为 $τ = 20$ 。
乘积 $c \cdot s_{1}$ ：其单一系数绝对值最大为 $20 \times 2 = 40$ 。
噪声 $e$ ：由 NOISE_BOUND = 15 限制，系数在 $[- 15, 15]$ 之间。

因此，多项式 $(c \cdot s_{1} - e)$ 的系数绝对值最大为 $40 + 15 = 55$ 。因为 $55$ 远远小于模数 $q = 65537$ 的一半（ $32768$ ），所以在这个计算过程中 绝对没有发生模运算溢出（Wrap-around）。我们可以直接在实数域中建立等式：

v = z - r = c \cdot s_{1} - e

其中 $v$ 是我们可以计算出的观测向量，目标是求出未知的 $s_{1}$ 。

2. 将多项式乘法转化为矩阵乘法

由于 $s_{1}$ 的每个分量都是长度为 $n = 64$ 的多项式，我们需要将 $c \cdot s_{1}$ 这种多项式乘法改写为矩阵与向量的乘法。

在模 $x^{64} + 1$ 的多项式环中，多项式乘法等价于负循环矩阵（Negacyclic Matrix）的乘法。对于一个给定的挑战多项式 $c = c_{0} + c_{1} x + \dots + c_{63} x^{63}$ ，我们可以构造一个 $64 \times 64$ 的矩阵 $M_{c}$ ：

M_{c} = [\begin{matrix} c_{0} & - c_{63} & \dots & - c_{1} \\ c_{1} & c_{0} & \dots & - c_{2} \\ ⋮ & ⋮ & ⋱ & ⋮ \\ c_{63} & c_{62} & \dots & c_{0} \end{matrix}]

这样，多项式等式 $v = c \cdot s_{1} - e$ 就变成了标准的矩阵方程：

\vec{v} = M_{c} \cdot {\vec{s}}_{1} - \vec{e}

3. 数据收集与矩阵堆叠 (Stacking)

单个签名的 64 个方程不足以精准过滤掉噪声 $e$ 的影响，因此我们需要向服务器发送多次无关的消息，收集多组 $(c, z, r)$ 。

由于私钥 $s_{1}$ 是由 $l = 2$ 个多项式组成的向量，我们需要对 $s_{1}$ 的这两个部分分别进行恢复。
假设我们收集了 $k$ 组签名数据（例如 $k = 20$ 组）。
对于 $s_{1}$ 的某一个多项式分量，我们将这 $k$ 个矩阵 $M_{c_{1}}, M_{c_{2}}, \dots, M_{c_{k}}$ 垂直堆叠，形成一个 $(64 k) \times 64$ 的大矩阵 $M_{s t a c k}$ 。
同样，将对应的观测向量 ${\vec{v}}_{1}, {\vec{v}}_{2}, \dots, {\vec{v}}_{k}$ 堆叠成一个长度为 $64 k$ 的大向量 $V_{s t a c k}$ 。

现在我们得到了一个巨大的超定方程组：

M_{s t a c k} \cdot {\vec{s}}_{1} = V_{s t a c k} + E_{s t a c k}

4. 最小二乘法 (OLS) 求解

由于噪声变量 $e$ 是通过 random.randint(-15, 15) 生成的，它在统计上符合均值为 $0$ 的均匀分布。这种零均值白噪声最适合用最小二乘法进行线性回归拟合。

我们寻找一个 ${\vec{s}}_{1}$ 的估计值，使得误差的平方和最小：

{\hat{s}}_{1} = (M_{s t a c k}^{T} M_{s t a c k})^{- 1} M_{s t a c k}^{T} V_{s t a c k}

在 Python 的 numpy 库中，这可以通过一行代码直接完成：

1	`s1_float, residuals, rank, s = numpy.linalg.lstsq(M_stack, V_stack, rcond=None)`

因为真实的 $s_{1}$ 系数一定是整数，而最小二乘法解出来的是浮点数，我们只需要 将 s1_float 中的每一项四舍五入到最近的整数，即可 $100 %$ 完美还原出原始的私钥 $s_{1}$ 。对 $s_{1}$ 的两个多项式分量分别执行此操作，恢复即告完成。

EXP

这里写一下直接通过 $t = A \cdot s_{1}$ 解私钥的 exp。

工具函数：

from server import Poly, q, n


# 剪切多项式高次项系数为零的项
def trim(poly):
    poly = [x % q for x in poly]
    while len(poly > 1) and poly[-1] == 0:
        poly.pop()
    return poly


# 减法
def sub(a, b):
    size = max(len(a), len(b))
    res = [0] * size
    for i in range(size):
        av = a[i] if i < len(a) else 0
        bv = b[i] if i < len(b) else 0
        res[i] = (av - bv) % q
    return trim(res)


# 乘法
def mul(a, b):
    res = [0] * (len(a) + len(b) - 1)
    for i, av in enumerate(a):
        for j, bv in enumerate(b):
            res[i + j] = (res[i + j] + av * bv) % q
    return trim(res)


# 长除法 返回(商, 余数)
def divmod(dividend, divisor):
    dividend = trim(dividend[:])
    divisor = trim(divisor[:])

    if len(dividend) < len(divisor):
        return [0], dividend

    quotient = [0] * (len(dividend) - len(divisor) + 1)

    divisor_lead_inv = pow(divisor[-1], -1, q)

    while dividend != [0] and len(dividend) >= len(divisor):
        shift = len(dividend) - len(divisor)
        factor = (dividend[-1] * divisor_lead_inv) % q
        quotient[shift] = factor

        for i, coefficient in enumerate(divisor):
            dividend[i + shift] = (dividend[i + shift] - factor * coefficient) % q

        dividend = trim(dividend)

    return trim(quotient), dividend


# exGCD
def invert_ring_poly(poly):
    modulus = [1] + [0] * (n - 1) + [1]

    old_r = modulus
    r = trim(list(poly.coeffs))

    old_t = [0]
    t = [1]

    while r != [0]:
        quotient, remainder = divmod(old_r, r)
        old_r, r = r, remainder
        old_t, t = t, sub(old_t, mul(quotient, t))

    scale = pow(old_r[0], -1, q)

    inverse = [(coefficient * scale) % q for coefficient in old_t]

    inverse = divmod(inverse, modulus)[1]
    inverse += [0] * (n - len(inverse))
    return Poly(inverse)

恢复私钥：

def recover_s1(A, t):
    a, b = A[0][0], A[0][1]
    c, d = A[1][0], A[1][1]
    t0, t1 = t[0], t[1]

    det = (a * d) - (b * c)

    det_inv = invert_ring_poly(det)

    s0 = det_inv * ((d * t0) - (b * t1))
    s1 = det_inv * ((a * t1) - (c * t0))

    return [s0, s1]

伪造签名：

from server import sign, TARGET_MSG

s = recover_s1(A, t)
c, z, r = sign(TARGET_MSG, A, s)
print(c, z)

ezRSA3

源码

from sage.all import gcd, is_prime, prod, proof
from Crypto.Util.number import bytes_to_long, getPrime
from Crypto.Random.random import sample
from secret import flag

proof.arithmetic(False)

e = 65537
n = 10000
k = 10

s = set()
while len(s) < n:
    s.add(getPrime(50))

sops = sorted(s)


def get_p():
    while True:
        p = 2 * prod(sample(sops, k)) - 1
        if is_prime(p) and gcd(p - 1, e) == 1:
            return p


def get_q():
    while True:
        q = getPrime(512)
        if gcd(q - 1, e) == 1:
            return q


p = get_p()
q = get_q()
N = p * q
m = bytes_to_long(flag)
assert m < N
c = pow(m, e, N)

print(f"{sops=}")
print(f"{N=}")
print(f"{c=}")

分析

题目首先生成了 10000 个小素数（每个 50bits）的集合用于生成大素数 p。具体公式为 $p = 2 \prod_{i = 1}^{10} P_{i} - 1$ （ $P_{i}$ 是被抽到的小素数，共 10 个），若最终的 p 不是素数就重新采样生成。

对生成公式变形，有 $p + 1 = 2 \prod_{i = 1}^{10} P_{i}$ 。其中 2 也是一个小素数，所以我们可以说 $p + 1$ 光滑（由 11 个小素数相乘得到）。调用 Williams 算法就能得到 p，进而分解 n。

Williams 算法

当 N 的某个素因子 p 满足(p+1)可以分解成若干较小素数的乘积时，该算法可能很快找到 p。在此算法中，Williams 引入了卢卡斯序列。给定一个整数 $A$ ，我们定义序列 $V_{k}$ ，满足：

$V_{0} = 2$
$V_{1} = A$
$V_{k} = A \cdot V_{k - 1} - V_{k - 2}$

且有性质：若判别式 $Δ = A^{2} - 4$ 在模 p 意义下是一个二次非剩余（不存在任何整数 $x$ 使得 $x^{2} \equiv Δ (\mod p)$ ），那么卢卡斯序列的第 $p + 1$ 项必定满足 $V_{p + 1} \equiv 2 (\mod p)$ 。

完整执行步骤：

随机选择一个整数 $A$ （当 $A = 2$ 时，卢卡斯序列会退化，所以一般从 3 开始遍历）。我们希望 $A^{2} - 4$ 是模 p 的二次非剩余。由于本步可能失败，所以需要枚举。
选择一个平滑限 $B$ ，计算 $B$ 以内所有素数幂的乘积（或直接计算 $B!$ ）。若 $p + 1$ 的所有素因子都小于 $B$ ，那么必定有 $E = k (p + 1)$ 。
根据卢卡斯数列的性质，由于 $E = k (p + 1)$ ，那么有 $V_{E} \equiv 2 (\mod p)$ ，也即 $V_{E} - 2 = m p$ 。
最终通过 $gcd (V_{E} - 2, N)$ 分解素数 $p$ 。

可以利用卢卡斯序列的性质（ $Trace (M^{k}) = V_{k}$ ）和矩阵快速幂加速计算 🥰

EXP

sops = []
N = 
c = 
e = 65537

Zn = Zmod(N)

for A in range(3, 20):
    M = matrix(Zn, [[A, -1], [1, 0]])
    M = M ^ 2

    for P in sops:
        M = M ^ P

    tr = M.trace()
    p = gcd(int(tr) - 2, N)

    if N % p == 0 and p > 1 and p < N:
        print(p)

LGC Attack

源码

import random
from Crypto.Util.number import getPrime, bytes_to_long


def generate_challenge():
    bits = 1024
    p = getPrime(bits)
    q = getPrime(bits)
    N = p * q

    hidden_p_bits = 502
    p_high = p >> hidden_p_bits

    a = random.randint(2, p - 1)
    c = random.randint(2, p - 1)

    flag = flag.ljust(64, b"\x00")
    seed = bytes_to_long(flag)
    assert seed < p

    def lcg(state):
        return a * (state - c) % p

    states = [seed]
    for _ in range(5):
        states.append(lcg(states[-1]))

    k = 256
    outputs = [s >> k for s in states]

    print("===== The Coppersmith & LLL Forge =====")
    print(f"N = {N}")
    print(f"p_high = {p_high}")
    print(f"a = {a}")
    print(f"c = {c}")
    print(f"outputs = {outputs}")


if __name__ == "__main__":
    generate_challenge()

分析

思路很明确，前半段是 copper，后半段是 LCG 演化的格密码问题。前半段的主要问题就在于未知比特太多了，sage 原生的 small_roots 很难直接恢复（或者直接闪退了，不知道怎么回事）。后来我尝试爆破中间的 12 位，压缩一下未知数长度，跑了几个小时也能出来，但最快的方法还是让 chatgpt 搓 copper 的格基（

恢复 p 之后，通过 LCG 的关系式整理，得到

s_{i} = a^{i} s_{0} - c (a + a^{2} + \dots + a^{i}) (\mod p)

记 $B = 2^{256}, H_{i} = B \cdot o u t p u t s$ ，那么 $s_{i} = H_{i} + e_{i}$ （特别地，当 $i = 0$ 时，记为 $s_{0} = H_{0} + x$ ， $x = f l a g$ ）。代入整理：

H_{i} + e_{i} \equiv a^{i} (H_{0} + x) - c (a + a^{2} + \dots + a^{i}) (\mod p)

移项，未知数全移到左边，已知量放在右边：

a^{i} x - e_{i} \equiv H_{i} - (a^{i} H_{0} - c (a + a^{2} + \dots + a^{i})) (\mod p)

记 $d_{i} = H_{i} - (a^{i} H_{0} - c (a + a^{2} + \dots + a^{i})) (\mod p)$ （已知量），提出模数，则最终化简为：

a_{i} x - k_{i} p - d_{i} \cdot 1 = e_{i}

把未知的系数和 1 汇总为向量：

(- k_{1}, - k_{2}, \dots, - k_{n}, x, 1)

构建目标向量，让 $e_{i}, x, 1$ 参与最终的长度约束：

V = (e_{1}, e_{2}, e_{3}, e_{4}, e_{5}, x, 1)

得到格基矩阵：

M = (\begin{matrix} p & 0 & 0 & \dots & 0 & 0 \\ 0 & p & 0 & \dots & 0 & 0 \\ ⋮ & ⋮ & ⋱ & ⋮ & ⋮ & ⋮ \\ 0 & 0 & 0 & p & 0 & 0 \\ a_{1} & a_{2} & a_{3} & \dots & 1 & 0 \\ - d_{1} & - d_{2} & - d_{3} & \dots & 0 & 1 \end{matrix})

做一下平衡，把目标向量和格基矩阵的最后一位换为 W，LLL 约简后就能提取出 x。

EXP

第一阶段（ChatGPT）

import ast
import math
import re

from fpylll import IntegerMatrix, LLL
from sage.all import PolynomialRing, ZZ

text = open("ce_shi_zhuan_yong_.py").read()
N = int(re.findall(r"N = (\d+)", text)[-1])
p_high = int(re.findall(r"p_high = (\d+)", text)[-1])
a = int(re.findall(r"a = (\d+)", text)[-1])
c = int(re.findall(r"c = (\d+)", text)[-1])
outputs = ast.literal_eval(re.findall(r"outputs = (\[.*\])", text)[-1])

# Coppersmith: f(x) = (p_high << 502) + x, with |x| < X.
A = p_high << 502
X = 1 << 502
m, t = 25, 25
dim = m + t + 1
M = IntegerMatrix(dim, dim)

for i in range(m):
    # N^(m-i) * (x+A)^i
    scale = N ** (m - i)
    for j in range(i + 1):
        M[i, j] = scale * math.comb(i, j) * A ** (i - j) * X**j

for j in range(t + 1):
    # x^j * (x+A)^m
    row = m + j
    for k in range(m + 1):
        M[row, k + j] = math.comb(m, k) * A ** (m - k) * X ** (k + j)

# A lower delta keeps this near-boundary instance lightweight enough locally.
LLL.reduction(M, delta=0.75)
R = PolynomialRing(ZZ, "x")
x = R.gen()
p = None
for row in range(12):
    h = sum(ZZ(M[row, j]) // X**j * x**j for j in range(dim) if M[row, j])
    for root, _ in h.roots():
        candidate = A + int(root)
        if 0 <= root < X and N % candidate == 0:
            p = candidate
            break
    if p is not None:
        break

assert p is not None
print(f"p = {p}")

第二阶段

from task import p, c, outputs, a as a_raw

a = []
for i in range(1, 6):
    temp = pow(a_raw, i, p)
    a.append(temp)

H = [x * W for x in outputs]
d = []
W = 2 ^ (256)


for i in range(5):
    t1 = (a[i] * H[0]) % p
    sum_a = 0
    for j in range(1, i + 1):
        sum_a = (sum_a + pow(a_raw, j, p)) % p
    d.append((H[i] - (t1 - c * sum_a)) % p)


L = Matrix(ZZ, 7, 7)
for i in range(5):
    L[i, i] = p
    L[5, i] = a[i]
    L[6, i] = -d[i]

L[5, 5] = 1
L[6, 6] = W

L = L.LLL()
print(L[0][-2])

true_ezgame

源码

from Crypto.Util.number import getPrime, inverse, GCD
from hashlib import sha256, sha512
from secret import FLAG
import secrets
import socketserver
import signal

banner = r"""
 ██████   █████                      █████████  ███████████ ███████████
░░██████ ░░███                      ███░░░░░███░█░░░███░░░█░░███░░░░░░█
 ░███░███ ░███   ██████  ████████  ███     ░░░ ░   ░███  ░  ░███   █ ░ 
 ░███░░███░███  ███░░███░░███░░███░███             ░███     ░███████   
 ░███ ░░██████ ░███████  ░███ ░███░███             ░███     ░███░░░█   
 ░███  ░░█████ ░███░░░   ░███ ░███░░███     ███    ░███     ░███  ░    
 █████  ░░█████░░██████  ░███████  ░░█████████     █████    █████      
░░░░░    ░░░░░  ░░░░░░   ░███░░░    ░░░░░░░░░     ░░░░░    ░░░░░       
                         ░███                                          
                         █████                                         
                        ░░░░░                                          
  ████████     █████     ████████   ████████                           
 ███░░░░███  ███░░░███  ███░░░░███ ███░░░░███                          
░░░    ░███ ███   ░░███░░░    ░███░███   ░░░                           
   ███████ ░███    ░███   ███████ ░█████████                           
  ███░░░░  ░███    ░███  ███░░░░  ░███░░░░███                          
 ███      █░░███   ███  ███      █░███   ░███                          
░██████████ ░░░█████░  ░██████████░░████████                           
░░░░░░░░░░    ░░░░░░   ░░░░░░░░░░  ░░░░░░░░                            



"""

MBIT = 512
ROUNDS = 40
RSA_BITS = 1024
RSA_E = 65537
MAX_INPUT = 32
ROUND_RANDOM_BYTES = 16

ROCK, SCISSORS, PAPER = 0, 1, 2
MOVES = ("rock", "scissors", "paper")
MOVE_TO_ID = {name.encode(): i for i, name in enumerate(MOVES)}


def rand_nonzero_bits(bits: int) -> int:
    while True:
        x = secrets.randbits(bits)
        if x != 0:
            return x

def crt(residues, moduli) -> int:
    a1, a2 = residues
    m1, m2 = moduli
    return (a1 + m1 * (((a2 - a1) * inverse(m1, m2)) % m2)) % (m1 * m2)

def H(i: int, r: bytes) -> int:
    assert 0 <= i <= 2
    return int.from_bytes(sha512(bytes([i]) + r).digest(), "big")


class Commitment:
    def __init__(self, nbits: int = RSA_BITS, e: int = RSA_E):
        assert RSA_BITS % 2 == 0
        while True:
            p = getPrime(nbits // 2)
            q = getPrime(nbits // 2)
            if GCD(p - 1, e) == 1 and GCD(q - 1, e) == 1:
                break

        self.p, self.q, self.e = p, q, e
        self.n = p * q
        self.dp = inverse(e, p - 1)
        self.dq = inverse(e, q - 1)

    def parameters(self):
        return self.n, self.e

    def _sample_mask(self) -> int:
        while True:
            mask = rand_nonzero_bits(MBIT)
            if GCD(mask, self.n) == 1:
                return mask

    def commit(self, value: int):
        mask = self._sample_mask()
        return pow(mask, self.e, self.n), value ^ mask

    def _recover_mask(self, token: int) -> int:
        rp = pow(token, self.dp, self.p)
        rq = pow(token, self.dq, self.q)
        mask = crt((rp, rq), (self.p, self.q))
        assert pow(mask, self.e, self.n) == token
        return mask

    def open(self, commitment):
        token, masked = commitment
        return masked ^ self._recover_mask(token)


COM = Commitment()


class EZ_GAME(socketserver.BaseRequestHandler):
    def readline(self, prompt: bytes = b"") -> bytes:
        self.send(prompt, newline=False)
        data = b""
        while len(data) < MAX_INPUT:
            ch = self.request.recv(1)
            if not ch or ch in b"\r\n":
                break
            data += ch
        return data.strip()

    def send(self, msg: bytes, newline: bool = True):
        try:
            if newline:
                msg += b"\n"
            self.request.sendall(msg)
        except OSError:
            pass

    def proof_of_work(self) -> bool:
        table = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
        proof = "".join(secrets.choice(table) for _ in range(20)).encode()
        digest = sha256(proof).hexdigest().encode()
        self.send(b"[+] sha256(XXX+" + proof[3:] + b") == " + digest)
        xxx = self.readline(b"[+] Plz Tell Me XXX: ")
        return len(xxx) == 3 and sha256(xxx + proof[3:]).hexdigest().encode() == digest

    @staticmethod
    def beats(player: int, dealer: int) -> bool:
        # rock beats scissors, scissors beats paper, paper beats rock
        return dealer == (player + 1) % 3

    def parse_move(self, data: bytes):
        return MOVE_TO_ID.get(data.strip().lower())

    def play_round(self, idx: int) -> bool:
        r = secrets.token_bytes(ROUND_RANDOM_BYTES)
        dealer = secrets.randbelow(3)
        commitment = COM.commit(H(dealer, r))

        self.send(f"[round {idx}/{ROUNDS}]".encode())
        self.send(f"r = {r.hex()}".encode())
        self.send(b"I have committed to my move. Now your turn.")
        self.send(f"commitment: {commitment}".encode())

        data = self.readline(b"your move [rock/scissors/paper]: ")
        player = self.parse_move(data)
        if player is None:
            self.send(b"Invalid move. Use rock, scissors, or paper.")
            return False

        self.send(f"I played {MOVES[dealer]}.".encode())
        self.send(f"You played {MOVES[player]}.".encode())
        return self.beats(player, dealer)

    def handle(self):
        signal.alarm(100)
        self.send(banner.encode("utf-8"))
        
        if not self.proof_of_work():
            self.request.close()
            return

        self.send(b"Welcome to NepCTF 2026")
        self.send(f"Beat me in RPS game for {ROUNDS} rounds.".encode())
        n, e = COM.parameters()
        self.send(f"parameters: n = {n}, e = {e}".encode())

        for i in range(1, ROUNDS + 1):
            if not self.play_round(i):
                self.send(b"You lose.")
                self.request.close()
                return
            self.send(b"You win this round.")

        self.send(b"You win the game!")
        self.send(b"flag: " + FLAG)
        self.request.close()


class ForkedServer(socketserver.ForkingMixIn, socketserver.TCPServer):
    allow_reuse_address = True


if __name__ == "__main__":
    HOST, PORT = "0.0.0.0", 10001
    print("HOST:PORT " + HOST + ":" + str(PORT), flush=True)
    with ForkedServer((HOST, PORT), EZ_GAME) as server:
        server.serve_forever()

分析

很有趣的题目 😍

连接靶机后首先是一个校验（也可能是个提示，因为这道题本身也需要爆破三种行为），要求爆破出满足指定哈希值的前三位字符。这一步直接写个循环就好。

答对后正式进入题目。大概是一个剪刀石头布的游戏，因为靶机不能直接给出本局的动作（这样就直接泄露了），但又需要保证自己已经出拳了，所以采用了一种基于 RSA 的信任机制。给出了 r, commitment，其中 r 是每轮随机生成的校验码，将其和本轮的手势代码拼接在一起（dealer + r）之后做哈希，接着生成掩码 mask，最终返回 r, commitment = (pow(mask, e, n), hash ^ mask)。

这题的漏洞在于 dealer 只有三个取值。我们枚举三种可能，记 hash = H(dealer, r)，计算 mask = (hash ^ mask) ^ hash，使用开头给出的公钥加密后与 commitment[0] 对照，靶机的出拳索引就对应相等的索引值。提前得到了靶机的信息，获胜自然是不在话下。

用 pwntools 实现 40 轮的自动化。

EXP

from pwn import *
from hashlib import sha256, sha512

HOST = "g9ddyleg-oeje-e7ou-nazd-6a5daa2930675-neptunus.nepctf.com"
PORT = 443


def H(i: int, r: bytes) -> int:
    assert 0 <= i <= 2
    return int.from_bytes(sha512(bytes([i]) + r).digest(), "big")


# 解决哈希校验
def solve_hash(suffix, target_hash):
    def to_bytes(i):
        return chr(i).encode()

    source = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    for i in source:
        bi = to_bytes(i)
        for j in source:
            bj = to_bytes(j)
            for k in source:
                bk = to_bytes(k)
                if sha256(bi + bj + bk + suffix).hexdigest().encode() == target_hash:
                    return bi + bj + bk


def main():
    io = remote(HOST, PORT, ssl=True)
    context.log_level = "debug"

    io.recvuntil(b"sha256(XXX+")
    suffix = io.recvuntil(b")")[:-1]
    io.recvuntil(b"== ")
    target_hash = io.recvline().strip()
    send = solve_hash(suffix, target_hash)
    io.sendlineafter(b"Plz Tell Me XXX: ", send)

    io.recvuntil(b"n = ")
    n = int(io.recvuntil(b",")[:-1])
    e = 65537

    for _ in range(40):
        io.recvuntil(b"r = ")
        r = io.recvline().strip()
        io.recvuntil(b"commitment: ")
        commitment = io.recvline().strip()

        r = bytes.fromhex(r.decode())
        cipher, masked = eval(commitment.decode())

        target_i = -1
        for i in range(3):
            hash = H(i, r)
            mask = masked ^ hash
            if cipher == pow(mask, e, n):
                target_i = i
                break

        MOVES = (b"rock", b"scissors", b"paper")
        my_move = MOVES[(target_i + 2) % 3]

        io.sendlineafter(b"your move [rock/scissors/paper]: ", my_move)

    io.interactive()


if __name__ == "__main__":
    main()

源码

#!/usr/bin/env python3
"""
Vector Blind RAG - Challenge Server
Provides the query oracle for the encrypted RAG system.

Usage: python server.py [port]
Default port: 8080

Requires: server_key.json (generated by generate.py, KEEP SECRET)
          challenge_data.json (generated by generate.py, public)
"""

import json
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn


# ─── Matrix arithmetic over Z_p ──────────────────────────────────────────────

def mat_vec_mul(vec, mat, p):
    """Row vector (1 x k) times matrix (k x m) modulo p. Returns row vector (1 x m)."""
    if not mat:
        return []
    m = len(mat[0])
    result = [0] * m
    for i, vi in enumerate(vec):
        if vi == 0:
            continue
        row = mat[i]
        for j in range(m):
            result[j] = (result[j] + vi * row[j]) % p
    return result


# ─── Query encryption (THE VULNERABLE FUNCTION) ──────────────────────────────

def encrypt_query(q, M1_inv, M2_inv, p):
    """Encrypt a query vector for ASPE-based search.

    Note: query vector is not split before encryption.
    This is a deliberate simplification over the full ASPE protocol
    where q would be randomly split as q = q1 + q2 and each half
    encrypted separately with M1^{-1} and M2^{-1}.
    """
    t1 = mat_vec_mul(q, M1_inv, p)
    t2 = mat_vec_mul(q, M2_inv, p)
    return t1, t2


# ─── Data loading ────────────────────────────────────────────────────────────

def load_data():
    """Load challenge data and server secret keys."""
    try:
        with open("challenge_data.json", "r") as f:
            challenge = json.load(f)
    except FileNotFoundError:
        print("ERROR: challenge_data.json not found. Run generate.py first.")
        sys.exit(1)

    try:
        with open("server_key.json", "r") as f:
            keys = json.load(f)
    except FileNotFoundError:
        print("ERROR: server_key.json not found. Run generate.py first.")
        sys.exit(1)

    p = int(challenge["p"])
    n = challenge["n"]
    database = challenge["database"]

    M1_inv = [[int(x) for x in row] for row in keys["M1_inv"]]
    M2_inv = [[int(x) for x in row] for row in keys["M2_inv"]]

    return p, n, database, M1_inv, M2_inv


# ─── HTTP Handler ────────────────────────────────────────────────────────────

class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
    """HTTP server with threading support for concurrent connections."""
    daemon_threads = True


class RequestHandler(BaseHTTPRequestHandler):
    """Handles /query (POST), /database (GET), and / (GET) endpoints."""

    # These are set by the server after loading data
    P = None
    N = None
    DATABASE = None
    M1_INV = None
    M2_INV = None

    def log_message(self, fmt, *args):
        """Suppress default access log to stderr."""
        pass

    def _send_json(self, status, data):
        body = json.dumps(data).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def _send_error(self, status, message):
        self._send_json(status, {"error": message})

    def do_GET(self):
        if self.path == "/" or self.path == "/health":
            self._send_json(200, {
                "status": "ok",
                "service": "Vector Blind RAG Query Oracle",
                "n": self.N,
            })
        elif self.path == "/database":
            with open("challenge_data.json", "r") as f:
                db = json.load(f)
            self._send_json(200, db)
        else:
            self._send_error(404, "not found")

    def do_POST(self):
        if self.path != "/query":
            self._send_error(404, "not found")
            return

        try:
            content_len = int(self.headers.get("Content-Length", 0))
            body = json.loads(self.rfile.read(content_len))
        except (ValueError, json.JSONDecodeError):
            self._send_error(400, "invalid JSON body")
            return

        q_raw = body.get("q")
        if q_raw is None or not isinstance(q_raw, list):
            self._send_error(400, f"missing 'q' field; must be a list of {self.N} integers")
            return
        if len(q_raw) != self.N:
            self._send_error(400, f"expected q of length {self.N}, got {len(q_raw)}")
            return

        try:
            q = [int(x) for x in q_raw]
        except (ValueError, TypeError):
            self._send_error(400, "q must contain valid integers")
            return

        t1, t2 = encrypt_query(q, self.M1_INV, self.M2_INV, self.P)
        self._send_json(200, {
            "t_q": [[str(x) for x in t1], [str(x) for x in t2]],
        })

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()


# ─── Main ────────────────────────────────────────────────────────────────────

def main():
    p, n, database, M1_inv, M2_inv = load_data()

    RequestHandler.P = p
    RequestHandler.N = n
    RequestHandler.DATABASE = database
    RequestHandler.M1_INV = M1_inv
    RequestHandler.M2_INV = M2_inv

    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
    server = ThreadingHTTPServer(("0.0.0.0", port), RequestHandler)
    print(f"Vector Blind RAG oracle listening on port {port}")
    print(f"  GET  /          - health check")
    print(f"  GET  /database  - download encrypted database")
    print(f"  POST /query      - submit query vector, get encrypted token")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.")
        server.shutdown()


if __name__ == "__main__":
    main()

分析

系统内置了两个矩阵 $M_{1}, M_{2}$ 作为密钥。一篇文本首先被转化为向量 $v$ （这部分我还没了解过，感觉挺神奇）。接着在有限域上，向量被随机拆分为两部分 $v = v_{1} + v_{2}$ ，分别用转置过的矩阵加密，生成密文对 $c = v \cdot M^{T}$ 。想要访问文档需要依靠其向量（文档被 GCM 方式加密过，要先处理向量得到密钥，再解密 c_d 得到原文）。

此外，系统向外暴露查询端口。传入查询向量 q，返回 $t = q \cdot M^{- 1}$ 。

根据矩阵乘法，把 q 构造为 $q = [1, 0, \dots, 0]$ 后传入，得到的就是 $M^{- 1}$ 的第 0 行。由此类推，只需要传入 64 组 q，就能把密钥 M 提取出来。接下来计算 $v_{1} = c_{1} \cdot (M_{1}^{- 1})^{T}$ 和 $v_{2}$ ，加起来后就能得到向量 v，解密就行了。

EXP

数据处理的脚本太容易出错了 😢

import requests
import sympy as sp
from hashlib import sha256
from Crypto.Cipher import AES

url = "https://omzgzwv2-zl0e-tp6o-pfdr-6a5dc4ab30760-neptunus.nepctf.com"
response = requests.get(f"{url}/database")
data = response.json()

p = int(data["p"])
n = int(data["n"])
database = data["database"]

M1_inv = []
M2_inv = []
for i in range(64):
    print(i + 1)
    q = ["0"] * n
    q[i] = "1"

    res = requests.post(f"{url}/query", json={"q": q}).json()
    M1_inv.append([int(x) for x in res["t_q"][0]])
    M2_inv.append([int(x) for x in res["t_q"][1]])

M1_inv_T = sp.Matrix(M1_inv).T
M2_inv_T = sp.Matrix(M2_inv).T

for i in range(21):
    print(i + 1)
    # 转化为行向量 sp 的 Int 类型需要额外转换
    c1 = sp.Matrix([int(x) for x in database[i]["c_v"][0]]).T
    c2 = sp.Matrix([int(x) for x in database[i]["c_v"][1]]).T

    v1 = (c1 * M1_inv_T).applyfunc(lambda x: x % p)
    v2 = (c2 * M2_inv_T).applyfunc(lambda x: x % p)

    v = [(v1[i] + v2[i]) % p for i in range(n)]

    key_bytes = b"".join((int(x) % 2**256).to_bytes(32, "little") for x in v)
    key = sha256(key_bytes).digest()

    enc_doc_hex = database[i]["c_d"]
    enc_doc = bytes.fromhex(enc_doc_hex)

    nonce = enc_doc[:12]
    cipher = enc_doc[12:-16]
    tag = enc_doc[-16:]

    dec = AES.new(key, AES.MODE_GCM, nonce)
    flag = dec.decrypt_and_verify(cipher, tag)
    print(flag.decode())

LeakyRAG

源码

#!/usr/bin/env python3
"""
LeakyRAG CTF — Server
"可搜索加密"的向量数据库，实际上分数泄漏导致向量可被完全重建。
"""
import json
import os
import secrets
from http.server import HTTPServer, BaseHTTPRequestHandler

import numpy as np

DIM = 64
FLAG = os.environ.get("FLAG", "flag{l34ky_v3ct0r_s34rch_1s_n0t_3ncrypt10n}")

# ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ====
# "可搜索加密" Embedding — 公开且确定性
# 比值编码：前 63 维编码字符，第 64 维为参考。
# v [i] / v [63] = exp((char_i - 128) / 64)
# 归一化不改变比值 → 重建向量后本地解码即得 flag。
# ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ====
def embed(text: str) -> np.ndarray:
    data = text.encode()
    v = np.ones(DIM, dtype=np.float64)
    n = min(len(data), DIM - 1)
    for i in range(n):
        ratio = np.exp((data[i] - 128) / 64.0)
        v[i] = ratio
    return v / np.linalg.norm(v)


def decode(v_norm: np.ndarray) -> str:
    """从归一化向量恢复文本"""
    v = np.array(v_norm, dtype=np.float64)
    v = v / np.linalg.norm(v)
    ref = v[-1]
    chars = []
    for i in range(DIM - 1):
        ratio = v[i] / ref
        char_code = int(round(np.log(ratio) * 64 + 128))
        if 32 <= char_code <= 126:
            chars.append(chr(char_code))
        else:
            break
    return ''.join(chars)


# ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ====
# 文档库
# ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ====
def init_docs():
    docs = {}
    # 普通文档
    samples = [
        "The quick brown fox jumps over the lazy dog",
        "Machine learning is revolutionizing the world",
        "Python is a popular programming language",
        "The capital of France is Paris",
        "Quantum computing uses qubits instead of bits",
        "Blockchain is a decentralized ledger technology",
        "Neural networks are inspired by biological neurons",
        "The Earth orbits around the Sun",
        "Encryption is the process of encoding information",
        "Docker containers provide isolated environments",
    ]
    for i, text in enumerate(samples):
        doc_id = f"doc_{i:03d}"
        docs[doc_id] = {
            "text": text,
            "vector": embed(text),
            "protected": False,
        }

    # Flag 文档
    docs["flag_doc"] = {
        "text": FLAG,
        "vector": embed(FLAG),
        "protected": True,
    }

    # 加一些干扰文档
    noise_texts = [
        f"Noise document number {i}: {secrets.token_hex(16)}"
        for i in range(50)
    ]
    for i, text in enumerate(noise_texts):
        doc_id = f"noise_{i:03d}"
        docs[doc_id] = {
            "text": text,
            "vector": embed(text),
            "protected": False,
        }

    return docs


docs = init_docs()


# ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ====
# HTTP Handler
# ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ==== ====
class Handler(BaseHTTPRequestHandler):
    def _json(self, data, status=200):
        body = json.dumps(data).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def _serve_file(self, filename: str, content_type: str):
        try:
            with open(filename, "rb") as f:
                body = f.read()
            self.send_response(200)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        except FileNotFoundError:
            self._json({"error": "Not found"}, 404)

    def _read_body(self):
        length = int(self.headers.get("Content-Length", 0))
        return json.loads(self.rfile.read(length)) if length else {}

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def do_GET(self):
        path = self.path.rstrip("/")

        # Serve HTML pages
        html_pages = {"": "index.html", "/index.html": "index.html",
                     "/search.html": "search.html", "/upload.html": "upload.html",
                     "/docs.html": "docs.html"}
        if path in html_pages:
            self._serve_file(html_pages[path], "text/html")
            return

        if path.startswith("/api/doc/"):
            doc_id = path.split("/api/doc/")[1]
            if doc_id in docs:
                doc = docs[doc_id]
                if doc["protected"]:
                    self._json({"error": "This document is protected by SecureAI encryption"}, 403)
                else:
                    self._json({
                        "doc_id": doc_id,
                        "text": doc["text"],
                        "vector": None,  # "加密保护"
                    })
            else:
                self._json({"error": "Document not found"}, 404)
        elif path == "/api/stats":
            self._json({"total_docs": len(docs), "dim": DIM})
        else:
            self._json({"error": "Not found"}, 404)

    def do_POST(self):
        path = self.path.rstrip("/")
        body = self._read_body()

        if path == "/api/search":
            vec = np.array(body.get("vector", []), dtype=np.float64)
            top_k = min(body.get("top_k", 5), 20)

            if vec.shape != (DIM,):
                self._json({"error": f"Vector must be {DIM}-dimensional"}, 400)
                return

            # 归一化
            norm = np.linalg.norm(vec)
            if norm > 0:
                vec = vec / norm

            # 计算余弦相似度
            results = []
            for doc_id, doc in docs.items():
                score = float(np.dot(vec, doc["vector"]))
                if doc["protected"]:
                    snippet = "[PROTECTED by SecureAI]"
                else:
                    snippet = doc["text"][:60]
                results.append({
                    "doc_id": doc_id,
                    "score": score,
                    "snippet": snippet,
                })

            results.sort(key=lambda x: -x["score"])
            self._json({"results": results[:top_k]})

        elif path == "/api/upload":
            text = body.get("text", "").strip()
            if not text or len(text) > 10000:
                self._json({"error": "Invalid text"}, 400)
                return

            doc_id = f"user_{secrets.token_hex(8)}"
            docs[doc_id] = {
                "text": text,
                "vector": embed(text),
                "protected": False,
            }
            self._json({"doc_id": doc_id, "message": "Document uploaded and indexed"})

        else:
            self._json({"error": "Not found"}, 404)

    def log_message(self, format, *args):
        pass  # 安静模式


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    server = HTTPServer(("0.0.0.0", port), Handler)
    print(f"SecureRAG running on port {port}")
    server.serve_forever()

分析

这题和上一道 Blind RAG 非常像，也是构造向量。

可以通过暴露的接口查询自定义向量和目标文本向量之间的余弦相似度。一开始我想的是如果把查询向量构造为 $q = [1, 0, \dots, 0]$ ，再对 snippet 值做匹配，就能得到 flag 向量的第一个分量。以此类推发 64 次请求，就能完整还原 flag 向量，再通过源码里的 decode() 函数还原为文本。

不过 top_k 最高是 20，如果没匹配到就完蛋。但细看源码，相关过滤机制是 top_k = min(body.get("top_k", 5), 20)，最终输出 results[:top_k]，并没有对负值做校验。所以传 top_k = -1 就能查到除最后一项外的所有条目。实际上手测试了一下，发现 flag 在某些维度竟然还真是垫底的。但是项目有上传功能，只需要上传 64 份在对应维度相似度最低的文档，让 flag 不再垫底，就能还原出 flag 向量，进而解题。

对于上传文档的构造，先看源码的实现：

def embed(text: str) -> np.ndarray:
    data = text.encode()
    v = np.ones(DIM, dtype=np.float64)
    n = min(len(data), DIM - 1)
    for i in range(n):
        ratio = np.exp((data[i] - 128) / 64.0)
        v[i] = ratio
    return v / np.linalg.norm(v)

即某个字符对应的 ASCII 码较小，对应位置的向量分量就小。那么对于第 i 维，把对应位置的字符设为 \x01，其它设为 \x7F，就能让上传的文档在指定维度垫底。

EXP

import requests
import numpy as np

url = "https://js5dgziq-bwag-nuxq-zkac-6a5df1cd31857-neptunus.nepctf.com/api"
DIM = 64


def decode(v_norm: np.ndarray) -> str:
    """从归一化向量恢复文本"""
    v = np.array(v_norm, dtype=np.float64)
    v = v / np.linalg.norm(v)
    ref = v[-1]
    chars = []
    for i in range(DIM - 1):
        ratio = v[i] / ref
        char_code = int(round(np.log(ratio) * 64 + 128))
        if 32 <= char_code <= 126:
            chars.append(chr(char_code))
        else:
            break
    return "".join(chars)


# 上传垫底文档
def upload_docs():
    for i in range(64):
        print(i + 1)
        s = ["\x7f"] * 64
        s[i] = "\x01"
        requests.post(f"{url}/upload", json={"text": "".join(s)}).raise_for_status()


def main():
    upload_docs()

    flag = []
    for i in range(64):
        print(i + 1)
        vec = [0.0] * 64
        vec[i] = 1.0

        payload = {"vector": vec, "top_k": -1}
        response = requests.post(f"{url}/search", json=payload)
        response.raise_for_status()
        data = response.json()

        score = 0.0
        for res in data.get("results", []):
            if res["snippet"] == "[PROTECTED by SecureAI]":
                score = res["score"]
                break
        flag.append(score)

    flag = decode(flag)
    print(flag)


if __name__ == "__main__":
    main()