#!/usr/bin/env python3 """独立校验SDK tar.gz的清单、文件长度和SHA，不解包或运行SDK代码。""" import argparse import gzip import hashlib import json import os from pathlib import Path import re import stat import tarfile MAX_ARCHIVE = 512 * 1024 * 1024 MAX_EXPANDED = 2 * 1024 * 1024 * 1024 MAX_MEMBER = 256 * 1024 * 1024 MAX_MANIFEST = 4 * 1024 * 1024 MAX_MEMBERS = 20_000 MAX_PATH_METADATA = 16 * 1024 * 1024 MAX_DEPTH = 32 CHUNK = 64 * 1024 MAX_EXTENSION = 64 * 1024 class ArchiveError(ValueError): """归档内容或完整性不符合SDK清单。""" def require(condition, message): if not condition: raise ArchiveError(message) class BoundedReader: def __init__(self, stream, limit, digest=None): self.stream, self.limit, self.digest, self.count = stream, limit, digest, 0 def read(self, size=-1): # Never let a caller allocate an unbounded buffer before checking limits. size = min(CHUNK if size < 0 else size, self.limit - self.count + 1) data = self.stream.read(size) self.count += len(data) require(self.count <= self.limit, "归档压缩或展开大小超过校验上限") if self.digest is not None: self.digest.update(data) return data class BoundedTarInfo(tarfile.TarInfo): def _proc_member(self, archive): # tarfile consumes PAX/GNU headers before exposing a member to callers. # Bound that earlier read too, including nested extension headers. archive.sdk_headers = getattr(archive, "sdk_headers", 0) + 1 archive.sdk_depth = getattr(archive, "sdk_depth", 0) + 1 try: require(archive.sdk_headers <= MAX_MEMBERS, "归档头部超过20000项") require(archive.sdk_depth <= MAX_DEPTH, "归档扩展头部嵌套超过32层") require(0 <= self.size <= MAX_MEMBER, "单个归档成员超过256MiB") if self.type in (tarfile.XHDTYPE, tarfile.XGLTYPE, tarfile.GNUTYPE_LONGNAME, tarfile.GNUTYPE_LONGLINK): require(self.size <= MAX_EXTENSION, "单个归档扩展头部超过64KiB") require(self.type != tarfile.GNUTYPE_SPARSE, "归档不允许稀疏文件") return super()._proc_member(archive) finally: archive.sdk_depth -= 1 def safe_name(name): require(isinstance(name, str), "归档路径必须是字符串") try: encoded = name.encode("utf-8") except UnicodeError as error: raise ArchiveError("归档路径编码无效") from error require(0 < len(encoded) <= 4096 and "\\" not in name and not any(ord(c) < 32 or ord(c) == 127 for c in name), f"归档路径无效：{name!r}") parts = name.split("/") require(len(parts) <= MAX_DEPTH, "归档路径层级超过32层") require(all(p and p not in (".", "..") and ":" not in p and not p.endswith((" ", ".")) and re.fullmatch(r"(?:CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(?:\..*)?", p, re.I) is None for p in parts), f"归档路径不安全：{name!r}") return parts def unique_object(pairs): result = {} for key, value in pairs: require(key not in result, f"清单含重复字段：{key!r}") result[key] = value return result def reject_constant(value): raise ArchiveError("SDK清单不允许非有限JSON数值") def verify_archive(path, expected_sha256=None): if expected_sha256 is not None: require(re.fullmatch(r"[0-9a-fA-F]{64}", expected_sha256) is not None, "预期SHA256必须是64位十六进制字符串") expected_sha256 = expected_sha256.lower() digest = hashlib.sha256() observed, seen, directories, files = {}, set(), set(), set() root, manifest, members = None, None, 0 path_metadata = 0 with Path(path).open("rb") as raw: before = os.fstat(raw.fileno()) require(stat.S_ISREG(before.st_mode) and before.st_size <= MAX_ARCHIVE, "输入须为不超过512MiB的普通归档文件") compressed = BoundedReader(raw, MAX_ARCHIVE, digest) with gzip.GzipFile(fileobj=compressed, mode="rb") as unpacked: expanded = BoundedReader(unpacked, MAX_EXPANDED) with tarfile.open(fileobj=expanded, mode="r|", tarinfo=BoundedTarInfo) as archive: for member in archive: members += 1 require(members <= MAX_MEMBERS, "归档成员超过20000项") parts = safe_name(member.name) if root is None: root = parts[0] require(parts[0] == root, "SDK归档必须只有一个顶层目录") path_metadata += len(member.name.encode("utf-8")) * 2 + 256 require(path_metadata <= MAX_PATH_METADATA, "归档路径元数据超过16MiB预算") key = member.name.casefold() require(key not in seen, f"重复或大小写冲突的归档路径：{member.name!r}") seen.add(key) for i in range(1, len(parts)): parent = "/".join(parts[:i]).casefold() require(parent not in files, "归档文件不能同时用作父目录") if parent not in directories: path_metadata += len(parent.encode("utf-8")) + 96 require(path_metadata <= MAX_PATH_METADATA, "归档路径元数据超过16MiB预算") directories.add(parent) if member.isdir(): require(member.size == 0 and key not in files, "目录成员无效") if key not in directories: path_metadata += len(key.encode("utf-8")) + 96 require(path_metadata <= MAX_PATH_METADATA, "归档路径元数据超过16MiB预算") directories.add(key) continue require(member.type in (tarfile.REGTYPE, tarfile.AREGTYPE) and not member.issparse(), "归档不允许链接、稀疏文件或特殊成员") require(len(parts) > 1 and key not in directories, "归档文件与目录冲突") require(0 <= member.size <= MAX_MEMBER, "单个归档成员超过256MiB") relative = "/".join(parts[1:]) is_manifest = relative == "bundle.json" if is_manifest: require(member.size <= MAX_MANIFEST, "SDK清单超过4MiB") hashed, length, chunks = hashlib.sha256(), 0, [] with archive.extractfile(member) as payload: while True: chunk = payload.read(CHUNK) if not chunk: break hashed.update(chunk) length += len(chunk) if is_manifest: chunks.append(chunk) require(length == member.size, f"归档文件被截断：{relative!r}") observed[relative] = (length, hashed.hexdigest()) files.add(key) if is_manifest: try: manifest = json.loads(b"".join(chunks), object_pairs_hook=unique_object, parse_constant=reject_constant) except (UnicodeError, json.JSONDecodeError, RecursionError) as error: raise ArchiveError("SDK清单JSON无效") from error # Drain the stream, including buffered tar padding. This verifies # gzip CRC/length and rejects an unnoticed second tar payload. while True: padding = archive.fileobj.read(CHUNK) if not padding: break require(not any(padding), "归档结束标记之后含非空数据") after = os.fstat(raw.fileno()) require((before.st_size, before.st_mtime_ns, before.st_ctime_ns) == (after.st_size, after.st_mtime_ns, after.st_ctime_ns), "校验期间归档文件发生变化") require(compressed.count == before.st_size, "归档压缩数据未完整读取") archive_sha = digest.hexdigest() require(expected_sha256 is None or archive_sha == expected_sha256, "归档SHA256与指定值不一致") require(isinstance(manifest, dict), "归档缺少bundle.json对象") require(type(manifest.get("schemaVersion")) is int and manifest["schemaVersion"] == 1, "不支持的SDK清单版本") require(isinstance(manifest.get("sdkVersion"), str) and 0 < len(manifest["sdkVersion"]) <= 256, "SDK版本无效") core = manifest.get("coreCommit") require(isinstance(core, str) and re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", core), "核心提交标识无效") inventory = manifest.get("files") require(isinstance(inventory, list) and len(inventory) <= MAX_MEMBERS, "SDK文件清单无效") declared = {} for item in inventory: require(isinstance(item, dict), "清单文件记录必须是对象") name = item.get("name") safe_name(name) require(name != "bundle.json" and name not in declared, "清单含重复文件或自引用") size, checksum = item.get("bytes"), item.get("sha256") require(type(size) is int and 0 <= size <= MAX_MEMBER, f"文件长度无效：{name!r}") require(isinstance(checksum, str) and re.fullmatch(r"[0-9a-f]{64}", checksum), f"文件SHA256无效：{name!r}") declared[name] = (size, checksum) require(set(observed) == set(declared) | {"bundle.json"}, "实际文件与SDK清单不一致") for name, expected in declared.items(): require(observed[name] == expected, f"文件长度或SHA256不匹配：{name!r}") platforms = manifest.get("platforms", {}) require(isinstance(platforms, dict) and len(platforms) <= 32 and all(isinstance(k, str) and len(k) <= 128 and isinstance(v, str) and len(v) <= 1024 for k, v in platforms.items()), "平台状态元数据无效") return {"status": "passed", "validation": "archive-integrity", "sdkVersion": manifest["sdkVersion"], "coreCommit": core, "archiveSha256": archive_sha, "archiveBytes": before.st_size, "expectedSha256Matched": expected_sha256 is not None, "verifiedFiles": len(declared), "archiveMembers": members, "platforms": platforms} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("archive", type=Path) parser.add_argument("--sha256", help="可选的外部预期归档SHA256") args = parser.parse_args() try: result = verify_archive(args.archive, args.sha256) except (ArchiveError, OSError, EOFError, tarfile.TarError, ValueError) as error: print(json.dumps({"status": "failed", "error": str(error)}, ensure_ascii=False)) return 1 print(json.dumps(result, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())