diff options
Diffstat (limited to '_posts')
| -rw-r--r-- | _posts/2026-11-01-compress.md | 1428 |
1 files changed, 1428 insertions, 0 deletions
diff --git a/_posts/2026-11-01-compress.md b/_posts/2026-11-01-compress.md new file mode 100644 index 0000000..a06ba85 --- /dev/null +++ b/_posts/2026-11-01-compress.md @@ -0,0 +1,1428 @@ +--- +layout: post +title: 关于压缩的各种探索 +tags: [压缩, 探索, Quine, zstd] +--- + + 压缩……在消除冗余的同时，又创造了新的混乱？<!--more--> + +# 对压缩比的追求 + 上次我在做[博客的离线模式](/2026/10/01/quine.html)时，为了降低带宽的消耗，我把我的博客压缩包又用7-Zip额外压缩了一遍，从大约16MiB压缩到了4MiB，效果感觉挺不错。另外我又试了试几个常见的压缩算法，比如Zip、zstd、BZip2等，全部使用最高等级压缩，但压缩效果都不如使用LZMA2的7-Zip好，速度甚至还比7-Zip慢，看来LZMA2还是相当强啊。 + 所以我在想，LZMA2是不是最强的压缩算法呢？虽然不同压缩算法可能对不同类型的数据效果不一样，但目前来看对我的数据来说，LZMA2看起来是最强的啊……不过这些只是常见算法，如果是用那些因为压缩速度慢而不太常见的算法，会不会比LZMA2更强呢？ +## ZPAQ + 我搜了一下，PAQ系列算法似乎是公认最强的，通过类似机器学习的算法花大量的时间达到更高的压缩比。其中比较知名的就是[ZPAQ](https://github.com/zpaq/zpaq)，这个程序已经有10年不更新了，当然对于算法来说这点倒也不是什么问题，毕竟LZMA2也是十几年前的算法。最终我试着用它压缩我的博客压缩包，用了一分钟左右的时间，压缩到了3.5MiB，比7-Zip压缩的少了大约10%，感觉效果还不错。 +## PAQ8PX + 当然ZPAQ只是PAQ系列中的其中一个算法，有些人为了追求更高的压缩比，依然在对其进行改进，据我所知，目前仍在更新、压缩效果也非常出色的应该是[PAQ8PX](https://github.com/hxim/paq8px)，我试着用v217版本压缩了一下，最终把压缩包压到了3.1MiB，真的是非常强。而代价也是极其高，首先压缩花了一个半小时，而且在这个过程中需要占用30GiB内存，更关键的问题是在解压时同样需要消耗这么多时间和资源，所以说这个算法真的是只能追求压缩比，基本上没办法正常使用…… +## NNCP + 除了PAQ系列之外，FFMPEG的作者[Fabrice Bellard](https://bellard.org)也做过一个使用机器学习的压缩算法，叫做[NNCP](https://bellard.org/nncp/)，使用了和现代LLM同款的Transformer模型。于是我编译了一下，试着在RTX4090上压缩我的压缩包，也差不多用了一个半小时，结果压缩比还没LZMA2高😂，给我压到了4.1MiB，也许是因为我的博客里有些图片，而NNCP貌似是专攻英文文本的，所以效果差了不少……又一次证明了LZMA2真的很强啊～ +## cmix + 后来我又听说还有一个叫做[cmix](https://github.com/byronknoll/cmix)的压缩算法，一样是基于机器学习，据说效果好像还不错，只不过在我实际用v21版本测试的时候不知道为什么总是压到后面的时候就会崩溃……这个项目也有2年没更新了，可能是代码有什么BUG。而且和PAQ8PX一样内存占用很高，用时非常长，但就从崩溃的位置来看，它压缩一半的结果还是比PAQ8PX更大，看来这个压缩算法的地位有点尴尬啊……不知道它是不是又是针对文本特化过的算法，但目前的结果来看感觉不如PAQ8PX。 +## 总结 + 最后PAQ8PX在这次测试中获得了最终的冠军🏆，我的博客压缩包里面包含的数据分布也算是有一定代表性吧，有一定的图片，大量的纯文本网页，和不少网站的结构应该很相似，所以这个压缩结果可能对不少数据都有一定的参考性。不过无论是PAQ8PX、NNCP还是cmix，它们压缩和解压需要的资源实在是太多了，不太适合一般情况下使用，至于ZPAQ虽然也是那种压缩和解压消耗相同的资源的算法，但作为十年前的东西来说，它在现在的速度还算是能在接受范围内，所以说如果想把不算太大的文件压到尽可能小，用ZPAQ看起来是最合适的选择，至于绝大多数的文件，还是用使用LZMA2算法的7-Zip吧～ + +# 关于自包含压缩包的继续研究 +## zstd Quine + 在上次我做博客压缩包的时候，AI告诉我LZMA2和zstd都可以做能包含大文件的Quine，但我当时却以“zstd感觉不是很知名”的理由选择了LZMA2。然而实际上zstd甚至都有专门的[RFC 8878](https://datatracker.ietf.org/doc/html/rfc8878)，甚至有不少网站都已经选择了zstd作为传输的压缩算法……既然如此，那我就让AI再简单研究一下zstd Quine吧，搜了一下以前也从来没人做过。 + AI看了一眼规范之后，说这个比7z Quine制作起来在结构上会更简单，但在计算校验值上会更困难，简单是因为zstd支持一种叫做“可跳过帧”的东西，有这个东西就能在构建的时候有更多的自由度，不过zstd用的是一种叫做“有限状态熵”的编码方式，这个要重新研究一下。至于困难是因为zstd默认使用XXH64来校验，相比于7z格式默认使用的CRC32来说，它是非线性的，因此很难像CRC32那样靠几个公式轻松计算出来。不过有一点好处是XXH64的计算速度比CRC32快不少，而且zstd只校验XXH64结果的低32位，因此即使是通过爆破方式，也用不了多长时间就能计算出来。当然这个值和7-Zip一样都是可选的，其实不算也不影响什么，但既然能算的话就算一下呗～ + 然后AI就开始想办法生成了，用了一个小时就做出来了，最开始它是直接做的`.zst`格式，但libarchive又开始出幺蛾子，显示`Input requires too many filters for decoding`错误，我试了一下[Gzip Quine](http://swtch.com/r.gz)用libarchive也有相同的问题……实在是令人难受，所以我又要求它改成`.tar.zst`格式，这样直接用Windows资源管理器也能直接打开，还增加了和我之前7z Quine一样的文件夹打包功能，说不定以后会用得上呢。 + 顺便一说，XXH64虽然说是非线性的，但作为非密码学哈希，倒也可以在不爆破的情况下构造值，于是以下就是最终的生成代码： +<details markdown="1"> +<summary markdown="span"> +Show Code +</summary> + +```python +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""Pack files/directories and a byte-identical copy of this .tar.zst archive. + +Example: python3 blogquine_tarzstd.py ./site ./site.tar.zst +One directory packs its contents; multiple inputs retain their basenames. +""" +from __future__ import annotations + +import argparse +import hashlib +import os +from pathlib import Path, PurePosixPath +import stat +import sys +import tempfile +import ctypes +import ctypes.util +import io +import shutil +import subprocess +import tarfile +from dataclasses import dataclass +from functools import lru_cache + +BLOCK = 65536 +MASK = (1 << 64) - 1 +P1, P2 = 0x9E3779B185EBCA87, 0xC2B2AE3D27D4EB4F +P3, P4 = 0x165667B19E3779F9, 0x85EBCA77C2B2AE63 +I1, I2 = pow(P1, -1, 1 << 64), pow(P2, -1, 1 << 64) +LL_BASE = tuple(range(16)) + (16,18,20,22,24,28,32,40,48,64,128,256,512,1024,2048,4096,8192,16384,32768,65536) +LL_BITS = (0,)*16 + (1,1,1,1,2,2,3,3,4,6,7,8,9,10,11,12,13,14,15,16) +ML_BASE = tuple(range(3,35)) + (35,37,39,41,43,47,51,59,67,83,99,131,259,515,1027,2051,4099,8195,16387,32771,65539) +ML_BITS = (0,)*32 + (1,1,1,1,2,2,3,3,4,4,5,7,8,9,10,11,12,13,14,15,16) + + +def _rol(x: int, n: int) -> int: + return ((x << n) | (x >> (64 - n))) & MASK + + +def _ror(x: int, n: int) -> int: + return ((x >> n) | (x << (64 - n))) & MASK + + +def _round(v: int, w: int) -> int: + return _rol((v + w * P2) & MASK, 31) * P1 & MASK + + +def checksum_for_size(size: int) -> int: + """XXH64 digest when all four terminal accumulators are zero.""" + if size < 32 or size % 32: + raise ValueError("decoded size must be a positive multiple of 32") + h = 0 + for _ in range(4): + h = (h * P1 + P4) & MASK + h = (h + size) & MASK + h ^= h >> 33 + h = h * P2 & MASK + h ^= h >> 29 + h = h * P3 & MASK + return h ^ (h >> 32) + + +def force_words(data: bytes | bytearray, offset: int) -> bytes: + """Force the four final lanes to zero using four unique aligned words.""" + if len(data) % 32 or offset % 8 or not 0 <= offset <= len(data)-32: + raise ValueError("invalid checksum patch position") + before = [(P1+P2)&MASK, P2, 0, (-P1)&MASK] + after = [0, 0, 0, 0] + view = memoryview(data) + for p in range(0, offset, 8): + lane = (p//8) % 4 + before[lane] = _round(before[lane], int.from_bytes(view[p:p+8], "little")) + for p in range(len(data)-8, offset+31, -8): + lane = (p//8) % 4 + w = int.from_bytes(view[p:p+8], "little") + after[lane] = (_ror(after[lane]*I1 & MASK, 31) - w*P2) & MASK + result = bytearray() + for p in range(offset, offset+32, 8): + lane = (p//8) % 4 + w = ((_ror(after[lane]*I1 & MASK, 31) - before[lane])*I2) & MASK + result += w.to_bytes(8, "little") + return bytes(result) + + +def _length_code(value, bases, widths): + for symbol, (base, width) in enumerate(zip(bases, widths)): + if base <= value < base + (1 << width): + return symbol, value-base, width + raise ValueError(f"unencodable Zstd length {value}") + + +def _literal_header(n): + if n < 32: + return bytes((n << 3,)) + if n < 4096: + return ((n << 4)|4).to_bytes(2,"little") + return ((n << 4)|12).to_bytes(3,"little") + + +def _sequence(ll, ml, distance): + if distance < 1: + raise ValueError("back reference must be positive") + lc, lv, ln = _length_code(ll, LL_BASE, LL_BITS) + mc, mv, mn = _length_code(ml, ML_BASE, ML_BITS) + oc = (distance+3).bit_length()-1 + if oc > 31: + raise ValueError("input needs a Zstd history window larger than 2 GiB") + ov = distance+3-(1 << oc) + bits = lv | (mv << ln) | (ov << (ln+mn)) | (1 << (ln+mn+oc)) + stream = bits.to_bytes((ln+mn+oc+8)//8, "little") + return bytes((1, 0x54, lc, oc, mc)) + stream + + +def _copy_block(literals, length, distance): + if len(literals)+length > 131072: + raise ValueError("Zstd decoded block exceeds 128 KiB") + seq = _sequence(len(literals), length, distance) + body = _literal_header(len(literals)) + literals + seq + return ((len(body)<<3)|4).to_bytes(3,"little") + body, seq + + +def _raw_header(size): + return (size << 3).to_bytes(3, "little") + + +def _parts(total): + """Nonempty pieces with length >=3 and bounded decoded block size.""" + out = [] + while total > BLOCK: + n = BLOCK + if total-n < 3: + n -= 3-(total-n) + out.append(n) + total -= n + if total: + if total < 3: + raise ValueError("internal block split too short") + out.append(total) + return out + + +def _header(name, size, mtime): + info = tarfile.TarInfo(name) + info.size, info.mode, info.mtime = size, 0o644, mtime + info.uid = info.gid = 0 + return info.tobuf(format=tarfile.PAX_FORMAT, encoding="utf-8", errors="surrogateescape") + + +def _window_descriptor(required): + if required > (1 << 31): + raise ValueError("input needs a history window larger than the supported 2 GiB") + for wd in range(256): + base = 1 << (10 + (wd >> 3)) + size = base + (base >> 3)*(wd & 7) + if size >= required: + return wd, size + raise ValueError("input exceeds the Zstandard frame window format") + + +@lru_cache(maxsize=1) +def _libzstd(): + path = ctypes.util.find_library("zstd") + if not path: + raise RuntimeError("libzstd is required (install the Zstandard library)") + lib = ctypes.CDLL(path) + lib.ZSTD_decompress.argtypes = [ctypes.c_void_p,ctypes.c_size_t,ctypes.c_void_p,ctypes.c_size_t] + lib.ZSTD_decompress.restype = ctypes.c_size_t + lib.ZSTD_isError.argtypes = [ctypes.c_size_t] + lib.ZSTD_isError.restype = ctypes.c_uint + lib.ZSTD_getErrorName.argtypes = [ctypes.c_size_t] + lib.ZSTD_getErrorName.restype = ctypes.c_char_p + lib.ZSTD_compressBound.argtypes = [ctypes.c_size_t] + lib.ZSTD_compressBound.restype = ctypes.c_size_t + lib.ZSTD_compress.argtypes = [ctypes.c_void_p,ctypes.c_size_t,ctypes.c_void_p,ctypes.c_size_t,ctypes.c_int] + lib.ZSTD_compress.restype = ctypes.c_size_t + return lib + + +def decompress(data: bytes, decoded_size: int) -> bytes: + """Decode and verify the one frame through the system reference library.""" + try: + lib = _libzstd() + except RuntimeError: + try: + import zstandard + except ImportError: + command = shutil.which("zstd") + if not command: + raise RuntimeError("verification needs libzstd, Python zstandard, or the zstd executable") from None + result = subprocess.run([command,"-q","-d","-c","--long=31"],input=data,capture_output=True) + if result.returncode: + raise RuntimeError("Zstandard verification: "+result.stderr.decode(errors="replace").strip()) + raw = result.stdout + else: + decoder = zstandard.ZstdDecompressor(max_window_size=1 << 31) + with decoder.stream_reader(io.BytesIO(data),read_across_frames=True) as reader: + raw = reader.read(decoded_size+1) + if reader.read(1): + raise RuntimeError("decoded data exceeds the declared size") + if len(raw) != decoded_size: + raise RuntimeError(f"decoded size mismatch: {len(raw)} != {decoded_size}") + return raw + dest = ctypes.create_string_buffer(decoded_size) + n = lib.ZSTD_decompress(dest, decoded_size, data, len(data)) + if lib.ZSTD_isError(n): + raise RuntimeError("Zstandard verification: " + lib.ZSTD_getErrorName(n).decode()) + if n != decoded_size: + raise RuntimeError(f"decoded size mismatch: {n} != {decoded_size}") + return dest.raw + + +@dataclass +class _Chunk: + kind: str + f: int + size: int + decoded: int + content: bytes = b"" + source: int = 0 + + +def _build_quine(prefix_tar: bytes, quine_name: str, *, mtime=0, compressed=None): + """Return (archive, info, decoded TAR), adding only a self member. + + The leading skippable frame is the quotation template. No seed TAR member + is created. The baseline stores source payloads once as raw Zstd literals. + """ + template_size = 512 + self_header = _header(quine_name, 1, mtime) + q = len(prefix_tar)+len(self_header) + chunks = [] + f, o = template_size+6, 0 + # Native prefix blocks may use a wider history than the quotation replay. + # Also, a block may regenerate at most min(window_size, 128 KiB). + max_distance = max(1,len(prefix_tar) if compressed is not None else 1) + raw_start = len(prefix_tar) if compressed is not None else 0 + prefix_pieces = _parts(q-raw_start) + if compressed is not None: + chunks.append(_Chunk("code",f,len(compressed),len(prefix_tar),compressed)) + f += len(compressed) + o += len(prefix_tar) + template_pieces = _parts(template_size) + # First output the actual TAR prefix, then the complete skippable template + # as the beginning of the embedded archive itself. + for size in prefix_pieces: + chunks.append(_Chunk("initial", f, 3+size, size, source=o)) + f += 3+size + o += size + source = 0 + for size in template_pieces: + chunks.append(_Chunk("template",f,3+size,size,source=source)) + f += 3+size + o += size + source += size + # Emit the physical frame header, then quote every preceding RAW block. + # The tiny header literals are fixed structural bytes, not new payload copies. + chunks.append(_Chunk("framecopy",f,9,6)) + f += 9 + o += 6 + if compressed is not None: + at = 0 + while at < len(compressed): + literal = compressed[at:at+BLOCK] + code = _raw_header(len(literal))+literal + chunks.append(_Chunk("code",f,len(code),len(literal),code)) + f += len(code) + o += len(literal) + at += len(literal) + for source_start,pieces in ((raw_start,prefix_pieces),(q,template_pieces)): + source = source_start + for size in pieces: + literal = _raw_header(size) + distance = o+3-source + code,_ = _copy_block(literal,size,distance) + chunks.append(_Chunk("code",f,len(code),3+size,code)) + f += len(code) + o += 3+size + source += size + max_distance = max(max_distance,distance) + # A quote/replay gear may itself use multiple RAW chunks. Its quoted source + # ends just before the FIRST new RAW payload, so no circular data is read. + for _ in range(64): + gap = f-(o-q) + if gap <= 0: + raise RuntimeError("invalid forward quotation gap") + start_o = o + group = _parts(gap+3) + source = 0 + for size in group: + chunks.append(_Chunk("gear",f,3+size,size,source=o-q)) + f += 3+size + o += size + for index,size in enumerate(group): + literal = _raw_header(size) if index else b"" + distance = o+len(literal)-(start_o+source) + code,_ = _copy_block(literal,size,distance) + chunks.append(_Chunk("code",f,len(code),len(literal)+size,code)) + f += len(code) + o += len(literal)+size + source += size + max_distance = max(max_distance,distance) + if f-(o-q) <= 24: + break + else: + raise RuntimeError("quotation gear normalization did not converge") + plant_pos,tail_pos = q+32,q+256 + choice = None + for empty in range(32): + jf = f+3*empty + gap = jf-(o-q) + distance = o-plant_pos + for length in range(max(3,gap+5),min(220,gap+40)): + jump,_ = _copy_block(b"",length,distance) + if length != gap+len(jump)+5: + continue + patch_f = jf+len(jump)+5 + if patch_f % 8: + continue + xo = o+length + match = 15 + for _ in range(10): + xcode,sx = _copy_block(bytes(32),match,xo+32-tail_pos) + new_match = len(sx)+8 + if new_match == match: + break + match = new_match + else: + continue + total = jf+len(jump)+len(xcode)+8 + choice = (total,empty,length,jump,xcode,sx,patch_f,xo) + break + if choice: + break + if choice is None: + raise RuntimeError("no aligned terminal block layout") + total,empty,length,jump,xcode,sx,patch_f,xo = choice + for _ in range(empty): + chunks.append(_Chunk("code",f,3,0,bytes(3))) + f += 3 + jump_output = o + chunks.append(_Chunk("code",f,len(jump),length,jump)) + f += len(jump) + o += length + chunks.append(_Chunk("code",f,len(xcode),32+len(sx)+8,xcode)) + f += len(xcode) + o += 32+len(sx)+8 + padding = (-total)%512+1024 + decoded_size = q+total+padding + if o != q+total or f+8 != total: + raise RuntimeError("terminal self-copy equation did not close") + checksum = checksum_for_size(decoded_size) + checksum_bytes = (checksum & 0xffffffff).to_bytes(4,"little") + rle = ((padding << 3)|3).to_bytes(3,"little") + b"\0" + max_distance = max(max_distance,jump_output-plant_pos,xo+32-tail_pos) + wd,window_size = _window_descriptor(max_distance) + frame_header = bytes.fromhex("28b52ffd04") + bytes((wd,)) + out = bytearray(total) + out[:8] = bytes.fromhex("502a4d18")+(template_size-8).to_bytes(4,"little") + out[template_size:template_size+6] = frame_header + for chunk in chunks: + if chunk.kind == "code": + out[chunk.f:chunk.f+chunk.size] = chunk.content + elif chunk.kind == "framecopy": + out[chunk.f:chunk.f+chunk.size] = _raw_header(6)+frame_header + else: + out[chunk.f:chunk.f+3] = _raw_header(chunk.decoded) + out[-8:-4],out[-4:] = rle,checksum_bytes + template = bytearray(out[:template_size]) + plant = out[jump_output-q:jump_output-q+length] + if len(plant) != length or 32+length > 256: + raise RuntimeError("skippable jump template overflow") + template[32:32+length] = plant + tail = sx+rle+checksum_bytes + template[256:256+len(tail)] = tail + note = b"\nZstandard quine template; checksum solved by lane inversion.\n" + template[320:] = (note*5)[:192] + out[:template_size] = template + self_header = _header(quine_name,total,mtime) + if len(prefix_tar)+len(self_header) != q: + raise ValueError("self header changed serialized length; input too large for this layout") + tar_prefix = prefix_tar+self_header + for chunk in chunks: + if chunk.kind == "initial": + out[chunk.f+3:chunk.f+chunk.size] = tar_prefix[chunk.source:chunk.source+chunk.decoded] + elif chunk.kind == "template": + out[chunk.f+3:chunk.f+chunk.size] = template[chunk.source:chunk.source+chunk.decoded] + elif chunk.kind == "gear": + out[chunk.f+3:chunk.f+chunk.size] = out[chunk.source:chunk.source+chunk.decoded] + raw = bytearray(decoded_size) + raw[:q] = tar_prefix + raw[q:q+total] = out + patch = force_words(raw,q+patch_f) + out[patch_f:patch_f+32] = patch + raw[q+patch_f:q+patch_f+32] = patch + archive = bytes(out) + actual = decompress(archive,decoded_size) + if actual != raw: + mismatch = next((p for p,(a,b) in enumerate(zip(actual,raw)) if a != b),min(len(actual),len(raw))) + raise RuntimeError(f"self reproduction mismatch at decoded offset {mismatch}") + info = dict(quine_name=quine_name,decoded_size=decoded_size,quine_offset=q, + frame_count=1,skippable_frames=1,compressed_size=total, + checksum=f"{checksum & 0xffffffff:08x}",xxh64=f"{checksum:016x}", + window_size=window_size,patch_offset=patch_f,block_count=len(chunks)+1, + prefix_size=len(prefix_tar),template_size=template_size, + payload_mode="compressed" if compressed is not None else "store", + compressed_prefix_size=len(compressed) if compressed is not None else None) + return archive,info,actual + + +def _compress_blocks(data: bytes, level: int) -> bytes: + """Independent source-prefix blocks, ready to execute after our frame header.""" + try: + lib = _libzstd() + except RuntimeError: + try: + import zstandard + except ImportError: + command = shutil.which("zstd") + if not command: + raise RuntimeError("compression needs libzstd, Python zstandard, or the zstd executable") from None + option = f"--fast={-level}" if level < 0 else f"-{level or 3}" + result = subprocess.run([command,"-q","--no-check","--ultra",option,"-c"],input=data,capture_output=True) + if result.returncode: + raise RuntimeError("Zstandard prefix compression: "+result.stderr.decode(errors="replace").strip()) + frame = bytearray(result.stdout) + else: + frame = bytearray(zstandard.ZstdCompressor(level=level,write_checksum=False).compress(data)) + else: + capacity = lib.ZSTD_compressBound(len(data)) + dest = ctypes.create_string_buffer(capacity) + result = lib.ZSTD_compress(dest,capacity,data,len(data),level) + if lib.ZSTD_isError(result): + raise RuntimeError("Zstandard prefix compression: "+lib.ZSTD_getErrorName(result).decode()) + frame = bytearray(dest.raw[:result]) + descriptor = frame[4] + single = bool(descriptor & 32) + dictionary_size = (0,1,2,4)[descriptor & 3] + fcs_flag = descriptor >> 6 + content_size = (1 if single else 0,2,4,8)[fcs_flag] + header_size = 5+(0 if single else 1)+dictionary_size+content_size + at = header_size + while True: + if at+3 > len(frame): + raise RuntimeError("truncated native prefix frame") + h = int.from_bytes(frame[at:at+3],"little") + last,kind,size = h & 1,(h >> 1) & 3,h >> 3 + if kind == 3: + raise RuntimeError("invalid native prefix block") + frame[at] &= 254 + at += 3+(1 if kind == 1 else size) + if last: + break + if at+(4 if descriptor & 4 else 0) != len(frame): + raise RuntimeError("unexpected native prefix frame suffix") + return bytes(frame[header_size:at]) + + +def build_quine(prefix_tar: bytes, quine_name: str, *, mtime=0, level=9, + verify_decoded=None): + """Build a seedless single-data-frame TAR quine. + + level=None stores the source prefix once verbatim. An integer enables native + prefix compression when its three encoded copies plus overhead are smaller + than the raw prefix. This avoids expanding incompressible payloads. Both + paths use the same algebraic checksum construction and contain no seed file. + After standard decoding and self-copy verification, an optional callback + receives (archive, decoded_tar) for additional checks without decoding again. + """ + if len(prefix_tar) % 512: + raise ValueError("prefix_tar must end on a 512-byte TAR boundary") + if not quine_name: + raise ValueError("self member name must be nonempty") + if level is not None and (not isinstance(level,int) or not -131072 <= level <= 22): + raise ValueError("compression level must be between -131072 and 22") + compressed = None + if level is not None and prefix_tar: + candidate = _compress_blocks(prefix_tar,level) + if 3*len(candidate)+2048 < len(prefix_tar): + compressed = candidate + result,info,decoded = _build_quine(prefix_tar,quine_name,mtime=mtime,compressed=compressed) + if verify_decoded is not None: + verify_decoded(result,decoded) + info["compression_level"] = level + return result,info + + +@dataclass +class Entry: + name: str + kind: str + mode: int + path: Path | None = None + link: str = "" + size: int = 0 + digest: bytes = b"" + + +def archive_name(value: str, *, empty: bool = False) -> str: + """Validate a relative POSIX member path without silently truncating it.""" + if "\0" in value or value.startswith("/") or ".." in value.split("/"): + raise ValueError(f"unsafe archive path: {value!r}") + if len(value) >= 2 and value[1] == ":": + raise ValueError(f"archive path must be relative: {value!r}") + name = str(PurePosixPath(value)) + if name == ".": + if empty: + return "" + raise ValueError("archive member name cannot be empty") + return name + + +def join_name(prefix: str, name: str) -> str: + return archive_name(f"{prefix}/{name}" if prefix else name) + + +def collect_entries(inputs: list[Path], output: Path, prefix: str, + reserved: str) -> tuple[list[Entry], bool]: + entries: list[Entry] = [] + names: dict[str, Entry] = {} + excluded = False + output_absolute = os.path.abspath(output) + if any(os.path.abspath(path) == output_absolute for path in inputs): + raise ValueError("output cannot also be an explicit input") + + def add(entry: Entry) -> None: + entry.name = archive_name(entry.name) + old = names.get(entry.name) + if old is not None: + if old.kind == entry.kind == "directory" and old.path is None: + old.path, old.mode = entry.path, entry.mode + return + raise ValueError(f"duplicate archive path: {entry.name}") + names[entry.name] = entry + entries.append(entry) + + def parents(name: str) -> None: + parts = PurePosixPath(name).parts + for end in range(1, len(parts)): + parent = "/".join(parts[:end]) + existing = names.get(parent) + if existing is not None: + if existing.kind != "directory": + raise ValueError(f"archive parent is not a directory: {parent}") + else: + add(Entry(parent, "directory", 0o755)) + + def walk(path: Path, name: str) -> None: + nonlocal excluded + if os.path.abspath(path) == output_absolute: + excluded = True + return + st = path.lstat() + parents(name) + mode = stat.S_IMODE(st.st_mode) + if stat.S_ISLNK(st.st_mode): + add(Entry(name, "symlink", mode, path, os.readlink(path))) + elif stat.S_ISDIR(st.st_mode): + add(Entry(name, "directory", mode, path)) + for child in sorted(path.iterdir(), key=lambda p: p.name): + walk(child, join_name(name, child.name)) + elif stat.S_ISREG(st.st_mode): + add(Entry(name, "file", mode, path)) + else: + raise ValueError(f"unsupported special file (not read): {path}") + + # Explicit parents also cover empty sources and custom self locations. + parents(reserved) + if prefix: + parents(prefix + "/placeholder") + for path in inputs: + st = path.lstat() # Fail for nonexistent inputs, including broken parents. + if len(inputs) == 1 and stat.S_ISDIR(st.st_mode): + for child in sorted(path.iterdir(), key=lambda p: p.name): + walk(child, join_name(prefix, child.name)) + else: + walk(path, join_name(prefix, path.name)) + if reserved in names: + raise ValueError(f"source collides with reserved archive member: {reserved}") + for name in names: + if name.startswith(reserved + "/"): + raise ValueError(f"reserved file would replace a directory: {reserved}") + return entries, excluded + + +def encode_entries(entries: list[Entry], mtime: int) -> bytes: + out = bytearray() + for entry in entries: + member = tarfile.TarInfo(entry.name) + member.mode, member.mtime = entry.mode, mtime + data = b"" + if entry.kind == "directory": + member.type = tarfile.DIRTYPE + elif entry.kind == "symlink": + member.type, member.linkname = tarfile.SYMTYPE, entry.link + else: + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) + assert entry.path is not None + with os.fdopen(os.open(entry.path, flags), "rb") as source: + before = os.fstat(source.fileno()) + if not stat.S_ISREG(before.st_mode): + raise ValueError(f"input is no longer a regular file: {entry.path}") + data = source.read() + after = os.fstat(source.fileno()) + if (len(data) != before.st_size or before.st_size != after.st_size or + before.st_mtime_ns != after.st_mtime_ns): + raise ValueError(f"input changed while reading: {entry.path}") + entry.size, entry.digest = len(data), hashlib.sha256(data).digest() + member.size = entry.size + out.extend(member.tobuf(format=tarfile.PAX_FORMAT, encoding="utf-8", + errors="surrogateescape")) + out.extend(data) + out.extend(bytes((-len(data)) % 512)) + return bytes(out) + + +def verify_archive(archive: bytes, raw: bytes, entries: list[Entry], + quine_name: str) -> None: + """Check members using the core's already checksum-verified decoded bytes.""" + if len(raw) % 512 or raw[-1024:] != bytes(1024): + raise ValueError("TAR is missing its complete zero end records") + with tarfile.open(fileobj=io.BytesIO(raw), mode="r:") as tar: + members = tar.getmembers() + by_name = {member.name.rstrip("/"): member for member in members} + if len(by_name) != len(members): + raise ValueError("duplicate member in generated TAR") + expected = {entry.name for entry in entries} | {quine_name} + if set(by_name) != expected: + raise ValueError("TAR member list does not match the input snapshot") + member = by_name[quine_name] + if not member.isfile() or tar.extractfile(member).read() != archive: + raise ValueError("Quine member is not byte-identical to the archive") + for entry in entries: + member = by_name[entry.name] + if entry.kind == "directory": + ok = member.isdir() + elif entry.kind == "symlink": + ok = member.issym() and member.linkname == entry.link + else: + ok = (member.isfile() and member.size == entry.size and + hashlib.sha256(tar.extractfile(member).read()).digest() == entry.digest) + if not ok or stat.S_IMODE(member.mode) != entry.mode: + raise ValueError(f"input entry did not survive packing: {entry.name}") + + +def atomic_write(path: Path, data: bytes, force: bool) -> None: + if path.exists() and not force: + raise FileExistsError(f"output exists; use --force to replace it: {path}") + temporary = None + try: + with tempfile.NamedTemporaryFile(prefix=".zstd-quine-", dir=path.parent, + delete=False) as output: + temporary = Path(output.name) + output.write(data) + output.flush() + os.fsync(output.fileno()) + if force: + os.replace(temporary, path) + else: + # link is atomic and cannot overwrite a file created after the check. + os.link(temporary, path) + temporary.unlink() + temporary = None + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("inputs", metavar="INPUT", nargs="+", type=Path) + parser.add_argument("output", metavar="OUTPUT", type=Path) + parser.add_argument("--quine-name", help="self member name (default: output basename)") + parser.add_argument("--src-dir", default="", help="archive prefix for source entries") + parser.add_argument("--quine-dir", default="", help="archive prefix for the self member") + parser.add_argument("--mtime", type=int, default=0, help="fixed TAR timestamp (default: 0)") + compression = parser.add_mutually_exclusive_group() + compression.add_argument("--level", type=int, default=9, + help="Zstd source compression level (default: 9; auto-falls back to storage)") + compression.add_argument("--store", dest="level", action="store_const", const=None, + help="store the source prefix without trying compression") + parser.add_argument("--force", action="store_true", help="replace an existing output after verification") + args = parser.parse_args(argv) + try: + if args.output.is_symlink(): + raise ValueError("output must not be a symbolic link") + if args.output.exists() and not args.output.is_file(): + raise ValueError("output must be a regular file path") + if args.output.exists() and not args.force: + raise FileExistsError(f"output exists; use --force to replace it: {args.output}") + if not args.output.parent.is_dir(): + raise ValueError(f"output directory does not exist: {args.output.parent}") + prefix = archive_name(args.src_dir, empty=True) + quine_dir = archive_name(args.quine_dir, empty=True) + quine = join_name(quine_dir, archive_name(args.quine_name or args.output.name)) + entries, excluded = collect_entries(args.inputs, args.output, prefix, quine) + prefix_tar = encode_entries(entries, args.mtime) + archive, info = build_quine( + prefix_tar, quine, mtime=args.mtime, level=args.level, + verify_decoded=lambda data, raw: verify_archive(data, raw, entries, quine), + ) + atomic_write(args.output, archive, args.force) + files = sum(entry.kind == "file" for entry in entries) + dirs = sum(entry.kind == "directory" for entry in entries) + links = sum(entry.kind == "symlink" for entry in entries) + print(f"[tar.zst quine] {files} files, {dirs} directories, {links} symlinks") + if excluded: + print("[tar.zst quine] excluded the output path from source traversal") + print(f"[tar.zst quine] wrote {args.output}: {len(archive)} bytes") + print(f"[layout] {info['frame_count']} checked data frame; payload mode: {info['payload_mode']}; " + f"window: {info['window_size']} bytes; XXH64: {info['checksum']}") + print("[verify] standard Zstd checksum, TAR contents and exact self: PASS") + print(f"[self] {quine}") + return 0 + except (OSError, ValueError, RuntimeError, tarfile.TarError) as error: + print(f"error: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) +``` +</details> + + 现在AI写的代码真的是越来越复杂了，真的是根本看不懂啊……但就是这样，很神奇的就把问题解决了。总之，zstd Quine也轻松实现了，也算是给自包含压缩包再提供一种新的格式吧。 + +## ZIP Quine + 后来闲的没事又让AI继续研究有没有可能做出大文件的ZIP Quine，结果它研究了一段时间还真让它研究出来了，当然[Ruben Van Mello](https://github.com/ruvmello)所说的窗口大小的问题依然存在，只不过AI用了某种方式绕过去了，而代价则是会存在成员数据有重叠的部分。比如在Python的标准库zipfile中，在[这里](https://github.com/python/cpython/blob/3fb88e94db547e9dbed97de68da87bcf9335a53a/Lib/zipfile/__init__.py#L2276)有检查自身成员压缩数据是不是越过了后续成员的局部头，导致我这种方法生成的ZIP Quine无法被解压，然而7-Zip和libarchive都能正常解压……总之这种方式生成的有一点点兼容性问题，但问题不是很严重，那么以下就是这种生成器的代码： +<details markdown="1"> +<summary markdown="span"> +Show Code +</summary> + +```python +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""Pack files/directories plus an exact copy of this ZIP archive. + +Example: python zip_quine.py ./site ./site.zip +One directory packs its contents; multiple inputs retain their basenames. +Python 3.10+; standard library only. Extract with 7-Zip or libarchive. +The archive deliberately shares physical bytes between overlapping members. +""" +from __future__ import annotations +import argparse +import io +import os +import stat +import struct +import sys +import tempfile +import time +import zipfile +import zlib +from pathlib import Path, PurePosixPath + +LBASE = list(range(3, 11)) + [11,13,15,17,19,23,27,31,35,43,51,59,67,83,99,115,131,163,195,227,258] +LBITS = [0]*8 + [1]*4 + [2]*4 + [3]*4 + [4]*4 + [5]*4 + [0] +DBASE = [1,2,3,4,5,7,9,13,17,25,33,49,65,97,129,193,257,385,513,769,1025,1537,2049,3073,4097,6145,8193,12289,16385,24577] +DBITS = [0]*4 + [v for n in range(1,14) for v in [n,n]] + +CL_ORDER = [16, 17, 18, 0, 8, 7, 9, 6, 10, 5, 11, 4, 12, 3, 13, 2, 14, 1] +CL_LEN = {0: 1, 1: 3, 17: 3, 18: 2} +# CL codes in stream order (MSB of the Huffman code first): +# 0 -> "0", 18 -> "10", 1 -> "110", 17 -> "111" +CL_CODE = {0: (0b0, 1), 18: (0b1, 2), 1: (0b011, 3), 17: (0b111, 3)} + +# (cell length, zero-length prefix, code lengths per unit copy). +# Exhaustively found for the fixed distance-8 match used by unit_bytes(). +CONNECTORS = [ + (576, 9, 129), (639, 9, 129), (927, 33, 118), (32872, 1, 128), + (33772, 49, 118), (39936, 17, 120), (39951, 17, 120), (39953, 9, 124), + (39999, 17, 120), (40732, 41, 108), (40944, 17, 120), (45559, 33, 124), + (45560, 33, 120), (49300, 1, 134), (49934, 57, 107), (50078, 49, 107), + (57356, 49, 118), (57373, 1, 129), (57374, 25, 117), (57572, 1, 133), + (57599, 41, 118), (57854, 41, 117), (58126, 57, 107), (58319, 49, 107), + (59264, 25, 117), (59390, 25, 117), (60536, 41, 120), (60540, 41, 118), + (61540, 9, 129), (61903, 49, 107), (63495, 25, 118), (63544, 33, 118), + (63608, 33, 117), (63743, 41, 118), (63943, 41, 110), (63984, 25, 118), + (64423, 25, 124), (64551, 25, 120), (65048, 49, 117), (65311, 33, 118), + (65420, 49, 118), (65436, 25, 120), +] + + +class Bits: + def __init__(self): + self.value = 0 + self.count = 0 + + def put(self, value, width): + assert 0 <= value < (1 << width), (value, width) + self.value |= value << self.count + self.count += width + + def align(self): + self.put(0, -self.count % 8) + + def bytes(self): + return self.value.to_bytes((self.count + 7) // 8, 'little') + + +def reverse(value, width): + return int(f'{value:0{width}b}'[::-1], 2) + + +def symbol(bits, s): + if s <= 143: + code, width = 0x30+s, 8 + elif s <= 255: + code, width = 0x190+s-144, 9 + elif s <= 279: + |