archive
原生 ZIP、IPA 等容器的安全检查、解压与创建。
archive 是 PythonIDE 的原生 ZIP/ZIP64 内核。它可检查、列出、读取、解压、创建和事务化更新 .zip、.ipa、.whl、.jar、.apk、.npz 等 ZIP 容器,不依赖 Python zipfile。
格式边界:支持 ZIP32、ZIP64、stored与deflate。加密、多磁盘、符号链接、特殊文件和未知压缩方法会以稳定错误码失败,不会静默跳过。
声明式 MiniApp 需要文件能力:
{
"capabilities": ["file_system"]
}
普通路径和所有输出必须位于当前 MiniApp 授权范围。系统分享页交付的准确 LaunchItem.path 是本次运行的只读输入例外;它可以被检查、读取或作为解压/更新副本的来源,但不能原地修改。
#模块概览
| 项 | 说明 |
|---|---|
| 导入 | import archive |
| 容器 | ZIP32、ZIP64 及常见 ZIP 衍生格式 |
| 压缩 | stored、deflate |
| 大文件 | 原生分块读写;大条目使用 extract_entry(),不会经 JSON/Base64 整包传输 |
| 写入 | 目录描述符锚定的同目录 staging、完整校验、原子安装;失败或取消保留旧目标 |
| 安全 | 工作区授权、逐级 openat/O_NOFOLLOW、Zip Slip、symlink、特殊文件、结构和资源预算校验;消费内容时校验 CRC |
| 生命周期 | operation ID、进度、主动取消、MiniApp 会话停止自动取消、旧会话拒绝提交 |
#快速开始
import archive
info = archive.inspect("/path/to/App.ipa")
print(info["file_count"], info["uncompressed_size"])
for entry in archive.list("/path/to/App.ipa", include_directories=False):
print(entry["path"], entry["compression"], entry["uncompressed_size"])
result = archive.extract(
"/path/to/App.ipa",
"/path/to/IPA-Contents",
overwrite=True,
)
print("已解压:", result["path"])
处理系统分享页交付的 IPA:
import os
import archive
import miniapp_runtime
request = miniapp_runtime.launch_request(required=True)
ipa_path = request.first.path
output = os.path.expanduser("~/Documents/IPA-Contents")
summary = archive.inspect(ipa_path)
result = archive.extract(ipa_path, output, overwrite=True)
print(summary["entry_count"], result["path"])
不要读取宿主环境变量寻找分享文件。只使用 LaunchItem.path;输出仍写入 MiniApp 工作区。
#API 速查
| API | 作用 |
|---|---|
inspect(path, **limits) | 校验容器结构并返回汇总 |
list(path, include_directories=True, **limits) | 返回结构已校验的条目与元数据 |
read_entry(path, entry_path, max_bytes=...) | 小条目直接读取为 bytes |
extract_entry(path, entry_path, destination_path, ...) | 大条目流式、原子写入文件 |
extract(path, destination_path, ...) | 事务化解压整个容器 |
create(source_path, destination_path, ...) | 从普通文件或目录创建 ZIP/ZIP64 |
update(path, operations, ...) | 一次原子提交多项增删改 |
transaction(path, **options) | 创建 ArchiveTransaction 组合事务 |
add(...) / replace(...) / remove(...) | 单项更新便利函数,内部复用 update() |
new_operation_id() | 生成 operation ID |
progress(operation_id) | 查询线程安全的进度快照 |
cancel(operation_id) | 请求取消当前调用者拥有的操作 |
ArchiveError | 原生错误;用稳定的 code 分支 |
所有长操作都可接收 operation_id 和同一组可调低的安全预算:
max_entries=100_000
max_entry_size=8 * 1024**3
max_total_size=16 * 1024**3
max_archive_size=8 * 1024**3
max_compression_ratio=200.0
调用者可以降低预算,不能高于平台硬上限。
#inspect 与 list
inspect(path, *, operation_id=None, **limits) 返回:
| 字段 | 说明 |
|---|---|
format | 当前为 zip |
path | 已授权、已检查的容器路径 |
archive_size | 容器自身字节数 |
entry_count | 文件与目录总条目数 |
file_count / directory_count | 文件数与目录数 |
compressed_size | 条目压缩大小总和 |
uncompressed_size | 条目展开大小总和 |
list(path, *, include_directories=True, operation_id=None, **limits) 返回确定性顺序的条目。可用字段包括:
{
"path": "Payload/App.app/Info.plist",
"type": "file",
"is_directory": False,
"compression": "deflate",
"compressed_size": 1234,
"uncompressed_size": 4096,
"crc32": "12ab34cd",
"mode": 0o644,
"modified_at": 1_725_000_000.0,
}
mode 和 modified_at 只在容器提供可信元数据时出现。include_directories=False 会在原生响应编码和 16 MiB 返回预算计算之前过滤目录;容器解析与安全检查仍覆盖全部条目。
inspect() 和 list() 只解析结构及元数据,不为验证 CRC 而解压所有内容,因此可以稳定处理大型 IPA。CRC 会在 read_entry()、extract_entry()、extract() 和 update() 实际消费条目内容时流式校验。Info-ZIP Unicode Path 0x7075 是经过版本、原始名称 CRC 和 UTF-8 校验后的规范路径,同样参与路径安全和碰撞检查。
#读取或解压单个条目
小文件可直接读为 Python bytes:
plist_bytes = archive.read_entry(
"/path/to/App.ipa",
"Payload/App.app/Info.plist",
)
read_entry() 经过 JSON/Base64 桥,因此 max_bytes 的默认值和硬上限均为 8 MiB。更大的文件必须流式写入:
result = archive.extract_entry(
"/path/to/App.ipa",
"Payload/App.app/AppBinary",
"/path/to/output/AppBinary",
overwrite=True,
preserve_metadata=True,
)
extract_entry() 只接受普通文件条目,先校验 CRC,再原子安装目标。preserve_metadata=True 会恢复安全的 0o777 权限位和修改时间,不恢复 setuid、setgid 或其他危险类型位。
#解压与创建
extract(path, destination_path, *, overwrite=False, operation_id=None, **limits)
- 目标必须是授权范围内的目录路径。
overwrite=False且目标存在时抛出destination_exists。- 覆盖时,旧目录保留到全部条目通过路径、CRC、预算和会话校验之后。
overwrite=True是整棵目录的原子替换,不与旧目录合并;提交前旧目录内部的并发修改也会随旧树一起被替换。- 失败、取消或会话过期都不会提交半成品。
- 这里的“原子”指进程运行期间的可见性与失败回滚:提交前只看见旧树,提交后只看见新树。它不承诺设备突然断电后的逐文件
fsync持久性;需要跨断电保存的业务结果应在操作成功后再使用平台的持久任务/备份策略。
create(source_path, destination_path, *, compression="deflate", include_root=False, overwrite=False, operation_id=None, **limits)
result = archive.create(
"/path/to/MyProject",
"/path/to/MyProject.zip",
compression="deflate",
include_root=True,
overwrite=True,
)
创建过程按相对路径排序,保留安全的可移植权限和修改时间,拒绝源目录中的 symlink、特殊文件以及位于源目录内部的输出路径。ZIP32 不足时自动写出 ZIP64。若 deflate 结果没有收益或超过压缩比预算,会安全回退为 stored。
#原子更新与事务
update() 将全部操作写入一个 staging 容器,对最终容器中保留及重写的全部 payload 完成结构和 CRC 自检后只提交一次。未修改条目会流式验证原内容 CRC,同时保留原压缩数据和可复用的 ZIP 元数据,避免无意义的解压再压缩。
result = archive.update(
"/path/to/App.ipa",
[
{
"action": "replace",
"entry_path": "Payload/App.app/Info.plist",
"source_path": "/path/to/new-Info.plist",
"compression": "deflate",
},
{
"action": "add",
"entry_path": "Metadata/report.json",
"data": b'{"checked":true}',
"mode": 0o644,
},
{
"action": "remove",
"entry_path": "Metadata/old.json",
},
],
)
add / replace 必须且只能提供一种内容来源:source_path、小型 data/data_base64,或 is_directory=True。大型内容使用 source_path 流式写入。compression 可为 preserve、stored 或 deflate。
也可用上下文管理器构建同一事务:
with archive.transaction("/path/to/data.zip") as tx:
tx.replace("config.json", data=b"{}")
tx.add("bin/tool", source_path="/path/to/tool", mode=0o755)
tx.remove("old.txt")
print(tx.result)
ArchiveTransaction.add()、replace()、remove() 只是收集通用 entry operation;最终仍由一个 update() 原子提交,并不是为某类 MiniApp 硬编码的 API。顶层 archive.add()、archive.replace()、archive.remove() 也是同一内核的便利入口。
传入 destination_path 可生成修改后的副本;省略时原子替换原容器。外部只读输入只能写成工作区内的新副本。
同一路径的更新在进程内串行化。原地更新会记录已打开源文件的完整文件指纹,并在原子交换后核对换出的旧对象;若 Files、另一个线程或其他进程已替换目标,会回滚交换并抛出 archive_conflict,不会静默覆盖较新的版本。目录解压覆盖只对目标根对象做 O(1) 比较交换,不在取消/会话提交门内遍历旧目录;旧树清理在提交后进行。
#进度与取消
Archive API 是同步调用。需要显示进度或取消时,在工作线程执行操作,并从另一个线程查询同一个 ID:
import archive
import threading
operation_id = archive.new_operation_id()
def extract_worker():
archive.extract(
"/path/to/data.zip",
"/path/to/output",
operation_id=operation_id,
)
worker = threading.Thread(target=extract_worker)
worker.start()
snapshot = archive.progress(operation_id)
print(snapshot["phase"], snapshot["fraction"])
archive.cancel(operation_id)
worker.join()
进度快照包括 phase、entries_completed、entries_total、bytes_completed、bytes_total、fraction、cancel_requested、finished,失败终态还包含 error_code。完成记录有界保留;未知或不属于当前运行的 ID 不会泄露其他会话状态。
MiniApp 停止时,宿主自动取消该会话拥有的操作。取消令牌、会话 lease 与最终 rename 共用线性化提交门:取消/停止先到达时不提交;提交点先到达时,停止会等待这次原子安装结束。停止返回后,旧运行不能再新增落盘结果。
为防止单个 MiniApp 或全局任务挤占文件描述符和内存,同一运行最多同时拥有 4 个 Archive 操作,宿主全局最多 16 个;超过限制立即返回 too_many_operations,不会排队形成隐藏延迟。
#安全预算
| 预算 | 默认及硬上限 |
|---|---|
| 条目数 | 100,000 |
| 单条目展开大小 | 8 GiB |
| 总展开大小 | 16 GiB |
| ZIP 文件大小 | 8 GiB |
| 单条目压缩比 | 200:1 |
read_entry() 内联数据 | 8 MiB |
| 单个 entry path | 1,024 UTF-8 bytes、最多 128 个组件 |
| 单个 path 组件 | 255 UTF-8 bytes |
| 中央目录元数据 | 32 MiB |
| C bridge 返回 JSON | 16 MiB |
| 写入磁盘保留余量 | 256 MiB |
| 同一运行并发操作 | 4 |
| 宿主全局并发操作 | 16 |
| 待回收 staging/quarantine 所有权记录 | 1,024 |
路径包含绝对路径、空组件、.、..、Windows 盘符、反斜杠、NUL、超出长度/层数限制、Unicode/大小写碰撞,或文件/目录冲突时,整个操作失败。不会自动修正不安全名称。路径冲突检查使用线性前缀索引,不会在 100,000 条目下退化为平方级扫描。POSIX external attributes、macOS host type 和 Info-ZIP ASi Unix 0x756e 会统一解释;symlink、特殊文件、重复字段、CRC 损坏或 local/central 类型冲突均拒绝。
extract()、extract_entry()、create() 和 update() 会在写 staging 前检查目标卷可用容量,并在流式写入期间继续执行硬计数。模块只按宿主私有日志回收自己登记、父目录与对象 identity/类型完全匹配的 staging/quarantine 残留,不按文件名前缀扫描,更不会删除普通用户文件或已被并发替换的对象。
POSIX 没有“仅当名称仍指向预期 inode 时才删除”的原子 unlinkat。因此嵌入式 Python 启动后,清理只会把已验证对象原子移到宿主私有 quarantine;物理删除延迟到下一次冷启动、Python 初始化之前的可信阶段。冷启动最多删除 128 个节点,但会在最多 1,025 条持久记录中独立扫描并轮转起点,无法打开的外部/security-scoped 父目录会保留所有权记录、写入诊断并在以后可访问的冷启动重试,不会阻塞后面的工作区记录。
运行期不会用 ftruncate 提前释放普通文件内容,因为同一 inode 可能还有用户创建或并发创建的硬链接,截断会损坏那些别名。代价是取消结果或被替换的旧容器/目录可能暂时占用磁盘,直到下次可信回收;单个容器受 8 GiB 上限、单棵展开目录受 16 GiB 上限约束。所有权日志最多保留 1,024 个常驻工作项,并另留一个只用于原子 quarantine 交接的内部槽位;达到上限时会在创建新 staging 之前返回 resource_limit,绝不会静默丢弃旧记录或留下未登记的临时对象。
#常见错误码
ArchiveError.code | 含义 |
|---|---|
invalid_input | 参数、operation ID 或更新操作无效 |
source_not_found | 输入容器、源文件或条目不存在 |
destination_exists | 未允许覆盖但目标存在 |
invalid_archive | ZIP 结构、边界或 local/central metadata 不一致 |
unsupported_container | 不是受支持的 ZIP 容器 |
multi_disk_unsupported | 多磁盘 ZIP 不受支持 |
encrypted_entry | 条目已加密 |
unsupported_compression | 不是 stored/deflate |
unsupported_entry_type / symlink_entry | 条目或源文件不是允许的普通文件/目录 |
unsafe_path / duplicate_entry / path_conflict | 路径不安全、重复或发生结构冲突 |
too_many_entries / entry_too_large / archive_too_large | 超出安全预算 |
too_many_operations | 当前运行或宿主已达到 Archive 并发上限 |
resource_limit | 待回收临时对象已达到有界所有权日志上限;冷启动回收后重试 |
metadata_too_large / result_too_large | 中央元数据或 bridge 返回超过固定内存预算 |
archive_conflict | 原地更新期间目标被另一个写入者修改,事务未覆盖对方版本 |
compression_ratio_exceeded | 压缩比超过预算 |
checksum_mismatch | 内容 CRC 校验失败 |
file_denied | 路径不在当前运行的授权范围 |
miniapp_session_expired | 发起操作的 MiniApp 会话已结束 |
cancelled | 主动取消或会话停止 |
io_error | 文件系统或底层流读写失败 |
#常见错误
| 错误写法 | 后果 | 修正 |
|---|---|---|
使用 zipfile.extractall() 处理不可信输入 | 绕过宿主路径、会话与资源预算 | 使用 archive.extract() |
用 read_entry() 读取几百 MiB 文件 | Base64/JSON 放大内存 | 使用 extract_entry() |
把 LaunchItem.path 当成可写目标 | 外部授权是只读的 | 输出到工作区副本 |
| 每次增改都解压整个包再重建 | 慢且丢失未修改条目元数据 | 使用一个 update() / transaction() |
在 AppUI body() 中运行长操作 | 重建时重复执行并阻塞交互 | 放进用户回调和工作线程 |
#相关文档
| 文档 | 用途 |
|---|---|
| 文件选择 | 由用户选择容器或输出位置 |
| MiniApp 原生能力 | 声明 file_system、读取 LaunchRequest 与理解会话授权 |
| background_download | 下载大型容器后交给 Archive |
#预期效果
合法 ZIP32/ZIP64 可稳定检查、读取、解压、创建和更新;未修改条目不会被无意义重编码。恶意路径、symlink、加密、多磁盘、未知压缩和炸弹会在结构检查阶段明确失败;CRC 损坏会在 read/extract/update 消费 payload 时明确失败。取消和旧会话不会提交半成品目标。