pub(crate) use zlib::make_module; #[pymodule] mod zlib { use crate::common::lock::PyMutex; use crate::vm::{ builtins::{PyBaseExceptionRef, PyBytes, PyBytesRef, PyIntRef, PyTypeRef}, function::{ArgBytesLike, OptionalArg}, IntoPyRef, PyResult, PyValue, VirtualMachine, }; use adler32::RollingAdler32 as Adler32; use crossbeam_utils::atomic::AtomicCell; use flate2::{ write::ZlibEncoder, Compress, Compression, Decompress, FlushCompress, FlushDecompress, Status, }; use std::io::Write; #[cfg(not(feature = "zlib"))] mod constants { pub const Z_NO_COMPRESSION: i32 = 0; pub const Z_BEST_COMPRESSION: i32 = 9; pub const Z_BEST_SPEED: i32 = 1; pub const Z_DEFAULT_COMPRESSION: i32 = -1; pub const Z_NO_FLUSH: i32 = 0; pub const Z_PARTIAL_FLUSH: i32 = 1; pub const Z_SYNC_FLUSH: i32 = 2; pub const Z_FULL_FLUSH: i32 = 3; // not sure what the value here means, but it's the only compression method zlibmodule // supports, so it doesn't really matter pub const Z_DEFLATED: i32 = 8; } #[cfg(feature = "zlib")] use libz_sys as constants; #[pyattr] use constants::{ Z_BEST_COMPRESSION, Z_BEST_SPEED, Z_DEFAULT_COMPRESSION, Z_DEFLATED as DEFLATED, Z_FULL_FLUSH, Z_NO_COMPRESSION, Z_NO_FLUSH, Z_PARTIAL_FLUSH, Z_SYNC_FLUSH, }; #[cfg(feature = "zlib")] #[pyattr] use libz_sys::{ Z_BLOCK, Z_DEFAULT_STRATEGY, Z_FILTERED, Z_FINISH, Z_FIXED, Z_HUFFMAN_ONLY, Z_RLE, Z_TREES, }; // copied from zlibmodule.c (commit 530f506ac91338) #[pyattr] const MAX_WBITS: u8 = 15; #[pyattr] const DEF_BUF_SIZE: usize = 16 * 1024; #[pyattr] const DEF_MEM_LEVEL: u8 = 8; #[pyattr(once)] fn error(vm: &VirtualMachine) -> PyTypeRef { vm.ctx.new_exception_type( "zlib", "error", Some(vec![vm.ctx.exceptions.exception_type.clone()]), ) } /// Compute an Adler-32 checksum of data. #[pyfunction] fn adler32(data: ArgBytesLike, begin_state: OptionalArg) -> u32 { data.with_ref(|data| { let begin_state = begin_state.map_or(1, |i| i.as_u32_mask()); let mut hasher = Adler32::from_value(begin_state); hasher.update_buffer(data); hasher.hash() }) } /// Compute a CRC-32 checksum of data. #[pyfunction] fn crc32(data: ArgBytesLike, begin_state: OptionalArg) -> u32 { crate::binascii::crc32(data, begin_state) } fn compression_from_int(level: Option) -> Option { match level.unwrap_or(Z_DEFAULT_COMPRESSION) { Z_DEFAULT_COMPRESSION => Some(Compression::default()), valid_level @ Z_NO_COMPRESSION..=Z_BEST_COMPRESSION => { Some(Compression::new(valid_level as u32)) } _ => None, } } /// Returns a bytes object containing compressed data. #[pyfunction] fn compress( data: ArgBytesLike, level: OptionalArg, vm: &VirtualMachine, ) -> PyResult { let compression = compression_from_int(level.into_option()) .ok_or_else(|| new_zlib_error("Bad compression level", vm))?; let mut encoder = ZlibEncoder::new(Vec::new(), compression); data.with_ref(|input_bytes| encoder.write_all(input_bytes).unwrap()); let encoded_bytes = encoder.finish().unwrap(); Ok(vm.ctx.new_bytes(encoded_bytes)) } enum InitOptions { Standard { header: bool, // [De]Compress::new_with_window_bits is only enabled for zlib; miniz_oxide doesn't // support wbits (yet?) #[cfg(feature = "zlib")] wbits: u8, }, #[cfg(feature = "zlib")] Gzip { wbits: u8 }, } impl InitOptions { fn decompress(self) -> Decompress { match self { #[cfg(not(feature = "zlib"))] Self::Standard { header } => Decompress::new(header), #[cfg(feature = "zlib")] Self::Standard { header, wbits } => Decompress::new_with_window_bits(header, wbits), #[cfg(feature = "zlib")] Self::Gzip { wbits } => Decompress::new_gzip(wbits), } } fn compress(self, level: Compression) -> Compress { match self { #[cfg(not(feature = "zlib"))] Self::Standard { header } => Compress::new(level, header), #[cfg(feature = "zlib")] Self::Standard { header, wbits } => { Compress::new_with_window_bits(level, header, wbits) } #[cfg(feature = "zlib")] Self::Gzip { wbits } => Compress::new_gzip(level, wbits), } } } fn header_from_wbits(wbits: OptionalArg, vm: &VirtualMachine) -> PyResult { let wbits = wbits.unwrap_or(MAX_WBITS as i8); let header = wbits > 0; let wbits = wbits.abs() as u8; match wbits { 9..=15 => Ok(InitOptions::Standard { header, #[cfg(feature = "zlib")] wbits, }), #[cfg(feature = "zlib")] 25..=31 => Ok(InitOptions::Gzip { wbits: wbits - 16 }), _ => Err(vm.new_value_error("Invalid initialization option".to_owned())), } } fn _decompress( mut data: &[u8], d: &mut Decompress, bufsize: usize, max_length: Option, is_flush: bool, vm: &VirtualMachine, ) -> PyResult<(Vec, bool)> { if data.is_empty() { return Ok((Vec::new(), true)); } let mut buf = Vec::new(); loop { let final_chunk = data.len() <= CHUNKSIZE; let chunk = if final_chunk { data } else { &data[..CHUNKSIZE] }; // if this is the final chunk, finish it let flush = if is_flush { if final_chunk { FlushDecompress::Finish } else { FlushDecompress::None } } else { FlushDecompress::Sync }; loop { let additional = if let Some(max_length) = max_length { std::cmp::min(bufsize, max_length - buf.capacity()) } else { bufsize }; if additional == 0 { return Ok((buf, false)); } buf.reserve_exact(additional); let prev_in = d.total_in(); let status = d .decompress_vec(chunk, &mut buf, flush) .map_err(|_| new_zlib_error("invalid input data", vm))?; let consumed = d.total_in() - prev_in; data = &data[consumed as usize..]; let stream_end = status == Status::StreamEnd; if stream_end || data.is_empty() { // we've reached the end of the stream, we're done buf.shrink_to_fit(); return Ok((buf, stream_end)); } else if !chunk.is_empty() && consumed == 0 { // we're gonna need a bigger buffer continue; } else { // next chunk break; } } } } /// Returns a bytes object containing the uncompressed data. #[pyfunction] fn decompress( data: ArgBytesLike, wbits: OptionalArg, bufsize: OptionalArg, vm: &VirtualMachine, ) -> PyResult> { data.with_ref(|data| { let bufsize = bufsize.unwrap_or(DEF_BUF_SIZE); let mut d = header_from_wbits(wbits, vm)?.decompress(); _decompress(data, &mut d, bufsize, None, false, vm).and_then(|(buf, stream_end)| { if stream_end { Ok(buf) } else { Err(new_zlib_error( "Error -5 while decompressing data: incomplete or truncated stream", vm, )) } }) }) } #[pyfunction] fn decompressobj(args: DecompressobjArgs, vm: &VirtualMachine) -> PyResult { #[allow(unused_mut)] let mut decompress = header_from_wbits(args.wbits, vm)?.decompress(); #[cfg(feature = "zlib")] if let OptionalArg::Present(dict) = args.zdict { dict.with_ref(|d| decompress.set_dictionary(d).unwrap()); } Ok(PyDecompress { decompress: PyMutex::new(decompress), eof: AtomicCell::new(false), unused_data: PyMutex::new(PyBytes::from(vec![]).into_ref(vm)), unconsumed_tail: PyMutex::new(PyBytes::from(vec![]).into_ref(vm)), }) } #[pyattr] #[pyclass(name = "Decompress")] #[derive(Debug, PyValue)] struct PyDecompress { decompress: PyMutex, eof: AtomicCell, unused_data: PyMutex, unconsumed_tail: PyMutex, } #[pyimpl] impl PyDecompress { #[pyproperty] fn eof(&self) -> bool { self.eof.load() } #[pyproperty] fn unused_data(&self) -> PyBytesRef { self.unused_data.lock().clone() } #[pyproperty] fn unconsumed_tail(&self) -> PyBytesRef { self.unconsumed_tail.lock().clone() } fn save_unused_input( &self, d: &mut Decompress, data: &[u8], stream_end: bool, orig_in: u64, vm: &VirtualMachine, ) { let leftover = &data[(d.total_in() - orig_in) as usize..]; if stream_end && !leftover.is_empty() { let mut unused_data = self.unused_data.lock(); let unused: Vec<_> = unused_data .as_bytes() .iter() .chain(leftover) .copied() .collect(); *unused_data = unused.into_pyref(vm); } } #[pymethod] fn decompress(&self, args: DecompressArgs, vm: &VirtualMachine) -> PyResult> { let max_length = if args.max_length == 0 { None } else { Some(args.max_length) }; let data = args.data.borrow_buf(); let data = &*data; let mut d = self.decompress.lock(); let orig_in = d.total_in(); let (ret, stream_end) = match _decompress(data, &mut d, DEF_BUF_SIZE, max_length, false, vm) { Ok((buf, true)) => { self.eof.store(true); (Ok(buf), true) } Ok((buf, false)) => (Ok(buf), false), Err(err) => (Err(err), false), }; self.save_unused_input(&mut d, data, stream_end, orig_in, vm); let leftover = if stream_end { b"" } else { &data[(d.total_in() - orig_in) as usize..] }; let mut unconsumed_tail = self.unconsumed_tail.lock(); if !leftover.is_empty() || !unconsumed_tail.is_empty() { *unconsumed_tail = PyBytes::from(leftover.to_owned()).into_ref(vm); } ret } #[pymethod] fn flush(&self, length: OptionalArg, vm: &VirtualMachine) -> PyResult> { let length = match length { OptionalArg::Present(l) => { if l <= 0 { return Err( vm.new_value_error("length must be greater than zero".to_owned()) ); } else { l as usize } } OptionalArg::Missing => DEF_BUF_SIZE, }; let mut data = self.unconsumed_tail.lock(); let mut d = self.decompress.lock(); let orig_in = d.total_in(); let (ret, stream_end) = match _decompress(&data, &mut d, length, None, true, vm) { Ok((buf, stream_end)) => (Ok(buf), stream_end), Err(err) => (Err(err), false), }; self.save_unused_input(&mut d, &data, stream_end, orig_in, vm); *data = PyBytes::from(Vec::new()).into_ref(vm); // TODO: drop the inner decompressor, somehow // if stream_end { // // } ret } } #[derive(FromArgs)] struct DecompressArgs { #[pyarg(positional)] data: ArgBytesLike, #[pyarg(any, default = "0")] max_length: usize, } #[derive(FromArgs)] struct DecompressobjArgs { #[pyarg(any, optional)] wbits: OptionalArg, #[cfg(feature = "zlib")] #[pyarg(any, optional)] zdict: OptionalArg, } #[pyfunction] fn compressobj( level: OptionalArg, // only DEFLATED is valid right now, it's w/e _method: OptionalArg, wbits: OptionalArg, // these aren't used. _mem_level: OptionalArg, // this is memLevel in CPython _strategy: OptionalArg, _zdict: OptionalArg, vm: &VirtualMachine, ) -> PyResult { let level = compression_from_int(level.into_option()) .ok_or_else(|| vm.new_value_error("invalid initialization option".to_owned()))?; let compress = header_from_wbits(wbits, vm)?.compress(level); Ok(PyCompress { inner: PyMutex::new(CompressInner { compress, unconsumed: Vec::new(), }), }) } #[derive(Debug)] struct CompressInner { compress: Compress, unconsumed: Vec, } #[pyattr] #[pyclass(name = "Compress")] #[derive(Debug, PyValue)] struct PyCompress { inner: PyMutex, } #[pyimpl] impl PyCompress { #[pymethod] fn compress(&self, data: ArgBytesLike, vm: &VirtualMachine) -> PyResult> { let mut inner = self.inner.lock(); data.with_ref(|b| inner.compress(b, vm)) } // TODO: mode argument isn't used #[pymethod] fn flush(&self, _mode: OptionalArg, vm: &VirtualMachine) -> PyResult> { self.inner.lock().flush(vm) } // TODO: This is an optional feature of Compress // #[pymethod] // #[pymethod(magic)] // #[pymethod(name = "__deepcopy__")] // fn copy(&self) -> Self { // todo!("") // } } const CHUNKSIZE: usize = u32::MAX as usize; impl CompressInner { fn compress(&mut self, data: &[u8], vm: &VirtualMachine) -> PyResult> { let orig_in = self.compress.total_in() as usize; let mut cur_in = 0; let unconsumed = std::mem::take(&mut self.unconsumed); let mut buf = Vec::new(); 'outer: for chunk in unconsumed.chunks(CHUNKSIZE).chain(data.chunks(CHUNKSIZE)) { while cur_in < chunk.len() { buf.reserve(DEF_BUF_SIZE); let status = self .compress .compress_vec(&chunk[cur_in..], &mut buf, FlushCompress::None) .map_err(|_| { self.unconsumed.extend_from_slice(&data[cur_in..]); new_zlib_error("error while compressing", vm) })?; cur_in = (self.compress.total_in() as usize) - orig_in; match status { Status::Ok => continue, Status::StreamEnd => break 'outer, _ => break, } } } self.unconsumed.extend_from_slice(&data[cur_in..]); buf.shrink_to_fit(); Ok(buf) } // TODO: flush mode (FlushDecompress) parameter fn flush(&mut self, vm: &VirtualMachine) -> PyResult> { let data = std::mem::take(&mut self.unconsumed); let mut data_it = data.chunks(CHUNKSIZE); let mut buf = Vec::new(); loop { let chunk = data_it.next().unwrap_or(&[]); if buf.len() == buf.capacity() { buf.reserve(DEF_BUF_SIZE); } let status = self .compress .compress_vec(chunk, &mut buf, FlushCompress::Finish) .map_err(|_| new_zlib_error("error while compressing", vm))?; match status { Status::StreamEnd => break, _ => continue, } } buf.shrink_to_fit(); Ok(buf) } } fn new_zlib_error(message: &str, vm: &VirtualMachine) -> PyBaseExceptionRef { vm.new_exception_msg(vm.class("zlib", "error"), message.to_owned()) } }