How to define timeouts

[tsafs]

First of all, thank you for this work.

Is there an official way to define a timeout of the JavaScript running in .eval()? I want to prevent malicious code to run for longer than a specific amount of time.

I dug through the tests and source code. The only way I see it done is to

• either wrap the to be evaluated JS in a JS wrapper that fails if the task takes too long
• or handle this from within Python and call CEngine::TerminateAllThreads if the task takes too long.

Thanks

[tsafs]

After a lot of fiddeling around, I seem to not get it work in any way. What follows is the most intuitive way I'd envision it to work. However, that doesn't work, because it seems that by the time engine.terminateAllThreads() is called that the isolate is already not anymore active.

The main Python thread is blocked by the long-running and blocking JavaScript process. I.e. the script is only looping through the tasks once the script finished.

Do you have any idea whether this is solvable or whether changes to the source of STPyV8 must be made?

import asyncio
import STPyV8

async def run_js(custom_js_code: str, timeout_ms: int):
    engine_storage = { "engine": None }

    async def timeout_task():
        print("Timeout task started")
        await asyncio.sleep(timeout_ms / 1000)
        print("Timeout task completed")

    def script_task():
        print("Script task started")
        with STPyV8.JSIsolate():
            with STPyV8.JSContext():
                engine_storage["engine"] = STPyV8.JSEngine()
                script = engine_storage["engine"].compile(custom_js_code)
                result = script.run()
                del engine_storage["engine"]
                print("Script task completed")
                return result

    timeout_task_future = asyncio.create_task(timeout_task())
    script_task_future = asyncio.create_task(asyncio.to_thread(script_task))

    tasks = [timeout_task_future, script_task_future]
    done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    # Get result from the first completed task
    for task in done:
        if task.exception():
            print(f"Task ended with exception: {task.exception()}")
        else:
            if (task == timeout_task_future):
                # Terminate all isolate threads if timeout was reached first
                engine_storage["engine"].terminateAllThreads()
                del engine_storage["engine"]

            elif task == script_task_future:
                # Handle script output
                print(f"Task result {task.result()}")

custom_js = """
// sleep for 5 seconds
var start = Date.now();
while(Date.now() - start < 5000);
"""

timeout = 1000

async def main():
    await run_js(custom_js, timeout)

asyncio.run(main())