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