# Author: Miguel Martinez Lopez
from rpyc.utils.server import ThreadedServer
from rpyc.utils.classic import DEFAULT_SERVER_PORT
from rpyc.core.service import Service, ModuleNamespace
from rpyc.lib.compat import execute, is_py3k
import sys
import os
import threading
import argparse
EXECUTED_PYTHON_FILE = False
def exec_python(filepath, namespace):
global EXECUTED_PYTHON_FILE
if EXECUTED_PYTHON_FILE:
raise Exception("exec_python can be used only one time")
EXECUTED_PYTHON_FILE = True
filepath = os.path.abspath(filepath)
sys.path = [os.path.dirname(filepath)] + sys.path[1:]
namespace["__file__"] = filepath
namespace["__name__"] = "__main__"
with open(filepath, 'rb') as file:
exec(compile(file.read(), filepath, 'exec'), namespace)
class PublicService(Service):
exposed_namespace = {}
def on_connect(self):
self._conn._config.update(dict(
allow_all_attrs = True,
allow_pickle = True,
allow_getattr = True,
allow_setattr = True,
allow_delattr = True,
import_custom_exceptions = True,
instantiate_custom_exceptions = True,
instantiate_oldstyle_exceptions = True,
))
# shortcuts
self._conn.modules = ModuleNamespace(self._conn.root.getmodule)
self._conn.eval = self._conn.root.eval
self._conn.execute = self._conn.root.execute
self._conn.namespace = self._conn.root.namespace
if is_py3k:
self._conn.builtin = self._conn.modules.builtins
else:
self._conn.builtin = self._conn.modules.__builtin__
self._conn.builtins = self._conn.builtin
def exposed_execute(self, text):
"""execute arbitrary code (using ``exec``)"""
execute(text, self.exposed_namespace)
def exposed_eval(self, text):
"""evaluate arbitrary code (using ``eval``)"""
return eval(text, self.exposed_namespace)
def exposed_getmodule(self, name):
"""imports an arbitrary module"""
return __import__(name, None, None, "*")
def exposed_getconn(self):
"""returns the local connection instance to the other side"""
return self._conn
parser = argparse.ArgumentParser(description='Remote debugging and testing')
parser.add_argument('filename', help="Path to script")
parser.add_argument('-p', '--port', action="store", dest="port", default=DEFAULT_SERVER_PORT, help="Remote interpreter port", type=int)
args = parser.parse_args()
thread = threading.Thread(target=lambda: ThreadedServer(PublicService, hostname = "localhost", port=args.port).start())
thread.daemon=True
thread.start()
exec_python(args.filename, PublicService.exposed_namespace)
Diff to Previous Revision
--- revision 2 2017-01-24 20:33:57
+++ revision 3 2017-01-24 20:34:52
@@ -11,15 +11,15 @@
import threading
import argparse
-EXECUTED_PYTHON = False
+EXECUTED_PYTHON_FILE = False
def exec_python(filepath, namespace):
- global EXECUTED_PYTHON
+ global EXECUTED_PYTHON_FILE
- if EXECUTED_PYTHON:
+ if EXECUTED_PYTHON_FILE:
raise Exception("exec_python can be used only one time")
- EXECUTED_PYTHON = True
+ EXECUTED_PYTHON_FILE = True
filepath = os.path.abspath(filepath)
sys.path = [os.path.dirname(filepath)] + sys.path[1:]
