#!/usr/bin/python2 # Copyright 2015 The Chromium OS Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """EC-3PO Interactive Console Interface console provides the console interface between the user and the interpreter. It handles the presentation of the EC console including editing methods as well as session-persistent command history. """ from __future__ import print_function import argparse import binascii # pylint: disable=cros-logging-import import logging import multiprocessing import os import pty import select import stat import sys import interpreter PROMPT = '> ' CONSOLE_INPUT_LINE_SIZE = 80 # Taken from the CONFIG_* with the same name. CONSOLE_MAX_READ = 100 # Max bytes to read at a time from the user. # The timeouts are really only useful for enhanced EC images, but otherwise just # serve as a delay for non-enhanced EC images. Therefore, we can keep this # value small enough so that there's not too much of a delay, but long enough # that enhanced EC images can respond in time. Once we've detected an enhanced # EC image, we can increase the timeout for stability just in case it takes a # bit longer to receive an ACK for some reason. NON_ENHANCED_EC_INTERROGATION_TIMEOUT = 0.3 # Maximum number of seconds to wait # for a response to an # interrogation of a non-enhanced # EC image. ENHANCED_EC_INTERROGATION_TIMEOUT = 1.0 # Maximum number of seconds to wait for # a response to an interrogation of an # enhanced EC image. class EscState(object): """Class which contains an enumeration for states of ESC sequences.""" ESC_START = 1 ESC_BRACKET = 2 ESC_BRACKET_1 = 3 ESC_BRACKET_3 = 4 ESC_BRACKET_8 = 5 class ControlKey(object): """Class which contains codes for various control keys.""" BACKSPACE = 0x08 CTRL_A = 0x01 CTRL_B = 0x02 CTRL_D = 0x04 CTRL_E = 0x05 CTRL_F = 0x06 CTRL_K = 0x0b CTRL_N = 0xe CTRL_P = 0x10 CARRIAGE_RETURN = 0x0d ESC = 0x1b class Console(object): """Class which provides the console interface between the EC and the user. This class essentially represents the console interface between the user and the EC. It handles all of the console editing behaviour Attributes: logger: A logger for this module. master_pty: File descriptor to the master side of the PTY. Used for driving output to the user and receiving user input. user_pty: A string representing the PTY name of the served console. cmd_pipe: A multiprocessing.Connection object which represents the console side of the command pipe. This must be a bidirectional pipe. Console commands and responses utilize this pipe. dbg_pipe: A multiprocessing.Connection object which represents the console's read-only side of the debug pipe. This must be a unidirectional pipe attached to the intepreter. EC debug messages use this pipe. oobm_queue: A multiprocessing.Queue which is used for out of band management for the interactive console. input_buffer: A string representing the current input command. input_buffer_pos: An integer representing the current position in the buffer to insert a char. partial_cmd: A string representing the command entered on a line before pressing the up arrow keys. esc_state: An integer represeting the current state within an escape sequence. line_limit: An integer representing the maximum number of characters on a line. history: A list of strings containing the past entered console commands. history_pos: An integer representing the current history buffer position. This index is used to show previous commands. prompt: A string representing the console prompt displayed to the user. enhanced_ec: A boolean indicating if the EC image that we are currently communicating with is enhanced or not. Enhanced EC images will support packed commands and host commands over the UART. This defaults to False until we perform some handshaking. interrogation_timeout: A float representing the current maximum seconds to wait for a response to an interrogation. """ def __init__(self, master_pty, user_pty, cmd_pipe, dbg_pipe): """Initalises a Console object with the provided arguments. Args: master_pty: File descriptor to the master side of the PTY. Used for driving output to the user and receiving user input. user_pty: A string representing the PTY name of the served console. cmd_pipe: A multiprocessing.Connection object which represents the console side of the command pipe. This must be a bidirectional pipe. Console commands and responses utilize this pipe. dbg_pipe: A multiprocessing.Connection object which represents the console's read-only side of the debug pipe. This must be a unidirectional pipe attached to the intepreter. EC debug messages use this pipe. """ logger = logging.getLogger('EC3PO.Console') self.logger = interpreter.LoggerAdapter(logger, {'pty': user_pty}) self.master_pty = master_pty self.user_pty = user_pty self.cmd_pipe = cmd_pipe self.dbg_pipe = dbg_pipe self.oobm_queue = multiprocessing.Queue() self.input_buffer = '' self.input_buffer_pos = 0 self.partial_cmd = '' self.esc_state = 0 self.line_limit = CONSOLE_INPUT_LINE_SIZE self.history = [] self.history_pos = 0 self.prompt = PROMPT self.enhanced_ec = False self.interrogation_timeout = NON_ENHANCED_EC_INTERROGATION_TIMEOUT def __str__(self): """Show internal state of Console object as a string.""" string = [] string.append('master_pty: %s' % self.master_pty) string.append('user_pty: %s' % self.user_pty) string.append('cmd_pipe: %s' % self.cmd_pipe) string.append('dbg_pipe: %s' % self.dbg_pipe) string.append('oobm_queue: %s' % self.oobm_queue) string.append('input_buffer: %s' % self.input_buffer) string.append('input_buffer_pos: %d' % self.input_buffer_pos) string.append('esc_state: %d' % self.esc_state) string.append('line_limit: %d' % self.line_limit) string.append('history: [\'' + '\', \''.join(self.history) + '\']') string.append('history_pos: %d' % self.history_pos) string.append('prompt: \'%s\'' % self.prompt) string.append('partial_cmd: \'%s\''% self.partial_cmd) return '\n'.join(string) def PrintHistory(self): """Print the history of entered commands.""" fd = self.master_pty # Make it pretty by figuring out how wide to pad the numbers. wide = (len(self.history) / 10) + 1 for i in range(len(self.history)): line = ' %*d %s\r\n' % (wide, i, self.history[i]) os.write(fd, line) def ShowPreviousCommand(self): """Shows the previous command from the history list.""" # There's nothing to do if there's no history at all. if not self.history: self.logger.debug('No history to print.') return # Don't do anything if there's no more history to show. if self.history_pos == 0: self.logger.debug('No more history to show.') return self.logger.debug('current history position: %d.', self.history_pos) # Decrement the history buffer position. self.history_pos -= 1 self.logger.debug('new history position.: %d', self.history_pos) # Save the text entered on the console if any. if self.history_pos == len(self.history)-1: self.logger.debug('saving partial_cmd: \'%s\'', self.input_buffer) self.partial_cmd = self.input_buffer # Backspace the line. for _ in range(self.input_buffer_pos): self.SendBackspace() # Print the last entry in the history buffer. self.logger.debug('printing previous entry %d - %s', self.history_pos, self.history[self.history_pos]) fd = self.master_pty prev_cmd = self.history[self.history_pos] os.write(fd, prev_cmd) # Update the input buffer. self.input_buffer = prev_cmd self.input_buffer_pos = len(prev_cmd) def ShowNextCommand(self): """Shows the next command from the history list.""" # Don't do anything if there's no history at all. if not self.history: self.logger.debug('History buffer is empty.') return fd = self.master_pty self.logger.debug('current history position: %d', self.history_pos) # Increment the history position. self.history_pos += 1 # Restore the partial cmd. if self.history_pos == len(self.history): self.logger.debug('Restoring partial command of \'%s\'', self.partial_cmd) # Backspace the line. for _ in range(self.input_buffer_pos): self.SendBackspace() # Print the partially entered command if any. os.write(fd, self.partial_cmd) self.input_buffer = self.partial_cmd self.input_buffer_pos = len(self.input_buffer) # Now that we've printed it, clear the partial cmd storage. self.partial_cmd = '' # Reset history position. self.history_pos = len(self.history) return self.logger.debug('new history position: %d', self.history_pos) if self.history_pos > len(self.history)-1: self.logger.debug('No more history to show.') self.history_pos -= 1 self.logger.debug('Reset history position to %d', self.history_pos) return # Backspace the line. for _ in range(self.input_buffer_pos): self.SendBackspace() # Print the newer entry from the history buffer. self.logger.debug('printing next entry %d - %s', self.history_pos, self.history[self.history_pos]) next_cmd = self.history[self.history_pos] os.write(fd, next_cmd) # Update the input buffer. self.input_buffer = next_cmd self.input_buffer_pos = len(next_cmd) self.logger.debug('new history position: %d.', self.history_pos) def SliceOutChar(self): """Remove a char from the line and shift everything over 1 column.""" fd = self.master_pty # Remove the character at the input_buffer_pos by slicing it out. self.input_buffer = self.input_buffer[0:self.input_buffer_pos] + \ self.input_buffer[self.input_buffer_pos+1:] # Write the rest of the line moved_col = os.write(fd, self.input_buffer[self.input_buffer_pos:]) # Write a space to clear out the last char moved_col += os.write(fd, ' ') # Update the input buffer position. self.input_buffer_pos += moved_col # Reset the cursor self.MoveCursor('left', moved_col) def HandleEsc(self, byte): """HandleEsc processes escape sequences. Args: byte: An integer representing the current byte in the sequence. """ # We shouldn't be handling an escape sequence if we haven't seen one. assert self.esc_state != 0 if self.esc_state is EscState.ESC_START: self.logger.debug('ESC_START') if byte == ord('['): self.esc_state = EscState.ESC_BRACKET return else: self.logger.error('Unexpected sequence. %c', byte) self.esc_state = 0 elif self.esc_state is EscState.ESC_BRACKET: self.logger.debug('ESC_BRACKET') # Left Arrow key was pressed. if byte == ord('D'): self.logger.debug('Left arrow key pressed.') self.MoveCursor('left', 1) self.esc_state = 0 # Reset the state. return # Right Arrow key. elif byte == ord('C'): self.logger.debug('Right arrow key pressed.') self.MoveCursor('right', 1) self.esc_state = 0 # Reset the state. return # Up Arrow key. elif byte == ord('A'): self.logger.debug('Up arrow key pressed.') self.ShowPreviousCommand() # Reset the state. self.esc_state = 0 # Reset the state. return # Down Arrow key. elif byte == ord('B'): self.logger.debug('Down arrow key pressed.') self.ShowNextCommand() # Reset the state. self.esc_state = 0 # Reset the state. return # For some reason, minicom sends a 1 instead of 7. /shrug # TODO(aaboagye): Figure out why this happens. elif byte == ord('1') or byte == ord('7'): self.esc_state = EscState.ESC_BRACKET_1 elif byte == ord('3'): self.esc_state = EscState.ESC_BRACKET_3 elif byte == ord('8'): self.esc_state = EscState.ESC_BRACKET_8 else: self.logger.error(r'Bad or unhandled escape sequence. got ^[%c\(%d)', chr(byte), byte) self.esc_state = 0 return elif self.esc_state is EscState.ESC_BRACKET_1: self.logger.debug('ESC_BRACKET_1') # HOME key. if byte == ord('~'): self.logger.debug('Home key pressed.') self.MoveCursor('left', self.input_buffer_pos) self.esc_state = 0 # Reset the state. self.logger.debug('ESC sequence complete.') return elif self.esc_state is EscState.ESC_BRACKET_3: self.logger.debug('ESC_BRACKET_3') # DEL key. if byte == ord('~'): self.logger.debug('Delete key pressed.') if self.input_buffer_pos != len(self.input_buffer): self.SliceOutChar() self.esc_state = 0 # Reset the state. elif self.esc_state is EscState.ESC_BRACKET_8: self.logger.debug('ESC_BRACKET_8') # END key. if byte == ord('~'): self.logger.debug('End key pressed.') self.MoveCursor('right', len(self.input_buffer) - self.input_buffer_pos) self.esc_state = 0 # Reset the state. self.logger.debug('ESC sequence complete.') return else: self.logger.error('Unexpected sequence. %c', byte) self.esc_state = 0 else: self.logger.error('Unexpected sequence. %c', byte) self.esc_state = 0 def ProcessInput(self): """Captures the input determines what actions to take.""" # There's nothing to do if the input buffer is empty. if len(self.input_buffer) == 0: return # Don't store 2 consecutive identical commands in the history. if (self.history and self.history[-1] != self.input_buffer or not self.history): self.history.append(self.input_buffer) # Split the command up by spaces. line = self.input_buffer.split(' ') self.logger.debug('cmd: %s', self.input_buffer) cmd = line[0].lower() # The 'history' command is a special case that we handle locally. if cmd == 'history': self.PrintHistory() return # Send the command to the interpreter. self.logger.debug('Sending command to interpreter.') self.cmd_pipe.send(self.input_buffer) def CheckForEnhancedECImage(self): """Performs an interrogation of the EC image. Send a SYN and expect an ACK. If no ACK or the response is incorrect, then assume that the current EC image that we are talking to is not enhanced. Returns: is_enhanced: A boolean indicating whether the EC responded to the interrogation correctly. """ # Send interrogation byte and wait for the response. self.logger.debug('Performing interrogation.') self.cmd_pipe.send(interpreter.EC_SYN) response = '' if self.dbg_pipe.poll(self.interrogation_timeout): response = self.dbg_pipe.recv() self.logger.debug('response: \'%s\'', binascii.hexlify(response)) else: self.logger.debug('Timed out waiting for EC_ACK') # Verify the acknowledgment. is_enhanced = response == interpreter.EC_ACK if is_enhanced: # Increase the interrogation timeout for stability purposes. self.interrogation_timeout = ENHANCED_EC_INTERROGATION_TIMEOUT self.logger.debug('Increasing interrogation timeout to %rs.', self.interrogation_timeout) else: # Reduce the timeout in order to reduce the perceivable delay. self.interrogation_timeout = NON_ENHANCED_EC_INTERROGATION_TIMEOUT self.logger.debug('Reducing interrogation timeout to %rs.', self.interrogation_timeout) return is_enhanced def HandleChar(self, byte): """HandleChar does a certain action when it receives a character. Args: byte: An integer representing the character received from the user. """ # Interrogate the EC every time we press the enter key. This is necessary # so that we know whether or not we should provide the console interface or # simply behave as a pass-thru. if byte == ControlKey.CARRIAGE_RETURN: self.enhanced_ec = self.CheckForEnhancedECImage() self.logger.debug('Enhanced EC image? %r', self.enhanced_ec) if not self.enhanced_ec: # Send everything straight to the EC to handle. self.cmd_pipe.send(chr(byte)) # Reset the input buffer. self.input_buffer = '' self.input_buffer_pos = 0 self.logger.debug('Reset input buffer.') return # Keep handling the ESC sequence if we're in the middle of it. if self.esc_state != 0: self.HandleEsc(byte) return # When we're at the end of the line, we should only allow going backwards, # backspace, carriage return, up, or down. The arrow keys are escape # sequences, so we let the escape...escape. if (self.input_buffer_pos >= self.line_limit and byte not in [ControlKey.CTRL_B, ControlKey.ESC, ControlKey.BACKSPACE, ControlKey.CTRL_A, ControlKey.CARRIAGE_RETURN, ControlKey.CTRL_P, ControlKey.CTRL_N]): return # If the input buffer is full we can't accept new chars. buffer_full = len(self.input_buffer) >= self.line_limit fd = self.master_pty # Carriage_Return/Enter if byte == ControlKey.CARRIAGE_RETURN: self.logger.debug('Enter key pressed.') # Put a carriage return/newline and the print the prompt. os.write(fd, '\r\n') # TODO(aaboagye): When we control the printing of all output, print the # prompt AFTER printing all the output. We can't do it yet because we # don't know how much is coming from the EC. # Print the prompt. os.write(fd, self.prompt) # Process the input. self.ProcessInput() # Now, clear the buffer. self.input_buffer = '' self.input_buffer_pos = 0 # Reset history buffer pos. self.history_pos = len(self.history) # Clear partial command. self.partial_cmd = '' # Backspace elif byte == ControlKey.BACKSPACE: self.logger.debug('Backspace pressed.') if self.input_buffer_pos > 0: # Move left 1 column. self.MoveCursor('left', 1) # Remove the character at the input_buffer_pos by slicing it out. self.SliceOutChar() self.logger.debug('input_buffer_pos: %d', self.input_buffer_pos) # Ctrl+A. Move cursor to beginning of the line elif byte == ControlKey.CTRL_A: self.logger.debug('Control+A pressed.') self.MoveCursor('left', self.input_buffer_pos) # Ctrl+B. Move cursor left 1 column. elif byte == ControlKey.CTRL_B: self.logger.debug('Control+B pressed.') self.MoveCursor('left', 1) # Ctrl+D. Delete a character. elif byte == ControlKey.CTRL_D: self.logger.debug('Control+D pressed.') if self.input_buffer_pos != len(self.input_buffer): # Remove the character by slicing it out. self.SliceOutChar() # Ctrl+E. Move cursor to end of the line. elif byte == ControlKey.CTRL_E: self.logger.debug('Control+E pressed.') self.MoveCursor('right', len(self.input_buffer) - self.input_buffer_pos) # Ctrl+F. Move cursor right 1 column. elif byte == ControlKey.CTRL_F: self.logger.debug('Control+F pressed.') self.MoveCursor('right', 1) # Ctrl+K. Kill line. elif byte == ControlKey.CTRL_K: self.logger.debug('Control+K pressed.') self.KillLine() # Ctrl+N. Next line. elif byte == ControlKey.CTRL_N: self.logger.debug('Control+N pressed.') self.ShowNextCommand() # Ctrl+P. Previous line. elif byte == ControlKey.CTRL_P: self.logger.debug('Control+P pressed.') self.ShowPreviousCommand() # ESC sequence elif byte == ControlKey.ESC: # Starting an ESC sequence self.esc_state = EscState.ESC_START # Only print printable chars. elif IsPrintable(byte): # Drop the character if we're full. if buffer_full: self.logger.debug('Dropped char: %c(%d)', byte, byte) return # Print the character. os.write(fd, chr(byte)) # Print the rest of the line (if any). extra_bytes_written = os.write(fd, self.input_buffer[self.input_buffer_pos:]) # Recreate the input buffer. self.input_buffer = (self.input_buffer[0:self.input_buffer_pos] + ('%c' % byte) + self.input_buffer[self.input_buffer_pos:]) # Update the input buffer position. self.input_buffer_pos += 1 + extra_bytes_written # Reset the cursor if we wrote any extra bytes. if extra_bytes_written: self.MoveCursor('left', extra_bytes_written) self.logger.debug('input_buffer_pos: %d', self.input_buffer_pos) def MoveCursor(self, direction, count): """MoveCursor moves the cursor left or right by count columns. Args: direction: A string that should be either 'left' or 'right' representing the direction to move the cursor on the console. count: An integer representing how many columns the cursor should be moved. Raises: AssertionError: If the direction is not equal to 'left' or 'right'. """ # If there's nothing to move, we're done. if not count: return fd = self.master_pty seq = '\033[' + str(count) if direction == 'left': # Bind the movement. if count > self.input_buffer_pos: count = self.input_buffer_pos seq += 'D' self.logger.debug('move cursor left %d', count) self.input_buffer_pos -= count elif direction == 'right': # Bind the movement. if (count + self.input_buffer_pos) > len(self.input_buffer): count = 0 seq += 'C' self.logger.debug('move cursor right %d', count) self.input_buffer_pos += count else: raise AssertionError(('The only valid directions are \'left\' and ' '\'right\'')) self.logger.debug('input_buffer_pos: %d', self.input_buffer_pos) # Move the cursor. if count != 0: os.write(fd, seq) def KillLine(self): """Kill the rest of the line based on the input buffer position.""" # Killing the line is killing all the text to the right. diff = len(self.input_buffer) - self.input_buffer_pos self.logger.debug('diff: %d', diff) # Diff shouldn't be negative, but if it is for some reason, let's try to # correct the cursor. if diff < 0: self.logger.warning('Resetting input buffer position to %d...', len(self.input_buffer)) self.MoveCursor('left', -diff) return if diff: self.MoveCursor('right', diff) for _ in range(diff): self.SendBackspace() self.input_buffer_pos -= diff self.input_buffer = self.input_buffer[0:self.input_buffer_pos] def SendBackspace(self): """Backspace a character on the console.""" os.write(self.master_pty, '\033[1D \033[1D') def IsPrintable(byte): """Determines if a byte is printable. Args: byte: An integer potentially representing a printable character. Returns: A boolean indicating whether the byte is a printable character. """ return byte >= ord(' ') and byte <= ord('~') def StartLoop(console): """Starts the infinite loop of console processing. Args: console: A Console object that has been properly initialzed. """ console.logger.info('EC Console is being served on %s.', console.user_pty) console.logger.debug(console) try: while True: # Check to see if pipes or the console are ready for reading. read_list = [console.master_pty, console.cmd_pipe, console.dbg_pipe] ready_for_reading = select.select(read_list, [], [])[0] for obj in ready_for_reading: if obj is console.master_pty: console.logger.debug('Input from user') # Convert to bytes so we can look for non-printable chars such as # Ctrl+A, Ctrl+E, etc. line = bytearray(os.read(console.master_pty, CONSOLE_MAX_READ)) for i in line: # Handle each character as it arrives. console.HandleChar(i) elif obj is console.cmd_pipe: data = console.cmd_pipe.recv() # Write it to the user console. console.logger.debug('|CMD|->\'%s\'', data) os.write(console.master_pty, data) elif obj is console.dbg_pipe: data = console.dbg_pipe.recv() # Write it to the user console. console.logger.debug('|DBG|->\'%s\'', data) os.write(console.master_pty, data) while not console.oobm_queue.empty(): console.logger.debug('OOBM queue ready for reading.') cmd = console.oobm_queue.get() console.logger.debug('cmd: %s', cmd) if cmd.startswith('loglevel'): console.logger.debug('Log level change request.') new_log_level = int(cmd.split(' ')[1]) console.logger.logger.setLevel(new_log_level) console.logger.info('Log level changed to %d.', new_log_level) # Forward the request to the interpreter as well. console.cmd_pipe.send(cmd) finally: # Close pipes. console.dbg_pipe.close() console.cmd_pipe.close() # Close file descriptor. os.close(console.master_pty) # Exit. sys.exit(0) def main(argv): """Kicks off the EC-3PO interactive console interface and interpreter. We create some pipes to communicate with an interpreter, instantiate an interpreter, create a PTY pair, and begin serving the console interface. Args: argv: A list of strings containing the arguments this module was called with. """ # Set up argument parser. parser = argparse.ArgumentParser(description=('Start interactive EC console ' 'and interpreter.')) parser.add_argument('ec_uart_pty', help=('The full PTY name that the EC UART' ' is present on. eg: /dev/pts/12')) parser.add_argument('--log-level', default='info', help='info, debug, warning, error, or critical') # Parse arguments. opts = parser.parse_args(argv) # Set logging level. opts.log_level = opts.log_level.lower() if opts.log_level == 'info': log_level = logging.INFO elif opts.log_level == 'debug': log_level = logging.DEBUG elif opts.log_level == 'warning': log_level = logging.WARNING elif opts.log_level == 'error': log_level = logging.ERROR elif opts.log_level == 'critical': log_level = logging.CRITICAL else: parser.error('Invalid log level. (info, debug, warning, error, critical)') # Start logging with a timestamp, module, and log level shown in each log # entry. logging.basicConfig(level=log_level, format=('%(asctime)s - %(module)s -' ' %(levelname)s - %(message)s')) # Create some pipes to communicate between the interpreter and the console. # The command pipe is bidirectional. cmd_pipe_interactive, cmd_pipe_interp = multiprocessing.Pipe() # The debug pipe is unidirectional from interpreter to console only. dbg_pipe_interactive, dbg_pipe_interp = multiprocessing.Pipe(duplex=False) # Create an interpreter instance. itpr = interpreter.Interpreter(opts.ec_uart_pty, cmd_pipe_interp, dbg_pipe_interp, log_level) # Spawn an interpreter process. itpr_process = multiprocessing.Process(target=interpreter.StartLoop, args=(itpr,)) # Make sure to kill the interpreter when we terminate. itpr_process.daemon = True # Start the interpreter. itpr_process.start() # Open a new pseudo-terminal pair (master_pty, user_pty) = pty.openpty() # Set the permissions to 660. os.chmod(os.ttyname(user_pty), (stat.S_IRGRP | stat.S_IWGRP | stat.S_IRUSR | stat.S_IWUSR)) # Create a console. console = Console(master_pty, os.ttyname(user_pty), cmd_pipe_interactive, dbg_pipe_interactive) # Start serving the console. StartLoop(console) if __name__ == '__main__': main(sys.argv[1:])