1import errno
  2import logging
  3import re
  4import socket
  5import sys
  6from collections.abc import Iterator
  7from pdb import Pdb
  8from types import FrameType
  9from typing import Any
 10
 11log = logging.getLogger(__name__)
 12
 13
 14def cry(message: str, stderr: Any = sys.__stderr__) -> None:
 15    log.critical(message)
 16    print(message, file=stderr)
 17    stderr.flush()
 18
 19
 20class LF2CRLF_FileWrapper:
 21    _send: Any  # Can be lambda or socket.sendall
 22
 23    def __init__(self, connection: socket.socket) -> None:
 24        self.connection = connection
 25        self.stream = fh = connection.makefile("rw")
 26        self.read = fh.read
 27        self.readline = fh.readline
 28        self.readlines = fh.readlines
 29        self.close = fh.close
 30        self.flush = fh.flush
 31        self.fileno = fh.fileno
 32        if hasattr(fh, "encoding"):
 33            self._send = lambda data: connection.sendall(data.encode(fh.encoding))
 34        else:
 35            self._send = connection.sendall
 36
 37    @property
 38    def encoding(self) -> str | None:
 39        return self.stream.encoding
 40
 41    def __iter__(self) -> Iterator[str]:
 42        return self.stream.__iter__()
 43
 44    def write(self, data: str, nl_rex: re.Pattern[str] = re.compile("\r?\n")) -> None:
 45        data = nl_rex.sub("\r\n", data)
 46        self._send(data)
 47
 48    def writelines(
 49        self, lines: list[str], nl_rex: re.Pattern[str] = re.compile("\r?\n")
 50    ) -> None:
 51        for line in lines:
 52            self.write(line, nl_rex)
 53
 54
 55class DevPdb(Pdb):
 56    """
 57    This will run pdb as a ephemeral telnet service. Once you connect no one
 58    else can connect. On construction this object will block execution till a
 59    client has connected.
 60
 61    Based on https://github.com/tamentis/rpdb I think ...
 62
 63    To use this::
 64
 65        DevPdb(host='0.0.0.0', port=4444).set_trace()
 66
 67    Then run: telnet 127.0.0.1 4444
 68    """
 69
 70    active_instance = None
 71
 72    def __init__(
 73        self, host: str, port: int, patch_stdstreams: bool = False, quiet: bool = False
 74    ) -> None:
 75        self._quiet = quiet
 76        listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 77        listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
 78        listen_socket.bind((host, port))
 79        if not self._quiet:
 80            cry(
 81                "DevPdb session open at {}:{}, waiting for connection ...".format(
 82                    *listen_socket.getsockname()
 83                )
 84            )
 85        listen_socket.listen(1)
 86        connection, address = listen_socket.accept()
 87        if not self._quiet:
 88            cry(f"DevPdb accepted connection from {repr(address)}.")
 89        self.handle = LF2CRLF_FileWrapper(connection)
 90        Pdb.__init__(self, completekey="tab", stdin=self.handle, stdout=self.handle)  # type: ignore[arg-type]
 91        self.backup = []
 92        if patch_stdstreams:
 93            for name in (
 94                "stderr",
 95                "stdout",
 96                "__stderr__",
 97                "__stdout__",
 98                "stdin",
 99                "__stdin__",
100            ):
101                self.backup.append((name, getattr(sys, name)))
102                setattr(sys, name, self.handle)
103        DevPdb.active_instance = self
104
105    def __restore(self) -> None:
106        if self.backup and not self._quiet:
107            cry(f"Restoring streams: {self.backup} ...")
108        for name, fh in self.backup:
109            setattr(sys, name, fh)
110        self.handle.close()
111        DevPdb.active_instance = None
112
113    def do_quit(self, arg: str) -> Any:
114        self.__restore()
115        return Pdb.do_quit(self, arg)
116
117    do_q = do_exit = do_quit
118
119    def set_trace(self, frame: FrameType | None = None) -> None:
120        if frame is None:
121            frame = sys._getframe().f_back
122        try:
123            Pdb.set_trace(self, frame)
124        except OSError as exc:
125            if exc.errno != errno.ECONNRESET:
126                raise
127
128
129def set_trace(
130    frame: FrameType | None = None,
131    host: str = "127.0.0.1",
132    port: int = 4444,
133    patch_stdstreams: bool = False,
134    quiet: bool = False,
135) -> None:
136    """
137    Opens a remote PDB over a host:port.
138    """
139    devpdb = DevPdb(
140        host=host, port=port, patch_stdstreams=patch_stdstreams, quiet=quiet
141    )
142    devpdb.set_trace(frame=frame)