8008596090

Committed 22 Feb 2024 05:30PM CUT coverage: 74.386% (+1.1%) from 73.253%

Build # 8008596090

Build Type

push

github

Committed by

web-flow

Commit Message

Fix gdb and other doctests (#2355)

* Fix gdb.debug doctests

* Accept bytes in ssh.which

But still return a `str` return for backwards-compatibility.

* CI: Make rpyc visible in gdb's python

* Fix not running all doctests

They appear to have to be a free standing block of `>>>` with an empty line above.

* Remove unnecessary text type check in ssh.which

Run Details

4639 of 7370 branches covered (62.94%)

12999 of 17475 relevant lines covered (74.39%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/pwnlib/gdb_api_bridge.py

"""GDB Python API bridge."""
import gdb

import socket
from threading import Condition
import time

from rpyc.core.protocol import Connection
from rpyc.core.service import Service
from rpyc.lib import spawn
from rpyc.lib.compat import select_error
from rpyc.utils.server import ThreadedServer


class ServeResult:
    """Result of serving requests on GDB thread."""
    def __init__(self):
        self.cv = Condition()
        self.done = False
        self.exc = None

    def set(self, exc):
        with self.cv:
            self.done = True
            self.exc = exc
            self.cv.notify()

    def wait(self):
        with self.cv:
            while not self.done:
                self.cv.wait()
            if self.exc is not None:
                raise self.exc


class GdbConnection(Connection):
    """A Connection implementation that serves requests on GDB thread.

    Serving on GDB thread might not be ideal from the responsiveness
    perspective, however, it is simple and reliable.
    """
    SERVE_TIME = 0.1  # Number of seconds to serve.
    IDLE_TIME = 0.1  # Number of seconds to wait after serving.

    def serve_gdb_thread(self, serve_result):
        """Serve requests on GDB thread."""
        try:
            deadline = time.time() + self.SERVE_TIME
            while True:
                timeout = deadline - time.time()
                if timeout < 0:
                    break
                super().serve(timeout=timeout)
        except Exception as exc:
            serve_result.set(exc)
        else:
            serve_result.set(None)

    def serve_all(self):
        """Modified version of rpyc.core.protocol.Connection.serve_all."""
        try:
            while not self.closed:
                serve_result = ServeResult()
                gdb.post_event(lambda: self.serve_gdb_thread(serve_result))
                serve_result.wait()
                time.sleep(self.IDLE_TIME)
        except (socket.error, select_error, IOError):
            if not self.closed:
                raise
        except EOFError:
            pass
        finally:
            self.close()


class GdbService(Service):
    """A public interface for Pwntools."""

    _protocol = GdbConnection  # Connection subclass.
    exposed_gdb = gdb  # ``gdb`` module.

    def exposed_set_breakpoint(self, client, has_stop, *args, **kwargs):
        """Create a breakpoint and connect it with the client-side mirror."""
        if has_stop:
            class Breakpoint(gdb.Breakpoint):
                def stop(self):
                    return client.stop()

            return Breakpoint(*args, **kwargs)
        return gdb.Breakpoint(*args, **kwargs)

    def exposed_set_finish_breakpoint(self, client, has_stop, has_out_of_scope, *args, **kwargs):
        """Create a finish breakpoint and connect it with the client-side mirror."""
        class FinishBreakpoint(gdb.FinishBreakpoint):
            if has_stop:
                def stop(self):
                    return client.stop()
            if has_out_of_scope:
                def out_of_scope(self):
                    client.out_of_scope()
        return FinishBreakpoint(*args, **kwargs)

    def exposed_quit(self):
        """Terminate GDB."""
        gdb.post_event(lambda: gdb.execute('quit'))


spawn(ThreadedServer(
    service=GdbService(),
    socket_path=socket_path,
    protocol_config={
        'allow_all_attrs': True,
        'allow_setattr': True,
    },
).start)

1	"""GDB Python API bridge."""
2	import gdb	×
3
4	import socket	×
5	from threading import Condition	×
6	import time	×
7
8	from rpyc.core.protocol import Connection	×
9	from rpyc.core.service import Service	×
10	from rpyc.lib import spawn	×
11	from rpyc.lib.compat import select_error	×
12	from rpyc.utils.server import ThreadedServer	×
13
14
15	class ServeResult:	×
16	"""Result of serving requests on GDB thread."""
17	def __init__(self):	×
18	self.cv = Condition()	×
19	self.done = False	×
20	self.exc = None	×
21
22	def set(self, exc):	×
23	with self.cv:	×
24	self.done = True	×
25	self.exc = exc	×
26	self.cv.notify()	×
27
28	def wait(self):	×
29	with self.cv:	×
30	while not self.done:	×
31	self.cv.wait()	×
32	if self.exc is not None:	×
33	raise self.exc	×
34
35
36	class GdbConnection(Connection):	×
37	"""A Connection implementation that serves requests on GDB thread.
38
39	Serving on GDB thread might not be ideal from the responsiveness
40	perspective, however, it is simple and reliable.
41	"""
42	SERVE_TIME = 0.1 # Number of seconds to serve.	×
43	IDLE_TIME = 0.1 # Number of seconds to wait after serving.	×
44
45	def serve_gdb_thread(self, serve_result):	×
46	"""Serve requests on GDB thread."""
47	try:	×
48	deadline = time.time() + self.SERVE_TIME	×
49	while True:	×
50	timeout = deadline - time.time()	×
51	if timeout < 0:	×
52	break	×
53	super().serve(timeout=timeout)	×
54	except Exception as exc:	×
55	serve_result.set(exc)	×
56	else:
57	serve_result.set(None)	×
58
59	def serve_all(self):	×
60	"""Modified version of rpyc.core.protocol.Connection.serve_all."""
61	try:	×
62	while not self.closed:	×
63	serve_result = ServeResult()	×
64	gdb.post_event(lambda: self.serve_gdb_thread(serve_result))	×
65	serve_result.wait()	×
66	time.sleep(self.IDLE_TIME)	×
67	except (socket.error, select_error, IOError):	×
68	if not self.closed:	×
69	raise	×
70	except EOFError:	×
71	pass	×
72	finally:
73	self.close()	×
74
75
76	class GdbService(Service):	×
77	"""A public interface for Pwntools."""
78
79	_protocol = GdbConnection # Connection subclass.	×
80	exposed_gdb = gdb # ``gdb`` module.	×
81
82	def exposed_set_breakpoint(self, client, has_stop, args, *kwargs):	×
83	"""Create a breakpoint and connect it with the client-side mirror."""
84	if has_stop:	×
85	class Breakpoint(gdb.Breakpoint):	×
86	def stop(self):	×
87	return client.stop()	×
88
89	return Breakpoint(args, *kwargs)	×
90	return gdb.Breakpoint(args, *kwargs)	×
91
92	def exposed_set_finish_breakpoint(self, client, has_stop, has_out_of_scope, args, *kwargs):	×
93	"""Create a finish breakpoint and connect it with the client-side mirror."""
94	class FinishBreakpoint(gdb.FinishBreakpoint):	×
95	if has_stop:	×
96	def stop(self):	×
97	return client.stop()	×
98	if has_out_of_scope:	×
99	def out_of_scope(self):	×
100	client.out_of_scope()	×
101	return FinishBreakpoint(args, *kwargs)	×
102
103	def exposed_quit(self):	×
104	"""Terminate GDB."""
105	gdb.post_event(lambda: gdb.execute('quit'))	×
106
107
108	spawn(ThreadedServer(	×
109	service=GdbService(),
110	socket_path=socket_path,
111	protocol_config={
112	'allow_all_attrs': True,
113	'allow_setattr': True,
114	},
115	).start)

Gallopsled / pwntools / 8008596090

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous