Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/streams/tls.py: 38%
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
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
99 @classmethod
100 async def wrap(
101 cls,
102 transport_stream: AnyByteStream,
103 *,
104 server_side: bool | None = None,
105 hostname: str | None = None,
106 ssl_context: ssl.SSLContext | None = None,
107 standard_compatible: bool = True,
108 ) -> TLSStream:
109 """
110 Wrap an existing stream with Transport Layer Security.
112 This performs a TLS handshake with the peer.
114 :param transport_stream: a bytes-transporting stream to wrap
115 :param server_side: ``True`` if this is the server side of the connection,
116 ``False`` if this is the client side (if omitted, will be set to ``False``
117 if ``hostname`` has been provided, ``False`` otherwise). Used only to create
118 a default context when an explicit context has not been provided.
119 :param hostname: host name of the peer (if host name checking is desired)
120 :param ssl_context: the SSLContext object to use (if not provided, a secure
121 default will be created)
122 :param standard_compatible: if ``False``, skip the closing handshake when
123 closing the connection, and don't raise an exception if the peer does the
124 same
125 :raises ~ssl.SSLError: if the TLS handshake fails
127 """
128 if server_side is None:
129 server_side = not hostname
131 if not ssl_context:
132 purpose = (
133 ssl.Purpose.CLIENT_AUTH if server_side else ssl.Purpose.SERVER_AUTH
134 )
135 ssl_context = ssl.create_default_context(purpose)
137 # Re-enable detection of unexpected EOFs if it was disabled by Python
138 if hasattr(ssl, "OP_IGNORE_UNEXPECTED_EOF"):
139 ssl_context.options &= ~ssl.OP_IGNORE_UNEXPECTED_EOF
141 bio_in = ssl.MemoryBIO()
142 bio_out = ssl.MemoryBIO()
144 # Resolve international host names using IDNA 2008.
145 # Otherwise wrap_bio() would resolve them with IDNA 2003.
146 if hostname is not None:
147 from .._core._sockets import idna2008_resolve
149 server_hostname: bytes | None = idna2008_resolve(hostname)
150 else:
151 server_hostname = None
153 # External SSLContext implementations may do blocking I/O in wrap_bio(),
154 # but the standard library implementation won't
155 if type(ssl_context) is ssl.SSLContext:
156 ssl_object = ssl_context.wrap_bio(
157 bio_in,
158 bio_out,
159 server_side=server_side,
160 server_hostname=server_hostname,
161 )
162 else:
163 ssl_object = await to_thread.run_sync(
164 ssl_context.wrap_bio,
165 bio_in,
166 bio_out,
167 server_side,
168 server_hostname,
169 None,
170 )
172 wrapper = cls(
173 transport_stream=transport_stream,
174 standard_compatible=standard_compatible,
175 _ssl_object=ssl_object,
176 _read_bio=bio_in,
177 _write_bio=bio_out,
178 )
179 await wrapper._call_sslobject_method(ssl_object.do_handshake)
180 return wrapper
182 async def _call_sslobject_method(
183 self, func: Callable[[Unpack[PosArgsT]], T_Retval], *args: Unpack[PosArgsT]
184 ) -> T_Retval:
185 while True:
186 try:
187 result = func(*args)
188 except ssl.SSLWantReadError:
189 try:
190 # Flush any pending writes first
191 if self._write_bio.pending:
192 await self.transport_stream.send(self._write_bio.read())
194 data = await self.transport_stream.receive()
195 except EndOfStream:
196 self._read_bio.write_eof()
197 except OSError as exc:
198 self._read_bio.write_eof()
199 self._write_bio.write_eof()
200 raise BrokenResourceError from exc
201 else:
202 self._read_bio.write(data)
203 except ssl.SSLWantWriteError:
204 await self.transport_stream.send(self._write_bio.read())
205 except ssl.SSLSyscallError as exc:
206 self._read_bio.write_eof()
207 self._write_bio.write_eof()
208 raise BrokenResourceError from exc
209 except ssl.SSLError as exc:
210 self._read_bio.write_eof()
211 self._write_bio.write_eof()
212 if isinstance(exc, ssl.SSLEOFError) or (
213 exc.strerror and "UNEXPECTED_EOF_WHILE_READING" in exc.strerror
214 ):
215 if self.standard_compatible:
216 raise BrokenResourceError from exc
217 else:
218 raise EndOfStream from None
220 raise
221 else:
222 # Flush any pending writes first
223 if self._write_bio.pending:
224 await self.transport_stream.send(self._write_bio.read())
226 return result
228 async def unwrap(self) -> tuple[AnyByteStream, bytes]:
229 """
230 Does the TLS closing handshake.
232 :return: a tuple of (wrapped byte stream, bytes left in the read buffer)
234 """
235 await self._call_sslobject_method(self._ssl_object.unwrap)
236 self._read_bio.write_eof()
237 self._write_bio.write_eof()
238 return self.transport_stream, self._read_bio.read()
240 async def aclose(self) -> None:
241 if self.standard_compatible:
242 try:
243 await self.unwrap()
244 except BaseException:
245 await aclose_forcefully(self.transport_stream)
246 raise
248 await self.transport_stream.aclose()
250 async def receive(self, max_bytes: int = 65536) -> bytes:
251 if max_bytes < 1:
252 raise ValueError("max_bytes must be a positive integer")
254 data = await self._call_sslobject_method(self._ssl_object.read, max_bytes)
255 if not data:
256 raise EndOfStream
258 return data
260 async def send(self, item: bytes) -> None:
261 await self._call_sslobject_method(self._ssl_object.write, item)
263 async def send_eof(self) -> None:
264 tls_version = self.extra(TLSAttribute.tls_version)
265 match = re.match(r"TLSv(\d+)(?:\.(\d+))?", tls_version)
266 if match:
267 major, minor = int(match.group(1)), int(match.group(2) or 0)
268 if (major, minor) < (1, 3):
269 raise NotImplementedError(
270 f"send_eof() requires at least TLSv1.3; current "
271 f"session uses {tls_version}"
272 )
274 raise NotImplementedError(
275 "send_eof() has not yet been implemented for TLS streams"
276 )
278 @property
279 def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
280 return {
281 **self.transport_stream.extra_attributes,
282 TLSAttribute.alpn_protocol: self._ssl_object.selected_alpn_protocol,
283 TLSAttribute.channel_binding_tls_unique: (
284 self._ssl_object.get_channel_binding
285 ),
286 TLSAttribute.cipher: self._ssl_object.cipher,
287 TLSAttribute.peer_certificate: lambda: self._ssl_object.getpeercert(False),
288 TLSAttribute.peer_certificate_binary: lambda: self._ssl_object.getpeercert(
289 True
290 ),
291 TLSAttribute.server_side: lambda: self._ssl_object.server_side,
292 TLSAttribute.shared_ciphers: lambda: (
293 self._ssl_object.shared_ciphers()
294 if self._ssl_object.server_side
295 else None
296 ),
297 TLSAttribute.standard_compatible: lambda: self.standard_compatible,
298 TLSAttribute.ssl_object: lambda: self._ssl_object,
299 TLSAttribute.tls_version: self._ssl_object.version,
300 }
303@dataclass(eq=False)
304class TLSListener(Listener[TLSStream]):
305 """
306 A convenience listener that wraps another listener and auto-negotiates a TLS session
307 on every accepted connection.
309 If the TLS handshake times out or raises an exception,
310 :meth:`handle_handshake_error` is called to do whatever post-mortem processing is
311 deemed necessary.
313 Supports only the :attr:`~TLSAttribute.standard_compatible` extra attribute.
315 :param Listener listener: the listener to wrap
316 :param ssl_context: the SSL context object
317 :param standard_compatible: a flag passed through to :meth:`TLSStream.wrap`
318 :param handshake_timeout: time limit for the TLS handshake
319 (passed to :func:`~anyio.fail_after`)
320 """
322 listener: Listener[Any]
323 ssl_context: ssl.SSLContext
324 standard_compatible: bool = True
325 handshake_timeout: float = 30
327 @staticmethod
328 async def handle_handshake_error(exc: BaseException, stream: AnyByteStream) -> None:
329 """
330 Handle an exception raised during the TLS handshake.
332 This method does 3 things:
334 #. Forcefully closes the original stream
335 #. Logs the exception (unless it was a cancellation exception) using the
336 ``anyio.streams.tls`` logger
337 #. Reraises the exception if it was a base exception or a cancellation exception
339 :param exc: the exception
340 :param stream: the original stream
342 """
343 await aclose_forcefully(stream)
345 # Log all except cancellation exceptions
346 if not isinstance(exc, get_cancelled_exc_class()):
347 # CPython (as of 3.11.5) returns incorrect `sys.exc_info()` here when using
348 # any asyncio implementation, so we explicitly pass the exception to log
349 # (https://github.com/python/cpython/issues/108668). Trio does not have this
350 # issue because it works around the CPython bug.
351 logging.getLogger(__name__).exception(
352 "Error during TLS handshake", exc_info=exc
353 )
355 # Only reraise base exceptions and cancellation exceptions
356 if not isinstance(exc, Exception) or isinstance(exc, get_cancelled_exc_class()):
357 raise
359 async def serve(
360 self,
361 handler: Callable[[TLSStream], Any],
362 task_group: TaskGroup | None = None,
363 ) -> None:
364 @wraps(handler)
365 async def handler_wrapper(stream: AnyByteStream) -> None:
366 from .. import fail_after
368 try:
369 with fail_after(self.handshake_timeout):
370 wrapped_stream = await TLSStream.wrap(
371 stream,
372 ssl_context=self.ssl_context,
373 standard_compatible=self.standard_compatible,
374 )
375 except BaseException as exc:
376 await self.handle_handshake_error(exc, stream)
377 else:
378 await handler(wrapped_stream)
380 await self.listener.serve(handler_wrapper, task_group)
382 async def aclose(self) -> None:
383 await self.listener.aclose()
385 @property
386 def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
387 return {
388 TLSAttribute.standard_compatible: lambda: self.standard_compatible,
389 }
392class TLSConnectable(ByteStreamConnectable):
393 """
394 Wraps another connectable and does TLS negotiation after a successful connection.
396 :param connectable: the connectable to wrap
397 :param hostname: host name of the server (if host name checking is desired)
398 :param ssl_context: the SSLContext object to use (if not provided, a secure default
399 will be created)
400 :param standard_compatible: if ``False``, skip the closing handshake when closing
401 the connection, and don't raise an exception if the server does the same
402 """
404 def __init__(
405 self,
406 connectable: AnyByteStreamConnectable,
407 *,
408 hostname: str | None = None,
409 ssl_context: ssl.SSLContext | None = None,
410 standard_compatible: bool = True,
411 ) -> None:
412 self.connectable = connectable
413 self.ssl_context: SSLContext = ssl_context or ssl.create_default_context(
414 ssl.Purpose.SERVER_AUTH
415 )
416 if not isinstance(self.ssl_context, ssl.SSLContext):
417 raise TypeError(
418 "ssl_context must be an instance of ssl.SSLContext, not "
419 f"{type(self.ssl_context).__name__}"
420 )
421 self.hostname = hostname
422 self.standard_compatible = standard_compatible
424 @override
425 async def connect(self) -> TLSStream:
426 stream = await self.connectable.connect()
427 try:
428 return await TLSStream.wrap(
429 stream,
430 hostname=self.hostname,
431 ssl_context=self.ssl_context,
432 standard_compatible=self.standard_compatible,
433 )
434 except BaseException:
435 await aclose_forcefully(stream)
436 raise