diff --git a/backend/handlers/terminal/start.py b/backend/handlers/terminal/start.py index 86bb322..62a487d 100644 --- a/backend/handlers/terminal/start.py +++ b/backend/handlers/terminal/start.py @@ -1,4 +1,5 @@ """Terminal WebSocket start handler.""" +# pylint: disable=duplicate-code # Auth/client setup pattern is intentional from flask import request from flask_socketio import emit, disconnect from config import logger, sessions, active_terminals diff --git a/backend/routes/containers/exec.py b/backend/routes/containers/exec.py index 84982bc..631c3f1 100644 --- a/backend/routes/containers/exec.py +++ b/backend/routes/containers/exec.py @@ -4,9 +4,8 @@ from config import logger, session_workdirs from utils.auth import check_auth from utils.docker_client import get_docker_client from utils.exec_helpers import ( - build_bash_command, - build_sh_command, - execute_in_container, + get_session_workdir, + execute_command_with_fallback, decode_output, extract_workdir ) @@ -29,28 +28,23 @@ def exec_container(container_id): return jsonify({'error': 'Cannot connect to Docker'}), 500 try: - container = client.containers.get(container_id) + # Get session working directory + session_key, current_workdir = get_session_workdir(token, container_id, session_workdirs) - # Get or initialize session working directory - session_key = f"{token}_{container_id}" - if session_key not in session_workdirs: - session_workdirs[session_key] = '/' - - current_workdir = session_workdirs[session_key] - is_cd_command = user_command.strip().startswith('cd ') - - # Try bash first, fallback to sh if bash doesn't exist - try: - bash_command = build_bash_command(current_workdir, user_command, is_cd_command) - exec_instance = execute_in_container(container, bash_command) - except Exception as bash_error: # pylint: disable=broad-exception-caught - logger.warning("Bash execution failed, trying sh: %s", bash_error) - sh_command = build_sh_command(current_workdir, user_command, is_cd_command) - exec_instance = execute_in_container(container, sh_command) + # Execute command with bash/sh fallback + exec_instance = execute_command_with_fallback( + client.containers.get(container_id), + current_workdir, + user_command, + user_command.strip().startswith('cd ') + ) # Decode and extract workdir from output - output = decode_output(exec_instance) - output, new_workdir = extract_workdir(output, current_workdir, is_cd_command) + output, new_workdir = extract_workdir( + decode_output(exec_instance), + current_workdir, + user_command.strip().startswith('cd ') + ) # Update session workdir session_workdirs[session_key] = new_workdir diff --git a/backend/utils/exec_helpers.py b/backend/utils/exec_helpers.py index 29eaae4..7bf21b3 100644 --- a/backend/utils/exec_helpers.py +++ b/backend/utils/exec_helpers.py @@ -1,4 +1,44 @@ """Helper functions for container exec operations.""" +from config import logger + + +def get_session_workdir(token, container_id, session_workdirs): + """Get or initialize session working directory. + + Args: + token: Session token + container_id: Container ID + session_workdirs: Session workdir dictionary + + Returns: + tuple: (session_key, current_workdir) + """ + session_key = f"{token}_{container_id}" + if session_key not in session_workdirs: + session_workdirs[session_key] = '/' + return session_key, session_workdirs[session_key] + + +def execute_command_with_fallback(container, current_workdir, user_command, is_cd_command): + """Execute command in container with bash/sh fallback. + + Args: + container: Docker container object + current_workdir: Current working directory + user_command: User's command + is_cd_command: Whether this is a cd command + + Returns: + Docker exec instance + """ + # Try bash first + try: + bash_command = build_bash_command(current_workdir, user_command, is_cd_command) + return execute_in_container(container, bash_command) + except Exception as bash_error: # pylint: disable=broad-exception-caught + logger.warning("Bash execution failed, trying sh: %s", bash_error) + sh_command = build_sh_command(current_workdir, user_command, is_cd_command) + return execute_in_container(container, sh_command) def build_bash_command(current_workdir, user_command, is_cd_command):