Stable SSH auth agent, that keeps the same location for auth socket and can handle many connects/disconnects. Wor

stable_auth_agent.py

# Stable SSH auth agent, that keeps the same location for auth socket 
# and can handle many connects/disconnects. 
# Works until any SSH connection with agent works.
import socket
import threading
import os
import select
import signal
import random
from pathlib import Path
import sys


def bridge(src_conn: socket.socket, dst_path: str):
    """Bidirectionally connect src_conn and dst_path socket."""
    try:
        dst_conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        dst_conn.connect(dst_path)
    except Exception as e:
        print(f"Failed to connect to {dst_path}: {e}")
        src_conn.close()
        return

    sockets = [src_conn, dst_conn]

    while True:
        rlist, _, _ = select.select(sockets, [], [])
        for s in rlist:
            data = s.recv(4096)
            if not data:
                for c in sockets:
                    c.close()  # One side closed — clean up
                return
            if s is src_conn:  # Forward data to the other side
                dst_conn.sendall(data)
            else:
                src_conn.sendall(data)


def main(listen_socket: Path):
    listen_socket.parent.mkdir(exist_ok=True, parents=True)

    symlinks_folder = listen_socket.with_suffix(".links")
    symlinks_folder.mkdir(exist_ok=True, parents=True)
    pid_file = listen_socket.with_suffix(".pid")

    if pid_file.exists():
        try:
            os.kill(int(pid_file.read_text()), signal.SIGINT)
        except BaseException:
            pass
    pid_file.write_text(str(os.getpid()))

    try:
        listen_socket.unlink()  # Clean up old bridge socket if it exists
    except FileNotFoundError:
        pass

    sock_path = Path(os.environ["SSH_AUTH_SOCK"])
    if sock_path.exists():
        symlinks_folder.joinpath(str(random.randint(10**9, 10**10))).symlink_to(
            sock_path
        )

    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server.bind(listen_socket.as_posix())
    server.listen()
    print('started listening to new socket')

    try:
        while True:
            conn, _ = server.accept()
            alive_links = []
            for f in symlinks_folder.iterdir():
                if f.is_symlink():
                    if f.resolve().exists():
                        alive_links.append(f)
                    else:
                        f.unlink()

            print(f"{alive_links=}")
            used_link = alive_links[0].as_posix()
            threading.Thread(target=bridge, args=(conn, used_link), daemon=True).start()
    except KeyboardInterrupt:
        pass
    finally:
        server.close()
        print(f"removing {listen_socket=} before exiting")
        pid_file.unlink()
        listen_socket.unlink()


if __name__ == "__main__":
    _, new_agent_socket = sys.argv[:]
    main(Path(new_agent_socket))

shall by used in .profile as

nohup python <file.py> /tmp/my_stable_socket &
export SSH_AUTH_SOCK=/tmp/my_stable_socket