agentcore_cli.utils.command_executor

agentcore_cli.utils.command_executor

Command execution utilities with security validation for AgentCore CLI.

This module provides a centralized utility for executing shell commands with security validation. It ensures that all subprocess calls consistently capture stdout and stderr while maintaining proper error handling and security validation for AgentCore CLI use cases.

The module is designed to be used as a utility for other modules in the AgentCore CLI.

execute_command(cmd, check=False, text=True, log_cmd=True, log_output=True)

Execute a shell command and capture all output with security validation.

This is a centralized utility to ensure all subprocess calls consistently capture stdout and stderr while maintaining proper error handling and security validation for AgentCore CLI use cases.

Parameters:

Name	Type	Description	Default
cmd	Union[list[str], str]	Command to execute, either as list of arguments or shell string	required
check	bool	Whether to raise an exception if command fails	False
text	bool	Whether to decode output as text (vs bytes)	True
log_cmd	bool	Whether to log the command being executed	True
log_output	bool	Whether to log command output	True

Returns:

Type	Description
tuple[int, str, str]	Tuple[int, str, str]: (return_code, stdout, stderr)

Source code in agentcore_cli/utils/command_executor.py

def execute_command(
    cmd: Union[list[str], str], check: bool = False, text: bool = True, log_cmd: bool = True, log_output: bool = True
) -> tuple[int, str, str]:
    """Execute a shell command and capture all output with security validation.

    This is a centralized utility to ensure all subprocess calls consistently capture
    stdout and stderr while maintaining proper error handling and security validation
    for AgentCore CLI use cases.

    Args:
        cmd: Command to execute, either as list of arguments or shell string
        check: Whether to raise an exception if command fails
        text: Whether to decode output as text (vs bytes)
        log_cmd: Whether to log the command being executed
        log_output: Whether to log command output

    Returns:
        Tuple[int, str, str]: (return_code, stdout, stderr)
    """
    if log_cmd:
        if isinstance(cmd, list):
            logger.info(f"Executing: {' '.join(cmd)}")
        else:
            logger.info(f"Executing: {cmd}")

    # Validate command for security
    is_valid, error_msg = _validate_command_security(cmd)
    if not is_valid:
        logger.warning(f"Command rejected by security validation: {error_msg}")
        return -1, "", f"Command rejected: {error_msg}"

    try:
        # Handle shell commands with pipes for AWS ECR authentication
        if isinstance(cmd, str) and "|" in cmd and "aws ecr get-login-password" in cmd:
            # Special case for ECR authentication command - requires shell=True for pipe
            # nosemgrep: subprocess-shell-true
            result = subprocess.run(cmd, shell=True, check=check, text=text, capture_output=True)  # nosec: B602 inputs are validated
        else:
            # Standard execution without shell for security
            result = subprocess.run(cmd, shell=False, check=check, text=text, capture_output=True)  # nosec: B603 inputs are validated

        if log_output:
            if result.returncode == 0:
                if result.stdout and result.stdout.strip():
                    logger.debug(f"Command output: {result.stdout.strip()}")
            else:
                if result.stderr and result.stderr.strip():
                    logger.error(f"Command error: {result.stderr.strip()}")

        return result.returncode, result.stdout, result.stderr
    except subprocess.CalledProcessError as e:
        logger.error(f"Command failed with exit code {e.returncode}: {e.stderr}")
        return e.returncode, e.stdout or "", e.stderr or ""
    except Exception as e:
        logger.error(f"Error executing command: {str(e)}")
        return -1, "", str(e)