Skip to content

cillow.Interpreter

Create an interpreter from a given python environment.

Examples:

>>> import cillow
>>>
>>> interpreter = cillow.Interpreter()
>>>
>>> interpreter.run_code("x = 1; x + 1")
Result(value=2)
>>>
>>> interpreter.install_requirements(["package-name"])

Don't trust LLMS? Concerned about arbitrary code execution? Take full control by limiting functionalities using patches.

To add patches, use the add_patches() function. To clear patches, use clear_patches().

Examples:

>>> import cillow
>>>
>>> import os
>>> from contextlib import contextmanager
>>>
>>> os_system_switchable = cillow.Switchable(os.system)
>>>
>>> @contextmanager
... def patch_os_system():
...     def disabled_os_system(command: str):
...         return "os.system has been disabled."
...
...     with os_system_switchable.switch_to(disabled_os_system):
...         yield
...
>>> cillow.add_patches(patch_os_system)  # Disable os.system
>>>
>>> interpreter = cillow.Interpreter()
>>> interpreter.run_code("import os;os.system('echo Hi')")
Result(value='os.system has been disabled.')
Source code in cillow/interpreter.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class Interpreter:
    """
    Create an interpreter from a given python environment.

    Examples:
        >>> import cillow
        >>>
        >>> interpreter = cillow.Interpreter()
        >>>
        >>> interpreter.run_code("x = 1; x + 1")
        Result(value=2)
        >>>
        >>> interpreter.install_requirements(["package-name"])

    Don't trust LLMS? Concerned about arbitrary code execution?
    Take full control by limiting functionalities using patches.

    To add patches, use the `add_patches()` function. To clear patches, use `clear_patches()`.

    Examples:
        >>> import cillow
        >>>
        >>> import os
        >>> from contextlib import contextmanager
        >>>
        >>> os_system_switchable = cillow.Switchable(os.system)
        >>>
        >>> @contextmanager
        ... def patch_os_system():
        ...     def disabled_os_system(command: str):
        ...         return "os.system has been disabled."
        ...
        ...     with os_system_switchable.switch_to(disabled_os_system):
        ...         yield
        ...
        >>> cillow.add_patches(patch_os_system)  # Disable os.system
        >>>
        >>> interpreter = cillow.Interpreter()
        >>> interpreter.run_code("import os;os.system('echo Hi')")
        Result(value='os.system has been disabled.')
    """

    def __init__(self, environment: PythonEnvironment = "$system") -> None:
        self.namespace: dict[str, Any] = {}
        self._import_hook = None
        if environment != "$system":
            self._import_hook = EnvironmentImportHook(environment)
            sys.meta_path.insert(0, self._import_hook)
            sys.path.insert(0, str(self._import_hook.site_packages))

    @property
    def environment(self) -> PythonEnvironment:
        """The current Python environment"""
        return getattr(self._import_hook, "environment", "$system")

    # fmt: off
    def run_command(
        self, *cmd: str, on_stream: Callable[[Stream], Any] | None = None
    ) -> None:
        # fmt: on
        """
        Run the given command.

        ⚠️ WARNING: This class allows execution of system commands and should be used with EXTREME CAUTION.

        - Never run commands with user-supplied or untrusted input
        - Always validate and sanitize any command arguments
        - Be aware of potential security risks, especially with privilege escalation

        Args:
            cmd: The command to run
            on_stream: The callback to capture streaming output.
        """
        on_stream = on_stream or default_stream_processor
        for line in shell.stream(*cmd):
            on_stream(Stream(type="cmd_exec", data=line))

    # fmt: off
    def install_requirements(
        self, *requirements: str, on_stream: Callable[[Stream], None] | None = None
    ) -> None:
        # fmt: on
        """
        Install the given requirements.

        Args:
            requirements: The requirements to install
            on_stream: The callback to capture streaming output. (defaults to print)
        """
        install_args = []
        if self._import_hook is not None:
            install_args = ["--python", str(self._import_hook.environment)]

        with NamedTemporaryFile(mode="w") as handler:
            # Writing requirements to file, instead of directly
            # passing them as arguments, to avoid arbitrary command execution
            handler.write("\n".join(requirements))
            handler.flush()
            install_args.extend(["-r", handler.name])

            self.run_command(*PIP_INSTALL_CMD, *install_args, on_stream=on_stream)

    def run_code(
        self, code: str, on_stream: Callable[[Stream | ByteStream], None] | None = None
    ) -> Result | ExceptionInfo:
        """
        Run the given code.

        Args:
            code: The code to run
            on_stream: The callback to capture streaming output.

        Returns:
            Result or ExceptionInfo dataclass instance
        """
        try:
            code_meta = CodeMeta.from_code(code, filename="interpreter-process")
        except Exception as exc:
            return ExceptionInfo(type=exc.__class__.__name__, message=str(exc))

        on_stream = on_stream or default_stream_processor
        if not is_auto_install_disabled() and (module_names := code_meta.module_names):
            to_install = (module_names - sys.stdlib_module_names) - get_installed_modules()
            if to_install:
                packages = [MODULE_TO_PACKAGE_MAP.get(name, name) for name in to_install]
                self.install_requirements(*packages, on_stream=on_stream)

        try:
            with patch.load_patches(on_stream=on_stream):
                if to_exec := code_meta.to_exec:
                    exec(to_exec, self.namespace, self.namespace)  # noqa: S102

                result_value = None
                if to_eval := code_meta.to_eval:
                    result_value = eval(to_eval, self.namespace, self.namespace)  # noqa: S307

            return Result(value=result_value)

        except Exception as exc:
            exc_info = {
                "type": exc.__class__.__name__,
                "message": str(exc),
            }
            if tb := exc.__traceback__:
                exc_info["where"] = format_traceback(tb)[-1].strip()

            return ExceptionInfo(**exc_info)

    def __del__(self) -> None:
        if self._import_hook:
            sys.meta_path.pop(0)
            sys.path.pop(0)

environment property

The current Python environment

install_requirements

Install the given requirements.

Parameters:

Name Type Description Default
requirements str

The requirements to install

 ()
on_stream Callable[[Stream], None] | None

The callback to capture streaming output. (defaults to print)

 None
Source code in cillow/interpreter.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def install_requirements(
    self, *requirements: str, on_stream: Callable[[Stream], None] | None = None
) -> None:
    # fmt: on
    """
    Install the given requirements.

    Args:
        requirements: The requirements to install
        on_stream: The callback to capture streaming output. (defaults to print)
    """
    install_args = []
    if self._import_hook is not None:
        install_args = ["--python", str(self._import_hook.environment)]

    with NamedTemporaryFile(mode="w") as handler:
        # Writing requirements to file, instead of directly
        # passing them as arguments, to avoid arbitrary command execution
        handler.write("\n".join(requirements))
        handler.flush()
        install_args.extend(["-r", handler.name])

        self.run_command(*PIP_INSTALL_CMD, *install_args, on_stream=on_stream)

run_code

Run the given code.

Parameters:

Name Type Description Default
code str

The code to run

 required
on_stream Callable[[Stream | ByteStream], None] | None

The callback to capture streaming output.

 None

Returns:

Type Description
Result | ExceptionInfo

Result or ExceptionInfo dataclass instance
Source code in cillow/interpreter.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def run_code(
    self, code: str, on_stream: Callable[[Stream | ByteStream], None] | None = None
) -> Result | ExceptionInfo:
    """
    Run the given code.

    Args:
        code: The code to run
        on_stream: The callback to capture streaming output.

    Returns:
        Result or ExceptionInfo dataclass instance
    """
    try:
        code_meta = CodeMeta.from_code(code, filename="interpreter-process")
    except Exception as exc:
        return ExceptionInfo(type=exc.__class__.__name__, message=str(exc))

    on_stream = on_stream or default_stream_processor
    if not is_auto_install_disabled() and (module_names := code_meta.module_names):
        to_install = (module_names - sys.stdlib_module_names) - get_installed_modules()
        if to_install:
            packages = [MODULE_TO_PACKAGE_MAP.get(name, name) for name in to_install]
            self.install_requirements(*packages, on_stream=on_stream)

    try:
        with patch.load_patches(on_stream=on_stream):
            if to_exec := code_meta.to_exec:
                exec(to_exec, self.namespace, self.namespace)  # noqa: S102

            result_value = None
            if to_eval := code_meta.to_eval:
                result_value = eval(to_eval, self.namespace, self.namespace)  # noqa: S307

        return Result(value=result_value)

    except Exception as exc:
        exc_info = {
            "type": exc.__class__.__name__,
            "message": str(exc),
        }
        if tb := exc.__traceback__:
            exc_info["where"] = format_traceback(tb)[-1].strip()

        return ExceptionInfo(**exc_info)

run_command

Run the given command.

⚠️ WARNING: This class allows execution of system commands and should be used with EXTREME CAUTION.

  • Never run commands with user-supplied or untrusted input
  • Always validate and sanitize any command arguments
  • Be aware of potential security risks, especially with privilege escalation

Parameters:

Name Type Description Default
cmd str

The command to run

 ()
on_stream Callable[[Stream], Any] | None

The callback to capture streaming output.

 None
Source code in cillow/interpreter.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def run_command(
    self, *cmd: str, on_stream: Callable[[Stream], Any] | None = None
) -> None:
    # fmt: on
    """
    Run the given command.

    ⚠️ WARNING: This class allows execution of system commands and should be used with EXTREME CAUTION.

    - Never run commands with user-supplied or untrusted input
    - Always validate and sanitize any command arguments
    - Be aware of potential security risks, especially with privilege escalation

    Args:
        cmd: The command to run
        on_stream: The callback to capture streaming output.
    """
    on_stream = on_stream or default_stream_processor
    for line in shell.stream(*cmd):
        on_stream(Stream(type="cmd_exec", data=line))