TermiNexus
794488b0f6
t.CUInt8T | t.CPtr[t.CPtr] 在定义侧按 PtrCount 累加得 i8**，但 stub/declare 侧(VpiReader 只把整段等于 t.CPtr/CPtr 的段认作 CPtr 标记)会坍缩为首段 i8，导致跨模块调用只传 1 字节、被调用方按 8 字节指针写入。ZlibTest 的 test_decompressobj 即由此触发 ASan heap-buffer-overflow：写入地址 = 压缩缓冲起址 -16(第 5 参栈槽残留的 compressed 指针被清零低字节) 改用 指针别名 | t.CPtr(两侧同为 i8**，与 win32sync.vp:65、posix.vp:199 同形)，覆盖 zinflate_stream.decompress、zdeflate_stream.compress/flush、pyzlib.append_bytes 共 4 处；不改任何运行期逻辑
702 lines
26 KiB
Plaintext
702 lines
26 KiB
Plaintext
# from string import memset, memcpy
|
||
# from stdio import malloc, free, calloc, realloc
|
||
from stdint import *
|
||
import zlib.zdeflate as zdeflate
|
||
import zlib.zinflate as zinflate
|
||
import zlib.zchecksum as zchecksum
|
||
import zlib.zdef as zdef
|
||
import zlib.zhuff as zhuff
|
||
import stdlib
|
||
import string
|
||
import memhub
|
||
import t, c
|
||
|
||
|
||
# ============================================================
|
||
# Constants - matching Python zlib module names and values
|
||
# ============================================================
|
||
|
||
# Compression levels
|
||
Z_NO_COMPRESSION: t.CDefine = 0
|
||
Z_BEST_SPEED: t.CDefine = 1
|
||
Z_BEST_COMPRESSION: t.CDefine = 9
|
||
Z_DEFAULT_COMPRESSION: t.CDefine = (-1)
|
||
|
||
# Compression methods
|
||
DEFLATED: t.CDefine = 8
|
||
|
||
# Flush modes
|
||
Z_NO_FLUSH: t.CDefine = 0
|
||
Z_PARTIAL_FLUSH: t.CDefine = 1
|
||
Z_SYNC_FLUSH: t.CDefine = 2
|
||
Z_FULL_FLUSH: t.CDefine = 3
|
||
Z_FINISH: t.CDefine = 4
|
||
Z_BLOCK: t.CDefine = 5
|
||
Z_TREES: t.CDefine = 6
|
||
|
||
# Strategies
|
||
Z_DEFAULT_STRATEGY: t.CDefine = 0
|
||
Z_FILTERED: t.CDefine = 1
|
||
Z_HUFFMAN_ONLY: t.CDefine = 2
|
||
Z_RLE: t.CDefine = 3
|
||
Z_FIXED: t.CDefine = 4
|
||
|
||
# Return codes
|
||
Z_OK: t.CDefine = 0
|
||
Z_STREAM_END: t.CDefine = 1
|
||
Z_NEED_DICT: t.CDefine = 2
|
||
Z_ERRNO: t.CDefine = (-1)
|
||
Z_STREAM_ERROR: t.CDefine = (-2)
|
||
Z_DATA_ERROR: t.CDefine = (-3)
|
||
Z_MEM_ERROR: t.CDefine = (-4)
|
||
Z_BUF_ERROR: t.CDefine = (-5)
|
||
Z_VERSION_ERROR: t.CDefine = (-6)
|
||
|
||
# Window / Buffer constants
|
||
MAX_WBITS: t.CDefine = 15
|
||
DEF_BUF_SIZE: t.CDefine = 16384
|
||
DEF_MEM_LEVEL: t.CDefine = 8
|
||
|
||
# Version
|
||
ZLIB_VERSION: t.CDefine = "1.3.2"
|
||
|
||
pyzlib_error_msg: t.CArray[t.CChar, 512] = ""
|
||
pyzlib_error_code_val: t.CInt = 0
|
||
|
||
# ============================================================
|
||
# Structures
|
||
# ============================================================
|
||
@t.Object
|
||
class Compress:
|
||
pool: memhub.MemManager | t.CPtr
|
||
stream: VOIDPTR
|
||
is_initialized: t.CInt
|
||
is_finished: t.CInt
|
||
level: t.CInt
|
||
method: t.CInt
|
||
wbits: t.CInt
|
||
memLevel: t.CInt
|
||
strategy: t.CInt
|
||
zdict: BYTEPTR
|
||
zdict_len: t.CSizeT
|
||
last_pos: t.CSizeT
|
||
input_buf: BYTEPTR
|
||
input_buf_len: t.CSizeT
|
||
input_buf_cap: t.CSizeT
|
||
header_written: t.CInt
|
||
|
||
def compress(self,
|
||
data: BYTEPTR, data_len: t.CSizeT,
|
||
out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not self.is_initialized:
|
||
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
||
return None
|
||
if self.is_finished:
|
||
set_error(Z_STREAM_ERROR, "compress object already finished")
|
||
return None
|
||
if not out_len:
|
||
set_error(Z_STREAM_ERROR, "out_len is None")
|
||
return None
|
||
if data_len == 0:
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p, 0, 1)
|
||
return p
|
||
while self.input_buf_len + data_len > self.input_buf_cap:
|
||
self.input_buf_cap *= 2
|
||
if self.pool == None:
|
||
new_buf: BYTEPTR = BYTEPTR(stdlib.realloc(self.input_buf, self.input_buf_cap))
|
||
else:
|
||
new_buf: BYTEPTR = BYTEPTR(self.pool.alloc(self.input_buf_cap))
|
||
if new_buf:
|
||
string.memcpy(new_buf, self.input_buf, self.input_buf_len)
|
||
self.pool.free(self.input_buf)
|
||
if not new_buf:
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p2: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p2, 0, 1)
|
||
return p2
|
||
self.input_buf = new_buf
|
||
string.memcpy(self.input_buf + self.input_buf_len, data, data_len)
|
||
self.input_buf_len += data_len
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p3: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p3, 0, 1)
|
||
return p3
|
||
|
||
def flush(self, mode: t.CInt, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not self.is_initialized:
|
||
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
||
return None
|
||
if not out_len:
|
||
set_error(Z_STREAM_ERROR, "out_len is None")
|
||
return None
|
||
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
|
||
if mode == Z_FINISH:
|
||
if not self.header_written:
|
||
self.header_written = 1
|
||
input_data: BYTEPTR = self.input_buf
|
||
input_len: t.CSizeT = self.input_buf_len
|
||
if input_len == 0:
|
||
s.writer.write_bits(1, 1)
|
||
s.writer.write_bits(1, 2)
|
||
lit_tree = zhuff.zhuff_tree()
|
||
lit_tree.build_fixed_lit_tree()
|
||
lit_tree.encode_symbol(256, c.Addr(s.writer))
|
||
else:
|
||
s.adler = zchecksum.zchecksum_adler32(input_data, input_len, 1)
|
||
s.compress_block(input_data, input_len, 1)
|
||
s.is_finished = 1
|
||
if s.wbits >= 0:
|
||
s.writer.align()
|
||
adler: t.CUInt32T = s.adler
|
||
s.writer.buf[s.writer.byte_pos + 0] = (adler >> 24) & 0xFF
|
||
s.writer.buf[s.writer.byte_pos + 1] = (adler >> 16) & 0xFF
|
||
s.writer.buf[s.writer.byte_pos + 2] = (adler >> 8) & 0xFF
|
||
s.writer.buf[s.writer.byte_pos + 3] = (adler >> 0) & 0xFF
|
||
s.writer.byte_pos += 4
|
||
self.is_finished = 1
|
||
total: t.CSizeT = s.writer.total()
|
||
if self.pool == None:
|
||
result: BYTEPTR = BYTEPTR(stdlib.malloc(total))
|
||
else:
|
||
result: BYTEPTR = BYTEPTR(self.pool.alloc(total))
|
||
if result: string.memcpy(result, s.writer.buf, total)
|
||
c.Set(c.Deref(out_len), total)
|
||
if self.pool == None:
|
||
stdlib.free(self.input_buf)
|
||
else:
|
||
self.pool.free(self.input_buf)
|
||
self.input_buf = None
|
||
self.input_buf_len = 0
|
||
self.input_buf_cap = 0
|
||
return result
|
||
elif mode == Z_SYNC_FLUSH or mode == Z_FULL_FLUSH:
|
||
input_data: BYTEPTR = self.input_buf
|
||
input_len: t.CSizeT = self.input_buf_len
|
||
if input_len > 0:
|
||
s.adler = zchecksum.zchecksum_adler32(input_data, input_len, s.adler)
|
||
s.compress_block(input_data, input_len, 0)
|
||
self.input_buf_len = 0
|
||
s.writer.write_bits(0, 1)
|
||
s.writer.write_bits(0, 2)
|
||
s.writer.align()
|
||
prev_pos: t.CSizeT = self.last_pos
|
||
total: t.CSizeT = s.writer.total()
|
||
delta: t.CSizeT = total - prev_pos
|
||
self.last_pos = s.writer.byte_pos
|
||
if delta == 0:
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p4: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p4, 0, 1)
|
||
return p4
|
||
|
||
if self.pool == None:
|
||
result: BYTEPTR = BYTEPTR(stdlib.malloc(delta))
|
||
else:
|
||
result: BYTEPTR = BYTEPTR(self.pool.alloc(delta))
|
||
if result: string.memcpy(result, s.writer.buf + prev_pos, delta)
|
||
c.Set(c.Deref(out_len), delta)
|
||
return result
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p5: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p5, 0, 1)
|
||
return p5
|
||
|
||
def copy(self) -> Compress | t.CPtr:
|
||
if not self.is_initialized:
|
||
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
||
return None
|
||
if self.pool == None:
|
||
copy_obj: Compress | t.CPtr = stdlib.calloc(1, Compress.__sizeof__())
|
||
else:
|
||
copy_obj: Compress | t.CPtr = self.pool.alloc(Compress.__sizeof__())
|
||
string.memset(copy_obj, 0, Compress.__sizeof__())
|
||
if not copy_obj:
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
copy_obj.pool = self.pool
|
||
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
|
||
copy_obj.stream = s.copy()
|
||
if not copy_obj.stream:
|
||
if self.pool == None:
|
||
stdlib.free(copy_obj)
|
||
else:
|
||
self.pool.free(copy_obj)
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
copy_obj.is_initialized = 1
|
||
copy_obj.is_finished = self.is_finished
|
||
copy_obj.level = self.level
|
||
copy_obj.method = self.method
|
||
copy_obj.wbits = self.wbits
|
||
copy_obj.memLevel = self.memLevel
|
||
copy_obj.strategy = self.strategy
|
||
copy_obj.last_pos = self.last_pos
|
||
copy_obj.header_written = self.header_written
|
||
if self.input_buf and self.input_buf_len > 0:
|
||
if self.pool == None:
|
||
copy_obj.input_buf = BYTEPTR(stdlib.calloc(1, self.input_buf_cap))
|
||
else:
|
||
copy_obj.input_buf = BYTEPTR(self.pool.alloc(self.input_buf_cap))
|
||
string.memset(copy_obj.input_buf, 0, self.input_buf_cap)
|
||
if copy_obj.input_buf:
|
||
string.memcpy(copy_obj.input_buf, self.input_buf, self.input_buf_len)
|
||
copy_obj.input_buf_cap = self.input_buf_cap
|
||
copy_obj.input_buf_len = self.input_buf_len
|
||
else:
|
||
if self.pool == None:
|
||
copy_obj.input_buf = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
copy_obj.input_buf = BYTEPTR(self.pool.alloc(256))
|
||
string.memset(copy_obj.input_buf, 0, 256)
|
||
copy_obj.input_buf_cap = 256
|
||
copy_obj.input_buf_len = 0
|
||
if self.zdict and self.zdict_len > 0:
|
||
copy_obj.zdict = clone_bytes(self.pool, self.zdict, self.zdict_len)
|
||
copy_obj.zdict_len = self.zdict_len
|
||
return copy_obj
|
||
|
||
def delete(self):
|
||
if self.is_initialized and self.stream:
|
||
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
|
||
s.destroy()
|
||
if self.pool == None:
|
||
stdlib.free(self.zdict)
|
||
stdlib.free(self.input_buf)
|
||
stdlib.free(self)
|
||
else:
|
||
self.pool.free(self.zdict)
|
||
self.pool.free(self.input_buf)
|
||
self.pool.free(self)
|
||
|
||
def __del__(self):
|
||
self.delete()
|
||
|
||
|
||
@t.Object
|
||
class Decompress:
|
||
pool: memhub.MemManager | t.CPtr
|
||
stream: VOIDPTR
|
||
is_initialized: t.CInt
|
||
eof: t.CInt
|
||
wbits: t.CInt
|
||
zdict: BYTEPTR
|
||
zdict_len: t.CSizeT
|
||
_unused_data: BYTEPTR
|
||
_unused_data_len: t.CSizeT
|
||
_unused_data_cap: t.CSizeT
|
||
_unconsumed_tail: BYTEPTR
|
||
_unconsumed_tail_len: t.CSizeT
|
||
_unconsumed_tail_cap: t.CSizeT
|
||
input_buf: BYTEPTR
|
||
input_buf_len: t.CSizeT
|
||
input_buf_cap: t.CSizeT
|
||
|
||
def decompress(self, data: BYTEPTR, data_len: t.CSizeT,
|
||
max_length: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not self.is_initialized:
|
||
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
||
return None
|
||
if not out_len:
|
||
set_error(Z_STREAM_ERROR, "out_len is None")
|
||
return None
|
||
if self.eof:
|
||
if data and data_len > 0:
|
||
append_bytes(self.pool, c.Addr(self._unused_data), c.Addr(self._unused_data_len),
|
||
c.Addr(self._unused_data_cap), data, data_len)
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p6: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p6, 0, 1)
|
||
return p6
|
||
self._unconsumed_tail_len = 0
|
||
if data and data_len > 0:
|
||
append_bytes(self.pool, c.Addr(self.input_buf), c.Addr(self.input_buf_len),
|
||
c.Addr(self.input_buf_cap), data, data_len)
|
||
if self.input_buf_len == 0:
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p7: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p7, 0, 1)
|
||
return p7
|
||
s: zinflate.zinflate_stream | t.CPtr = zinflate.zinflate_create(self.pool, self.wbits)
|
||
if not s:
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
if self.zdict and self.zdict_len > 0:
|
||
s.set_dictionary(self.zdict, self.zdict_len)
|
||
out: t.CUInt8T | t.CPtr = None
|
||
err: t.CInt = s.decompress(self.input_buf, self.input_buf_len,
|
||
max_length, c.Addr(out), out_len)
|
||
if err != 0:
|
||
s.destroy()
|
||
set_error(Z_DATA_ERROR, "decompression failed")
|
||
return None
|
||
if s.is_finished:
|
||
self.eof = 1
|
||
if s.input_pos < s.input_len:
|
||
append_bytes(self.pool, c.Addr(self._unused_data), c.Addr(self._unused_data_len),
|
||
c.Addr(self._unused_data_cap),
|
||
s.input_data + s.input_pos,
|
||
s.input_len - s.input_pos)
|
||
self.input_buf_len = 0
|
||
else:
|
||
consumed: t.CSizeT = s.input_pos
|
||
if consumed < self.input_buf_len:
|
||
remaining: t.CSizeT = self.input_buf_len - consumed
|
||
string.memmove(self.input_buf, self.input_buf + consumed, remaining)
|
||
self.input_buf_len = remaining
|
||
else:
|
||
self.input_buf_len = 0
|
||
s.destroy()
|
||
return out
|
||
|
||
def flush(self, length: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not self.is_initialized:
|
||
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
||
return None
|
||
if not out_len:
|
||
set_error(Z_STREAM_ERROR, "out_len is None")
|
||
return None
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p8: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p8, 0, 1)
|
||
return p8
|
||
|
||
def copy(self) -> Decompress | t.CPtr:
|
||
if not self.is_initialized:
|
||
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
||
return None
|
||
if self.pool == None:
|
||
copy_obj: Decompress | t.CPtr = stdlib.calloc(1, Decompress.__sizeof__())
|
||
else:
|
||
copy_obj: Decompress | t.CPtr = self.pool.alloc(Decompress.__sizeof__())
|
||
string.memset(copy_obj, 0, Decompress.__sizeof__())
|
||
if not copy_obj:
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
copy_obj.pool = self.pool
|
||
copy_obj.is_initialized = 1
|
||
copy_obj.eof = self.eof
|
||
copy_obj.wbits = self.wbits
|
||
if self.zdict and self.zdict_len > 0:
|
||
copy_obj.zdict = clone_bytes(self.pool, self.zdict, self.zdict_len)
|
||
copy_obj.zdict_len = self.zdict_len
|
||
if self._unused_data and self._unused_data_len > 0:
|
||
copy_obj._unused_data = clone_bytes(self.pool, self._unused_data, self._unused_data_len)
|
||
copy_obj._unused_data_len = self._unused_data_len
|
||
copy_obj._unused_data_cap = self._unused_data_len
|
||
if self._unconsumed_tail and self._unconsumed_tail_len > 0:
|
||
copy_obj._unconsumed_tail = clone_bytes(self.pool, self._unconsumed_tail, self._unconsumed_tail_len)
|
||
copy_obj._unconsumed_tail_len = self._unconsumed_tail_len
|
||
copy_obj._unconsumed_tail_cap = self._unconsumed_tail_len
|
||
if self.input_buf and self.input_buf_len > 0:
|
||
copy_obj.input_buf = clone_bytes(self.pool, self.input_buf, self.input_buf_len)
|
||
copy_obj.input_buf_len = self.input_buf_len
|
||
copy_obj.input_buf_cap = self.input_buf_cap
|
||
else:
|
||
if self.pool == None:
|
||
copy_obj.input_buf = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
copy_obj.input_buf = BYTEPTR(self.pool.alloc(256))
|
||
string.memset(copy_obj.input_buf, 0, 256)
|
||
copy_obj.input_buf_cap = 256
|
||
copy_obj.input_buf_len = 0
|
||
return copy_obj
|
||
|
||
def delete(self):
|
||
if self.pool == None:
|
||
stdlib.free(self.zdict)
|
||
stdlib.free(self._unused_data)
|
||
stdlib.free(self._unconsumed_tail)
|
||
stdlib.free(self.input_buf)
|
||
stdlib.free(self)
|
||
else:
|
||
self.pool.free(self.zdict)
|
||
self.pool.free(self._unused_data)
|
||
self.pool.free(self._unconsumed_tail)
|
||
self.pool.free(self.input_buf)
|
||
self.pool.free(self)
|
||
|
||
def unused_data(self, length: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
#if not self:
|
||
# if length:
|
||
# c.Set(c.Deref(length), 0)
|
||
# return None
|
||
if length:
|
||
c.Set(c.Deref(length), self._unused_data_len)
|
||
return self._unused_data
|
||
|
||
def unconsumed_tail(self, length: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
#if not self:
|
||
# if length:
|
||
# c.Set(c.Deref(length), 0)
|
||
# return None
|
||
if length:
|
||
c.Set(c.Deref(length), self._unconsumed_tail_len)
|
||
return self._unconsumed_tail
|
||
|
||
# def eof(self) -> t.CInt:
|
||
# # if not obj: return 0
|
||
# return self.eof
|
||
|
||
|
||
|
||
def set_error(code: int, msg: str):
|
||
global pyzlib_error_code_val, pyzlib_error_msg
|
||
pyzlib_error_code_val = code
|
||
if msg:
|
||
string.strncpy(pyzlib_error_msg, msg, pyzlib_error_msg.__sizeof__() - 1)
|
||
pyzlib_error_msg[pyzlib_error_msg.__sizeof__() - 1] = '\0'
|
||
else:
|
||
pyzlib_error_msg[0] = '\0'
|
||
|
||
def clone_bytes(pool: memhub.MemManager | t.CPtr, src: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
|
||
if not src or length == 0: return None
|
||
if pool == None:
|
||
dst: BYTEPTR = BYTEPTR(stdlib.malloc(length))
|
||
else:
|
||
dst: BYTEPTR = BYTEPTR(pool.alloc(length))
|
||
if dst: string.memcpy(dst, src, length)
|
||
return dst
|
||
|
||
# buf 是"指针的指针"（i8**）：不要写成 `BYTE | t.CPtr[t.CPtr]`——联合里嵌
|
||
# `t.CPtr[...]` 下标时声明侧（StubGen/VpiReader）不认作 CPtr 标记，会坍缩成
|
||
# 首段 BYTE = i8，与定义侧 i8** 不一致（跨模块调用即 ABI 错位，见
|
||
# includes/zlib/zinflate.vp decompress 处说明）。改用"指针别名 | t.CPtr"。
|
||
def append_bytes(pool: memhub.MemManager | t.CPtr, buf: BYTEPTR | t.CPtr, length: t.CSizeT | t.CPtr, cap: t.CSizeT | t.CPtr,
|
||
data: BYTEPTR, data_len: t.CSizeT) -> t.CInt:
|
||
if not data or data_len == 0: return 0
|
||
needed: t.CSizeT = c.Deref(length) + data_len
|
||
if needed > c.Deref(cap):
|
||
new_cap: t.CSizeT = c.Deref(cap) * 2
|
||
if new_cap < needed: new_cap = needed
|
||
if pool == None:
|
||
new_buf: BYTEPTR = BYTEPTR(stdlib.realloc(c.Deref(buf), new_cap))
|
||
else:
|
||
new_buf: BYTEPTR = BYTEPTR(pool.alloc(new_cap))
|
||
if new_buf:
|
||
string.memcpy(new_buf, c.Deref(buf), c.Deref(length))
|
||
pool.free(c.Deref(buf))
|
||
if not new_buf: return -1
|
||
c.Set(c.Deref(buf), new_buf)
|
||
c.Set(c.Deref(cap), new_cap)
|
||
string.memcpy(c.Deref(buf) + c.Deref(length), data, data_len)
|
||
c.Set(c.Deref(length), c.Deref(length) + data_len)
|
||
return 0
|
||
|
||
|
||
def shrink_to_fit(pool: memhub.MemManager | t.CPtr, buf: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
|
||
if length == 0:
|
||
if pool == None:
|
||
stdlib.free(buf)
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
pool.free(buf)
|
||
p: BYTEPTR = BYTEPTR(pool.alloc(1))
|
||
string.memset(p, 0, 1)
|
||
return p
|
||
if pool == None:
|
||
result: BYTEPTR = BYTEPTR(stdlib.realloc(buf, length))
|
||
return result if result else buf
|
||
else:
|
||
result: BYTEPTR = BYTEPTR(pool.alloc(length))
|
||
if result:
|
||
string.memcpy(result, buf, length)
|
||
pool.free(buf)
|
||
return result
|
||
return buf
|
||
|
||
|
||
def runtime_version() -> str:
|
||
return ZLIB_VERSION
|
||
|
||
def get_error() -> str:
|
||
return pyzlib_error_msg
|
||
|
||
def get_error_code() -> t.CInt:
|
||
return pyzlib_error_code_val
|
||
|
||
def clear_error():
|
||
global pyzlib_error_msg, pyzlib_error_code_val
|
||
pyzlib_error_msg[0] = '\0'
|
||
pyzlib_error_code_val = 0
|
||
|
||
def compress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT,
|
||
level: t.CInt, wbits: t.CInt, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not data and data_len > 0:
|
||
set_error(Z_STREAM_ERROR, "data is None but data_len > 0")
|
||
return None
|
||
if not out_len:
|
||
set_error(Z_STREAM_ERROR, "out_len is None")
|
||
return None
|
||
actual_wbits: t.CInt = wbits
|
||
if wbits > MAX_WBITS:
|
||
actual_wbits = -MAX_WBITS
|
||
raw_result: UINT8PTR = zdeflate.zdeflate_one_shot(pool, data, data_len, level, actual_wbits, out_len)
|
||
if not raw_result:
|
||
set_error(Z_DATA_ERROR, "compression failed")
|
||
return None
|
||
if wbits > MAX_WBITS:
|
||
gzip_len: t.CSizeT = 10 + c.Deref(out_len) + 8
|
||
if pool == None:
|
||
gzip_buf: BYTEPTR = BYTEPTR(stdlib.malloc(gzip_len))
|
||
else:
|
||
gzip_buf: BYTEPTR = BYTEPTR(pool.alloc(gzip_len))
|
||
if not gzip_buf:
|
||
if pool == None:
|
||
stdlib.free(raw_result)
|
||
else:
|
||
pool.free(raw_result)
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
pos: t.CSizeT = 0
|
||
gzip_buf[pos + 0] = 0x1F
|
||
gzip_buf[pos + 1] = 0x8B
|
||
gzip_buf[pos + 2] = 0x08
|
||
gzip_buf[pos + 3] = 0x00
|
||
gzip_buf[pos + 4] = 0x00
|
||
gzip_buf[pos + 5] = 0x00
|
||
gzip_buf[pos + 6] = 0x00
|
||
gzip_buf[pos + 7] = 0x00
|
||
gzip_buf[pos + 8] = 0x00
|
||
gzip_buf[pos + 9] = 0xFF
|
||
pos += 10
|
||
|
||
string.memcpy(gzip_buf + pos, raw_result, c.Deref(out_len))
|
||
pos += c.Deref(out_len)
|
||
if pool == None:
|
||
stdlib.free(raw_result)
|
||
else:
|
||
pool.free(raw_result)
|
||
|
||
crc: t.CUInt32T = zchecksum.zchecksum_crc32(data, data_len, 0)
|
||
gzip_buf[pos + 0] = (crc) & 0xFF
|
||
gzip_buf[pos + 1] = (crc >> 8) & 0xFF
|
||
gzip_buf[pos + 2] = (crc >> 16) & 0xFF
|
||
gzip_buf[pos + 3] = (crc >> 24) & 0xFF
|
||
gzip_buf[pos + 4] = (data_len) & 0xFF
|
||
gzip_buf[pos + 5] = (data_len >> 8) & 0xFF
|
||
gzip_buf[pos + 6] = (data_len >> 16) & 0xFF
|
||
gzip_buf[pos + 7] = (data_len >> 24) & 0xFF
|
||
pos += 8
|
||
c.Set(c.Deref(out_len), pos)
|
||
return shrink_to_fit(pool, gzip_buf, pos)
|
||
return shrink_to_fit(pool, raw_result, c.Deref(out_len))
|
||
|
||
|
||
def decompress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT,
|
||
wbits: t.CInt, bufsize: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not data and data_len > 0:
|
||
set_error(Z_STREAM_ERROR, "data is None but data_len > 0")
|
||
return None
|
||
if not out_len:
|
||
set_error(Z_STREAM_ERROR, "out_len is None")
|
||
return None
|
||
result: UINT8PTR = zinflate.zinflate_one_shot(pool, data, data_len, wbits, bufsize, out_len)
|
||
if not result:
|
||
set_error(Z_DATA_ERROR, "decompression failed")
|
||
return None
|
||
return shrink_to_fit(pool, result, c.Deref(out_len))
|
||
|
||
def compressobj(pool: memhub.MemManager | t.CPtr, level: t.CInt, method: t.CInt, wbits: t.CInt,
|
||
memLevel: t.CInt, strategy: t.CInt,
|
||
zdict: BYTEPTR, zdict_len: t.CSizeT) -> Compress | t.CPtr:
|
||
if pool == None:
|
||
obj: Compress | t.CPtr = stdlib.calloc(1, Compress.__sizeof__())
|
||
else:
|
||
obj: Compress | t.CPtr = pool.alloc(Compress.__sizeof__())
|
||
string.memset(obj, 0, Compress.__sizeof__())
|
||
if not obj:
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
obj.pool = pool
|
||
actual_wbits: t.CInt = wbits
|
||
if wbits > MAX_WBITS: actual_wbits = wbits - 16
|
||
s: zdeflate.zdeflate_stream | t.CPtr = zdeflate.zdeflate_create(pool, level, actual_wbits, memLevel, strategy)
|
||
if not s:
|
||
if pool == None:
|
||
stdlib.free(obj)
|
||
else:
|
||
pool.free(obj)
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
obj.stream = s
|
||
obj.is_initialized = 1
|
||
obj.level = level
|
||
obj.method = method
|
||
obj.wbits = wbits
|
||
obj.memLevel = memLevel
|
||
obj.strategy = strategy
|
||
if pool == None:
|
||
obj.input_buf = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
obj.input_buf = BYTEPTR(pool.alloc(256))
|
||
string.memset(obj.input_buf, 0, 256)
|
||
obj.input_buf_cap = 256
|
||
obj.input_buf_len = 0
|
||
obj.header_written = 0
|
||
if zdict and zdict_len > 0:
|
||
s.set_dictionary(zdict, zdict_len)
|
||
obj.zdict = clone_bytes(pool, zdict, zdict_len)
|
||
obj.zdict_len = zdict_len
|
||
return obj
|
||
|
||
|
||
def decompressobj(pool: memhub.MemManager | t.CPtr, wbits: t.CInt, zdict: BYTEPTR, zdict_len: t.CSizeT) -> Decompress | t.CPtr:
|
||
if pool == None:
|
||
obj: Decompress | t.CPtr = stdlib.calloc(1, Decompress.__sizeof__())
|
||
else:
|
||
obj: Decompress | t.CPtr = pool.alloc(Decompress.__sizeof__())
|
||
string.memset(obj, 0, Decompress.__sizeof__())
|
||
if not obj:
|
||
set_error(Z_MEM_ERROR, "out of memory")
|
||
return None
|
||
obj.pool = pool
|
||
actual_wbits: t.CInt = wbits
|
||
if wbits > MAX_WBITS: actual_wbits = wbits - 16
|
||
obj.stream = None
|
||
obj.is_initialized = 1
|
||
obj.wbits = actual_wbits
|
||
if zdict and zdict_len > 0:
|
||
obj.zdict = clone_bytes(pool, zdict, zdict_len)
|
||
obj.zdict_len = zdict_len
|
||
if pool == None:
|
||
obj.input_buf = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
obj.input_buf = BYTEPTR(pool.alloc(256))
|
||
string.memset(obj.input_buf, 0, 256)
|
||
obj.input_buf_cap = 256
|
||
obj.input_buf_len = 0
|
||
return obj
|
||
|
||
def zlib_adler32(data: BYTEPTR, length: t.CSizeT, value: t.CUnsignedLong) -> t.CUnsignedLong:
|
||
return zchecksum.zchecksum_adler32(data, length, UINT32(value))
|
||
|
||
def zlib_crc32(data: BYTEPTR, length: t.CSizeT, value: t.CUnsignedLong) -> t.CUnsignedLong:
|
||
return zchecksum.zchecksum_crc32(data, length, UINT32(value))
|