154 lines
7.9 KiB
Python
154 lines
7.9 KiB
Python
"""Synthetic malformed PAK tests; no engine, extracted assets or native-build claims."""
|
|
from dataclasses import replace
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zlib
|
|
from contextlib import redirect_stdout
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT/'tools'))
|
|
import pak_inventory as p
|
|
|
|
|
|
def chunk(tag, data): return tag+struct.pack('>I', len(data))+data
|
|
|
|
def entry(name=b'file.txt', offset=56, stored=3, expanded=3, flags=b'\0\0', reserved=b'\0'*6, kind=1):
|
|
return bytes([kind, len(name)])+name+struct.pack('<III', offset, stored, expanded)+reserved+flags+b'\0'*4
|
|
|
|
def archive(data=b'abc', tree=None, head=None):
|
|
if tree is None: tree=b'\0\0'+struct.pack('<I', 1)+entry(stored=len(data), expanded=len(data))
|
|
if head is None: head=b'\x03\0\x01\0'+b'\0'*24
|
|
body=b'PAC1'+chunk(b'HEAD',head)+chunk(b'DATA',data)+chunk(b'FILE',tree)
|
|
return b'FORM'+struct.pack('>I',len(body))+body
|
|
|
|
|
|
class PakTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory(prefix='erri-pak-test-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.home=Path(self.temp.name);self.path=self.home/'synthetic.pak'
|
|
|
|
def inspect(self, data, limits=p.Limits()):
|
|
self.path.write_bytes(data)
|
|
return p.inventory(self.path, limits)
|
|
|
|
def refused(self, data, limits=p.Limits()):
|
|
with self.assertRaises((p.PakError,OSError)): self.inspect(data,limits)
|
|
|
|
def test_raw_and_compressed_nested_hashes_without_extraction(self):
|
|
text=b'synthetic text '*200
|
|
compressed=zlib.compress(text)
|
|
tree=b'\0\0'+struct.pack('<I',2)+entry()
|
|
tree+=b'\0\x06folder'+struct.pack('<I',1)+entry(b'data.c',59,len(compressed),len(text),b'\x01\x06')
|
|
data=archive(b'abc'+compressed,tree)
|
|
report=self.inspect(data)
|
|
self.assertEqual(report['entry_count'],2)
|
|
self.assertEqual(report['tree_node_count'],3)
|
|
self.assertEqual(report['archive_sha256'],hashlib.sha256(data).hexdigest())
|
|
rows={r['path']:r for r in report['entries']}
|
|
self.assertEqual(rows['folder/data.c']['sha256'],hashlib.sha256(text).hexdigest())
|
|
self.assertFalse(report['platform_or_resource_semantics_verified'])
|
|
self.assertFalse(report['release_qualified'])
|
|
self.assertEqual(list(self.home.iterdir()),[self.path])
|
|
|
|
def test_truncation_at_every_byte_is_rejected(self):
|
|
original=archive()
|
|
for length in range(len(original)):
|
|
with self.subTest(length=length): self.refused(original[:length])
|
|
|
|
def test_header_versions_chunks_lengths_and_trailing_bytes(self):
|
|
original=archive()
|
|
for offset in (0,4,8,12,16,48):
|
|
corrupt=bytearray(original);corrupt[offset]^=1;self.refused(bytes(corrupt))
|
|
self.refused(archive(head=b'\x04\0\x01\0'+b'\0'*24))
|
|
self.refused(archive(head=b'\x03\0\x01\0'+b'\0'*23+b'\1'))
|
|
self.refused(archive(head=b'\0'))
|
|
self.refused(original+b'extra')
|
|
self.refused(archive(tree=b'\0\0'+struct.pack('<I',1)+entry()+b'extra'))
|
|
|
|
def test_paths_duplicates_case_and_file_directory_conflicts(self):
|
|
for name in (b'..',b'.',b'',b'/root',b'a/b',b'a\\b',b'C:x',b'a\0b',b'a\nb',b'a.',b'a ',b'CON.txt',b'\xff', 'e\u0301'.encode()):
|
|
with self.subTest(name=name):self.refused(archive(tree=b'\0\0'+struct.pack('<I',1)+entry(name)))
|
|
for first,second in ((b'file',b'file'),(b'FILE',b'file')):
|
|
tree=b'\0\0'+struct.pack('<I',2)+entry(first)+entry(second,59)
|
|
self.refused(archive(b'abcdef',tree))
|
|
tree=b'\0\0'+struct.pack('<I',2)+entry(b'folder')+b'\0\x06folder'+struct.pack('<I',0)
|
|
self.refused(archive(tree=tree))
|
|
|
|
def test_descriptor_bounds_overlap_gaps_and_unsupported_flags(self):
|
|
variants=[entry(offset=0),entry(offset=58),entry(stored=4),entry(stored=3,expanded=2),
|
|
entry(flags=b'\1\x09'),entry(reserved=b'\1'+b'\0'*5),entry(kind=2)]
|
|
for raw in variants:self.refused(archive(tree=b'\0\0'+struct.pack('<I',1)+raw))
|
|
self.refused(archive(b'abcdef',b'\0\0'+struct.pack('<I',2)+entry(b'a')+entry(b'b',56)))
|
|
self.refused(archive(b'abcdef',b'\0\0'+struct.pack('<I',1)+entry()))
|
|
self.refused(archive(tree=b'\1\0'+struct.pack('<I',1)+entry()))
|
|
|
|
def test_zlib_exact_sizes_checksums_trailing_and_truncated_streams(self):
|
|
original=b'x'*1000;compressed=zlib.compress(original)
|
|
for data,size in ((compressed[:-1],1000),(compressed+b'extra',1000),(compressed+compressed,2000),
|
|
(compressed,999),(compressed,1001),(b'bad-stream',1000)):
|
|
tree=b'\0\0'+struct.pack('<I',1)+entry(stored=len(data),expanded=size,flags=b'\1\6')
|
|
self.refused(archive(data,tree))
|
|
damaged=bytearray(compressed);damaged[-1]^=1
|
|
tree=b'\0\0'+struct.pack('<I',1)+entry(stored=len(damaged),expanded=1000,flags=b'\1\6')
|
|
self.refused(archive(bytes(damaged),tree))
|
|
|
|
def test_large_decompression_streamed_in_bounded_blocks(self):
|
|
text=b'abcd'*(p.BLOCK//2+17);data=zlib.compress(text)
|
|
tree=b'\0\0'+struct.pack('<I',1)+entry(stored=len(data),expanded=len(text),flags=b'\1\6')
|
|
report=self.inspect(archive(data,tree),replace(p.Limits(),expansion_ratio=2048))
|
|
self.assertEqual(report['entries'][0]['sha256'],hashlib.sha256(text).hexdigest())
|
|
|
|
def test_preflight_limits_and_small_declared_bomb(self):
|
|
self.refused(archive(),replace(p.Limits(),archive_bytes=20))
|
|
self.refused(archive(),replace(p.Limits(),tree_bytes=1))
|
|
self.refused(archive(),replace(p.Limits(),entry_bytes=2))
|
|
self.refused(archive(),replace(p.Limits(),total_bytes=2))
|
|
tree=b'\0\0'+struct.pack('<I',2)+entry(b'a')+entry(b'b',59)
|
|
self.refused(archive(b'abcdef',tree),replace(p.Limits(),entries=1))
|
|
tree=b'\0\0'+struct.pack('<I',1)+b'\0\1a'+struct.pack('<I',1)+b'\0\1b'+struct.pack('<I',1)+entry()
|
|
self.refused(archive(tree=tree),replace(p.Limits(),depth=1))
|
|
payload=zlib.compress(b'x'*2_000_000)
|
|
tree=b'\0\0'+struct.pack('<I',1)+entry(stored=len(payload),expanded=2_000_000,flags=b'\1\6')
|
|
self.refused(archive(payload,tree),replace(p.Limits(),expansion_ratio=2))
|
|
tree=b'\0\0'+struct.pack('<I',1)+entry(stored=len(payload),expanded=3,flags=b'\1\6')
|
|
self.refused(archive(payload,tree)) # output limited to declared size + one, not 2 MB
|
|
|
|
def test_fifo_and_symlink_inputs_refused_without_blocking(self):
|
|
os.mkfifo(self.path)
|
|
with self.assertRaises(p.PakError):p.inventory(self.path)
|
|
self.path.unlink();self.path.symlink_to(self.home/'missing')
|
|
with self.assertRaises(OSError):p.inventory(self.path)
|
|
|
|
def test_json_output_is_new_and_no_partial_report_on_failure(self):
|
|
self.path.write_bytes(archive());output=self.home/'report.json'
|
|
with redirect_stdout(io.StringIO()):self.assertEqual(p.main([str(self.path),'--output',str(output)]),0)
|
|
before=output.read_bytes()
|
|
with redirect_stdout(io.StringIO()):self.assertEqual(p.main([str(self.path),'--output',str(output)]),1)
|
|
self.assertEqual(output.read_bytes(),before)
|
|
bad=self.home/'bad.json';self.path.write_bytes(b'bad')
|
|
with redirect_stdout(io.StringIO()):self.assertEqual(p.main([str(self.path),'--output',str(bad)]),1)
|
|
self.assertFalse(bad.exists())
|
|
|
|
|
|
def test_changed_archive_during_hashing_is_refused(self):
|
|
self.path.write_bytes(archive())
|
|
original=p.content_hash
|
|
def change(stream, item):
|
|
result=original(stream,item)
|
|
with self.path.open('ab') as output:output.write(b'changed')
|
|
return result
|
|
with patch.object(p,'content_hash',side_effect=change):
|
|
with self.assertRaisesRegex(p.PakError,'changed during inspection'):p.inventory(self.path)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|