Merge pull request #3042 from fanninpm/uu-encoding

Add uuencoding/decoding code to binascii.rs

Lib/test/test_uu.py

@@ -92,8 +92,6 @@ class UUTest(unittest.TestCase):
             uu.decode(inp, out)
             self.assertEqual(out.getvalue(), plaintext)
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
     def test_truncatedinput(self):
         inp = io.BytesIO(b"begin 644 t1\n" + encodedtext)
         out = io.BytesIO()
@@ -239,8 +237,6 @@ class UUFileTest(unittest.TestCase):
             s = f.read()
         self.assertEqual(s, plaintext)
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
     def test_decodetwice(self):
         # Verify that decode() will refuse to overwrite an existing file
         with open(self.tmpin, 'wb') as f:

vm/src/stdlib/binascii.rs

@@ -177,31 +177,110 @@ mod decl {
         encoded
     }
 
-    #[pyfunction]
-    fn a2b_uu(s: SerializedData, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
-        s.with_ref(|b| {
-            let mut buf;
-            let b = if memchr::memchr(b'\n', b).is_some() {
-                buf = b.to_vec();
-                buf.retain(|c| *c != b'\n');
-                &buf
-            } else {
-                b
-            };
-            // TODO: RUSTPYTHON, implement actual uuencoding code
-            base64::decode(b)
-        })
-        .map_err(|err| vm.new_value_error(format!("error decoding uuencode: {}", err)))
+    #[inline]
+    fn uu_a2b_read(c: &u8, vm: &VirtualMachine) -> PyResult<u8> {
+        // Check the character for legality
+        // The 64 instead of the expected 63 is because
+        // there are a few uuencodes out there that use
+        // '`' as zero instead of space.
+        if !(0x20..=0x60).contains(c) {
+            if [b'\r', b'\n'].contains(c) {
+                return Ok(0);
+            }
+            return Err(vm.new_value_error("Illegal char".to_string()));
+        }
+        Ok((*c - 0x20) & 0x3f)
     }
 
     #[pyfunction]
-    fn b2a_uu(data: ArgBytesLike, NewlineArg { newline }: NewlineArg) -> Vec<u8> {
-        #[allow(clippy::redundant_closure)] // https://stackoverflow.com/questions/63916821
-        // TODO: RUSTPYTHON, implement actual uuencoding code
-        let mut encoded = data.with_ref(|b| base64::encode(b)).into_bytes();
-        if newline {
-            encoded.push(b'\n');
+    fn a2b_uu(s: SerializedData, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
+        s.with_ref(|b| {
+            // First byte: binary data length (in bytes)
+            let length = if b.is_empty() {
+                ((-0x20i32) & 0x3fi32) as usize
+            } else {
+                ((b[0] - 0x20) & 0x3f) as usize
+            };
+
+            // Allocate the buffer
+            let mut res = Vec::<u8>::with_capacity(length);
+            let trailing_garbage_error = || Err(vm.new_value_error("Trailing garbage".to_string()));
+
+            for chunk in b.get(1..).unwrap_or_default().chunks(4) {
+                let char_a = chunk.get(0).map_or(Ok(0), |x| uu_a2b_read(x, vm))?;
+                let char_b = chunk.get(1).map_or(Ok(0), |x| uu_a2b_read(x, vm))?;
+                let char_c = chunk.get(2).map_or(Ok(0), |x| uu_a2b_read(x, vm))?;
+                let char_d = chunk.get(3).map_or(Ok(0), |x| uu_a2b_read(x, vm))?;
+
+                if res.len() < length {
+                    res.push(char_a << 2 | char_b >> 4);
+                } else if char_a != 0 || char_b != 0 {
+                    return trailing_garbage_error();
+                }
+
+                if res.len() < length {
+                    res.push((char_b & 0xf) | char_c >> 2);
+                } else if char_c != 0 {
+                    return trailing_garbage_error();
+                }
+
+                if res.len() < length {
+                    res.push((char_c & 0x3) << 6 | char_d);
+                } else if char_d != 0 {
+                    return trailing_garbage_error();
+                }
+            }
+
+            let remaining_length = length - res.len();
+            if remaining_length > 0 {
+                res.extend(vec![0; remaining_length]);
+            }
+            Ok(res)
+        })
+    }
+
+    #[derive(FromArgs)]
+    struct BacktickArg {
+        #[pyarg(named, default = "true")]
+        backtick: bool,
+    }
+
+    #[pyfunction]
+    fn b2a_uu(
+        data: ArgBytesLike,
+        BacktickArg { backtick }: BacktickArg,
+        vm: &VirtualMachine,
+    ) -> PyResult<Vec<u8>> {
+        #[inline]
+        fn uu_b2a(num: u8, backtick: bool) -> u8 {
+            if backtick && num != 0 {
+                0x60
+            } else {
+                0x20 + num
+            }
         }
-        encoded
+
+        data.with_ref(|b| {
+            let length = b.len();
+            if length > 45 {
+                return Err(vm.new_value_error("At most 45 bytes at once".to_string()));
+            }
+            let mut res = Vec::<u8>::with_capacity(2 + ((length + 2) / 3) * 4);
+            res.push(uu_b2a(length as u8, backtick));
+
+            for chunk in b.chunks(3) {
+                let char_a = *chunk.get(0).unwrap_or(&0);
+                let char_b = *chunk.get(1).unwrap_or(&0);
+                let char_c = *chunk.get(2).unwrap_or(&0);
+
+                res.push(uu_b2a(char_a >> 2, backtick));
+                res.push(uu_b2a((char_a & 0x3) << 4 | char_b >> 4, backtick));
+                res.push(uu_b2a((char_b & 0xf) << 2 | char_c >> 6, backtick));
+                res.push(uu_b2a(char_c & 0x3f, backtick));
+            }
+
+            res.push(0xau8);
+            Ok(res)
+        })
     }
 }