# Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # http://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. """Module for processing CLI args.""" import os import logging from awscli.compat import six from botocore import xform_name from botocore.compat import OrderedDict, json from awscli import utils from awscli import SCALAR_TYPES, COMPLEX_TYPES from awscli.paramfile import get_paramfile, ResourceLoadingError from awscli.paramfile import PARAMFILE_DISABLED LOG = logging.getLogger('awscli.argprocess') class ParamError(Exception): def __init__(self, cli_name, message): """ :type cli_name: string :param cli_name: The complete cli argument name, e.g. "--foo-bar". It should include the leading hyphens if that's how a user would specify the name. :type message: string :param message: The error message to display to the user. """ full_message = ("Error parsing parameter '%s': %s" % (cli_name, message)) super(ParamError, self).__init__(full_message) self.cli_name = cli_name self.message = message class ParamSyntaxError(Exception): pass class ParamUnknownKeyError(Exception): def __init__(self, key, valid_keys): valid_keys = ', '.join(valid_keys) full_message = ( "Unknown key '%s', valid choices " "are: %s" % (key, valid_keys)) super(ParamUnknownKeyError, self).__init__(full_message) def unpack_argument(session, service_name, operation_name, cli_argument, value): """ Unpack an argument's value from the commandline. This is part one of a two step process in handling commandline arguments. Emits the load-cli-arg event with service, operation, and parameter names. Example:: load-cli-arg.ec2.describe-instances.foo """ param_name = getattr(cli_argument, 'name', 'anonymous') value_override = session.emit_first_non_none_response( 'load-cli-arg.%s.%s.%s' % (service_name, operation_name, param_name), param=cli_argument, value=value, service_name=service_name, operation_name=operation_name) if value_override is not None: value = value_override return value def uri_param(event_name, param, value, **kwargs): """Handler that supports param values from URIs. """ cli_argument = param qualified_param_name = '.'.join(event_name.split('.')[1:]) if qualified_param_name in PARAMFILE_DISABLED or \ getattr(cli_argument, 'no_paramfile', None): return else: return _check_for_uri_param(cli_argument, value) def _check_for_uri_param(param, value): if isinstance(value, list) and len(value) == 1: value = value[0] try: return get_paramfile(value) except ResourceLoadingError as e: raise ParamError(param.cli_name, six.text_type(e)) def detect_shape_structure(param): stack = [] return _detect_shape_structure(param, stack) def _detect_shape_structure(param, stack): if param.name in stack: return 'recursive' else: stack.append(param.name) try: if param.type_name in SCALAR_TYPES: return 'scalar' elif param.type_name == 'structure': sub_types = [_detect_shape_structure(p, stack) for p in param.members.values()] # We're distinguishing between structure(scalar) # and structure(scalars), because for the case of # a single scalar in a structure we can simplify # more than a structure(scalars). if len(sub_types) == 1 and all(p == 'scalar' for p in sub_types): return 'structure(scalar)' elif len(sub_types) > 1 and all(p == 'scalar' for p in sub_types): return 'structure(scalars)' else: return 'structure(%s)' % ', '.join(sorted(set(sub_types))) elif param.type_name == 'list': return 'list-%s' % _detect_shape_structure(param.member, stack) elif param.type_name == 'map': if param.value.type_name in SCALAR_TYPES: return 'map-scalar' else: return 'map-%s' % _detect_shape_structure(param.value, stack) finally: stack.pop() def unpack_cli_arg(cli_argument, value): """ Parses and unpacks the encoded string command line parameter and returns native Python data structures that can be passed to the Operation. :type cli_argument: :class:`awscli.arguments.BaseCLIArgument` :param cli_argument: The CLI argument object. :param value: The value of the parameter. This can be a number of different python types (str, list, etc). This is the value as it's specified on the command line. :return: The "unpacked" argument than can be sent to the `Operation` object in python. """ return _unpack_cli_arg(cli_argument.argument_model, value, cli_argument.cli_name) def _unpack_cli_arg(argument_model, value, cli_name): if argument_model.type_name in SCALAR_TYPES: return unpack_scalar_cli_arg( argument_model, value, cli_name) elif argument_model.type_name in COMPLEX_TYPES: return _unpack_complex_cli_arg( argument_model, value, cli_name) else: return six.text_type(value) def _unpack_complex_cli_arg(argument_model, value, cli_name): type_name = argument_model.type_name if type_name == 'structure' or type_name == 'map': if value.lstrip()[0] == '{': try: return json.loads(value, object_pairs_hook=OrderedDict) except ValueError as e: raise ParamError( cli_name, "Invalid JSON: %s\nJSON received: %s" % (e, value)) raise ParamError(cli_name, "Invalid JSON:\n%s" % value) elif type_name == 'list': if isinstance(value, six.string_types): if value.lstrip()[0] == '[': return json.loads(value, object_pairs_hook=OrderedDict) elif isinstance(value, list) and len(value) == 1: single_value = value[0].strip() if single_value and single_value[0] == '[': return json.loads(value[0], object_pairs_hook=OrderedDict) try: # There's a couple of cases remaining here. # 1. It's possible that this is just a list of strings, i.e # --security-group-ids sg-1 sg-2 sg-3 => ['sg-1', 'sg-2', 'sg-3'] # 2. It's possible this is a list of json objects: # --filters '{"Name": ..}' '{"Name": ...}' member_shape_model = argument_model.member return [_unpack_cli_arg(member_shape_model, v, cli_name) for v in value] except (ValueError, TypeError) as e: # The list params don't have a name/cli_name attached to them # so they will have bad error messages. We're going to # attach the parent parameter to this error message to provide # a more helpful error message. raise ParamError(cli_name, value[0]) def unpack_scalar_cli_arg(argument_model, value, cli_name=''): # Note the cli_name is used strictly for error reporting. It's # not required to use unpack_scalar_cli_arg if argument_model.type_name == 'integer' or argument_model.type_name == 'long': return int(value) elif argument_model.type_name == 'float' or argument_model.type_name == 'double': # TODO: losing precision on double types return float(value) elif argument_model.type_name == 'blob' and \ argument_model.serialization.get('streaming'): file_path = os.path.expandvars(value) file_path = os.path.expanduser(file_path) if not os.path.isfile(file_path): msg = 'Blob values must be a path to a file.' raise ParamError(cli_name, msg) return open(file_path, 'rb') elif argument_model.type_name == 'boolean': if isinstance(value, six.string_types) and value.lower() == 'false': return False return bool(value) else: return value class ParamShorthand(object): # To add support for a new shape: # # * Add it to SHORTHAND_SHAPES below, key is the shape structure # value is the name of the method to call. # * Implement parse method. # * Implement _doc_. This is used to generate # the docs for this shorthand syntax. SHORTHAND_SHAPES = { 'structure(scalars)': '_key_value_parse', 'structure(scalar)': '_special_key_value_parse', 'structure(list-scalar, scalar)': '_struct_scalar_list_parse', 'map-scalar': '_key_value_parse', 'list-structure(scalar)': '_list_scalar_parse', 'list-structure(scalars)': '_list_key_value_parse', 'list-structure(list-scalar, scalar)': '_list_scalar_list_parse', } def __init__(self): pass def __call__(self, cli_argument, value, **kwargs): """Attempt to parse shorthand syntax for values. This is intended to be hooked up as an event handler (hence the **kwargs). Given ``param`` object and its string ``value``, figure out if we can parse it. If we can parse it, we return the parsed value (typically some sort of python dict). :type cli_argument: :class:`awscli.arguments.BaseCLIArgument` :param cli_argument: The CLI argument object. :type param: :class:`botocore.parameters.Parameter` :param param: The parameter object (includes various metadata about the parameter). :type value: str :param value: The value for the parameter type on the command line, e.g ``--foo this_value``, value would be ``"this_value"``. :returns: If we can parse the value we return the parsed value. If it looks like JSON, we return None (which tells the event emitter to use the default ``unpack_cli_arg`` provided that no other event handlers can parsed the value). If we run into an error parsing the value, a ``ParamError`` will be raised. """ parse_method = self.get_parse_method_for_param(cli_argument, value) if parse_method is None: return else: try: LOG.debug("Using %s for param %s", parse_method, cli_argument.cli_name) parsed = getattr(self, parse_method)( cli_argument.argument_model, value) except ParamSyntaxError as e: docgen = ParamShorthandDocGen() example_usage = docgen.generate_shorthand_example(cli_argument) raise ParamError(cli_argument.cli_name, "should be: %s" % example_usage) except (ParamError, ParamUnknownKeyError) as e: # The shorthand parse methods don't have the cli_name, # so any ParamError won't have this value. To accomodate # this, ParamErrors are caught and reraised with the cli_name # injected. raise ParamError(cli_argument.cli_name, str(e)) return parsed def get_parse_method_for_param(self, cli_argument, value=None): # We first need to make sure this is a parameter that qualifies # for simplification. The first short-circuit case is if it looks # like json we immediately return. if isinstance(value, list): check_val = value[0] else: check_val = value if isinstance(check_val, six.string_types) and check_val.strip().startswith( ('[', '{')): LOG.debug("Param %s looks like JSON, not considered for " "param shorthand.", cli_argument.py_name) return structure = detect_shape_structure(cli_argument.argument_model) # If this looks like shorthand then we log the detected structure # to help with debugging why the shorthand may not work, for # example list-structure(list-structure(scalars)) LOG.debug('Detected structure: {0}'.format(structure)) parse_method = self.SHORTHAND_SHAPES.get(structure) return parse_method def _get_example_fn(self, param): doc_fn = None shape_structure = detect_shape_structure(param) method = self.SHORTHAND_SHAPES.get(shape_structure) if method: doc_fn = getattr(self, '_docs' + method, None) return doc_fn def add_example_fn(self, arg_name, help_command, **kwargs): """ Adds a callable to the ``example_fn`` attribute of the parameter if the parameter type is supported by shorthand syntax. This callable should return a string containing just the example and not any of the ReST formatting that might be required in the docs. """ argument = help_command.arg_table[arg_name] model = argument.argument_model LOG.debug('Adding example fn for: %s' % arg_name) doc_fn = self._get_example_fn(model) # XXX: fix this, don't set attributes on argument objects. argument.example_fn = doc_fn def _list_scalar_list_parse(self, param, value): # Think something like ec2.DescribeInstances.Filters. # We're looking for key=val1,val2,val3,key2=val1,val2. parsed = [] for v in value: struct = self._struct_scalar_list_parse(param.member, v) parsed.append(struct) return parsed def _struct_scalar_list_parse(self, param, value): # Create a mapping of argument name -> argument object args = {} for member_name, arg in param.members.items(): # Arg name -> arg object lookup args[member_name] = arg parts = self._split_on_commas(value) current_parsed = {} current_key = None for part in parts: current = part.split('=', 1) if len(current) == 2: # This is a key/value pair. current_key = current[0].strip() if current_key not in args: raise ParamUnknownKeyError(current_key, args.keys()) current_value = unpack_scalar_cli_arg(args[current_key], current[1].strip()) if args[current_key].type_name == 'list': current_parsed[current_key] = current_value.split(',') else: current_parsed[current_key] = current_value elif current_key is not None: # This is a value which we associate with the current_key, # so key1=val1,val2 # ^ # | # val2 is associated with key1. current_value = unpack_scalar_cli_arg(args[current_key], current[0]) current_parsed[current_key].append(current_value) else: raise ParamSyntaxError(part) return current_parsed def _list_scalar_parse(self, param, value): single_param_name = list(param.member.members.keys())[0] parsed = [] # We know that value is a list in this case. for v in value: parsed.append({single_param_name: v}) return parsed def _list_key_value_parse(self, param, value): # param is a list param. # param.member is the struct param. struct_param = param.member parsed = [] for v in value: single_struct_param = self._key_value_parse(struct_param, v) parsed.append(single_struct_param) return parsed def _special_key_value_parse(self, param, value): # This is a special key value parse that can do the normal # key=value parsing, *but* supports a few additional conveniences # when working with a structure with a single element. # Precondition: param is a shape of structure(scalar) if self._is_special_case_key_value(param, value): # We have an even shorter shorthand syntax for structure # of scalars of a single element with a member name of # 'Value'. return {'Value': value} else: return self._key_value_parse(param, value) def _is_special_case_key_value(self, param, value): members = param.members if len(param.members) == 1: if list(members.keys())[0] == 'Value' and \ '=' not in value: return True return False def _key_value_parse(self, param, value): # The expected structure is: # key=value,key2=value # that is, csv key value pairs, where the key and values # are separated by '='. All of this should be whitespace # insensitive. parsed = OrderedDict() parts = self._split_on_commas(value) valid_names = self._create_name_to_params(param) for part in parts: try: key, value = part.split('=', 1) except ValueError: raise ParamSyntaxError(part) key = key.strip() value = value.strip() if valid_names and key not in valid_names: raise ParamUnknownKeyError(key, valid_names) if valid_names: sub_param = valid_names[key] if sub_param is not None: # TODO: you are here. unpack_scalar_cli_arg takes # the cli_name, but we don't have it. What are our # options? value = unpack_scalar_cli_arg(sub_param, value) parsed[key] = value return parsed def _create_name_to_params(self, param): if param.type_name == 'structure': return dict([(member_name, p) for member_name, p in param.members.items()]) elif param.type_name == 'map' and hasattr(param.key, 'enum'): return dict([(v, None) for v in param.key.enum]) def _split_on_commas(self, value): try: return utils.split_on_commas(value) except ValueError as e: raise ParamSyntaxError(six.text_type(e)) class ParamShorthandDocGen(object): """Documentation generator for param shorthand syntax.""" SHORTHAND_SHAPES = ParamShorthand.SHORTHAND_SHAPES def supports_shorthand(self, cli_argument): """Checks if a CLI argument supports shorthand syntax.""" if cli_argument.argument_model is not None: structure = detect_shape_structure(cli_argument.argument_model) return structure in self.SHORTHAND_SHAPES return False def generate_shorthand_example(self, cli_argument): """Generate documentation for a CLI argument. :type cli_argument: awscli.arguments.BaseCLIArgument :param cli_argument: The CLI argument which to generate documentation for. """ structure = detect_shape_structure(cli_argument.argument_model) parse_method_name = self.SHORTHAND_SHAPES.get(structure) doc_method_name = '_docs%s' % parse_method_name method = getattr(self, doc_method_name) doc_string = method(cli_argument) return doc_string def _docs_list_scalar_parse(self, cli_argument): cli_name = cli_argument.cli_name structure_members = cli_argument.argument_model.member.members # We know based on the SHORTHAND_SHAPES that this is a # structure with a single member, so we can safely say: member_name = list(structure_members.keys())[0] return '%s %s1 %s2 %s3' % (cli_name, member_name, member_name, member_name) def _docs_key_value_parse(self, cli_argument): cli_name = cli_argument.cli_name model = cli_argument.argument_model s = '%s ' % cli_name if model.type_name == 'structure': members_dict = model.members member_names = list(members_dict.keys()) s += ','.join(['%s=value' % name for name in member_names]) elif model.type_name == 'map': s += 'key_name=string,key_name2=string' if self._has_enum_values(model.key): enum_values = self._get_enum_values(model.key) s += '\nWhere valid key names are:\n' for value in enum_values: s += ' %s\n' % value return s def _docs_list_key_value_parse(self, cli_argument): s = "Key value pairs, with multiple values separated by a space.\n" s += '%s ' % cli_argument.cli_name members = cli_argument.argument_model.member.members pair = ','.join(['%s=%s' % (member_name, shape.type_name) for member_name, shape in members.items()]) pair += ' %s' % pair s += pair return s def _docs_list_scalar_list_parse(self, cli_argument): s = ('Key value pairs, where values are separated by commas, ' 'and multiple pairs are separated by spaces.\n') s += '%s ' % cli_argument.cli_name pair = self._generate_struct_list_scalar_docs( cli_argument.argument_model.member.members) pair += ' %s' % pair s += pair return s def _docs_struct_scalar_list_parse(self, cli_argument): s = ('Key value pairs, where values are separated by commas.\n') s += '%s ' % cli_argument.cli_name s += self._generate_struct_list_scalar_docs( cli_argument.argument_model.members) return s def _generate_struct_list_scalar_docs(self, members_dict): scalar_params = list(self._get_scalar_params(members_dict)) list_params = list(self._get_list_params(members_dict)) pair = '' for member_name, param in scalar_params: pair += '%s=%s1,' % (member_name, param.type_name) for member_name, param in list_params[:-1]: param_type = param.member.type_name pair += '%s=%s1,%s2,' % (member_name, param_type, param_type) member_name, last_param = list_params[-1] param_type = last_param.member.type_name pair += '%s=%s1,%s2' % (member_name, param_type, param_type) return pair def _get_scalar_params(self, members_dict): for key, value in members_dict.items(): if value.type_name in SCALAR_TYPES: yield (key, value) def _get_list_params(self, members_dict): for key, value in members_dict.items(): if value.type_name == 'list': yield (key, value) def _has_enum_values(self, model): return 'enum' in model.metadata def _get_enum_values(self, model): return model.metadata['enum'] def _docs_special_key_value_parse(self, cli_argument): members = cli_argument.argument_model.members if len(members) == 1 and 'Value' in members: # Returning None will indicate that we don't have # any examples to generate, and the entire examples section # should be skipped for this arg. return None else: return self._docs_key_value_parse(cli_argument)