Coverage for /pythoncovmergedfiles/medio/medio/src/idna/idna/codec.py: 84%

1from __future__ import annotations 

2 

3import codecs 

4from typing import Any 

5 

6from .core import IDNAError, _max_domain_length, _unicode_dots_re, alabel, decode, encode, ulabel 

7 

8 

9class Codec(codecs.Codec): 

10 """Stateless IDNA 2008 codec. 

11 

12 Implements the :class:`codecs.Codec` protocol so that the whole-domain 

13 encoder (:func:`idna.encode`) and decoder (:func:`idna.decode`) are 

14 accessible through the standard codec machinery as ``"idna2008"``. 

15 

16 Only the ``"strict"`` error handler is supported; any other handler 

17 raises :exc:`~idna.IDNAError`. 

18 """ 

19 

20 def encode(self, data: str, errors: str = "strict") -> tuple[bytes, int]: # ty: ignore[invalid-method-override] 

21 if errors != "strict": 

22 raise IDNAError(f'Unsupported error handling "{errors}"', code="unsupported_errors") 

23 

24 if not data: 

25 return b"", 0 

26 

27 return encode(data), len(data) 

28 

29 def decode(self, data: bytes, errors: str = "strict") -> tuple[str, int]: # ty: ignore[invalid-method-override] 

30 if errors != "strict": 

31 raise IDNAError(f'Unsupported error handling "{errors}"', code="unsupported_errors") 

32 

33 if not data: 

34 return "", 0 

35 

36 return decode(data), len(data) 

37 

38 

39class IncrementalEncoder(codecs.BufferedIncrementalEncoder): 

40 """Incremental IDNA 2008 encoder. 

41 

42 Buffers a partial trailing label across calls until either the next 

43 label separator is seen or ``final=True``, so that streamed input is 

44 encoded one whole label at a time. Any of the four Unicode label 

45 separators (``U+002E``, ``U+3002``, ``U+FF0E``, ``U+FF61``) ends a 

46 label; the result always uses ``U+002E`` as the separator. 

47 

48 The 253-octet domain length limit (254 with a trailing dot) that 

49 :func:`idna.encode` enforces is applied to the accumulated output, so 

50 that streaming a name and encoding it in one shot either both succeed 

51 with the same result or both raise :exc:`~idna.IDNAError`. 

52 

53 Only the ``"strict"`` error handler is supported. 

54 """ 

55 

56 def __init__(self, errors: str = "strict") -> None: 

57 super().__init__(errors) 

58 self._emitted = 0 # octets returned so far 

59 self._trailing_dot = False # whether the output so far ends with "." 

60 

61 def reset(self) -> None: 

62 super().reset() 

63 self._emitted = 0 

64 self._trailing_dot = False 

65 

66 def getstate(self) -> Any: 

67 if not self.buffer and not self._emitted: 

68 return 0 

69 return (self.buffer, self._emitted, self._trailing_dot) 

70 

71 def setstate(self, state: Any) -> None: 

72 if state: 

73 self.buffer, self._emitted, self._trailing_dot = state 

74 else: 

75 self.reset() 

76 

77 def _buffer_encode(self, data: str, errors: str, final: bool) -> tuple[bytes, int]: # ty: ignore[invalid-method-override] 

78 if errors != "strict": 

79 raise IDNAError(f'Unsupported error handling "{errors}"', code="unsupported_errors") 

80 

81 result_bytes = b"" 

82 size = 0 

83 if data: 

84 labels = _unicode_dots_re.split(data) 

85 trailing_dot = b"" 

86 if labels: 

87 if not labels[-1]: 

88 trailing_dot = b"." 

89 del labels[-1] 

90 elif not final: 

91 # Keep potentially unfinished label until the next call 

92 del labels[-1] 

93 if labels: 

94 trailing_dot = b"." 

95 

96 result = [] 

97 for label in labels: 

98 result.append(alabel(label)) 

99 if size: 

100 size += 1 

101 size += len(label) 

102 

103 result_bytes = b".".join(result) + trailing_dot 

104 size += len(trailing_dot) 

105 

106 # Mirror encode(): the whole name may not exceed 253 octets, or 254 

107 # when it ends with a dot. Until the input is final a trailing dot 

108 # may still arrive, so only the 254-octet ceiling applies before then. 

109 self._emitted += len(result_bytes) 

110 if result_bytes: 

111 self._trailing_dot = result_bytes.endswith(b".") 

112 may_end_with_dot = self._trailing_dot or not final 

113 if self._emitted > _max_domain_length + may_end_with_dot: 

114 raise IDNAError("Domain too long", code="domain_too_long") 

115 return result_bytes, size 

116 

117 

118class IncrementalDecoder(codecs.BufferedIncrementalDecoder): 

119 """Incremental IDNA 2008 decoder. 

120 

121 Buffers a partial trailing label across calls until either the next 

122 label separator is seen or ``final=True``, so that streamed input is 

123 decoded one whole label at a time. 

124 

125 The 254-octet input length limit that :func:`idna.decode` enforces is 

126 applied to the accumulated input, so that streaming a name and decoding 

127 it in one shot either both succeed with the same result or both raise 

128 :exc:`~idna.IDNAError`. 

129 

130 Only the ``"strict"`` error handler is supported. 

131 """ 

132 

133 def __init__(self, errors: str = "strict") -> None: 

134 super().__init__(errors) 

135 self._consumed = 0 # input octets consumed so far 

136 

137 def reset(self) -> None: 

138 super().reset() 

139 self._consumed = 0 

140 

141 def getstate(self) -> tuple[bytes, int]: 

142 return (self.buffer, self._consumed) 

143 

144 def setstate(self, state: tuple[bytes, int]) -> None: 

145 self.buffer, self._consumed = state 

146 

147 def _buffer_decode(self, data: Any, errors: str, final: bool) -> tuple[str, int]: # ty: ignore[invalid-method-override] 

148 if errors != "strict": 

149 raise IDNAError(f'Unsupported error handling "{errors}"', code="unsupported_errors") 

150 

151 if not data: 

152 return ("", 0) 

153 

154 if not isinstance(data, str): 

155 try: 

156 data = str(data, "ascii") 

157 except UnicodeDecodeError as err: 

158 raise IDNAError("Invalid ASCII in A-label", code="invalid_ascii") from err 

159 

160 # Mirror decode(), which rejects input longer than 254 characters 

161 # (253 plus a possible trailing dot) before looking at any label. 

162 # ``data`` is the unconsumed buffer plus the new input, so this is 

163 # the total seen so far. 

164 if self._consumed + len(data) > _max_domain_length + 1: 

165 raise IDNAError("Domain too long", code="domain_too_long") 

166 

167 labels = _unicode_dots_re.split(data) 

168 trailing_dot = "" 

169 if labels: 

170 if not labels[-1]: 

171 trailing_dot = "." 

172 del labels[-1] 

173 elif not final: 

174 # Keep potentially unfinished label until the next call 

175 del labels[-1] 

176 if labels: 

177 trailing_dot = "." 

178 

179 result = [] 

180 size = 0 

181 for label in labels: 

182 result.append(ulabel(label)) 

183 if size: 

184 size += 1 

185 size += len(label) 

186 

187 result_str = ".".join(result) + trailing_dot 

188 size += len(trailing_dot) 

189 self._consumed += size 

190 return (result_str, size) 

191 

192 

193class StreamWriter(Codec, codecs.StreamWriter): 

194 pass 

195 

196 

197class StreamReader(Codec, codecs.StreamReader): 

198 pass 

199 

200 

201def search_function(name: str) -> codecs.CodecInfo | None: 

202 """Codec search function registered with :mod:`codecs`. 

203 

204 Returns a :class:`codecs.CodecInfo` for the ``"idna2008"`` codec name 

205 so that ``str.encode("idna2008")`` and ``bytes.decode("idna2008")`` 

206 invoke the IDNA 2008 codec defined in this module. 

207 

208 :param name: The codec name being looked up. 

209 :returns: A :class:`codecs.CodecInfo` instance if ``name`` is 

210 ``"idna2008"``, otherwise ``None``. 

211 """ 

212 if name != "idna2008": 

213 return None 

214 return codecs.CodecInfo( 

215 name=name, 

216 encode=Codec().encode, 

217 decode=Codec().decode, # type: ignore 

218 incrementalencoder=IncrementalEncoder, 

219 incrementaldecoder=IncrementalDecoder, 

220 streamwriter=StreamWriter, 

221 streamreader=StreamReader, 

222 ) 

223 

224 

225codecs.register(search_function)