/usr/lib/python3/dist-packages/pudb/remote.py is in python3-pudb 2015.4.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | # -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
# mostly stolen from celery.contrib.rdb
import errno
import os
import socket
import sys
from pudb.debugger import Debugger
__all__ = ['PUDB_RDB_HOST', 'PUDB_RDB_PORT', 'default_port',
'debugger', 'set_trace']
default_port = 6899
PUDB_RDB_HOST = os.environ.get('CELERY_RDB_HOST') or '127.0.0.1'
PUDB_RDB_PORT = int(os.environ.get('CELERY_RDB_PORT') or default_port)
#: Holds the currently active debugger.
_current = [None]
_frame = getattr(sys, '_getframe')
NO_AVAILABLE_PORT = """\
{self.ident}: Couldn't find an available port.
Please specify one using the CELERY_RDB_PORT environment variable.
"""
BANNER = """\
{self.ident}: Please telnet into {self.host} {self.port}.
{self.ident}: Waiting for client...
"""
SESSION_STARTED = '{self.ident}: Now in session with {self.remote_addr}.'
SESSION_ENDED = '{self.ident}: Session with {self.remote_addr} ended.'
class RemoteDebugger(Debugger):
me = 'pudb'
_prev_outs = None
_sock = None
def __init__(self, host=PUDB_RDB_HOST, port=PUDB_RDB_PORT,
port_search_limit=100, out=sys.stdout, term_size=None):
self.active = True
self.out = out
self._prev_handles = sys.stdin, sys.stdout
self._sock, this_port = self.get_avail_port(
host, port, port_search_limit)
self._sock.setblocking(1)
self._sock.listen(1)
self.ident = '{0}:{1}'.format(self.me, this_port)
self.host = host
self.port = this_port
self.say(BANNER.format(self=self))
self._client, address = self._sock.accept()
self._client.setblocking(1)
self.remote_addr = ':'.join(str(v) for v in address)
self.say(SESSION_STARTED.format(self=self))
self._handle = sys.stdin = sys.stdout = self._client.makefile('rwb', 0)
import telnetlib as tn
self._handle.write(tn.IAC + tn.WILL + tn.SGA)
resp = self._handle.read(3)
assert resp == tn.IAC + tn.DO + tn.SGA
self._handle.write(tn.IAC + tn.WILL + tn.ECHO)
resp = self._handle.read(3)
assert resp == tn.IAC + tn.DO + tn.ECHO
Debugger.__init__(self, stdin=self._handle, stdout=self._handle,
term_size=term_size)
def get_avail_port(self, host, port, search_limit=100, skew=+0):
this_port = None
for i in range(search_limit):
_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
this_port = port + i
try:
_sock.bind((host, this_port))
except socket.error as exc:
if exc.errno in [errno.EADDRINUSE, errno.EINVAL]:
continue
raise
else:
return _sock, this_port
else:
raise Exception(NO_AVAILABLE_PORT.format(self=self))
def say(self, m):
print(m, file=self.out)
def _close_session(self):
self.stdin, self.stdout = sys.stdin, sys.stdout = self._prev_handles
self._handle.close()
self._client.close()
self._sock.close()
self.active = False
self.say(SESSION_ENDED.format(self=self))
def do_continue(self, arg):
self._close_session()
self.set_continue()
return 1
do_c = do_cont = do_continue
def do_quit(self, arg):
self._close_session()
self.set_quit()
return 1
def set_quit(self):
# this raises a BdbQuit exception that we are unable to catch.
sys.settrace(None)
def debugger(term_size=None):
"""Return the current debugger instance (if any),
or creates a new one."""
rdb = _current[0]
if rdb is None or not rdb.active:
rdb = _current[0] = RemoteDebugger(term_size=term_size)
return rdb
def set_trace(frame=None, term_size=None):
"""Set breakpoint at current location, or a specified frame"""
if frame is None:
frame = _frame().f_back
return debugger(term_size=term_size).set_trace(frame)
|