[#29] subprocess로 코드 실행

app/main.py

@@ -36,7 +36,6 @@
 # 핸들러 등록
 exception_handlers.setup_exception_handlers(app)
 
-
 @app.get("/edupi-assist/health-check", response_class=JSONResponse)
 def root():
     return JSONResponse(

app/route/execute/service/execute_service.py

@@ -1,92 +1,42 @@
-import os
 import re
 import subprocess
-import tempfile
-import textwrap
 
-from RestrictedPython import compile_restricted, PrintCollector
-
-from app.config.restricted_python_config import RestrictedPythonConfig
 from app.route.execute.exception.code_execute_error import CodeExecuteError
 from app.route.execute.exception.code_syntax_error import CodeSyntaxError
-from app.route.execute.exception.input_size_matching_error import InputSizeMatchingError
 from app.web.exception.enum.error_enum import ErrorEnum
+from app.web.exception.task_fail_exception import TaskFailException
+
+FORBIDDEN_IMPORTS = ["os", "sys", "subprocess", "shutil"]
 
 
 def execute_code(source_code: str, user_input: str):
-    code = textwrap.dedent(source_code)
-    restricted_config = RestrictedPythonConfig(user_input)
+    if _contains_forbidden_imports(source_code):
+        # 보안상 실행 안함
+        raise CodeExecuteError(ErrorEnum.CODE_EXEC_SECURITY_ERROR)
 
     try:
-        byte_code = compile_restricted(code, filename="<string>", mode="exec")
-        restricted_locals = restricted_config.get_limited_locals()
-        restricted_globals = restricted_config.get_limited_globals()
-
-        exec(byte_code, restricted_globals, restricted_locals)
-
-        return _get_print_result(restricted_locals)
-
-    except InputSizeMatchingError as e:
-        raise CodeExecuteError(e.error_enum)
-
-    except SyntaxError as e:
-        error = _get_error_message(source_code)
-        raise CodeSyntaxError(ErrorEnum.CODE_SYNTAX_ERROR, {"error": error} if e.args else {})
+        process = subprocess.run(
+            args=["python3", "-c", source_code],
+            input=user_input,
+            capture_output=True,    # stdout, stderr 별도의 Pipe에서 처리
+            timeout=3,  # limit child process execute time
+            check=True,  # CalledProcessError exception if return_code is 0
+            text=True
+        )
+        return process.stdout
+
+    # 프로세스 실행 중 비정상 종료
+    except subprocess.CalledProcessError as e:
+        raise CodeSyntaxError(ErrorEnum.CODE_SYNTAX_ERROR, {"error": e.stderr})
 
     except Exception as e:
-        error = _get_error_message(source_code)
-        raise CodeExecuteError(ErrorEnum.CODE_EXEC_ERROR, {"error": error} if e.args else {})
-
-
-def _get_print_result(restricted_locals):
-    """_print 변수가 존재할 경우, 그 결과 반환."""
-    if "_print" in restricted_locals:
-        if isinstance(restricted_locals["_print"], PrintCollector):
-            return "".join(restricted_locals["_print"].txt)  # 리스트 요소를 합쳐 하나의 문자열로
-
-    return ""
-
-
-def _get_error_message(code):
-    """ 잘못된 코드의 에러 메시지를 추출 ex)'1:26: E999 SyntaxError: '(' was never closed' """
-    code = textwrap.dedent(code)
-    temp_file_path = _create_temp_file_with_code(code)
-
-    result = _run_flake8(temp_file_path)
-    os.remove(temp_file_path)
-
-    if result.returncode != 0:
-        return _extract_error_message(result.stdout)
-
-    return None
-
-
-def _create_temp_file_with_code(code) -> str:
-    """ 임시 파일에 코드 저장후, 파일 경로 반환 """
-    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as temp_file:
-        temp_file.write(code)
-        return temp_file.name
-
-
-def _run_flake8(temp_file_path: str) -> subprocess.CompletedProcess:
-    """ Flake8을 실행하여 코드 검사 """
-    result = subprocess.run(
-        ['flake8', temp_file_path],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        text=True
-    )
-    return result
-
+        raise TaskFailException(ErrorEnum.CODE_EXEC_SERVER_ERROR, dict(e.args))
 
-def _extract_error_message(error_string) -> str:
-    # [행:열: error message] 형태로 추출
-    pattern = r"(\d+:\d+: [^\n]+)"
 
-    # 정규 표현식을 사용하여 매칭된 부분 추출
-    match = re.search(pattern, error_string)
+def _contains_forbidden_imports(code: str) -> bool:
+    # 금지된 모듈이 코드에 있는지 확인
+    for module in FORBIDDEN_IMPORTS:
+        if re.search(rf'\bimport\s+{module}\b', code):
+            return True
+    return False
 
-    if match:
-        return match.group(1)  # 매칭된 부분을 반환
-    else:
-        return "No match found."

app/web/exception/enum/error_enum.py

@@ -4,18 +4,21 @@
 class ErrorEnum(Enum):
 
     # 400
-    TASK_FAIL = "CS-400001", "잠시 후 다시 시도 해주세요"
-    CODE_SYNTAX_ERROR = "CS-400002", "잘못된 문법입니다."
-    CODE_CORRECT_FAIL = ("CS-400003", "코드 교정에 실패 했습니다.")
-    UNKNOWN_ERROR = ("CS-400999", "처리가 필요한 에러입니다.")
+    TASK_FAIL = "CS-400001", "There is a problem with the service login"
+    CODE_SYNTAX_ERROR = "CS-400002", "The code is incorrect syntax"
+    CODE_CORRECT_FAIL = "CS-400003", "code correct fail"
+    UNKNOWN_ERROR = "CS-400999", "The unexpected error"
 
-    CODE_EXEC_ERROR = "CS-400004", "지원하지 않는 형식입니다."
-    INPUT_SIZE_MATCHING_ERROR = "CS-400005", "사용자 입력 개수가 일치하지 않습니다."
-    CODE_VISUALIZE_ERROR = "CS-400006", "아직 시각화할 수 없는 문법이 포함되어 있습니다."
+    CODE_EXEC_ERROR = "CS-400004", "The format is not supported for security reasons"
+    CODE_EXEC_SECURITY_ERROR = "CS-400004", "The format is not supported for security reasons"
+    INPUT_SIZE_MATCHING_ERROR = "CS-400005", "The number of user inputs does not match."
+    CODE_VISUALIZE_ERROR = "CS-400006", "It contains syntax that we can't visualize yet."
 
     #500
-    OPENAI_SERVER_ERROR = ("CS-504001", "Open AI internal server error")
-    OPENAI_MAX_TOKEN_LIMIT = ("CS-504002", "Open AI max token limit")
+    CODE_EXEC_SERVER_ERROR = "CS-503001", "There is a problem with the execute service login"
+    OPENAI_SERVER_ERROR = "CS-504001", "Open AI internal server error"
+    OPENAI_MAX_TOKEN_LIMIT = "CS-504002", "Open AI max token limit"
+
 
     def __init__(self, code, detail):
         self.code = code

app/web/exception/task_fail_exception.py

@@ -0,0 +1,14 @@
+from starlette import status
+
+from app.web.exception.base_exception import BaseCustomException
+from app.web.exception.enum.error_enum import ErrorEnum
+
+
+class TaskFailException(BaseCustomException):
+    def __init__(self, error_enum: ErrorEnum, result: dict = None):
+        super().__init__(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            error_enum=error_enum,
+            result={} if result is None else result
+        )
+

app/web/exception_handlers.py

@@ -12,6 +12,7 @@
 from app.web.exception.enum.error_enum import ErrorEnum
 from app.web.exception.invalid_exception import InvalidException
 from app.route.advice.exception.openai_exception import OpenaiException
+from app.web.exception.task_fail_exception import TaskFailException
 from app.web.models.error_response import ErrorResponse
 
 
@@ -29,7 +30,8 @@ async def invalid_exception_handler(request: Request, exc: InvalidException):
         )
 
     @app.exception_handler(OpenaiException)
-    async def openai_exception_handler(request: Request, exc: OpenaiException):
+    @app.exception_handler(TaskFailException)
+    async def openai_exception_handler(request: Request, exc: OpenaiException | TaskFailException):
         logger.info(
             f"{exc.error_enum.code} - {exc.error_enum.detail}\n")
 

tests/conftest.py

@@ -0,0 +1,2 @@
+def test_placeholder():
+    assert True

tests/route/execute/test_execute_service.py

@@ -0,0 +1,7 @@
+from app.route.execute.service import execute_service
+
+
+def test__contains_forbidden_imports():
+    code = "import os\n"
+
+    assert execute_service._contains_forbidden_imports(code) is True