Add uuencoding/decoding code to binascii.rs

The code is based on the PyPy implementation. It may not work properly.

vm/src/stdlib/binascii.rs

@@ -177,31 +177,109 @@ 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);
+
+            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 Err(vm.new_value_error("Trailing garbage".to_string()));
+                }
+
+                if res.len() < length {
+                    res.push((char_b & 0xf) | char_c >> 2);
+                } else if char_c != 0 {
+                    return Err(vm.new_value_error("Trailing garbage".to_string()));
+                }
+
+                if res.len() < length {
+                    res.push((char_c & 0x3) << 6 | char_d);
+                } else if char_d != 0 {
+                    return Err(vm.new_value_error("Trailing garbage".to_string()));
+                }
+            }
+
+            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,
+    }
+
+    #[inline]
+    fn uu_b2a_write(res: &mut Vec<u8>, num: u8, backtick: bool) {
+        if backtick && num != 0 {
+            res.push(0x60);
+        } else {
+            res.push(0x20 + num);
         }
-        encoded
+    }
+
+    #[pyfunction]
+    fn b2a_uu(
+        data: ArgBytesLike,
+        BacktickArg { backtick }: BacktickArg,
+        vm: &VirtualMachine,
+    ) -> PyResult<Vec<u8>> {
+        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);
+            uu_b2a_write(&mut res, 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);
+
+                uu_b2a_write(&mut res, char_a >> 2, backtick);
+                uu_b2a_write(&mut res, (char_a & 0x3) << 4 | char_b >> 4, backtick);
+                uu_b2a_write(&mut res, (char_b & 0xf) << 2 | char_c >> 6, backtick);
+                uu_b2a_write(&mut res, char_c & 0x3f, backtick);
+            }
+
+            res.push(0xau8);
+            Ok(res)
+        })
     }
 }