[BUG] - exec_command can deadlock transport when its ChannelStdinFile is garbage collected at the wrong time

import atexit
import gc
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import threading
import time

import paramiko
import traceback
from paramiko.channel import Channel

_orig_Channel__close_internal = Channel._close_internal


def Channel__close_internal(self):
    # Wait for stdin to go out of scope on the main thread.
    time.sleep(1.0)

    # Trigger garbage collection to run the `ChannelStdinFile` finalizer.
    gc.collect()

    return _orig_Channel__close_internal(self)


Channel._close_internal = Channel__close_internal


def start_sshd(workdir):
    """Starts an sshd instance to test against."""
    sshd_path = shutil.which("sshd") or "/usr/sbin/sshd"
    if not os.path.exists(sshd_path):
        sys.exit("no sshd binary found")

    # Pick a free port.
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()

    host_key = os.path.join(workdir, "host_key")
    client_key = os.path.join(workdir, "client_key")
    authorized_keys = os.path.join(workdir, "authorized_keys")

    subprocess.check_call(["ssh-keygen", "-q", "-t", "ed25519", "-f", host_key, "-N", ""])
    subprocess.check_call(["ssh-keygen", "-q", "-t", "ed25519", "-f", client_key, "-N", ""])
    shutil.copy(client_key + ".pub", authorized_keys)
    os.chmod(host_key, 0o600)
    os.chmod(client_key, 0o600)

    config_path = os.path.join(workdir, "sshd_config")
    with open(config_path, "w") as f:
        f.write(
            textwrap.dedent(
                f"""\
                Port {port}
                ListenAddress 127.0.0.1
                HostKey {host_key}
                AuthorizedKeysFile {authorized_keys}
                PubkeyAuthentication yes
                PasswordAuthentication no
                KbdInteractiveAuthentication no
                UsePAM no
                StrictModes no
                PidFile {workdir}/sshd.pid
                """
            )
        )

    log_path = os.path.join(workdir, "sshd.log")
    proc = subprocess.Popen([sshd_path, "-D", "-f", config_path, "-E", log_path])
    atexit.register(proc.terminate)

    for _ in range(100):
        if os.path.exists(log_path) and "Server listening" in open(log_path).read():
            break
        time.sleep(0.05)
    else:
        sys.exit("sshd never came up, check %s" % log_path)

    return port, client_key


def first_ssh_command(host, port, key_path, cmd):
    """Creates an SSHClient and runs a command."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    # `SSHClient._auth` creates a reference cycle:
    #     exception -> traceback -> _auth stack frame -> saved_exception local variable -> exception
    # The traceback also retains this stack frame and keeps stdin alive.
    client.connect(
        host,
        port=port,
        username=os.environ["USER"],
        key_filename=key_path,
        look_for_keys=False,
        allow_agent=False,
        timeout=10,
    )

    stdin, stdout, stderr = client.exec_command(cmd)
    try:
        stdout.read()
        stdout.channel.recv_exit_status()
    finally:
        stdin.close()
        stdout.close()
        stderr.close()

    return client


def second_ssh_command(client, cmd):
    """Runs a command on an existing SSHClient."""
    # Expected to hang here if the transport thread is deadlocked.
    stdin, stdout, stderr = client.exec_command(cmd)
    try:
        stdout.read()
        stdout.channel.recv_exit_status()
    finally:
        stdin.close()
        stdout.close()
        stderr.close()


def main():
    workdir = tempfile.mkdtemp(prefix="paramiko_deadlock_repro_")
    atexit.register(shutil.rmtree, workdir, ignore_errors=True)

    port, key_path = start_sshd(workdir)

    client = first_ssh_command("127.0.0.1", port, key_path, "true")
    print("first_ssh_command completed")
    # second_ssh_command will hang if the transport thread is deadlocked.
    second_ssh_command(client, "true")
    print("second_ssh_command completed")


def start_watchdog_thread():
    def watchdog():
        # Wait for the main thread to hang inside `second_ssh_command`.
        time.sleep(20)

        # If the main thread hasn't exited by now, we are in a deadlock.
        # Dump all threads.
        print("*** DEADLOCK ***")
        names = {t.ident: t.name for t in threading.enumerate()}
        for ident, frame in sys._current_frames().items():
            print("Thread %s:" % (names.get(ident, "?")))
            traceback.print_stack(frame)
            print()

        sys.exit(1)

    threading.Thread(target=watchdog, daemon=True).start()


if __name__ == "__main__":
    start_watchdog_thread()
    main()

Thread Thread-2:
  File "/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/threading.py", line 1014, in _bootstrap
    self._bootstrap_inner()
  File "/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/threading.py", line 1043, in _bootstrap_inner
    self.run()
  File ".../repro_venv/lib/python3.13/site-packages/paramiko/transport.py", line 2085, in run
    self._channel_handler_table[ptype](chan, m)
  File ".../repro_venv/lib/python3.13/site-packages/paramiko/channel.py", line 1179, in _handle_close
    msgs = self._close_internal()
  File ".../paramiko_deadlock_repro.py", line 25, in Channel__close_internal
    gc.collect()
  File ".../repro_venv/lib/python3.13/site-packages/paramiko/file.py", line 67, in __del__
    self.close()
  File ".../repro_venv/lib/python3.13/site-packages/paramiko/channel.py", line 1390, in close
    self.channel.shutdown_write()
  File ".../repro_venv/lib/python3.13/site-packages/paramiko/channel.py", line 989, in shutdown_write
    self.shutdown(1)
  File ".../repro_venv/lib/python3.13/site-packages/paramiko/channel.py", line 959, in shutdown
    self.lock.acquire()