diff options
Diffstat (limited to '_posts/2026-11-01-compress.md')
| -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: + code, width = s-256, 7 + else: + code, width = 0xc0+s-280, 8 + bits.put(reverse(code, width), width) + + +def number(value, bases, widths): + for i, (base, width) in enumerate(zip(bases, widths)): + if base <= value < base + (1 << width): + return i, value - base, width + raise ValueError(value) + + +def copy_code(length, distance): + assert 3 <= length and 1 <= distance <= 32768 + bits = Bits() + bits.put(2, 3) + remaining = length + while remaining: + size = min(258, remaining) + if 0 < remaining - size < 3: + size -= 3 - (remaining - size) + slot, extra, width = number(size, LBASE, LBITS) + symbol(bits, 257 + slot) + bits.put(extra, width) + slot, extra, width = number(distance, DBASE, DBITS) + bits.put(reverse(slot, 5), 5) + bits.put(extra, width) + remaining -= size + symbol(bits, 256) + bits.put(0, 3) + bits.align() + bits.put(0, 16) + bits.put(0xffff, 16) + return bits.bytes() + + +def raw_header(size): + assert 0 <= size <= 65535 + return b'\0' + struct.pack('<HH', size, size ^ 0xffff) + + +def crc(data): + return zlib.crc32(data) + + +# --------------------------------------------------------------- connector +def unit_bytes(lc, final=False): + code = b'\x82\xd1\x80\0' if final else b'\x82\xd1\0\0' + return code + struct.pack('<HH', lc, lc ^ 0xffff) + + +def connector_tail(unit, m): + w = Bits() + w.put(4, 3) # BFINAL=0, BTYPE=10 + w.put(0, 5) # HLIT=0 -> 257 literal lengths + w.put(29, 5) # HDIST=29 -> 30 distance codes + w.put(14, 4) # HCLEN=14 -> 18 CL lengths + for s in CL_ORDER: + w.put(CL_LEN.get(s, 0), 3) + assert w.count == 71 + w.put(0, m) # m zero lengths (m == 1 mod 8) + assert w.count % 8 == 0 + for b in unit: + w.put(b, 8) + return w.bytes() + + +def payload_head(m, c1, final): + window_end = m + 2 * c1 + first_free = max(257, window_end) + count = 287 - first_free + if count < 2: + raise ValueError('no free distance codes') + w = Bits() + for i in range(count): + w.put(*(CL_CODE[1] if i < 2 else CL_CODE[0])) + w.put(1, 1) # EOB (literal code 1) + w.put(1 if final else 0, 3) # stored block header + w.align() + return w.bytes() + + +def geometry(lc, m, c1): + head = payload_head(m, c1, False) + return dict(lc=lc, m=m, unit=unit_bytes(lc), p_head=len(head)+4, + k=8+(71+m)//8, head=head, + head_final=payload_head(m, c1, True)) + + +UINT32 = 0xffffffff +UTF8_FLAG = 0x800 + + +def _stamp(epoch): + if epoch <= 315532800: + return 0, 33 + t = time.gmtime(min(epoch, 4354819198)) + return ((t.tm_hour << 11) | (t.tm_min << 5) | (t.tm_sec // 2), + ((t.tm_year - 1980) << 9) | (t.tm_mon << 5) | t.tm_mday) + + +def _extra_padding(size): + if not size: + return b'' + if size < 4: + raise ValueError('extra padding needs at least four bytes') + return struct.pack('<HH', 0xca05, size - 4) + bytes(size - 4) + + +def _local(e, *, extra=b''): + tm, dt = _stamp(e['mtime']) + return (struct.pack('<IHHHHHIIIHH', 0x04034b50, 20, UTF8_FLAG, + e['method'], tm, dt, e['crc'], e['csize'], + e['size'], len(e['encoded']), len(extra)) + + e['encoded'] + extra) + + +def _central(e, *, extra=b'', extra_size=None): + tm, dt = _stamp(e['mtime']) + attrs = ((e['mode'] & 0xffff) << 16) | (0x10 if e['is_dir'] else 0) + return (struct.pack('<IHHHHHHIIIHHHHHII', 0x02014b50, + (3 << 8) | 20, 20, UTF8_FLAG, e['method'], tm, dt, + e['crc'], e['csize'], e['size'], len(e['encoded']), + len(extra) if extra_size is None else extra_size, + 0, 0, 0, attrs, e['local_offset']) + + e['encoded'] + extra) + + +def _entry(item, mtime): + e = dict(item) + e['is_dir'] = bool(e.get('is_dir', False)) + name = archive_name(e['name']) + ('/' if e['is_dir'] else '') + encoded = name.encode('utf8') + if len(encoded) > 60000: + raise ValueError(f'UTF-8 archive path exceeds 60000 bytes: {name!r}') + data = bytes(e.get('data', b'')) + if e['is_dir'] and data: + raise ValueError('directory entries cannot have data') + e.update(name=name, encoded=encoded, data=data, size=len(data), + mtime=e.get('mtime', mtime), crc=crc(data), method=0, + mode=e.get('mode', (stat.S_IFDIR | 0o755) if e['is_dir'] + else (stat.S_IFREG | 0o644)), csize=len(data)) + if len(data) >= UINT32: + raise ValueError('ZIP64 is not implemented: a member is too large') + return e + + +def _large_plan(e): + best = None + base = 30 + len(e['encoded']) + for values in CONNECTORS: + g = geometry(*values) + lc, ph, k = g['lc'], g['p_head'], g['k'] + first_cap, middle_cap, last_cap = lc - base - 5 - k, lc - ph - k, lc - ph - 8 + if min(first_cap, middle_cap, last_cap) < 4: + continue + count = 2 + max(0, (e['size'] - first_cap - last_cap + middle_cap - 1) // middle_cap) + left = e['size'] - (count - 2) * middle_cap + first = min(first_cap, left - 4) + if first < 1: + continue + extra = first_cap - first + if 0 < extra < 4: + first -= 4 - extra + extra = 4 + last = left - first + if not 1 <= last <= last_cap: + continue + physical = count * lc + (count - 1) * 8 + candidate = dict(g, count=count, first=first, middle=middle_cap, + last=last, extra=extra, physical=physical, + csize=e['size'] + 5 + (count - 1) * (k + 8 + ph)) + if best is None or physical < best['physical']: + best = candidate + if best is None: + raise ValueError(f'cannot fit payload layout for {e["name"]!r}') + return best + + +def _solve_patch(baseline): + # The final four bytes are a unique free field in the archive comment. + # CRC differences do not depend on the prefix bytes, so this needs only + # two full-archive CRC scans and 32 tiny four-byte computations. + zero = crc(bytes(4)) + cols = [crc((1 << i).to_bytes(4, 'little')) ^ zero for i in range(32)] + rows = [sum(((value >> bit) & 1) << col for col, value in enumerate(cols)) + | (((baseline >> bit) & 1) << 32) for bit in range(32)] + for col in range(32): + pivot = next((r for r in range(col, 32) if rows[r] >> col & 1), None) + if pivot is None: + raise RuntimeError('CRC patch matrix unexpectedly singular') + rows[col], rows[pivot] = rows[pivot], rows[col] + for r in range(32): + if r != col and rows[r] >> col & 1: + rows[r] ^= rows[col] + return sum(((rows[i] >> 32) & 1) << i for i in range(32)).to_bytes(4, 'little') + + +def build(entries, self_name, *, mtime=0): + """Return ZIP bytes containing the inputs and a byte-identical self member. + + Each entry has name, data, is_dir, mode and optional mtime. Unix type bits + in mode are preserved, including symlinks whose data is their target text. + """ + items = [_entry(e, mtime) for e in entries] + me = _entry(dict(name=self_name, data=b'', mode=stat.S_IFREG | 0o644), mtime) + if len(me['encoded']) > 32000: + raise ValueError('self member name is too long for the bootstrap window') + names = [me['name'].rstrip('/').casefold()] + [e['name'].rstrip('/').casefold() for e in items] + if len(set(names)) != len(names): + raise ValueError('duplicate or case-colliding member names') + if len(items) + 1 >= 65535: + raise ValueError('ZIP64 is not implemented: too many members') + me.update(method=8, crc=0, local_offset=0) + h = 30 + len(me['encoded']) + 260 + p0 = h + 5 + plant_at = 30 + len(me['encoded']) + 12 + first = copy_code(p0, p0) + jump = None + for length in range(3, 4096): + code = copy_code(length, 2 * p0 - plant_at) + if length == len(first) + len(code) + 5: + jump = code + break + if jump is None: + raise ValueError('no bootstrap layout for this self name') + if len(first) + len(jump) + 5 > 248: + raise ValueError('self member name needs a bootstrap template larger than 248 bytes; use a shorter name') + prefix_size = h + 5 + p0 + len(first) + len(jump) + 5 + cells, pending, pending_entries = [], bytearray(), [] + + def flush_small(): + nonlocal pending, pending_entries + if pending_entries: + for e, relative in pending_entries: + e['cell_index'], e['cell_relative'] = len(cells), relative + cells.append(bytearray(pending) + bytes(8)) + pending, pending_entries = bytearray(), [] + + for e in items: + record_size = 30 + len(e['encoded']) + len(e['data']) + if record_size <= 65527: + if len(pending) + record_size > 65527: + flush_small() + pending_entries.append((e, len(pending))) + pending += _local(e) + e['data'] + continue + flush_small() + g = _large_plan(e) + e.update(method=8, csize=g['csize'], cell_index=len(cells), cell_relative=0) + tail, at = connector_tail(g['unit'], g['m']), g['first'] + cell = (_local(e, extra=_extra_padding(g['extra'])) + + b'\0' + struct.pack('<HH', at, at ^ 65535) + + e['data'][:at] + tail) + if len(cell) != g['lc']: + raise RuntimeError('first payload cell length mismatch') + cells.append(bytearray(cell)) + for index in range(1, g['count']): + last = index == g['count'] - 1 + size = g['last'] if last else g['middle'] + cell = ((g['head_final'] if last else g['head']) + + struct.pack('<HH', size, size ^ 65535) + + e['data'][at:at + size]) + at += size + cell += bytes(g['lc'] - len(cell)) if last else tail + if len(cell) != g['lc']: + raise RuntimeError('payload cell length mismatch') + cells.append(bytearray(cell)) + if at != e['size']: + raise RuntimeError('payload split length mismatch') + flush_small() + offsets, position = [], prefix_size + for cell in cells: + offsets.append(position) + position += len(cell) + 8 + for e in items: + e['local_offset'] = offsets[e['cell_index']] + e['cell_relative'] + + # Directory cell boundaries are hidden inside a valid custom extra TLV. + # Its data is the 8-byte seed in the cell plus the next actual 8-byte unit. + all_entries = [me] + items + batches, current, size = [], [], 0 + for e in all_entries: + record_size = 46 + len(e['encoded']) + if size + record_size > 65000 and current: + batches.append(current) + current, size = [], 0 + current.append(e) + size += record_size + batches.append(current) + cd_start = position if cells else prefix_size + cd_size = sum(46 + len(e['encoded']) for e in all_entries) + 20 * (len(batches) - 1) + directory_sizes = [sum(46 + len(e['encoded']) for e in batch) + + (12 if i < len(batches) - 1 else 26) + for i, batch in enumerate(batches)] + total = (prefix_size + sum(map(len, cells)) + sum(directory_sizes) + + 8 * (len(cells) + len(batches) - 1)) + if max(total, cd_start, cd_size) >= UINT32: + raise ValueError('ZIP64 is not implemented: archive exceeds ZIP32 limits') + me.update(size=total, csize=total - h) + eocd = struct.pack('<IHHHHIIH', 0x06054b50, 0, 0, len(all_entries), + len(all_entries), cd_size, cd_start, 4) + for i, batch in enumerate(batches): + buf = bytearray() + for j, e in enumerate(batch): + if i < len(batches) - 1 and j == len(batch) - 1: + partial = struct.pack('<HH', 0xca04, 16) + bytes(8) + buf += _central(e, extra=partial, extra_size=20) + else: + buf += _central(e) + if i == len(batches) - 1: + buf += eocd + bytes(4) + if len(buf) != directory_sizes[i]: + raise RuntimeError('directory cell length mismatch') + cells.append(buf) + if any(not 1 <= len(cell) <= 65535 for cell in cells): + raise ValueError('cell exceeds Deflate stored-block bounds') + units = [unit_bytes(len(cell), final=i == len(cells)-1) + for i, cell in enumerate(cells)] + for i in range(len(cells) - 1): + cells[i][-8:] = units[i + 1] + extra = bytearray(256) + extra[:8] = b'VACUUM04' + first_raw = raw_header(len(cells[0])) + if len(cells) == 1: + first_raw = b'\x01' + first_raw[1:] + plant = first + jump + first_raw + extra[8:8 + len(plant)] = plant + header = _local(me, extra=struct.pack('<HH', 0xca03, len(extra)) + extra) + if len(header) != h: + raise RuntimeError('self header length mismatch') + archive = bytearray(header + raw_header(p0) + header + raw_header(p0) + + first + jump + first_raw) + for i, cell in enumerate(cells): + if i: + archive += units[i] + archive += cell + if len(archive) != total: + raise RuntimeError('archive length mismatch') + patch = _solve_patch(crc(archive)) + archive[-4:] = patch + if crc(archive) != 0: + raise RuntimeError('archive CRC patch failed') + return bytes(archive) + + +def archive_name(value, *, empty=False): + value = str(value).replace('\\', '/') + if value.startswith('/') or '\0' in value: + raise ValueError(f'invalid archive path: {value!r}') + parts = value.split('/') + if any(part == '..' or ':' in part for part in parts): + raise ValueError(f'archive paths must be relative, without .. or drive names: {value!r}') + parts = [part for part in parts if part not in ('', '.')] + name = '/'.join(parts) + if not name and not empty: + raise ValueError('archive member name must not be empty') + if len(name.encode('utf-8')) > 65535: + raise ValueError('UTF-8 archive path exceeds the ZIP filename limit') + return name + + +def join_name(prefix, name): + return archive_name(f'{prefix}/{name}' if prefix else name) + + +def path_key(path): + return os.path.normcase(str(Path(path).resolve())) + + +def read_snapshot(path): + flags = os.O_RDONLY | getattr(os, 'O_BINARY', 0) | getattr(os, 'O_NOFOLLOW', 0) + with os.fdopen(os.open(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: {path}') + if before.st_size >= 0xffffffff: + raise ValueError(f'file exceeds this generator\'s ZIP32 size limit: {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: {path}') + return data + + +def collect_entries(inputs, output, prefix, reserved, mtime=0): + entries, names, case_names = [], {}, {} + excluded = False + output_key = path_key(output) + if any(path_key(path) == output_key for path in inputs): + raise ValueError('output cannot also be an explicit input') + + def add(name, kind, mode, path=None): + name = archive_name(name) + old = names.get(name) + if old is not None: + if kind == 'directory' and old['is_dir'] and old.get('_path') is None: + old['_path'], old['mode'] = path, mode + return + raise ValueError(f'duplicate archive path: {name}') + if name.casefold() in case_names: + raise ValueError(f'archive paths collide on case-insensitive systems: {name}') + entry = dict(name=name, data=b'', mode=mode, mtime=mtime, + is_dir=kind == 'directory', kind=kind, _path=path) + entries.append(entry) + names[name] = entry + case_names[name.casefold()] = name + + def parents(name): + 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 not existing['is_dir']: + raise ValueError(f'archive parent is not a directory: {parent}') + else: + add(parent, 'directory', stat.S_IFDIR | 0o755) + + def walk(path, name): + nonlocal excluded + if path_key(path) == output_key: + excluded = True + return + st = path.lstat() + parents(name) + if stat.S_ISLNK(st.st_mode): + add(name, 'symlink', stat.S_IFLNK | stat.S_IMODE(st.st_mode), path) + elif stat.S_ISDIR(st.st_mode): + if getattr(path, 'is_junction', lambda: False)(): + raise ValueError(f'directory junctions are not followed: {path}') + add(name, 'directory', stat.S_IFDIR | stat.S_IMODE(st.st_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(name, 'file', stat.S_IFREG | stat.S_IMODE(st.st_mode), path) + else: + raise ValueError(f'unsupported special file: {path}') + + parents(reserved) + if prefix: + parents(prefix + '/placeholder') + for path in inputs: + st = path.lstat() + 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 or path.resolve().name)) + if reserved.casefold() in case_names: + raise ValueError(f'source collides with reserved archive member: {reserved}') + if any(name.casefold().startswith(reserved.casefold()+'/') for name in names): + raise ValueError(f'reserved self member would replace a directory: {reserved}') + for entry in entries: + path = entry.pop('_path') + if entry['kind'] == 'file': + entry['data'] = read_snapshot(path) + elif entry['kind'] == 'symlink': + entry['data'] = os.readlink(path).encode('utf-8') + return entries, excluded + + +def verify_archive(archive, entries, self_name): + expected = {entry['name']+('/' if entry['is_dir'] else ''):entry['data'] for entry in entries} + expected[self_name] = archive + with zipfile.ZipFile(io.BytesIO(archive)) as reader: + infos = reader.infolist() + if len(infos) != len(expected) or {i.filename for i in infos} != set(expected): + raise ValueError('generated member list does not match the input snapshot') + for member in infos: + local = struct.unpack_from('<IHHHHHIIIHH', archive, member.header_offset) + if (local[0] != 0x04034b50 or local[3] != member.compress_type + or local[6:9] != (member.CRC, member.compress_size, member.file_size)): + raise ValueError(f'inconsistent ZIP local header: {member.filename}') + at = member.header_offset + 30 + local[-2] + local[-1] + wire = memoryview(archive)[at:at+member.compress_size] + original = memoryview(expected[member.filename]) + if member.file_size != len(original): + raise ValueError(f'wrong member size: {member.filename}') + checksum, position = 0, 0 + def check_chunk(decoded): + nonlocal checksum, position + if decoded != original[position:position+len(decoded)]: + raise ValueError(f'member data mismatch: {member.filename}') + checksum = zlib.crc32(decoded, checksum) + position += len(decoded) + inflater = zlib.decompressobj(-15) if member.compress_type == 8 else None + if member.compress_type not in (0, 8): + raise ValueError(f'unexpected ZIP compression method: {member.filename}') + for start in range(0, len(wire), 1 << 20): + pending = wire[start:start+(1 << 20)] + while len(pending): + decoded = inflater.decompress(pending, 1 << 20) if inflater else pending + check_chunk(decoded) + pending = inflater.unconsumed_tail if inflater else b'' + if inflater: + while True: + decoded = inflater.decompress(b'', 1 << 20) + if not decoded: + break + check_chunk(decoded) + if inflater and (not inflater.eof or inflater.unused_data or inflater.unconsumed_tail): + raise ValueError(f'incomplete or trailing Deflate data: {member.filename}') + if position != len(original) or checksum != member.CRC: + raise ValueError(f'member length or CRC mismatch: {member.filename}') + + +def atomic_write(path, data, force=False): + if path.exists() and not force: + raise FileExistsError(f'output exists; use --force to replace it: {path}') + temporary = None + try: + with tempfile.NamedTemporaryFile(prefix='.zip-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: + os.link(temporary, path) + temporary.unlink() + temporary = None + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + 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 timestamp; ZIP dates clamp to 1980..2107 (default: 0)') + 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) + self_dir = archive_name(args.quine_dir, empty=True) + self_name = join_name(self_dir, archive_name(args.quine_name or args.output.name)) + entries, excluded = collect_entries(args.inputs, args.output, prefix, self_name, args.mtime) + archive = build(entries, self_name, mtime=args.mtime) + verify_archive(archive, entries, self_name) + atomic_write(args.output, archive, args.force) + files = sum(e['kind'] == 'file' for e in entries) + dirs = sum(e['is_dir'] for e in entries) + links = sum(e['kind'] == 'symlink' for e in entries) + print(f'[ZIP quine] {files} files, {dirs} directories, {links} symlinks') + if excluded: + print('[ZIP quine] excluded the output path from source traversal') + print(f'[ZIP quine] wrote {args.output}: {len(archive)} bytes') + print('[verify] member contents, CRC32 and exact self: PASS') + print(f'[self] {self_name}') + return 0 + except (OSError, ValueError, RuntimeError, OverflowError, zipfile.BadZipFile, + zlib.error, struct.error) as error: + print(f'error: {error}', file=sys.stderr) + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) +``` +</details> + +## 更多格式——RAR Quine + 其实大多数常用的压缩算法都和ZIP差不多，核心思想基本上都是LZ77那套用距离和长度来表示重复内容，然后再用各种熵编码来尽可能用少的比特存储数据……所以说RAR Quine理论上也可以存在，毕竟原理都差不多。同样，我也试着让AI做了一下，依旧轻轻松松就做出来了。不过毕竟RAR不是开源软件，生成的代码我就不放出来了，这里有一份生成后的RAR Quine：[quine.rar](/assets/quine.rar)，如果有人有兴趣，也可以研究一下这个档案。 + +# 感想 + 我明明对压缩的原理也只是了解点皮毛，但是有了AI之后能做的事情真的是太可怕了，各种只能在想象中的东西都被轻松实现……真的是越来越离不开AI了，再看看最近的AI又发现了大量基础设施的高危漏洞，很明显，人被AI淘汰的时代可能越来越近了，到时候……还有继续记录的意义吗？
\ No newline at end of file |