Socket-like object providing TLS-in-TLS. Only methods that are used by websockets are implemented.
| 554 | |
| 555 | |
| 556 | class SSLSSLSocket: |
| 557 | """ |
| 558 | Socket-like object providing TLS-in-TLS. |
| 559 | |
| 560 | Only methods that are used by websockets are implemented. |
| 561 | |
| 562 | """ |
| 563 | |
| 564 | recv_bufsize = 65536 |
| 565 | |
| 566 | def __init__( |
| 567 | self, |
| 568 | sock: socket.socket, |
| 569 | ssl_context: ssl_module.SSLContext, |
| 570 | server_hostname: str | None = None, |
| 571 | ) -> None: |
| 572 | self.incoming = ssl_module.MemoryBIO() |
| 573 | self.outgoing = ssl_module.MemoryBIO() |
| 574 | self.ssl_socket = sock |
| 575 | self.ssl_object = ssl_context.wrap_bio( |
| 576 | self.incoming, |
| 577 | self.outgoing, |
| 578 | server_hostname=server_hostname, |
| 579 | ) |
| 580 | self.run_io(self.ssl_object.do_handshake) |
| 581 | |
| 582 | def run_io(self, func: Callable[..., T], *args: Any) -> T: |
| 583 | while True: |
| 584 | want_read = False |
| 585 | want_write = False |
| 586 | try: |
| 587 | result = func(*args) |
| 588 | except ssl_module.SSLWantReadError: |
| 589 | want_read = True |
| 590 | except ssl_module.SSLWantWriteError: # pragma: no cover |
| 591 | want_write = True |
| 592 | |
| 593 | # Write outgoing data in all cases. |
| 594 | data = self.outgoing.read() |
| 595 | if data: |
| 596 | self.ssl_socket.sendall(data) |
| 597 | |
| 598 | # Read incoming data and retry on SSLWantReadError. |
| 599 | if want_read: |
| 600 | data = self.ssl_socket.recv(self.recv_bufsize) |
| 601 | if data: |
| 602 | self.incoming.write(data) |
| 603 | else: |
| 604 | self.incoming.write_eof() |
| 605 | continue |
| 606 | # Retry after writing outgoing data on SSLWantWriteError. |
| 607 | if want_write: # pragma: no cover |
| 608 | continue |
| 609 | # Return result if no error happened. |
| 610 | return result |
| 611 | |
| 612 | def recv(self, buflen: int) -> bytes: |
| 613 | try: |
no outgoing calls
no test coverage detected
searching dependent graphs…