meta-openembedded/meta-python/recipes-devtools/python/files/CVE-2026-42304_p5.patch

From 4aa0fdb33f2a12db2d234754b9b03d74df8eaf9d Mon Sep 17 00:00:00 2001
From: Tomas Illuminati <tomas.illuminati@owasp.org>
Date: Tue, 21 Apr 2026 17:26:49 -0300
Subject: [PATCH] names: fix changes

CVE: CVE-2026-42304
Upstream-Status: Backport [https://github.com/twisted/twisted/commit/9df6d960d3569751ebb5567093fe1d1d9f63ca54]

Backport Changes:
- Retained the 25.5.0 Optional and Union typing imports while removing
  Final; Sequence remains sourced from collections.abc after p4.

(cherry picked from commit 9df6d960d3569751ebb5567093fe1d1d9f63ca54)
Signed-off-by: Hetvi Thakar <hthakar@cisco.com>
---
 src/twisted/names/dns.py           | 37 +++++++------
 src/twisted/names/test/test_dns.py | 85 +++++++++++++-----------------
 2 files changed, 56 insertions(+), 66 deletions(-)

diff --git a/src/twisted/names/dns.py b/src/twisted/names/dns.py
index 869ffec76..ca5079454 100644
--- a/src/twisted/names/dns.py
+++ b/src/twisted/names/dns.py
@@ -20,7 +20,7 @@ from collections.abc import Sequence
 from contextlib import contextmanager
 from io import BytesIO
 from itertools import chain
-from typing import Final, Optional, SupportsInt, Union, overload
+from typing import Optional, SupportsInt, Union, overload
 
 from zope.interface import Attribute, Interface, implementer
 
@@ -448,15 +448,6 @@ def readPrecisely(file, l):
     return buff
 
 
-MAX_COMPRESSION_POINTERS_PER_MESSAGE: Final = 1000
-"""
-Cap the total number of compression-pointer dereferences performed while
-decoding a single DNS message.  A hostile peer can otherwise craft a packet
-in which every record name chases a long compression chain, forcing
-C{O(N*M)} work and stalling the reactor.
-"""
-
-
 class DNSDecodeError(ValueError):
     """
     Raised when a DNS message cannot be decoded because it violates a
@@ -484,7 +475,7 @@ class _DecodeContext:
 
     __slots__ = ("jumps", "maxJumps")
 
-    def __init__(self, maxJumps: int = MAX_COMPRESSION_POINTERS_PER_MESSAGE) -> None:
+    def __init__(self, maxJumps: int = 1000) -> None:
         self.jumps = 0
         self.maxJumps = maxJumps
 
@@ -644,16 +635,15 @@ class Name:
     @ivar name: A byte string giving the name.
     @type name: L{bytes}
 
-    @cvar maxCompressionPointers: Per-message cap on the total number of
+    @ivar maxCompressionPointers: Per-message cap on the total number of
         compression-pointer dereferences L{decode} will follow before
-        raising L{DNSDecodeError}.  Defined as a class attribute so
-        subclasses (and, in the future, individual instances) may override
-        it to tune the trade-off between tolerance for legitimately
-        verbose messages and resistance to denial-of-service attacks.
-    @type maxCompressionPointers: L{int}
+        raising L{DNSDecodeError}.  Defaults to C{1000}.  Override it on
+        a subclass or individual instance to tune the trade-off between
+        tolerance for legitimately verbose messages and resistance to
+        denial-of-service attacks.
     """
 
-    maxCompressionPointers: int = MAX_COMPRESSION_POINTERS_PER_MESSAGE
+    maxCompressionPointers: int = 1000
 
     def __init__(self, name: bytes | str = b""):
         """
@@ -2610,8 +2600,17 @@ class Message(tputil.FancyEqMixin):
         header fields.
     @ivar _sectionNames: The names of attributes representing the record
         sections of this message.
+
+    @ivar maxCompressionPointers: Per-message cap on the total number of
+        compression-pointer dereferences L{decode} will follow across every
+        name in the message before raising L{DNSDecodeError}.  Defaults to
+        C{1000}.  Override it on a subclass or individual instance to tune
+        the trade-off between tolerance for legitimately verbose messages
+        and resistance to denial-of-service attacks.
     """
 
+    maxCompressionPointers: int = 1000
+
     compareAttributes = (
         "id",
         "answer",
@@ -2830,7 +2829,7 @@ class Message(tputil.FancyEqMixin):
         # performed across every name in this message.  It is installed on
         # the private context variable so nested record decoders pick it up
         # without needing to thread it through each signature.
-        decodeContext = _DecodeContext(maxJumps=Name.maxCompressionPointers)
+        decodeContext = _DecodeContext(maxJumps=self.maxCompressionPointers)
         with _installDecodeContext(decodeContext):
             self.queries = []
             for i in range(nqueries):
diff --git a/src/twisted/names/test/test_dns.py b/src/twisted/names/test/test_dns.py
index 9626115ab..3be6b4546 100644
--- a/src/twisted/names/test/test_dns.py
+++ b/src/twisted/names/test/test_dns.py
@@ -354,60 +354,51 @@ class NameTests(unittest.TestCase):
 
     def test_rejectTooManyCompressionPointers(self):
         """
-        L{Name.decode} raises L{dns.DNSDecodeError} when the number of
-        compression-pointer dereferences taken for a single message exceeds
-        the limit carried by the shared L{dns._DecodeContext} installed
-        through the private L{dns._decodeContextVar}.
-        """
-        # Five distinct pointers chained end-to-end, terminated by a zero
-        # label byte.  With a maxJumps of three the fourth dereference must
-        # trip the safety limit.
+        L{Name.decode} raises L{dns.DNSDecodeError} when it would have to
+        follow more than L{Name.maxCompressionPointers} compression
+        pointers to finish decoding a name.
+        """
+        # Four distinct pointers chained end-to-end, terminated by a zero
+        # label byte.  With maxCompressionPointers of three the fourth
+        # dereference must trip the safety limit.
         payload = b"\xc0\x02\xc0\x04\xc0\x06\xc0\x08\x00"
-        context = dns._DecodeContext(maxJumps=3)
-        with dns._installDecodeContext(context):
-            self.assertRaises(
-                dns.DNSDecodeError,
-                dns.Name().decode,
-                BytesIO(payload),
-            )
+        name = dns.Name()
+        name.maxCompressionPointers = 3
+        self.assertRaises(
+            dns.DNSDecodeError, name.decode, BytesIO(payload)
+        )
 
-    def test_compressionPointerCounterIsShared(self):
+    def test_decodeRecoversAfterDNSDecodeError(self):
         """
-        The L{dns._DecodeContext} counter accumulates across successive
-        L{Name.decode} calls, so that a message whose individual names are
-        each within bounds is still rejected when their aggregate exceeds
-        the configured limit.  This mirrors production: L{Message.decode}
-        invokes L{Name.decode} many times against the same stream under one
-        shared context.
+        After L{Name.decode} raises L{dns.DNSDecodeError}, subsequent
+        L{Name.decode} calls continue to work.  No residual
+        compression-pointer counter leaks across calls, so a legitimate
+        name decoded right after a hostile one still succeeds.
         """
-        payload = b"\xc0\x02\xc0\x04\x00"
-        context = dns._DecodeContext(maxJumps=3)
-
-        with dns._installDecodeContext(context):
-            stream = BytesIO(payload)
-            dns.Name().decode(stream)
-            self.assertEqual(context.jumps, 2)
-
-            stream.seek(0)
-            self.assertRaises(
-                dns.DNSDecodeError,
-                dns.Name().decode,
-                stream,
-            )
+        # First, force a DNSDecodeError by decoding a payload that
+        # exceeds the configured limit.
+        hostile = dns.Name()
+        hostile.maxCompressionPointers = 3
+        self.assertRaises(
+            dns.DNSDecodeError,
+            hostile.decode,
+            BytesIO(b"\xc0\x02\xc0\x04\xc0\x06\xc0\x08\x00"),
+        )
 
-    def test_decodeWithoutContextIsBackwardsCompatible(self):
-        """
-        L{Name.decode} continues to work when called with no active
-        L{dns._decodeContextVar}, using a fresh per-call counter seeded
-        from L{dns.Name.maxCompressionPointers} so existing callers are
-        unaffected.
-        """
-        name = dns.Name()
+        # Then prove the process has not been poisoned: a legitimate
+        # name still decodes normally, both with a fresh instance and
+        # with the instance that just errored.
         stream = BytesIO()
         dns.Name(b"example.org").encode(stream)
+
+        fresh = dns.Name()
         stream.seek(0)
-        name.decode(stream)
-        self.assertEqual(name.name, b"example.org")
+        fresh.decode(stream)
+        self.assertEqual(fresh.name, b"example.org")
+
+        stream.seek(0)
+        hostile.decode(stream)
+        self.assertEqual(hostile.name, b"example.org")
 
     def test_equality(self):
         """
@@ -823,7 +814,7 @@ class MessageTests(unittest.SynchronousTestCase):
         L{Message.decode} installs a shared compression-pointer counter and
         raises L{dns.DNSDecodeError} when the aggregate number of pointer
         dereferences across every record in the message exceeds
-        L{dns.MAX_COMPRESSION_POINTERS_PER_MESSAGE}.
+        L{dns.Message.maxCompressionPointers}.
         """
         chainLength = 100
         numRecords = 8000