From 0c13737586cde92720c41aced8bf98352328e900 Mon Sep 17 00:00:00 2001
From: Aslan Osmanov <osmnvaslan@gmail.com>
Date: Sat, 24 Oct 2026 14:04:51 +0000
Subject: [PATCH] gh-160308: Buffer input in the punycode incremental codec
 until final

Punycode encodes the positions of non-ASCII characters relative to the whole string, so chunks cannot be transformed independently. Use BufferedIncrementalEncoder/Decoder and only transform the input once final is true.
---
 Lib/encodings/punycode.py                     | 23 +++++++++---
 Lib/test/test_codecs.py                       | 37 +++++++++++++++++++
 ...-10-24-14-04-51.gh-issue-160308.Pq7nYc.rst |  3 ++
 3 files changed, 58 insertions(+), 5 deletions(-)
 create mode 100644 Misc/NEWS.d/next/Library/2026-10-24-14-04-51.gh-issue-160308.Pq7nYc.rst

diff --git a/Lib/encodings/punycode.py b/Lib/encodings/punycode.py
index 279245f435e1791..102915084849c06 100644
--- a/Lib/encodings/punycode.py
+++ b/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
diff --git a/Lib/test/test_codecs.py b/Lib/test/test_codecs.py
index 37ac57063c35390..edccde2ae8bb9f0 100644
--- a/Lib/test/test_codecs.py
+++ b/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)
diff --git a/Misc/NEWS.d/next/Library/2026-10-24-14-04-51.gh-issue-160308.Pq7nYc.rst b/Misc/NEWS.d/next/Library/2026-10-24-14-04-51.gh-issue-160308.Pq7nYc.rst
new file mode 100644
index 000000000000000..309add21c988dd9
--- /dev/null
+++ b/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.