Switching from clear text to encrypted and back again.
(self)
| 3442 | try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_2, False) |
| 3443 | |
| 3444 | def test_starttls(self): |
| 3445 | """Switching from clear text to encrypted and back again.""" |
| 3446 | msgs = (b"msg 1", b"MSG 2", b"STARTTLS", b"MSG 3", b"msg 4", b"ENDTLS", b"msg 5", b"msg 6") |
| 3447 | |
| 3448 | server = ThreadedEchoServer(CERTFILE, |
| 3449 | starttls_server=True, |
| 3450 | chatty=True, |
| 3451 | connectionchatty=True) |
| 3452 | wrapped = False |
| 3453 | with server: |
| 3454 | s = socket.socket() |
| 3455 | s.setblocking(True) |
| 3456 | s.connect((HOST, server.port)) |
| 3457 | if support.verbose: |
| 3458 | sys.stdout.write("\n") |
| 3459 | for indata in msgs: |
| 3460 | if support.verbose: |
| 3461 | sys.stdout.write( |
| 3462 | " client: sending %r...\n" % indata) |
| 3463 | if wrapped: |
| 3464 | conn.write(indata) |
| 3465 | outdata = conn.read() |
| 3466 | else: |
| 3467 | s.send(indata) |
| 3468 | outdata = s.recv(1024) |
| 3469 | msg = outdata.strip().lower() |
| 3470 | if indata == b"STARTTLS" and msg.startswith(b"ok"): |
| 3471 | # STARTTLS ok, switch to secure mode |
| 3472 | if support.verbose: |
| 3473 | sys.stdout.write( |
| 3474 | " client: read %r from server, starting TLS...\n" |
| 3475 | % msg) |
| 3476 | conn = test_wrap_socket(s) |
| 3477 | wrapped = True |
| 3478 | elif indata == b"ENDTLS" and msg.startswith(b"ok"): |
| 3479 | # ENDTLS ok, switch back to clear text |
| 3480 | if support.verbose: |
| 3481 | sys.stdout.write( |
| 3482 | " client: read %r from server, ending TLS...\n" |
| 3483 | % msg) |
| 3484 | s = conn.unwrap() |
| 3485 | wrapped = False |
| 3486 | else: |
| 3487 | if support.verbose: |
| 3488 | sys.stdout.write( |
| 3489 | " client: read %r from server\n" % msg) |
| 3490 | if support.verbose: |
| 3491 | sys.stdout.write(" client: closing connection.\n") |
| 3492 | if wrapped: |
| 3493 | conn.write(b"over\n") |
| 3494 | else: |
| 3495 | s.send(b"over\n") |
| 3496 | if wrapped: |
| 3497 | conn.close() |
| 3498 | else: |
| 3499 | s.close() |
| 3500 | |
| 3501 | def test_socketserver(self): |
nothing calls this directly
no test coverage detected