826f4f39da04bd1b97e77addee38f8db4221f17a

Build # 826f4f39da04bd1b97e77addee38f8db4221f17a

Build Type

push

github-actions

Committed by GitHub

Commit Message

Create SECURITY.md

Run Details

3867 of 6353 branches covered (60.87%)

12184 of 16570 relevant lines covered (73.53%)

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_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,
    },
).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_quit(self):	×
93	"""Terminate GDB."""
94	gdb.post_event(lambda: gdb.execute('quit'))	×
95
96
97	spawn(ThreadedServer(	×
98	service=GdbService(),
99	socket_path=socket_path,
100	protocol_config={
101	'allow_all_attrs': True,
102	},
103	).start)

Gallopsled / pwntools / 826f4f39da04bd1b97e77addee38f8db4221f17a

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