From 2505ad5f95831e62c821aef1ed14054655e9dc46 Mon Sep 17 00:00:00 2001 From: dot Date: Mon, 12 Oct 2026 01:21:23 +0900 Subject: [PATCH 1/2] Implement Python 3.15 base64 encoding and decoding Follow CPython 3.15.0 final d220409a13b77de7a6c30abe16ea90db01528edc for base64 options, alphabets, padding and canonical decoding. Keep reverse-table caches in traced module-native state and preserve module-bound callable semantics. Add bounded Python regressions without editing Lib. Assisted-by: OpenAI Codex:model identifier unavailable --- Cargo.lock | 7 - Cargo.toml | 1 - crates/common/Cargo.toml | 3 +- crates/common/src/binascii.rs | 187 ++++++--- crates/stdlib/src/binascii.rs | 194 +++++++++- crates/vm/src/builtins/builtin_func.rs | 4 +- crates/vm/src/builtins/module.rs | 31 +- .../snippets/stdlib_binascii_base64.py | 363 ++++++++++++++++++ 8 files changed, 701 insertions(+), 89 deletions(-) create mode 100644 extra_tests/snippets/stdlib_binascii_base64.py diff --git a/Cargo.lock b/Cargo.lock index 9a0b3b46c9b..cf5e0f9c0cd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -320,12 +320,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf" -[[package]] -name = "base64" -version = "0.23.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" - [[package]] name = "base64ct" version = "1.8.3" @@ -3761,7 +3755,6 @@ name = "rustpython-common" version = "0.6.1" dependencies = [ "ascii", - "base64", "bitflags 2.13.2", "blake2b_simd", "blake2s_simd", diff --git a/Cargo.toml b/Cargo.toml index 792f2b981af..0b55f2e8277 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -198,7 +198,6 @@ phf = { version = "0.14.0", default-features = false, features = ["macros"]} adler32 = "1.2.0" approx = "0.5.1" ascii = { version = "1.1", default-features = false } -base64 = "0.23" blake2b_simd = "1.0.4" blake2s_simd = "1.0.4" bitflags = "2.11.0" diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml index 25247e4d010..ea31322575e 100644 --- a/crates/common/Cargo.toml +++ b/crates/common/Cargo.toml @@ -10,7 +10,7 @@ license.workspace = true [features] default = ["std"] -binascii = ["dep:base64", "dep:crc32fast"] +binascii = ["dep:crc32fast"] bz2 = ["std", "dep:bzip2"] cjk-codecs = [] inet = ["std"] @@ -39,7 +39,6 @@ rustpython-unicode = { workspace = true } rustpython-wtf8 = { workspace = true } ascii = { workspace = true } -base64 = { workspace = true, optional = true } bitflags = { workspace = true } blake2b_simd = { workspace = true, optional = true } blake2s_simd = { workspace = true, optional = true } diff --git a/crates/common/src/binascii.rs b/crates/common/src/binascii.rs index 2a262d3a28b..73a17c1b01e 100644 --- a/crates/common/src/binascii.rs +++ b/crates/common/src/binascii.rs @@ -7,7 +7,6 @@ //! the caller's responsibility (see `mod.rs`). use alloc::{vec, vec::Vec}; -use base64::Engine; /// The base64 pad character `=`. pub const PAD: u8 = 61u8; @@ -31,13 +30,17 @@ pub enum Error { TooLong, /// `a2b_base64`: base64 decoding failed. Base64(Base64DecodeError), + /// The base64 output would exceed the maximum Python bytes length. + TooMuchBase64Data, + /// Reserving the transform output failed. + Memory, } /// Mirrors CPython's `binascii_a2b_base64_impl` (`Modules/binascii.c`) error /// paths one-for-one, so the caller can build the exact Python error message. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Base64DecodeError { - /// A `=` at `quad_pos == 0, i == 0`, strict mode. + /// Padding before any decoded output, in strict mode. LeadingPaddingNotAllowed, /// A `=` outside the padding-completion window, strict mode. ExcessPaddingNotAllowed, @@ -51,6 +54,10 @@ pub enum Base64DecodeError { InvalidLastSymbol { index: usize }, /// `quad_pos != 0` and the trailing padding does not complete the quad. IncorrectPadding, + /// Padding is forbidden when decoding unpadded input in strict mode. + PaddingNotAllowed, + /// The final symbol contains non-zero unused bits. + NonZeroPaddingBits, } const fn hex_nibble(n: u8) -> u8 { @@ -230,45 +237,53 @@ pub fn crc_hqx(bytes: &[u8], init: u32) -> u32 { crc } -/// `a2b_base64`. A line-by-line port of `binascii_a2b_base64_impl` -/// (CPython `Modules/binascii.c`): padding characters within the completion -/// window (`quad_pos >= 2 && quad_pos + pads <= 4`) are always skipped — -/// even in strict mode — and non-strict mode additionally skips *every* -/// pad and every non-alphabet byte outside that window. Errors therefore -/// only fire in strict mode (mid-loop) or from the two post-loop length -/// checks, which apply unconditionally. -pub fn a2b_base64(b: &[u8], strict_mode: bool) -> Result, Error> { - // Converts between ASCII and base-64 characters. The index of a given number yields the - // number in ASCII while the value of said index yields the number in base-64. For example - // "=" is 61 in ASCII but 0 (since it's the pad character) in base-64, so BASE64_TABLE[61] == 0 - #[rustfmt::skip] - const BASE64_TABLE: [i8; 256] = [ - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,62, -1,-1,-1,63, - 52,53,54,55, 56,57,58,59, 60,61,-1,-1, -1, 0,-1,-1, /* Note PAD->0 */ - -1, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,10, 11,12,13,14, - 15,16,17,18, 19,20,21,22, 23,24,25,-1, -1,-1,-1,-1, - -1,26,27,28, 29,30,31,32, 33,34,35,36, 37,38,39,40, - 41,42,43,44, 45,46,47,48, 49,50,51,-1, -1,-1,-1,-1, - - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, -1,-1,-1,-1, - ]; +/// RFC 4648's standard base64 alphabet. +pub const BASE64_ALPHABET: &[u8; 64] = + b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + +/// Build the decoder table, with later duplicate symbols taking precedence. +#[must_use] +pub const fn base64_reverse_table(alphabet: &[u8; 64]) -> [u8; 256] { + let mut table = [u8::MAX; 256]; + let mut index = 0; + while index < 64 { + table[alphabet[index] as usize] = index as u8; + index += 1; + } + // Padding always takes precedence, even in a custom alphabet. + table[PAD as usize] = 64; + table +} - let mut decoded: Vec = Vec::with_capacity(b.len().div_ceil(4) * 3); +/// Options for CPython 3.15's base64 decoder. +pub struct Base64DecodeOptions<'a> { + pub strict_mode: bool, + pub padded: bool, + pub table: &'a [u8; 256], + pub ignored: &'a [bool; 256], + pub canonical: bool, +} - let mut quad_pos: u32 = 0; // position in the quad, 0..=3 - let mut pads: u32 = 0; +/// `a2b_base64`, following CPython 3.15's `Modules/binascii.c`. +pub fn a2b_base64(b: &[u8], options: Base64DecodeOptions<'_>) -> Result, Error> { + let Base64DecodeOptions { + strict_mode, + padded, + table, + ignored, + canonical, + } = options; + + let mut decoded: Vec = Vec::new(); + decoded + .try_reserve_exact(b.len().div_ceil(4) * 3) + .map_err(|_| Error::Memory)?; + + let mut quad_pos: usize = 0; // position in the quad, 0..=3 + let mut pads: usize = 0; let mut left_char: u8 = 0; - for (i, &el) in b.iter().enumerate() { - if el == PAD { + for &el in b { + if padded && el == PAD { pads += 1; if quad_pos >= 2 && quad_pos + pads <= 4 { continue; @@ -276,7 +291,7 @@ pub fn a2b_base64(b: &[u8], strict_mode: bool) -> Result, Error> { // RFC 4648 section-3.3: implementations MAY ignore a pad // character present before the end of the encoded data, and // MAY ignore excess pad characters. Non-strict mode does. - if !strict_mode { + if !strict_mode || ignored[PAD as usize] { continue; } if quad_pos == 1 { @@ -284,22 +299,26 @@ pub fn a2b_base64(b: &[u8], strict_mode: bool) -> Result, Error> { // `quad_pos == 1` check below. break; } - return Err(Error::Base64(if quad_pos == 0 && i == 0 { + return Err(Error::Base64(if quad_pos == 0 && decoded.is_empty() { Base64DecodeError::LeadingPaddingNotAllowed } else { Base64DecodeError::ExcessPaddingNotAllowed })); } - let binary_char = BASE64_TABLE[el as usize]; - if binary_char < 0 { - if strict_mode { - return Err(Error::Base64(Base64DecodeError::OnlyBase64DataAllowed)); + let binary_char = table[el as usize]; + if binary_char >= 64 { + if strict_mode && !ignored[el as usize] { + return Err(Error::Base64(if el == PAD { + Base64DecodeError::PaddingNotAllowed + } else { + Base64DecodeError::OnlyBase64DataAllowed + })); } continue; } - if pads > 0 && strict_mode { + if pads > 0 && strict_mode && !ignored[PAD as usize] { return Err(Error::Base64(if quad_pos + pads == 4 { Base64DecodeError::ExcessDataAfterPadding } else { @@ -312,21 +331,21 @@ pub fn a2b_base64(b: &[u8], strict_mode: bool) -> Result, Error> { match quad_pos { 0 => { quad_pos = 1; - left_char = binary_char as u8; + left_char = binary_char; } 1 => { quad_pos = 2; - decoded.push((left_char << 2) | (binary_char >> 4) as u8); - left_char = (binary_char & 0x0f) as u8; + decoded.push((left_char << 2) | (binary_char >> 4)); + left_char = binary_char & 0x0f; } 2 => { quad_pos = 3; - decoded.push((left_char << 4) | (binary_char >> 2) as u8); - left_char = (binary_char & 0x03) as u8; + decoded.push((left_char << 4) | (binary_char >> 2)); + left_char = binary_char & 0x03; } 3 => { quad_pos = 0; - decoded.push((left_char << 6) | binary_char as u8); + decoded.push((left_char << 6) | binary_char); left_char = 0; } _ => unsafe { @@ -344,24 +363,72 @@ pub fn a2b_base64(b: &[u8], strict_mode: bool) -> Result, Error> { })); } - if quad_pos != 0 && quad_pos + pads < 4 { + if padded && quad_pos != 0 && quad_pos + pads < 4 { return Err(Error::Base64(Base64DecodeError::IncorrectPadding)); } + if canonical && left_char != 0 { + return Err(Error::Base64(Base64DecodeError::NonZeroPaddingBits)); + } Ok(decoded) } -/// `b2a_base64`. -#[must_use] -pub fn b2a_base64(data: &[u8], newline: bool) -> Vec { - // https://stackoverflow.com/questions/63916821 - let mut encoded = base64::engine::general_purpose::STANDARD - .encode(data) - .into_bytes(); +/// `b2a_base64`, including CPython 3.15's wrapping and custom alphabets. +pub fn b2a_base64( + data: &[u8], + padded: bool, + wrapcol: usize, + newline: bool, + alphabet: &[u8; 64], +) -> Result, Error> { + let pads = (3 - data.len() % 3) % 3; + let mut size = data + .len() + .div_ceil(3) + .checked_mul(4) + .ok_or(Error::TooMuchBase64Data)?; + if !padded { + size -= pads; + } + let width = if wrapcol == 0 { + usize::MAX + } else { + wrapcol.max(4) / 4 * 4 + }; + let newlines = size.saturating_sub(1) / width; + size = size + .checked_add(newlines) + .and_then(|n| n.checked_add(usize::from(newline))) + .filter(|&n| isize::try_from(n).is_ok()) + .ok_or(Error::TooMuchBase64Data)?; + let mut encoded = Vec::new(); + encoded.try_reserve_exact(size).map_err(|_| Error::Memory)?; + let mut column = 0; + let mut emit = |symbol| { + if column == width { + encoded.push(b'\n'); + column = 0; + } + encoded.push(symbol); + column += 1; + }; + for chunk in data.chunks(3) { + let value = (u32::from(chunk[0]) << 16) + | (u32::from(*chunk.get(1).unwrap_or(&0)) << 8) + | u32::from(*chunk.get(2).unwrap_or(&0)); + for index in 0..=chunk.len() { + emit(alphabet[((value >> (18 - index * 6)) & 63) as usize]); + } + if padded { + for _ in chunk.len() + 1..4 { + emit(PAD); + } + } + } if newline { encoded.push(b'\n'); } - encoded + Ok(encoded) } #[inline] diff --git a/crates/stdlib/src/binascii.rs b/crates/stdlib/src/binascii.rs index 73c20506dd7..dab01ceae25 100644 --- a/crates/stdlib/src/binascii.rs +++ b/crates/stdlib/src/binascii.rs @@ -10,12 +10,65 @@ use rustpython_vm::{VirtualMachine, builtins::PyBaseExceptionRef}; mod decl { use super::new_binascii_error; use crate::vm::{ - PyResult, VirtualMachine, - builtins::PyTypeRef, - function::{ArgAsciiBuffer, ArgBytesLike, ArgIndex, OptionalArg}, + AsObject, Py, PyObjectRef, PyRef, PyResult, VirtualMachine, + builtins::{PyBytes, PyBytesRef, PyDict, PyModule, PyTypeRef}, + function::{ + ArgAsciiBuffer, ArgBytesLike, ArgIndex, ArgIntoBool, ItemDoc, OptionalArg, PyMethodDef, + PyMethodFlags, + }, }; use rustpython_common::binascii; + pub(crate) fn module_exec(vm: &VirtualMachine, module: &Py) -> PyResult<()> { + __module_exec(vm, module); + module.set_native_state(vm.ctx.new_dict().into()); + const DECODE: &PyMethodDef = &PyMethodDef::new_const( + "a2b_base64", + a2b_base64, + PyMethodFlags::empty(), + ItemDoc::static_text( + "a2b_base64($module, data, /, *, strict_mode=, padded=True, alphabet=BASE64_ALPHABET, ignorechars=, canonical=False)\n--\n\nDecode a line of base64 data.", + ), + ); + let function = DECODE.build_bound_function(&vm.ctx, module.to_owned().into()); + function + .as_object() + .set_attr("__module__", vm.ctx.new_str("binascii"), vm)?; + module.dict().set_item("a2b_base64", function.into(), vm)?; + Ok(()) + } + + fn reverse_table( + module: &Py, + alphabet: &PyBytesRef, + vm: &VirtualMachine, + ) -> PyResult<[u8; 256]> { + let bytes: &[u8; 64] = alphabet + .as_bytes() + .try_into() + .map_err(|_| vm.new_value_error("alphabet must have length 64"))?; + let cache = module + .native_state() + .ok_or_else(|| vm.new_runtime_error("Missing binascii state"))? + .downcast::() + .map_err(|_| vm.new_runtime_error("Broken binascii cache"))?; + // The actual bytes object is the key: subclass hash/equality callbacks + // are part of CPython's cache lookup and insertion semantics. + if let Some(value) = cache.get_item_opt(alphabet.as_object(), vm)? { + return value + .downcast_ref::() + .and_then(|value| value.as_bytes().try_into().ok()) + .ok_or_else(|| vm.new_runtime_error("Broken binascii cache")); + } + let table = binascii::base64_reverse_table(bytes); + cache.set_item( + alphabet.as_object(), + vm.ctx.new_bytes(table.to_vec()).into(), + vm, + )?; + Ok(table) + } + #[pyattr(name = "Error", once)] pub(super) fn error_type(vm: &VirtualMachine) -> PyTypeRef { vm.ctx.new_exception_type( @@ -86,30 +139,137 @@ mod decl { data.with_ref(|bytes| binascii::crc_hqx(bytes, crc.into_int_ref().as_u32_mask())) } - #[derive(FromArgs)] - struct NewlineArg { - #[pyarg(named, default = true)] - newline: bool, + #[pyattr(name = "BASE64_ALPHABET")] + fn base64_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx.new_bytes(binascii::BASE64_ALPHABET.to_vec()) + } + + #[pyattr(name = "URLSAFE_BASE64_ALPHABET")] + fn urlsafe_base64_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx + .new_bytes(b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_".to_vec()) } #[derive(FromArgs)] struct A2bBase64Args { #[pyarg(positional)] data: ArgAsciiBuffer, - #[pyarg(named, default)] - strict_mode: bool, + #[pyarg(named, optional)] + strict_mode: OptionalArg, + #[pyarg(named, default = ArgIntoBool::TRUE, py_default = "True")] + padded: ArgIntoBool, + #[pyarg(named, optional, py_default = "BASE64_ALPHABET")] + alphabet: OptionalArg, + #[pyarg(named, optional)] + ignorechars: OptionalArg, + #[pyarg(named, optional, py_default = "False")] + canonical: OptionalArg, } - #[pyfunction] - fn a2b_base64(args: A2bBase64Args, vm: &VirtualMachine) -> PyResult> { - let A2bBase64Args { data, strict_mode } = args; - data.with_ref(|b| binascii::a2b_base64(b, strict_mode)) + fn a2b_base64( + module: PyRef, + args: A2bBase64Args, + vm: &VirtualMachine, + ) -> PyResult> { + let A2bBase64Args { + data, + strict_mode, + padded, + alphabet, + ignorechars, + canonical, + } = args; + let strict_mode = match strict_mode { + OptionalArg::Present(value) => value.into_bool(), + OptionalArg::Missing => matches!(&ignorechars, OptionalArg::Present(_)), + }; + let result = (|| { + // Convert the last Clinic argument while both acquired buffers are alive. + let canonical = match canonical { + OptionalArg::Present(value) => value.try_to_bool(vm)?, + OptionalArg::Missing => false, + }; + const DEFAULT_TABLE: [u8; 256] = + binascii::base64_reverse_table(binascii::BASE64_ALPHABET); + let table = match &alphabet { + OptionalArg::Present(value) => reverse_table(&module, value, vm)?, + OptionalArg::Missing => DEFAULT_TABLE, + }; + let mut ignored = [false; 256]; + if strict_mode && let OptionalArg::Present(value) = &ignorechars { + value.with_ref(|bytes| { + for &symbol in bytes { + ignored[symbol as usize] = true; + } + }); + } + data.with_ref(|bytes| { + binascii::a2b_base64( + bytes, + binascii::Base64DecodeOptions { + strict_mode, + padded: padded.into_bool(), + table: &table, + ignored: &ignored, + canonical, + }, + ) + }) .map_err(|e| new_binascii_error(e, vm)) + })(); + // Match Argument Clinic cleanup order, including failures in __bool__. + drop(data); + drop(ignorechars); + result + } + + #[derive(FromArgs)] + struct B2aBase64Args { + #[pyarg(positional)] + data: ArgBytesLike, + #[pyarg(named, default = ArgIntoBool::TRUE, py_default = "True")] + padded: ArgIntoBool, + #[pyarg(named, default = 0)] + wrapcol: usize, + #[pyarg(named, default = ArgIntoBool::TRUE, py_default = "True")] + newline: ArgIntoBool, + #[pyarg(named, optional, py_default = "BASE64_ALPHABET")] + alphabet: OptionalArg, } #[pyfunction] - fn b2a_base64(data: ArgBytesLike, NewlineArg { newline }: NewlineArg) -> Vec { - data.with_ref(|bytes| binascii::b2a_base64(bytes, newline)) + fn b2a_base64(args: B2aBase64Args, vm: &VirtualMachine) -> PyResult> { + let B2aBase64Args { + data, + padded, + wrapcol, + newline, + alphabet, + } = args; + let copy_alphabet = |bytes: &[u8]| -> PyResult<[u8; 64]> { + bytes + .try_into() + .map_err(|_| vm.new_value_error("alphabet must have length 64")) + }; + let alphabet_bytes = match &alphabet { + OptionalArg::Present(value) => value.with_ref(copy_alphabet), + OptionalArg::Missing => Ok(*binascii::BASE64_ALPHABET), + }; + let result = alphabet_bytes.and_then(|alphabet_bytes| { + data.with_ref(|bytes| { + binascii::b2a_base64( + bytes, + padded.into_bool(), + wrapcol, + newline.into_bool(), + &alphabet_bytes, + ) + }) + .map_err(|e| new_binascii_error(e, vm)) + }); + drop(data); + drop(alphabet); + result } #[derive(FromArgs)] @@ -182,6 +342,8 @@ fn new_binascii_error(error: Error, vm: &VirtualMachine) -> PyBaseExceptionRef { Error::TrailingGarbage => "Trailing garbage".to_owned(), Error::TooLong => "At most 45 bytes at once".to_owned(), Error::Base64(e) => base64_message(e), + Error::TooMuchBase64Data => "Too much data for base64".to_owned(), + Error::Memory => return vm.no_memory_error(), }; vm.new_exception_msg(decl::error_type(vm), message.into()) } @@ -199,5 +361,7 @@ fn base64_message(error: Base64DecodeError) -> String { "Invalid base64-encoded string: number of data characters ({index}) cannot be 1 more than a multiple of 4" ), Base64DecodeError::IncorrectPadding => "Incorrect padding".to_owned(), + Base64DecodeError::PaddingNotAllowed => "Padding not allowed".to_owned(), + Base64DecodeError::NonZeroPaddingBits => "Non-zero padding bits".to_owned(), } } diff --git a/crates/vm/src/builtins/builtin_func.rs b/crates/vm/src/builtins/builtin_func.rs index e0399987049..8d50d469df5 100644 --- a/crates/vm/src/builtins/builtin_func.rs +++ b/crates/vm/src/builtins/builtin_func.rs @@ -140,7 +140,7 @@ impl Representable for PyNativeFunction { if let Some(bound) = zelf .zelf .as_ref() - .filter(|b| !b.class().is(vm.ctx.types.module_type)) + .filter(|b| !b.class().is_subtype(vm.ctx.types.module_type)) { Ok(format!( "", @@ -169,7 +169,7 @@ impl PyNativeFunction { fn __qualname__(zelf: NativeFunctionOrMethod, vm: &VirtualMachine) -> PyResult { let zelf = zelf.0; let qualname = if let Some(bound) = &zelf.zelf { - if bound.class().is(vm.ctx.types.module_type) { + if bound.class().is_subtype(vm.ctx.types.module_type) { return Ok(vm.ctx.intern_str(zelf.value.name).to_owned()); } let prefix = if bound.class().is(vm.ctx.types.type_type) { diff --git a/crates/vm/src/builtins/module.rs b/crates/vm/src/builtins/module.rs index ac01c901c17..e7f9e2bfc9f 100644 --- a/crates/vm/src/builtins/module.rs +++ b/crates/vm/src/builtins/module.rs @@ -82,7 +82,7 @@ impl PyModuleDef { } } -#[pyclass(module = false, name = "module")] +#[pyclass(module = false, name = "module", traverse = "manual")] // The dict lives in the object extension, not the payload. The offset is // the dict cell at the front of that extension. #[pymember(name = "__dict__", offset = ::rustpython_vm::object::dict_member_offset())] @@ -90,12 +90,26 @@ impl PyModuleDef { pub struct PyModule { // PyObject *md_dict; pub def: Option<&'static PyModuleDef>, - // state: Any + native_state: crate::object::PyAtomicRef>, // weaklist // for logging purposes after md_dict is cleared pub name: Option<&'static PyStrInterned>, } +// SAFETY: Native state is the only owned Python reference in the payload. +// The module dictionary is visited separately by the object extension. +unsafe impl crate::object::Traverse for PyModule { + fn traverse(&self, tracer: &mut crate::object::TraverseFn<'_>) { + self.native_state.traverse(tracer); + } + + fn clear(&mut self, out: &mut Vec) { + if let Some(state) = self.native_state.store(None) { + out.push(state); + } + } +} + impl PyPayload for PyModule { #[inline] fn class(ctx: &Context) -> &'static Py { @@ -119,6 +133,7 @@ impl PyModule { pub const fn new() -> Self { Self { def: None, + native_state: crate::object::PyAtomicRef::new_empty(), name: None, } } @@ -127,10 +142,22 @@ impl PyModule { pub const fn from_def(def: &'static PyModuleDef) -> Self { Self { def: Some(def), + native_state: crate::object::PyAtomicRef::new_empty(), name: Some(def.name), } } + /// Interpreter-local native state, retained and traversed with the module. + #[must_use] + pub fn native_state(&self) -> Option { + self.native_state.load_owned() + } + + /// Replace native state without exposing it in the Python module dictionary. + pub fn set_native_state(&self, state: PyObjectRef) { + drop(self.native_state.store(Some(state))); + } + pub fn __init_dict_from_def(vm: &VirtualMachine, module: &Py) { let doc = module.def.unwrap().doc.map(|doc| doc.to_owned()); module.init_dict(module.name.unwrap(), doc, vm); diff --git a/extra_tests/snippets/stdlib_binascii_base64.py b/extra_tests/snippets/stdlib_binascii_base64.py new file mode 100644 index 00000000000..4daac629a14 --- /dev/null +++ b/extra_tests/snippets/stdlib_binascii_base64.py @@ -0,0 +1,363 @@ +"""Bounded CPython 3.15 base64 API regressions (no other binascii codecs).""" + +import binascii + +from testutils import assert_raises + +encode = binascii.b2a_base64 +decode = binascii.a2b_base64 +alphabet = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" +assert binascii.BASE64_ALPHABET == alphabet +assert binascii.URLSAFE_BASE64_ALPHABET == alphabet[:-2] + b"-_" + +# RFC 4648 vectors and CPython's four-column wrapping normalization. +for raw, expected in ( + (b"", b""), + (b"f", b"Zg=="), + (b"fo", b"Zm8="), + (b"foo", b"Zm9v"), + (b"foob", b"Zm9vYg=="), + (b"fooba", b"Zm9vYmE="), + (b"foobar", b"Zm9vYmFy"), + (b"www.python.org", b"d3d3LnB5dGhvbi5vcmc="), +): + for factory in bytes, bytearray, memoryview: + for padded in False, True: + unwrapped = expected if padded else expected.rstrip(b"=") + for width in 0, 1, 3, 4, 5, 7, 8, 11, 76: + normalized = max(4, width // 4 * 4) if width else 1000 + wrapped = b"\n".join( + unwrapped[i : i + normalized] + for i in range(0, len(unwrapped), normalized) + ) + for newline in False, True: + result = encode( + factory(raw), padded=padded, wrapcol=width, newline=newline + ) + assert result == wrapped + (b"\n" if newline else b"") + assert ( + decode( + factory(result), + padded=padded, + ignorechars=b"\n", + canonical=True, + ) + == raw + ) + +# Input and custom alphabet asymmetry: decoding accepts ASCII str, but its +# alphabet must be bytes. Encoders accept any contiguous bytes-like alphabet. +raw = bytes(range(64)) +standard = encode(raw, newline=False) +for custom in alphabet[::-1], alphabet[:-2] + b"-_", bytes(range(128, 192)): + expected = standard.translate(bytes.maketrans(alphabet, custom)) + for factory in bytes, bytearray, memoryview: + assert encode(raw, alphabet=factory(custom), newline=False) == expected + assert decode(expected, alphabet=custom, canonical=True) == raw +assert decode("Zm9v", alphabet=alphabet) == b"foo" +assert encode(b"\0", alphabet=b"A" * 64, newline=False) == b"AA==" +assert decode(b"AA==", alphabet=b"A" * 64) == b"\xff" +assert encode(b"\0", alphabet=b"=" * 64, newline=False) == b"====" +assert decode(b"====", alphabet=b"=" * 64) == b"" + +# Supplied ignorechars enables strictness by default; alphabet characters are +# never ignored. Explicit non-strict mode still ignores arbitrary non-alphabet. +assert decode(b" Zm9v!", strict_mode=False, ignorechars=b"") == b"foo" +assert decode(b" Zm9v!", ignorechars=b" !") == b"foo" +assert decode(b"Zm9v", ignorechars=b"Zm9v") == b"foo" +for value in b" ", b"!", b"\xff": + assert decode(value) == b"" + with assert_raises(binascii.Error): + decode(value, ignorechars=b"") + +for value, message in ( + (b"=", "Leading padding not allowed"), + (b" =Zm9v", "Leading padding not allowed"), + (b"Zm9v=", "Excess padding not allowed"), + (b"Zg===", "Excess padding not allowed"), + (b"Zg=a", "Discontinuous padding not allowed"), + (b"Zg==a", "Excess data after padding"), + (b"Zg", "Incorrect padding"), + (b"!", "Only base64 data is allowed"), +): + try: + decode(value, ignorechars=b" ") + except binascii.Error as error: + assert str(error) == message + else: + assert False, (value, message) + +for value, expected in (b"YQ==", b"a"), (b"YWI=", b"ab"), (b"YWJj", b"abc"): + assert decode(value.replace(b"=", b""), padded=False, strict_mode=True) == expected + assert decode(value, padded=False, ignorechars=b"=") == expected + if b"=" in value: + try: + decode(value, padded=False, strict_mode=True) + except binascii.Error as error: + assert str(error) == "Padding not allowed" + else: + assert False +assert decode(b"Y=WJj", ignorechars=b"=") == b"abc" +assert decode(b"YQ==YQ==") == b"a\x06\x10" +for value in b"AB==", b"AP==", b"AAB=", b"AAD=": + decode(value) + for padded in False, True: + try: + decode( + value if padded else value.rstrip(b"="), padded=padded, canonical=True + ) + except binascii.Error as error: + assert str(error) == "Non-zero padding bits" + else: + assert False +for value in b"A", b"A=", b"AAAAA": + with assert_raises(binascii.Error): + decode(value, padded=False) + +for function in encode, decode: + for invalid in b"", alphabet[:-1], alphabet + b"?": + with assert_raises(ValueError): + function(b"", alphabet=invalid) + for invalid in None, "", []: + with assert_raises(TypeError): + function(b"", alphabet=invalid) + with assert_raises(TypeError): + function(data=b"") + with assert_raises(TypeError): + function(b"", False) +for invalid in bytearray(alphabet), memoryview(alphabet): + with assert_raises(TypeError): + decode(b"", alphabet=invalid) +for invalid in None, "", []: + with assert_raises(TypeError): + decode(b"", ignorechars=invalid) +with assert_raises(ValueError): + encode(b"", wrapcol=-1) +with assert_raises(OverflowError): + encode(b"", wrapcol=1 << 128) +with assert_raises(TypeError): + encode(b"", wrapcol=4.0) +with assert_raises(ValueError): + decode("é") + + +class Width: + def __index__(self): + return 4 + + +assert encode(b"foobar", wrapcol=Width(), newline=False) == b"Zm9v\nYmFy" + +# Conversion order is independent of keyword order and canonical runs after +# ignorechars acquisition, before alphabet length validation. +events = [] + + +class Flag: + def __init__(self, name, value): + self.name = name + self.value = value + + def __bool__(self): + events.append(self.name) + return self.value + + +assert ( + encode(b"f", newline=Flag("newline", False), padded=Flag("padded", False)) == b"Zg" +) +assert events == ["padded", "newline"] +events.clear() +assert ( + decode( + b"Zg", + canonical=Flag("canonical", True), + padded=Flag("padded", False), + strict_mode=Flag("strict", True), + ) + == b"f" +) +assert events == ["strict", "padded", "canonical"] +events.clear() +with assert_raises(ValueError): + decode(b"", alphabet=b"", canonical=Flag("canonical", False)) +assert events == ["canonical"] + + +# The decoder cache uses the original bytes object, including subclass hooks. +class CachedAlphabet(bytes): + def __hash__(self): + events.append("hash") + return 314159265 + + def __eq__(self, other): + events.append("equal") + return bytes.__eq__(self, other) + + +custom = CachedAlphabet(alphabet[1:] + alphabet[:1]) +encoded = encode(b"abc", alphabet=custom, newline=False) +events.clear() +assert decode(encoded, alphabet=custom) == b"abc" +assert events == ["hash", "hash"] +events.clear() +assert decode(encoded, alphabet=custom) == b"abc" +assert events == ["hash"] +events.clear() +assert decode(encoded, alphabet=CachedAlphabet(custom)) == b"abc" +assert events == ["hash", "equal"] + + +class UnhashableAlphabet(bytes): + __hash__ = None + + +with assert_raises(TypeError): + decode(b"", alphabet=UnhashableAlphabet(alphabet)) +with assert_raises(ValueError): + decode(b"", alphabet=UnhashableAlphabet(b"")) +assert encode(b"abc", alphabet=UnhashableAlphabet(alphabet)) == b"YWJj\n" + + +class HashError(Exception): + pass + + +class BrokenAlphabet(bytes): + def __hash__(self): + raise HashError + + +with assert_raises(HashError): + decode(b"!", alphabet=BrokenAlphabet(alphabet), strict_mode=True) + +# Keep the exact owning module alive through a saved builtin, independently of +# sys.modules. A fresh module's cache cycle must remain visible to the collector. +import gc +import importlib.util +import sys +import weakref + +assert decode.__self__ is binascii +assert decode.__module__ == "binascii" +assert decode.__name__ == "a2b_base64" +original = sys.modules.pop("binascii") +try: + assert decode(encoded, alphabet=custom) == b"abc" +finally: + sys.modules["binascii"] = original + +spec = importlib.util.find_spec("binascii") +original = sys.modules.pop("binascii") +try: + isolated = importlib.util.module_from_spec(spec) + spec.loader.exec_module(isolated) +finally: + sys.modules["binascii"] = original +assert isolated is not binascii +assert isolated.a2b_base64.__self__ is isolated +cycle_key = CachedAlphabet(alphabet) +cycle_key.owner = isolated +assert isolated.a2b_base64(b"YWJj", alphabet=cycle_key) == b"abc" +module_ref = weakref.ref(isolated) +del cycle_key, isolated +for _ in range(3): + gc.collect() +assert module_ref() is None +assert not hasattr(binascii, "_reverse_table_cache") + + +# Hash callbacks can safely reenter the same module before decoding begins. +class ReentrantAlphabet(bytes): + def __hash__(self): + assert decode(b"Zm9v") == b"foo" + assert encode(b"foo", wrapcol=4) == b"Zm9v\n" + events.append("reenter") + return 271828182 + + +events.clear() +assert decode(b"YWJj", alphabet=ReentrantAlphabet(alphabet)) == b"abc" +assert events == ["reenter", "reenter"] +assert "ignorechars=" in decode.__text_signature__ +assert "wrapcol=0" in encode.__text_signature__ + + +# Acquired input and auxiliary exports survive conversions and cache callbacks, +# and are released in Clinic order, including a canonical conversion failure. +class Export: + def __init__(self, name, data): + self.name = name + self.data = data + + def __buffer__(self, flags): + events.append("acquire " + self.name) + return memoryview(self.data) + + def __release_buffer__(self, view): + events.append("release " + self.name) + + +class BrokenFlag: + def __bool__(self): + events.append("canonical error") + raise HashError + + +events.clear() +assert ( + encode(Export("data", b"foo"), alphabet=Export("alphabet", alphabet)) == b"Zm9v\n" +) +assert events == [ + "acquire data", + "acquire alphabet", + "release data", + "release alphabet", +] +events.clear() +assert decode(Export("data", b"Zm9v"), ignorechars=Export("ignored", b"")) == b"foo" +assert events == ["acquire data", "acquire ignored", "release data", "release ignored"] +events.clear() +with assert_raises(HashError): + decode( + Export("data", b""), ignorechars=Export("ignored", b""), canonical=BrokenFlag() + ) +assert events == [ + "acquire data", + "acquire ignored", + "canonical error", + "release data", + "release ignored", +] + +# Aliased buffer exporters do not require nested backing-store locks. +shared = bytearray(alphabet) +assert encode(shared, alphabet=shared) == encode(bytes(shared), alphabet=bytes(shared)) +shared = bytearray(b"Zm9v") +assert decode(shared, ignorechars=shared) == b"foo" + +for length in (0, 1, 2, 3, 4, 5, 56, 57, 58, 63, 64, 65, 255, 256, 257): + payload = (bytes(range(256)) * 2)[:length] + for padded in False, True: + for width in 0, 4, 76: + encoded = encode(payload, padded=padded, wrapcol=width) + assert ( + decode(encoded, padded=padded, ignorechars=b"\n", canonical=True) + == payload + ) + + +# Bound module functions retain their metadata for module subclasses too. +class ModuleSubclass(type(binascii)): + pass + + +original_class = binascii.__class__ +try: + binascii.__class__ = ModuleSubclass + assert repr(decode) == "" + assert decode.__qualname__ == decode.__name__ == "a2b_base64" + assert decode.__self__ is binascii + assert decode.__reduce__() == "a2b_base64" +finally: + binascii.__class__ = original_class From 9d459d2b2d8d51e0bf949d6ebc23c1fb03f56558 Mon Sep 17 00:00:00 2001 From: dot Date: Mon, 12 Oct 2026 14:42:42 +0900 Subject: [PATCH 2/2] Complete binascii codecs for canonical Python 3.15 tests Migrate Lib/test/test_binascii.py unchanged from CPython 3.15.0 final using scripts/update_lib and replace the superseded custom base64 snippet. Implement the base32, Base85, Ascii85 and hex ignorechars APIs required by the canonical suite. Share the existing module-bound original-key alphabet cache across base32/base64/base85 and preserve argument and buffer cleanup semantics. Keep the existing base64 kernels and VM state integration unchanged. Assisted-by: OpenAI Codex:model identifier unavailable --- Lib/test/test_binascii.py | 1181 ++++++++++++++++- crates/stdlib/src/binascii.rs | 400 +++++- crates/stdlib/src/binascii/base_n.rs | 431 ++++++ .../snippets/stdlib_binascii_base64.py | 363 ----- 4 files changed, 1955 insertions(+), 420 deletions(-) create mode 100644 crates/stdlib/src/binascii/base_n.rs delete mode 100644 extra_tests/snippets/stdlib_binascii_base64.py diff --git a/Lib/test/test_binascii.py b/Lib/test/test_binascii.py index 82eabf4bb06..cedbdc61f18 100644 --- a/Lib/test/test_binascii.py +++ b/Lib/test/test_binascii.py @@ -4,14 +4,17 @@ import binascii import array import re -from test.support import bigmemtest, _1G, _4G +import sys +from test.support import bigmemtest, _1G, _4G, check_impl_detail from test.support.hypothesis_helper import hypothesis # Note: "*_hex" functions are aliases for "(un)hexlify" -b2a_functions = ['b2a_base64', 'b2a_hex', 'b2a_qp', 'b2a_uu', +b2a_functions = ['b2a_ascii85', 'b2a_base32', 'b2a_base64', 'b2a_base85', + 'b2a_hex', 'b2a_qp', 'b2a_uu', 'hexlify'] -a2b_functions = ['a2b_base64', 'a2b_hex', 'a2b_qp', 'a2b_uu', +a2b_functions = ['a2b_ascii85', 'a2b_base32', 'a2b_base64', 'a2b_base85', + 'a2b_hex', 'a2b_qp', 'a2b_uu', 'unhexlify'] all_functions = a2b_functions + b2a_functions + ['crc32', 'crc_hqx'] @@ -23,6 +26,8 @@ class BinASCIITest(unittest.TestCase): rawdata = b"The quick brown fox jumps over the lazy dog.\r\n" # Be slow so we don't depend on other modules rawdata += bytes(range(256)) + rawdata += b'\0'*32 + rawdata += b' '*32 rawdata += b"\r\nHello world.\n" def setUp(self): @@ -41,6 +46,40 @@ def test_exceptions(self): self.assertIsSubclass(binascii.Error, Exception) self.assertIsSubclass(binascii.Incomplete, Exception) + def test_constants(self): + self.assertEqual(binascii.BASE64_ALPHABET, + b'ABCDEFGHIJKLMNOPQRSTUVWXYZ' + b'abcdefghijklmnopqrstuvwxyz' + b'0123456789+/') + self.assertEqual(binascii.URLSAFE_BASE64_ALPHABET, + binascii.BASE64_ALPHABET[:-2] + b'-_') + self.assertEqual(binascii.UU_ALPHABET, bytes(range(32, 32+64))) + self.assertEqual(binascii.CRYPT_ALPHABET, + b'./0123456789' + b'ABCDEFGHIJKLMNOPQRSTUVWXYZ' + b'abcdefghijklmnopqrstuvwxyz') + self.assertEqual(binascii.BINHEX_ALPHABET, + b'!"#$%&\'()*+,-012345689' + b'@ABCDEFGHIJKLMNPQRSTUVXYZ[' + b'`abcdefhijklmpqr') + + self.assertEqual(binascii.BASE85_ALPHABET, + b'0123456789' + b'ABCDEFGHIJKLMNOPQRSTUVWXYZ' + b'abcdefghijklmnopqrstuvwxyz' + b'!#$%&()*+-;<=>?@^_`{|}~') + self.assertEqual(binascii.ASCII85_ALPHABET, bytes(range(33, 33+85))) + self.assertEqual(binascii.Z85_ALPHABET, + b'0123456789' + b'abcdefghijklmnopqrstuvwxyz' + b'ABCDEFGHIJKLMNOPQRSTUVWXYZ' + b'.-:+=^!/*?&<>()[]{}@%$#') + + self.assertEqual(binascii.BASE32_ALPHABET, + b'ABCDEFGHIJKLMNOPQRSTUVWXYZ234567') + self.assertEqual(binascii.BASE32HEX_ALPHABET, + b'0123456789ABCDEFGHIJKLMNOPQRSTUV') + def test_functions(self): # Check presence of all functions for name in all_functions: @@ -117,62 +156,170 @@ def addnoise(line): # empty strings. TBD: shouldn't it raise an exception instead ? self.assertEqual(binascii.a2b_base64(self.type2test(fillers)), b'') - def test_base64_strict_mode(self): - # Test base64 with strict mode on - def _assertRegexTemplate(assert_regex: str, data: bytes, non_strict_mode_expected_result: bytes): + def test_base64_bad_padding(self): + # Test malformed padding + def _assertRegexTemplate(assert_regex, data, + non_strict_mode_expected_result, **kwargs): + data = self.type2test(data) with self.assertRaisesRegex(binascii.Error, assert_regex): - binascii.a2b_base64(self.type2test(data), strict_mode=True) - self.assertEqual(binascii.a2b_base64(self.type2test(data), strict_mode=False), + binascii.a2b_base64(data, strict_mode=True, **kwargs) + self.assertEqual(binascii.a2b_base64(data, strict_mode=False), non_strict_mode_expected_result) - self.assertEqual(binascii.a2b_base64(self.type2test(data)), + self.assertEqual(binascii.a2b_base64(data, strict_mode=True, + ignorechars=b' ='), + non_strict_mode_expected_result) + self.assertEqual(binascii.a2b_base64(data), non_strict_mode_expected_result) - def assertExcessData(data, non_strict_mode_expected_result: bytes): - _assertRegexTemplate(r'(?i)Excess data', data, non_strict_mode_expected_result) - - def assertNonBase64Data(data, non_strict_mode_expected_result: bytes): - _assertRegexTemplate(r'(?i)Only base64 data', data, non_strict_mode_expected_result) + def assertLeadingPadding(*args, **kwargs): + _assertRegexTemplate(r'(?i)Leading padding', *args, **kwargs) - def assertLeadingPadding(data, non_strict_mode_expected_result: bytes): - _assertRegexTemplate(r'(?i)Leading padding', data, non_strict_mode_expected_result) + def assertDiscontinuousPadding(*args): + _assertRegexTemplate(r'(?i)Discontinuous padding', *args) - def assertDiscontinuousPadding(data, non_strict_mode_expected_result: bytes): - _assertRegexTemplate(r'(?i)Discontinuous padding', data, non_strict_mode_expected_result) + def assertExcessPadding(*args): + _assertRegexTemplate(r'(?i)Excess padding', *args) - def assertExcessPadding(data, non_strict_mode_expected_result: bytes): - _assertRegexTemplate(r'(?i)Excess padding', data, non_strict_mode_expected_result) + def assertInvalidLength(data, *args, length=None, **kwargs): + if length is None: + length = len(data.split(b'=', 1)[0].replace(b' ', b'')) + assert_regex = fr"(?i)Invalid.+number of data characters \({length}\)" + _assertRegexTemplate(assert_regex, data, *args, **kwargs) - # Test excess data exceptions assertExcessPadding(b'ab===', b'i') assertExcessPadding(b'ab====', b'i') - assertNonBase64Data(b'ab==:', b'i') - assertExcessData(b'abc=a', b'i\xb7\x1a') - assertNonBase64Data(b'abc=:', b'i\xb7') - assertNonBase64Data(b'ab==\n', b'i') assertExcessPadding(b'abc==', b'i\xb7') assertExcessPadding(b'abc===', b'i\xb7') assertExcessPadding(b'abc====', b'i\xb7') assertExcessPadding(b'abc=====', b'i\xb7') - # Test non-base64 data exceptions - assertNonBase64Data(b'\nab==', b'i') - assertNonBase64Data(b'ab:(){:|:&};:==', b'i') - assertNonBase64Data(b'a\nb==', b'i') - assertNonBase64Data(b'a\x00b==', b'i') - - # Test malformed padding assertLeadingPadding(b'=', b'') assertLeadingPadding(b'==', b'') assertLeadingPadding(b'===', b'') assertLeadingPadding(b'====', b'') assertLeadingPadding(b'=====', b'') + assertLeadingPadding(b'=abcd', b'i\xb7\x1d') + assertLeadingPadding(b'==abcd', b'i\xb7\x1d') + assertLeadingPadding(b'===abcd', b'i\xb7\x1d') + assertLeadingPadding(b'====abcd', b'i\xb7\x1d') + assertLeadingPadding(b'=====abcd', b'i\xb7\x1d') + assertLeadingPadding(b' =abcd', b'i\xb7\x1d', ignorechars=b' ') + + assertInvalidLength(b'a=b==', b'i') + assertInvalidLength(b'a=bc=', b'i\xb7') + assertInvalidLength(b'a=bc==', b'i\xb7') + assertInvalidLength(b'a=bcd', b'i\xb7\x1d') + assertInvalidLength(b'a=bcd=', b'i\xb7\x1d') + assertInvalidLength(b' a=b==', b'i', ignorechars=b' ') + assertInvalidLength(b'abcde=f==', b'i\xb7\x1dy') + assertInvalidLength(b' abcde=f==', b'i\xb7\x1dy', ignorechars=b' ') + assertDiscontinuousPadding(b'ab=c=', b'i\xb7') - assertDiscontinuousPadding(b'ab=ab==', b'i\xb6\x9b') + assertDiscontinuousPadding(b'ab=cd', b'i\xb7\x1d') + assertDiscontinuousPadding(b'ab=cd==', b'i\xb7\x1d') + assertExcessPadding(b'abcd=', b'i\xb7\x1d') assertExcessPadding(b'abcd==', b'i\xb7\x1d') assertExcessPadding(b'abcd===', b'i\xb7\x1d') assertExcessPadding(b'abcd====', b'i\xb7\x1d') assertExcessPadding(b'abcd=====', b'i\xb7\x1d') + assertExcessPadding(b'abcd==', b'i\xb7\x1d') + assertExcessPadding(b'abcd===', b'i\xb7\x1d') + assertExcessPadding(b'abcd====', b'i\xb7\x1d') + assertExcessPadding(b'abcd=====', b'i\xb7\x1d') + assertExcessPadding(b'abcd=efgh', b'i\xb7\x1dy\xf8!') + assertExcessPadding(b'abcd==efgh', b'i\xb7\x1dy\xf8!') + assertExcessPadding(b'abcd===efgh', b'i\xb7\x1dy\xf8!') + assertExcessPadding(b'abcd====efgh', b'i\xb7\x1dy\xf8!') + assertExcessPadding(b'abcd=====efgh', b'i\xb7\x1dy\xf8!') + + def test_a2b_base64_padded(self): + a2b_base64 = binascii.a2b_base64 + t = self.type2test + def check(data, expected): + if b'=' in data: + with self.assertRaisesRegex(binascii.Error, 'Padding not allowed'): + a2b_base64(t(data), padded=False, strict_mode=True) + self.assertEqual(a2b_base64(t(data), padded=False, ignorechars=b'='), + expected) + data = data.replace(b'=', b'') + self.assertEqual(a2b_base64(t(data), padded=False), expected) + self.assertEqual(a2b_base64(t(data), padded=False, strict_mode=True), + expected) + + check(b'', b'') + check(b'YQ==', b'a') + check(b'YWI=', b'ab') + check(b'YWJj', b'abc') + check(b'Y=WJj', b'abc') + check(b'YW=Jj', b'abc') + check(b'YWJ=j', b'abc') + + def _common_test_ignorechars(self, func): + eq = self.assertEqual + empty = self.type2test(b'') + data = self.type2test(b'\n \n') + ignorechars = self.type2test(b' \n') + eq(func(empty, ignorechars=ignorechars), b'') + eq(func(empty, ignorechars=empty), b'') + eq(func(data, ignorechars=ignorechars), b'') + with self.assertRaises(binascii.Error): + func(data, ignorechars=empty) + with self.assertRaises(binascii.Error): + func(data, ignorechars=ignorechars[1:]) + with self.assertRaises(binascii.Error): + func(data, ignorechars=ignorechars[:-1]) + with self.assertRaises(TypeError): + func(empty, ignorechars='') + with self.assertRaises(TypeError): + func(empty, ignorechars=[]) + with self.assertRaises(TypeError): + func(empty, ignorechars=None) + + def test_base64_invalidchars(self): + # Test non-base64 data exceptions + self._common_test_ignorechars(binascii.a2b_base64) + def assertNonBase64Data(data, expected, ignorechars): + data = self.type2test(data) + assert_regex = r'(?i)Only base64 data' + self.assertEqual(binascii.a2b_base64(data), expected) + with self.assertRaisesRegex(binascii.Error, assert_regex): + binascii.a2b_base64(data, strict_mode=True) + with self.assertRaisesRegex(binascii.Error, assert_regex): + binascii.a2b_base64(data, ignorechars=b'') + self.assertEqual(binascii.a2b_base64(data, ignorechars=ignorechars), + expected) + self.assertEqual(binascii.a2b_base64(data, strict_mode=False, ignorechars=b''), + expected) + + assertNonBase64Data(b'\nab==', b'i', ignorechars=b'\n') + assertNonBase64Data(b'ab:(){:|:&};:==', b'i', ignorechars=b':;(){}|&') + assertNonBase64Data(b'a\nb==', b'i', ignorechars=b'\n') + assertNonBase64Data(b'a\x00b==', b'i', ignorechars=b'\x00') + assertNonBase64Data(b'ab:==', b'i', ignorechars=b':') + assertNonBase64Data(b'ab=:=', b'i', ignorechars=b':') + assertNonBase64Data(b'ab==:', b'i', ignorechars=b':') + assertNonBase64Data(b'abc=:', b'i\xb7', ignorechars=b':') + assertNonBase64Data(b'ab==\n', b'i', ignorechars=b'\n') + assertNonBase64Data(b'a\nb==', b'i', ignorechars=bytearray(b'\n')) + assertNonBase64Data(b'a\nb==', b'i', ignorechars=memoryview(b'\n')) + + self.assertEqual(binascii.a2b_base64(b'+A-/B_', ignorechars=b'+/-_'), + b'\xf8\x0f\xc1') + self.assertEqual(binascii.a2b_base64(b'+A-/B_', ignorechars=b'+/-_', + alphabet=binascii.URLSAFE_BASE64_ALPHABET), + b'\x03\xe0\x7f') + + # Same cell in the cache: '\r' >> 3 == '\n' >> 3. + data = self.type2test(b'\r\n') + with self.assertRaises(binascii.Error): + binascii.a2b_base64(data, ignorechars=b'\r') + self.assertEqual(binascii.a2b_base64(data, ignorechars=b'\r\n'), b'') + # Same bit mask in the cache: '*' & 31 == '\n' & 31. + data = self.type2test(b'*\n') + with self.assertRaises(binascii.Error): + binascii.a2b_base64(data, ignorechars=b'*') + self.assertEqual(binascii.a2b_base64(data, ignorechars=b'*\n'), b'') def test_base64_excess_data(self): # Test excess data exceptions @@ -183,6 +330,9 @@ def assertExcessData(data, expected): binascii.a2b_base64(data, strict_mode=True) self.assertEqual(binascii.a2b_base64(data, strict_mode=False), expected) + self.assertEqual(binascii.a2b_base64(data, strict_mode=True, + ignorechars=b'='), + expected) self.assertEqual(binascii.a2b_base64(data), expected) assertExcessData(b'ab==c=', b'i\xb7') @@ -191,9 +341,15 @@ def assertExcessData(data, expected): def test_base64errors(self): # Test base64 with invalid padding - def assertIncorrectPadding(data): + def assertIncorrectPadding(data, strict_mode=True): + data = self.type2test(data) + with self.assertRaisesRegex(binascii.Error, r'(?i)Incorrect padding'): + binascii.a2b_base64(data) with self.assertRaisesRegex(binascii.Error, r'(?i)Incorrect padding'): - binascii.a2b_base64(self.type2test(data)) + binascii.a2b_base64(data, strict_mode=False) + if strict_mode: + with self.assertRaisesRegex(binascii.Error, r'(?i)Incorrect padding'): + binascii.a2b_base64(data, strict_mode=True) assertIncorrectPadding(b'ab') assertIncorrectPadding(b'ab=') @@ -201,16 +357,22 @@ def assertIncorrectPadding(data): assertIncorrectPadding(b'abcdef') assertIncorrectPadding(b'abcdef=') assertIncorrectPadding(b'abcdefg') - assertIncorrectPadding(b'a=b=') - assertIncorrectPadding(b'a\nb=') + assertIncorrectPadding(b'a=b=', strict_mode=False) + assertIncorrectPadding(b'a\nb=', strict_mode=False) # Test base64 with invalid number of valid characters (1 mod 4) - def assertInvalidLength(data): + def assertInvalidLength(data, strict_mode=True): n_data_chars = len(re.sub(br'[^A-Za-z0-9/+]', br'', data)) + data = self.type2test(data) expected_errmsg_re = \ r'(?i)Invalid.+number of data characters.+' + str(n_data_chars) with self.assertRaisesRegex(binascii.Error, expected_errmsg_re): - binascii.a2b_base64(self.type2test(data)) + binascii.a2b_base64(data) + with self.assertRaisesRegex(binascii.Error, expected_errmsg_re): + binascii.a2b_base64(data, strict_mode=False) + if strict_mode: + with self.assertRaisesRegex(binascii.Error, expected_errmsg_re): + binascii.a2b_base64(data, strict_mode=True) assertInvalidLength(b'a') assertInvalidLength(b'a=') @@ -218,7 +380,911 @@ def assertInvalidLength(data): assertInvalidLength(b'a===') assertInvalidLength(b'a' * 5) assertInvalidLength(b'a' * (4 * 87 + 1)) - assertInvalidLength(b'A\tB\nC ??DE') # only 5 valid characters + assertInvalidLength(b'A\tB\nC ??DE', # only 5 valid characters + strict_mode=False) + + def test_base64_canonical(self): + # https://datatracker.ietf.org/doc/html/rfc4648.html#section-3.5 + # Decoders MAY reject encoded data if the pad bits are not zero. + + # Without canonical=True, non-zero padding bits are accepted + self.assertEqual(binascii.a2b_base64(self.type2test(b'AB==')), b'\x00') + self.assertEqual(binascii.a2b_base64(self.type2test(b'AB=='), + strict_mode=True), b'\x00') + + # 2 data chars + "==": last char has 4 padding bits + # 'A' = 0, 'B' = 1 -> leftover 0001 (non-zero) + with self.assertRaises(binascii.Error): + binascii.a2b_base64(self.type2test(b'AB=='), canonical=True) + # 'A' = 0, 'P' = 15 -> leftover 1111 (non-zero) + with self.assertRaises(binascii.Error): + binascii.a2b_base64(self.type2test(b'AP=='), canonical=True) + + # 3 data chars + "=": last char has 2 padding bits + # 'A' = 0, 'A' = 0, 'B' = 1 -> leftover 01 (non-zero) + with self.assertRaises(binascii.Error): + binascii.a2b_base64(self.type2test(b'AAB='), canonical=True) + # 'A' = 0, 'A' = 0, 'D' = 3 -> leftover 11 (non-zero) + with self.assertRaises(binascii.Error): + binascii.a2b_base64(self.type2test(b'AAD='), canonical=True) + + # Verify that zero padding bits are accepted + binascii.a2b_base64(self.type2test(b'AA=='), canonical=True) + binascii.a2b_base64(self.type2test(b'AAA='), canonical=True) + + # Full quads with no padding have no leftover bits -- always valid + binascii.a2b_base64(self.type2test(b'AAAA'), canonical=True) + + @hypothesis.given(payload=hypothesis.strategies.binary()) + @hypothesis.example(b'') + @hypothesis.example(b'\x00') + @hypothesis.example(b'\xff\xff') + @hypothesis.example(b'abc') + def test_base64_canonical_roundtrip(self, payload): + # The encoder must always produce canonical output. + encoded = binascii.b2a_base64(payload, newline=False) + decoded = binascii.a2b_base64(encoded, canonical=True) + self.assertEqual(decoded, payload) + + def test_base64_alphabet(self): + alphabet = (b'!"#$%&\'()*+,-012345689@' + b'ABCDEFGHIJKLMNPQRSTUVXYZ[`abcdefhijklmpqr') + data = self.type2test(self.rawdata) + encoded = binascii.b2a_base64(data, alphabet=alphabet) + trans = bytes.maketrans(binascii.BASE64_ALPHABET, alphabet) + expected = binascii.b2a_base64(data).translate(trans) + self.assertEqual(encoded, expected) + self.assertEqual(binascii.a2b_base64(encoded, alphabet=alphabet), self.rawdata) + self.assertEqual(binascii.b2a_base64(data, alphabet=self.type2test(alphabet)), expected) + + data = self.type2test(b'') + self.assertEqual(binascii.b2a_base64(data, alphabet=alphabet), b'\n') + self.assertEqual(binascii.a2b_base64(data, alphabet=alphabet), b'') + + for func in binascii.b2a_base64, binascii.a2b_base64: + with self.assertRaises(TypeError): + func(data, alphabet=None) + with self.assertRaises(TypeError): + func(data, alphabet=alphabet.decode()) + with self.assertRaises(ValueError): + func(data, alphabet=alphabet[:-1]) + with self.assertRaises(ValueError): + func(data, alphabet=alphabet+b'?') + with self.assertRaises(TypeError): + binascii.a2b_base64(data, alphabet=bytearray(alphabet)) + + def test_ascii85_valid(self): + # Test Ascii85 with valid data + ASCII85_PREFIX = b"<~" + ASCII85_SUFFIX = b"~>" + + # Interleave blocks of 4 null bytes and 4 spaces into test data + rawdata = bytearray() + rawlines, i = [], 0 + for k in range(1, len(self.rawdata) + 1): + b = b"\0\0\0\0" if k & 1 else b" " + b = b + self.rawdata[i:i + k] + b = b" " if k & 1 else b"\0\0\0\0" + rawdata += b + rawlines.append(b) + i += k + if i >= len(self.rawdata): + break + + # Test core parameter combinations + params = (False, False), (False, True), (True, False), (True, True) + for foldspaces, adobe in params: + lines = [] + for rawline in rawlines: + b = self.type2test(rawline) + a = binascii.b2a_ascii85(b, foldspaces=foldspaces, adobe=adobe) + lines.append(a) + res = bytearray() + for line in lines: + a = self.type2test(line) + b = binascii.a2b_ascii85(a, foldspaces=foldspaces, adobe=adobe) + res += b + self.assertEqual(res, rawdata) + + # Inputs with length 1 mod 5 end with a 1-char group, which is + # an encoding violation per the PLRM spec. + error_params = [ + (b"a", False, False, b""), + (b"xbw", False, False, b"wx"), + (b"<~c~>", False, True, b""), + (b"{d ~>", False, True, b" {"), + (b"ye", True, False, b""), + (b"z\x01y\x00f", True, False, b"\x00\x01"), + (b"<~FCfN8yg~>", True, True, b""), + (b"FE;\x03#8zFCf\x02N8yh~>", True, True, b"\x02\x03"), + ] + for a, foldspaces, adobe, ignorechars in error_params: + kwargs = {"foldspaces": foldspaces, "adobe": adobe, "ignorechars": ignorechars} + with self.assertRaises(binascii.Error): + binascii.a2b_ascii85(self.type2test(a), **kwargs) + + def test_ascii85_invalid(self): + # Test Ascii85 with invalid characters interleaved + lines, i = [], 0 + for k in range(1, len(self.rawdata) + 1): + b = self.type2test(self.rawdata[i:i + k]) + a = binascii.b2a_ascii85(b) + lines.append(a) + i += k + if i >= len(self.rawdata): + break + + fillers = bytearray() + valid = b"!\"#$%&'()*+,-./0123456789:;<=>?@" \ + b"ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstu" + b"z" + for i in range(256): + if i not in valid: + fillers.append(i) + def addnoise(line): + res = bytearray() + for i in range(len(line)): + res.append(line[i]) + for j in range(i, len(fillers), len(line)): + res.append(fillers[j]) + return res + res = bytearray() + for line in map(addnoise, lines): + a = self.type2test(line) + b = binascii.a2b_ascii85(a, ignorechars=fillers) + res += b + self.assertEqual(res, self.rawdata) + + # Test Ascii85 with only invalid characters + fillers = self.type2test(fillers) + b = binascii.a2b_ascii85(fillers, ignorechars=fillers) + self.assertEqual(b, b"") + + def test_ascii85_errors(self): + def _assertRegexTemplate(assert_regex, data, **kwargs): + with self.assertRaisesRegex(binascii.Error, assert_regex): + binascii.a2b_ascii85(self.type2test(data), **kwargs) + + def assertMissingDelimiter(data): + _assertRegexTemplate(r"(?i)end with b'~>'", data, adobe=True) + + def assertOverflow(data): + _assertRegexTemplate(r"(?i)Ascii85 overflow", data) + + def assertInvalidSpecial(data): + _assertRegexTemplate(r"(?i)'[yz]'.+5-tuple", data, foldspaces=True) + + def assertInvalidChar(data, **kwargs): + _assertRegexTemplate(r"(?i)Non-Ascii85 digit", data, **kwargs) + + # Test Ascii85 with missing delimiters + assertMissingDelimiter(b"") + assertMissingDelimiter(b"a") + assertMissingDelimiter(b"<~") + assertMissingDelimiter(b"<~!~") + assertMissingDelimiter(b"<~abc>") + assertMissingDelimiter(b"<~has delimiter but not terminal~> !") + + # Test Ascii85 with out-of-range encoded value + assertOverflow(b"t") + assertOverflow(b"s9") + assertOverflow(b"s8X") + assertOverflow(b"s8W.") + assertOverflow(b's8W-"') + assertOverflow(b"s8W-!u") + assertOverflow(b"s8W-!s8W-!zs8X") + + # Test Ascii85 with misplaced short form groups + assertInvalidSpecial(b"ay") + assertInvalidSpecial(b"az") + assertInvalidSpecial(b"aby") + assertInvalidSpecial(b"ayz") + assertInvalidSpecial(b"abcz") + assertInvalidSpecial(b"abcdy") + assertInvalidSpecial(b"y!and!z!then!!y") + + # Test Ascii85 with non-ignored invalid characters + assertInvalidChar(b"j\n") + assertInvalidChar(b" ", ignorechars=b"") + assertInvalidChar(b" valid\x02until\x03", ignorechars=b"\x00\x01\x02\x04") + assertInvalidChar(b"\tFCb", ignorechars=b"\n") + assertInvalidChar(b"xxxB\nP\thU'D v/F+", ignorechars=b" \n\tv") + + def _common_test_wrapcol(self, func): + eq = self.assertEqual + data = self.data + expected = func(data) + eq(func(data, wrapcol=0), expected) + eq(func(data, wrapcol=8000), expected) + eq(func(b'', wrapcol=0), func(b'')) + eq(func(b'', wrapcol=1), func(b'')) + eq(func(data, wrapcol=sys.maxsize), expected) + if check_impl_detail(): + eq(func(data, wrapcol=sys.maxsize*2), expected) + with self.assertRaises(OverflowError): + func(data, wrapcol=2**1000) + with self.assertRaises(ValueError): + func(data, wrapcol=-80) + with self.assertRaises(TypeError): + func(data, wrapcol=80.0) + with self.assertRaises(TypeError): + func(data, wrapcol='80') + with self.assertRaises(TypeError): + func(data, wrapcol=None) + + def test_ascii85_wrapcol(self): + # Test Ascii85 splitting lines + self._common_test_wrapcol(binascii.b2a_ascii85) + + def assertEncode(a_expected, data, n, adobe=False): + b = self.type2test(data) + a = binascii.b2a_ascii85(b, adobe=adobe, wrapcol=n) + self.assertEqual(a, a_expected) + + def assertDecode(data, b_expected, adobe=False): + a = self.type2test(data) + b = binascii.a2b_ascii85(a, adobe=adobe, ignorechars=b"\n") + self.assertEqual(b, b_expected) + + tests = [ + (b"", 0, b"", b"<~~>"), + (b"", 1, b"", b"<~\n~>"), + (b"a", 0, b"@/", b"<~@/~>"), + (b"a", 1, b"@\n/", b"<~\n@/\n~>"), + (b"a", 2, b"@/", b"<~\n@/\n~>"), + (b"a", 3, b"@/", b"<~@\n/~>"), + (b"a", 4, b"@/", b"<~@/\n~>"), + (b"a", 5, b"@/", b"<~@/\n~>"), + (b"a", 6, b"@/", b"<~@/~>"), + (b"a", 7, b"@/", b"<~@/~>"), + (b"a", 123, b"@/", b"<~@/~>"), + (b"this is a test", 7, b"FD,B0+D\nGm>@3BZ\n'F*%", + b"<~FD,B0\n+DGm>@3\nBZ'F*%\n~>"), + (b"a test!!!!!!! ", 11, b"@3BZ'F*&QK+\nX&!P+WqmM+9", + b"<~@3BZ'F*&Q\nK+X&!P+WqmM\n+9~>"), + (b"\0" * 56, 7, b"zzzzzzz\nzzzzzzz", b"<~zzzzz\nzzzzzzz\nzz~>"), + ] + for b, n, a, a_wrap in tests: + assertEncode(a, b, n) + assertEncode(a_wrap, b, n, adobe=True) + assertDecode(a, b) + assertDecode(a_wrap, b, adobe=True) + + def test_ascii85_pad(self): + # Test Ascii85 with encode padding + rawdata = b"n1n3tee\n ch@rAcTer$" + for i in range(1, len(rawdata) + 1): + padding = -i % 4 + b = rawdata[:i] + a_pad = binascii.b2a_ascii85(self.type2test(b), pad=True) + b_pad = binascii.a2b_ascii85(self.type2test(a_pad)) + b_pad_expected = b + b"\0" * padding + self.assertEqual(b_pad, b_pad_expected) + + # Test Ascii85 short form groups with encode padding + def assertShortPad(data, expected, **kwargs): + data = self.type2test(data) + res = binascii.b2a_ascii85(data, **kwargs) + self.assertEqual(res, expected) + + assertShortPad(b"\0", b"!!", pad=False) + assertShortPad(b"\0", b"z", pad=True) + assertShortPad(b"\0" * 2, b"z", pad=True) + assertShortPad(b"\0" * 3, b"z", pad=True) + assertShortPad(b"\0" * 4, b"z", pad=True) + assertShortPad(b"\0" * 5, b"zz", pad=True) + assertShortPad(b"\0" * 6, b"z!!!") + assertShortPad(b" " * 7, b"y+", + foldspaces=True, adobe=True, pad=True) + assertShortPad(b"\0\0\0\0abcd \0\0", b"<~z@:E_Wy\nz~>", + foldspaces=True, adobe=True, wrapcol=9, pad=True) + + def test_ascii85_ignorechars(self): + # Test Ascii85 with ignored characters + def assertIgnore(data, expected, ignorechars=b"", **kwargs): + data = self.type2test(data) + ignore = self.type2test(ignorechars) + with self.assertRaisesRegex(binascii.Error, r"(?i)Non-Ascii85 digit"): + binascii.a2b_ascii85(data, **kwargs) + res = binascii.a2b_ascii85(data, ignorechars=ignorechars, **kwargs) + self.assertEqual(res, expected) + + assertIgnore(b"\n", b"", ignorechars=b"\n") + assertIgnore(b"<~ ~>", b"", ignorechars=b" ", adobe=True) + assertIgnore(b"z|z", b"\0" * 8, ignorechars=b"|||") # repeats don't matter + assertIgnore(b"zz!!|", b"\0" * 9, ignorechars=b"|!z") # ignore only if invalid + assertIgnore(b"<~B P~@~>", b"hi", ignorechars=b" <~>", adobe=True) + assertIgnore(b"zy}", b"\0\0\0\0", ignorechars=b"zy}") + assertIgnore(b"zy}", b"\0\0\0\0 ", ignorechars=b"zy}", foldspaces=True) + + def test_base85_valid(self): + # Test base85 with valid data + lines, i = [], 0 + for k in range(1, len(self.rawdata) + 1): + b = self.type2test(self.rawdata[i:i + k]) + a = binascii.b2a_base85(b) + lines.append(a) + i += k + if i >= len(self.rawdata): + break + res = bytes() + for line in lines: + a = self.type2test(line) + b = binascii.a2b_base85(a) + res += b + self.assertEqual(res, self.rawdata) + + # Test decoding inputs with different length + # 1-char groups are rejected (encoding violation) + with self.assertRaises(binascii.Error): + binascii.a2b_base85(self.type2test(b'a')) + self.assertEqual(binascii.a2b_base85(self.type2test(b'ab')), b'q') + self.assertEqual(binascii.a2b_base85(self.type2test(b'abc')), b'qa') + self.assertEqual(binascii.a2b_base85(self.type2test(b'abcd')), + b'qa\x9e') + self.assertEqual(binascii.a2b_base85(self.type2test(b'abcde')), + b'qa\x9e\xb6') + # 6-char input = full 5-char group + trailing 1-char group (rejected) + with self.assertRaises(binascii.Error): + binascii.a2b_base85(self.type2test(b'abcdef')) + self.assertEqual(binascii.a2b_base85(self.type2test(b'abcdefg')), + b'qa\x9e\xb6\x81') + + def test_base85_errors(self): + def _assertRegexTemplate(assert_regex, data, **kwargs): + with self.assertRaisesRegex(binascii.Error, assert_regex): + binascii.a2b_base85(self.type2test(data), **kwargs) + + def assertNonBase85Data(data): + _assertRegexTemplate(r"(?i)bad base85 character", data) + + def assertOverflow(data): + _assertRegexTemplate(r"(?i)base85 overflow", data) + + assertNonBase85Data(b"\xda") + assertNonBase85Data(b"00\0\0") + assertNonBase85Data(b"Z )*") + assertNonBase85Data(b"bY*jNb0Hyq\n") + + # Test base85 with out-of-range encoded value + assertOverflow(b"}") + assertOverflow(b"|O") + assertOverflow(b"|Nt") + assertOverflow(b"|NsD") + assertOverflow(b"|NsC1") + assertOverflow(b"|NsC0~") + assertOverflow(b"|NsC0|NsC0|NsD0") + + def test_base85_wrapcol(self): + self._common_test_wrapcol(binascii.b2a_base85) + b = self.type2test(b'www.python.org') + self.assertEqual(binascii.b2a_base85(b, wrapcol=10), + b'cXxL#aCvlS\nZ*DGca%T') + self.assertEqual(binascii.b2a_base85(b, wrapcol=14), + b'cXxL#aCvlS\nZ*DGca%T') + self.assertEqual(binascii.b2a_base85(b, wrapcol=1), + b'cXxL#\naCvlS\nZ*DGc\na%T') + + def test_base85_pad(self): + # Test base85 with encode padding + rawdata = b"n1n3Tee\n ch@rAc\te\r$" + for i in range(1, len(rawdata) + 1): + padding = -i % 4 + b = rawdata[:i] + a_pad = binascii.b2a_base85(self.type2test(b), pad=True) + b_pad = binascii.a2b_base85(self.type2test(a_pad)) + b_pad_expected = b + b"\0" * padding + self.assertEqual(b_pad, b_pad_expected) + + def test_base85_ignorechars(self): + a2b_base85 = binascii.a2b_base85 + self._common_test_ignorechars(a2b_base85) + eq = self.assertEqual + eq(a2b_base85(b'VPa\n !s\n', ignorechars=b' \n'), b'abcd') + eq(a2b_base85(b'VPa\n !s\n', ignorechars=bytearray(b' \n')), b'abcd') + + eq(a2b_base85(b'A~[B];C', ignorechars=b';[]~'), b'"1\xa3\x15') + eq(a2b_base85(b'A~[B];C', ignorechars=b';[]~', + alphabet=binascii.Z85_ALPHABET), b'r\xd8dv') + + def test_base85_alphabet(self): + alphabet = (b'0123456789abcdefghijklmnopqrstuvwxyz' + b'ABCDEFGHIJKLMNOPQRSTUVWXYZ.-:+=^!/*?&<>()[]{}@%$#') + data = self.type2test(self.rawdata) + encoded = binascii.b2a_base85(data, alphabet=alphabet) + trans = bytes.maketrans(binascii.BASE85_ALPHABET, alphabet) + expected = binascii.b2a_base85(data).translate(trans) + self.assertEqual(encoded, expected) + self.assertEqual(binascii.a2b_base85(encoded, alphabet=alphabet), self.rawdata) + self.assertEqual(binascii.b2a_base85(data, alphabet=self.type2test(alphabet)), expected) + + data = self.type2test(b'') + self.assertEqual(binascii.b2a_base85(data, alphabet=alphabet), b'') + self.assertEqual(binascii.a2b_base85(data, alphabet=alphabet), b'') + + for func in binascii.b2a_base85, binascii.a2b_base85: + with self.assertRaises(TypeError): + func(data, alphabet=None) + with self.assertRaises(TypeError): + func(data, alphabet=alphabet.decode()) + with self.assertRaises(ValueError): + func(data, alphabet=alphabet[:-1]) + with self.assertRaises(ValueError): + func(data, alphabet=alphabet+b'?') + with self.assertRaises(TypeError): + binascii.a2b_base64(data, alphabet=bytearray(alphabet)) + + def test_base85_canonical(self): + # Non-canonical encodings are accepted without canonical=True + self.assertEqual(binascii.a2b_base85(b'VF'), b'a') + + # 1-char partial groups are always rejected (encoding violation: + # no conforming encoder produces them) + with self.assertRaises(binascii.Error): + binascii.a2b_base85(b'V') + with self.assertRaises(binascii.Error): + binascii.a2b_base85(b'0') + + # Verify round-trip: encode then decode with canonical=True works + for data in [b'a', b'ab', b'abc', b'abcd', b'abcde', + b'\x00', b'\xff', b'\x00\x00', b'\xff\xff\xff']: + encoded = binascii.b2a_base85(data) + decoded = binascii.a2b_base85(encoded, canonical=True) + self.assertEqual(decoded, data) + + # Test non-canonical rejection for each partial group size + # (2-char/1-byte, 3-char/2-byte, 4-char/3-byte). + # Incrementing the last digit by 1 produces a non-canonical + # encoding. For 4-char groups (n_pad=1) a +1 can change the + # output byte, so we use b'ab\x00' whose canonical form allows + # a +1 that still decodes to the same 3 bytes. + for data in [b'a', b'ab', b'ab\x00']: + canonical_enc = binascii.b2a_base85(data) + non_canonical = (canonical_enc[:-1] + + bytes([canonical_enc[-1] + 1])) + # Same decoded output without canonical check + self.assertEqual(binascii.a2b_base85(non_canonical), data) + # Rejected with canonical=True + with self.assertRaises(binascii.Error): + binascii.a2b_base85(non_canonical, canonical=True) + + # Boundary bytes: \x00 and \xff for each partial group size + for data in [b'\x00', b'\x00\x00', b'\x00\x00\x00', + b'\xff', b'\xff\xff', b'\xff\xff\xff']: + canonical_enc = binascii.b2a_base85(data) + binascii.a2b_base85(canonical_enc, canonical=True) + + # Full 5-char groups are always canonical (no padding bits) + self.assertEqual( + binascii.a2b_base85(b'VPa!s', canonical=True), b'abcd') + + # Empty input is valid + self.assertEqual(binascii.a2b_base85(b'', canonical=True), b'') + + @hypothesis.given(payload=hypothesis.strategies.binary()) + @hypothesis.example(b'') + @hypothesis.example(b'\x00') + @hypothesis.example(b'\xff\xff') + @hypothesis.example(b'abc') + def test_base85_canonical_roundtrip(self, payload): + encoded = binascii.b2a_base85(payload) + decoded = binascii.a2b_base85(encoded, canonical=True) + self.assertEqual(decoded, payload) + + @hypothesis.given(payload=hypothesis.strategies.binary(min_size=1, max_size=3)) + @hypothesis.example(b'\x00') + @hypothesis.example(b'\xff') + @hypothesis.example(b'ab\x00') + def test_base85_canonical_unique(self, payload): + # For a partial group, sweeping all 85 last-digit values should + # yield exactly one encoding that both decodes to the original + # payload AND passes canonical=True. + hypothesis.assume(len(payload) % 4 != 0) + canonical_enc = binascii.b2a_base85(payload) + table = binascii.BASE85_ALPHABET + accepted = [] + for digit in table: + candidate = canonical_enc[:-1] + bytes([digit]) + try: + result = binascii.a2b_base85(candidate, canonical=True) + if result == payload: + accepted.append(candidate) + except binascii.Error: + pass + self.assertEqual(accepted, [canonical_enc]) + + def test_ascii85_canonical(self): + # Non-canonical encodings are accepted without canonical=True + self.assertEqual(binascii.a2b_ascii85(b'@0'), b'a') + + # 1-char partial groups are always rejected (PLRM encoding violation) + with self.assertRaises(binascii.Error): + binascii.a2b_ascii85(b'@') + with self.assertRaises(binascii.Error): + binascii.a2b_ascii85(b'!') + + # Verify round-trip: encode then decode with canonical=True works + for data in [b'a', b'ab', b'abc', b'abcd', b'abcde', + b'\x00', b'\xff', b'\x00\x00', b'\xff\xff\xff']: + encoded = binascii.b2a_ascii85(data) + decoded = binascii.a2b_ascii85(encoded, canonical=True) + self.assertEqual(decoded, data) + + # Test non-canonical rejection for each partial group size. + # See test_base85_canonical for why b'ab\x00' is used for 3 bytes. + for data in [b'a', b'ab', b'ab\x00']: + canonical_enc = binascii.b2a_ascii85(data) + non_canonical = (canonical_enc[:-1] + + bytes([canonical_enc[-1] + 1])) + self.assertEqual(binascii.a2b_ascii85(non_canonical), data) + with self.assertRaises(binascii.Error): + binascii.a2b_ascii85(non_canonical, canonical=True) + + # Full 5-char groups are always canonical + self.assertEqual( + binascii.a2b_ascii85(b'@:E_W', canonical=True), b'abcd') + + # 'z' is the canonical form for all-zero groups per the PLRM. + # '!!!!!' decodes identically but is non-canonical. + self.assertEqual(binascii.a2b_ascii85(b'!!!!!'), b'\x00' * 4) + self.assertEqual(binascii.a2b_ascii85(b'z'), b'\x00' * 4) + self.assertEqual( + binascii.a2b_ascii85(b'z', canonical=True), b'\x00' * 4) + with self.assertRaises(binascii.Error): + binascii.a2b_ascii85(b'!!!!!', canonical=True) + # Multiple groups: z + !!!!! should fail + with self.assertRaises(binascii.Error): + binascii.a2b_ascii85(b'z!!!!!', canonical=True) + # Multiple z groups are fine + self.assertEqual( + binascii.a2b_ascii85(b'zz', canonical=True), b'\x00' * 8) + + # Empty input is valid + self.assertEqual(binascii.a2b_ascii85(b'', canonical=True), b'') + + # Adobe-wrapped with canonical + self.assertEqual( + binascii.a2b_ascii85(b'<~@:E_W~>', canonical=True, adobe=True), + b'abcd') + + @hypothesis.given(payload=hypothesis.strategies.binary()) + @hypothesis.example(b'') + @hypothesis.example(b'\x00') + @hypothesis.example(b'\x00\x00\x00\x00') # triggers z abbreviation + @hypothesis.example(b'\xff\xff') + @hypothesis.example(b'abc') + def test_ascii85_canonical_roundtrip(self, payload): + encoded = binascii.b2a_ascii85(payload) + decoded = binascii.a2b_ascii85(encoded, canonical=True) + self.assertEqual(decoded, payload) + + @hypothesis.given(payload=hypothesis.strategies.binary(min_size=1, max_size=3)) + @hypothesis.example(b'\x00') + @hypothesis.example(b'\xff') + @hypothesis.example(b'ab\x00') + def test_ascii85_canonical_unique(self, payload): + hypothesis.assume(len(payload) % 4 != 0) + canonical_enc = binascii.b2a_ascii85(payload) + # Ascii85 alphabet: '!' (33) through 'u' (117) + accepted = [] + for digit in range(33, 118): + candidate = canonical_enc[:-1] + bytes([digit]) + try: + result = binascii.a2b_ascii85(candidate, canonical=True) + if result == payload: + accepted.append(candidate) + except binascii.Error: + pass + self.assertEqual(accepted, [canonical_enc]) + + def test_base32_valid(self): + # Test base32 with valid data + lines = [] + step = 0 + i = 0 + while i < len(self.rawdata): + b = self.type2test(self.rawdata[i:i + step]) + a = binascii.b2a_base32(b) + lines.append(a) + i += step + step += 1 + res = bytes() + for line in lines: + a = self.type2test(line) + b = binascii.a2b_base32(a) + res += b + self.assertEqual(res, self.rawdata) + + def test_base32_errors(self): + def _fixPadding(data): + fixed = data.replace(b"=", b"") + len_8 = len(fixed) % 8 + p = 8 - len_8 if len_8 else 0 + return fixed + b"=" * p + + def _assertRegexTemplate(assert_regex, data, good_padding_result=None, **kwargs): + with self.assertRaisesRegex(binascii.Error, assert_regex): + binascii.a2b_base32(self.type2test(data), **kwargs) + if good_padding_result: + fixed = self.type2test(_fixPadding(data)) + self.assertEqual(binascii.a2b_base32(fixed), good_padding_result) + + def assertNonBase32Data(*args): + _assertRegexTemplate(r"(?i)Only base32 data", *args) + + def assertExcessData(*args): + _assertRegexTemplate(r"(?i)Excess data", *args) + + def assertExcessPadding(*args): + _assertRegexTemplate(r"(?i)Excess padding", *args) + + def assertLeadingPadding(*args, **kwargs): + _assertRegexTemplate(r"(?i)Leading padding", *args, **kwargs) + + def assertIncorrectPadding(*args): + _assertRegexTemplate(r"(?i)Incorrect padding", *args) + + def assertDiscontinuousPadding(*args): + _assertRegexTemplate(r"(?i)Discontinuous padding", *args) + + def assertInvalidLength(data, *args, length=None, **kwargs): + if length is None: + length = len(data.split(b'=', 1)[0].replace(b' ', b'')) + assert_regex = fr"(?i)Invalid.+number of data characters \({length}\)" + _assertRegexTemplate(assert_regex, data, *args, **kwargs) + + assertNonBase32Data(b"a") + assertNonBase32Data(b"AA-") + assertNonBase32Data(b"ABCDE==!") + assertNonBase32Data(b"ab:(){:|:&};:==") + + assertExcessData(b"AB======C") + assertExcessData(b"AB======CD") + assertExcessData(b"ABCD====E") + assertExcessData(b"ABCDE===FGH") + assertExcessData(b"ABCDEFG=H") + assertExcessData(b"432Z====55555555") + + assertExcessData(b"BE======EF", b"\t\x08") + assertExcessData(b"BEEF====C", b"\t\x08Q") + assertExcessData(b"BEEFC===AK", b"\t\x08Q\x01") + assertExcessData(b"BEEFCAK=E", b"\t\x08Q\x01D") + + assertExcessPadding(b"BE=======", b"\t") + assertExcessPadding(b"BE========", b"\t") + assertExcessPadding(b"BEEF=====", b"\t\x08") + assertExcessPadding(b"BEEF======", b"\t\x08") + assertExcessPadding(b"BEEFC====", b"\t\x08Q") + assertExcessPadding(b"BEEFC=====", b"\t\x08Q") + assertExcessPadding(b"BEEFCAK==", b"\t\x08Q\x01") + assertExcessPadding(b"BEEFCAK===", b"\t\x08Q\x01") + assertExcessPadding(b"BEEFCAKE=", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE==", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE===", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE====", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE=====", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE======", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE=======", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE========", b"\t\x08Q\x01D") + assertExcessPadding(b"BEEFCAKE=========", b"\t\x08Q\x01D") + + assertLeadingPadding(b"=", b"") + assertLeadingPadding(b"==", b"") + assertLeadingPadding(b"===", b"") + assertLeadingPadding(b"====", b"") + assertLeadingPadding(b"=====", b"") + assertLeadingPadding(b"======", b"") + assertLeadingPadding(b"=======", b"") + assertLeadingPadding(b"========", b"") + assertLeadingPadding(b"=========", b"") + assertLeadingPadding(b"=BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"==BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"===BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"====BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"=====BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"======BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"=======BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"========BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b"=========BEEFCAKE", b"\t\x08Q\x01D") + assertLeadingPadding(b" =BEEFCAKE", ignorechars=b' ') + + assertIncorrectPadding(b"AB") + assertIncorrectPadding(b"ABCD") + assertIncorrectPadding(b"ABCDE") + assertIncorrectPadding(b"ABCDEFG") + + assertIncorrectPadding(b"BE=", b"\t") + assertIncorrectPadding(b"BE==", b"\t") + assertIncorrectPadding(b"BE===", b"\t") + assertIncorrectPadding(b"BE====", b"\t") + assertIncorrectPadding(b"BE=====", b"\t") + assertIncorrectPadding(b"BEEF=", b"\t\x08") + assertIncorrectPadding(b"BEEF==", b"\t\x08") + assertIncorrectPadding(b"BEEF===", b"\t\x08") + assertIncorrectPadding(b"BEEFC=", b"\t\x08Q") + assertIncorrectPadding(b"BEEFC==", b"\t\x08Q") + + assertDiscontinuousPadding(b"BE=EF===", b"\t\x08") + assertDiscontinuousPadding(b"BE==EF==", b"\t\x08") + assertDiscontinuousPadding(b"BEEF=C==", b"\t\x08Q") + assertDiscontinuousPadding(b"BEEFC=AK", b"\t\x08Q\x01") + + assertInvalidLength(b"A") + assertInvalidLength(b"ABC") + assertInvalidLength(b"ABCDEF") + assertInvalidLength(b"ABCDEFGHI") + assertInvalidLength(b"ABCDEFGHIJK") + assertInvalidLength(b"ABCDEFGHIJKLMN") + + assertInvalidLength(b"A=") + assertInvalidLength(b"A==") + assertInvalidLength(b"A===") + assertInvalidLength(b"A====") + assertInvalidLength(b"A=====") + assertInvalidLength(b"A======") + assertInvalidLength(b"ABC=") + assertInvalidLength(b"ABC==") + assertInvalidLength(b"ABC===") + assertInvalidLength(b"ABC====") + assertInvalidLength(b"ABCDEF=") + + assertInvalidLength(b"B=E=====", b"\t") + assertInvalidLength(b"B==E====", b"\t") + assertInvalidLength(b"BEE=F===", b"\t\x08") + assertInvalidLength(b"BEE==F==", b"\t\x08") + assertInvalidLength(b"BEEFCA=K", b"\t\x08Q\x01") + assertInvalidLength(b"BEEFCA=====K", b"\t\x08Q\x01") + + assertInvalidLength(b" A", ignorechars=b' ') + assertInvalidLength(b" ABC", ignorechars=b' ') + assertInvalidLength(b" ABCDEF", ignorechars=b' ') + assertInvalidLength(b" ABCDEFGHI", ignorechars=b' ') + assertInvalidLength(b" ABCDEFGHIJK", ignorechars=b' ') + assertInvalidLength(b" ABCDEFGHIJKLMN", ignorechars=b' ') + assertInvalidLength(b" A=======", ignorechars=b' ') + assertInvalidLength(b" ABC=====", ignorechars=b' ') + assertInvalidLength(b" ABCDEF==", ignorechars=b' ') + + def test_base32_canonical(self): + # https://datatracker.ietf.org/doc/html/rfc4648.html#section-3.5 + # Decoders MAY reject encoded data if the pad bits are not zero. + + # Without canonical=True, non-zero padding bits are accepted + self.assertEqual(binascii.a2b_base32(self.type2test(b'AB======')), + b'\x00') + + # 2 data chars + "======": last char has 2 padding bits + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AB======'), canonical=True) + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AD======'), canonical=True) + + # 4 data chars + "====": last char has 4 padding bits + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AAAB===='), canonical=True) + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AAAP===='), canonical=True) + + # 5 data chars + "===": last char has 1 padding bit + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AAAAB==='), canonical=True) + + # 7 data chars + "=": last char has 3 padding bits + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AAAAAAB='), canonical=True) + with self.assertRaises(binascii.Error): + binascii.a2b_base32(self.type2test(b'AAAAAAH='), canonical=True) + + # Verify that zero padding bits are accepted + binascii.a2b_base32(self.type2test(b'AA======'), canonical=True) + binascii.a2b_base32(self.type2test(b'AAAA===='), canonical=True) + binascii.a2b_base32(self.type2test(b'AAAAA==='), canonical=True) + binascii.a2b_base32(self.type2test(b'AAAAAAA='), canonical=True) + + # Full octet with no padding -- always valid + binascii.a2b_base32(self.type2test(b'AAAAAAAA'), canonical=True) + + @hypothesis.given(payload=hypothesis.strategies.binary()) + @hypothesis.example(b'') + @hypothesis.example(b'\x00') + @hypothesis.example(b'\xff\xff') + @hypothesis.example(b'abc') + def test_base32_canonical_roundtrip(self, payload): + encoded = binascii.b2a_base32(payload) + decoded = binascii.a2b_base32(encoded, canonical=True) + self.assertEqual(decoded, payload) + + def test_a2b_base32_padded(self): + a2b_base32 = binascii.a2b_base32 + t = self.type2test + def check(data, expected): + if b'=' in data: + with self.assertRaisesRegex(binascii.Error, 'Padding not allowed'): + a2b_base32(t(data), padded=False) + self.assertEqual(a2b_base32(t(data), padded=False, ignorechars=b'='), + expected) + data = data.replace(b'=', b'') + self.assertEqual(a2b_base32(t(data), padded=False), expected) + + check(b'', b'') + check(b'ME======', b'a') + check(b'MFRA====', b'ab') + check(b'MFRGG===', b'abc') + check(b'MFRGGZA=', b'abcd') + check(b'MFRGGZDF', b'abcde') + check(b'M=FRGGZDF', b'abcde') + check(b'MF=RGGZDF', b'abcde') + check(b'MFR=GGZDF', b'abcde') + check(b'MFRG=GZDF', b'abcde') + check(b'MFRGG=ZDF', b'abcde') + check(b'MFRGGZ=DF', b'abcde') + check(b'MFRGGZD=F', b'abcde') + + def test_b2a_base32_padded(self): + b2a_base32 = binascii.b2a_base32 + t = self.type2test + self.assertEqual(b2a_base32(t(b''), padded=False), b'') + self.assertEqual(b2a_base32(t(b'a'), padded=False), b'ME') + self.assertEqual(b2a_base32(t(b'ab'), padded=False), b'MFRA') + self.assertEqual(b2a_base32(t(b'abc'), padded=False), b'MFRGG') + self.assertEqual(b2a_base32(t(b'abcd'), padded=False), b'MFRGGZA') + self.assertEqual(b2a_base32(t(b'abcde'), padded=False), b'MFRGGZDF') + + def test_base32_wrapcol(self): + self._common_test_wrapcol(binascii.b2a_base32) + b = self.type2test(b'www.python.org') + self.assertEqual(binascii.b2a_base32(b, wrapcol=16), + b'O53XOLTQPF2GQ33O\nFZXXEZY=') + self.assertEqual(binascii.b2a_base32(b, wrapcol=23), + b'O53XOLTQPF2GQ33O\nFZXXEZY=') + self.assertEqual(binascii.b2a_base32(b, wrapcol=1), + b'O53XOLTQ\nPF2GQ33O\nFZXXEZY=') + + def test_base32_ignorechars(self): + a2b_base32 = binascii.a2b_base32 + self._common_test_ignorechars(a2b_base32) + eq = self.assertEqual + eq(a2b_base32(b'MFRG\n GZDF\n', ignorechars=b' \n'), b'abcde') + eq(a2b_base32(b'MFRG\n GZDF\n', ignorechars=bytearray(b' \n')), b'abcde') + eq(a2b_base32(b'M=======FRGGZDF', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MF======RGGZDF', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFR=====GGZDF', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFRG====GZDF', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFRGG===ZDF', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFRGGZ==DF', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFRGGZD=F', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFRGGZDF=', ignorechars=b'='), b'abcde') + eq(a2b_base32(b'MFRA======', ignorechars=b'='), b'ab') + + eq(a2b_base32(b'A1B3C5W7Z9', ignorechars=b'19WZ'), b'\x00v.\xdb\xf9') + eq(a2b_base32(b'A1B3C5W7Z9', ignorechars=b'19WZ', + alphabet=binascii.BASE32HEX_ALPHABET), b'PV6\x14\xe9') + + def test_base32_alphabet(self): + alphabet = b'0Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8Ii9JjKk' + data = self.type2test(self.rawdata) + encoded = binascii.b2a_base32(data, alphabet=alphabet) + trans = bytes.maketrans(binascii.BASE32_ALPHABET, alphabet) + expected = binascii.b2a_base32(data).translate(trans) + self.assertEqual(encoded, expected) + self.assertEqual(binascii.a2b_base32(encoded, alphabet=alphabet), self.rawdata) + self.assertEqual(binascii.b2a_base32(data, alphabet=self.type2test(alphabet)), expected) + + data = self.type2test(b'') + self.assertEqual(binascii.b2a_base32(data, alphabet=alphabet), b'') + self.assertEqual(binascii.a2b_base32(data, alphabet=alphabet), b'') + + for func in binascii.b2a_base32, binascii.a2b_base32: + with self.assertRaises(TypeError): + func(data, alphabet=None) + with self.assertRaises(TypeError): + func(data, alphabet=alphabet.decode()) + with self.assertRaises(ValueError): + func(data, alphabet=alphabet[:-1]) + with self.assertRaises(ValueError): + func(data, alphabet=alphabet+b'?') + with self.assertRaises(TypeError): + binascii.a2b_base32(data, alphabet=bytearray(alphabet)) def test_uu(self): MAX_UU = 45 @@ -325,6 +1391,15 @@ def test_hex_separator(self): expected1 = s.hex(':').encode('ascii') self.assertEqual(binascii.b2a_hex(self.type2test(s), ':'), expected1) + def test_hex_ignorechars(self): + a2b_hex = binascii.a2b_hex + self._common_test_ignorechars(a2b_hex) + self._common_test_ignorechars(binascii.unhexlify) + eq = self.assertEqual + eq(a2b_hex(b'A B\nC D\n', ignorechars=b' \n'), b'\xab\xcd') + eq(a2b_hex(b'A B\nC D\n', ignorechars=bytearray(b' \n')), b'\xab\xcd') + eq(a2b_hex(b'aBcD', ignorechars=b'ac'), b'\xab\xcd') + def test_qp(self): type2test = self.type2test a2b_qp = binascii.a2b_qp @@ -499,6 +1574,32 @@ def test_b2a_base64_newline(self): b'aGVsbG8=\n') self.assertEqual(binascii.b2a_base64(b, newline=False), b'aGVsbG8=') + b = self.type2test(b'') + self.assertEqual(binascii.b2a_base64(b), b'\n') + self.assertEqual(binascii.b2a_base64(b, newline=True), b'\n') + self.assertEqual(binascii.b2a_base64(b, newline=False), b'') + + def test_b2a_base64_padded(self): + b2a_base64 = binascii.b2a_base64 + t = self.type2test + self.assertEqual(b2a_base64(t(b''), padded=False), b'\n') + self.assertEqual(b2a_base64(t(b'a'), padded=False), b'YQ\n') + self.assertEqual(b2a_base64(t(b'ab'), padded=False), b'YWI\n') + self.assertEqual(b2a_base64(t(b'abc'), padded=False), b'YWJj\n') + + def test_b2a_base64_wrapcol(self): + self._common_test_wrapcol(binascii.b2a_base64) + b = self.type2test(b'www.python.org') + self.assertEqual(binascii.b2a_base64(b, wrapcol=8), + b'd3d3LnB5\ndGhvbi5v\ncmc=\n') + self.assertEqual(binascii.b2a_base64(b, wrapcol=11), + b'd3d3LnB5\ndGhvbi5v\ncmc=\n') + self.assertEqual(binascii.b2a_base64(b, wrapcol=8, newline=False), + b'd3d3LnB5\ndGhvbi5v\ncmc=') + self.assertEqual(binascii.b2a_base64(b, wrapcol=1), + b'd3d3\nLnB5\ndGhv\nbi5v\ncmc=\n') + b = self.type2test(b'') + self.assertEqual(binascii.b2a_base64(b, wrapcol=8, newline=False), b'') @hypothesis.given( binary=hypothesis.strategies.binary(), diff --git a/crates/stdlib/src/binascii.rs b/crates/stdlib/src/binascii.rs index dab01ceae25..715410f14b6 100644 --- a/crates/stdlib/src/binascii.rs +++ b/crates/stdlib/src/binascii.rs @@ -1,4 +1,6 @@ -// spell-checker:ignore hexlify unhexlify uuencodes rlecode rledecode +// spell-checker:ignore hexlify unhexlify uuencodes rlecode rledecode ABCDEFGHIJKLMNPQRSTUVXYZ abcdefhijklmpqr + +mod base_n; pub(super) use decl::crc32; pub(crate) use decl::module_def; @@ -8,7 +10,7 @@ use rustpython_vm::{VirtualMachine, builtins::PyBaseExceptionRef}; #[pymodule(name = "binascii")] mod decl { - use super::new_binascii_error; + use super::{base_n, new_binascii_error}; use crate::vm::{ AsObject, Py, PyObjectRef, PyRef, PyResult, VirtualMachine, builtins::{PyBytes, PyBytesRef, PyDict, PyModule, PyTypeRef}, @@ -30,23 +32,47 @@ mod decl { "a2b_base64($module, data, /, *, strict_mode=, padded=True, alphabet=BASE64_ALPHABET, ignorechars=, canonical=False)\n--\n\nDecode a line of base64 data.", ), ); - let function = DECODE.build_bound_function(&vm.ctx, module.to_owned().into()); - function - .as_object() - .set_attr("__module__", vm.ctx.new_str("binascii"), vm)?; - module.dict().set_item("a2b_base64", function.into(), vm)?; + const DECODE32: &PyMethodDef = &PyMethodDef::new_const( + "a2b_base32", + a2b_base32, + PyMethodFlags::empty(), + ItemDoc::static_text( + "a2b_base32($module, data, /, *, padded=True, alphabet=BASE32_ALPHABET, ignorechars=b'', canonical=False)\n--\n\nDecode a line of base32 data.", + ), + ); + const DECODE85: &PyMethodDef = &PyMethodDef::new_const( + "a2b_base85", + a2b_base85, + PyMethodFlags::empty(), + ItemDoc::static_text( + "a2b_base85($module, data, /, *, alphabet=BASE85_ALPHABET, ignorechars=b'', canonical=False)\n--\n\nDecode a line of Base85 data.", + ), + ); + for (name, definition) in [ + ("a2b_base64", DECODE), + ("a2b_base32", DECODE32), + ("a2b_base85", DECODE85), + ] { + let function = definition.build_bound_function(&vm.ctx, module.to_owned().into()); + function + .as_object() + .set_attr("__module__", vm.ctx.new_str("binascii"), vm)?; + module.dict().set_item(name, function.into(), vm)?; + } Ok(()) } fn reverse_table( module: &Py, alphabet: &PyBytesRef, + size: usize, + padchar: Option, vm: &VirtualMachine, ) -> PyResult<[u8; 256]> { - let bytes: &[u8; 64] = alphabet - .as_bytes() - .try_into() - .map_err(|_| vm.new_value_error("alphabet must have length 64"))?; + let bytes = alphabet.as_bytes(); + if bytes.len() != size { + return Err(vm.new_value_error(format!("alphabet must have length {size}"))); + } let cache = module .native_state() .ok_or_else(|| vm.new_runtime_error("Missing binascii state"))? @@ -60,7 +86,7 @@ mod decl { .and_then(|value| value.as_bytes().try_into().ok()) .ok_or_else(|| vm.new_runtime_error("Broken binascii cache")); } - let table = binascii::base64_reverse_table(bytes); + let table = base_n::reverse_table(bytes, padchar); cache.set_item( alphabet.as_object(), vm.ctx.new_bytes(table.to_vec()).into(), @@ -69,6 +95,53 @@ mod decl { Ok(table) } + #[pyattr(name = "CRYPT_ALPHABET")] + fn crypt_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx + .new_bytes(b"./0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz".to_vec()) + } + + #[pyattr(name = "UU_ALPHABET")] + fn uu_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx.new_bytes( + b" !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_".to_vec(), + ) + } + + #[pyattr(name = "BINHEX_ALPHABET")] + fn binhex_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx.new_bytes( + b"!\"#$%&'()*+,-012345689@ABCDEFGHIJKLMNPQRSTUVXYZ[`abcdefhijklmpqr".to_vec(), + ) + } + + #[pyattr(name = "BASE85_ALPHABET")] + fn base85_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx.new_bytes(b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz!#$%&()*+-;<=>?@^_`{|}~".to_vec()) + } + + #[pyattr(name = "ASCII85_ALPHABET")] + fn ascii85_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx.new_bytes(b"!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstu".to_vec()) + } + + #[pyattr(name = "Z85_ALPHABET")] + fn z85_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx.new_bytes(b"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ.-:+=^!/*?&<>()[]{}@%$#".to_vec()) + } + + #[pyattr(name = "BASE32_ALPHABET")] + fn base32_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx + .new_bytes(b"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567".to_vec()) + } + + #[pyattr(name = "BASE32HEX_ALPHABET")] + fn base32hex_alphabet(vm: &VirtualMachine) -> PyBytesRef { + vm.ctx + .new_bytes(b"0123456789ABCDEFGHIJKLMNOPQRSTUV".to_vec()) + } + #[pyattr(name = "Error", once)] pub(super) fn error_type(vm: &VirtualMachine) -> PyTypeRef { vm.ctx.new_exception_type( @@ -107,12 +180,24 @@ mod decl { Ok(data.with_ref(|bytes| binascii::hexlify(bytes, sep, bytes_per_sep))) } + #[derive(FromArgs)] + struct UnhexlifyArgs { + #[pyarg(positional)] + hexstr: ArgAsciiBuffer, + #[pyarg(named, optional, py_default = "b''")] + ignorechars: OptionalArg, + } + #[pyfunction(name = "a2b_hex")] #[pyfunction] - fn unhexlify(hexstr: ArgAsciiBuffer, vm: &VirtualMachine) -> PyResult> { - hexstr - .with_ref(binascii::unhexlify) - .map_err(|e| new_binascii_error(e, vm)) + fn unhexlify(args: UnhexlifyArgs, vm: &VirtualMachine) -> PyResult> { + let ignorechars = ignored_argument(&args.ignorechars); + let result = args + .hexstr + .with_ref(|data| base_n::decode_hex(data, &ignorechars)); + drop(args.hexstr); + drop(args.ignorechars); + base_n_result(result, vm) } #[derive(FromArgs)] @@ -192,7 +277,7 @@ mod decl { const DEFAULT_TABLE: [u8; 256] = binascii::base64_reverse_table(binascii::BASE64_ALPHABET); let table = match &alphabet { - OptionalArg::Present(value) => reverse_table(&module, value, vm)?, + OptionalArg::Present(value) => reverse_table(&module, value, 64, Some(b'='), vm)?, OptionalArg::Missing => DEFAULT_TABLE, }; let mut ignored = [false; 256]; @@ -272,6 +357,287 @@ mod decl { result } + fn base_n_result( + result: Result, base_n::Error>, + vm: &VirtualMachine, + ) -> PyResult> { + result.map_err(|error| match error { + base_n::Error::InvalidAlphabet(size) => { + vm.new_value_error(format!("alphabet must have length {size}")) + } + base_n::Error::Codec(message) => vm.new_exception_msg(error_type(vm), message.into()), + base_n::Error::Memory => vm.no_memory_error(), + }) + } + + fn encoder_alphabet( + argument: &OptionalArg, + default: &[u8], + ) -> Result, base_n::Error> { + match argument { + OptionalArg::Present(value) => { + value.with_ref(|bytes| base_n::copy_alphabet(bytes, default.len())) + } + OptionalArg::Missing => base_n::copy_alphabet(default, default.len()), + } + } + + fn ignored_argument(argument: &OptionalArg) -> [bool; 256] { + match argument { + OptionalArg::Present(value) => value.with_ref(base_n::ignored), + OptionalArg::Missing => [false; 256], + } + } + + fn canonical_argument( + argument: &OptionalArg, + vm: &VirtualMachine, + ) -> PyResult { + match argument { + OptionalArg::Present(value) => value.try_to_bool(vm), + OptionalArg::Missing => Ok(false), + } + } + + #[derive(FromArgs)] + struct B2aBase32Args { + #[pyarg(positional)] + data: ArgBytesLike, + #[pyarg(named, default = ArgIntoBool::TRUE, py_default = "True")] + padded: ArgIntoBool, + #[pyarg(named, default = 0)] + wrapcol: usize, + #[pyarg(named, optional, py_default = "BASE32_ALPHABET")] + alphabet: OptionalArg, + } + + #[derive(FromArgs)] + struct A2bBase32Args { + #[pyarg(positional)] + data: ArgAsciiBuffer, + #[pyarg(named, default = ArgIntoBool::TRUE, py_default = "True")] + padded: ArgIntoBool, + #[pyarg(named, optional, py_default = "BASE32_ALPHABET")] + alphabet: OptionalArg, + #[pyarg(named, optional, py_default = "b''")] + ignorechars: OptionalArg, + #[pyarg(named, optional, py_default = "False")] + canonical: OptionalArg, + } + + fn a2b_base32( + module: PyRef, + args: A2bBase32Args, + vm: &VirtualMachine, + ) -> PyResult> { + let A2bBase32Args { + data, + padded, + alphabet, + ignorechars, + canonical, + } = args; + let result = (|| { + // This is the last Clinic conversion. Keep acquired exports alive + // and reach ordered cleanup even when __bool__ raises. + let canonical = canonical_argument(&canonical, vm)?; + let table = match &alphabet { + OptionalArg::Present(value) => reverse_table(&module, value, 32, Some(b'='), vm)?, + OptionalArg::Missing => base_n::reverse_table(base_n::BASE32, Some(b'=')), + }; + let ignore = ignored_argument(&ignorechars); + let result = data.with_ref(|data| { + base_n::decode32(data, &table, &ignore, padded.into_bool(), canonical) + }); + base_n_result(result, vm) + })(); + // Match Argument Clinic cleanup: release data before auxiliary exports. + drop(data); + drop(ignorechars); + result + } + + #[pyfunction] + fn b2a_base32(args: B2aBase32Args, vm: &VirtualMachine) -> PyResult> { + let B2aBase32Args { + data, + padded, + wrapcol, + alphabet, + } = args; + let result = encoder_alphabet(&alphabet, base_n::BASE32).and_then(|alphabet_bytes| { + data.with_ref(|data| { + base_n::encode32(data, padded.into_bool(), wrapcol, &alphabet_bytes) + }) + }); + drop(data); + drop(alphabet); + base_n_result(result, vm) + } + + #[derive(FromArgs)] + struct B2aBase85Args { + #[pyarg(positional)] + data: ArgBytesLike, + #[pyarg(named, default = ArgIntoBool::FALSE, py_default = "False")] + pad: ArgIntoBool, + #[pyarg(named, default = 0)] + wrapcol: usize, + #[pyarg(named, optional, py_default = "BASE85_ALPHABET")] + alphabet: OptionalArg, + } + + #[derive(FromArgs)] + struct A2bBase85Args { + #[pyarg(positional)] + data: ArgAsciiBuffer, + #[pyarg(named, optional, py_default = "BASE85_ALPHABET")] + alphabet: OptionalArg, + #[pyarg(named, optional, py_default = "b''")] + ignorechars: OptionalArg, + #[pyarg(named, optional, py_default = "False")] + canonical: OptionalArg, + } + + fn a2b_base85( + module: PyRef, + args: A2bBase85Args, + vm: &VirtualMachine, + ) -> PyResult> { + let A2bBase85Args { + data, + alphabet, + ignorechars, + canonical, + } = args; + let result = (|| { + // This is the last Clinic conversion. Keep acquired exports alive + // and reach ordered cleanup even when __bool__ raises. + let canonical = canonical_argument(&canonical, vm)?; + let table = match &alphabet { + OptionalArg::Present(value) => reverse_table(&module, value, 85, None, vm)?, + OptionalArg::Missing => base_n::reverse_table(base_n::BASE85, None), + }; + let ignore = ignored_argument(&ignorechars); + let result = data.with_ref(|data| { + base_n::decode85(data, &table, &ignore, canonical, false, false, false) + }); + base_n_result(result, vm) + })(); + drop(data); + drop(ignorechars); + result + } + + #[pyfunction] + fn b2a_base85(args: B2aBase85Args, vm: &VirtualMachine) -> PyResult> { + let B2aBase85Args { + data, + pad, + wrapcol, + alphabet, + } = args; + let result = encoder_alphabet(&alphabet, base_n::BASE85).and_then(|alphabet_bytes| { + data.with_ref(|data| { + base_n::encode85( + data, + pad.into_bool(), + wrapcol, + &alphabet_bytes, + false, + false, + false, + ) + }) + }); + drop(data); + drop(alphabet); + base_n_result(result, vm) + } + + #[derive(FromArgs)] + struct B2aAscii85Args { + #[pyarg(positional)] + data: ArgBytesLike, + #[pyarg(named, default = ArgIntoBool::FALSE, py_default = "False")] + foldspaces: ArgIntoBool, + #[pyarg(named, default = 0)] + wrapcol: usize, + #[pyarg(named, default = ArgIntoBool::FALSE, py_default = "False")] + pad: ArgIntoBool, + #[pyarg(named, default = ArgIntoBool::FALSE, py_default = "False")] + adobe: ArgIntoBool, + } + + #[derive(FromArgs)] + struct A2bAscii85Args { + #[pyarg(positional)] + data: ArgAsciiBuffer, + #[pyarg(named, default = ArgIntoBool::FALSE, py_default = "False")] + foldspaces: ArgIntoBool, + #[pyarg(named, default = ArgIntoBool::FALSE, py_default = "False")] + adobe: ArgIntoBool, + #[pyarg(named, optional, py_default = "b''")] + ignorechars: OptionalArg, + #[pyarg(named, optional, py_default = "False")] + canonical: OptionalArg, + } + + #[pyfunction] + fn a2b_ascii85(args: A2bAscii85Args, vm: &VirtualMachine) -> PyResult> { + let A2bAscii85Args { + data, + foldspaces, + adobe, + ignorechars, + canonical, + } = args; + let result = (|| { + // This is the last Clinic conversion. Keep acquired exports alive + // and reach ordered cleanup even when __bool__ raises. + let canonical = canonical_argument(&canonical, vm)?; + let ignore = ignored_argument(&ignorechars); + let result = data.with_ref(|data| { + base_n::decode85( + data, + &base_n::reverse_table(base_n::ASCII85, None), + &ignore, + canonical, + true, + foldspaces.into_bool(), + adobe.into_bool(), + ) + }); + base_n_result(result, vm) + })(); + drop(data); + drop(ignorechars); + result + } + + #[pyfunction] + fn b2a_ascii85(args: B2aAscii85Args, vm: &VirtualMachine) -> PyResult> { + let B2aAscii85Args { + data, + foldspaces, + wrapcol, + pad, + adobe, + } = args; + let result = data.with_ref(|data| { + base_n::encode85( + data, + pad.into_bool(), + wrapcol, + base_n::ASCII85, + true, + foldspaces.into_bool(), + adobe.into_bool(), + ) + }); + base_n_result(result, vm) + } + #[derive(FromArgs)] struct A2bQpArgs { #[pyarg(any)] diff --git a/crates/stdlib/src/binascii/base_n.rs b/crates/stdlib/src/binascii/base_n.rs new file mode 100644 index 00000000000..fb966a9c3d4 --- /dev/null +++ b/crates/stdlib/src/binascii/base_n.rs @@ -0,0 +1,431 @@ +// Scalar native equivalents of CPython 3.15 Modules/binascii.c base-N codecs. + +pub(super) const BASE32: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"; +pub(super) const BASE85: &[u8] = + b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz!#$%&()*+-;<=>?@^_`{|}~"; +pub(super) const ASCII85: &[u8] = + b"!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstu"; + +#[derive(Debug)] +pub(super) enum Error { + InvalidAlphabet(usize), + Codec(String), + Memory, +} + +type Result = core::result::Result; + +fn codec(message: impl Into) -> Error { + Error::Codec(message.into()) +} + +fn buffer(size: usize, too_large: &str) -> Result> { + if size > isize::MAX as usize { + return Err(codec(too_large)); + } + let mut result = Vec::new(); + result.try_reserve_exact(size).map_err(|_| Error::Memory)?; + Ok(result) +} + +fn checked_alphabet(alphabet: &[u8], size: usize) -> Result<&[u8]> { + if alphabet.len() != size { + return Err(Error::InvalidAlphabet(size)); + } + Ok(alphabet) +} + +pub(super) fn copy_alphabet(alphabet: &[u8], size: usize) -> Result> { + checked_alphabet(alphabet, size)?; + let mut out = buffer(size, "Too large alphabet")?; + out.extend_from_slice(alphabet); + Ok(out) +} + +pub(super) fn reverse_table(alphabet: &[u8], padchar: Option) -> [u8; 256] { + let mut table = [u8::MAX; 256]; + for (digit, &ch) in alphabet.iter().enumerate() { + // Duplicate characters are allowed; their last occurrence wins. + table[usize::from(ch)] = digit as u8; + } + if let Some(padchar) = padchar { + table[usize::from(padchar)] = alphabet.len() as u8; + } + table +} + +pub(super) fn ignored(ignorechars: &[u8]) -> [bool; 256] { + let mut table = [false; 256]; + for &ch in ignorechars { + table[usize::from(ch)] = true; + } + table +} + +fn rounded_width(width: usize, quantum: usize) -> usize { + if width == 0 { + 0 + } else { + width.max(quantum) / quantum * quantum + } +} + +fn wrapped_size(size: usize, width: usize, newline: bool, too_large: &str) -> Result { + let breaks = if width == 0 || size == 0 { + 0 + } else { + (size - 1) / width + }; + size.checked_add(breaks) + .and_then(|n| n.checked_add(usize::from(newline))) + .filter(|n| isize::try_from(*n).is_ok()) + .ok_or_else(|| codec(too_large)) +} + +fn wrap(mut input: Vec, width: usize, newline: bool, too_large: &str) -> Result> { + let size = wrapped_size(input.len(), width, newline, too_large)?; + if width != 0 && input.len() > width { + let mut out = buffer(size, too_large)?; + for (index, chunk) in input.chunks(width).enumerate() { + if index != 0 { + out.push(b'\n'); + } + out.extend_from_slice(chunk); + } + input = out; + } + if newline { + input.try_reserve(1).map_err(|_| Error::Memory)?; + input.push(b'\n'); + } + Ok(input) +} + +pub(super) fn encode32( + data: &[u8], + padded: bool, + width: usize, + alphabet: &[u8], +) -> Result> { + let (size, in_group, out_group, too_large) = (32, 5, 8, "Too much data for base32"); + let bits = 5; + checked_alphabet(alphabet, size)?; + let full_size = data + .len() + .div_ceil(in_group) + .checked_mul(out_group) + .ok_or_else(|| codec(too_large))?; + let pads = (in_group - data.len() % in_group) % in_group * out_group / in_group; + let encoded_size = full_size - if padded { 0 } else { pads }; + let width = rounded_width(width, out_group); + wrapped_size(encoded_size, width, false, too_large)?; + let mut out = buffer(encoded_size, too_large)?; + let mut value = 0u32; + let mut available = 0u32; + let mask = (1u32 << bits) - 1; + for &byte in data { + value = (value << 8) | u32::from(byte); + available += 8; + while available >= bits { + available -= bits; + out.push(alphabet[((value >> available) & mask) as usize]); + } + value &= (1u32 << available) - 1; + } + if available != 0 { + out.push(alphabet[((value << (bits - available)) & mask) as usize]); + } + if padded { + out.resize(encoded_size, b'='); + } + wrap(out, width, false, too_large) +} + +pub(super) fn decode32( + data: &[u8], + table: &[u8; 256], + ignorechars: &[bool; 256], + padded: bool, + canonical: bool, +) -> Result> { + let (size, group, decoded_group, name) = (32, 8, 5, "base32"); + let bits = 5; + let ignore = ignorechars; + let capacity = data + .len() + .div_ceil(group) + .checked_mul(decoded_group) + .ok_or_else(|| codec(format!("Too much {name} data")))?; + let mut out = buffer(capacity, &format!("Too much {name} data"))?; + let mut pos = 0usize; + let mut pads = 0usize; + let mut value = 0u32; + let mut available = 0u32; + for &ch in data { + if padded && ch == b'=' { + pads += 1; + let allowed_pos = matches!(pos, 2 | 4 | 5 | 7); + if allowed_pos && pos + pads <= group { + continue; + } + if ignore[usize::from(b'=')] { + continue; + } + if matches!(pos, 1 | 3 | 6) { + break; + } + return Err(codec(if pos == 0 && out.is_empty() { + "Leading padding not allowed" + } else { + "Excess padding not allowed" + })); + } + let digit = table[usize::from(ch)]; + if usize::from(digit) >= size { + if !ignore[usize::from(ch)] { + return Err(codec(if ch == b'=' { + "Padding not allowed".to_owned() + } else { + format!("Only {name} data is allowed") + })); + } + continue; + } + if pads != 0 && !ignore[usize::from(b'=')] { + return Err(codec(if pos + pads == group { + "Excess data after padding" + } else { + "Discontinuous padding not allowed" + })); + } + value = (value << bits) | u32::from(digit); + available += bits; + if available >= 8 { + available -= 8; + out.push((value >> available) as u8); + value &= (1u32 << available) - 1; + } + pos += 1; + if pos == group { + pos = 0; + pads = 0; + } + } + if matches!(pos, 1 | 3 | 6) { + let count = out.len() / decoded_group * group + pos; + return Err(codec(format!( + "Invalid base32-encoded string: number of data characters ({count}) cannot be 1, 3, or 6 more than a multiple of 8" + ))); + } + if padded && pos != 0 && pos + pads < group { + return Err(codec("Incorrect padding")); + } + if canonical && value != 0 { + return Err(codec("Non-zero padding bits")); + } + Ok(out) +} + +fn digits85(mut value: u32, alphabet: &[u8]) -> [u8; 5] { + let mut out = [0; 5]; + for slot in out.iter_mut().rev() { + *slot = alphabet[(value % 85) as usize]; + value /= 85; + } + out +} + +pub(super) fn encode85( + data: &[u8], + pad: bool, + width: usize, + alphabet: &[u8], + ascii85: bool, + foldspaces: bool, + adobe: bool, +) -> Result> { + checked_alphabet(alphabet, 85)?; + let too_large = if ascii85 { + "Too much data for Ascii85" + } else { + "Too much data for Base85" + }; + let full_size = data + .len() + .div_ceil(4) + .checked_mul(5) + .and_then(|n| n.checked_add(if adobe { 4 } else { 0 })) + .ok_or_else(|| codec(too_large))?; + let size = full_size + - if !pad && !data.len().is_multiple_of(4) { + 4 - data.len() % 4 + } else { + 0 + }; + let width = if ascii85 { + if adobe && width == 1 { 2 } else { width } + } else { + rounded_width(width, 5) + }; + wrapped_size(size, width, false, too_large)?; + let mut out = buffer(size, too_large)?; + if adobe { + out.extend_from_slice(b"<~"); + } + for chunk in data.chunks(4) { + let mut block = [0; 4]; + block[..chunk.len()].copy_from_slice(chunk); + let value = u32::from_be_bytes(block); + if ascii85 && value == 0 && (pad || chunk.len() == 4) { + out.push(b'z'); + } else if ascii85 && foldspaces && value == 0x20202020 && chunk.len() == 4 { + out.push(b'y'); + } else { + let digits = digits85(value, alphabet); + let count = if pad || chunk.len() == 4 { + 5 + } else { + chunk.len() + 1 + }; + out.extend_from_slice(&digits[..count]); + } + } + if adobe { + out.extend_from_slice(b"~>"); + } + let mut out = wrap(out, width, false, too_large)?; + if adobe && out.len() >= 3 && out[out.len() - 2] == b'\n' { + let len = out.len(); + out[len - 3] = b'\n'; + out[len - 2] = b'~'; + } + Ok(out) +} + +pub(super) fn decode85( + mut data: &[u8], + table: &[u8; 256], + ignorechars: &[bool; 256], + canonical: bool, + ascii85: bool, + foldspaces: bool, + adobe: bool, +) -> Result> { + if adobe { + if !data.ends_with(b"~>") { + return Err(codec("Ascii85 encoded byte sequences must end with b'~>'")); + } + data = &data[..data.len() - 2]; + if data.starts_with(b"<~") { + data = &data[2..]; + } + } + let ignore = ignorechars; + let shortcuts = if ascii85 { + data.iter().filter(|ch| matches!(ch, b'y' | b'z')).count() + } else { + 0 + }; + let size = (data.len() - shortcuts) + .div_ceil(5) + .checked_mul(4) + .and_then(|n| shortcuts.checked_mul(4).and_then(|m| n.checked_add(m))) + .ok_or_else(|| codec("Too much Ascii85 data"))?; + let mut out = buffer(size, "Too much Ascii85 data")?; + let mut value = 0u32; + let mut pos = 0usize; + let mut from_z = false; + let mut index = 0usize; + let mut tail_len = 4usize; + while index < data.len() || pos != 0 { + let (ch, digit) = if index < data.len() { + let ch = data[index]; + (ch, table[usize::from(ch)]) + } else { + tail_len -= 1; + (0, 84) + }; + if digit < 85 { + value = value + .checked_mul(85) + .and_then(|n| n.checked_add(u32::from(digit))) + .ok_or_else(|| { + codec(if ascii85 { + "Ascii85 overflow".to_owned() + } else { + format!("Base85 overflow in hunk starting at byte {}", index / 5 * 5) + }) + })?; + pos += 1; + } else if ascii85 && (ch == b'z' || (ch == b'y' && foldspaces)) { + if pos != 0 { + return Err(codec(format!( + "'{}' inside Ascii85 5-tuple", + char::from(ch) + ))); + } + value = if ch == b'y' { 0x20202020 } else { 0 }; + from_z = ch == b'z'; + pos = 5; + } else if !ignore[usize::from(ch)] { + return Err(codec(if ascii85 { + format!("Non-Ascii85 digit found: {}", char::from(ch)) + } else { + format!("bad Base85 character at position {index}") + })); + } + index += 1; + if pos != 5 { + continue; + } + if tail_len == 0 { + return Err(codec(if ascii85 { + "Incomplete Ascii85 group" + } else { + "Incomplete Base85 group" + })); + } + out.extend_from_slice(&value.to_be_bytes()[..tail_len]); + if canonical { + if ascii85 && tail_len == 4 && value == 0 && !from_z { + return Err(codec("Non-canonical encoding, use 'z' for all-zero groups")); + } + if tail_len < 4 { + let pads = 4 - tail_len; + let top = value >> (pads * 8) << (pads * 8); + let divisor = 85u32.pow(pads as u32); + if top / divisor != value / divisor { + return Err(codec("Non-zero padding bits")); + } + } + } + pos = 0; + value = 0; + from_z = false; + } + Ok(out) +} + +pub(super) fn decode_hex(data: &[u8], ignorechars: &[bool; 256]) -> Result> { + let ignore = ignorechars; + let mut out = buffer(data.len() / 2, "Too much hexadecimal data")?; + let mut high = None; + for &ch in data { + let digit = match ch { + b'0'..=b'9' => ch - b'0', + b'a'..=b'f' => ch - b'a' + 10, + b'A'..=b'F' => ch - b'A' + 10, + _ if ignore[usize::from(ch)] => continue, + _ => return Err(codec("Non-hexadecimal digit found")), + }; + if let Some(hi) = high.take() { + out.push((hi << 4) | digit); + } else { + high = Some(digit); + } + } + if high.is_some() { + return Err(codec("Odd number of hexadecimal digits")); + } + Ok(out) +} diff --git a/extra_tests/snippets/stdlib_binascii_base64.py b/extra_tests/snippets/stdlib_binascii_base64.py deleted file mode 100644 index 4daac629a14..00000000000 --- a/extra_tests/snippets/stdlib_binascii_base64.py +++ /dev/null @@ -1,363 +0,0 @@ -"""Bounded CPython 3.15 base64 API regressions (no other binascii codecs).""" - -import binascii - -from testutils import assert_raises - -encode = binascii.b2a_base64 -decode = binascii.a2b_base64 -alphabet = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" -assert binascii.BASE64_ALPHABET == alphabet -assert binascii.URLSAFE_BASE64_ALPHABET == alphabet[:-2] + b"-_" - -# RFC 4648 vectors and CPython's four-column wrapping normalization. -for raw, expected in ( - (b"", b""), - (b"f", b"Zg=="), - (b"fo", b"Zm8="), - (b"foo", b"Zm9v"), - (b"foob", b"Zm9vYg=="), - (b"fooba", b"Zm9vYmE="), - (b"foobar", b"Zm9vYmFy"), - (b"www.python.org", b"d3d3LnB5dGhvbi5vcmc="), -): - for factory in bytes, bytearray, memoryview: - for padded in False, True: - unwrapped = expected if padded else expected.rstrip(b"=") - for width in 0, 1, 3, 4, 5, 7, 8, 11, 76: - normalized = max(4, width // 4 * 4) if width else 1000 - wrapped = b"\n".join( - unwrapped[i : i + normalized] - for i in range(0, len(unwrapped), normalized) - ) - for newline in False, True: - result = encode( - factory(raw), padded=padded, wrapcol=width, newline=newline - ) - assert result == wrapped + (b"\n" if newline else b"") - assert ( - decode( - factory(result), - padded=padded, - ignorechars=b"\n", - canonical=True, - ) - == raw - ) - -# Input and custom alphabet asymmetry: decoding accepts ASCII str, but its -# alphabet must be bytes. Encoders accept any contiguous bytes-like alphabet. -raw = bytes(range(64)) -standard = encode(raw, newline=False) -for custom in alphabet[::-1], alphabet[:-2] + b"-_", bytes(range(128, 192)): - expected = standard.translate(bytes.maketrans(alphabet, custom)) - for factory in bytes, bytearray, memoryview: - assert encode(raw, alphabet=factory(custom), newline=False) == expected - assert decode(expected, alphabet=custom, canonical=True) == raw -assert decode("Zm9v", alphabet=alphabet) == b"foo" -assert encode(b"\0", alphabet=b"A" * 64, newline=False) == b"AA==" -assert decode(b"AA==", alphabet=b"A" * 64) == b"\xff" -assert encode(b"\0", alphabet=b"=" * 64, newline=False) == b"====" -assert decode(b"====", alphabet=b"=" * 64) == b"" - -# Supplied ignorechars enables strictness by default; alphabet characters are -# never ignored. Explicit non-strict mode still ignores arbitrary non-alphabet. -assert decode(b" Zm9v!", strict_mode=False, ignorechars=b"") == b"foo" -assert decode(b" Zm9v!", ignorechars=b" !") == b"foo" -assert decode(b"Zm9v", ignorechars=b"Zm9v") == b"foo" -for value in b" ", b"!", b"\xff": - assert decode(value) == b"" - with assert_raises(binascii.Error): - decode(value, ignorechars=b"") - -for value, message in ( - (b"=", "Leading padding not allowed"), - (b" =Zm9v", "Leading padding not allowed"), - (b"Zm9v=", "Excess padding not allowed"), - (b"Zg===", "Excess padding not allowed"), - (b"Zg=a", "Discontinuous padding not allowed"), - (b"Zg==a", "Excess data after padding"), - (b"Zg", "Incorrect padding"), - (b"!", "Only base64 data is allowed"), -): - try: - decode(value, ignorechars=b" ") - except binascii.Error as error: - assert str(error) == message - else: - assert False, (value, message) - -for value, expected in (b"YQ==", b"a"), (b"YWI=", b"ab"), (b"YWJj", b"abc"): - assert decode(value.replace(b"=", b""), padded=False, strict_mode=True) == expected - assert decode(value, padded=False, ignorechars=b"=") == expected - if b"=" in value: - try: - decode(value, padded=False, strict_mode=True) - except binascii.Error as error: - assert str(error) == "Padding not allowed" - else: - assert False -assert decode(b"Y=WJj", ignorechars=b"=") == b"abc" -assert decode(b"YQ==YQ==") == b"a\x06\x10" -for value in b"AB==", b"AP==", b"AAB=", b"AAD=": - decode(value) - for padded in False, True: - try: - decode( - value if padded else value.rstrip(b"="), padded=padded, canonical=True - ) - except binascii.Error as error: - assert str(error) == "Non-zero padding bits" - else: - assert False -for value in b"A", b"A=", b"AAAAA": - with assert_raises(binascii.Error): - decode(value, padded=False) - -for function in encode, decode: - for invalid in b"", alphabet[:-1], alphabet + b"?": - with assert_raises(ValueError): - function(b"", alphabet=invalid) - for invalid in None, "", []: - with assert_raises(TypeError): - function(b"", alphabet=invalid) - with assert_raises(TypeError): - function(data=b"") - with assert_raises(TypeError): - function(b"", False) -for invalid in bytearray(alphabet), memoryview(alphabet): - with assert_raises(TypeError): - decode(b"", alphabet=invalid) -for invalid in None, "", []: - with assert_raises(TypeError): - decode(b"", ignorechars=invalid) -with assert_raises(ValueError): - encode(b"", wrapcol=-1) -with assert_raises(OverflowError): - encode(b"", wrapcol=1 << 128) -with assert_raises(TypeError): - encode(b"", wrapcol=4.0) -with assert_raises(ValueError): - decode("é") - - -class Width: - def __index__(self): - return 4 - - -assert encode(b"foobar", wrapcol=Width(), newline=False) == b"Zm9v\nYmFy" - -# Conversion order is independent of keyword order and canonical runs after -# ignorechars acquisition, before alphabet length validation. -events = [] - - -class Flag: - def __init__(self, name, value): - self.name = name - self.value = value - - def __bool__(self): - events.append(self.name) - return self.value - - -assert ( - encode(b"f", newline=Flag("newline", False), padded=Flag("padded", False)) == b"Zg" -) -assert events == ["padded", "newline"] -events.clear() -assert ( - decode( - b"Zg", - canonical=Flag("canonical", True), - padded=Flag("padded", False), - strict_mode=Flag("strict", True), - ) - == b"f" -) -assert events == ["strict", "padded", "canonical"] -events.clear() -with assert_raises(ValueError): - decode(b"", alphabet=b"", canonical=Flag("canonical", False)) -assert events == ["canonical"] - - -# The decoder cache uses the original bytes object, including subclass hooks. -class CachedAlphabet(bytes): - def __hash__(self): - events.append("hash") - return 314159265 - - def __eq__(self, other): - events.append("equal") - return bytes.__eq__(self, other) - - -custom = CachedAlphabet(alphabet[1:] + alphabet[:1]) -encoded = encode(b"abc", alphabet=custom, newline=False) -events.clear() -assert decode(encoded, alphabet=custom) == b"abc" -assert events == ["hash", "hash"] -events.clear() -assert decode(encoded, alphabet=custom) == b"abc" -assert events == ["hash"] -events.clear() -assert decode(encoded, alphabet=CachedAlphabet(custom)) == b"abc" -assert events == ["hash", "equal"] - - -class UnhashableAlphabet(bytes): - __hash__ = None - - -with assert_raises(TypeError): - decode(b"", alphabet=UnhashableAlphabet(alphabet)) -with assert_raises(ValueError): - decode(b"", alphabet=UnhashableAlphabet(b"")) -assert encode(b"abc", alphabet=UnhashableAlphabet(alphabet)) == b"YWJj\n" - - -class HashError(Exception): - pass - - -class BrokenAlphabet(bytes): - def __hash__(self): - raise HashError - - -with assert_raises(HashError): - decode(b"!", alphabet=BrokenAlphabet(alphabet), strict_mode=True) - -# Keep the exact owning module alive through a saved builtin, independently of -# sys.modules. A fresh module's cache cycle must remain visible to the collector. -import gc -import importlib.util -import sys -import weakref - -assert decode.__self__ is binascii -assert decode.__module__ == "binascii" -assert decode.__name__ == "a2b_base64" -original = sys.modules.pop("binascii") -try: - assert decode(encoded, alphabet=custom) == b"abc" -finally: - sys.modules["binascii"] = original - -spec = importlib.util.find_spec("binascii") -original = sys.modules.pop("binascii") -try: - isolated = importlib.util.module_from_spec(spec) - spec.loader.exec_module(isolated) -finally: - sys.modules["binascii"] = original -assert isolated is not binascii -assert isolated.a2b_base64.__self__ is isolated -cycle_key = CachedAlphabet(alphabet) -cycle_key.owner = isolated -assert isolated.a2b_base64(b"YWJj", alphabet=cycle_key) == b"abc" -module_ref = weakref.ref(isolated) -del cycle_key, isolated -for _ in range(3): - gc.collect() -assert module_ref() is None -assert not hasattr(binascii, "_reverse_table_cache") - - -# Hash callbacks can safely reenter the same module before decoding begins. -class ReentrantAlphabet(bytes): - def __hash__(self): - assert decode(b"Zm9v") == b"foo" - assert encode(b"foo", wrapcol=4) == b"Zm9v\n" - events.append("reenter") - return 271828182 - - -events.clear() -assert decode(b"YWJj", alphabet=ReentrantAlphabet(alphabet)) == b"abc" -assert events == ["reenter", "reenter"] -assert "ignorechars=" in decode.__text_signature__ -assert "wrapcol=0" in encode.__text_signature__ - - -# Acquired input and auxiliary exports survive conversions and cache callbacks, -# and are released in Clinic order, including a canonical conversion failure. -class Export: - def __init__(self, name, data): - self.name = name - self.data = data - - def __buffer__(self, flags): - events.append("acquire " + self.name) - return memoryview(self.data) - - def __release_buffer__(self, view): - events.append("release " + self.name) - - -class BrokenFlag: - def __bool__(self): - events.append("canonical error") - raise HashError - - -events.clear() -assert ( - encode(Export("data", b"foo"), alphabet=Export("alphabet", alphabet)) == b"Zm9v\n" -) -assert events == [ - "acquire data", - "acquire alphabet", - "release data", - "release alphabet", -] -events.clear() -assert decode(Export("data", b"Zm9v"), ignorechars=Export("ignored", b"")) == b"foo" -assert events == ["acquire data", "acquire ignored", "release data", "release ignored"] -events.clear() -with assert_raises(HashError): - decode( - Export("data", b""), ignorechars=Export("ignored", b""), canonical=BrokenFlag() - ) -assert events == [ - "acquire data", - "acquire ignored", - "canonical error", - "release data", - "release ignored", -] - -# Aliased buffer exporters do not require nested backing-store locks. -shared = bytearray(alphabet) -assert encode(shared, alphabet=shared) == encode(bytes(shared), alphabet=bytes(shared)) -shared = bytearray(b"Zm9v") -assert decode(shared, ignorechars=shared) == b"foo" - -for length in (0, 1, 2, 3, 4, 5, 56, 57, 58, 63, 64, 65, 255, 256, 257): - payload = (bytes(range(256)) * 2)[:length] - for padded in False, True: - for width in 0, 4, 76: - encoded = encode(payload, padded=padded, wrapcol=width) - assert ( - decode(encoded, padded=padded, ignorechars=b"\n", canonical=True) - == payload - ) - - -# Bound module functions retain their metadata for module subclasses too. -class ModuleSubclass(type(binascii)): - pass - - -original_class = binascii.__class__ -try: - binascii.__class__ = ModuleSubclass - assert repr(decode) == "" - assert decode.__qualname__ == decode.__name__ == "a2b_base64" - assert decode.__self__ is binascii - assert decode.__reduce__() == "a2b_base64" -finally: - binascii.__class__ = original_class