257 lines
10 KiB
Python
257 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only strict inventory of the observed Enfusion FORM/PAC1 PAK format.
|
|
Names, declared sizes and verified content hashes only; never extract payloads.
|
|
"""
|
|
import argparse
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import struct
|
|
import unicodedata
|
|
import zlib
|
|
|
|
BLOCK = 1024 * 1024
|
|
|
|
|
|
class PakError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Limits:
|
|
archive_bytes: int = 4 * 1024**3
|
|
tree_bytes: int = 32 * 1024**2
|
|
entries: int = 250000 # files plus directories
|
|
depth: int = 64
|
|
entry_bytes: int = 1024**3
|
|
total_bytes: int = 8 * 1024**3
|
|
expansion_ratio: int = 1024
|
|
|
|
def validate(self):
|
|
if any(type(value) is not int or value <= 0 for value in vars(self).values()):
|
|
raise PakError('All limits must be positive integers')
|
|
if self.depth > 128:
|
|
raise PakError('Maximum supported tree depth is 128')
|
|
|
|
|
|
def exact(stream, count):
|
|
data = stream.read(count)
|
|
if len(data) != count:
|
|
raise PakError('Truncated archive')
|
|
return data
|
|
|
|
|
|
def component(raw):
|
|
try:
|
|
name = raw.decode('utf-8')
|
|
except UnicodeDecodeError as error:
|
|
raise PakError('Non-UTF-8 entry name') from error
|
|
if (not name or name in {'.', '..'} or any(char in name for char in '/\\:')
|
|
or any(ord(char) < 32 or ord(char) == 127 for char in name)
|
|
or name.endswith((' ', '.')) or unicodedata.normalize('NFC', name) != name
|
|
or re.fullmatch(r'(?:CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(?:\..*)?', name, re.I)):
|
|
raise PakError('Unsafe or noncanonical entry name')
|
|
return name
|
|
|
|
|
|
def parse_tree(tree, data_start, data_end, limits):
|
|
if len(tree) < 6 or tree[:2] != b'\0\0':
|
|
raise PakError('Unsupported FILE root header')
|
|
offset = 6
|
|
seen = set()
|
|
entries = []
|
|
count_seen = 0
|
|
total = 0
|
|
|
|
def take(size):
|
|
nonlocal offset
|
|
if offset + size > len(tree):
|
|
raise PakError('Truncated FILE entry')
|
|
result = tree[offset:offset+size]
|
|
offset += size
|
|
return result
|
|
|
|
def walk(parent, count, depth):
|
|
nonlocal count_seen, total
|
|
if depth > limits.depth or count > limits.entries - count_seen:
|
|
raise PakError('Directory depth/count limit exceeded')
|
|
for _ in range(count):
|
|
count_seen += 1
|
|
if count_seen > limits.entries:
|
|
raise PakError('Entry count limit exceeded')
|
|
kind, length = take(2)
|
|
name = component(take(length))
|
|
path = parent + name
|
|
key = path.casefold()
|
|
if key in seen:
|
|
raise PakError('Duplicate or case-colliding file/directory path: '+path)
|
|
seen.add(key)
|
|
if kind == 0:
|
|
children = struct.unpack('<I', take(4))[0]
|
|
walk(path+'/', children, depth+1)
|
|
elif kind == 1:
|
|
raw = take(24)
|
|
start, stored, size = struct.unpack_from('<III', raw)
|
|
flags = (raw[18], raw[19])
|
|
if raw[12:18] != b'\0'*6 or flags not in {(0, 0), (1, 6)}:
|
|
raise PakError('Unsupported FILE descriptor flags/reserved bytes: '+path)
|
|
if start < data_start or start > data_end or stored > data_end-start:
|
|
raise PakError('Payload outside DATA bounds: '+path)
|
|
if flags == (0, 0) and stored != size:
|
|
raise PakError('Raw payload stored/expanded size mismatch: '+path)
|
|
if flags == (1, 6) and stored == 0:
|
|
raise PakError('Empty compressed stream: '+path)
|
|
if size > limits.entry_bytes or size > max(1, stored)*limits.expansion_ratio:
|
|
raise PakError('Entry expansion/size limit exceeded: '+path)
|
|
total += size
|
|
if total > limits.total_bytes:
|
|
raise PakError('Aggregate expanded-data limit exceeded')
|
|
entries.append({'path': path, 'offset': start, 'stored_bytes': stored,
|
|
'content_bytes': size, 'compression': 'zlib' if flags[0] else 'raw',
|
|
'opaque_tail_hex': raw[20:24].hex()})
|
|
else:
|
|
raise PakError('Unsupported FILE node kind')
|
|
|
|
walk('', struct.unpack_from('<I', tree, 2)[0], 0)
|
|
if offset != len(tree):
|
|
raise PakError('Unparsed trailing FILE bytes')
|
|
# Observed archives have contiguous, non-overlapping, completely indexed DATA.
|
|
# Refuse sparse/unindexed/deduplicated variants instead of guessing semantics.
|
|
cursor = data_start
|
|
for entry in sorted(entries, key=lambda row: (row['offset'], row['stored_bytes'])):
|
|
if entry['offset'] != cursor:
|
|
raise PakError('Overlapping or unindexed DATA extent')
|
|
cursor += entry['stored_bytes']
|
|
if cursor != data_end:
|
|
raise PakError('Unindexed trailing DATA bytes')
|
|
return entries, total, count_seen
|
|
|
|
|
|
def content_hash(stream, entry):
|
|
stream.seek(entry['offset'])
|
|
remaining = entry['stored_bytes']
|
|
expected = entry['content_bytes']
|
|
produced = 0
|
|
hasher = hashlib.sha256()
|
|
decoder = zlib.decompressobj() if entry['compression'] == 'zlib' else None
|
|
|
|
def accept(data):
|
|
nonlocal produced
|
|
produced += len(data)
|
|
if produced > expected:
|
|
raise PakError('Expanded data exceeds declared size: '+entry['path'])
|
|
hasher.update(data)
|
|
|
|
try:
|
|
while remaining:
|
|
data = exact(stream, min(BLOCK, remaining))
|
|
remaining -= len(data)
|
|
if decoder is None:
|
|
accept(data)
|
|
continue
|
|
while data:
|
|
previous = data
|
|
output = decoder.decompress(data, min(BLOCK, expected-produced+1))
|
|
accept(output)
|
|
if decoder.unused_data or (decoder.eof and remaining):
|
|
raise PakError('Trailing/concatenated compressed payload: '+entry['path'])
|
|
data = decoder.unconsumed_tail
|
|
if data == previous and not output:
|
|
raise PakError('Compressed decoder made no progress')
|
|
if decoder is not None:
|
|
while not decoder.eof:
|
|
output = decoder.decompress(b'', min(BLOCK, expected-produced+1))
|
|
if not output:
|
|
break
|
|
accept(output)
|
|
if not decoder.eof or decoder.unused_data or decoder.unconsumed_tail:
|
|
raise PakError('Incomplete/extra compressed payload: '+entry['path'])
|
|
except zlib.error as error:
|
|
raise PakError('Invalid zlib stream: '+entry['path']) from error
|
|
if produced != expected:
|
|
raise PakError('Expanded size mismatch: '+entry['path'])
|
|
return hasher.hexdigest()
|
|
|
|
|
|
def inventory(path, limits=Limits()):
|
|
limits.validate()
|
|
path = Path(path)
|
|
descriptor = os.open(path, os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW)
|
|
with os.fdopen(descriptor, 'rb') as stream:
|
|
before = os.fstat(stream.fileno())
|
|
if not stat.S_ISREG(before.st_mode) or before.st_size > limits.archive_bytes:
|
|
raise PakError('Input must be a regular archive within its size limit')
|
|
header = exact(stream, 12)
|
|
if header[:4] != b'FORM' or header[8:] != b'PAC1' or struct.unpack_from('>I', header, 4)[0] != before.st_size-8:
|
|
raise PakError('Unsupported or inconsistent FORM/PAC1 header')
|
|
offset = 12
|
|
chunks = {}
|
|
for expected in (b'HEAD', b'DATA', b'FILE'):
|
|
stream.seek(offset)
|
|
tag, size = struct.unpack('>4sI', exact(stream, 8))
|
|
if tag != expected or size > before.st_size-offset-8:
|
|
raise PakError('Unsupported chunk order or out-of-bounds chunk')
|
|
chunks[tag] = (offset+8, size)
|
|
offset += 8+size
|
|
if offset != before.st_size:
|
|
raise PakError('Unknown trailing archive chunks/bytes')
|
|
start, size = chunks[b'HEAD']
|
|
if size != 28:
|
|
raise PakError('Unsupported HEAD size')
|
|
stream.seek(start)
|
|
head = exact(stream, size)
|
|
if head[:4] != b'\x03\x00\x01\x00' or head[24:] != b'\0'*4:
|
|
raise PakError('Unsupported HEAD version/flags')
|
|
start, size = chunks[b'FILE']
|
|
if size > limits.tree_bytes:
|
|
raise PakError('FILE metadata exceeds size limit')
|
|
stream.seek(start)
|
|
tree = exact(stream, size)
|
|
data_start, data_size = chunks[b'DATA']
|
|
entries, expanded, nodes = parse_tree(tree, data_start, data_start+data_size, limits)
|
|
for entry in entries:
|
|
entry['sha256'] = content_hash(stream, entry)
|
|
stream.seek(0)
|
|
archive_hash = hashlib.sha256()
|
|
for block in iter(lambda: stream.read(BLOCK), b''):
|
|
archive_hash.update(block)
|
|
after = os.fstat(stream.fileno())
|
|
if (before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns):
|
|
raise PakError('Archive changed during inspection')
|
|
return {'format': 'FORM/PAC1 observed HEAD 03000100', 'archive': str(path),
|
|
'archive_bytes': before.st_size, 'archive_sha256': archive_hash.hexdigest(),
|
|
'head_hex': head.hex(), 'entry_count': len(entries), 'tree_node_count': nodes,
|
|
'expanded_bytes': expanded, 'entries': sorted(entries, key=lambda row: row['path']),
|
|
'limits': vars(limits), 'structural_inventory_verified': True,
|
|
'platform_or_resource_semantics_verified': False, 'release_qualified': False,
|
|
'scope': 'Bounded structural/content-hash inspection only; no extraction, native load or gameplay proof'}
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('archive', type=Path)
|
|
parser.add_argument('--output', type=Path, help='New JSON report only; never overwritten')
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
if args.output and (args.output.exists() or args.output.is_symlink()):
|
|
raise PakError('Output already exists; refusing overwrite')
|
|
report = inventory(args.archive)
|
|
text = json.dumps(report, indent=2, sort_keys=True)+'\n'
|
|
if args.output:
|
|
with args.output.open('x') as output:
|
|
output.write(text)
|
|
else:
|
|
print(text, end='')
|
|
return 0
|
|
except (OSError, ValueError, struct.error) as error:
|
|
print(json.dumps({'structural_inventory_verified': False, 'error': str(error), 'release_qualified': False}))
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|