a0c9da07acc34cf1eb7f532923c6c375420a52a1

pending completion

Build # a0c9da07acc34cf1eb7f532923c6c375420a52a1

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Fix ssh.process not setting ssh_process.cwd (#2241)

* Fix ssh.process not setting ssh_process.cwd

The cwd wasn't passed to the ssh_process created in ssh.process,
so the ssh_process.cwd attribute was never set.

This passes the cwd along now.
```
s = ssh(..)
io = s.process("pwd", cwd="/tmp")
io.cwd # "."
io.recvall() # b"/tmp\n"
```

* Update changelog

Run Details

3876 of 6362 branches covered (60.92%)

7 of 7 new or added lines in 1 file covered. (100.0%)

12204 of 16597 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_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,
    },
).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	},
114	).start)

Gallopsled / pwntools / a0c9da07acc34cf1eb7f532923c6c375420a52a1

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