58 lines
1.4 KiB
Python
58 lines
1.4 KiB
Python
|
from nullptr.store import CHUNK_MAGIC, ChunkHeader, StoreUnpickler
|
||
|
from hexdump import hexdump
|
||
|
from io import BytesIO
|
||
|
class FakeStore:
|
||
|
def get(self, typ, sym, create=False):
|
||
|
return None
|
||
|
|
||
|
class StoreAnalyzer:
|
||
|
def __init__(self, verbose=False):
|
||
|
self.verbose = verbose
|
||
|
|
||
|
def load_obj(self, f, sz):
|
||
|
buf = BytesIO(f.read(sz))
|
||
|
p = StoreUnpickler(buf, FakeStore())
|
||
|
obj = p.load()
|
||
|
return obj
|
||
|
print(obj.symbol, type(obj).__name__)
|
||
|
|
||
|
def run(self, f):
|
||
|
lastpos = 0
|
||
|
pos = 0
|
||
|
objs = {}
|
||
|
result = True
|
||
|
f.seek(0)
|
||
|
while True:
|
||
|
lastpos = pos
|
||
|
pos = f.tell()
|
||
|
m = f.read(8)
|
||
|
if len(m) < 8:
|
||
|
break
|
||
|
if m != CHUNK_MAGIC:
|
||
|
print(f'missing magic at {pos}')
|
||
|
result = False
|
||
|
self.investigate(f, lastpos)
|
||
|
break
|
||
|
f.seek(-8, 1)
|
||
|
h = ChunkHeader.parse(f)
|
||
|
if self.verbose:
|
||
|
print(h, pos)
|
||
|
if h.in_use:
|
||
|
obj = self.load_obj(f, h.used)
|
||
|
kobj = obj.symbol, type(obj).__name__
|
||
|
if kobj in objs:
|
||
|
print(f'Double object {kobj} prev {objs[kobj]} latest {h}')
|
||
|
result = False
|
||
|
objs[kobj] = h
|
||
|
else:
|
||
|
f.seek(h.used, 1)
|
||
|
f.seek(h.size - h.used, 1)
|
||
|
return result
|
||
|
|
||
|
def investigate(self, f, lastpos):
|
||
|
print(f'dumping 1024 bytes from {lastpos}')
|
||
|
f.seek(lastpos, 0)
|
||
|
d = f.read(1024)
|
||
|
|
||
|
hexdump(d)
|
||
|
print(d.index(CHUNK_MAGIC))
|