Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/streams/tls.py: 37%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3__all__ = (
4 "TLSAttribute",
5 "TLSConnectable",
6 "TLSListener",
7 "TLSStream",
8)
10import logging
11import re
12import ssl
13import sys
14from collections.abc import Callable, Mapping
15from dataclasses import dataclass, field
16from functools import wraps
17from ssl import SSLContext
18from typing import Any, TypeAlias, TypeVar
20from .. import (
21 BrokenResourceError,
22 EndOfStream,
23 aclose_forcefully,
24 get_cancelled_exc_class,
25 to_thread,
26)
27from .._core._typedattr import TypedAttributeSet, typed_attribute
28from ..abc import (
29 AnyByteStream,
30 AnyByteStreamConnectable,
31 ByteStream,
32 ByteStreamConnectable,
33 Listener,
34 TaskGroup,
35)
37if sys.version_info >= (3, 11):
38 from typing import TypeVarTuple, Unpack
39else:
40 from typing_extensions import TypeVarTuple, Unpack
42if sys.version_info >= (3, 12):
43 from typing import override
44else:
45 from typing_extensions import override
47T_Retval = TypeVar("T_Retval")
48PosArgsT = TypeVarTuple("PosArgsT")
49_PCTRTT: TypeAlias = tuple[tuple[str, str], ...]
50_PCTRTTT: TypeAlias = tuple[_PCTRTT, ...]
53class TLSAttribute(TypedAttributeSet):
54 """Contains Transport Layer Security related attributes."""
56 #: the selected ALPN protocol
57 alpn_protocol: str | None = typed_attribute()
58 #: the channel binding for type ``tls-unique``
59 channel_binding_tls_unique: bytes = typed_attribute()
60 #: the selected cipher
61 cipher: tuple[str, str, int] = typed_attribute()
62 #: the peer certificate in dictionary form (see :meth:`ssl.SSLSocket.getpeercert`
63 # for more information)
64 peer_certificate: None | (dict[str, str | _PCTRTTT | _PCTRTT]) = typed_attribute()
65 #: the peer certificate in binary form
66 peer_certificate_binary: bytes | None = typed_attribute()
67 #: ``True`` if this is the server side of the connection
68 server_side: bool = typed_attribute()
69 #: ciphers shared by the client during the TLS handshake (``None`` if this is the
70 #: client side)
71 shared_ciphers: list[tuple[str, str, int]] | None = typed_attribute()
72 #: the :class:`~ssl.SSLObject` used for encryption
73 ssl_object: ssl.SSLObject = typed_attribute()
74 #: ``True`` if this stream does (and expects) a closing TLS handshake when the
75 #: stream is being closed
76 standard_compatible: bool = typed_attribute()
77 #: the TLS protocol version (e.g. ``TLSv1.2``)
78 tls_version: str = typed_attribute()
81@dataclass(eq=False)
82class TLSStream(ByteStream):
83 """
84 A stream wrapper that encrypts all sent data and decrypts received data.
86 This class has no public initializer; use :meth:`wrap` instead.
87 All extra attributes from :class:`~TLSAttribute` are supported.
89 :var AnyByteStream transport_stream: the wrapped stream
91 """
93 transport_stream: AnyByteStream
94 standard_compatible: bool
95 _ssl_object: ssl.SSLObject
96 _read_bio: ssl.MemoryBIO
97 _write_bio: ssl.MemoryBIO
98 _broken: bool = field(default=False, init=False)
100 @classmethod
101 async def wrap(
102 cls,
103 transport_stream: AnyByteStream,
104 *,
105 server_side: bool | None = None,
106 hostname: str | None = None,
107 ssl_context: ssl.SSLContext | None = None,
108 standard_compatible: bool = True,
109 ) -> TLSStream:
110 """
111 Wrap an existing stream with Transport Layer Security.
113 This performs a TLS handshake with the peer.
115 :param transport_stream: a bytes-transporting stream to wrap
116 :param server_side: ``True`` if this is the server side of the connection,
117 ``False`` if this is the client side (if omitted, will be set to ``False``
118 if ``hostname`` has been provided, ``False`` otherwise). Used only to create
119 a default context when an explicit context has not been provided.
120 :param hostname: host name of the peer (if host name checking is desired)
121 :param ssl_context: the SSLContext object to use (if not provided, a secure
122 default will be created)
123 :param standard_compatible: if ``False``, skip the closing handshake when
124 closing the connection, and don't raise an exception if the peer does the
125 same
126 :raises ~ssl.SSLError: if the TLS handshake fails
128 """
129 if server_side is None:
130 server_side = not hostname
132 if not ssl_context:
133 purpose = (
134 ssl.Purpose.CLIENT_AUTH if server_side else ssl.Purpose.SERVER_AUTH
135 )
136 ssl_context = ssl.create_default_context(purpose)
138 # Re-enable detection of unexpected EOFs if it was disabled by Python
139 if hasattr(ssl, "OP_IGNORE_UNEXPECTED_EOF"):
140 ssl_context.options &= ~ssl.OP_IGNORE_UNEXPECTED_EOF
142 bio_in = ssl.MemoryBIO()
143 bio_out = ssl.MemoryBIO()
145 # Resolve international host names using IDNA 2008.
146 # Otherwise wrap_bio() would resolve them with IDNA 2003.
147 if hostname is not None:
148 from .._core._sockets import idna2008_resolve
150 server_hostname: bytes | None = idna2008_resolve(hostname)
151 else:
152 server_hostname = None
154 # External SSLContext implementations may do blocking I/O in wrap_bio(),
155 # but the standard library implementation won't
156 if type(ssl_context) is ssl.SSLContext:
157 ssl_object = ssl_context.wrap_bio(
158 bio_in,
159 bio_out,
160 server_side=server_side,
161 server_hostname=server_hostname,
162 )
163 else:
164 ssl_object = await to_thread.run_sync(
165 ssl_context.wrap_bio,
166 bio_in,
167 bio_out,
168 server_side,
169 server_hostname,
170 None,
171 )
173 wrapper = cls(
174 transport_stream=transport_stream,
175 standard_compatible=standard_compatible,
176 _ssl_object=ssl_object,
177 _read_bio=bio_in,
178 _write_bio=bio_out,
179 )
180 await wrapper._call_sslobject_method(ssl_object.do_handshake)
181 return wrapper
183 async def _flush_write_bio(self) -> None:
184 if self._write_bio.pending:
185 data = self._write_bio.read()
186 try:
187 await self.transport_stream.send(data)
188 except BaseException:
189 # SSLObject.write() already consumed this record's sequence
190 # number, so once it is lost the peer can't decrypt anything
191 # sent after it. Like trio, treat the stream as broken.
192 self._broken = True
193 raise
195 async def _call_sslobject_method(
196 self, func: Callable[[Unpack[PosArgsT]], T_Retval], *args: Unpack[PosArgsT]
197 ) -> T_Retval:
198 if self._broken:
199 raise BrokenResourceError
201 while True:
202 try:
203 result = func(*args)
204 except ssl.SSLWantReadError:
205 try:
206 # Flush any pending writes first
207 await self._flush_write_bio()
209 data = await self.transport_stream.receive()
210 except EndOfStream:
211 self._read_bio.write_eof()
212 except OSError as exc:
213 self._read_bio.write_eof()
214 self._write_bio.write_eof()
215 raise BrokenResourceError from exc
216 else:
217 self._read_bio.write(data)
218 except ssl.SSLWantWriteError:
219 await self._flush_write_bio()
220 except ssl.SSLSyscallError as exc:
221 self._read_bio.write_eof()
222 self._write_bio.write_eof()
223 raise BrokenResourceError from exc
224 except ssl.SSLError as exc:
225 self._read_bio.write_eof()
226 self._write_bio.write_eof()
227 if isinstance(exc, ssl.SSLEOFError) or (
228 exc.strerror and "UNEXPECTED_EOF_WHILE_READING" in exc.strerror
229 ):
230 if self.standard_compatible:
231 raise BrokenResourceError from exc
232 else:
233 raise EndOfStream from None
235 raise
236 else:
237 # Flush any pending writes first
238 await self._flush_write_bio()
240 return result
242 async def unwrap(self) -> tuple[AnyByteStream, bytes]:
243 """
244 Does the TLS closing handshake.
246 :return: a tuple of (wrapped byte stream, bytes left in the read buffer)
248 """
249 await self._call_sslobject_method(self._ssl_object.unwrap)
250 self._read_bio.write_eof()
251 self._write_bio.write_eof()
252 return self.transport_stream, self._read_bio.read()
254 async def aclose(self) -> None:
255 if self.standard_compatible and not self._broken:
256 try:
257 await self.unwrap()
258 except BaseException:
259 await aclose_forcefully(self.transport_stream)
260 raise
262 await self.transport_stream.aclose()
264 async def receive(self, max_bytes: int = 65536) -> bytes:
265 if max_bytes < 1:
266 raise ValueError("max_bytes must be a positive integer")
268 data = await self._call_sslobject_method(self._ssl_object.read, max_bytes)
269 if not data:
270 raise EndOfStream
272 return data
274 async def send(self, item: bytes) -> None:
275 await self._call_sslobject_method(self._ssl_object.write, item)
277 async def send_eof(self) -> None:
278 tls_version = self.extra(TLSAttribute.tls_version)
279 match = re.match(r"TLSv(\d+)(?:\.(\d+))?", tls_version)
280 if match:
281 major, minor = int(match.group(1)), int(match.group(2) or 0)
282 if (major, minor) < (1, 3):
283 raise NotImplementedError(
284 f"send_eof() requires at least TLSv1.3; current "
285 f"session uses {tls_version}"
286 )
288 raise NotImplementedError(
289 "send_eof() has not yet been implemented for TLS streams"
290 )
292 @property
293 def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
294 return {
295 **self.transport_stream.extra_attributes,
296 TLSAttribute.alpn_protocol: self._ssl_object.selected_alpn_protocol,
297 TLSAttribute.channel_binding_tls_unique: (
298 self._ssl_object.get_channel_binding
299 ),
300 TLSAttribute.cipher: self._ssl_object.cipher,
301 TLSAttribute.peer_certificate: lambda: self._ssl_object.getpeercert(False),
302 TLSAttribute.peer_certificate_binary: lambda: self._ssl_object.getpeercert(
303 True
304 ),
305 TLSAttribute.server_side: lambda: self._ssl_object.server_side,
306 TLSAttribute.shared_ciphers: lambda: (
307 self._ssl_object.shared_ciphers()
308 if self._ssl_object.server_side
309 else None
310 ),
311 TLSAttribute.standard_compatible: lambda: self.standard_compatible,
312 TLSAttribute.ssl_object: lambda: self._ssl_object,
313 TLSAttribute.tls_version: self._ssl_object.version,
314 }
317@dataclass(eq=False)
318class TLSListener(Listener[TLSStream]):
319 """
320 A convenience listener that wraps another listener and auto-negotiates a TLS session
321 on every accepted connection.
323 If the TLS handshake times out or raises an exception,
324 :meth:`handle_handshake_error` is called to do whatever post-mortem processing is
325 deemed necessary.
327 Supports only the :attr:`~TLSAttribute.standard_compatible` extra attribute.
329 :param Listener listener: the listener to wrap
330 :param ssl_context: the SSL context object
331 :param standard_compatible: a flag passed through to :meth:`TLSStream.wrap`
332 :param handshake_timeout: time limit for the TLS handshake
333 (passed to :func:`~anyio.fail_after`)
334 """
336 listener: Listener[Any]
337 ssl_context: ssl.SSLContext
338 standard_compatible: bool = True
339 handshake_timeout: float = 30
341 @staticmethod
342 async def handle_handshake_error(exc: BaseException, stream: AnyByteStream) -> None:
343 """
344 Handle an exception raised during the TLS handshake.
346 This method does 3 things:
348 #. Forcefully closes the original stream
349 #. Logs the exception (unless it was a cancellation exception) using the
350 ``anyio.streams.tls`` logger
351 #. Reraises the exception if it was a base exception or a cancellation exception
353 :param exc: the exception
354 :param stream: the original stream
356 """
357 await aclose_forcefully(stream)
359 # Log all except cancellation exceptions
360 if not isinstance(exc, get_cancelled_exc_class()):
361 # CPython (as of 3.11.5) returns incorrect `sys.exc_info()` here when using
362 # any asyncio implementation, so we explicitly pass the exception to log
363 # (https://github.com/python/cpython/issues/108668). Trio does not have this
364 # issue because it works around the CPython bug.
365 logging.getLogger(__name__).exception(
366 "Error during TLS handshake", exc_info=exc
367 )
369 # Only reraise base exceptions and cancellation exceptions
370 if not isinstance(exc, Exception) or isinstance(exc, get_cancelled_exc_class()):
371 raise # noqa: PLE0704
373 async def serve(
374 self,
375 handler: Callable[[TLSStream], Any],
376 task_group: TaskGroup | None = None,
377 ) -> None:
378 @wraps(handler)
379 async def handler_wrapper(stream: AnyByteStream) -> None:
380 from .. import fail_after
382 try:
383 with fail_after(self.handshake_timeout):
384 wrapped_stream = await TLSStream.wrap(
385 stream,
386 ssl_context=self.ssl_context,
387 standard_compatible=self.standard_compatible,
388 )
389 except BaseException as exc:
390 await self.handle_handshake_error(exc, stream)
391 else:
392 await handler(wrapped_stream)
394 await self.listener.serve(handler_wrapper, task_group)
396 async def aclose(self) -> None:
397 await self.listener.aclose()
399 @property
400 def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
401 return {
402 TLSAttribute.standard_compatible: lambda: self.standard_compatible,
403 }
406class TLSConnectable(ByteStreamConnectable):
407 """
408 Wraps another connectable and does TLS negotiation after a successful connection.
410 :param connectable: the connectable to wrap
411 :param hostname: host name of the server (if host name checking is desired)
412 :param ssl_context: the SSLContext object to use (if not provided, a secure default
413 will be created)
414 :param standard_compatible: if ``False``, skip the closing handshake when closing
415 the connection, and don't raise an exception if the server does the same
416 """
418 def __init__(
419 self,
420 connectable: AnyByteStreamConnectable,
421 *,
422 hostname: str | None = None,
423 ssl_context: ssl.SSLContext | None = None,
424 standard_compatible: bool = True,
425 ) -> None:
426 self.connectable = connectable
427 self.ssl_context: SSLContext = ssl_context or ssl.create_default_context(
428 ssl.Purpose.SERVER_AUTH
429 )
430 if not isinstance(self.ssl_context, ssl.SSLContext):
431 raise TypeError(
432 "ssl_context must be an instance of ssl.SSLContext, not "
433 f"{type(self.ssl_context).__name__}"
434 )
435 self.hostname = hostname
436 self.standard_compatible = standard_compatible
438 @override
439 async def connect(self) -> TLSStream:
440 stream = await self.connectable.connect()
441 try:
442 return await TLSStream.wrap(
443 stream,
444 hostname=self.hostname,
445 ssl_context=self.ssl_context,
446 standard_compatible=self.standard_compatible,
447 )
448 except BaseException:
449 await aclose_forcefully(stream)
450 raise