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/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/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..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,14 +10,138 @@ 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::{ - 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.", + ), + ); + 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 = 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"))? + .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 = base_n::reverse_table(bytes, padchar); + cache.set_item( + alphabet.as_object(), + vm.ctx.new_bytes(table.to_vec()).into(), + vm, + )?; + 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( @@ -54,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)] @@ -86,30 +224,418 @@ 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, + } + + 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, 64, Some(b'='), 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 a2b_base64(args: A2bBase64Args, vm: &VirtualMachine) -> PyResult> { - let A2bBase64Args { data, strict_mode } = args; - data.with_ref(|b| binascii::a2b_base64(b, strict_mode)) + 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 + } + + 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_base64(data: ArgBytesLike, NewlineArg { newline }: NewlineArg) -> Vec { - data.with_ref(|bytes| binascii::b2a_base64(bytes, newline)) + 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)] @@ -182,6 +708,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 +727,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/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/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);