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