Skip to content

Commit e3f6b7c

Browse files
authored
Wrote a quick remoting script because I needed one
1 parent aa4c995 commit e3f6b7c

File tree

1 file changed

+68
-0
lines changed

1 file changed

+68
-0
lines changed

scripts/remote.py

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
"""
2+
A slightly better way to remote with GDB/GEF
3+
4+
gdb -ex 'source /path/to/gef-extras/scripts/remote.py' -ex rpyc-remote -ex quit
5+
"""
6+
7+
from typing import Any
8+
9+
import rpyc
10+
import gdb
11+
import sys
12+
import contextlib
13+
14+
__AUTHOR__ = "hugsy"
15+
__VERSION__ = 0.1
16+
17+
18+
class GefRemoteService(rpyc.Service):
19+
"""The RPYC service for interacting with GEF"""
20+
21+
def exposed_gdb(self, cmd: str) -> str:
22+
return gdb.execute(cmd, to_string=True)
23+
24+
def exposed_gef(self, cmd: str) -> Any:
25+
return eval(cmd)
26+
27+
28+
class DisableStreamBufferContext(contextlib.ContextDecorator):
29+
"""Because stream buffering doesn't play well with rpyc"""
30+
def __enter__(self) -> None:
31+
info("Backuping context")
32+
self.old_stream_buffer = gef.ui.stream_buffer
33+
self.old_redirect_fd = gef.ui.redirect_fd
34+
gef.ui.stream_buffer = sys.stdout
35+
gef.ui.redirect_fd = None
36+
return self
37+
38+
def __exit__(self, _) -> bool:
39+
info("Restoring context")
40+
gef.ui.stream_buffer = self.old_stream_buffer
41+
gef.ui.redirect_fd = self.old_redirect_fd
42+
return False
43+
44+
45+
@register_external_command
46+
class GefRemoteCommand(GenericCommand):
47+
"""A better way of remoting to GDB, using rpyc"""
48+
49+
_cmdline_ = "rpyc-remote"
50+
_aliases_ = []
51+
_syntax_ = f"{_cmdline_:s}"
52+
_example_ = f"{_cmdline_:s}"
53+
54+
def __init__(self) -> None:
55+
super().__init__(prefix=False)
56+
self["host"] = ("0.0.0.0", "The interface to listen on")
57+
self["port"] = (12345, "The port to listen on")
58+
return
59+
60+
def do_invoke(self, _) -> None:
61+
with DisableStreamBufferContext():
62+
info(f"Listening on {self['host']}:{self['port']}, press Ctrl+C to stop")
63+
server = rpyc.utils.server.ThreadedServer(GefRemoteService, port=12345)
64+
try:
65+
server.start()
66+
except KeyboardInterrupt:
67+
info("Stopping")
68+
server.close()

0 commit comments

Comments
 (0)