#!/usr/bin/env python3 import os import random imp...

Question

#!/usr/bin/env python3
import os
import random
import socket
import socketserver
import argparse

def load_flag():
for p in ("/flag", "./flag.txt"):
if os.path.exists(p):
with open(p, "r", encoding="utf-8") as f:
return f.read().strip()
return "flag{test_flag}"

FLAG = load_flag()

class CathylinFourPlus:
def init(self, r, key):
self.r = r
self.S = (0xC, 0x5, 0x6, 0xB,
0x9, 0x0, 0xA, 0xD,
0x3, 0xE, 0xF, 0x8,
0x4, 0x7, 0x1, 0x2)
self.P = (0, 5, 10, 15,
4, 9, 14, 3,
8, 13, 2, 7,
12, 1, 6, 11)
self.RC = (0x0F0F, 0x3333, 0x55AA)
self.key = key
assert len(key) == self.r + 1

text
@staticmethod
def deblock(m):
    return [(m >> 12) & 0xF, (m >> 8) & 0xF, (m >> 4) & 0xF, m & 0xF]

@staticmethod
def block(nibbles):
    return (nibbles[0] << 12) | (nibbles[1] << 8) | (nibbles[2] << 4) | nibbles[3]

def s_layer(self, x):
    n = self.deblock(x)
    n = [self.S[i] for i in n]
    return self.block(n)

def permute_bits(self, x):
    bits = [(x >> (15 - i)) & 1 for i in range(16)]
    new_bits = [0] * 16
    for i in range(16):
        new_bits[self.P[i]] = bits[i]
    out = 0
    for b in new_bits:
        out = (out << 1) | b
    return out & 0xFFFF

def encrypt(self, m):
    p = m & 0xFFFF
    for t in range(self.r - 1):
        p ^= self.key[t]
        p ^= self.RC[t]
        p = self.s_layer(p)
        p = self.permute_bits(p)
    p ^= self.key[self.r - 1]
    p ^= self.RC[self.r - 1]
    p = self.s_layer(p)
    c = p ^ self.key[-1]
    return c & 0xFFFF

def parse_hex16(s: str) -> int:
s = s.strip().lower()
if s.startswith("0x"):
s = s[2:]
if not (1 <= len(s) <= 4) or any(ch not in "0123456789abcdef" for ch in s):
raise ValueError("bad hex")
return int(s, 16) & 0xFFFF

def serve_one_client(rfile, wfile):
# 每连接随机 key / mask
r = 3
key = [random.getrandbits(16) for _ in range(r + 1)]
enc = CathylinFourPlus(r, key)
mask = random.getrandbits(16)

text
def wln(msg=""):
    wfile.write((msg + "\n").encode())
    wfile.flush()

def write_prompt():
    wfile.write(b"Your input: ")
    wfile.flush()

wln("=== CathylinFour++ (SPN 16-bit) ===")
wln("Input: hex (e.g. 0x1234). Commands: h/help, k, q")
wln(f"mask = 0x{mask:04x}")
wln("Goal: submit (K3 ^ mask) to get the flag.")
wln("----------------------------------")

qcnt = 0
QMAX = 4096

while True:
    write_prompt()
    line = rfile.readline()
    if not line:
        break
    x = line.decode(errors="ignore").strip().lower()

    if x in ("q", "quit", "exit"):
        wln("Bye.")
        break

    if x in ("h", "help", "?"):
        wln("Commands:")
        wln("  <hex>   encrypt a 16-bit block, e.g. 0x1234")
        wln("  k       submit key proof (K3 ^ mask)")
        wln("  q       quit")
        continue

    if x == "k":
        wfile.write(b"Your key proof (hex): ")
        wfile.flush()
        ki = rfile.readline()
        if not ki:
            break
        try:
            proof = parse_hex16(ki.decode(errors="ignore"))
        except Exception:
            wln("Bad format. Bye.")
            break

        if proof == (key[-1] ^ mask):
            wln("You can smile: " + FLAG)
        else:
            wln("Wrong. Bye.")
        break

    try:
        a = parse_hex16(x)
    except Exception:
        wln("Bad format. Tips: 0x???? (1~4 hex digits).")
        continue

    qcnt += 1
    if qcnt > QMAX:
        wln("Too many queries. Bye.")
        break

    c = enc.encrypt(a)
    wln("Output: " + f"0x{c:04x}")

class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
allow_reuse_address = True
daemon_threads = True

class Handler(socketserver.BaseRequestHandler):
def handle(self):
# 每连接超时：防止挂死占用线程
self.request.settimeout(60)
rfile = self.request.makefile("rb", buffering=0)
wfile = self.request.makefile("wb", buffering=0)
try:
serve_one_client(rfile, wfile)
except (socket.timeout, BrokenPipeError, ConnectionResetError):
pass
finally:
try:
rfile.close()
except Exception:
pass
try:
wfile.close()
except Exception:
pass

def main():
ap = argparse.ArgumentParser()
ap.add_argument("--listen", default=None, help="listen address, e.g. 0.0.0.0")
ap.add_argument("--port", type=int, default=37344)
args = ap.parse_args()

text
if args.listen:
    with ThreadedTCPServer((args.listen, args.port), Handler) as srv:
        srv.serve_forever()
else:
    # 本地 stdin/stdout 单机调试模式
    import sys
    serve_one_client(sys.stdin.buffer, sys.stdout.buffer)

if name == "main":
main() 这是一道ctf 的crypto题目提示：此题不是正常flag

Answer 1

这题的本质：你有一个16-bit、3 轮的 SPN加密 oracle，最后一轮是

做完最后一层 S 盒后，直接 c = state ^ K3
服务端把 mask 给你看，你只要提交 K3 ^ mask 就出 flag

所以目标就是在同一条连接里把最后一轮子密钥 K3（16-bit）搞出来。

关键点：XOR 轮密钥/轮常量不影响差分。因此可以对最后一轮做经典“猜子密钥 + 逆 S 盒”的差分恢复。

用什么差分（这题最省事的一组）

对这套 S/P（代码里那两个表），如果把“前两轮（到最后一轮 S 盒输入前）”当作一个 2-round SPN 函数，可以离线算到下面这些高概率差分模式（我直接给你可用的）：

我们分别用 4 个明文差分 ΔP，每个用来恢复 K3 的一个 nibble：

恢复哪一段 K3	选 ΔP	期望“最后一轮 S 盒输入差分”在该 nibble 上是	同时要求哪些 nibble 的密文差分为 0（用于过滤）
K3[15:12]（最高 nibble，pos=0）	0x0D00	0x4	密文差分 nibble1、nibble3 为 0
K3[11:8]（pos=1）	0x7000	0x2	密文差分 nibble0、nibble3 为 0
K3[7:4]（pos=2）	0x9000	0x4	密文差分 nibble1、nibble3 为 0
K3[3:0]（pos=3，最低 nibble）	0xD000	0x4	密文差分 nibble0、nibble2 为 0

然后对通过过滤的对 (C, C')，对目标 nibble 猜 k in [0..15]，算：

text
u  = invS( C_nib  ^ k )
u' = invS( C'_nib ^ k )
如果 (u ^ u') == 期望差分 => 给这个 k +1 分

得分最高的 k 就是该 nibble 的 K3。

注意：这题的 S 盒里，输入差分为 0x1 会有“键等价”的坑，所以上面我特意选了在目标 nibble 上差分是 0x2 或 0x4 的组合，能稳定唯一恢复。

直接可用的 exploit（不需要满 4096 次查询）

下面脚本默认每个 ΔP 跑 N=400 对（总查询 4 * 2 * 400 = 3200 < 4096，留足余量），基本稳定。

把 HOST/PORT 改成题目地址即可：

python
#!/usr/bin/env python3
import socket
import re
import random

S = [0xC, 0x5, 0x6, 0xB,
     0x9, 0x0, 0xA, 0xD,
     0x3, 0xE, 0xF, 0x8,
     0x4, 0x7, 0x1, 0x2]

INV = [0]*16
for i,v in enumerate(S):
    INV[v] = i

def nib(x, pos):
    """pos: 0..3 from MS nibble to LS nibble"""
    return (x >> (12 - 4*pos)) & 0xF

def set_nib(x, pos, v):
    sh = 12 - 4*pos
    x &= ~(0xF << sh)
    x |= (v & 0xF) << sh
    return x

class Oracle:
    def __init__(self, host, port):
        self.s = socket.create_connection((host, port))
        self.rf = self.s.makefile("rb", buffering=0)
        self.wf = self.s.makefile("wb", buffering=0)
        self.mask = None

    def read_banner_and_mask(self):
        buf = b""
        # 读到出现 mask 行为止
        while True:
            line = self.rf.readline()
            if not line:
                raise RuntimeError("connection closed while reading banner")
            buf += line
            m = re.search(rb"mask\s*=\s*0x([0-9a-fA-F]{4})", buf)
            if m:
                self.mask = int(m.group(1), 16)
                return

    def enc(self, pt16):
        self.wf.write(f"0x{pt16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()
        line = self.rf.readline()
        if not line:
            raise RuntimeError("connection closed during query")
        m = re.search(rb"Output:\s*0x([0-9a-fA-F]{4})", line)
        if not m:
            raise RuntimeError(f"unexpected oracle line: {line!r}")
        return int(m.group(1), 16)

    def submit(self, proof16):
        # 直接连发，不用等提示
        self.wf.write(b"k\n")
        self.wf.write(f"0x{proof16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()
        # 读几行把 flag 打出来
        out = b""
        while True:
            line = self.rf.readline()
            if not line:
                break
            out += line
        return out.decode(errors="ignore")

def recover_one_nibble(oracle, deltaP, target_pos, expected_diff, inactive_zero_positions, N=400):
    scores = [0]*16
    kept = 0

    for _ in range(N):
        p = random.getrandbits(16)
        p2 = p ^ deltaP
        c  = oracle.enc(p)
        c2 = oracle.enc(p2)
        cd = c ^ c2

        # 过滤：要求某些 nibble 的密文差分为 0
        ok = True
        for ip in inactive_zero_positions:
            if nib(cd, ip) != 0:
                ok = False
                break
        if not ok:
            continue

        kept += 1
        cn  = nib(c,  target_pos)
        cn2 = nib(c2, target_pos)

        for k in range(16):
            u  = INV[cn  ^ k]
            u2 = INV[cn2 ^ k]
            if (u ^ u2) == expected_diff:
                scores[k] += 1

    best = max(range(16), key=lambda k: scores[k])
    scores_sorted = sorted(scores, reverse=True)
    margin = scores_sorted[0] - scores_sorted[1]
    return best, kept, margin, scores

def main():
    HOST = "127.0.0.1"
    PORT = 37344

    o = Oracle(HOST, PORT)
    o.read_banner_and_mask()
    print(f"[+] mask = 0x{o.mask:04x}")

    # 四个配置：各恢复一个 nibble
    # pos: 0..3 (MS nibble -> LS nibble)
    jobs = [
        # recover K3 nibble0 (MS)
        dict(deltaP=0x0D00, pos=0, a=0x4, inactive=[1,3]),
        # recover K3 nibble1
        dict(deltaP=0x7000, pos=1, a=0x2, inactive=[0,3]),
        # recover K3 nibble2
        dict(deltaP=0x9000, pos=2, a=0x4, inactive=[1,3]),
        # recover K3 nibble3 (LS)
        dict(deltaP=0xD000, pos=3, a=0x4, inactive=[0,2]),
    ]

    k3_nibs = [0,0,0,0]
    for j in jobs:
        k, kept, margin, scores = recover_one_nibble(
            o,
            deltaP=j["deltaP"],
            target_pos=j["pos"],
            expected_diff=j["a"],
            inactive_zero_positions=j["inactive"],
            N=400,          # 总查询 3200 次，< 4096
        )
        k3_nibs[j["pos"]] = k
        print(f"[+] pos={j['pos']}  deltaP=0x{j['deltaP']:04x}  kept={kept}  best=0x{k:x}  margin={margin}")

    K3 = (k3_nibs[0]<<12) | (k3_nibs[1]<<8) | (k3_nibs[2]<<4) | (k3_nibs[3])
    print(f"[+] Recovered K3 = 0x{K3:04x}")

    proof = K3 ^ o.mask
    print(f"[+] Sending proof = K3 ^ mask = 0x{proof:04x}")
    resp = o.submit(proof)
    print(resp)

if __name__ == "__main__":
    main()

你可能关心的“此题不是正常 flag”

服务端打印的是：You can smile: + FLAG，FLAG 可能不是 flag{...} 这种格式（出题人提示的意思大概率就是这个）。脚本拿到服务端返回啥就会原样打印。

如果你把题目远程的 HOST:PORT 给我，我也可以帮你把脚本改成更适合那边 IO 的版本（比如 pwntools、并发批量发送、更快更稳）。

Answer 2

你给的这个服务端脚本里只写死了默认端口 37344，并没有写死 HOST（它要么是本地 stdin/stdout 模式，要么你启动时用 --listen 指定监听地址）。所以解题脚本里 HOST 需要你自己填：

本地测试一般用：127.0.0.1:37344
远程就是题目平台给你的域名/IP + 37344

我把 exploit 改成了命令行传 host/port，你直接用就行。

1）本地怎么跑（可选）

先把你那份 server 脚本跑起来（监听 TCP）：

bash
python3 server.py --listen 127.0.0.1 --port 37344

2）解题脚本（solve.py）

python
#!/usr/bin/env python3
import socket
import re
import random
import argparse

S = [0xC, 0x5, 0x6, 0xB,
     0x9, 0x0, 0xA, 0xD,
     0x3, 0xE, 0xF, 0x8,
     0x4, 0x7, 0x1, 0x2]

INV = [0]*16
for i, v in enumerate(S):
    INV[v] = i

def nib(x, pos):
    """pos: 0..3 from MS nibble to LS nibble"""
    return (x >> (12 - 4*pos)) & 0xF

class Oracle:
    def __init__(self, host, port, timeout=10):
        self.s = socket.create_connection((host, port), timeout=timeout)
        self.rf = self.s.makefile("rb", buffering=0)
        self.wf = self.s.makefile("wb", buffering=0)
        self.mask = None

    def read_banner_and_mask(self):
        buf = b""
        while True:
            line = self.rf.readline()
            if not line:
                raise RuntimeError("connection closed while reading banner")
            buf += line
            m = re.search(rb"mask\s*=\s*0x([0-9a-fA-F]{4})", buf)
            if m:
                self.mask = int(m.group(1), 16)
                return

    def enc(self, pt16):
        self.wf.write(f"0x{pt16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()
        line = self.rf.readline()
        if not line:
            raise RuntimeError("connection closed during query")
        m = re.search(rb"Output:\s*0x([0-9a-fA-F]{4})", line)
        if not m:
            raise RuntimeError(f"unexpected oracle line: {line!r}")
        return int(m.group(1), 16)

    def submit(self, proof16):
        self.wf.write(b"k\n")
        self.wf.write(f"0x{proof16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()
        out = b""
        while True:
            line = self.rf.readline()
            if not line:
                break
            out += line
        return out.decode(errors="ignore")

def recover_one_nibble(oracle, deltaP, target_pos, expected_diff, inactive_zero_positions, N=400):
    scores = [0]*16
    kept = 0

    for _ in range(N):
        p = random.getrandbits(16)
        p2 = p ^ deltaP
        c  = oracle.enc(p)
        c2 = oracle.enc(p2)
        cd = c ^ c2

        # 过滤：要求某些 nibble 的密文差分为 0
        ok = True
        for ip in inactive_zero_positions:
            if nib(cd, ip) != 0:
                ok = False
                break
        if not ok:
            continue

        kept += 1
        cn  = nib(c,  target_pos)
        cn2 = nib(c2, target_pos)

        for k in range(16):
            u  = INV[cn  ^ k]
            u2 = INV[cn2 ^ k]
            if (u ^ u2) == expected_diff:
                scores[k] += 1

    best = max(range(16), key=lambda k: scores[k])
    scores_sorted = sorted(scores, reverse=True)
    margin = scores_sorted[0] - scores_sorted[1]
    return best, kept, margin

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("host", nargs="?", default="127.0.0.1")
    ap.add_argument("port", nargs="?", type=int, default=37344)
    ap.add_argument("-n", "--pairs", type=int, default=400, help="pairs per job (default 400)")
    args = ap.parse_args()

    o = Oracle(args.host, args.port)
    o.read_banner_and_mask()
    print(f"[+] mask = 0x{o.mask:04x}")

    # 四个配置：各恢复一个 nibble（总查询约 4*2*pairs，默认 3200 < 4096）
    jobs = [
        dict(deltaP=0x0D00, pos=0, a=0x4, inactive=[1, 3]),  # K3[15:12]
        dict(deltaP=0x7000, pos=1, a=0x2, inactive=[0, 3]),  # K3[11:8]
        dict(deltaP=0x9000, pos=2, a=0x4, inactive=[1, 3]),  # K3[7:4]
        dict(deltaP=0xD000, pos=3, a=0x4, inactive=[0, 2]),  # K3[3:0]
    ]

    k3_nibs = [0, 0, 0, 0]
    for j in jobs:
        k, kept, margin = recover_one_nibble(
            o,
            deltaP=j["deltaP"],
            target_pos=j["pos"],
            expected_diff=j["a"],
            inactive_zero_positions=j["inactive"],
            N=args.pairs,
        )
        k3_nibs[j["pos"]] = k
        print(f"[+] pos={j['pos']} deltaP=0x{j['deltaP']:04x} kept={kept} best=0x{k:x} margin={margin}")

    K3 = (k3_nibs[0] << 12) | (k3_nibs[1] << 8) | (k3_nibs[2] << 4) | k3_nibs[3]
    print(f"[+] Recovered K3 = 0x{K3:04x}")

    proof = K3 ^ o.mask
    print(f"[+] Sending proof = 0x{proof:04x}")
    resp = o.submit(proof)
    print(resp)

if __name__ == "__main__":
    main()

3）怎么用

本地：

bash
python3 solve.py 127.0.0.1 37344

远程（把 HOST 换成题目给的域名/IP）：

bash
python3 solve.py HOST 37344

题目提示“不是正常flag”没关系：服务端回什么这脚本就原样打印什么（可能不是 flag{...} 格式）。

如果你把题目平台给的那个 HOST（域名/IP）发我一句，我也可以顺手帮你确认脚本在那边的输出格式是否需要微调（有些平台会多几行 banner 或者输出不在同一行）。

Answer 3

这个报错很正常：你在 read_banner_and_mask() 里读到 mask 那一行就返回了，但服务端 mask 后面还有几行 banner（比如 Goal: ...、分割线），导致你第一次 enc() 的 readline() 读到的还是 banner 行，于是 regex 匹配不到 Output: 就直接 raise 了。

把 Oracle.enc() 改成“一直读到出现 Output 为止（中间的 banner 行全部丢掉）”就好了。同理提交 k 的时候也建议做成“读到连接关闭/或读到结果行”。

下面是最小补丁（你直接替换 Oracle.enc 就能跑）：

python
def enc(self, pt16):
    self.wf.write(f"0x{pt16 & 0xFFFF:04x}\n".encode())
    self.wf.flush()

    while True:
        line = self.rf.readline()
        if not line:
            raise RuntimeError("connection closed during query")

        # 这里用 search，所以即使是 "Your input: Output: 0x...." 也能匹配
        m = re.search(rb"Output:\s*0x([0-9a-fA-F]{4})", line)
        if m:
            return int(m.group(1), 16)

        # 否则就是 banner/提示/残留行，丢掉继续读

如果你想更稳一点（避免 k 提交时也被残留行影响），也可以把 submit() 写成“读到包含 flag/You can smile/Wrong 之类关键字就返回”，不过通常你现在这个 submit() 读到断开就够用了。

你改完再跑一次应该就能继续往下打分恢复 K3 了。要是之后出现“margin 很小/偶尔猜错 nibble”，把 -n/--pairs 调大点（比如 600）即可，但别超过 4096 查询总数。

Answer 4

你这个 IndentationError 是因为你把 enc() 的函数体粘贴进去后，没有在 def enc(...): 下面再缩进一层（Python 必须缩进）。

最省事：你直接把你文件里的 整个 Oracle 类替换成下面这一段（保证缩进完全正确），其它地方不用动。

python
class Oracle:
    def __init__(self, host, port, timeout=10):
        self.s = socket.create_connection((host, port), timeout=timeout)
        self.rf = self.s.makefile("rb", buffering=0)
        self.wf = self.s.makefile("wb", buffering=0)
        self.mask = None

    def read_banner_and_mask(self):
        buf = b""
        while True:
            line = self.rf.readline()
            if not line:
                raise RuntimeError("connection closed while reading banner")
            buf += line
            m = re.search(rb"mask\s*=\s*0x([0-9a-fA-F]{4})", buf)
            if m:
                self.mask = int(m.group(1), 16)
                return

    def enc(self, pt16):
        # 发送明文
        self.wf.write(f"0x{pt16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()

        # 一直读到真正的 Output 行（中间 banner/提示行全部跳过）
        while True:
            line = self.rf.readline()
            if not line:
                raise RuntimeError("connection closed during query")

            m = re.search(rb"Output:\s*0x([0-9a-fA-F]{4})", line)
            if m:
                return int(m.group(1), 16)
            # 否则丢掉继续读

    def submit(self, proof16):
        self.wf.write(b"k\n")
        self.wf.write(f"0x{proof16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()

        out = b""
        while True:
            line = self.rf.readline()
            if not line:
                break
            out += line
        return out.decode(errors="ignore")

为什么会这样

你之前的报错 unexpected oracle line: b'Goal: ...' 是因为 banner 没读完；我这个 enc() 会自动把 banner 行吃掉，直到读到 Output: 0x....。

如果你还遇到第二个坑

如果你现在的 main() 里还是写着：

python
k, kept, margin, scores = recover_one_nibble(...)

但你的 recover_one_nibble() 只 return best, kept, margin（3 个值），那会再报 “not enough values to unpack”。
解决：把那行改成 3 个变量：

python
k, kept, margin = recover_one_nibble(...)

你改完再跑一次就应该能继续恢复 K3 了。要是你愿意，把你当前 recover_one_nibble 和 main 那几行也贴出来，我可以直接帮你对齐成一份可运行的完整 solve.py。

Answer 5

下面给你一份完整可运行的 solve.py（已修好 banner/提示行导致的读取错位问题；并且默认查询次数保证 <4096）。

把它保存为 solve.py，然后运行：

本地：python3 solve.py 127.0.0.1 37344
远程：python3 solve.py 题目给的HOST 37344

python
#!/usr/bin/env python3
import socket
import re
import random
import argparse

# ------------------------
# S-box and inverse S-box
# ------------------------
S = [0xC, 0x5, 0x6, 0xB,
     0x9, 0x0, 0xA, 0xD,
     0x3, 0xE, 0xF, 0x8,
     0x4, 0x7, 0x1, 0x2]

INV = [0] * 16
for i, v in enumerate(S):
    INV[v] = i

def nib(x, pos):
    """pos: 0..3 from MS nibble to LS nibble"""
    return (x >> (12 - 4 * pos)) & 0xF

# ------------------------
# Oracle wrapper
# ------------------------
class Oracle:
    def __init__(self, host, port, timeout=10):
        self.s = socket.create_connection((host, port), timeout=timeout)
        self.rf = self.s.makefile("rb", buffering=0)
        self.wf = self.s.makefile("wb", buffering=0)
        self.mask = None

    def read_mask_only(self):
        """
        只读到 mask 行就返回。
        注意：服务端之后还会输出几行 banner + 一个没有换行的 prompt，
        所以这里不能贪多读，否则会卡在 readline() 等换行。
        """
        buf = b""
        while True:
            line = self.rf.readline()
            if not line:
                raise RuntimeError("connection closed while reading banner")
            buf += line
            m = re.search(rb"mask\s*=\s*0x([0-9a-fA-F]{4})", buf)
            if m:
                self.mask = int(m.group(1), 16)
                return

    def enc(self, pt16):
        """
        发送明文后，不断读取直到匹配到 Output: 0x....
        这样可以自动吃掉 banner 残留行/提示行/以及 'Your input: ' 这种无换行 prompt 拼接的行。
        """
        self.wf.write(f"0x{pt16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()

        while True:
            line = self.rf.readline()
            if not line:
                raise RuntimeError("connection closed during query")

            m = re.search(rb"Output:\s*0x([0-9a-fA-F]{4})", line)
            if m:
                return int(m.group(1), 16)
            # 否则是 banner/提示/残留行，丢掉继续读

    def submit(self, proof16):
        # 进入提交模式
        self.wf.write(b"k\n")
        self.wf.write(f"0x{proof16 & 0xFFFF:04x}\n".encode())
        self.wf.flush()

        # 读到断开为止（服务端会 break 然后关闭连接）
        out = b""
        while True:
            chunk = self.rf.readline()
            if not chunk:
                break
            out += chunk
        return out.decode(errors="ignore")

# ------------------------
# Differential last-round nibble recovery
# ------------------------
def recover_one_nibble(oracle, deltaP, target_pos, expected_diff, inactive_zero_positions, N=400):
    scores = [0] * 16
    kept = 0

    for _ in range(N):
        p = random.getrandbits(16)
        p2 = p ^ deltaP

        c  = oracle.enc(p)
        c2 = oracle.enc(p2)
        cd = c ^ c2

        # 过滤：要求某些 nibble 的密文差分为 0
        ok = True
        for ip in inactive_zero_positions:
            if nib(cd, ip) != 0:
                ok = False
                break
        if not ok:
            continue

        kept += 1
        cn  = nib(c,  target_pos)
        cn2 = nib(c2, target_pos)

        # 猜最后一轮子密钥该 nibble（4-bit）
        for k in range(16):
            u  = INV[cn  ^ k]
            u2 = INV[cn2 ^ k]
            if (u ^ u2) == expected_diff:
                scores[k] += 1

    best = max(range(16), key=lambda k: scores[k])
    top2 = sorted(scores, reverse=True)[:2]
    margin = top2[0] - top2[1] if len(top2) == 2 else top2[0]
    return best, kept, margin, scores

# ------------------------
# Main
# ------------------------
def main():
    ap = argparse.ArgumentParser(description="CathylinFour++ solver: recover K3 then submit K3^mask")
    ap.add_argument("host", nargs="?", default="127.0.0.1")
    ap.add_argument("port", nargs="?", type=int, default=37344)
    ap.add_argument("-n", "--pairs", type=int, default=400, help="pairs per job (default 400). total queries ~= 8*pairs")
    args = ap.parse_args()

    # 保证不超过 QMAX=4096（每 pair 需要 2 次 enc；4 个 job => 8 次 enc）
    max_pairs = 4096 // 8  # 512
    if args.pairs > max_pairs:
        print(f"[!] pairs too large, cap to {max_pairs} to stay under 4096 queries.")
        args.pairs = max_pairs

    o = Oracle(args.host, args.port)
    o.read_mask_only()
    print(f"[+] mask = 0x{o.mask:04x}")

    # 4 个 job：分别恢复 K3 的 4 个 nibble（pos 0..3 从高到低）
    jobs = [
        dict(deltaP=0x0D00, pos=0, a=0x4, inactive=[1, 3]),  # K3[15:12]
        dict(deltaP=0x7000, pos=1, a=0x2, inactive=[0, 3]),  # K3[11:8]
        dict(deltaP=0x9000, pos=2, a=0x4, inactive=[1, 3]),  # K3[7:4]
        dict(deltaP=0xD000, pos=3, a=0x4, inactive=[0, 2]),  # K3[3:0]
    ]

    k3_nibs = [0, 0, 0, 0]
    for j in jobs:
        k, kept, margin, scores = recover_one_nibble(
            o,
            deltaP=j["deltaP"],
            target_pos=j["pos"],
            expected_diff=j["a"],
            inactive_zero_positions=j["inactive"],
            N=args.pairs
        )
        k3_nibs[j["pos"]] = k
        print(f"[+] pos={j['pos']} deltaP=0x{j['deltaP']:04x} kept={kept} best=0x{k:x} margin={margin}")

    K3 = (k3_nibs[0] << 12) | (k3_nibs[1] << 8) | (k3_nibs[2] << 4) | k3_nibs[3]
    print(f"[+] Recovered K3 = 0x{K3:04x}")

    proof = K3 ^ o.mask
    print(f"[+] Sending proof = K3 ^ mask = 0x{proof:04x}")

    resp = o.submit(proof)
    print(resp)

if __name__ == "__main__":
    main()

如果你跑出来 margin 很小（比如 0/1）导致某个 nibble 偶尔不稳，就把 -n 稍微加大一点（例如 -n 450，总查询约 3600 仍然 <4096）。

Question

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Python

Thought

Choosing the number of pairs and ensuring accuracy

Ensuring correct pairs and handling socket interactions

Implementing robust socket handling and prompt reading

Handling prompt and output with chunked reading

Optimizing query execution with pipelining

Clarifying server-client interaction

Optimizing query handling and attack simulation

Handling proof submission and reading response

Explaining cryptanalysis and flag extraction

Answer

用什么差分（这题最省事的一组）

直接可用的 exploit（不需要满 4096 次查询）

你可能关心的“此题不是正常 flag”

Question

Python

Thought

Clarifying host:port and script modifications

Clarifying host:port settings

Answer

1）本地怎么跑（可选）

2）解题脚本（solve.py）

3）怎么用

Question

Thought

Refining read logic for prompt handling

Handling newline and prompt byte issues

Improving line reading with output matching

Fixing readline handling and banner skipping

Answer

Question

Thought

Answer

为什么会这样

如果你还遇到第二个坑

Question

Python

Python

Thought

Ensuring stability and refining reading behavior

Finalizing script and handling edge cases

Answer

Share this Q&A