Switching from clear text to encrypted and back again.
(self)
| 3725 | try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_2, False) |
| 3726 | |
| 3727 | def test_starttls(self): |
| 3728 | """Switching from clear text to encrypted and back again.""" |
| 3729 | msgs = (b"msg 1", b"MSG 2", b"STARTTLS", b"MSG 3", b"msg 4", b"ENDTLS", b"msg 5", b"msg 6") |
| 3730 | |
| 3731 | server = ThreadedEchoServer(CERTFILE, |
| 3732 | starttls_server=True, |
| 3733 | chatty=True, |
| 3734 | connectionchatty=True) |
| 3735 | wrapped = False |
| 3736 | with server: |
| 3737 | s = socket.socket() |
| 3738 | s.setblocking(True) |
| 3739 | s.connect((HOST, server.port)) |
| 3740 | if support.verbose: |
| 3741 | sys.stdout.write("\n") |
| 3742 | for indata in msgs: |
| 3743 | if support.verbose: |
| 3744 | sys.stdout.write( |
| 3745 | " client: sending %r...\n" % indata) |
| 3746 | if wrapped: |
| 3747 | conn.write(indata) |
| 3748 | outdata = conn.read() |
| 3749 | else: |
| 3750 | s.send(indata) |
| 3751 | outdata = s.recv(1024) |
| 3752 | msg = outdata.strip().lower() |
| 3753 | if indata == b"STARTTLS" and msg.startswith(b"ok"): |
| 3754 | # STARTTLS ok, switch to secure mode |
| 3755 | if support.verbose: |
| 3756 | sys.stdout.write( |
| 3757 | " client: read %r from server, starting TLS...\n" |
| 3758 | % msg) |
| 3759 | conn = test_wrap_socket(s) |
| 3760 | wrapped = True |
| 3761 | elif indata == b"ENDTLS" and msg.startswith(b"ok"): |
| 3762 | # ENDTLS ok, switch back to clear text |
| 3763 | if support.verbose: |
| 3764 | sys.stdout.write( |
| 3765 | " client: read %r from server, ending TLS...\n" |
| 3766 | % msg) |
| 3767 | s = conn.unwrap() |
| 3768 | wrapped = False |
| 3769 | else: |
| 3770 | if support.verbose: |
| 3771 | sys.stdout.write( |
| 3772 | " client: read %r from server\n" % msg) |
| 3773 | if support.verbose: |
| 3774 | sys.stdout.write(" client: closing connection.\n") |
| 3775 | if wrapped: |
| 3776 | conn.write(b"over\n") |
| 3777 | else: |
| 3778 | s.send(b"over\n") |
| 3779 | if wrapped: |
| 3780 | conn.close() |
| 3781 | else: |
| 3782 | s.close() |
| 3783 | |
| 3784 | def test_socketserver(self): |
nothing calls this directly
no test coverage detected