[网鼎杯 2024 semifinal]noisy-nfsr 解题

flag 不在附件里，无法获取，只能解题，以下是解题过程，本地工具api铺助完成，水平小菜，可以扩宽眼界和学习作用，实际无法帮助提交，题目从NSSCTF平台获取。

---

noisy-nfsr 解题报告

1. 题目概览

服务是一个"提交随机挑战 + 会话密钥"的 oracle：

1. 先过一个 4 字符（[A-Za-z0-9]）的 Proof-of-Work。
2. 服务器随机生成：
   • 一个12-bitLFSR（seed, mask，bit11 置 1）用于产生噪声；
   • 两个128-bitLFSR（seeds[0..1],masks[0..1]，bit127 置 1）。
3. 打印两个 128-bit 的masks（seeds不打印）。
4. 选择参数(n, m)，随后可反复选择：
   • [H]it：用当前seeds重置两个 128-bit LFSR，输出m字节密钥流；
   • [S]tand：提交猜的seeds，全对则打印 flag；
   • [Q]uit：退出。

目标：在最多n次交互内恢复两个 128-bitseeds。

2. 组件分析

2.1 LFSR

def __next__(self): b = (self.state & self.mask).bit_count() & 1 self.state = ((self.state << 1) | b) & (2**128 - 1) return b

状态左移、低位补反馈位。记状态转移为T，第t拍反馈位：

b_t = <mask, T^t seed> (GF(2) 内积)

反馈位等于输出位，因此seed -> 输出序列是 GF(2) 线性映射。

2.2 NFSR（组合器）

def __next__(self): b0, b1 = self.lfsr0(1), self.lfsr1(1) return b0 ^ b1 ^ next(self.noise)

输出out = L0 ^ L1 ^ y，其中L0/L1是两个 128-bit LFSR 的比特流，y是噪声流。

2.3 噪声生成器

def NOISE(lfsr, p=2/3): while True: if lfsr(1): yield int(rand() > p) # 以 1/3 概率得到 1 else: yield int(rand() <= p) # 以 2/3 概率得到 1

记 12-bit LFSR 第g拍输出为c_g，噪声位为y_g：

P(y_g = 1^c_g) = 2/3, P(y_g = c_g) = 1/3

即y_g = c_g ^ e_g，其中e_g是以2/3概率取 1 的错误位。

这是本题唯一的信息泄漏点：噪声与一个 12-bit LFSR 强相关。

2.4 三个关键结构性质

1. 每次Hit都会用同一seeds重置两个 128-bit LFSR，但noise生成器只创建一次、在整个连接内持续前进。
2. 每次Hit输出m字节，即P = 8m比特。
3. 第k次Hit的第t位为：

O[k*P + t] = S[t] ^ y_{k*P+t}, S[t] = L0[t] ^ L1[t]

其中S[t]每次都一样，只有噪声在推进。

3. 攻击思路

3.1 差分：消掉未知的 S

把相隔恰好一个Hit的两位异或：

U[g] := O[g] ^ O[g+P] = y_g ^ y_{g+P}

S被完全消掉，且U只与噪声有关：

U[g] = (c_g ^ c_{g+P}) ^ (e_g ^ e_{g+P}) = C[g] ^ f_g

其中C[g] = c_g ^ c_{g+P}，f_g = e_g ^ e_{g+P}。

由P(e=1)=2/3：

P(f_g = 0) = (2/3)^2 + (1/3)^2 = 5/9 P(f_g = 1) = 4/9

即U是C的一个"以 5/9 正确"的带噪观测。

3.2 相关性恢复 12-bit 噪声 LFSR

候选噪声 LFSR 由(mask, seed)（各 12-bit，最高位固定为 1）确定。对候选(M, s)：

c_g = <v_g, s>, v_g = (Tᵀ)^g M

共轭递推（adjoint）：

def adj(y, M): return (y >> 1) ^ (M if (y & 1) else 0)

于是：

C[g] = c_g ^ c_{g+P} = <v_g ^ v_{g+P}, s> = <w_g, s>

用相关性给候选打分：

score(s) = Σ_g [<w_g, s> == U[g]] = N/2 + (1/2) Σ_g (-1)^{U[g] ^ <w_g,s>}

对正确候选该项期望约为N/9，随机候选约为 0。

关键优化：固定M时w_g只依赖M，先做统计：

H[w] = Σ_{g : w_g = w} (-1)^{U[g]}

再对 4096 长的H做一次Fast Walsh–Hadamard Transform：

FWT(H)[s] = Σ_w H[w] (-1)^{<w, s>} = Σ_g (-1)^{U[g] ^ <w_g, s>}

一次 FWT 就同时评估该 mask 下全部 2048 个合法 seed。

外层枚举 2048 个 mask，总计约2048 × (build + O(2¹²·12))，几秒完成。

3.3 多数表决恢复 S

得到正确噪声 LFSRc后，用ŷ_g = 1 ^ c_g作为噪声预测（正确率 2/3），构造：

pred_g = O[g] ^ (1 ^ c_g) = S[t] ^ (1 ^ e_g)

对固定的t，跨所有Hit做多数表决：

S[t] = majority_k ( O[k*P + t] ^ (1 ^ c_{k*P+t}) )

每一票正确率 2/3，h = n-1 = 255票下误码率约10⁻⁸/bit，2048 位几乎必然全对。

3.4 线性求解两个 128-bit 种子

已知两个掩码，L0/L1输出对seed线性。对每一t：

<alpha0_t, seed0> ^ <alpha1_t, seed1> = S[t]

其中alpha0_t / alpha1_t分别是两个 128-bit mask 的 adjoint 轨道(Tᵀ)^t mask。

叠加"seed 最高位为 1"这一仿射约束（把seed = 2¹²⁷ | x代入），得到254 个未知数、P = 2048个方程的超定 GF(2) 线性系统，用位掩码高斯消元求解。

4. 两处退化情形及处理

4.1 种子映射非单射（必须换连接）

极少数随机掩码对会令映射(seed0, seed1) -> S出现核（实测约15%），即存在(δ0, δ1) ≠ 0使L0(δ0) ^ L1(δ1) ≡ 0，从而两个不同的种子对产生完全相同的密钥流，信息论上无法区分。

处理：在读入masks后立即计算该映射的秩rank_sub，若< 254直接重连，换一组掩码。

4.2 噪声 LFSR 的 P-周期歧义

相关性只能确定C[g] = c_g ^ c_{g+P}，因此噪声 LFSR 只被确定到相差一个P-周期序列d（且c ^ d仍是合法 12-bit LFSR 输出）。

用 V-一致性过滤：一个候选只有当它导出的解满足线性系统完全一致（残差为 0）时才接受；若这样的候选不唯一，说明本连接存在等价解释，直接重连。

5. 参数选择

项	值	说明
args索引	1	对应(n, m) = (256, 256)
P = 8m	2048	每次Hit的密钥流长度（bit）
h = n-1	255	先做 255 次Hit，最后一次用于提交
N	12000	相关性使用的U长度；正解约N/9 ≈ 1333，随机约±110
候选阈值	G - 5√N	保证纳入真解，同时排除随机峰

6. 实现

解题脚本：solve_cn.py

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ noisy-nfsr 解题脚本 【密码学分析】 每次 Hit 输出 S[t] = L0[t] ^ L1[t] ^ y[t]，其中： * L0、L1 是两个 128-bit LFSR，**每次 Hit 都用同一组 seeds 重置**； * y 是噪声流，由一个独立的 12-bit LFSR 产生（其 seed/mask 都是 12 位、 最高位 bit11 为 1，且不打印出来）。 12-bit 噪声 LFSR 的第 g 拍输出 c_g 让噪声位满足： P(y_g = 1 ^ c_g) = 2/3 , P(y_g = c_g) = 1/3 也就是 y_g = c_g ^ e_g，e_g 以 2/3 概率取 1。 因为两个 128-bit LFSR 每次 Hit 都重置、而噪声发生器在整条连接里持续前进， 所以把相隔一个 Hit（P = 8*m 位）的两比特异或即可消掉 S： O[g] ^ O[g+P] = y_g ^ y_{g+P} = (c_g ^ c_{g+P}) ^ (e_g ^ e_{g+P}) 这就得到 C[g] = c_g ^ c_{g+P} 的一个“带噪观测”，于是可以用相关性攻击恢复 那个 12-bit 噪声 LFSR（对每个候选 mask 做一次快速沃尔什-哈达玛变换，一次性 评估全部 4096 个 seed）。 知道噪声 LFSR 后就能预测 y，对每条 Hit 做多数表决还原 S[t]（每票 2/3 正确）， 再解 GF(2) 线性方程组 L0[seed0][t] ^ L1[seed1][t] = S[t] 得到两个 128-bit 种子（两者最高位恒为 1）。 【两处退化情形】 1. 映射 (seed0, seed1) -> S 偶尔不是单射（约 15% 的随机掩码对），会出现多个 等价种子对，信息论上无法区分 —— 用 `rank_sub` 提前检测，命中就换连接。 2. 噪声 LFSR 只能确定到一个 P-周期的等价类 —— 枚举所有候选，只接受能令线性 系统**一致且唯一**的那个，否则重连。 """ import sys import os import re import time import socket import hashlib import string import itertools import subprocess import shutil import tempfile try: # Windows 控制台（cp936/cp1252）下避免打印崩溃 sys.stdout.reconfigure(encoding='utf-8', errors='replace') sys.stderr.reconfigure(encoding='utf-8', errors='replace') except Exception: pass # args 取索引 1 -> (n, m) = (256, 256) ARGS_INDEX = 1 M_BYTES = 256 P_HIT = 8 * M_BYTES # 每次 Hit 产出的密钥流比特数 N_HITS = 256 # 主循环次数 n H_HITS = N_HITS - 1 # 先做 255 次 Hit，最后一次用于提交 N_RECOVER = 12000 # 相关性攻击使用的 U 长度 MAX_TRIES = 12 # 最多重连次数 PLACEHOLDER_FLAG = 'flag{no_local_flag_file__seeds_are_correct}' # -------------------------------------------------------------------------- # GF(2) 基础工具 # -------------------------------------------------------------------------- def adj(y, M): """共轭（adjoint）LFSR 的一步：计算 (T^T) y。""" return (y >> 1) ^ (M if (y & 1) else 0) def fwt(H): """原地快速沃尔什-哈达玛变换（长度 4096）。""" mm = 1 n = len(H) while mm < n: for i in range(0, n, mm * 2): for j in range(i, i + mm): x = H[j] y = H[j + mm] H[j] = x + y H[j + mm] = x - y mm *= 2 return H def _seq_info(M): """返回掩码 M 的共轭轨道 (序列, 预周期长度, 周期长度)。""" seen = {} v = [] y = M while True: s = seen.get(y, -1) if s >= 0: return v, s, len(v) - s seen[y] = len(v) v.append(y) y = adj(y, M) def _build_H(M, signs, P, N): """累加 H[w] = Σ_{g : v_g ^ v_{g+P} = w} signs[g]。""" v, start, L = _seq_info(M) # 周期性部分：r 为 (g-start) mod L wres = [v[start + r] ^ v[start + ((r + P) % L)] for r in range(L)] H = [0] * 4096 # 预周期（start 之前）的部分单独处理 for g in range(start): t2 = g + P w = v[g] ^ v[t2] if t2 < len(v) else v[g] ^ v[start + ((t2 - start) % L)] H[w] += signs[g] r = 0 for g in range(start, N): H[wres[r]] += signs[g] r += 1 if r == L: r = 0 return H def scan(U, P, N): """对每个 12-bit mask 返回 (最高相关分, mask, 最优 seed)。""" signs = [1 if x == 0 else -1 for x in U] res = [] for M in range(2048, 4096): # seed/mask 最高位为 1，只需枚举 2048..4095 H = fwt(_build_H(M, signs, P, N)) bi = 2048 bv = H[2048] for s in range(2049, 4096): if H[s] > bv: bv = H[s] bi = s res.append((bv, M, bi)) return res def enumerate_hi(M, U, P, N, thr): """列出 mask M 下所有相关分 >= thr 的 seed。""" signs = [1 if x == 0 else -1 for x in U] H = fwt(_build_H(M, signs, P, N)) return [(H[s], M, s) for s in range(2048, 4096) if H[s] >= thr] def rank_sub(mask0, mask1, P): """(seed0,seed1) -> S 映射的秩（两个种子最高位固定为 1）。""" LOW = (1 << 127) - 1 a0, a1 = mask0, mask1 piv = {} for _ in range(P): c = (a0 & LOW) | ((a1 & LOW) << 127) a0 = (a0 >> 1) ^ (mask0 if (a0 & 1) else 0) a1 = (a1 >> 1) ^ (mask1 if (a1 & 1) else 0) cc = c while cc: b = cc.bit_length() - 1 if b in piv: cc ^= piv[b] else: piv[b] = cc break return len(piv) def solve_residual(R, P, mask0, mask1): """解线性系统，返回 (不一致方程条数, seed0, seed1)。""" LOW = (1 << 127) - 1 a0, a1 = mask0, mask1 piv = {} resid = 0 for t in range(P): # 第 t 个方程：<alpha0_t, seed0> ^ <alpha1_t, seed1> = R[t] rhs = R[t] ^ ((a0 >> 127) & 1) ^ ((a1 >> 127) & 1) # 去掉最高位（恒为 1）的贡献 coeff = (a0 & LOW) | ((a1 & LOW) << 127) a0 = (a0 >> 1) ^ (mask0 if (a0 & 1) else 0) a1 = (a1 >> 1) ^ (mask1 if (a1 & 1) else 0) c, r = coeff, rhs while c: b = c.bit_length() - 1 pv = piv.get(b) if pv is None: piv[b] = (c, r) break c ^= pv[0] r ^= pv[1] else: if r: resid += 1 # 系数消光但右端为 1 -> 不一致 # 回代：按主元从低位到高位求值 sol = 0 for b in sorted(piv): c, r = piv[b] if r ^ ((c & sol).bit_count() & 1): sol |= (1 << b) return resid, (sol & LOW) | (1 << 127), ((sol >> 127) & LOW) | (1 << 127) def _majority_R(out, h, P, mask, seed): """用候选噪声 LFSR 预测噪声，再对每条 Hit 做多数表决得到 R。""" cnt = [0] * P st = seed t = 0 for g in range(h * P): b = (st & mask).bit_count() & 1 # 12-bit 噪声 LFSR 输出 c_g st = ((st << 1) | b) & 0xFFF cnt[t] += out[g] ^ (1 ^ b) # 预测 ŷ = 1 ^ c_g t += 1 if t == P: t = 0 half = h / 2.0 return bytearray(1 if c > half else 0 for c in cnt) def solve_keystream(out, h, P, N, mask0, mask1, verbose=True): """恢复两个种子；无法唯一确定时返回 None（由调用方重连）。""" # 差分：U[g] = O[g] ^ O[g+P] = y_g ^ y_{g+P} U = bytes(out[g] ^ out[g + P] for g in range(N)) t0 = time.time() per = scan(U, P, N) if verbose: print("[*] 噪声相关性扫描完成，用时 %.1fs" % (time.time() - t0)) G = max(bv for bv, _, _ in per) thr = G - 5 * int(N ** 0.5) # 阈值：真解约在 G 附近，随机峰约 ±√N cands = [] for bv, M, s in per: if bv >= thr: cands += enumerate_hi(M, U, P, N, thr) if verbose: print("[*] 阈值以上共有 %d 个候选噪声 LFSR" % len(cands)) # 只保留能让线性系统“完全一致”的候选 good = [] for val, M, s in cands: R = _majority_R(out, h, P, M, s) resid, x0, x1 = solve_residual(R, P, mask0, mask1) if resid == 0: good.append((x0, x1)) if len(good) == 1: return good[0] if verbose: print("[!] 一致候选有 %d 个（存在等价解）-> 重连" % len(good)) return None def solve_pow(suffix, target_hex): """暴力 4 字符前缀使 sha256(prefix + suffix) == target。""" target = bytes.fromhex(target_hex) alph = (string.ascii_letters + string.digits).encode() suf = suffix.encode() for tup in itertools.product(alph, repeat=4): cand = bytes(tup) if hashlib.sha256(cand + suf).digest() == target: return cand.decode() return None # -------------------------------------------------------------------------- # 传输层：socket（远程）或子进程（本地） # -------------------------------------------------------------------------- class SocketTube: def __init__(self, host, port, timeout=300): self.s = socket.create_connection((host, port), timeout=timeout) self.buf = b"" def recvuntil(self, marker): if isinstance(marker, str): marker = marker.encode() while marker not in self.buf: d = self.s.recv(65536) if not d: raise EOFError("连接已关闭") self.buf += d i = self.buf.index(marker) + len(marker) res, self.buf = self.buf[:i], self.buf[i:] return res def sendline(self, x): if isinstance(x, str): x = x.encode() self.s.sendall(x + b"\n") def close(self): try: self.s.close() except OSError: pass class ProcTube: def __init__(self, argv, cwd): env = dict(os.environ) env['PYTHONIOENCODING'] = 'utf-8' # 子进程要打印 '🏁'，控制台可能是 gbk env['PYTHONUTF8'] = '1' self.p = subprocess.Popen(argv, cwd=cwd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env) self.buf = b"" def recvuntil(self, marker): if isinstance(marker, str): marker = marker.encode() while marker not in self.buf: d = self.p.stdout.read1(65536) if not d: raise EOFError("子进程已退出；最后输出: %r" % self.buf[-600:]) self.buf += d i = self.buf.index(marker) + len(marker) res, self.buf = self.buf[:i], self.buf[i:] return res def sendline(self, x): if isinstance(x, str): x = x.encode() self.p.stdin.write(x + b"\n") self.p.stdin.flush() def close(self): try: self.p.kill() except OSError: pass # -------------------------------------------------------------------------- # 一次完整交互 # -------------------------------------------------------------------------- def run_attempt(tube, verbose=True): """跑一次协议流程，成功返回 flag 字符串，否则返回 None。""" # 1) Proof-of-Work banner = tube.recvuntil('XXXX>') m = re.search(rb'sha256\(XXXX \+ (.{16})\) == ([0-9a-f]{64})', banner) if not m: raise RuntimeError("无法解析 PoW: %r" % banner[:200]) suffix, target = m.group(1).decode(), m.group(2).decode() if verbose: print("[*] 解 PoW，后缀 = %r" % suffix) tube.sendline(solve_pow(suffix, target)) # 2) 读取打印出来的两个 128-bit 掩码 chunk = tube.recvuntil('args>') nums = re.findall(rb'\d{10,}', chunk) if len(nums) < 2: raise RuntimeError("无法解析 masks: %r" % chunk[:200]) mask0, mask1 = int(nums[0]), int(nums[1]) if verbose: print("[*] mask0=%d mask1=%d" % (mask0, mask1)) # 3) 秩门控：非单射就换连接（否则种子对不唯一） if rank_sub(mask0, mask1, P_HIT) != 254: if verbose: print("[!] 该掩码对下种子映射非单射 -> 重连") return None # 4) 选参数，收集 255 次 Hit 的密钥流 tube.sendline(str(ARGS_INDEX)) tube.recvuntil('| inp>') # 先吃掉第一个菜单提示 out = bytearray() for _ in range(H_HITS): tube.sendline('h') resp = tube.recvuntil('| inp>') hx = re.search(rb"bits\.hex\(\) = '([0-9a-f]+)'", resp) out += bytes.fromhex(hx.group(1).decode()) if verbose: print("[*] 已收集 %d 字节密钥流" % len(out)) # 字节 -> 比特（每字节高位在前，和程序 self(8) 的顺序一致） bits = bytearray() for byte in out: for k in range(7, -1, -1): bits.append((byte >> k) & 1) # 5) 恢复种子 seeds = solve_keystream(bits, H_HITS, P_HIT, N_RECOVER, mask0, mask1, verbose) if seeds is None: return None seed0, seed1 = seeds if verbose: print("[*] 恢复出 seed0=%d seed1=%d" % (seed0, seed1)) # 6) 提交并读 flag tube.sendline('s') tube.recvuntil('| seed>') tube.sendline(str(seed0)) tube.recvuntil('| seed>') tube.sendline(str(seed1)) try: tail = tube.recvuntil('| Bye') except EOFError: # task.py 崩了（裸 except 会打印 '| Nah' 后退出） tail = bytes(getattr(tube, 'buf', b'')) if verbose: print("[!] 子进程/服务在 '| Bye' 之前就退出了") txt = tail.decode('utf-8', 'replace') if verbose: print(txt.encode('ascii', 'replace').decode('ascii')) # 控制台安全打印 m = re.search(r'flag\{[^}\n]*\}', txt) if m: return m.group(0).strip() m = re.search('\U0001F3C1\\s*(.+)', txt) # 🏁 之后的内容 return m.group(1).strip() if m else None def _make_local_dir(src_dir, flag_path=None): """把 task.py / utils.py 复制到临时目录，并： - 把 POSIX 专有的 signal.alarm() 换成 pass，使题目在 Windows 也能跑； - 把裸 except: 换成会打印真实 traceback 的版本（便于排错）； - 保证存在一个 flag 文件（没有就写占位）。 用户自己的文件不会被修改。""" d = tempfile.mkdtemp(prefix='nfsr_local_') shutil.copy(os.path.join(src_dir, 'utils.py'), d) with open(os.path.join(src_dir, 'task.py'), 'r', encoding='utf-8') as f: src = f.read() src = re.sub(r'^[ \t]*signal\.alarm\([^)]*\)[ \t]*$', ' pass # signal.alarm() 已被移除（仅 POSIX 可用）', src, flags=re.M) src = re.sub(r'^([ \t]*)except:[ \t]*$', r'\1except Exception:\n\1 import traceback; traceback.print_exc()', src, flags=re.M) with open(os.path.join(d, 'task.py'), 'w', encoding='utf-8') as f: f.write(src) flag_src = flag_path or os.path.join(src_dir, 'flag') if os.path.exists(flag_src): shutil.copy(flag_src, os.path.join(d, 'flag')) src_desc = flag_src else: with open(os.path.join(d, 'flag'), 'w') as f: f.write(PLACEHOLDER_FLAG) src_desc = '占位符（本地没有 flag 文件，仅验证种子是否正确）' print("[*] 临时副本目录: %s (flag: %s)" % (d, src_desc)) return d def main(): args = sys.argv[1:] local = '--local' in args args = [a for a in args if a != '--local'] flag_path = None if '--flag' in args: k = args.index('--flag') flag_path = args[k + 1] del args[k:k + 2] run_dir = None if local: if not os.path.exists('task.py'): print("[-] --local 需要当前目录下有 task.py（以及 utils.py）") print("[-] 当前目录: %s" % os.getcwd()) return 3 run_dir = _make_local_dir(os.getcwd(), flag_path) for attempt in range(1, MAX_TRIES + 1): print("[*] === 第 %d/%d 次尝试 ===" % (attempt, MAX_TRIES)) if local: tube = ProcTube([sys.executable, '-u', 'task.py'], cwd=run_dir) else: if len(args) < 2: print("用法: solve_cn.py <host> <port> （或 --local [--flag 文件]）") return 2 tube = SocketTube(args[0], int(args[1])) try: flag = run_attempt(tube) except KeyboardInterrupt: raise except Exception as e: import traceback print("[!] 第 %d 次尝试失败: %r" % (attempt, e)) traceback.print_exc() flag = None finally: tube.close() if flag: print("[+] FLAG: %s" % flag) if flag == PLACEHOLDER_FLAG: print("[i] 这只是占位符：本地没找到 flag 文件，但种子已正确恢复。") print("[i] 连远程（solve_cn.py <host> <port>）或用 --flag <文件> 指定真 flag。") return 0 print("[-] 已用尽尝试次数") return 1 if __name__ == "__main__": sys.exit(main())

solve_cn.py 命令行使用方法：

---

两种模式：

1. 远程模式（连真实 CTF 服务器）

python solve_cn.py <host> <port>

示例：

python solve_cn.py 127.0.0.1 8080

2. 本地模式（在本地跑 task.py 验证逻辑）

python solve_cn.py --local

需要当前目录下有task.py和utils.py。

可选参数：

python solve_cn.py --local --flag <flag文件路径>

指定本地 flag 文件，没有的话用占位符flag{no_local_flag_file__seeds_are_correct}。

---

参数说明：

参数	说明
<host>	远程服务器地址
<port>	远程服务器端口
--local	本地运行模式（复制 task.py 到临时目录，自动移除 POSIX 专有的 signal.alarm）
--flag <路径>	指定 flag 文件路径（仅本地模式用）

---