diff options
| -rw-r--r-- | _data/other_repo_list.csv | 21 | ||||
| -rw-r--r-- | _posts/2026-11-01-compress.md | 1428 | ||||
| -rw-r--r-- | _tools/blogquine.py | 1120 | ||||
| -rw-r--r-- | _tools/envs_post-receive | 2 | ||||
| -rw-r--r-- | _tools/serv00_post-receive | 2 | ||||
| -rw-r--r-- | assets/quine.rar | bin | 0 -> 300 bytes |
6 files changed, 1929 insertions, 644 deletions
diff --git a/_data/other_repo_list.csv b/_data/other_repo_list.csv index 4503052..4c35840 100644 --- a/_data/other_repo_list.csv +++ b/_data/other_repo_list.csv @@ -27,7 +27,6 @@ https://git.Apture.io/mayx/blog http://219.157.255.213:25311/mayx/blog http://116.236.50.103:8789/mayx/blog https://repo.gusdya.net/mayx/blog -https://gitea.synapsetec.cn/mayx/blog http://gitea.yunshanghub.com:8081/mayx/blog http://152.69.204.151:3000/mayx/blog http://187.216.152.151:9999/mayx/blog @@ -68,7 +67,6 @@ http://39.101.74.135:5000/mayx/blog_cn http://123.57.130.140:3000/mayx/blog_cn http://118.24.46.223:3000/mayx/blog_cn http://s3.v100.vip:31057/mayx/blog_cn -http://101.33.225.95:3000/mayx/blog_cn http://45.55.138.82:3000/mayx/blog_cn http://118.24.129.148:3000/mayx/blog_cn http://120.48.141.82:3000/mayx/blog_cn @@ -119,9 +117,7 @@ http://tian-you.top:7020/mayx/blog http://34.17.182.140/mayx-gogs/blog https://mcgit.place/mayx/blog http://git.instal.org.cn/mayx/blog -https://git.smart-family.net/mayx/blog https://git.chiasson.cloud/mayx/blog -https://git.everestlinux.org/mayx/blog https://repo.qruize.com/mayx/blog https://git.xneon.org/mayx/blog https://lab.dutt.ch/mayx/blog @@ -132,11 +128,8 @@ https://gitea.deliverables.io/mayx/blog http://23.182.16.150:3000/mayx/blog https://git.ritonquilol.fr/mayx/blog https://bis127.vse.cz/mayx/blog -https://app.gitpasha.com/mayx/blog https://git.jhammah.com/mayx/blog -https://git.kemomimi.site/mayx/blog https://gitea.kolesarhome.com/mayx/blog -https://gitbucket.aint-no.info/mayx/blog https://git.wikeo.ovh/mayx/blog https://devcraft-dot.tech/mayx/blog https://a-t-g.ru/mayx/blog @@ -176,11 +169,9 @@ https://git.uj-s.com/mayx/blog https://git.albiobola.nl/mayx/blog https://git.83channel.net/mayx/blog https://repo.paperless.rs/mayx/blog -https://bolha.dev/mayx/blog https://git.checkerwars.com/mayx/blog https://git.broderlands.com/mayx/blog https://gitea.schwegmann.tech/mayx/blog -https://veretium.com/mayx/blog https://gitea.kcm-nixi.cn:32443/mayx/blog https://git.riskreduction.net/mayx/blog https://git.twitchmagic.pro/mayx/blog @@ -189,11 +180,19 @@ https://git.imagitech.mx/mayx/blog https://github.jogjacloud.com/mayx/blog https://git.stormofgalaxy.com/mayx/blog https://git.jibai.dev/mayx/blog -https://git.danfis.cz/mayx/blog https://dev.rasadyar.com/mayx/blog https://git.dew.vn/mayx/blog https://git.recstar.club/mayx/blog -http://www.1flow1.net/mayx/blog https://git.ric-rac.org/mayx/blog https://git.yan.best/mayx/blog https://forgejo.cgr-international.com/mayx/blog +https://yang-gitea.umersoft.com/mayx/blog +https://synthesist.d4d.host/mayx/blog +https://git.takits.me/mayx/blog +https://gitea.zazlab.freeddns.org/mayx/blog +https://gitweb.cn/mayx/blog +https://forge.harusato.com/mayx/blog +https://forgejo.siriusdevops.com/mayx/blog +https://git.tessdev.fr/mayx/blog +https://gitlab.navy.mi.th/mayx/blog +https://forgejo.quantum-forge.io/mayx/blog 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 diff --git a/_tools/blogquine.py b/_tools/blogquine.py index 70636bd..770ac2b 100644 --- a/_tools/blogquine.py +++ b/_tools/blogquine.py @@ -1,110 +1,83 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- +"""LZMA2 7z quine 生成器，仅依赖 Python 标准库。 +用法：python blogquine.py 源目录 输出.7z [--quine-name 名称] + [--src-dir 前缀目录] [--quine-dir 自身所在目录] +解压得到源文件及输出归档自身；内容原样存储，辅助信息位于归档内部。 +""" import argparse import binascii +from dataclasses import dataclass import lzma import os +from pathlib import Path import struct -import sys import time -# ============================================================ -# Part 1: 最小 LZMA1 range coder（只做编码） -# -# 设计目标：只实现 quine 构造需要的 token：match / rep0-match， -# 编码字节与历史内容无关（不维护输出历史）， -# 这是"先算结构、后填数据"两阶段装配的基础。 -# 参考：Igor Pavlov 的 LZMA SDK (LzmaEnc.c / LzmaDec.c)。 -# ============================================================ - -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 +SIG = b"7z\xbc\xaf'\x1c" +VER = b"\x00\x04" +CHUNK = 1 << 16 # LZMA2 未压缩 chunk 上限 +MAX_CHUNK_OUTPUT = 1 << 21 # LZMA2 压缩 chunk 解压上限 +MAX_MATCH = 273 +PROPS_BYTE = 0x5D # lc=3, lp=0, pb=2 +FILETIME_EPOCH_OFFSET = 116444736000000000 +PROB_INIT = 1024 class RangeEncoder: - """LZMA 的区间编码器。low 是 64 位（要容纳进位），range 是 32 位。""" + """LZMA 区间编码器；所有整数截断与 SDK 的 UInt32 运算一致。""" def __init__(self): self.low = 0 self.range = 0xFFFFFFFF self.cache = 0 - self.cache_size = 1 # 初值 1 -> 第一个输出字节恒为 0x00 + self.cache_size = 1 self.buf = bytearray() def _shift_low(self): - if (self.low & 0xFFFFFFFF) < 0xFF000000 or (self.low >> 32) != 0: + carry = self.low >> 32 + if (self.low & 0xFFFFFFFF) < 0xFF000000 or carry: temp = self.cache - while True: - self.buf.append((temp + (self.low >> 32)) & 0xFF) + while self.cache_size: + self.buf.append((temp + carry) & 0xFF) temp = 0xFF self.cache_size -= 1 - if self.cache_size == 0: - break self.cache = (self.low >> 24) & 0xFF self.cache_size += 1 - # 注意：C 里 `(UInt32)low << 8` 会截断回 32 位，必须照做， - # 否则 low 的高位会一直累积，进位位 (low>>32) 就不止 0/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.low = (self.low << 8) & 0xFFFFFFFF + + def _normalize(self): + while self.range < 1 << 24: 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): + def encode_bit(self, probs, index, bit): + probability = probs[index] + bound = (self.range >> 11) * probability + if bit: + self.low += bound + self.range -= bound + probs[index] -= probability >> 5 + else: + self.range = bound + probs[index] += (2048 - probability) >> 5 + self._normalize() + + def encode_direct_bits(self, value, bits): + for shift in range(bits - 1, -1, -1): self.range >>= 1 - if (value >> i) & 1: + if value >> shift & 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): - """普通位树：MSB 优先，节点下标从 1 开始。""" - 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): - """反向位树：LSB 优先，用于 SpecPos / Align。""" - m = 1 - for i in range(num_bits): - bit = symbol & 1 - symbol >>= 1 - self.encode_bit(probs, off + m, bit) - m = (m << 1) | bit + self._normalize() + + def bit_tree(self, probs, offset, bits, symbol, reverse=False): + node = 1 + shifts = range(bits) if reverse else range(bits - 1, -1, -1) + for shift in shifts: + bit = symbol >> shift & 1 + self.encode_bit(probs, offset + node, bit) + node = (node << 1) | bit def finish(self): for _ in range(5): @@ -112,602 +85,487 @@ class RangeEncoder: return bytes(self.buf) -def _pos_slot_and_bits(d): - """把 0 基距离 d (=真实距离-1) 拆成 (posSlot, 低位数, 低位值)。""" - if d < 4: - return d, 0, 0 - # posSlot >= 4 时：numDirectBits=(slot>>1)-1, base=(2|(slot&1))<<numDirectBits - 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) +class LengthEncoder: + """普通 match 和 rep0 使用独立概率，但共享长度编码逻辑。""" + + def __init__(self): + self.choice = [PROB_INIT] * 2 + self.low = [PROB_INIT] * 32 + self.mid = [PROB_INIT] * 32 + self.high = [PROB_INIT] * 256 + + def encode(self, rc, pos_state, length): + value = length - 2 + rc.encode_bit(self.choice, 0, value >= 8) + if value < 8: + rc.bit_tree(self.low, pos_state << 3, 3, value) + else: + rc.encode_bit(self.choice, 1, value >= 16) + if value < 16: + rc.bit_tree(self.mid, pos_state << 3, 3, value - 8) + else: + rc.bit_tree(self.high, 0, 8, value - 16) class LzmaEncoder: - def __init__(self, pb=2): - """ - 不维护输出假历史：match 的编码字节只取决于 (dist,len,pos)，与历史内容无关， - 因此 quine 构造（先算结构、后填数据）不需要逐字节复制假历史。 - """ - self.pb = pb - self.pos_mask = (1 << pb) - 1 + """只编码 match/rep0，编码结果不依赖历史字节，供布局后填充数据。""" + + def __init__(self, out_pos=0): self.rc = RangeEncoder() - self.pos = 0 # 已输出字节数 + self.pos = out_pos 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) # 114 - 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): - """choice 是 2 元素数组：choice[0]=choice 位，choice[1]=choice2 位。""" - 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) - - # ---------- 对外 token ---------- - 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) + self.is_match = [PROB_INIT] * (12 << 4) + self.is_rep = [PROB_INIT] * 12 + self.is_rep_g0 = [PROB_INIT] * 12 + self.rep0_long = [PROB_INIT] * (12 << 4) + self.pos_slot = [PROB_INIT] * (4 << 6) + self.spec_pos = [PROB_INIT] * 114 + self.align = [PROB_INIT] * 16 + self.match_len = LengthEncoder() + self.rep_len = LengthEncoder() + + def _write_dist(self, dist, length): + if not 1 <= dist <= 0xFFFFFFFF: + raise ValueError("LZMA distance must be 1..2^32-1") + value = dist - 1 + bits = max(0, value.bit_length() - 2) + slot = value if value < 4 else 2 * (bits + 1) + (value >> bits & 1) + self.rc.bit_tree(self.pos_slot, min(length - 2, 3) << 6, 6, 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) + base = (2 | (slot & 1)) << bits + extra = value - base + if slot < 14: + self.rc.bit_tree(self.spec_pos, base - slot - 1, bits, extra, True) 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)) + self.rc.encode_direct_bits(extra >> 4, bits - 4) + self.rc.bit_tree(self.align, 0, 4, extra & 15, True) 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) # 非 rep - 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 + if not 2 <= length <= MAX_MATCH: + raise ValueError("LZMA match length must be 2..273") + self.rc.encode_bit(self.is_match, (self.state << 4) + (self.pos & 3), 1) + self.rc.encode_bit(self.is_rep, self.state, 0) + self.match_len.encode(self.rc, self.pos & 3, length) + self._write_dist(dist, length) + self.state = 7 if self.state < 7 else 10 self.pos += length def rep_match(self, length): - """rep0 匹配：只使用最近一次 match 的距离（quine 构造只需要这种）。""" - 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 + if not 2 <= length <= MAX_MATCH: + raise ValueError("LZMA rep0 length must be 2..273") + index = (self.state << 4) + (self.pos & 3) + self.rc.encode_bit(self.is_match, index, 1) + self.rc.encode_bit(self.is_rep, self.state, 1) + self.rc.encode_bit(self.is_rep_g0, self.state, 0) + self.rc.encode_bit(self.rep0_long, index, 1) + self.rep_len.encode(self.rc, self.pos & 3, length) + self.state = 8 if self.state < 7 else 11 self.pos += length def finish(self): return self.rc.finish() -# ============================================================ -# Part 2: 7z 格式原语 -# ============================================================ - -SIG = b"7z\xbc\xaf'\x1c" -VER = b"\x00\x04" -MAX_MATCH = 273 # LZMA1 单个 match 长度上限 -# FILETIME 纪元（1601-01-01）与 Unix 纪元（1970-01-01）之差，单位 100ns -FILETIME_EPOCH_OFFSET = 116444736000000000 - +def crc32(data, value=0): + return binascii.crc32(data, value) & 0xFFFFFFFF -def crc32(b, v=0): - return binascii.crc32(b, v) & 0xFFFFFFFF - -def varint(v): - """7z UINT64 变长编码：首字节高位连续 n 个 1 + 值的 (7-n) 个高位，再跟 n 字节 LE。""" - if v < 0x80: - return bytes([v]) - for n in range(1, 9): - if v < (1 << (8 * n + 7 - n)): - break - else: - raise ValueError("varint too big") - first = ((0xFF << (8 - n)) & 0xFF) | ((v >> (8 * n)) & ((1 << (7 - n)) - 1)) - return bytes([first]) + (v & ((1 << (8 * n)) - 1)).to_bytes(n, "little") +def varint(value): + """7z UINT64：首字节给出后续小端字节数，最多九字节。""" + if not 0 <= value < 1 << 64: + raise ValueError("7z UINT64 out of range") + for size in range(8): + if value < 1 << (7 * (size + 1)): + first = ((0xFF << (8 - size)) & 0xFF) | (value >> (8 * size)) + return bytes([first]) + (value & ((1 << (8 * size)) - 1)).to_bytes(size, "little") + return b"\xff" + value.to_bytes(8, "little") 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") + if not 1 <= payload_len <= CHUNK: + raise ValueError("LZMA2 stored chunk must contain 1..65536 bytes") + return bytes([1 if first else 2]) + (payload_len - 1).to_bytes(2, "big") def matches_for(dist, total): - """dist 固定、总长 total 的 token 序列：首个 match + 后续 rep0。 - 长度按 273 上限切分，并避开"只剩 1 字节"（rep0 最小长度为 2）。""" - 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 - - -# ============================================================ -# Part 3: 多文件 7z quine 构造器 -# ============================================================ - -PROPS_BYTE = 0x5D # lc=3, lp=0, pb=2 -CHUNK = 65536 # LZMA2 单 chunk 解压上限 -SEED_NOTE = ("\nMayx's Blog!").encode("utf-8") + """固定距离的首个 match 加 rep0；分段时不留下单字节。""" + if total < 2: + raise ValueError("match sequence must output at least two bytes") + tokens = [] + while total: + length = min(MAX_MATCH, total) + if total - length == 1: + length -= 1 + tokens.append(("r0", length) if tokens else ("m", dist, length)) + total -= length + return tokens 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") + for prop in range(41): + size = 0xFFFFFFFF if prop == 40 else (2 | (prop & 1)) << (prop // 2 + 11) + if size >= maxdist: + return prop, size + raise ValueError("LZMA2 dictionary distance exceeds 2^32-1") def lzma_chunk(tokens, out_pos): - """编一个 LZMA chunk（0xC0: state+props reset，无 dict reset，无 end marker）。 - match token 的编码字节只取决于 (dist,len,pos)，与历史内容无关（不维护输出历史）。""" - 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] + """重置状态和概率，保留字典；21-bit 解压长度，不写结束标记。""" + enc = LzmaEncoder(out_pos) + for token in tokens: + if token[0] == "m": + enc.match(token[1], token[2]) + elif token[0] == "r0": + enc.rep_match(token[1]) else: - enc.rep_match(t[1]) - total += t[1] + raise ValueError("unsupported LZMA token: %r" % (token[0],)) 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 + total = enc.pos - out_pos + if not 1 <= total <= MAX_CHUNK_OUTPUT or not 1 <= len(data) <= CHUNK: + raise ValueError("LZMA2 compressed chunk size out of range") + unpacked, packed = total - 1, len(data) - 1 + header = (bytes([0xC0 | (unpacked >> 16)]) + + (unpacked & 0xFFFF).to_bytes(2, "big") + + packed.to_bytes(2, "big") + bytes([PROPS_BYTE])) + return header + data, total + + +@dataclass(frozen=True) +class Entry: + name: str + is_dir: bool + data: bytes = b"" + + +@dataclass(frozen=True) +class Layout: + prefix_size: int + chunks: tuple + samples: bytes + padding: int + matches: bytes + header: bytes + crc_offset: int + dict_size: int + + @property + def packed_size(self): + return sum(self.chunks) + 3 * len(self.chunks) + len(self.matches) + 1 + + +class LayoutError(RuntimeError): + """当前填充量没有满足布局或 CRC 定点，换一个填充量重试。""" + + +def split_sizes(size, limit): + """每片至少两字节，避免 match 无法表达的一字节尾片。""" + if size < 2: + raise ValueError("分片总长度必须至少为 2") + sizes = [limit] * (size // limit) + tail = size % limit + if tail == 1: + sizes[-1] -= 1 + tail = 2 + if tail: + sizes.append(tail) + return tuple(sizes) + + +def bitfield(flags): + data = bytearray((len(flags) + 7) // 8) + for i, flag in enumerate(flags): + if flag: + data[i // 8] |= 1 << (7 - i % 8) + return bytes(data) + + +def archive_path(value, allow_empty=False): + value = value.replace("\\", "/") + if not value and allow_empty: + return "" + parts = value.split("/") + if any(part in ("", ".", "..") for part in parts) or any(c in value for c in ("\0", ":")): + raise ValueError("归档内路径必须是相对路径，不能包含空段、. 或 ..：%r" % value) + return "/".join(parts) + + +def encode_segments(segments, out_pos): + """把 (历史起点, 长度) 转成 match 流；跨 chunk 时保留字典。""" + result, tokens = bytearray(), [] + used, max_distance = 0, 0 + for source, length in segments: + for size in split_sizes(length, MAX_CHUNK_OUTPUT): + if used + size > MAX_CHUNK_OUTPUT: + block, _ = lzma_chunk(tokens, out_pos) + result += block + out_pos += used + tokens, used = [], 0 + distance = out_pos + used - source + if distance <= 0: + raise LayoutError("回引源尚未输出") + max_distance = max(max_distance, distance) + tokens.extend(matches_for(distance, size)) + used += size + source += size + if tokens: + block, _ = lzma_chunk(tokens, out_pos) + result += block + return bytes(result), max_distance + + +def solve_fixed_crc(data, groups, targets): + """CRC 是 GF(2) 仿射函数；同步所有镜像，解三个 CRC 的 96-bit 定点。""" + def vector(values): + return sum(value << (32 * i) for i, value in enumerate(values)) + + base = vector(targets()) + columns = [] + for copies in groups: + for bit in range(32): + positions = [copy[bit // 8] for copy in copies] + for pos in positions: + data[pos] ^= 1 << (bit % 8) + columns.append(vector(targets()) ^ base) + for pos in positions: + data[pos] ^= 1 << (bit % 8) + + rows = [] + for bit in range(96): + coeff = 1 << bit + for j, column in enumerate(columns): + coeff ^= ((column >> bit) & 1) << j + rows.append([coeff, (base >> bit) & 1]) + pivots = [] + for column in range(96): + rank = len(pivots) + pivot = next((i for i in range(rank, 96) if rows[i][0] >> column & 1), None) + if pivot is None: + continue + rows[rank], rows[pivot] = rows[pivot], rows[rank] + for i in range(96): + if i != rank and rows[i][0] >> column & 1: + rows[i][0] ^= rows[rank][0] + rows[i][1] ^= rows[rank][1] + pivots.append(column) + if any(not coeff and rhs for coeff, rhs in rows): + raise LayoutError("CRC 定点方程无解") + solution = sum(rows[i][1] << column for i, column in enumerate(pivots)) + values = [(solution >> (32 * i)) & 0xFFFFFFFF for i in range(3)] + for copies, value in zip(groups, values): + packed = struct.pack("<I", value) + for copy in copies: + for i, pos in enumerate(copy): + data[pos] = packed[i] + if tuple(values) != targets(): + raise RuntimeError("CRC 定点校验失败") class BlogQuine: - def __init__(self, root, quine_name="MayxBlog.7z", seed_name="_quine_seed.bin", - src_dir=""): - self.quine_name = quine_name - self.seed_name = seed_name - # 源文件在归档内的前缀目录（如 "src" → 归档内出现 src/hello.txt）；"" = 根目录 - self.src_dir = src_dir.strip("/") - # 程序运行时间 → FILETIME（1601-01-01 起 100ns 计数）， - # 本次运行只取一次，作为归档内所有文件条目的修改时间字段。 - self.mtime_ft = time.time_ns() // 100 + FILETIME_EPOCH_OFFSET - # walk_entries: walk 顺序（父先子后、同级 dirs 在前）的 (rel, is_dir)。 - # 子流顺序 = 非空条目在此列表中的顺序，必须自始至终保持同一顺序。 - self.walk_entries = [] - self.files = [] # (relpath, data bytes)，顺序 = walk_entries 中文件序 - # src_dir 的各级父目录先作为目录条目挂入（父先子后）， - # 这样归档结构完整：如 src_dir="public/src" → "public"、"public/src" - 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)) - 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 entries(self): - ent = [{"name": self.seed_name, "dir": False}] - ent += [{"name": r, "dir": isd} for r, isd in self.walk_entries] - ent.append({"name": self.quine_name, "dir": False}) - return ent - - # ---------- 7z header ---------- - def build_header(self, n, total, d, entries, sub_sizes): - n_sub = len(sub_sizes) - N = len(entries) - h = bytearray() - h += b"\x01\x04" # kHeader, kMainStreamsInfo - h += b"\x06" + varint(0) + varint(1) + b"\x09" + varint(n) + b"\x00" # PackInfo - h += b"\x07" # UnpackInfo - h += b"\x0b" + varint(1) + b"\x00" # kFolder: 1, local - h += varint(1) + b"\x21\x21" + varint(1) + bytes([self.dict_prop]) - h += b"\x0c" + varint(d + total) # kCodersUnpackSize - h += b"\x00" # end UnpackInfo - h += b"\x08" # SubStreamsInfo - h += b"\x0d" + varint(n_sub) # 子流数 - h += b"\x09" + b"".join(varint(s) for s in sub_sizes[:-1]) # 前 n_sub-1 个大小 - h += b"\x0a" + b"\x01" # kCRC: AllDefined - crc_base = len(h) - h += b"\x00" * (4 * n_sub) # CRC 槽（quine 的待定点） - h += b"\x00\x00" # end SSI, end MainStreamsInfo - h += b"\x05" + varint(N) # FilesInfo - bits = bytearray((N + 7) // 8) # kEmptyStream 位域（MSB 优先） - for i, e in enumerate(entries): - if e["dir"]: - bits[i // 8] |= 1 << (7 - i % 8) - h += b"\x0e" + varint(len(bits)) + bytes(bits) - # kMTime：所有条目（文件与目录）写入修改时间（= 本次程序运行时间） - mt = b"\x01\x00" + struct.pack("<Q", self.mtime_ft) * N - h += b"\x14" + varint(len(mt)) + mt - names = bytearray(b"\x00") # kName: external=0 - for e in entries: - names += e["name"].encode("utf-16-le") + b"\x00\x00" - h += b"\x11" + varint(len(names)) + bytes(names) - attrs = b"".join(struct.pack("<I", 0x10 if e["dir"] else 0x20) - for e in entries) - h += b"\x15" + varint(2 + len(attrs)) + b"\x01\x00" + attrs # kAttributes - h += b"\x00\x00" # end FilesInfo, end Header - return bytes(h), crc_base - - # ---------- 结构计算 ---------- - def layout(self, d_seed, maxdist_hint): - content_len = len(self.content) - d = d_seed + content_len - # C1 链：k 个 uncompressed chunk，载荷总长 d+35 - Ls = [] - rem = d + 35 - while rem > CHUNK: - Ls.append(CHUNK) - rem -= CHUNK - Ls.append(rem) - assert Ls[-1] >= 2, "末 chunk 载荷 %d 太小，调整 d_seed" % Ls[-1] - k = len(Ls) - hdrA = store_hdr(CHUNK, first=False) # 0x02 FF FF - hdrB = store_hdr(Ls[-1], first=False) # 末 chunk 头（可能与 A 相同） - self.dict_prop, dict_size = dict_prop_for(maxdist_hint) - # 子流大小（quine 的 total 在迭代后填）：seed, 各文件, quine - file_sizes = [len(data) for _, data in self.files] - - T, h = 100, 100 - for _ in range(30): - chunks = [] - o, f = 0, 32 - S = 0 - for j, L in enumerate(Ls): # C1a..C1k - chunks.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 == d + 35 - # repro：复现 file[35 : 32+3k+d+35) = pay0 hdr1 pay1 ... hdr(k-1) pay(k-1) - S = 0 - for j, L in enumerate(Ls): - if j > 0: - # chunk j 的 3 字节头：从 seed 开头样本复制 - srcpos = 0 if (j < k - 1 or L == CHUNK) else 3 - dist_h = (d + 35 + S + 3 * (j - 1)) - srcpos - hb, hu = lzma_chunk([("m", dist_h, 3)], o) - assert hu == 3 - chunks.append({"kind": "lzma", "foff": f, "size": len(hb), - "ooff": o, "olen": hu, "bytes": hb}) - o += hu - f += len(hb) - # chunk j 的载荷：从 W[S:S+L) 复制 - cb, cu = lzma_chunk(matches_for(d + 35 + 3 * j, L), o) - assert cu == L - chunks.append({"kind": "lzma", "foff": f, "size": len(cb), - "ooff": o, "olen": cu, "bytes": cb}) - o += cu - f += len(cb) - S += L - # gears: store(x)+copy(x) 把 slip 压到 <=16 - while f - (o - d) > 16: - s = f - (o - d) - x = min(s + 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}) - o += cu - f += len(cb) - # jump: 单 match 从 W[6:6+y)（seed 里的种植串）复制，使 slip = -3 - s = f - (o - d) - y = s + 17 - for _ in range(10): - jb, ju = lzma_chunk([("m", o - 6, y)], o) - y2 = s + 3 + len(jb) - if y2 == y: - break - y = y2 - assert ju == y and y2 == y, "jump 不动点失败" - assert y + 6 <= d_seed, "种植串 %d+6 超过 seed %d" % (y, d_seed) - chunks.append({"kind": "lzma", "foff": f, "size": len(jb), - "ooff": o, "olen": ju, "bytes": jb, "jump": True}) - o += ju - f += len(jb) - # gadget: 真空 store(T) + match(T,T) + 结束标记 - assert f == (o - d) - 3, "slip=%d 应为 -3" % (f - (o - d)) - chunks.append({"kind": "store", "foff": f, "size": 3 + T, - "ooff": o, "olen": T, "vac": True}) - f += 3 + T - o += T - gb, gu = lzma_chunk(matches_for(T, T), o) - assert gu == T - chunks.append({"kind": "lzma", "foff": f, "size": len(gb), - "ooff": o, "olen": gu, "bytes": gb, "gmatch": True}) - f += len(gb) - o += gu - # LZMA2 流结束标记（0x00）：7z t 需要它确认流完整结束 - chunks.append({"kind": "term", "foff": f, "size": 1, - "ooff": o, "olen": 0, "bytes": b"\x00"}) - f += 1 - n = f - 32 - total = f + h - sub_sizes = [d_seed] + file_sizes + [total] - entries = self.entries() - hdr, crc_base = self.build_header(n, total, d, entries, sub_sizes) - h_new = len(hdr) - T_new = len(gb) + 1 + h_new - if T_new == T and h_new == h: - assert o == d + f + h, "输出终点 %d != d+total %d" % (o, d + f + h) - return {"d": d, "d_seed": d_seed, "k": k, "Ls": Ls, - "chunks": chunks, "header": hdr, "crc_base": crc_base, - "n": n, "total": f + h, "T": T, "h": h, - "hdrA": hdrA, "hdrB": hdrB, "jump_y": y, - "n_sub": len(sub_sizes), "sub_sizes": sub_sizes, - "entries": entries, "dict_size": dict_size} - T, h = T_new, h_new - raise RuntimeError("layout 不收敛") - - # ---------- 装配 ---------- - def assemble(self, lay): - F = bytearray(lay["total"]) - d, h = lay["d"], lay["h"] - d_seed = lay["d_seed"] - hoff = lay["total"] - h - # pass A: sig + 所有非 payload 字节 - F[0:6] = SIG - F[6:8] = VER - struct.pack_into("<Q", F, 12, lay["n"]) # NextHeaderOffset - struct.pack_into("<Q", F, 20, h) # NextHeaderSize - jump_c = vac_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)) - if c.get("vac"): - vac_c = c - else: - F[c["foff"]:c["foff"] + c["size"]] = c["bytes"] - if c.get("jump"): - jump_c = c - F[hoff:hoff + h] = lay["header"] - assert jump_c and vac_c - # pass B: seed = hdrA ‖ hdrB ‖ 种植串 ‖ 填充 - y = lay["jump_y"] - plant = bytes(F[jump_c["ooff"] - d: jump_c["ooff"] - d + y]) - seed = bytearray(d_seed) - seed[0:3] = lay["hdrA"] - seed[3:6] = lay["hdrB"] - seed[6:6 + y] = plant - pad = (SEED_NOTE * (d_seed // len(SEED_NOTE) + 2))[:d_seed - 6 - y] - seed[6 + y:] = pad - # pass C: 各 store 的 payload - P = bytes(seed) + self.content + bytes(F[0:35]) # = W[0:d+35) - assert len(P) == d + 35 - 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("vac"): - fo = ooff - d - F[fo:fo + olen] = F[fo + olen:fo + 2 * olen] - else: - fo = ooff - d - F[c["foff"] + 3:c["foff"] + 3 + olen] = F[fo:fo + olen] - return F, bytes(seed) - - # ---------- CRC 定点（96x96 GF(2)） ---------- - 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，各出现两处 - dpos = crc_base + 4 * (n_sub - 1) - ncopy = 32 + 3 * k + d # C1k 载荷里 F[0:35) 副本起点 - groups = [[hoff + dpos, hcopy + dpos], - [28, ncopy + 28], - [8, ncopy + 8]] - Fv = memoryview(F) - - def targets(): - return (crc32(Fv), - crc32(Fv[hoff:hoff + h]), - crc32(Fv[12:32])) - - base = targets() - basevec = base[0] | (base[1] << 32) | (base[2] << 64) - cols = [] - for g in range(3): - for kb in range(32): - poss = [p + kb // 8 for p in groups[g]] - bb = kb % 8 - for p in poss: - F[p] ^= (1 << bb) - t = targets() - for p in poss: - F[p] ^= (1 << bb) - cols.append((t[0] ^ base[0]) | ((t[1] ^ base[1]) << 32) - | ((t[2] ^ base[2]) << 64)) - # (I+A) x = basevec；方程 i: x_i ^ sum_j A[i][j] x_j = base_i - eqs = [] - for i in range(96): - coeff = 0 - for j in range(96): - if (cols[j] >> i) & 1: - coeff |= 1 << j - coeff ^= 1 << i # 对角元 = 1 ^ A[i][i] - eqs.append([coeff, (basevec >> i) & 1]) - for j in range(96): - p = next((i for i in range(j, 96) if (eqs[i][0] >> j) & 1), None) - if p is None: - raise RuntimeError("GF(2) 奇异 @bit%d" % j) - eqs[j], eqs[p] = eqs[p], eqs[j] - for i in range(96): - if i != j and ((eqs[i][0] >> j) & 1): - eqs[i][0] ^= eqs[j][0] - eqs[i][1] ^= eqs[j][1] - x = 0 - for i in range(96): - assert eqs[i][0] == (1 << i) - if eqs[i][1]: - x |= 1 << i - vals = [x & 0xFFFFFFFF, (x >> 32) & 0xFFFFFFFF, (x >> 64) & 0xFFFFFFFF] - for g in range(3): - for p in groups[g]: - F[p:p + 4] = struct.pack("<I", vals[g]) - t = targets() - assert t == tuple(vals), "CRC 定点失败: %s != %s" % (t, vals) - return vals - - def build(self, maxdist_hint=None): - if maxdist_hint is None: - # 最大距离 ~ jump 的 dist ≈ 2d + repro/gear 余量 - maxdist_hint = 2 * (len(self.content) + 128) + (1 << 21) - d_seed = 64 - last_err = None - for _ in range(80): - d = d_seed + len(self.content) - if (d + 35) % CHUNK == 1: # L_k = 1 无法编 match - d_seed += 1 - continue + """单层 LZMA2：解码(P) = content || Q，无独立的辅助文件条目。 + + Q = V || P || H，V = S32 || H副本 || chunk头样本 || pad || M副本 || 00。 + P 先 store(content || V)，再由 M 回引已输出的字节，复现 P || H。 + 内容按原样存储；M 的编码不依赖历史字节，可先求布局，再填头部与 CRC。 + """ + + def __init__(self, root, quine_name="MayxBlog.7z", src_dir="", exclude=None): + root = Path(root).resolve() + if not root.is_dir(): + raise ValueError("源目录不存在：%s" % root) + self.quine_name = archive_path(quine_name) + prefix = archive_path(src_dir.strip("/\\"), allow_empty=True) + excluded = Path(exclude).resolve() if exclude is not None else None + self.entries, self.files = [], [] + names = {} + + def add(name, is_dir, data=b""): + key = name.casefold() + if key in names: + if is_dir and names[key].is_dir and names[key].name == name: + return + raise ValueError("归档内路径冲突：%s" % name) + entry = Entry(name, is_dir, data) + names[key] = entry + self.entries.append(entry) + if not is_dir: + self.files.append(entry) + + def add_parents(name): + parts = name.split("/") + for i in range(1, len(parts)): + add("/".join(parts[:i]), True) + + if prefix: + add_parents(prefix) + add(prefix, True) + + def walk_error(error): + raise error + + for directory, dirs, files in os.walk(root, onerror=walk_error): + dirs.sort() + children = [(name, True) for name in dirs] + children += [(name, False) for name in sorted(files)] + for name, is_dir in children: + path = Path(directory) / name + if not is_dir and path.resolve() == excluded: + continue + rel = path.relative_to(root).as_posix() + name = archive_path(prefix + "/" + rel if prefix else rel) + add_parents(name) + add(name, is_dir, b"" if is_dir else path.read_bytes()) + + add_parents(self.quine_name) + if self.quine_name.casefold() in names: + raise ValueError("quine 文件名与源条目冲突：%s" % self.quine_name) + self.entries.append(Entry(self.quine_name, False)) + self.content = b"".join(entry.data for entry in self.files) + self.stream_files = [entry for entry in self.files if entry.data] + self.stream_sizes = b"".join(varint(len(entry.data)) for entry in self.stream_files) + self.stream_crcs = b"".join(struct.pack("<I", crc32(entry.data)) for entry in self.stream_files) + self.mtime = time.time_ns() // 100 + FILETIME_EPOCH_OFFSET + + def build_header(self, prefix_size, packed_size, archive_size, dict_prop): + h = bytearray(b"\x01\x04\x06" + varint(prefix_size - 32) + + b"\x01\x09" + varint(packed_size) + b"\x00\x07\x0b\x01\x00") + h += b"\x01\x21\x21\x01" + bytes([dict_prop]) # 一个 LZMA2 coder + h += b"\x0c" + varint(len(self.content) + archive_size) + b"\x00" + h += b"\x08\x0d" + varint(len(self.stream_files) + 1) + b"\x09" + self.stream_sizes + h += b"\x0a\x01" + self.stream_crcs + crc_offset = len(h) + h += b"\x00" * 4 + b"\x00\x00\x05" + varint(len(self.entries)) + + def prop(kind, data): + return bytes([kind]) + varint(len(data)) + data + + empty = [entry.is_dir or (not entry.data and i != len(self.entries) - 1) + for i, entry in enumerate(self.entries)] + h += prop(0x0E, bitfield(empty)) + if any(flag and not entry.is_dir for flag, entry in zip(empty, self.entries)): + empty_files = [not entry.is_dir for flag, entry in zip(empty, self.entries) if flag] + h += prop(0x0F, bitfield(empty_files)) + h += prop(0x14, b"\x01\x00" + struct.pack("<Q", self.mtime) * len(self.entries)) + names = b"".join(entry.name.encode("utf-16-le") + b"\x00\x00" for entry in self.entries) + attrs = b"".join(struct.pack("<I", 0x10 if entry.is_dir else 0x20) for entry in self.entries) + h += prop(0x11, b"\x00" + names) + h += prop(0x15, b"\x01\x00" + attrs) + return bytes(h + b"\x00\x00"), crc_offset + + def layout(self, padding): + header_size, match_size = 256, 256 + seen = set() + dc = len(self.content) + for _ in range(40): + prefix_size = 32 + header_size + 12 + padding + match_size + 1 + output_size = dc + prefix_size + chunks = split_sizes(output_size, CHUNK) + samples = (store_hdr(chunks[0], True) + store_hdr(CHUNK) + + store_hdr(CHUNK - 1) + store_hdr(chunks[-1])) + segments, offset = [], 0 + for i, size in enumerate(chunks): + sample = 0 if i == 0 else 1 if size == CHUNK else 2 if size == CHUNK - 1 else 3 + segments += [(dc + 32 + header_size + 3 * sample, 3), (offset, size)] + offset += size + segments += [(dc + 32 + header_size + 12 + padding, match_size + 1), + (dc + 32, header_size)] + matches, max_distance = encode_segments(segments, output_size) + dict_prop, dict_size = dict_prop_for(max_distance) + packed_size = output_size + 3 * len(chunks) + len(matches) + 1 + header, crc_offset = self.build_header( + prefix_size, packed_size, prefix_size + packed_size + header_size, dict_prop) + key = len(header), len(matches) + if key == (header_size, match_size): + return Layout(prefix_size, chunks, samples, padding, matches, + header, crc_offset, dict_size) + if key in seen: + break + seen.add(key) + header_size, match_size = key + raise LayoutError("布局没有收敛") + + def assemble(self, layout): + start = struct.pack("<QQI", layout.prefix_size - 32 + layout.packed_size, + len(layout.header), 0) + signature = SIG + VER + b"\x00" * 4 + start + prefix = (signature + layout.header + layout.samples + b"\x00" * layout.padding + + layout.matches + b"\x00") + initial = self.content + prefix + data, offset = bytearray(prefix), 0 + for i, size in enumerate(layout.chunks): + data += store_hdr(size, first=i == 0) + initial[offset:offset + size] + offset += size + data += layout.matches + b"\x00" + layout.header + if len(prefix) != layout.prefix_size or offset != len(initial): + raise RuntimeError("装配布局不一致") + + def stored_positions(output_offset, size): + result, consumed, physical = [], 0, layout.prefix_size + for length in layout.chunks: + begin = max(output_offset, consumed) + end = min(output_offset + size, consumed + length) + result.extend(range(physical + 3 + begin - consumed, physical + 3 + end - consumed)) + consumed += length + physical += 3 + length + if len(result) != size: + raise RuntimeError("无法定位 store 载荷") + return result + + slot = layout.crc_offset + hoff = layout.prefix_size + layout.packed_size + groups = [[list(range(32 + slot, 36 + slot)), + stored_positions(len(self.content) + 32 + slot, 4), + list(range(hoff + slot, hoff + slot + 4))]] + for slot in (28, 8): + groups.append([list(range(slot, slot + 4)), stored_positions(len(self.content) + slot, 4)]) + view = memoryview(data) + solve_fixed_crc(data, groups, lambda: (crc32(view), crc32(view[hoff:]), crc32(view[12:32]))) + return bytes(data) + + def verify(self, data, layout): + dec = lzma.LZMADecompressor(format=lzma.FORMAT_RAW, + filters=[{"id": lzma.FILTER_LZMA2, "dict_size": layout.dict_size}]) + expected = len(self.content) + len(data) + packed = data[layout.prefix_size:layout.prefix_size + layout.packed_size] + decoded = dec.decompress(packed, max_length=expected + 1) + view = memoryview(decoded) + if (not dec.eof or dec.unused_data or len(decoded) != expected + or view[:len(self.content)] != self.content or view[len(self.content):] != data): + raise RuntimeError("自解码校验失败") + + def build(self): + last_error = None + for padding in range(256): try: - lay = self.layout(d_seed, maxdist_hint) - F, seed = self.assemble(lay) - crcs = self.solve_crc(F, lay, seed) - return bytes(F), lay, seed, crcs - except (RuntimeError, AssertionError) as e: - last_err = e - d_seed += 1 - raise RuntimeError("build 多次失败: %s" % last_err) - - -def verify(F, lay, seed, files): - """解码 LZMA2 流并逐文件校验。""" - d, d_seed = lay["d"], lay["d_seed"] - dec = lzma.LZMADecompressor( - format=lzma.FORMAT_RAW, - filters=[{"id": lzma.FILTER_LZMA2, "dict_size": lay["dict_size"]}]) - W = dec.decompress(F[32:32 + lay["n"]]) - assert len(W) == d + len(F), "|W|=%d 期望 %d" % (len(W), d + len(F)) - ok = True - if W[:d_seed] != seed: - print("[verify] seed 不匹配") - ok = False - pos = d_seed - for rel, data in files: - if W[pos:pos + len(data)] != data: - print("[verify] 文件不匹配:", rel) - ok = False - pos += len(data) - assert pos == d - if W[d:] != F: - print("[verify] quine 自复制不匹配") - ok = False - print("[verify] 自解码: |W|=%d, seed/文件/quine 全部匹配: %s" % (len(W), ok)) - return ok + layout = self.layout(padding) + data = self.assemble(layout) + self.verify(data, layout) + return data + except LayoutError as error: + last_error = error + raise RuntimeError("无法找到布局/CRC 定点：%s" % last_error) def main(): - ap = argparse.ArgumentParser(description="多文件 7z 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="源文件在归档内的前缀目录（如 src；默认根目录）") - ap.add_argument("--quine-dir", default="", - help="quine 在归档内所在的目录（如 public；默认根目录）") - args = ap.parse_args() - quine_name = args.quine_name or os.path.basename(args.output) - if args.quine_dir: - # 归档内路径用 / 分隔（与 walk 收集的目录条目一致） - quine_dir = args.quine_dir.strip("/") - quine_name = quine_dir + "/" + quine_name - - bq = BlogQuine(args.srcdir, quine_name=quine_name, seed_name=args.seed_name, - src_dir=args.src_dir) - print("[blogquine] %d 个文件, %d 个目录, 内容 %d 字节" - % (len(bq.files), len(bq.dirs), len(bq.content))) - F, lay, seed, crcs = bq.build() - with open(args.output, "wb") as fp: - fp.write(F) - print("[blogquine] written %s: %d 字节 (d=%d, k=%d chunks, n=%d, h=%d, T=%d)" - % (args.output, len(F), lay["d"], lay["k"], lay["n"], lay["h"], lay["T"])) - print("[blogquine] CRC 定点: D(quine)=%08x N(hdr)=%08x S(sig)=%08x" % tuple(crcs)) - ok = verify(F, lay, seed, bq.files) - sys.exit(0 if ok else 1) + parser = argparse.ArgumentParser(description="LZMA2 7z quine 生成器") + parser.add_argument("srcdir", help="要打包的源目录") + parser.add_argument("output", help="输出的 7z 文件") + parser.add_argument("--quine-name", help="归档内自身文件名，默认使用输出文件名") + parser.add_argument("--src-dir", default="", help="源文件在归档内的前缀目录") + parser.add_argument("--quine-dir", default="", help="归档内自身文件所在目录") + args = parser.parse_args() + try: + output = Path(args.output) + name = args.quine_name or output.name + if args.quine_dir: + name = archive_path(args.quine_dir.strip("/\\")) + "/" + name + quine = BlogQuine(args.srcdir, name, args.src_dir, exclude=output) + data = quine.build() # 验证成功后才写出，失败时保留已有输出。 + output.write_bytes(data) + except (OSError, ValueError, RuntimeError, lzma.LZMAError) as error: + parser.exit(1, "blogquine: %s\n" % error) + print("[blogquine] %d 个文件，%d 个目录，内容 %d 字节" % + (len(quine.files), sum(entry.is_dir for entry in quine.entries), len(quine.content))) + print("[blogquine] 已写入 %s：%d 字节；自解码通过" % (output, len(data))) if __name__ == "__main__": diff --git a/_tools/envs_post-receive b/_tools/envs_post-receive index 7fee4c9..3c85de2 100644 --- a/_tools/envs_post-receive +++ b/_tools/envs_post-receive @@ -1,6 +1,6 @@ #!/bin/bash cd /home/mayx/ -rm -rf public_html +rm -rf public_html blog/* git --work-tree=/home/mayx/blog --git-dir=/home/mayx/blog.git checkout -f cd blog mkdir Mabbs diff --git a/_tools/serv00_post-receive b/_tools/serv00_post-receive index ab45da0..dfbb126 100644 --- a/_tools/serv00_post-receive +++ b/_tools/serv00_post-receive @@ -1,6 +1,6 @@ #!/bin/bash cd /home/Mayx/ -rm -rf domains/mayx.serv00.net/public_html/ +rm -rf domains/mayx.serv00.net/public_html/ blog/* git --work-tree=/home/Mayx/blog --git-dir=/home/Mayx/repo/git/pub/mayx checkout -f cd blog mkdir Mabbs diff --git a/assets/quine.rar b/assets/quine.rar Binary files differnew file mode 100644 index 0000000..9632b6f --- /dev/null +++ b/assets/quine.rar |