mirror of https://github.com/hwchase17/langchain
community[major], experimental[patch]: Remove Python REPL from community (#22904)
Remove the REPL from community, and suggest an alternative import from langchain_experimental. Fix for this issue: https://github.com/langchain-ai/langchain/issues/14345 This is not a bug in the code or an actual security risk. The python REPL itself is behaving as expected. The PR is done to appease blanket security policies that are just looking for the presence of exec in the code. --------- Co-authored-by: Erick Friis <erick@langchain.dev>pull/22912/head
Eugene Yurtsev
4 months ago
committed by
GitHub
parent
9a877c7adb
commit
c72bcda4f2
@ -1,71 +1,17 @@
|
||||
import functools
|
||||
import logging
|
||||
import multiprocessing
|
||||
import sys
|
||||
from io import StringIO
|
||||
from typing import Dict, Optional
|
||||
|
||||
from langchain_core.pydantic_v1 import BaseModel, Field
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=None)
|
||||
def warn_once() -> None:
|
||||
"""Warn once about the dangers of PythonREPL."""
|
||||
logger.warning("Python REPL can execute arbitrary code. Use with caution.")
|
||||
|
||||
|
||||
class PythonREPL(BaseModel):
|
||||
"""Simulates a standalone Python REPL."""
|
||||
|
||||
globals: Optional[Dict] = Field(default_factory=dict, alias="_globals")
|
||||
locals: Optional[Dict] = Field(default_factory=dict, alias="_locals")
|
||||
|
||||
@classmethod
|
||||
def worker(
|
||||
cls,
|
||||
command: str,
|
||||
globals: Optional[Dict],
|
||||
locals: Optional[Dict],
|
||||
queue: multiprocessing.Queue,
|
||||
) -> None:
|
||||
old_stdout = sys.stdout
|
||||
sys.stdout = mystdout = StringIO()
|
||||
try:
|
||||
exec(command, globals, locals)
|
||||
sys.stdout = old_stdout
|
||||
queue.put(mystdout.getvalue())
|
||||
except Exception as e:
|
||||
sys.stdout = old_stdout
|
||||
queue.put(repr(e))
|
||||
|
||||
def run(self, command: str, timeout: Optional[int] = None) -> str:
|
||||
"""Run command with own globals/locals and returns anything printed.
|
||||
Timeout after the specified number of seconds."""
|
||||
|
||||
# Warn against dangers of PythonREPL
|
||||
warn_once()
|
||||
|
||||
queue: multiprocessing.Queue = multiprocessing.Queue()
|
||||
|
||||
# Only use multiprocessing if we are enforcing a timeout
|
||||
if timeout is not None:
|
||||
# create a Process
|
||||
p = multiprocessing.Process(
|
||||
target=self.worker, args=(command, self.globals, self.locals, queue)
|
||||
)
|
||||
|
||||
# start it
|
||||
p.start()
|
||||
|
||||
# wait for the process to finish or kill it after timeout seconds
|
||||
p.join(timeout)
|
||||
|
||||
if p.is_alive():
|
||||
p.terminate()
|
||||
return "Execution timed out"
|
||||
else:
|
||||
self.worker(command, self.globals, self.locals, queue)
|
||||
# get the result from the worker function
|
||||
return queue.get()
|
||||
def __getattr__(name: str) -> Any:
|
||||
if name in "PythonREPL":
|
||||
raise AssertionError(
|
||||
"PythonREPL has been deprecated from langchain_community due to being "
|
||||
"flagged by security scanners. See: "
|
||||
"https://github.com/langchain-ai/langchain/issues/14345 "
|
||||
"If you need to use it, please use the version "
|
||||
"from langchain_experimental. "
|
||||
"from langchain_experimental.utilities.python import PythonREPL."
|
||||
)
|
||||
raise AttributeError(f"module {__name__} has no attribute {name}")
|
||||
|
||||
Loading…
Reference in New Issue