EEP: Making QuestionFunctional serializable

[johnjosephhorton]

QuestionFunctional allows the user to have a Python function that can generate an answer in response to scenario and agent attributes. However, this makes it not safely serializable. This example shows a way we could do it using RestrictedPython.

1. When instantiated, we grab the source code for the function
2. When serialized, the source code and function name are stored as strings
3. When deserialized, the source code and function are added to the object but not activated unless user calls activate function that uses the RestrcitedPython version
4. We have an "activate_dangerously" function that allows any function.

@zer0dss can you review and give your thoughts on safety?

import inspect
import textwrap

from RestrictedPython import compile_restricted, safe_globals, limited_builtins
from RestrictedPython.Eval import default_guarded_getitem
from RestrictedPython.Guards import safe_builtins, full_write_guard, guarded_iter_unpack_sequence

class QuestionFunctionlRunningException(Exception):
    pass

class QuestionFunctionlActivatedException(Exception):
    pass

class DummyQuestion:

    def __init__(self, func: None):
        if func is not None:
            self._func = func
            self.activated = True
            self.source_code = inspect.getsource(func)
            self.function_name = func.__name__
        else:
            self._func = None
            self.activated = False
            self.source_code = None
            self.function_name = None

    def func(self, *args, **kwargs):
        if self.activated:
            try:
                return self._func(*args, **kwargs)
            except Exception as e:
                # This is likely to be a RestrictedPython error
                print("An error occurred:", e)
                raise QuestionFunctionlRunningException(
                    textwrap.dedent("""\
                The reason you received an exception is that the function might not safe to run.
                edsl uses RestrictedPython <https://pypi.org/project/RestrictedPython/>.
                If you you are ***sure*** it is OK to run, you can activate the function with the `activate_dangerously` method
                >>> q.activate_dangerously()
                """))
        else:
            raise QuestionFunctionlActivatedException("Function not activated. Please activate it first with `activate` method.")

    def to_dict(self):
        return {"function_source_code": self.source_code, 'function_name': self.function_name}
    
    def activate(self) -> None:
        """Activate the function using RestrictedPython."""
        safe_env = safe_globals.copy()
        safe_env['__builtins__'] = {**safe_builtins, 'getattr': getattr}

        byte_code = compile_restricted(self.source_code, '<string>', 'exec')
        loc = {}
        try:
            exec(byte_code, safe_env, loc)
            self._func = loc[self.function_name]
            self.activated = True
        except Exception as e:
            print("An error occurred:", e)

    def activate_dangerously(self) -> None:
        """Activate the function without checking the safety."""
        exec(self.source_code, globals(), locals())
        self._func = locals()[self.function_name]
        self.activated = True

    @classmethod
    def from_dict(cls, data:dict, activate:bool=False):
        """Create a `QuestionFunctional` from a dictionary."""
        new_q = cls(None)
        # Not allowing loops for now
        #safe_env['_getiter_'] = iter
        #safe_env['_iter_unpack_sequence_'] = guarded_iter_unpack_sequence
        #safe_env['_write_'] = full_write_guard  # Optional, for variable assignments within the loop
        #safe_env['_inplacevar_'] = _inplacevar_  # Handle augmented assignments
        new_q.source_code = data["function_source_code"]
        new_q.function_name = data["function_name"]
        new_q.activated = False
        if activate:
            new_q.activate()
        return new_q

def user_function(x):
    y = 0
    for i in range(100):
        y += x
    return y

q1 = DummyQuestion(user_function)

q2 = DummyQuestion.from_dict(q1.to_dict())
try:
    print(q2.func(12))
except QuestionFunctionlActivatedException as e:
    print("Doesn't work because the function is not activated yet.")
    
try:
    q2.activate()
    print(q2.func(12))
except QuestionFunctionlRunningException as e:
    print("Doesn't work because the function has a loop.")
    
    
q2.activate_dangerously()
print(q2.func(12))    

[zer0dss]

@johnjosephhorton using 'getattr': getattr is not safe in extreme cases. By default inside safe_builtins getattr is defined as safer_getattr and we can use that.
Using safe_env['__builtins__'] = {**safe_builtins} is safer.

[johnjosephhorton]

I see - want to re-factor this code to do that? We might also consider re-factor "activate()" to allow for different levels of security. E.g., one that allows loops, for example.