gh-148594: Clear stale OpenSSL errors before TLS. I/O. - #159641

"""Stale OpenSSL error queue in a server thread drops the next TLS connection.

A server accepts and wraps sockets in one thread and reads them in a single
worker thread (the usual accept-then-hand-off thread-pool layout).
One client completes the TLS handshake and closes without sending anything.
The worker's read of that socket fails, and leaves an error on the worker
thread's OpenSSL error queue. After that, every fresh connection whose first
read has to wait fails in the worker instead of waiting.

    python3 repro.py [--probe] [--clear] [--same-thread] [--no-tickets] [--tls12]
                     [-n N] [--delay S]

  --probe        first run one client that handshakes and closes, sending nothing
  --clear        call ERR_clear_error() (via ctypes) in the worker before each read
  --same-thread  accept, wrap and read in one thread (no hand-off)
  --no-tickets   server sends no TLS 1.3 session tickets (num_tickets = 0)
  --tls12        cap the server at TLS 1.2

The server prints, per outcome, the worker thread's error queue (first..last
entry, via ctypes) before and after the read; 0x0..0x0 is empty.

Standard library only; needs the `openssl` CLI to make a throwaway self-signed pair.
"""
import argparse
import collections
import ctypes
import os
import queue
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time

p = argparse.ArgumentParser()
p.add_argument("--probe", action="store_true")
p.add_argument("--clear", action="store_true")
p.add_argument("--same-thread", action="store_true")
p.add_argument("--no-tickets", action="store_true")
p.add_argument("--tls12", action="store_true")
p.add_argument("-n", type=int, default=20)
p.add_argument("--delay", type=float, default=0.2)
args = p.parse_args()

# ERR_* from the libcrypto the ssl module actually uses: looked up through the
# _ssl extension, whose dependencies dlsym() searches, or through the
# interpreter itself when _ssl is built in. None if not exported.
try:
    import _ssl
    crypto = ctypes.CDLL(getattr(_ssl, "__file__", None))
    crypto.ERR_peek_error.restype = ctypes.c_ulong
    crypto.ERR_peek_last_error.restype = ctypes.c_ulong
    crypto.ERR_clear_error
except (AttributeError, OSError):
    crypto = None
if args.clear and crypto is None:
    sys.exit("--clear: ERR_clear_error is not reachable through _ssl on this build")

tmp = tempfile.mkdtemp()
cert, key = os.path.join(tmp, "c.pem"), os.path.join(tmp, "k.pem")
subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
                "-keyout", key, "-out", cert, "-days", "1", "-subj", "/CN=localhost"],
               check=True, capture_output=True)

sctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
sctx.load_cert_chain(cert, key)
shutil.rmtree(tmp)  # loaded; the throwaway key is not needed on disk any more
if args.no_tickets:
    sctx.num_tickets = 0
if args.tls12:
    sctx.maximum_version = ssl.TLSVersion.TLSv1_2
cctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
cctx.check_hostname = False
cctx.verify_mode = ssl.CERT_NONE

lsock = socket.create_server(("127.0.0.1", 0))
port = lsock.getsockname()[1]
handoff = queue.Queue()
log = collections.Counter()


def accept_and_wrap():
    conn, _ = lsock.accept()
    conn.settimeout(5)  # a read that has to wait goes through SSL_get_error
    try:
        return sctx.wrap_socket(conn, server_side=True)  # handshake here
    except (OSError, ssl.SSLError) as e:
        log["handshake " + type(e).__name__] += 1
        conn.close()


def queue_state():
    if crypto is None:
        return "n/a"
    return "%#x..%#x" % (crypto.ERR_peek_error(), crypto.ERR_peek_last_error())


def serve_one(s):
    if args.clear:
        crypto.ERR_clear_error()
    before = queue_state()
    try:
        data = s.recv(1024)
        if data:
            s.sendall(b"HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\nok")
            log["read ok"] += 1
        else:
            log["read eof (ragged EOF suppressed); queue after: %s" % queue_state()] += 1
    except (OSError, ssl.SSLError) as e:
        log["read %s; queue before: %s, after: %s"
            % (type(e).__name__, before, queue_state())] += 1
    finally:
        s.close()


def acceptor():
    while True:
        s = accept_and_wrap()
        if s:
            handoff.put(s)


def worker():
    while True:
        serve_one(handoff.get())


def single_thread():
    while True:
        s = accept_and_wrap()
        if s:
            serve_one(s)


if args.same_thread:
    threading.Thread(target=single_thread, daemon=True).start()
else:
    threading.Thread(target=acceptor, daemon=True).start()
    threading.Thread(target=worker, daemon=True).start()


def connect():
    return cctx.wrap_socket(socket.create_connection(("127.0.0.1", port), timeout=5))


if args.probe:
    connect().close()  # handshake, then close without sending anything
    time.sleep(0.2)

answered = 0
for i in range(args.n):
    c = connect()
    time.sleep(args.delay)  # the request arrives after the server's first read
    try:
        c.sendall(b"GET / HTTP/1.0\r\n\r\n")
        answered += c.recv(100).startswith(b"HTTP/1.0 200")
    except OSError:
        pass
    c.close()

time.sleep(0.2)
print("%s  %s  %s  answered %d of %d"
      % (sys.version.split()[0], ssl.OPENSSL_VERSION, " ".join(sys.argv[1:]) or "(no flags)",
         answered, args.n))
for k, v in sorted(log.items()):
    print("  server: %-72s %d" % (k, v))