11532588290

Committed 26 Oct 2024 02:29PM CUT coverage: 73.585% (+2.1%) from 71.491%

Build # 11532588290

Build Type

push

github

Committed by

peace-maker

Commit Message

Merge branch 'dev' into interactive

Run Details

3767 of 6364 branches covered (59.19%)

1463 of 2234 new or added lines in 69 files covered. (65.49%)

85 existing lines in 22 files now uncovered.

13246 of 18001 relevant lines covered (73.58%)

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	×
UNCOV 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
NEW 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."""
NEW 94	class FinishBreakpoint(gdb.FinishBreakpoint):	×
NEW 95	if has_stop:	×
NEW 96	def stop(self):	×
NEW 97	return client.stop()	×
NEW 98	if has_out_of_scope:	×
NEW 99	def out_of_scope(self):	×
NEW 100	client.out_of_scope()	×
NEW 101	return FinishBreakpoint(args, *kwargs)	×
102
UNCOV 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 / 11532588290

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