461 lines
18 KiB
Plaintext
461 lines
18 KiB
Plaintext
from stdint import *
|
|
import zlib.zchecksum as zchecksum
|
|
import zlib.zdef as zdef
|
|
import zlib.zhuff as zhuff
|
|
import string
|
|
import memhub
|
|
import t, c
|
|
|
|
|
|
class zinflate_stream:
|
|
pool: memhub.MemManager*
|
|
reader: zdef.zbit_reader
|
|
|
|
window: BYTEPTR
|
|
window_pos: int
|
|
window_size: int
|
|
|
|
wbits: int
|
|
is_finished: int
|
|
is_initialized: int
|
|
|
|
header_parsed: int
|
|
is_gzip: int
|
|
|
|
adler: uint32
|
|
crc: uint32
|
|
|
|
output: BYTEPTR
|
|
output_len: sizet
|
|
output_cap: sizet
|
|
|
|
input_data: const BYTEPTR
|
|
input_len: sizet
|
|
input_pos: sizet
|
|
|
|
max_length: sizet
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
def output_byte(self, byte: BYTE):
|
|
if self.max_length > 0 and self.output_len >= self.max_length: return
|
|
|
|
if self.output_len >= self.output_cap:
|
|
new_cap: sizet = self.output_cap * 2
|
|
if new_cap < 256: new_cap = 256
|
|
new_buf: BYTEPTR = BYTEPTR(zdef.zdef_alloc(self.pool, new_cap))
|
|
if not new_buf: return
|
|
string.memcpy(new_buf, self.output, self.output_len)
|
|
zdef.zdef_free(self.pool, self.output)
|
|
self.output = new_buf
|
|
self.output_cap = new_cap
|
|
self.output[self.output_len] = byte
|
|
self.output_len += 1
|
|
self.window[self.window_pos & (zdef.ZDEFLATE_WINDOW_SIZE - 1)] = byte
|
|
self.window_pos += 1
|
|
|
|
|
|
def parse_header(self) -> int:
|
|
if self.header_parsed: return 0
|
|
if self.wbits < 0:
|
|
self.header_parsed = 1
|
|
self.is_gzip = 0
|
|
return 0
|
|
if self.input_pos + 2 > self.input_len: return -3
|
|
b0: BYTE = self.input_data[self.input_pos]
|
|
b1: BYTE = self.input_data[self.input_pos + 1]
|
|
if b0 == 0x1F and b1 == 0x8B:
|
|
self.is_gzip = 1
|
|
if self.input_pos + 10 > self.input_len: return -3
|
|
self.input_pos += 10
|
|
# FLG is at byte 4 of the gzip header (offset 3), i.e. self.input_pos - 7
|
|
flg: BYTE = self.input_data[self.input_pos - 7]
|
|
if flg & 0x04:
|
|
if self.input_pos + 2 > self.input_len: return -3
|
|
xlen: UINT = self.input_data[self.input_pos] | (self.input_data[self.input_pos + 1] << 8)
|
|
self.input_pos += 2 + xlen
|
|
if flg & 0x08:
|
|
while self.input_pos < self.input_len and self.input_data[self.input_pos] != 0: self.input_pos += 1
|
|
self.input_pos += 1
|
|
if flg & 0x10:
|
|
while self.input_pos < self.input_len and self.input_data[self.input_pos] != 0: self.input_pos += 1
|
|
self.input_pos += 1
|
|
if flg & 0x02:
|
|
self.input_pos += 2
|
|
if self.input_pos > self.input_len: return -3
|
|
else:
|
|
self.is_gzip = 0
|
|
if (b0 * 256 + b1) % 31 != 0: return -3
|
|
cm: int = b0 & 0x0F
|
|
if cm != 8: return -3
|
|
self.input_pos += 2
|
|
self.header_parsed = 1
|
|
return 0
|
|
|
|
|
|
def read_bits_from_input(self, nbits: int, out: UINTPTR) -> int:
|
|
val: UINT = 0
|
|
for i in range(int(nbits)):
|
|
if self.input_pos >= self.input_len: return -3
|
|
if self.input_data[self.input_pos] & (UINT(1) << (self.reader.bit_pos)):
|
|
val |= (UINT(1) << i)
|
|
self.reader.bit_pos += 1
|
|
if self.reader.bit_pos == 8:
|
|
self.reader.bit_pos = 0
|
|
self.input_pos += 1
|
|
*out = val
|
|
return 0
|
|
|
|
|
|
def read_huffman(self, dt: zhuff.zhuff_decode_tree*) -> int:
|
|
node: int = dt.root
|
|
while dt.nodes[node].symbol == -1:
|
|
bit: UINT
|
|
if self.read_bits_from_input(1, &bit) != 0: return -1
|
|
dir: int = int(bit)
|
|
if dt.nodes[node].children[dir] == -1: return -3
|
|
node = dt.nodes[node].children[dir]
|
|
return dt.nodes[node].symbol
|
|
|
|
def inflate_block_stored(self) -> int:
|
|
if self.reader.bit_pos != 0:
|
|
self.reader.bit_pos = 0
|
|
self.input_pos += 1
|
|
if self.input_pos + 4 > self.input_len: return -3
|
|
length: UINT = self.input_data[self.input_pos] | (self.input_data[self.input_pos + 1] << 8)
|
|
nlen: UINT = self.input_data[self.input_pos + 2] | (self.input_data[self.input_pos + 3] << 8)
|
|
self.input_pos += 4
|
|
if (length ^ nlen) != 0xFFFF: return -3
|
|
if self.input_pos + length > self.input_len: return -3
|
|
i: uint
|
|
for i in range(int(length)):
|
|
self.output_byte(self.input_data[self.input_pos])
|
|
self.input_pos += 1
|
|
return 0
|
|
|
|
|
|
def inflate_block_fixed(self) -> int:
|
|
lit_tree = zhuff.zhuff_tree()
|
|
dist_tree = zhuff.zhuff_tree()
|
|
lit_tree.build_fixed_lit_tree()
|
|
dist_tree.build_fixed_dist_tree()
|
|
|
|
lit_dt: zhuff.zhuff_decode_tree* = zdef.zdef_alloc(self.pool, zhuff.zhuff_decode_tree.__sizeof__())
|
|
dist_dt: zhuff.zhuff_decode_tree* = zdef.zdef_alloc(self.pool, zhuff.zhuff_decode_tree.__sizeof__())
|
|
if not lit_dt or not dist_dt:
|
|
if lit_dt: zdef.zdef_free(self.pool, lit_dt)
|
|
if dist_dt: zdef.zdef_free(self.pool, dist_dt)
|
|
return -4
|
|
string.memset(lit_dt, 0, zhuff.zhuff_decode_tree.__sizeof__())
|
|
string.memset(dist_dt, 0, zhuff.zhuff_decode_tree.__sizeof__())
|
|
lit_dt.build_decode_tree(&lit_tree)
|
|
dist_dt.build_decode_tree(&dist_tree)
|
|
|
|
result: int = 0
|
|
while True:
|
|
if self.max_length > 0 and self.output_len >= self.max_length: break
|
|
|
|
sym: int = self.read_huffman(lit_dt)
|
|
if sym < 0:
|
|
result = -3
|
|
break
|
|
if sym == 256: break
|
|
if sym < 256:
|
|
self.output_byte(BYTE(sym))
|
|
else:
|
|
len_idx: int = sym - 257
|
|
if len_idx < 0 or len_idx >= 29:
|
|
result = -3
|
|
break
|
|
length: int = zdef.zdeflate_len_base[len_idx]
|
|
extra: int = zdef.zdeflate_len_extra_bits[len_idx]
|
|
if extra > 0:
|
|
extra_val: UINT
|
|
if self.read_bits_from_input(extra, &extra_val) != 0:
|
|
result = -3
|
|
break
|
|
length += extra_val
|
|
|
|
dist_sym: int = self.read_huffman(dist_dt)
|
|
if dist_sym < 0 or dist_sym >= 30:
|
|
result = -3
|
|
break
|
|
dist: int = zdef.zdeflate_dist_base[dist_sym]
|
|
extra = zdef.zdeflate_dist_extra_bits[dist_sym]
|
|
if extra > 0:
|
|
extra_val: UINT
|
|
if self.read_bits_from_input(extra, &extra_val) != 0:
|
|
result = -3
|
|
break
|
|
dist += extra_val
|
|
|
|
for i in range(int(length)):
|
|
if self.max_length > 0 and self.output_len >= self.max_length: break
|
|
src_pos: int = (self.window_pos - dist) & (zdef.ZDEFLATE_WINDOW_SIZE - 1)
|
|
byte: BYTE = self.window[src_pos]
|
|
self.output_byte(byte)
|
|
|
|
zdef.zdef_free(self.pool, lit_dt)
|
|
zdef.zdef_free(self.pool, dist_dt)
|
|
return result
|
|
|
|
|
|
def inflate_block_dynamic(self) -> int:
|
|
hlit_val: UINT
|
|
hdist_val: UINT
|
|
hclen_val: UINT
|
|
if self.read_bits_from_input(5, &hlit_val) != 0: return -3
|
|
if self.read_bits_from_input(5, &hdist_val) != 0: return -3
|
|
if self.read_bits_from_input(4, &hclen_val) != 0: return -3
|
|
|
|
hlit: int = int(hlit_val) + 257
|
|
hdist: int = int(hdist_val) + 1
|
|
hclen: int = int(hclen_val) + 4
|
|
|
|
cl_lengths: t.CArray[int, 19]
|
|
string.memset(cl_lengths, 0, cl_lengths.__sizeof__())
|
|
for i in range(int(hclen)):
|
|
len_val: UINT
|
|
if self.read_bits_from_input(3, &len_val) != 0: return -3
|
|
cl_lengths[zdef.zdeflate_codelen_order[i]] = int(len_val)
|
|
|
|
cl_tree = zhuff.zhuff_tree()
|
|
cl_tree.build_tree_from_lengths(cl_lengths, 19, 7)
|
|
|
|
cl_dt = zhuff.zhuff_decode_tree()
|
|
cl_dt.build_decode_tree(&cl_tree)
|
|
|
|
total: int = hlit + hdist
|
|
all_lengths: int* = zdef.zdef_alloc(self.pool, total * int.__sizeof__())
|
|
if not all_lengths: return -4
|
|
|
|
idx: int = 0
|
|
while idx < total:
|
|
sym: int = self.read_huffman(&cl_dt)
|
|
if sym < 0 or sym > 18:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
if sym < 16:
|
|
all_lengths[idx] = sym
|
|
idx += 1
|
|
elif sym == 16:
|
|
rep: UINT
|
|
if self.read_bits_from_input(2, &rep) != 0:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
rep += 3
|
|
if idx == 0 or idx + rep > total:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
prev: int = all_lengths[idx - 1]
|
|
i: uint
|
|
for i in range(int(rep)):
|
|
all_lengths[idx] = prev
|
|
idx += 1
|
|
elif sym == 17:
|
|
rep: uint
|
|
if self.read_bits_from_input(3, &rep) != 0:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
rep += 3
|
|
if idx + rep > total:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
i: uint
|
|
for i in range(int(rep)):
|
|
all_lengths[idx] = 0
|
|
idx += 1
|
|
else:
|
|
rep: uint
|
|
if self.read_bits_from_input(7, &rep) != 0:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
rep += 11
|
|
if idx + rep > total:
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
return -3
|
|
i: uint
|
|
for i in range(int(rep)):
|
|
all_lengths[idx] = 0
|
|
idx += 1
|
|
|
|
lit_tree = zhuff.zhuff_tree()
|
|
dist_tree = zhuff.zhuff_tree()
|
|
|
|
lit_lengths: t.CArray[int, 288]
|
|
string.memset(lit_lengths, 0, lit_lengths.__sizeof__())
|
|
for i in range(int(hlit)):
|
|
lit_lengths[i] = all_lengths[i]
|
|
lit_tree.build_tree_from_lengths(lit_lengths, hlit, zdef.ZDEFLATE_MAX_BITS)
|
|
|
|
dist_lengths: t.CArray[int, 32]
|
|
string.memset(dist_lengths, 0, dist_lengths.__sizeof__())
|
|
for i in range(int(hdist)):
|
|
dist_lengths[i] = all_lengths[hlit + i]
|
|
dist_tree.build_tree_from_lengths(dist_lengths, hdist, zdef.ZDEFLATE_MAX_BITS)
|
|
|
|
zdef.zdef_free(self.pool, all_lengths)
|
|
|
|
lit_dt = zhuff.zhuff_decode_tree()
|
|
dist_dt = zhuff.zhuff_decode_tree()
|
|
lit_dt.build_decode_tree(&lit_tree)
|
|
dist_dt.build_decode_tree(&dist_tree)
|
|
|
|
while True:
|
|
if self.max_length > 0 and self.output_len >= self.max_length: return 0
|
|
sym: int = self.read_huffman(&lit_dt)
|
|
if sym < 0: return -3
|
|
if sym == 256: return 0
|
|
if sym < 256:
|
|
self.output_byte(BYTE(sym))
|
|
else:
|
|
len_idx: int = sym - 257
|
|
if len_idx < 0 or len_idx >= 29: return -3
|
|
length: int = zdef.zdeflate_len_base[len_idx]
|
|
extra: int = zdef.zdeflate_len_extra_bits[len_idx]
|
|
if extra > 0:
|
|
extra_val: UINT
|
|
if self.read_bits_from_input(extra, &extra_val) != 0: return -3
|
|
length += extra_val
|
|
dist_sym: int = self.read_huffman(&dist_dt)
|
|
if dist_sym < 0 or dist_sym >= 30: return -3
|
|
dist: int = zdef.zdeflate_dist_base[dist_sym]
|
|
extra = zdef.zdeflate_dist_extra_bits[dist_sym]
|
|
if extra > 0:
|
|
extra_val: UINT
|
|
if self.read_bits_from_input(extra, &extra_val) != 0: return -3
|
|
dist += extra_val
|
|
for i in range(int(length)):
|
|
if self.max_length > 0 and self.output_len >= self.max_length: return 0
|
|
src_pos: int = (self.window_pos - dist) & (zdef.ZDEFLATE_WINDOW_SIZE - 1)
|
|
byte: BYTE = self.window[src_pos]
|
|
self.output_byte(byte)
|
|
|
|
def inflate_stream(self) -> int:
|
|
err: int = self.parse_header()
|
|
if err != 0: return err
|
|
while not self.is_finished:
|
|
bfinal_val: UINT
|
|
btype_val: UINT
|
|
if self.read_bits_from_input(1, &bfinal_val) != 0: return -3
|
|
if self.read_bits_from_input(2, &btype_val) != 0: return -3
|
|
bfinal: int = int(bfinal_val)
|
|
btype: int = int(btype_val)
|
|
if btype == 0:
|
|
err = self.inflate_block_stored()
|
|
elif btype == 1:
|
|
err = self.inflate_block_fixed()
|
|
elif btype == 2:
|
|
err = self.inflate_block_dynamic()
|
|
else:
|
|
return -3
|
|
if err != 0: return err
|
|
if bfinal: self.is_finished = 1
|
|
if self.max_length > 0 and self.output_len >= self.max_length: return 0
|
|
if self.is_gzip:
|
|
if self.reader.bit_pos != 0:
|
|
self.reader.bit_pos = 0
|
|
self.input_pos += 1
|
|
if self.input_pos + 8 <= self.input_len:
|
|
self.crc = ((uint32(self.input_data[self.input_pos + 0]) << 0) |
|
|
(uint32(self.input_data[self.input_pos + 1]) << 8) |
|
|
(uint32(self.input_data[self.input_pos + 2]) << 16) |
|
|
(uint32(self.input_data[self.input_pos + 3]) << 24))
|
|
self.input_pos += 8
|
|
elif self.wbits >= 0:
|
|
if self.reader.bit_pos != 0:
|
|
self.reader.bit_pos = 0
|
|
self.input_pos += 1
|
|
if self.input_pos + 4 <= self.input_len:
|
|
self.adler = ((uint32(self.input_data[self.input_pos + 0]) << 24) |
|
|
(uint32(self.input_data[self.input_pos + 1]) << 16) |
|
|
(uint32(self.input_data[self.input_pos + 2]) << 8) |
|
|
(uint32(self.input_data[self.input_pos + 3]) << 0))
|
|
self.input_pos += 4
|
|
return 0
|
|
|
|
# out is a "pointer to pointer" (i8**). Do NOT write `t.CUInt8T | t.CPtr[t.CPtr]`: with an embedded `t.CPtr[...]` subscript the definition side accumulates PtrCount to i8**, but the stub/declare side (StubGen/VpiReader) only recognizes a segment exactly equal to `t.CPtr`/`CPtr` as the CPtr marker, collapsing it to t.CUInt8T = i8 -> cross-module ABI mismatch (caller passes 1 byte, callee writes an 8-byte pointer -> ASan heap-buffer-overflow).
|
|
# Use "pointer alias | t.CPtr" instead: both sides append one pointer level and resolve to i8**.
|
|
# Same-shaped precedents: includes/w32/win32sync.vp:65 `Context: VOIDPTR | t.CPtr`,
|
|
# includes/posix.vp:199 `argv: CHARPTR | t.CPtr`.
|
|
def decompress(self, data: UINT8PTR, length: sizet,
|
|
max_length: sizet, out: BYTEPTR*, out_len: sizet*) -> int:
|
|
if not self.is_initialized: return -2
|
|
self.input_data = data
|
|
self.input_len = length
|
|
self.input_pos = 0
|
|
self.reader.bit_pos = 0
|
|
self.output_len = 0
|
|
self.max_length = max_length
|
|
err: int = self.inflate_stream()
|
|
if err != 0: return err
|
|
*out_len = self.output_len
|
|
if self.output_len == 0:
|
|
*out = UINT8PTR(zdef.zdef_alloc(self.pool, 1))
|
|
else:
|
|
dst: UINT8PTR = UINT8PTR(zdef.zdef_alloc(self.pool, self.output_len))
|
|
*out = dst
|
|
if dst:
|
|
string.memcpy(dst, self.output, self.output_len)
|
|
return 0
|
|
|
|
def set_dictionary(self, dict: UINT8PTR, length: sizet) -> int:
|
|
if not dict: return -2
|
|
use: sizet = length
|
|
if use > zdef.ZDEFLATE_WINDOW_SIZE:
|
|
dict += length - zdef.ZDEFLATE_WINDOW_SIZE
|
|
use = zdef.ZDEFLATE_WINDOW_SIZE
|
|
string.memcpy(self.window, dict, use)
|
|
self.window_pos = int(use)
|
|
return 0
|
|
|
|
def copy(self) -> zinflate_stream*:
|
|
z: zinflate_stream* = zdef.zdef_alloc(self.pool, zinflate_stream.__sizeof__())
|
|
if not z: return None
|
|
string.memcpy(z, self, zinflate_stream.__sizeof__())
|
|
z.window = BYTEPTR(zdef.zdef_alloc(self.pool, zdef.ZDEFLATE_WINDOW_SIZE))
|
|
z.output = BYTEPTR(zdef.zdef_alloc(self.pool, self.output_cap))
|
|
if not z.window or not z.output:
|
|
zinflate_stream.destroy(z)
|
|
return None
|
|
string.memcpy(z.window, self.window, zdef.ZDEFLATE_WINDOW_SIZE)
|
|
string.memcpy(z.output, self.output, self.output_cap)
|
|
return z
|
|
|
|
def destroy(self):
|
|
if not self: return
|
|
if self.window: zdef.zdef_free(self.pool, self.window)
|
|
if self.output: zdef.zdef_free(self.pool, self.output)
|
|
zdef.zdef_free(self.pool, self)
|
|
|
|
|
|
def zinflate_one_shot(pool: memhub.MemManager*, data: UINT8PTR, length: sizet,
|
|
wbits: int, bufsize: sizet, out_len: sizet*) -> uint8*:
|
|
s: zinflate_stream* = zinflate_create(pool, wbits)
|
|
if not s: return None
|
|
out: uint8* = None
|
|
err: int = int(s.decompress(data, length, 0, &out, out_len))
|
|
s.destroy()
|
|
if err != 0: return None
|
|
return out
|
|
|
|
|
|
def zinflate_create(pool: memhub.MemManager*, wbits: int) -> zinflate_stream*:
|
|
s: zinflate_stream* = zdef.zdef_alloc(pool, zinflate_stream.__sizeof__())
|
|
if not s: return None
|
|
string.memset(s, 0, zinflate_stream.__sizeof__())
|
|
s.pool = pool
|
|
s.wbits = wbits
|
|
s.is_initialized = 1
|
|
s.adler = 1
|
|
s.window = BYTEPTR(zdef.zdef_alloc(pool, zdef.ZDEFLATE_WINDOW_SIZE))
|
|
s.output = BYTEPTR(zdef.zdef_alloc(pool, 256))
|
|
s.output_cap = 256
|
|
s.output_len = 0
|
|
s.max_length = 0
|
|
if not s.window or not s.output:
|
|
s.destroy()
|
|
return None
|
|
return s
|