NepCTF LeakyRAG / SecureRAG 题解

题目类型：Crypto / Web
核心漏洞：余弦相似度分数泄漏导致秘密向量可被完整重建
关键词：向量、余弦相似度、Top-K、坐标微扰、浮点数侧信道

1. 题目介绍

需要分析的是附件中的 SecureRAG 向量搜索服务。

题目将自己包装成一个“安全的可搜索加密系统”：

• 文档内容会被转成 64 维向量；
• 普通文档可以读取；
• flag 文档被标记为受保护文档；
• 无法直接读取 flag 文档的任何信息
• 但是可以提交任意 64 维查询向量；
• 搜索接口会返回查询向量与各文档向量的余弦相似度分数。

服务器声称：只要隐藏正文和向量，搜索过程就是安全的。

我们的目标就是证明这个说法是错误的。

2. 解题思路

虽然服务器不直接给出 flag 的向量，但它会返回：

$$ \text{score}=\frac{q\cdot x}{\lVert q\rVert\lVert x\rVert} $$

其中：

• $q$ 是我们提交的查询向量；
• $x$ 是 flag 对应的秘密向量；

服务器中的文档向量已经被归一化，因此 $\lVert x\rVert=1$。

我们先找到一个能让 flag_doc 稳定出现在 Top-20 中的查询向量，然后每次只修改查询向量的一个坐标。通过比较修改前后的相似度分数，就能计算出秘密向量对应坐标的值。

重复 64 次后，整个秘密向量都会被恢复。

最后再利用题目自定义的向量编码公式，把秘密向量转换回 flag 字符串。

3. 附件审计

3.1 向量维度

源码中定义：

DIM = 64

所有文档都被编码成 64 维向量。

也就是说，一个文档最终会变成：

$$ (v_0,v_1,v_2,\ldots,v_{63}) $$

总共 64 个浮点数。

3.2 flag 文档确实无法直接读取

flag 文档的初始化代码如下：

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

访问文档接口时，如果文档被标记为 protected，服务器会返回 403：

if doc["protected"]:
    self._json(
        {"error": "This document is protected by SecureAI encryption"},
        403,
    )

所以下述请求无效：

/api/doc/flag_doc

但是，保护措施只隐藏了正文，没有隐藏搜索分数。

3.3 搜索接口泄漏精确分数

搜索接口的核心代码如下：

vec = np.array(body.get("vector", []), dtype=np.float64)

norm = np.linalg.norm(vec)
if norm > 0:
    vec = vec / norm

for doc_id, doc in docs.items():
    score = float(np.dot(vec, doc["vector"]))

靶机会先把查询向量归一化，然后计算它与文档向量的点积。由于文档向量也已经被归一化，所以这里的点积就是余弦相似度。

返回结果中包含：

{
  "doc_id": "flag_doc",
  "score": 0.123456789,
  "snippet": "[PROTECTED by SecureAI]"
}

正文虽然被替换成了 [PROTECTED by SecureAI]，但是 doc_id 和精确的 score 仍然会返回。

这就是整道题最关键的泄漏。

4. 相似度分数为什么会泄漏秘密向量（应该是基础中的基础吧）

假设 flag 的秘密向量为：

$$ x=(x_0,x_1,\ldots,x_{63}) $$

我们不知道任何一个 $x_i$。

现在提交一个自己知道的查询向量：

$$ a=(a_0,a_1,\ldots,a_{63}) $$

服务器返回分数 $s_0$：

$$ s_0=rac{a\cdot x}{\lVert a\rVert} $$

因为秘密向量 $x$ 已经归一化，所以 $\lVert x\rVert=1$。

将公式两边乘以 $\lVert a\rVert$：

$$ s_0\lVert a\rVert=a\cdot x $$

也就是说，服务器虽然没有直接告诉我们 $x$，但它告诉了我们 $a\cdot x$。

这相当于对秘密向量做了一次线性测量。

所以只要可以构造足够多组有规律的查询，就可以反推出秘密向量的所有坐标。

5. Top-20 的魔咒

5.1 为什么不能直接查询标准基向量

最直观的想法是依次查询：

$$ e_0=(1,0,0,\ldots,0) $$

$$ e_1=(0,1,0,\ldots,0) $$

一直到：

$$ e_{63}=(0,0,\ldots,1) $$

如果服务器每次都返回 flag_doc，那么查询 $e_i$ 时的分数就直接等于 $x_i$。

但是题目只返回相似度最高的 20 个文档：

top_k = min(body.get("top_k", 5), 20)

所以数据库中一共有：

• 10 个普通文档；
• 1 个 flag 文档；
• 50 个噪声文档。

总数为 61。

当我们查询某个标准基向量时，flag_doc 不一定能进入 Top-20。如果它没有进入结果列表，我们就得不到它的分数。

所以需要先解决“如何让 flag 文档稳定出现在 Top-20”这个问题。

5.2 随机寻找锚点向量

随机生成一个 64 维向量，并把它归一化：

$$ a=\frac{r}{\lVert r\rVert} $$

然后把 $a$ 发送。

如果 flag_doc 没有出现在 Top-20，就重新生成一个随机向量继续尝试。

如果 flag_doc 出现了，而且排名比较靠前，例如前 5 名，就把这个向量保存为“锚点向量”。

之所以要求排名尽量靠前，是因为后面需要轻微修改这个向量。如果 flag 原本排在第 20 名，哪怕只改变一点点，它也可能掉出 Top-20。

如果 flag 排在第 1 到第 5 名，轻微扰动后通常仍然会留在返回结果中。

脚本还会计算 flag 与第 20 名之间的分数差：

$$ \text{margin} = ext{flag score}-\text{20th score} $$

这个差值越大，锚点就越稳定。

粗略地说，如果 61 个文档在随机方向上的排名近似均匀，那么 flag 进入 Top-20 的概率大约为：

$$ \frac{20}{61}\approx32.8\% $$

进入前 5 的概率大约为：

$$ \frac{5}{61}\approx8.2\% $$

（但是，我们从概率学上来说，实际文档向量之间并不完全独立，因此这只是近似值，不是严格概率证明）

6. 单坐标微扰攻击

这是整道题最重要的数学部分。

6.1 锚点查询

设锚点向量为：

$$ a=(a_0,a_1,\ldots,a_{63}) $$

秘密向量为：

$$ x=(x_0,x_1,\ldots,x_{63}) $$

靶机返回锚点分数：

$$ s_0=rac{a\cdot x}{\lVert a\rVert} $$

所以：

$$ a\cdot x=s_0\lVert a\rVert $$

6.2 只修改第 $i$ 个坐标

定义第 $i$ 个标准基向量：

$$ e_i=(0,0,\ldots,1,\ldots,0) $$

只有第 $i$ 个位置是 1，其余位置都是 0。

选择一个很小的数 $\varepsilon$，构造新查询：

$$ q_i=a+\varepsilon e_i $$

这表示只把锚点向量的第 $i$ 个坐标增加 $\varepsilon$。

服务器会对它归一化，并返回新的分数 $s_i$：

$$ s_i= \frac{(a+\varepsilon e_i)\cdot x} {\lVert a+\varepsilon e_i\rVert} $$

把分母乘到等式左边：

$$ s_i\lVert a+\varepsilon e_i\rVert =(a+\varepsilon e_i)\cdot x $$

利用点积的分配律：

$$ (a+\varepsilon e_i)\cdot x =a\cdot x+\varepsilon(e_i\cdot x) $$

而 $e_i$ 只有第 $i$ 个位置为 1，所以：

$$ e_i\cdot x=x_i $$

因此：

$$ s_i\lVert a+\varepsilon e_i\rVert =a\cdot x+\varepsilon x_i $$

前面已经得到：

$$ a\cdot x=s_0\lVert a\rVert $$

代入：

$$ s_i\lVert a+\varepsilon e_i\rVert =s_0\lVert a\rVert+\varepsilon x_i $$

移项：

$$ \varepsilon x_i= s_i\lVert a+\varepsilon e_i\rVert -s_0\lVert a\rVert $$

最终得到秘密向量第 $i$ 个坐标：

$$ \boxed{ x_i= \frac{ s_i\lVert a+\varepsilon e_i\rVert -s_0\lVert a\rVert }{\varepsilon} } $$

公式右边的所有量都是已知的：

• $s_i$：服务器返回；
• $s_0$：服务器返回；
• $a$：我们自己生成；
• $\varepsilon$：我们自己设置；
• 两个向量长度：可以在本地计算。

因此，$x_i$ 可以被直接算出。

从 $i=0$ 一直计算到 $i=63$，就能恢复完整秘密向量：

$$ x=(x_0,x_1,\ldots,x_{63}) $$

7. 为什么要选择合适的 $\varepsilon$

脚本默认从：

$$ \varepsilon=10^{-3} $$

开始尝试。

$\varepsilon$ 不能太大，因为修改太大会改变查询方向，导致 flag_doc 掉出 Top-20。

$\varepsilon$ 也不能无限小，因为公式中需要除以 $\varepsilon$：

$$ x_i=\frac{\text{两个非常接近的浮点数之差}}{\varepsilon} $$

当 $\varepsilon$ 过小时，浮点数舍入误差会被放大。

所以这是一个平衡，$10^{-3}$ 通常比较合适。

恢复完成后，脚本还会检查秘密向量的长度是否接近 1：

$$ \lVert x\rVert\approx1 $$

8. 从秘密向量解码 flag

恢复向量后，还需要理解题目是如何把字符串编码成向量的。

8.1 源码中的编码方式

核心代码如下：

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)

初始向量的所有坐标都是 1：

$$ v=(1,1,\ldots,1) $$

对于字符串中的第 $i$ 个字节 $c_i$，服务器设置：

$$ v_i= \exp\left(\frac{c_i-128}{64}\right) $$

第 64 维，也就是下标 63 的坐标，没有被字符覆盖，因此在归一化前始终为：

$$ v_{63}=1 $$

随后整个向量会被归一化：

$$ x=\frac{v}{\lVert v\rVert} $$

8.2 归一化为什么不会破坏字符信息

归一化后：

$$ x_i=\frac{v_i}{\lVert v\rVert} $$

$$ x_{63}=\frac{v_{63}}{\lVert v\rVert} $$

计算二者比值：

$$ \frac{x_i}{x_{63}} = \frac{v_i/\lVert v\rVert} {v_{63}/\lVert v\rVert} $$

上下的归一化系数会抵消：

$$ \frac{x_i}{x_{63}}=rac{v_i}{v_{63}} $$

由于 $v_{63}=1$：

$$ \frac{x_i}{x_{63}} = \exp\left(\frac{c_i-128}{64}\right) $$

对两边取自然对数：

$$ \ln\left(\frac{x_i}{x_{63}}\right) =rac{c_i-128}{64} $$

两边乘以 64：

$$ 64\ln\left(\frac{x_i}{x_{63}}\right) =c_i-128 $$

最终得到字符编码：

$$ \boxed{ c_i= 64\ln\left(\frac{x_i}{x_{63}}\right)+128 } $$

由于浮点运算会产生轻微误差，需要四舍五入：

$$ c_i= \operatorname{round} \left( 64\ln\left(\frac{x_i}{x_{63}}\right)+128 \right) $$

再通过：

chr(c_i)

把整数转换成字符，就能逐字还原 flag

以上就是原理分析。

9. 完整攻击流程

完整流程可以整理为以下步骤：

1. 请求 /api/stats，确认目标是 64 维向量服务；
2. 随机生成一个 64 维单位向量；
3. 请求 /api/search，寻找 flag_doc；
4. 如果 flag 不在 Top-20，就继续随机；
5. 尽量找到 flag 排名前 5、且与第 20 名有一定分差的锚点；
6. 保存锚点向量 $a$ 和对应分数 $s_0$；
7. 对每个坐标 $i$ 构造 $a+\varepsilon e_i$；
8. 获取对应分数 $s_i$；
9. 使用公式恢复 $x_i$；
10. 重复 64 次，恢复完整秘密向量；
11. 使用第 64 维作为参考坐标；
12. 根据比值编码公式恢复每个字符；
13. 输出 flag。

除去寻找锚点所需的随机查询，恢复向量本身只需要 64 次查询。

10. EXP

下面是完整 EXP：

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import math
import random
import sys
from typing import Iterable, Sequence

import requests

DIM = 64
TOP_K = 20


class ExploitError(RuntimeError):

def l2_norm(v: Sequence[float]) -> float:
    return math.sqrt(math.fsum(x * x for x in v))


def search(
    session: requests.Session,
    base_url: str,
    vector: Sequence[float],
) -> list[dict]:
    response = session.post(
        f"{base_url.rstrip('/')}/api/search",
        json={"vector": list(vector), "top_k": TOP_K},
        timeout=20,
    )
    response.raise_for_status()

    payload = response.json()
    if "error" in payload:
        raise ExploitError(str(payload["error"]))

    results = payload.get("results")
    if not isinstance(results, list):
        raise ExploitError(f"Unexpected API response: {payload!r}")

    return results


def locate_flag(results: Iterable[dict]) -> tuple[int, float] | None:
    for rank, item in enumerate(results, start=1):
        if item.get("doc_id") == "flag_doc":
            return rank, float(item["score"])
    return None


def random_unit_vector(rng: random.Random) -> list[float]:
    while True:
        vector = [rng.gauss(0.0, 1.0) for _ in range(DIM)]
        norm = l2_norm(vector)
        if norm > 0:
            return [x / norm for x in vector]


def find_anchor(
    session: requests.Session,
    base_url: str,
    rng: random.Random,
    attempts: int = 400,
    target_rank: int = 5,
) -> tuple[list[float], float, int, float]:
    best: tuple[list[float], float, int, float] | None = None

    for attempt in range(1, attempts + 1):
        anchor = random_unit_vector(rng)
        results = search(session, base_url, anchor)
        found = locate_flag(results)

        if found is None:
            continue

        rank, score = found
        boundary_score = float(results[-1]["score"])
        margin = score - boundary_score

        if (
            best is None
            or rank < best[2]
            or (rank == best[2] and margin > best[3])
        ):
            best = (anchor, score, rank, margin)

        if rank <= target_rank and margin > 1e-4:
            print(
                f"[+] Anchor found after {attempt} tries: "
                f"rank={rank}, Top-20 margin={margin:.6g}"
            )
            return anchor, score, rank, margin

    if best is not None:
        anchor, score, rank, margin = best
        print(
            f"[!] No rank <= {target_rank} anchor found; "
            f"using best rank={rank}, margin={margin:.6g}"
        )
        return anchor, score, rank, margin

    raise ExploitError(
        f"flag_doc never entered Top-{TOP_K} "
        f"after {attempts} random queries"
    )


def recover_vector(
    session: requests.Session,
    base_url: str,
    anchor: Sequence[float],
    anchor_score: float,
) -> list[float]:
    anchor_norm = l2_norm(anchor)
    anchor_dot = anchor_score * anchor_norm
    for epsilon in (1e-3, 1e-4, 1e-5):
        recovered: list[float] = []
        print(
            f"[*] Recovering {DIM} coordinates "
            f"with epsilon={epsilon:g}"
        )

        for i in range(DIM):
            query = list(anchor)
            query[i] += epsilon

            results = search(session, base_url, query)
            found = locate_flag(results)

            if found is None:
                print(
                    f"[!] flag_doc left Top-{TOP_K} "
                    f"at coordinate {i}; shrinking epsilon"
                )
                break

            _, perturbed_score = found
            query_norm = l2_norm(query)

            # x_i = (
            #     s_i * ||a + epsilon * e_i||
            #     - s_0 * ||a||
            # ) / epsilon
            coordinate = (
                perturbed_score * query_norm - anchor_dot
            ) / epsilon

            recovered.append(coordinate)
            print(
                f"\r    coordinate {i + 1:02d}/{DIM}",
                end="",
                flush=True,
            )
        else:
            print()

            recovered_norm = l2_norm(recovered)
            if (
                not math.isfinite(recovered_norm)
                or not (0.999 <= recovered_norm <= 1.001)
            ):
                print(
                    "[!] Recovered vector norm is "
                    f"{recovered_norm:.12f}; "
                    "trying smaller epsilon"
                )
                continue
            return [x / recovered_norm for x in recovered]

        print()

    raise ExploitError(
        f"Could not keep flag_doc in Top-{TOP_K} during perturbation"
    )


def decode_flag(vector: Sequence[float]) -> str:
    reference = vector[DIM - 1]
    if reference <= 0:
        raise ExploitError(
            f"Invalid recovered reference coordinate: {reference}"
        )

    chars: list[str] = []
    recovered_codes: list[int] = []

    for coordinate in vector[: DIM - 1]:
        if coordinate <= 0:
            break

        code = round(
            math.log(coordinate / reference) * 64.0 + 128.0
        )
        recovered_codes.append(code)

        if not 32 <= code <= 126:
            break

        chars.append(chr(code))

    text = "".join(chars)
    if not text:
        raise ExploitError(
            "Decoded no printable text; "
            f"first codes: {recovered_codes[:10]}"
        )

    return text


def solve(base_url: str, seed: int | None = None) -> str:
    session = requests.Session()
    rng = random.Random(seed)

    stats_response = session.get(
        f"{base_url.rstrip('/')}/api/stats",
        timeout=20,
    )
    stats_response.raise_for_status()
    info = stats_response.json()

    if int(info.get("dim", -1)) != DIM:
        raise ExploitError(f"Unexpected target stats: {info!r}")

    print(
        f"[+] Target online: {info.get('total_docs', '?')} docs, "
        f"{info.get('dim')} dimensions"
    )

    anchor, anchor_score, _, _ = find_anchor(
        session,
        base_url,
        rng,
    )

    secret_vector = recover_vector(
        session,
        base_url,
        anchor,
        anchor_score,
    )

    return decode_flag(secret_vector)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("url", help="challenge base URL")
    parser.add_argument(
        "--seed",
        type=int,
        default=None,
        help="optional random seed",
    )
    args = parser.parse_args()

    try:
        flag = solve(args.url, args.seed)
    except (
        requests.RequestException,
        ValueError,
        ExploitError,
    ) as exc:
        print(f"[-] Exploit failed: {exc}", file=sys.stderr)
        return 1

    print(f"\nFLAG: {flag}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

这道题最有意思的地方在于，服务器表面上只泄漏了一个普通的“搜索分数”，但红队通过精心设计查询，可以把这个分数接口变成一台逐坐标扫描秘密向量的测量仪。