限制IO访问

@contextlib.contextmanager
def restricted_io():
    original_open = builtins.open

    def restricted_open(*args, **kwargs):
        raise PermissionError("File IO operations are forbidden.")

    builtins.open = restricted_open
    try:
        yield
    finally:
        builtins.open = original_open

class ScriptEngineHandler:
    def with_restricted_context(func):
        from pp_utils import restricted_io
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with restricted_io():
                return func(*args, **kwargs)

        return wrapper

    @with_restricted_context
    def handle(self, req: fc3context.EngineRequest):
        raise NotImplementedError

这个是

class ExecHandler(ScriptEngineHandler):
    @ScriptEngineHandler.with_restricted_context
    def handle(self, req: fc3context.EngineRequest):
        code = req.raw_code
        if not code:
            return None
        loader = importlib.machinery.SourceFileLoader("user_module", "/path/to/user_module.py")
        loader.get_code = lambda _: compile(code, loader.name, 'exec')
        spec = importlib.util.spec_from_loader(loader.name, loader)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        params_count = check_function(module, 'main')
        setattr(module, '_ctx_node_', fc3context.NodeResponse(getattr(req, '_ctx_node_', None)))
        algo_module = importlib.import_module('pp_algo')
        # 将参数写入模块的上下文，这样代码中就可以直接使用 「_ctx_node_」
        setattr(algo_module, '_ctx_node_', fc3context.NodeResponse(getattr(req, '_ctx_node_', None)))
        if params_count == 0:
            return module.main()
        else:
            return module.main(req.event)

这里是将参数写入request, 后续可将request中的变量写入上下文

class EngineRequest:
    def __init__(self, event: Optional[dict] = None):
        self.raw_code = None
        self.script_name = None
        self.script_engine: Optional[Literal['py310', 'eval', 'exec']] = None
        if not event:
            return
        for key, value in event.items():
            setattr(self, key, value)

推荐您阅读更多有关于“”的文章