diff --git a/app/main.py b/app/main.py index d235c45..870286a 100644 --- a/app/main.py +++ b/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( diff --git a/app/route/execute/service/execute_service.py b/app/route/execute/service/execute_service.py index 13e4465..4c7144c 100644 --- a/app/route/execute/service/execute_service.py +++ b/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="", 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." diff --git a/app/web/exception/enum/error_enum.py b/app/web/exception/enum/error_enum.py index bdd988b..b76b9ab 100644 --- a/app/web/exception/enum/error_enum.py +++ b/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 diff --git a/app/web/exception/task_fail_exception.py b/app/web/exception/task_fail_exception.py new file mode 100644 index 0000000..5e22ff8 --- /dev/null +++ b/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 + ) + diff --git a/app/web/exception_handlers.py b/app/web/exception_handlers.py index 7f15e5f..a94a763 100644 --- a/app/web/exception_handlers.py +++ b/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") diff --git a/tests/conftest.py b/tests/conftest.py index e69de29..601241a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -0,0 +1,2 @@ +def test_placeholder(): + assert True \ No newline at end of file diff --git a/tests/route/execute/service/test_execute_service.py b/tests/route/execute/service/test_execute_service.py deleted file mode 100644 index a5dbef9..0000000 --- a/tests/route/execute/service/test_execute_service.py +++ /dev/null @@ -1,57 +0,0 @@ -import os - -from app.route.execute.service import execute_service - - -def test__create_temp_file_with_code(): - # 테스트할 코드 샘플 - sample_code = "print('Hello, World!')" - - temp_file_path = execute_service._create_temp_file_with_code(sample_code) - - try: - assert os.path.isfile(temp_file_path) - - # 파일 내용을 확인 - with open(temp_file_path, 'r') as temp_file: - file_content = temp_file.read() - assert file_content == sample_code - finally: - # 테스트 후 임시 파일 삭제 - if os.path.isfile(temp_file_path): - os.remove(temp_file_path) - - -def test__run_flake8_fail(): - sample_code = '\nprint(a)\n' - temp_file_path = execute_service._create_temp_file_with_code(sample_code) - - result = execute_service._run_flake8(temp_file_path) - # 테스트 후 임시 파일 삭제 - if os.path.isfile(temp_file_path): - os.remove(temp_file_path) - - print(result.stdout) - assert result.returncode != 0 - - -def test__run_flake8_success(): - sample_code = '\nprint("hello world")\n' - temp_file_path = execute_service._create_temp_file_with_code(sample_code) - - result = execute_service._run_flake8(temp_file_path) - # 테스트 후 임시 파일 삭제 - if os.path.isfile(temp_file_path): - os.remove(temp_file_path) - - assert result.returncode == 0 - - -def test__extract_error_message(): - origin = "/var/folders/0n/yb899qnj0vddx24q31c2t45h0000gn/T/tmpigduybq_.py:2:7: F821 undefined name 'a'" - - result = execute_service._extract_error_message(origin) - - assert result == "2:7: F821 undefined name 'a'" - - diff --git a/tests/route/execute/test_execute_service.py b/tests/route/execute/test_execute_service.py new file mode 100644 index 0000000..6d67928 --- /dev/null +++ b/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