gh-160308: Buffer input in the punycode incremental codec until final

Lib/encodings/punycode.py

@@ -223,15 +223,28 @@ def decode(self, input, errors='strict'):
         res = punycode_decode(input, errors)
         return res, len(input)
 
-class IncrementalEncoder(codecs.IncrementalEncoder):
-    def encode(self, input, final=False):
-        return punycode_encode(input)
+# Punycode encodes the positions of non-ASCII characters relative to the
+# whole string, so a chunk cannot be transformed on its own: buffer the
+# input until final is true.
 
-class IncrementalDecoder(codecs.IncrementalDecoder):
+class IncrementalEncoder(codecs.BufferedIncrementalEncoder):
+    def _buffer_encode(self, input, errors, final):
+        if not final:
+            return b'', 0
+        return punycode_encode(input), len(input)
+
+class IncrementalDecoder(codecs.BufferedIncrementalDecoder):
     def decode(self, input, final=False):
         if self.errors not in ('strict', 'replace', 'ignore'):
             raise UnicodeError(f"Unsupported error handling: {self.errors}")
-        return punycode_decode(input, self.errors)
+        if isinstance(input, str):
+            input = input.encode('ascii')
+        return super().decode(bytes(input), final)
+
+    def _buffer_decode(self, input, errors, final):
+        if not final:
+            return '', 0
+        return punycode_decode(input, errors), len(input)
 
 class StreamWriter(Codec,codecs.StreamWriter):
     pass

Lib/test/test_codecs.py

@@ -1405,6 +1405,43 @@ def test_decode_invalid(self):
                 else:
                     self.assertEqual(puny.decode("punycode", errors), expected)
 
+    def test_incremental_encode(self):
+        for uni, puny in punycode_testcases:
+            with self.subTest(uni=uni):
+                encoder = codecs.getincrementalencoder("punycode")()
+                result = b"".join(encoder.encode(c) for c in uni)
+                result += encoder.encode("", final=True)
+                self.assertEqual(result, uni.encode("punycode"))
+                result = b"".join(codecs.iterencode(uni, "punycode"))
+                self.assertEqual(result, uni.encode("punycode"))
+
+    def test_incremental_decode(self):
+        for uni, puny in punycode_testcases:
+            with self.subTest(uni=uni):
+                decoder = codecs.getincrementaldecoder("punycode")()
+                result = "".join(decoder.decode(bytes([b])) for b in puny)
+                result += decoder.decode(b"", final=True)
+                self.assertEqual(result, uni)
+                chunks = (bytes([b]) for b in puny)
+                result = "".join(codecs.iterdecode(chunks, "punycode"))
+                self.assertEqual(result, uni)
+
+    def test_incremental_state(self):
+        encoder = codecs.getincrementalencoder("punycode")()
+        self.assertEqual(encoder.encode("b\xfc"), b"")
+        state = encoder.getstate()
+        encoder.reset()
+        self.assertEqual(encoder.encode("cher", final=True), b"cher-")
+        encoder.setstate(state)
+        self.assertEqual(encoder.encode("cher", final=True), b"bcher-kva")
+
+        decoder = codecs.getincrementaldecoder("punycode")()
+        self.assertEqual(decoder.decode(b"bcher-"), "")
+        self.assertEqual(decoder.decode("kva", final=True), "b\xfccher")
+        decoder = codecs.getincrementaldecoder("punycode")("ignore")
+        self.assertEqual(decoder.decode(b"xn--", final=False), "")
+        self.assertEqual(decoder.decode(b"w&", final=True), "xn-")
+
 
 utf_7_imap_testcases = [
     # (unicode, modified UTF-7)

Misc/NEWS.d/next/Library/2026-10-24-14-04-51.gh-issue-160308.Pq7nYc.rst

@@ -0,0 +1,3 @@
+The incremental encoder and decoder of the ``punycode`` codec now buffer
+their input until *final* is true, instead of transforming each chunk
+separately and producing wrong results for chunked input.