1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
From e33f7fc231845487f969a9c0fbf7956226ac8dfa Mon Sep 17 00:00:00 2001
From: Tom Most <twm@freecog.net>
Date: Sun, 13 Mar 2022 23:51:52 -0700
Subject: [PATCH] Reject malformed chunk sizes
Upstream-Status: Backport [https://github.com/twisted/twisted/commit/0275152f147506c82868ff1dabd9bf655ab67946]
CVE: CVE-2022-24801
Signed-off-by: Gyorgy Sarvari <skandigraun@gmail.com>
---
src/twisted/web/http.py | 35 +++++++++++++++++++++++----
src/twisted/web/test/test_http.py | 40 +++++++++++++++++++++++++++++++
2 files changed, 71 insertions(+), 4 deletions(-)
diff --git a/src/twisted/web/http.py b/src/twisted/web/http.py
index 5316d81..940ff9f 100644
--- a/src/twisted/web/http.py
+++ b/src/twisted/web/http.py
@@ -108,7 +108,7 @@ import tempfile
import time
import warnings
from io import BytesIO
-from typing import AnyStr, Callable, Optional
+from typing import AnyStr, Callable, Optional, Tuple
from urllib.parse import (
ParseResultBytes,
unquote_to_bytes as unquote,
@@ -410,7 +410,33 @@ def toChunk(data):
return (networkString(f"{len(data):x}"), b"\r\n", data, b"\r\n")
-def fromChunk(data):
+def _ishexdigits(b: bytes) -> bool:
+ """
+ Is the string case-insensitively hexidecimal?
+
+ It must be composed of one or more characters in the ranges a-f, A-F
+ and 0-9.
+ """
+ for c in b:
+ if c not in b'0123456789abcdefABCDEF':
+ return False
+ return bool(b)
+
+
+def _hexint(b: bytes) -> int:
+ """
+ Decode a hexadecimal integer.
+
+ Unlike L{int(b, 16)}, this raises L{ValueError} when the integer has
+ a prefix like C{b'0x'}, C{b'+'}, or C{b'-'}, which is desirable when
+ parsing network protocols.
+ """
+ if not _ishexdigits(b):
+ raise ValueError(b)
+ return int(b, 16)
+
+
+def fromChunk(data: bytes) -> Tuple[bytes, bytes]:
"""
Convert chunk to string.
@@ -422,7 +448,7 @@ def fromChunk(data):
byte string.
"""
prefix, rest = data.split(b"\r\n", 1)
- length = int(prefix, 16)
+ length = _hexint(prefix)
if length < 0:
raise ValueError("Chunk length must be >= 0, not %d" % (length,))
if rest[length : length + 2] != b"\r\n":
@@ -1883,8 +1909,9 @@ class _ChunkedTransferDecoder:
endOfLengthIndex = self._buffer.find(b";", 0, eolIndex)
if endOfLengthIndex == -1:
endOfLengthIndex = eolIndex
+ rawLength = self._buffer[0:endOfLengthIndex]
try:
- length = int(self._buffer[0:endOfLengthIndex], 16)
+ length = _hexint(rawLength)
except ValueError:
raise _MalformedChunkedDataError("Chunk-size must be an integer.")
diff --git a/src/twisted/web/test/test_http.py b/src/twisted/web/test/test_http.py
index e686aeb..201991f 100644
--- a/src/twisted/web/test/test_http.py
+++ b/src/twisted/web/test/test_http.py
@@ -4472,3 +4472,43 @@ class HTTPClientSanitizationTests(unittest.SynchronousTestCase):
transport.value().splitlines(),
[b": ".join([sanitizedBytes, sanitizedBytes])],
)
+
+
+class HexHelperTests(unittest.SynchronousTestCase):
+ """
+ Test the L{http._hexint} and L{http._ishexdigits} helper functions.
+ """
+
+ badStrings = (b"", b"0x1234", b"feds", b"-123" b"+123")
+
+ def test_isHex(self):
+ """
+ L{_ishexdigits()} returns L{True} for nonempy bytestrings containing
+ hexadecimal digits.
+ """
+ for s in (b"10", b"abcdef", b"AB1234", b"fed", b"123467890"):
+ self.assertIs(True, http._ishexdigits(s))
+
+ def test_decodes(self):
+ """
+ L{_hexint()} returns the integer equivalent of the input.
+ """
+ self.assertEqual(10, http._hexint(b"a"))
+ self.assertEqual(0x10, http._hexint(b"10"))
+ self.assertEqual(0xABCD123, http._hexint(b"abCD123"))
+
+ def test_isNotHex(self):
+ """
+ L{_ishexdigits()} returns L{False} for bytestrings that don't contain
+ hexadecimal digits, including the empty string.
+ """
+ for s in self.badStrings:
+ self.assertIs(False, http._ishexdigits(s))
+
+ def test_decodeNotHex(self):
+ """
+ L{_hexint()} raises L{ValueError} for bytestrings that can't
+ be decoded.
+ """
+ for s in self.badStrings:
+ self.assertRaises(ValueError, http._hexint, s)
|
