Exercise 3 solution import struct, zlib MAGIC = b"TBX1" VERSION = 1 HEADER = struct.Struct("<4sBBH") # magic, version, flags, entry count ENTRY = struct.Struct(" 255: raise FormatError("entry name too long") def write_tbx(entries, compress=False): if len(entries) > MAX_ENTRIES: raise FormatError("too many entries") seen, body = set(), [] for name, data in entries: check_name(name) if name in seen: raise FormatError(f"duplicate name {name!r}") seen.add(name) if len(data) > MAX_ORIGINAL: raise FormatError(f"{name!r} is over {MAX_ORIGINAL} bytes") stored = zlib.compress(data, 9) if compress else data raw_name = name.encode("utf-8") body.append(ENTRY.pack(len(raw_name), len(stored), len(data), zlib.crc32(data))) body.append(raw_name) body.append(stored) flags = FLAG_ZLIB if compress else 0 return HEADER.pack(MAGIC, VERSION, flags, len(entries)) + b"".join(body) class Reader: def __init__(self, buf): self.buf = bytes(buf) self.pos = 0 def remaining(self): return len(self.buf) - self.pos def take(self, n): if n < 0 or n > self.remaining(): raise FormatError(f"need {n} bytes at offset {self.pos}, {self.remaining()} left") chunk = self.buf[self.pos:self.pos + n] self.pos += n return chunk def unpack(self, st): return st.unpack(self.take(st.size)) def inflate(data, expected): d = zlib.decompressobj() try: out = d.decompress(data, expected + 1) except zlib.error as e: raise FormatError(f"corrupt zlib data: {e}") from e if len(out) != expected or not d.eof or d.unused_data: raise FormatError(f"zlib data does not match the declared size {expected}") return out def read_tbx(buf): if len(buf) > MAX_FILE: raise FormatError(f"file over {MAX_FILE} bytes") r = Reader(buf) magic, version, flags, count = r.unpack(HEADER) if magic != MAGIC: raise FormatError(f"bad magic {magic!r}") if version != VERSION: raise FormatError(f"unsupported version {version}") if flags & ~KNOWN_FLAGS: raise FormatError(f"unknown flag bits {flags & ~KNOWN_FLAGS:#04x}") if count > MAX_ENTRIES: raise FormatError(f"entry count {count} over limit {MAX_ENTRIES}") if count * (ENTRY.size + 1) > r.remaining(): raise FormatError(f"entry count {count} cannot fit in {r.remaining()} bytes") out, seen = [], set() for _ in range(count): start = r.pos name_len, stored, original, crc = r.unpack(ENTRY) try: name = r.take(name_len).decode("utf-8") except UnicodeDecodeError as e: raise FormatError(f"entry at {start}: name is not UTF-8") from e check_name(name) if name in seen: raise FormatError(f"duplicate name {name!r}") seen.add(name) if original > MAX_ORIGINAL: raise FormatError(f"{name!r}: declared size {original} over limit") data = r.take(stored) if flags & FLAG_ZLIB: data = inflate(data, original) elif stored != original: raise FormatError(f"{name!r}: stored size {stored} != original size {original}") if zlib.crc32(data) != crc: raise FormatError(f"{name!r}: checksum mismatch") out.append((name, data)) if r.remaining(): raise FormatError(f"{r.remaining()} trailing bytes at offset {r.pos}") return out import random, tracemalloc from collections import Counter def make_entries(rng): entries = [] for i in range(rng.randint(0, 6)): size = rng.randint(0, 300) data = rng.randbytes(size) if rng.random() < 0.5 else bytes(size) entries.append((f"file{i}.bin", data)) return entries def selftest(seed=27, rounds=300, mutations=2000): rng = random.Random(seed) report = Counter() seeds = [] for _ in range(rounds): # round trips entries = make_entries(rng) for compress in (False, True): blob = write_tbx(entries, compress) report["round trips"] += 1 if read_tbx(blob) != entries: report["round trip FAILED"] += 1 seeds.append(blob) for _ in range(mutations): # random multi-byte damage blob = bytearray(rng.choice(seeds)) if not blob: continue for _ in range(rng.randint(1, 4)): blob[rng.randrange(len(blob))] = rng.randrange(256) if rng.random() < 0.3: del blob[rng.randrange(len(blob)):] try: read_tbx(bytes(blob)) report["accepted"] += 1 except FormatError: report["rejected"] += 1 except Exception as e: report["CRASH " + type(e).__name__] += 1 return report report = selftest() print(dict(report)) print("same seed, same result:", selftest() == report) bomb = zlib.compress(bytes(50_000_000), 9) liar = HEADER.pack(MAGIC, VERSION, FLAG_ZLIB, 1) + ENTRY.pack(1, len(bomb), 400_000, 0) + b"b" + bomb tracemalloc.start() try: read_tbx(liar) except FormatError as e: print("bomb ->", e) peak = tracemalloc.get_traced_memory()[1] tracemalloc.stop() print("peak under 5 MB:", peak < 5_000_000) # {'round trips': 600, 'rejected': 1985, 'accepted': 15} # same seed, same result: True # bomb -> zlib data does not match the declared size 400000 # peak under 5 MB: True (Run as written; the output matched the comments.) Explanation: the generator is seeded with random.Random(seed), so the whole run is repeatable. The report counts round trips and then outcomes of the damaged files; any exception other than FormatError appears as a CRASH key. The accepted files are damage that happened to leave a valid archive (for example a changed name byte, or a byte replaced by the same value). tracemalloc shows the capped decompressor never builds the 50-million-byte result.