mirror of
https://github.com/Mabbs/mabbs.github.io
synced 2026-10-01 06:17:18 +08:00
303f5df01c
- /_data/other_repo_list.csv - /_data/links.csv - /_posts/2026-10-01-quine.md - /_tools/blogquine.py - /assets/js/pjax.js
1352 lines
53 KiB
Markdown
1352 lines
53 KiB
Markdown
---
|
||
layout: post
|
||
title: 如何制作一个“完整”的博客压缩包？
|
||
tags: [压缩包, Quine, LZMA2, 7-Zip]
|
||
---
|
||
|
||
让AI做出真正的“完整”将不再是难事……<!--more-->
|
||
|
||
# 起因
|
||
在上次[用AI制作了各种东西](/2026/09/01/vibe-coding2.html)之后，我已经完全理解了AI确实是无所不能的。既然如此，那就让它帮我解决曾经未能解决的事情吧？
|
||
去年，我为了让下载全站压缩包的按钮不断链，让这个压缩包也包含它本身而研究了[ZIP Quine](/2025/09/01/quine.html)，但限于DEFLATE的回溯窗口大小没能做到……但那是人做的东西，人还是太弱小了，现在换AI来试试，也许一切将变得不一样？
|
||
|
||
# 制作基于LZMA2的博客压缩包
|
||
首先，我把[Ruben Van Mello](https://github.com/ruvmello)写的那篇论文《[A Generator for Recursive Zip Files](https://www.mdpi.com/2076-3417/14/21/9797)》以及生成器[zip-quine-generator](https://github.com/ruvmello/zip-quine-generator)发给了AI，问它这个限制是不是真的，是不是真的无法创造出超过32KiB的ZIP Quine？没过多久它分析完了，告诉我确实存在这样的问题，但并不是无法解决的，最简单的办法就是换个回溯窗口更大的压缩算法。我看了一下它给我列出的几个算法，看起来LZMA2和zstd比较符合要求。不过zstd感觉不是很知名，一般的解压软件应该处理不了，所以我就选择让它基于LZMA2来制作了。
|
||
不过LZMA2只是算法，还得选个容器，考虑到7z算是最出名，而且LZMA2本来就是7-Zip的作者开发的，所以就直接让它做7z版本的了。另外因为[Nate Choe](https://github.com/NateChoe1)做了通过扩展欧几里得求逆元来计算CRC32的[PR](https://github.com/ruvmello/zip-quine-generator/pull/3)，所以我也告诉它要用这种方法在多项式时间内解出CRC32的值，而不是爆破。就这样AI花了两个小时左右成功把代码写出来了，效果非常完美，另外它还偷懒地将计算CRC32的算法改成了高斯消元法😆，因为7z格式有3个相互影响的CRC32值，用扩展欧几里得会更复杂一点，不过对7z来说不写入文件的CRC32也不影响，只是会不校验罢了，至少它还是按我要求做了……但其实这个问题也并不是完全不能用扩展欧几里得算法，后来我又强烈要求了一下，AI还是给我写出来了。以下是代码片段，有兴趣的人可以参考一下（不过这个看起来写的就是很复杂，所以我自己只用了高斯消元法）：
|
||
```python
|
||
RU_POLY = 0x104C11DB7 # x^32+x^26+x^23+...+1（含 x^32 首项）
|
||
|
||
|
||
def _find_probe(p):
|
||
ret = 0
|
||
for i in range(64):
|
||
if (1 << i) & p:
|
||
ret = i
|
||
return ret
|
||
|
||
|
||
def _mul_raw(p1, p2):
|
||
"""多项式乘法（无模）。"""
|
||
ret = 0
|
||
probe = _find_probe(p1)
|
||
for i in range(64):
|
||
if p2 & (1 << i):
|
||
assert probe + i < 64, "多项式乘法溢出"
|
||
ret ^= p1 << i
|
||
return ret
|
||
|
||
|
||
def _poly_divmod(dividend, divisor):
|
||
"""多项式长除法，返回 (商, 余)。"""
|
||
probe = _find_probe(divisor)
|
||
probe_bit = 1 << probe
|
||
quot = 0
|
||
rem = dividend
|
||
for i in range(63 - probe, -1, -1):
|
||
if rem & (probe_bit << i):
|
||
quot |= 1 << i
|
||
rem ^= divisor << i
|
||
return quot, rem
|
||
|
||
|
||
def _mul(p1, p2, mod):
|
||
"""p1*p2 % mod。"""
|
||
return _poly_divmod(_mul_raw(p1, p2), mod)[1]
|
||
|
||
|
||
def _xgcd(p1, p2):
|
||
"""扩展欧几里得：返回 (k1,k2,gcd)，满足 p1*k1+p2*k2=gcd。"""
|
||
if _find_probe(p1) < _find_probe(p2):
|
||
k1, k2, g = _xgcd(p2, p1)
|
||
return k2, k1, g
|
||
if p2 == 0:
|
||
return p1, 0, p1
|
||
q, r = _poly_divmod(p1, p2)
|
||
c1, c2, g = _xgcd(p2, r)
|
||
return c2, c1 ^ _mul_raw(c2, q), g
|
||
|
||
|
||
def _minv(p, mod):
|
||
"""p 在环上的乘法逆元；不存在时返回 0。"""
|
||
k1, _, g = _xgcd(p, mod)
|
||
if g != 1:
|
||
return 0
|
||
return _poly_divmod(k1, mod)[1]
|
||
|
||
|
||
def _bitrev32(x):
|
||
"""32 位比特反转。CRC 的输出循环冗余（binascii 反射）↔ 本环（非反射）互为 bit-reverse。"""
|
||
return int(format(x & 0xFFFFFFFF, "032b")[::-1], 2)
|
||
|
||
|
||
def _ring_pow_x(k):
|
||
"""环内 x^k mod P（P=RU_POLY），二进制幂 O(log k)。变量列就是单个环幂。"""
|
||
res, base, mod, e = 1, 2, RU_POLY, k
|
||
while e:
|
||
if e & 1:
|
||
res = _mul(res, base, mod)
|
||
base = _mul(base, base, mod)
|
||
e >>= 1
|
||
return res
|
||
|
||
|
||
def solve_crc_system(files):
|
||
"""解析解 CRC 定点系统
|
||
|
||
- 常数列 ：对“变量清零”的数据做一次 binascii.crc32，再 bit-reverse 32 位
|
||
（binascii 反射 CRC 与本环非反射 CRC 互为 bit-reverse，见 _bitrev32）。
|
||
- 变量列 ：单个环幂 x^(32 + 8*(len-pos-4))，二进制幂 O(log N) 一步算出；
|
||
同一变量出现在多处时按位异或累加。
|
||
各自 O(#rows) 次 CRC + 少量环幂即可建出整矩阵，Gauss 消元不变。
|
||
"""
|
||
n = len(files)
|
||
mod = RU_POLY
|
||
matrix = [[0] * (n + 1) for _ in range(n)]
|
||
|
||
for file in range(n):
|
||
data = files[file][0]
|
||
offsets = files[file][1]
|
||
N = len(data)
|
||
zero = bytearray(data)
|
||
for pos in offsets: # 未知量占 4 字节，清零后求常数项
|
||
zero[pos:pos + 4] = b"\x00\x00\x00\x00"
|
||
matrix[file][n] = _bitrev32(crc32(bytes(zero)))
|
||
for pos, vid in offsets.items():
|
||
rest = N - pos - 4 # 变量被 4 字节（x^32）+ 其后字节（x^8/个）推到底
|
||
matrix[file][vid] ^= _ring_pow_x(32 + 8 * rest)
|
||
matrix[file][file] ^= 1
|
||
|
||
for sr in range(n):
|
||
ex = sr
|
||
while ex < n and matrix[ex][sr] == 0:
|
||
ex += 1
|
||
if ex == n:
|
||
raise RuntimeError("CRC 多项式系统奇异 @row%d" % sr)
|
||
if ex != sr:
|
||
matrix[sr], matrix[ex] = matrix[ex], matrix[sr]
|
||
inv = _minv(matrix[sr][sr], mod)
|
||
for er in range(sr + 1, n):
|
||
mult = _mul(inv, matrix[er][sr], mod)
|
||
for ec in range(n + 1):
|
||
matrix[er][ec] ^= _mul(mult, matrix[sr][ec], mod)
|
||
|
||
res = [0] * n
|
||
for row in range(n - 1, -1, -1):
|
||
v = matrix[row][n]
|
||
for col in range(row + 1, n):
|
||
v ^= _mul(res[col], matrix[row][col], mod)
|
||
res[row] = _mul(v, _minv(matrix[row][row], mod), mod)
|
||
|
||
return [_impl_val(res[i]) for i in range(n)]
|
||
|
||
def _impl_val(math_val):
|
||
imp = 0
|
||
for bit in range(32):
|
||
if math_val & (1 << bit):
|
||
imp |= 1 << (31 - bit)
|
||
return imp
|
||
|
||
class BlogQuine:
|
||
def solve_crc(self, F, lay, seed):
|
||
T, h, d = lay["T"], lay["h"], lay["d"]
|
||
k, n_sub, crc_base = lay["k"], lay["n_sub"], lay["crc_base"]
|
||
hoff = lay["total"] - h
|
||
vout_f = lay["total"] - 2 * T
|
||
hcopy = vout_f + (T - h)
|
||
# 已知 CRC：seed + 各文件（quine 的是未知量 D）
|
||
known = [crc32(seed)] + [crc32(data) for _, data in self.files]
|
||
assert len(known) == n_sub - 1
|
||
for i, e in enumerate(known):
|
||
struct.pack_into("<I", F, hoff + crc_base + 4 * i, e)
|
||
struct.pack_into("<I", F, hcopy + crc_base + 4 * i, e)
|
||
# 未知量 D N S：D=整个文件、N=头部 F[hoff:hoff+h]、S=签名 F[12:32]
|
||
dpos = crc_base + 4 * (n_sub - 1)
|
||
ncopy = 32 + 3 * k + d # C1k 载荷里 F[0:35) 副本起点
|
||
groups = [[hoff + dpos, hcopy + dpos], # D 出现两处
|
||
[28, ncopy + 28], # N
|
||
[8, ncopy + 8]] # S
|
||
|
||
# 构造 3 个“文件”喂给解析解系统：
|
||
# 文件0 = 整个 F，三个未知量各自出现在 groups 两处
|
||
# 文件1 = 头部块，只有 D 出现在相对 dpos（另一处 hcopy+dpos 在头外）
|
||
# 文件2 = 签名块，只有 N 出现在相对 16（28-12）
|
||
whole = bytes(F)
|
||
hdr = bytes(F[hoff:hoff + h])
|
||
sig = bytes(F[12:32])
|
||
f0_off = {hoff + dpos: 0, hcopy + dpos: 0,
|
||
28: 1, ncopy + 28: 1,
|
||
8: 2, ncopy + 8: 2}
|
||
f1_off = {dpos: 0}
|
||
f2_off = {28 - 12: 1}
|
||
vals = solve_crc_system([(whole, f0_off), (hdr, f1_off), (sig, f2_off)])
|
||
|
||
for g in range(3):
|
||
for p in groups[g]:
|
||
F[p:p + 4] = struct.pack("<I", vals[g])
|
||
real = (crc32(F), crc32(F[hoff:hoff + h]), crc32(F[12:32]))
|
||
assert real == tuple(vals), "CRC 定点失败: %s != %s" % (real, vals)
|
||
return vals
|
||
```
|
||
|
||
于是我第一时间就把原来的TGZ压缩命令换掉，换成了AI给我写的[blogquine.py](https://github.com/Mabbs/mabbs.github.io/blob/master/_tools/blogquine.py)，现在就可以通过[这里](https://mayx.eu.org/MayxBlog.7z)下载到“完整”包含我博客所有内容的压缩包了。
|
||
不过唯一的问题就是这样做出来的压缩包并没有压缩😂，相当于给做成了普通的归档了。当然我的博客本身倒是不大，没压缩也多不了多少空间，但相比于能做出“完整”的效果来说，这点浪费的空间也是小问题了。
|
||
## 对TXZ格式的尝试
|
||
在做完7z格式的压缩包之后，我发现了一个问题，虽然7z确实很流行，但是在Linux下解压起来有点麻烦，7-Zip历史上主要面向Windows，Linux上长期以来更多依赖p7zip等第三方移植，因此生态集成度不如tar.xz，想要解压7z文件还得额外安装。
|
||
不过Linux下也有个支持LZMA2算法的压缩软件，那就是前些年出过[后门](https://tukaani.org/xz-backdoor/)的XZ Utils，配合tar就可以做出TXZ（tar.xz）文件，甚至用我[博客终端](https://mabbs.github.io/linux/)中的BusyBox也能解压。我想了一下反正有AI，干脆一句话让AI帮我把blogquine.py改成tar.xz格式的好了，结果倒也没费多少功夫，AI就这样写出来了：
|
||
|
||
<details markdown="1">
|
||
<summary markdown="span">
|
||
Show Code
|
||
</summary>
|
||
|
||
```python
|
||
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import argparse
|
||
import binascii
|
||
import io
|
||
import lzma
|
||
import os
|
||
import struct
|
||
import sys
|
||
import tarfile
|
||
import time
|
||
import ctypes
|
||
import ctypes.util
|
||
|
||
# ============================================================
|
||
# Part 1: 最小 LZMA1 range coder（只做编码）
|
||
# ============================================================
|
||
|
||
kNumBitModelTotalBits = 11
|
||
kBitModelTotal = 1 << kNumBitModelTotalBits # 2048
|
||
kNumMoveBits = 5
|
||
kTopValue = 1 << 24 # 0x1000000
|
||
|
||
kNumStates = 12
|
||
kNumLitStates = 7
|
||
kNumPosBitsMax = 4
|
||
kNumLenToPosStates = 4
|
||
kNumAlignBits = 4
|
||
kEndPosModelIndex = 14
|
||
kNumFullDistances = 1 << (kEndPosModelIndex >> 1) # 128
|
||
kMatchMinLen = 2
|
||
kNumLowLenBits = 3
|
||
kNumMidLenBits = 3
|
||
kNumHighLenBits = 8
|
||
kNumLowLenSymbols = 1 << kNumLowLenBits # 8
|
||
kNumMidLenSymbols = 1 << kNumMidLenBits # 8
|
||
kNumPosSlotBits = 6
|
||
|
||
PROB_INIT = kBitModelTotal >> 1 # 1024
|
||
|
||
|
||
class RangeEncoder:
|
||
"""LZMA 的区间编码器。low 是 64 位（要容纳进位），range 是 32 位。"""
|
||
|
||
def __init__(self):
|
||
self.low = 0
|
||
self.range = 0xFFFFFFFF
|
||
self.cache = 0
|
||
self.cache_size = 1 # 初值 1 -> 第一个输出字节恒为 0x00
|
||
self.buf = bytearray()
|
||
|
||
def _shift_low(self):
|
||
if (self.low & 0xFFFFFFFF) < 0xFF000000 or (self.low >> 32) != 0:
|
||
temp = self.cache
|
||
while True:
|
||
self.buf.append((temp + (self.low >> 32)) & 0xFF)
|
||
temp = 0xFF
|
||
self.cache_size -= 1
|
||
if self.cache_size == 0:
|
||
break
|
||
self.cache = (self.low >> 24) & 0xFF
|
||
self.cache_size += 1
|
||
self.low = ((self.low & 0xFFFFFFFF) << 8) & 0xFFFFFFFF
|
||
|
||
def encode_bit(self, probs, idx, bit):
|
||
p = probs[idx]
|
||
bound = (self.range >> kNumBitModelTotalBits) * p
|
||
if bit == 0:
|
||
self.range = bound
|
||
probs[idx] = p + ((kBitModelTotal - p) >> kNumMoveBits)
|
||
else:
|
||
self.low += bound
|
||
self.range -= bound
|
||
probs[idx] = p - (p >> kNumMoveBits)
|
||
while self.range < kTopValue:
|
||
self.range = (self.range << 8) & 0xFFFFFFFF
|
||
self._shift_low()
|
||
|
||
def encode_direct_bits(self, value, num_bits):
|
||
for i in range(num_bits - 1, -1, -1):
|
||
self.range >>= 1
|
||
if (value >> i) & 1:
|
||
self.low += self.range
|
||
while self.range < kTopValue:
|
||
self.range = (self.range << 8) & 0xFFFFFFFF
|
||
self._shift_low()
|
||
|
||
def bittree_encode(self, probs, off, num_bits, symbol):
|
||
m = 1
|
||
for i in range(num_bits - 1, -1, -1):
|
||
bit = (symbol >> i) & 1
|
||
self.encode_bit(probs, off + m, bit)
|
||
m = (m << 1) | bit
|
||
|
||
def bittree_reverse_encode(self, probs, off, num_bits, symbol):
|
||
m = 1
|
||
for i in range(num_bits):
|
||
bit = symbol & 1
|
||
symbol >>= 1
|
||
self.encode_bit(probs, off + m, bit)
|
||
m = (m << 1) | bit
|
||
|
||
def finish(self):
|
||
for _ in range(5):
|
||
self._shift_low()
|
||
return bytes(self.buf)
|
||
|
||
|
||
def _pos_slot_and_bits(d):
|
||
if d < 4:
|
||
return d, 0, 0
|
||
for slot in range(4, 64):
|
||
n = (slot >> 1) - 1
|
||
base = (2 | (slot & 1)) << n
|
||
if d < base + (1 << n):
|
||
return slot, n, d - base
|
||
raise ValueError("distance too large: %d" % d)
|
||
|
||
|
||
# pb 必须与写进每个 LZMA2 chunk 头的 PROPS_BYTE(0x5D) 一致。
|
||
# lc/lp 只影响 literal 编码的概率表索引，而本编码器不实现 literal 路径，故无需常量。
|
||
PB = 2
|
||
|
||
|
||
class LzmaEncoder:
|
||
"""只实现 quine 用到的两条路径：match 与 rep0-match。
|
||
|
||
literal / rep-g1 / rep-g2 三条分支从未被构造，因此它们对应的概率表
|
||
（p_lit、p_is_rep_g1、p_is_rep_g2）以及 prev_byte 都不必存在。
|
||
reps 同理：编码器从不回读（rep0 的距离由解码器自行维护），故不保存。
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.pos_mask = (1 << PB) - 1
|
||
self.rc = RangeEncoder()
|
||
self.pos = 0
|
||
self.state = 0
|
||
n = PROB_INIT
|
||
self.p_is_match = [n] * (kNumStates << kNumPosBitsMax)
|
||
self.p_is_rep = [n] * kNumStates
|
||
self.p_is_rep_g0 = [n] * kNumStates
|
||
self.p_rep0_long = [n] * (kNumStates << kNumPosBitsMax)
|
||
self.p_pos_slot = [n] * (kNumLenToPosStates << kNumPosSlotBits)
|
||
self.p_spec_pos = [n] * (kNumFullDistances - kEndPosModelIndex)
|
||
self.p_align = [n] * (1 << kNumAlignBits)
|
||
self.p_len_choice = [n] * 2
|
||
self.p_len_low = [n] * (16 * kNumLowLenSymbols)
|
||
self.p_len_mid = [n] * (16 * kNumMidLenSymbols)
|
||
self.p_len_high = [n] * (1 << kNumHighLenBits)
|
||
self.p_rep_len_choice = [n] * 2
|
||
self.p_rep_len_low = [n] * (16 * kNumLowLenSymbols)
|
||
self.p_rep_len_mid = [n] * (16 * kNumMidLenSymbols)
|
||
self.p_rep_len_high = [n] * (1 << kNumHighLenBits)
|
||
|
||
@property
|
||
def pos_state(self):
|
||
return self.pos & self.pos_mask
|
||
|
||
def _encode_len(self, choice, low, mid, high, length):
|
||
ps = self.pos_state
|
||
l = length - kMatchMinLen
|
||
if l < kNumLowLenSymbols:
|
||
self.rc.encode_bit(choice, 0, 0)
|
||
self.rc.bittree_encode(low, ps << kNumLowLenBits, kNumLowLenBits, l)
|
||
else:
|
||
self.rc.encode_bit(choice, 0, 1)
|
||
l -= kNumLowLenSymbols
|
||
if l < kNumMidLenSymbols:
|
||
self.rc.encode_bit(choice, 1, 0)
|
||
self.rc.bittree_encode(mid, ps << kNumMidLenBits, kNumMidLenBits, l)
|
||
else:
|
||
self.rc.encode_bit(choice, 1, 1)
|
||
self.rc.bittree_encode(high, 0, kNumHighLenBits, l - kNumMidLenSymbols)
|
||
|
||
def _write_dist(self, dist, lts):
|
||
d = dist - 1
|
||
slot, n, low_bits = _pos_slot_and_bits(d)
|
||
self.rc.bittree_encode(self.p_pos_slot, lts << kNumPosSlotBits,
|
||
kNumPosSlotBits, slot)
|
||
if slot >= 4:
|
||
if slot < kEndPosModelIndex:
|
||
base = (2 | (slot & 1)) << n
|
||
off = base - slot - 1
|
||
self.rc.bittree_reverse_encode(self.p_spec_pos, off, n, low_bits)
|
||
else:
|
||
self.rc.encode_direct_bits(low_bits >> kNumAlignBits, n - kNumAlignBits)
|
||
self.rc.bittree_reverse_encode(self.p_align, 0, kNumAlignBits,
|
||
low_bits & ((1 << kNumAlignBits) - 1))
|
||
|
||
def match(self, dist, length):
|
||
assert kMatchMinLen <= length <= 273, length
|
||
ps = self.pos_state
|
||
self.rc.encode_bit(self.p_is_match, (self.state << kNumPosBitsMax) + ps, 1)
|
||
self.rc.encode_bit(self.p_is_rep, self.state, 0)
|
||
lts = min(length - kMatchMinLen, kNumLenToPosStates - 1)
|
||
self._encode_len(self.p_len_choice,
|
||
self.p_len_low, self.p_len_mid, self.p_len_high, length)
|
||
self._write_dist(dist, lts)
|
||
self.state = 7 if self.state < kNumLitStates else 10
|
||
self.pos += length
|
||
|
||
def rep_match(self, length):
|
||
assert kMatchMinLen <= length <= 273, length
|
||
ps = self.pos_state
|
||
self.rc.encode_bit(self.p_is_match, (self.state << kNumPosBitsMax) + ps, 1)
|
||
self.rc.encode_bit(self.p_is_rep, self.state, 1)
|
||
self.rc.encode_bit(self.p_is_rep_g0, self.state, 0)
|
||
self.rc.encode_bit(self.p_rep0_long, (self.state << kNumPosBitsMax) + ps, 1)
|
||
self._encode_len(self.p_rep_len_choice,
|
||
self.p_rep_len_low, self.p_rep_len_mid, self.p_rep_len_high, length)
|
||
self.state = 8 if self.state < kNumLitStates else 11
|
||
self.pos += length
|
||
|
||
def finish(self):
|
||
return self.rc.finish()
|
||
|
||
|
||
# ============================================================
|
||
# Part 2: tar + xz 格式原语
|
||
# ============================================================
|
||
|
||
MAX_MATCH = 273 # LZMA1 单个 match 长度上限
|
||
CHUNK = 65536 # LZMA2 单 chunk 解压上限
|
||
PROPS_BYTE = 0x5D # lc=3, lp=0, pb=2
|
||
SEED_NOTE = ("\nMayx's Blog!").encode("utf-8")
|
||
|
||
# xz 格式常量
|
||
XZ_MAGIC = b"\xFD7zXZ\x00"
|
||
XZ_FOOTER_MAGIC = b"YZ"
|
||
XZ_CHECK_CRC64 = 0x04
|
||
LZMA2_FILTER_ID = 0x21
|
||
CHECK_SIZE = 8 # CRC64 占 8 字节
|
||
_XZ_FLAGS = bytes([0x00, XZ_CHECK_CRC64]) # Stream Flags: reserved + check_type
|
||
|
||
# tar 格式常量
|
||
TAR_BLOCK = 512
|
||
|
||
|
||
def round_up_512(n):
|
||
"""长度按 tar 块 512 字节向上取整。"""
|
||
return (n + TAR_BLOCK - 1) // TAR_BLOCK * TAR_BLOCK
|
||
|
||
|
||
def crc32(b):
|
||
return binascii.crc32(b) & 0xFFFFFFFF
|
||
|
||
|
||
# ---------- CRC64 (ECMA-182, reflected) ----------
|
||
|
||
_CRC64_POLY = 0xC96C5795D7870F42 # reflected ECMA-182
|
||
_CRC64_INIT = 0xFFFFFFFFFFFFFFFF
|
||
_CRC64_XOROUT = 0xFFFFFFFFFFFFFFFF
|
||
_CRC64_NON_REFLECTED = 0x142F0E1EBA9EA3693 # 非反射多项式 (65 bit)
|
||
_CRC64_MOD_BIT = 1 << 64 # x^64 对应的 bit
|
||
|
||
|
||
_liblzma = None
|
||
try:
|
||
_lzma_name = ctypes.util.find_library('lzma')
|
||
if _lzma_name:
|
||
_liblzma = ctypes.CDLL(_lzma_name)
|
||
_liblzma.lzma_crc64.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.c_uint64]
|
||
_liblzma.lzma_crc64.restype = ctypes.c_uint64
|
||
# 验证：CRC64("123456789") 应为 0x995dc9bbdf1939fa
|
||
if _liblzma.lzma_crc64(b'123456789', 9, 0) != 0x995dc9bbdf1939fa:
|
||
_liblzma = None
|
||
except (OSError, AttributeError):
|
||
_liblzma = None
|
||
|
||
|
||
def _make_crc64_table():
|
||
table = []
|
||
for i in range(256):
|
||
crc = i
|
||
for _ in range(8):
|
||
crc = (crc >> 1) ^ _CRC64_POLY if (crc & 1) else crc >> 1
|
||
table.append(crc)
|
||
return table
|
||
|
||
|
||
_CRC64_TABLE = _make_crc64_table()
|
||
|
||
|
||
def _crc64_pure(data):
|
||
"""纯 Python CRC64（ECMA-182 反射，init/xorout = 0xFFFF...FFFF）。"""
|
||
crc = _CRC64_INIT
|
||
for byte in data:
|
||
crc = (crc >> 8) ^ _CRC64_TABLE[(crc ^ byte) & 0xFF]
|
||
return crc ^ _CRC64_XOROUT
|
||
|
||
|
||
def _crc64_lib(data):
|
||
"""liblzma CRC64（init=0，内部处理 init/xorout）。"""
|
||
return _liblzma.lzma_crc64(data, len(data), 0)
|
||
|
||
|
||
crc64 = _crc64_lib if _liblzma else _crc64_pure
|
||
|
||
|
||
# ---------- GF(2^64) 多项式运算 ----------
|
||
# 用于 CRC64 自引用求解。CRC64 的线性贡献可表示为
|
||
# bit_reverse(contribution(D)) = D_poly * P (mod G_non)
|
||
# 其中 D_poly = bit_reverse(D)，P 是位置决定的多项式。
|
||
# 方程 D_poly * (1 ^ P1 ^ P2) = bit_reverse(CRC64(W)) 通过多项式逆元求解。
|
||
|
||
def _bit_reverse64(v):
|
||
r = 0
|
||
for _ in range(64):
|
||
r = (r << 1) | (v & 1)
|
||
v >>= 1
|
||
return r
|
||
|
||
|
||
def _poly_mul_mod(a, b):
|
||
"""GF(2) 多项式乘法 mod G（非反射）。"""
|
||
result = 0
|
||
while b:
|
||
if b & 1:
|
||
result ^= a
|
||
b >>= 1
|
||
a <<= 1
|
||
if a & _CRC64_MOD_BIT:
|
||
a ^= _CRC64_NON_REFLECTED
|
||
return result
|
||
|
||
|
||
def _poly_pow(base, exp):
|
||
"""多项式幂：base^exp mod G。"""
|
||
result = 1
|
||
while exp:
|
||
if exp & 1:
|
||
result = _poly_mul_mod(result, base)
|
||
base = _poly_mul_mod(base, base)
|
||
exp >>= 1
|
||
return result
|
||
|
||
|
||
def _poly_minv(p):
|
||
"""多项式逆元 p^(-1) mod G，用扩展欧几里得算法。"""
|
||
if p == 0:
|
||
return 0
|
||
old_r, r = p, _CRC64_NON_REFLECTED
|
||
old_s, s = 1, 0
|
||
while r:
|
||
# poly divmod(old_r, r) -> (q, rem)
|
||
probe_r = r.bit_length() - 1
|
||
rem = old_r
|
||
q = 0
|
||
while rem.bit_length() - 1 >= probe_r and rem:
|
||
shift = rem.bit_length() - 1 - probe_r
|
||
q |= 1 << shift
|
||
rem ^= r << shift
|
||
old_r, r = r, rem
|
||
# s = old_s ^ q*s (raw 多项式乘法)
|
||
prod = 0
|
||
a, b = q, s
|
||
while b:
|
||
if b & 1:
|
||
prod ^= a
|
||
b >>= 1
|
||
a <<= 1
|
||
old_s, s = s, old_s ^ prod
|
||
if old_r != 1:
|
||
return 0
|
||
# old_s mod G
|
||
probe_m = _CRC64_NON_REFLECTED.bit_length() - 1
|
||
rem = old_s
|
||
while rem.bit_length() - 1 >= probe_m and rem:
|
||
shift = rem.bit_length() - 1 - probe_m
|
||
rem ^= _CRC64_NON_REFLECTED << shift
|
||
return rem
|
||
|
||
|
||
def store_hdr(payload_len, first=False):
|
||
"""LZMA2 uncompressed chunk 头（3 字节，大端 size-1）。"""
|
||
assert 1 <= payload_len <= 65536
|
||
return bytes([0x01 if first else 0x02]) + (payload_len - 1).to_bytes(2, "big")
|
||
|
||
|
||
def matches_for(dist, total):
|
||
"""dist 固定、总长 total 的 token 序列：首个 match + 后续 rep0。"""
|
||
assert total >= 2
|
||
toks = []
|
||
first = min(MAX_MATCH, total)
|
||
if total - first == 1:
|
||
first -= 1
|
||
toks.append(("m", dist, first))
|
||
total -= first
|
||
while total > 0:
|
||
l = min(MAX_MATCH, total)
|
||
if total - l == 1:
|
||
l -= 1
|
||
toks.append(("r0", l))
|
||
total -= l
|
||
return toks
|
||
|
||
|
||
def _add_gear(chunks, o, f, q):
|
||
"""追加一组标准 gear：store 载荷 x = slip+3，紧跟一个 lzma 把那 x 字节复制一遍。
|
||
|
||
lzma chunk 解压出的字节数正好等于 x，所以齿轮转完之后 slip (f - (o-q)) 精确
|
||
归约为 len(cb)（很小的正数，x 被 CHUNK 截断时则按 CHUNK 缩减）。
|
||
_layout_iterate 反复套用把 slip 压到 <= 16，_trim_layouts 用它做相位微调。
|
||
返回追加后的 (o, f)。
|
||
"""
|
||
x = min(f - (o - q) + 3, CHUNK)
|
||
chunks.append({"kind": "store", "foff": f, "size": 3 + x,
|
||
"ooff": o, "olen": x})
|
||
o += x
|
||
f += 3 + x
|
||
cb, cu = lzma_chunk(matches_for(x, x), o)
|
||
chunks.append({"kind": "lzma", "foff": f, "size": len(cb),
|
||
"ooff": o, "olen": cu, "bytes": cb})
|
||
return o + cu, f + len(cb)
|
||
|
||
|
||
def _forward_vac(F, vac, q):
|
||
"""vac_store：把紧随其后的等长字节块前移覆盖自身。"""
|
||
fo = vac["ooff"] - q
|
||
olen = vac["olen"]
|
||
F[fo:fo + olen] = F[fo + olen:fo + 2 * olen]
|
||
|
||
|
||
def dict_prop_for(maxdist):
|
||
"""选最小的 LZMA2 dict prop 使字典 >= maxdist。"""
|
||
for p in range(41):
|
||
if (2 | (p & 1)) << (p // 2 + 11) >= maxdist:
|
||
return p, (2 | (p & 1)) << (p // 2 + 11)
|
||
raise ValueError("distance too large")
|
||
|
||
|
||
def lzma_chunk(tokens, out_pos):
|
||
"""编一个 LZMA chunk（0xC0: state+props reset，无 dict reset，无 end marker）。"""
|
||
enc = LzmaEncoder()
|
||
enc.pos = out_pos
|
||
total = 0
|
||
for t in tokens:
|
||
if t[0] == "m":
|
||
enc.match(t[1], t[2])
|
||
total += t[2]
|
||
else:
|
||
enc.rep_match(t[1])
|
||
total += t[1]
|
||
data = enc.finish()
|
||
assert total - 1 < 65536 and len(data) - 1 < 65536
|
||
hdr = bytes([0xC0]) + (total - 1).to_bytes(2, "big") + \
|
||
(len(data) - 1).to_bytes(2, "big") + bytes([PROPS_BYTE])
|
||
return hdr + data, total
|
||
|
||
|
||
# ---------- tar 原语 ----------
|
||
|
||
def tar_header(name, size, mtime, mode=0o644, typeflag=b'0', uid=0, gid=0):
|
||
"""构建 512 字节 POSIX ustar tar header。
|
||
注意：tar 数值字段包含 null 终止符，slice 范围必须精确匹配字段宽度。
|
||
mode/uid/gid 各 8 字节（7 octal + NUL），size/mtime 各 12 字节（11 octal + NUL）。
|
||
"""
|
||
h = bytearray(512)
|
||
name_bytes = name.encode('utf-8')[:100]
|
||
h[0:len(name_bytes)] = name_bytes
|
||
h[100:108] = b'%07o\x00' % mode # 8 bytes (field 100-107)
|
||
h[108:116] = b'%07o\x00' % uid # 8 bytes (field 108-115)
|
||
h[116:124] = b'%07o\x00' % gid # 8 bytes (field 116-123)
|
||
h[124:136] = b'%011o\x00' % size # 12 bytes (field 124-135)
|
||
h[136:148] = b'%011o\x00' % mtime # 12 bytes (field 136-147)
|
||
h[148:156] = b' ' # checksum placeholder (8 spaces)
|
||
h[156:157] = typeflag
|
||
h[257:263] = b'ustar\x00'
|
||
h[263:265] = b'00'
|
||
assert len(h) == 512, "header length %d != 512" % len(h)
|
||
# 计算校验和：所有字节之和，checksum 字段视为空格
|
||
chksum = sum(h) & 0o7777777
|
||
h[148:156] = b'%06o\x00 ' % chksum # 6 octal + NUL + space = 8 bytes
|
||
assert len(h) == 512, "header length %d != 512 after checksum" % len(h)
|
||
return bytes(h)
|
||
|
||
|
||
# ---------- xz 原语 ----------
|
||
|
||
def xz_varint(v):
|
||
"""xz 变长整数编码：每字节 7 位数据（LSB first），MSB 为续位。"""
|
||
b = []
|
||
while v >= 0x80:
|
||
b.append((v & 0x7F) | 0x80)
|
||
v >>= 7
|
||
b.append(v & 0x7F)
|
||
return bytes(b)
|
||
|
||
|
||
def xz_stream_header():
|
||
"""12 字节 xz stream header (CHECK_CRC64)。
|
||
Stream Flags = [reserved(0x00), check_type(CRC64=0x04)]。
|
||
"""
|
||
flags = _XZ_FLAGS
|
||
crc = crc32(flags)
|
||
return XZ_MAGIC + flags + struct.pack('<I', crc)
|
||
|
||
|
||
def xz_block_header(dict_size):
|
||
"""xz block header（单 LZMA2 filter）。
|
||
返回 (header_bytes, 实际字典大小)；header 长度即 block header 总长度。
|
||
"""
|
||
prop, real_dict = dict_prop_for(dict_size)
|
||
# filter flags: filter_id(1) + props_size(1) + props(1=prop byte)
|
||
filter_flags = bytes([LZMA2_FILTER_ID, 1, prop])
|
||
# block header content: flags(1, 0x00=1 filter) + filter_flags
|
||
content = bytes([0x00]) + filter_flags
|
||
# 填充到 (1 + len(content) + 4) 是 4 的倍数
|
||
total = 1 + len(content) + 4
|
||
pad = (4 - total % 4) % 4
|
||
content += b'\x00' * pad
|
||
bh = 1 + len(content) + 4
|
||
assert bh % 4 == 0, "bh not 4-aligned: %d" % bh
|
||
crc = crc32(bytes([bh // 4 - 1]) + content)
|
||
header = bytes([bh // 4 - 1]) + content + struct.pack('<I', crc)
|
||
assert len(header) == bh
|
||
return header, real_dict
|
||
|
||
|
||
def xz_index(records):
|
||
"""xz index。records = [(unpadded_size, uncompressed_size), ...]。"""
|
||
data = bytes([0x00]) # index indicator
|
||
data += xz_varint(len(records))
|
||
for unpadded, uncompressed in records:
|
||
data += xz_varint(unpadded)
|
||
data += xz_varint(uncompressed)
|
||
# 填充到 (len(data) + 4) 是 4 的倍数
|
||
pad = (4 - (len(data) + 4) % 4) % 4
|
||
data += b'\x00' * pad
|
||
return data + struct.pack('<I', crc32(data))
|
||
|
||
|
||
def xz_stream_footer(backward_size):
|
||
"""12 字节 xz stream footer (CHECK_CRC64)。
|
||
backward_size = (index_size / 4) - 1。
|
||
Stream Flags = [reserved(0x00), check_type(CRC64=0x04)]。
|
||
CRC32 覆盖 Backward Size + Stream Flags（6 字节，不含 CRC32 和 Footer Magic）。
|
||
"""
|
||
flags = _XZ_FLAGS
|
||
crc_data = struct.pack('<I', backward_size) + flags # 6 bytes, no magic
|
||
crc = crc32(crc_data)
|
||
return struct.pack('<I', crc) + crc_data + XZ_FOOTER_MAGIC
|
||
|
||
|
||
# ============================================================
|
||
# Part 3: 多文件 tar.xz quine 构造器
|
||
# ============================================================
|
||
|
||
class TarXzQuine:
|
||
def __init__(self, root, quine_name="quine.tar.xz", seed_name=".this_is_mayx_blog",
|
||
src_dir=""):
|
||
self.quine_name = quine_name
|
||
self.seed_name = seed_name
|
||
self.src_dir = src_dir.strip("/")
|
||
self.mtime = int(time.time())
|
||
# walk 源目录
|
||
self.walk_entries = []
|
||
self.files = []
|
||
if self.src_dir:
|
||
parts = self.src_dir.split("/")
|
||
for i in range(1, len(parts) + 1):
|
||
self.walk_entries.append(("/".join(parts[:i]), True))
|
||
for dirpath, dirnames, filenames in os.walk(root):
|
||
dirnames.sort()
|
||
for dn in dirnames:
|
||
rel = os.path.relpath(os.path.join(dirpath, dn), root).replace(os.sep, "/")
|
||
self.walk_entries.append((self._prefixed(rel), True))
|
||
for fn in sorted(filenames):
|
||
p = os.path.join(dirpath, fn)
|
||
rel = os.path.relpath(p, root).replace(os.sep, "/")
|
||
with open(p, "rb") as fp:
|
||
data = fp.read()
|
||
self.walk_entries.append((self._prefixed(rel), False))
|
||
self.files.append((self._prefixed(rel), data))
|
||
# rel -> data 映射，避免下面三处按 rel 回头线性搜索整个 files 列表
|
||
self.file_map = dict(self.files)
|
||
self.dirs = [r for r, isd in self.walk_entries if isd]
|
||
self.content = b"".join(data for _, data in self.files)
|
||
|
||
def _prefixed(self, rel):
|
||
return self.src_dir + "/" + rel if self.src_dir else rel
|
||
|
||
def tar_entries(self, d_seed, xz_size):
|
||
"""计算 tar 归档中每个条目的 (header, data_size, is_dir)。
|
||
顺序：seed, walk_entries (interleaved files+dirs), quine。
|
||
"""
|
||
entries = []
|
||
# seed 文件
|
||
seed_hdr = tar_header(self.seed_name, d_seed, self.mtime)
|
||
entries.append({"name": self.seed_name, "hdr": seed_hdr,
|
||
"data_size": d_seed, "is_dir": False})
|
||
# walk 顺序的文件和目录
|
||
for rel, isd in self.walk_entries:
|
||
if isd:
|
||
hdr = tar_header(rel, 0, self.mtime, mode=0o755, typeflag=b'5')
|
||
entries.append({"name": rel, "hdr": hdr,
|
||
"data_size": 0, "is_dir": True})
|
||
else:
|
||
# file_map 与 walk_entries 在 __init__ 里同步构造，这里必有
|
||
data = self.file_map[rel]
|
||
entries.append({"name": rel, "hdr": tar_header(rel, len(data), self.mtime),
|
||
"data_size": len(data), "is_dir": False})
|
||
# quine 文件（放在最后）
|
||
qhdr = tar_header(self.quine_name, xz_size, self.mtime)
|
||
entries.append({"name": self.quine_name, "hdr": qhdr,
|
||
"data_size": xz_size, "is_dir": False})
|
||
return entries
|
||
|
||
# ---------- 结构计算 ----------
|
||
def _compute_q(self, d_seed):
|
||
"""计算 xz_bytes 在 tar 中的偏移 q。
|
||
q = seed + files + dirs 的 tar 占用 + quine tar header。
|
||
q 不依赖 xz_size（quine 的 data 在 q 之后）。"""
|
||
q = 0
|
||
# seed 文件
|
||
q += TAR_BLOCK # seed tar header
|
||
q += round_up_512(d_seed) # seed data + pad
|
||
# walk 顺序的文件和目录
|
||
for rel, isd in self.walk_entries:
|
||
q += TAR_BLOCK # tar header
|
||
if not isd:
|
||
q += round_up_512(len(self.file_map[rel]))
|
||
# quine tar header
|
||
q += TAR_BLOCK
|
||
return q
|
||
|
||
def _trim_layouts(self, base, o_g, f_g, q, w_plant_off, d_seed):
|
||
"""枚举 trim 配置，产出一批「相位」不同的候选布局。
|
||
|
||
trim = [可选的裸 store chunk，载荷长度 x0] + 一组标准 gear
|
||
（store 载荷 x = s+3，紧接着一个 lzma 复制这 x 字节）。
|
||
|
||
裸 store 把 slip 抬高 3、把 f 抬高 3+x0；随后这组 gear 把 slip 重新归一到
|
||
len(cb)，jump 再归一到 -3。净效果是 (o_pre, f_pre) 被整体平移
|
||
δ ≈ s_g + 6 + len(cb) + 3 + x0，而 δ 决定了自洽方程的模 4 相位。
|
||
旧实现 δ 恒为 0，相位一旦不对就永远无解 —— 而 d_seed += 1 根本不改变
|
||
布局（q 只在跨 512 字节块时才变），所以那 80 次重试是纯粹的无效空转。
|
||
|
||
注意：不能只插一个裸的 lzma 复制块。lzma 复制块的输出必须等于
|
||
F[o-q : o-q+olen]，而这只能靠前面那个「载荷 = F[o-q : o-q+x] 且 x = s+3」
|
||
的 store chunk 把字节搬到正确位置来实现；裸插复制块解压出来和 F 对不上。
|
||
|
||
返回 [(chunks, o_before_jump, f_before_jump, y, jb), ...]，按文件体积升序。
|
||
"""
|
||
s_g = f_g - (o_g - q)
|
||
out = []
|
||
seen = set()
|
||
for x0 in range(0, s_g + 4):
|
||
chunks = list(base)
|
||
o, f = o_g, f_g
|
||
if x0:
|
||
# 裸 store：载荷 = F[o-q : o-q+x0]，要求 x0 <= s+3 以免读到尚未定稿的字节
|
||
chunks.append({"kind": "store", "foff": f, "size": 3 + x0,
|
||
"ooff": o, "olen": x0})
|
||
o += x0
|
||
f += 3 + x0
|
||
o, f = _add_gear(chunks, o, f, q) # 一组标准 gear
|
||
# jump 不动点：y = s + 3 + len(jb)
|
||
s = f - (o - q)
|
||
y = max(2, s + 17)
|
||
jb = ju = None
|
||
for _ in range(64):
|
||
try:
|
||
jb, ju = lzma_chunk([("m", o - w_plant_off, y)], o)
|
||
except AssertionError:
|
||
jb = None
|
||
break
|
||
y2 = s + 3 + len(jb)
|
||
if y2 == y:
|
||
break
|
||
if y2 < 2 or y2 > 273:
|
||
jb = None
|
||
break
|
||
y = y2
|
||
if jb is None or ju != y:
|
||
continue
|
||
if y + w_plant_off - TAR_BLOCK > d_seed:
|
||
continue
|
||
key = (o + y, f + len(jb))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append((chunks, o, f, y, jb))
|
||
|
||
out.sort(key=lambda t: t[2] + len(t[4]))
|
||
return out
|
||
|
||
def _solve_gadget(self, chunks_pre, o_j, f_j, y, jb, q, bh, FEW,
|
||
block_hdr, dict_size, hdrA, hdrB, d_seed, k):
|
||
"""在一个给定相位的布局上求解 gadget（T / gb / suffix / tz / k）。
|
||
|
||
关键简化：tz 只依赖 u = gb + suffix 与 k；T = u + tz + 4 也只依赖 (u, k)；
|
||
而 gb 又由 T 唯一确定（len(lzma_chunk(matches_for(T, T))))。
|
||
因此只需一维枚举 u 即可遍历全部解，既不漏解也远快于原来的三重暴力。
|
||
"""
|
||
o_pre = o_j + y
|
||
f_pre = f_j + len(jb)
|
||
if f_pre != (o_pre - q) - 3: # jump 应把 slip 归一到 -3
|
||
return None
|
||
|
||
chunks = list(chunks_pre)
|
||
chunks.append({"kind": "lzma", "foff": f_j, "size": len(jb),
|
||
"ooff": o_j, "olen": y, "bytes": jb, "jump": True})
|
||
|
||
best = None
|
||
for u in range(8, 512):
|
||
# 512k 必须落在 [f_pre + 2u + 2059, f_pre + 2u + 3595)
|
||
lo = f_pre + 2 * u + 2059
|
||
hi = f_pre + 2 * u + 3595
|
||
for k_cand in range(max(1, (lo - 512) // 512), (hi + 512) // 512 + 2):
|
||
num = k_cand * 512 - (o_pre + 2 * u + 8 - q - 1024)
|
||
if num % 3 != 0:
|
||
continue
|
||
tz_cand = num // 3
|
||
if tz_cand < 1024:
|
||
continue
|
||
T = u + tz_cand + 4
|
||
if T < 2:
|
||
continue
|
||
gb, gu = lzma_chunk(matches_for(T, T), o_pre + T)
|
||
gb_len = len(gb)
|
||
suffix = u - gb_len
|
||
if suffix < 1:
|
||
continue
|
||
|
||
# 精确自检（不再用带 9 字节偏差的预筛公式）
|
||
f_final = f_pre + 3 + T + gb_len + 3 + tz_cand + 1
|
||
n = f_final - (12 + bh)
|
||
bp = (4 - (12 + bh + n) % 4) % 4
|
||
unpadded = bh + n + CHECK_SIZE
|
||
idx = xz_index([(unpadded, q + k_cand * 512 + 1024)])
|
||
ix = len(idx)
|
||
suffix_actual = bp + CHECK_SIZE + ix + 12
|
||
if suffix_actual != suffix:
|
||
continue
|
||
xz_size = 12 + bh + n + bp + CHECK_SIZE + ix + 12
|
||
xz_pad_actual = round_up_512(xz_size)
|
||
if xz_pad_actual != k_cand * 512:
|
||
continue
|
||
total_w_actual = q + xz_pad_actual + 1024
|
||
o_final = o_pre + 2 * T + tz_cand
|
||
if o_final != total_w_actual:
|
||
continue
|
||
|
||
if best is not None and xz_size >= best["xz_size"]:
|
||
continue # 已找到更小的解，跳过
|
||
|
||
backward_size = ix // 4 - 1
|
||
final = list(chunks)
|
||
final.append({"kind": "store", "foff": f_pre, "size": 3 + T,
|
||
"ooff": o_pre, "olen": T, "vac": True})
|
||
final.append({"kind": "lzma", "foff": f_pre + 3 + T,
|
||
"size": gb_len, "ooff": o_pre + T, "olen": gu,
|
||
"bytes": gb})
|
||
final.append({"kind": "store",
|
||
"foff": f_pre + 3 + T + gb_len,
|
||
"size": 3 + tz_cand, "ooff": o_pre + 2 * T,
|
||
"olen": tz_cand, "trailing": True})
|
||
final.append({"kind": "term",
|
||
"foff": f_pre + 3 + T + gb_len + 3 + tz_cand,
|
||
"size": 1, "ooff": o_final, "olen": 0,
|
||
"bytes": b"\x00"})
|
||
best = {
|
||
"q": q, "d_seed": d_seed, "k": k,
|
||
"chunks": final, "block_hdr": block_hdr, "bh": bh,
|
||
"n": n, "xz_size": xz_size,
|
||
"T": T, "suffix": suffix_actual,
|
||
"bp": bp, "ix": ix,
|
||
"dict_size": dict_size,
|
||
"FEW": FEW, "hdrA": hdrA,
|
||
"hdrB": hdrB, "jump_y": y, "index": idx,
|
||
"stream_hdr": xz_stream_header(),
|
||
"footer": xz_stream_footer(backward_size),
|
||
"entries": self.tar_entries(d_seed, xz_size),
|
||
}
|
||
return best
|
||
|
||
def _layout_iterate(self, d_seed, maxdist_hint):
|
||
"""核心布局。
|
||
|
||
阶段 1a：C1 / repro / gears（只依赖 q，所有 trim 配置共用，只算一次）
|
||
阶段 1b：插入 trim 微调块，平移 (o_pre, f_pre) 改变模 4 相位
|
||
阶段 2 ：按 u = gb + suffix 一维搜索自洽解
|
||
"""
|
||
block_hdr, dict_size = xz_block_header(maxdist_hint)
|
||
bh = len(block_hdr)
|
||
FEW = 12 + bh + 3
|
||
w_hdrA_off = TAR_BLOCK + 12 + bh
|
||
w_hdrB_off = TAR_BLOCK + 12 + bh + 3
|
||
w_plant_off = TAR_BLOCK + 12 + bh + 6
|
||
|
||
q = self._compute_q(d_seed)
|
||
|
||
# ---------- 阶段 1a：C1 + repro + gears ----------
|
||
base = []
|
||
o, f = 0, 12 + bh
|
||
Ls = []
|
||
rem = q + FEW
|
||
while rem > CHUNK:
|
||
Ls.append(CHUNK)
|
||
rem -= CHUNK
|
||
Ls.append(rem)
|
||
assert Ls[-1] >= 2, "末 C1 chunk 载荷 %d 太小" % Ls[-1]
|
||
k = len(Ls)
|
||
hdrA = store_hdr(CHUNK, first=False)
|
||
hdrB = store_hdr(Ls[-1], first=False)
|
||
|
||
S = 0
|
||
for j, L in enumerate(Ls):
|
||
base.append({"kind": "store", "foff": f, "size": 3 + L,
|
||
"ooff": o, "olen": L, "first": j == 0,
|
||
"c1": True, "S": S})
|
||
o += L
|
||
f += 3 + L
|
||
S += L
|
||
assert o == q + FEW
|
||
|
||
S = 0
|
||
for j, L in enumerate(Ls):
|
||
if j > 0:
|
||
srcpos = w_hdrA_off if (j < k - 1 or L == CHUNK) else w_hdrB_off
|
||
dist_h = (q + FEW + S + 3 * (j - 1)) - srcpos
|
||
hb, hu = lzma_chunk([("m", dist_h, 3)], o)
|
||
base.append({"kind": "lzma", "foff": f, "size": len(hb),
|
||
"ooff": o, "olen": hu, "bytes": hb})
|
||
o += hu
|
||
f += len(hb)
|
||
cb, cu = lzma_chunk(matches_for(q + FEW + 3 * j, L), o)
|
||
base.append({"kind": "lzma", "foff": f, "size": len(cb),
|
||
"ooff": o, "olen": cu, "bytes": cb})
|
||
o += cu
|
||
f += len(cb)
|
||
S += L
|
||
|
||
# gears：把 slip 压到 <= 16
|
||
while f - (o - q) > 16:
|
||
o, f = _add_gear(base, o, f, q)
|
||
|
||
o_g, f_g = o, f
|
||
|
||
# ---------- 阶段 1b + 阶段 2 ----------
|
||
# 相位按文件体积升序枚举；第一个成功的相位通常已经足够好，但不同相位的
|
||
# (u, tz) 不同，xz_size 仍可能相差几百字节，所以再多看几个取最小的。
|
||
tried = 0
|
||
best = None
|
||
since_hit = 0
|
||
for chunks_pre, o_j, f_j, y, jb in self._trim_layouts(
|
||
base, o_g, f_g, q, w_plant_off, d_seed):
|
||
lay = self._solve_gadget(chunks_pre, o_j, f_j, y, jb, q, bh, FEW,
|
||
block_hdr, dict_size, hdrA, hdrB,
|
||
d_seed, k)
|
||
tried += 1
|
||
if lay is not None and (best is None
|
||
or lay["xz_size"] < best["xz_size"]):
|
||
best = lay
|
||
if best is not None:
|
||
since_hit += 1
|
||
if since_hit >= 10:
|
||
break
|
||
elif tried >= 80: # 一直无解，放弃后续相位
|
||
break
|
||
if best is not None:
|
||
return best
|
||
|
||
raise RuntimeError("layout 不收敛: 无自洽解 (q=%d, f_pre=%d, o_pre=%d)"
|
||
% (q, f_g, o_g))
|
||
|
||
# ---------- 装配 ----------
|
||
def assemble(self, lay):
|
||
"""构建 xz 文件 F 和 seed。"""
|
||
F = bytearray(lay["xz_size"])
|
||
q = lay["q"]
|
||
d_seed = lay["d_seed"]
|
||
bh = lay["bh"]
|
||
FEW = lay["FEW"]
|
||
|
||
# bytearray 已全零初始化：block padding、CRC64 占位、trailing-zero 载荷
|
||
# 都不必再显式写 0（后面几处赋值只会写到各自独立的区间，不会交叉覆盖）。
|
||
|
||
# pass A: 写 xz stream header + block header + 所有 chunk bytes + suffix
|
||
F[0:12] = lay["stream_hdr"]
|
||
F[12:12+bh] = lay["block_hdr"]
|
||
|
||
jump_c = None
|
||
for c in lay["chunks"]:
|
||
if c["kind"] == "store":
|
||
F[c["foff"]:c["foff"] + 3] = store_hdr(c["olen"],
|
||
first=c.get("first", False))
|
||
else:
|
||
F[c["foff"]:c["foff"] + c["size"]] = c["bytes"]
|
||
if c.get("jump"):
|
||
jump_c = c
|
||
|
||
# block padding + CRC64(pass D 填) + index + footer
|
||
bp_off = 12 + bh + lay["n"]
|
||
ix_off = bp_off + lay["bp"] + CHECK_SIZE
|
||
F[ix_off:ix_off + lay["ix"]] = lay["index"]
|
||
sf_off = ix_off + lay["ix"]
|
||
F[sf_off:sf_off + 12] = lay["footer"]
|
||
|
||
assert jump_c is not None
|
||
|
||
# pass B: 构建 seed
|
||
y = lay["jump_y"]
|
||
plant = bytes(F[jump_c["ooff"] - q: jump_c["ooff"] - q + y])
|
||
seed = bytearray(d_seed)
|
||
seed[0:12] = lay["stream_hdr"]
|
||
seed[12:12+bh] = lay["block_hdr"]
|
||
seed[12+bh:12+bh+3] = lay["hdrA"]
|
||
seed[12+bh+3:12+bh+6] = lay["hdrB"]
|
||
seed[12+bh+6:12+bh+6+y] = plant
|
||
pad = (SEED_NOTE * (d_seed // len(SEED_NOTE) + 2))[:d_seed - 12 - bh - 6 - y]
|
||
seed[12+bh+6+y:] = pad
|
||
|
||
# 构建 tar 前缀 T（含 seed data）
|
||
T = self._build_tar_prefix_with_seed(lay, seed)
|
||
F_prefix = bytes(F[0:FEW])
|
||
P = T + F_prefix
|
||
assert len(P) == q + FEW, "P 长度 %d != q+FEW %d" % (len(P), q + FEW)
|
||
|
||
# pass C: 填充 store payloads
|
||
for c in lay["chunks"]:
|
||
if c["kind"] != "store":
|
||
continue
|
||
ooff, olen = c["ooff"], c["olen"]
|
||
if c.get("c1"):
|
||
F[c["foff"] + 3:c["foff"] + 3 + olen] = P[c["S"]:c["S"] + olen]
|
||
elif c.get("trailing"):
|
||
# 载荷是尾部零：F 全零初始化，且唯一会写到 chunk 之外的 vac_store
|
||
# 目标是 [f_pre+3, f_pre+3+T)（f_pre == (o_pre-q)-3 已由 _solve_gadget
|
||
# 断言），与本区块不重叠，因此无需写入。
|
||
continue
|
||
elif c.get("vac"):
|
||
_forward_vac(F, c, q)
|
||
else:
|
||
# 其余 store 的载荷 = 输出流自身对应位置的字节（自引用复制）
|
||
fo = ooff - q
|
||
F[c["foff"] + 3:c["foff"] + 3 + olen] = F[fo:fo + olen]
|
||
|
||
# pass D: 求解 CRC64 自引用
|
||
self._solve_crc64(F, lay, T)
|
||
|
||
return bytes(F), bytes(seed)
|
||
|
||
def _build_tar_prefix_with_seed(self, lay, seed):
|
||
"""构建 tar 前缀 T（含 seed data）。"""
|
||
entries = lay["entries"]
|
||
T = bytearray()
|
||
for e in entries[:-1]:
|
||
T += e["hdr"]
|
||
if not e["is_dir"] and e["data_size"] > 0:
|
||
data = seed if e["name"] == self.seed_name else self.file_map[e["name"]]
|
||
T += data
|
||
T += b'\x00' * (round_up_512(e["data_size"]) - e["data_size"])
|
||
T += entries[-1]["hdr"] # quine tar header
|
||
return bytes(T)
|
||
|
||
def _solve_crc64(self, F, lay, T_prefix):
|
||
"""求解 CRC64 自引用。
|
||
|
||
CRC64 值 D = CRC64(W)，W = T + F + 尾部零。
|
||
D 在 W 中出现两次：F 的 CRC64 字段 + vac_store 副本。
|
||
|
||
解法（多项式逆元）：
|
||
CRC64 的线性贡献可表示为 GF(2^64) 多项式乘法：
|
||
bit_reverse(contribution(D)) = D_poly * P (mod G)
|
||
其中 D_poly = bit_reverse(D)，P = x^(64+trailing) mod G。
|
||
方程 D_poly * (1 ^ P1 ^ P2) = bit_reverse(CRC64(W)) (mod G)
|
||
用扩展欧几里得求 (1 ^ P1 ^ P2) 的逆元，一步求解。
|
||
"""
|
||
q = lay["q"]
|
||
bh = lay["bh"]
|
||
n = lay["n"]
|
||
bp = lay["bp"]
|
||
xz_size = lay["xz_size"]
|
||
T = lay["T"]
|
||
|
||
# CRC64 位置 1：在 F 中（block padding 后）
|
||
crc_f_off = 12 + bh + n + bp
|
||
crc_w1 = q + crc_f_off
|
||
|
||
# CRC64 位置 2：vac_store 副本
|
||
crc_f_copy = crc_f_off - T
|
||
crc_w2 = q + crc_f_copy
|
||
|
||
# 构造 W = T + F + 尾部零（先把两处 CRC64 置 0）
|
||
xz_pad = round_up_512(xz_size)
|
||
tz = xz_pad - xz_size + 1024
|
||
W = bytearray(T_prefix) + F + b'\x00' * tz
|
||
W[crc_w1:crc_w1 + CHECK_SIZE] = b'\x00' * CHECK_SIZE
|
||
W[crc_w2:crc_w2 + CHECK_SIZE] = b'\x00' * CHECK_SIZE
|
||
|
||
# CRC64(W) 的多项式表示
|
||
crc_poly = _bit_reverse64(crc64(bytes(W)))
|
||
|
||
# P1 = x^(64 + 位置1之后的bit数) mod G
|
||
trailing1 = (len(W) - crc_w1 - 8) * 8
|
||
P1 = _poly_pow(2, 64 + trailing1)
|
||
# P2 = x^(64 + 位置2之后的bit数) mod G
|
||
trailing2 = (len(W) - crc_w2 - 8) * 8
|
||
P2 = _poly_pow(2, 64 + trailing2)
|
||
|
||
# D_poly = crc_poly * inv(1 ^ P1 ^ P2)
|
||
coeff = 1 ^ P1 ^ P2
|
||
inv = _poly_minv(coeff)
|
||
if inv == 0:
|
||
raise RuntimeError("CRC64 多项式无逆元")
|
||
D_poly = _poly_mul_mod(crc_poly, inv)
|
||
D = _bit_reverse64(D_poly)
|
||
|
||
# 写入 D 到 F 的 CRC64 位置
|
||
struct.pack_into('<Q', F, crc_f_off, D)
|
||
|
||
# 重新执行 vac_store 副本
|
||
_forward_vac(F, next(c for c in lay["chunks"] if c.get("vac")), q)
|
||
|
||
# 验证
|
||
W2 = bytearray(T_prefix) + F + b'\x00' * tz
|
||
assert crc64(bytes(W2)) == D, "CRC64 求解验证失败"
|
||
|
||
def build(self, maxdist_hint=None):
|
||
if maxdist_hint is None:
|
||
maxdist_hint = 2 * (len(self.content) + 128) + (1 << 21)
|
||
last_err = None
|
||
# d_seed 在同一 512 字节块内不改变 q（seed 数据按 512 对齐），所以旧的
|
||
# "d_seed += 1" 重试完全不会改变布局，只是空转 80 次。真正需要 d_seed
|
||
# 变化的情况只有两种：seed 装不下 jump 种植串，或需要跨块改变 q 作为兜底。
|
||
for d_seed in (64, 96, 128, 160, 192, 256, 320, 384, 448,
|
||
576, 704, 832, 960, 1088, 1600, 2112, 2624, 3648):
|
||
try:
|
||
lay = self._layout_iterate(d_seed, maxdist_hint)
|
||
F, seed = self.assemble(lay)
|
||
return F, lay, seed
|
||
except (RuntimeError, AssertionError, ValueError) as e:
|
||
last_err = e
|
||
raise RuntimeError("build 多次失败: %s" % last_err)
|
||
|
||
|
||
# ============================================================
|
||
# Part 4: 验证 + main
|
||
# ============================================================
|
||
|
||
def verify(F, lay, files, quine_name):
|
||
"""验证：解压 xz → 解析 tar → 检查文件 + quine 自复制。"""
|
||
# 1. 解压 xz
|
||
try:
|
||
dec = lzma.LZMADecompressor(format=lzma.FORMAT_XZ)
|
||
tar_data = dec.decompress(F)
|
||
except Exception as e:
|
||
print("[verify] xz 解压失败: %s" % e)
|
||
return False
|
||
|
||
# 2. 检查 quine 自复制
|
||
if tar_data[lay["q"]:lay["q"] + lay["xz_size"]] != F:
|
||
print("[verify] quine 自复制不匹配")
|
||
a = tar_data[lay["q"]:lay["q"] + lay["xz_size"]]
|
||
for i in range(min(len(a), len(F))):
|
||
if a[i] != F[i]:
|
||
print("[verify] 首个不匹配在偏移 %d: tar=%02x vs xz=%02x" % (i, a[i], F[i]))
|
||
break
|
||
return False
|
||
|
||
# 3. 解析 tar
|
||
try:
|
||
tf = tarfile.open(fileobj=io.BytesIO(tar_data))
|
||
members = tf.getmembers()
|
||
except Exception as e:
|
||
print("[verify] tar 解析失败: %s" % e)
|
||
return False
|
||
|
||
print("[verify] tar 条目: %s" % [m.name for m in members])
|
||
|
||
# 4. 检查源文件
|
||
ok = True
|
||
for rel, data in files:
|
||
try:
|
||
f = tf.extractfile(rel)
|
||
if f is None:
|
||
print("[verify] 文件 %s 是目录或不存在" % rel)
|
||
ok = False
|
||
continue
|
||
extracted = f.read()
|
||
if extracted != data:
|
||
print("[verify] 文件 %s 内容不匹配" % rel)
|
||
ok = False
|
||
except KeyError:
|
||
print("[verify] 文件 %s 不在 tar 中" % rel)
|
||
ok = False
|
||
|
||
# 5. 检查 quine 文件
|
||
try:
|
||
qf = tf.extractfile(quine_name)
|
||
if qf:
|
||
qdata = qf.read()
|
||
if qdata != F:
|
||
print("[verify] quine 文件内容不匹配 (%d vs %d)" % (len(qdata), len(F)))
|
||
ok = False
|
||
else:
|
||
print("[verify] quine 文件匹配!")
|
||
except KeyError:
|
||
print("[verify] quine 文件不在 tar 中")
|
||
ok = False
|
||
|
||
print("[verify] 结果: %s" % ("通过" if ok else "失败"))
|
||
return ok
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="多文件 tar.xz quine 生成器")
|
||
ap.add_argument("srcdir")
|
||
ap.add_argument("output")
|
||
ap.add_argument("--quine-name", default=None,
|
||
help="归档内 quine 文件名（默认 = output 的文件名）")
|
||
ap.add_argument("--seed-name", default=".this_is_mayx_blog")
|
||
ap.add_argument("--src-dir", default="",
|
||
help="源文件在归档内的前缀目录")
|
||
ap.add_argument("--quine-dir", default="",
|
||
help="quine 在归档内所在目录")
|
||
args = ap.parse_args()
|
||
|
||
quine_name = args.quine_name or os.path.basename(args.output)
|
||
if args.quine_dir:
|
||
quine_dir = args.quine_dir.strip("/")
|
||
quine_name = quine_dir + "/" + quine_name
|
||
|
||
bq = TarXzQuine(args.srcdir, quine_name=quine_name, seed_name=args.seed_name,
|
||
src_dir=args.src_dir)
|
||
print("[blogquine-tarxz] %d 个文件, %d 个目录, 内容 %d 字节"
|
||
% (len(bq.files), len(bq.dirs), len(bq.content)))
|
||
F, lay, seed = bq.build()
|
||
with open(args.output, "wb") as fp:
|
||
fp.write(F)
|
||
print("[blogquine-tarxz] written %s: %d 字节 (q=%d, k=%d chunks, n=%d, xz_size=%d)"
|
||
% (args.output, len(F), lay["q"], lay["k"], lay["n"], lay["xz_size"]))
|
||
print("[blogquine-tarxz] T=%d, suffix=%d (bp=%d, ix=%d), dict_size=%d"
|
||
% (lay["T"], lay["suffix"], lay["bp"], lay["ix"], lay["dict_size"]))
|
||
ok = verify(F, lay, bq.files, quine_name)
|
||
sys.exit(0 if ok else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
```
|
||
</details>
|
||
|
||
不过对我来说，相比于TXZ格式我还是更喜欢7z一点，一是因为LZMA2本来就是7-Zip的作者发明的，XZ感觉像是摘桃子的，二是TXZ这个名字听起来有点怪，感觉不像压缩包，三是XZ Utils出过后门，尽管整个Linux社区都在使用XZ，但是我还是稍微有点偏见，所以这份TXZ的生成器我就粘贴出来给需要的人吧，我博客用7z格式就好了。
|
||
|
||
# 感想
|
||
以前总是有人说AI没有创新能力，只是对曾经在网络上出现的东西进行重组，但人何尝不是这样呢？像这次制作的7z/TXZ Quine生成器在整个网络上没有任何公开信息，当然我也知道这并不是理论上的创新，但谁说组合创新不是创新呢？再看看最近OpenAI又解决了一大堆数学难题，完全可以相信AI是真的拥有智能，所以我相信总有一天AI将能完成人类能做的所有事情，人类将不再需要额外的思考，只需要做自己想做的事情吧。 |