Home | History | Annotate | Download | only in cros 
      1 # Copyright 2015 The Chromium OS Authors. All rights reserved.
      2 # Use of this source code is governed by a BSD-style license that can be
      3 # found in the LICENSE file.
      4 
      5 import collections
      6 import dbus
      7 import logging
      8 import pipes
      9 import re
     10 import shlex
     11 
     12 from autotest_lib.client.bin import utils
     13 from autotest_lib.client.common_lib import error
     14 
     15 
     16 # Represents the result of a dbus-send call.  |sender| refers to the temporary
     17 # bus name of dbus-send, |responder| to the remote process, and |response|
     18 # contains the parsed response.
     19 DBusSendResult = collections.namedtuple('DBusSendResult', ['sender',
     20                                                            'responder',
     21                                                            'response'])
     22 # Used internally.
     23 DictEntry = collections.namedtuple('DictEntry', ['key', 'value'])
     24 
     25 
     26 def _build_token_stream(headerless_dbus_send_output):
     27     """A tokenizer for dbus-send output.
     28 
     29     The output is basically just like splitting on whitespace, except that
     30     strings are kept together by " characters.
     31 
     32     @param headerless_dbus_send_output: list of lines of dbus-send output
     33             without the meta-information prefix.
     34     @return list of tokens in dbus-send output.
     35     """
     36     return shlex.split(' '.join(headerless_dbus_send_output))
     37 
     38 
     39 def _parse_value(token_stream):
     40     """Turn a stream of tokens from dbus-send output into native python types.
     41 
     42     @param token_stream: output from _build_token_stream() above.
     43 
     44     """
     45     if len(token_stream) == 0:
     46       # Return None for dbus-send output with no return values.
     47       return None
     48     # Assumes properly tokenized output (strings with spaces handled).
     49     # Assumes tokens are pre-stripped
     50     token_type = token_stream.pop(0)
     51     if token_type == 'variant':
     52         token_type = token_stream.pop(0)
     53     if token_type == 'object':
     54         token_type = token_stream.pop(0)  # Should be 'path'
     55     token_value = token_stream.pop(0)
     56     INT_TYPES = ('int16', 'uint16', 'int32', 'uint32',
     57                  'int64', 'uint64', 'byte')
     58     if token_type in INT_TYPES:
     59         return int(token_value)
     60     if token_type == 'string' or token_type == 'path':
     61         return token_value  # shlex removed surrounding " chars.
     62     if token_type == 'boolean':
     63         return token_value == 'true'
     64     if token_type == 'double':
     65         return float(token_value)
     66     if token_type == 'array':
     67         values = []
     68         while token_stream[0] != ']':
     69             values.append(_parse_value(token_stream))
     70         token_stream.pop(0)
     71         if values and all([isinstance(x, DictEntry) for x in values]):
     72             values = dict(values)
     73         return values
     74     if token_type == 'dict':
     75         assert token_value == 'entry('
     76         key = _parse_value(token_stream)
     77         value = _parse_value(token_stream)
     78         assert token_stream.pop(0) == ')'
     79         return DictEntry(key=key, value=value)
     80     raise error.TestError('Unhandled DBus type found: %s' % token_type)
     81 
     82 
     83 def _parse_dbus_send_output(dbus_send_stdout):
     84     """Turn dbus-send output into usable Python types.
     85 
     86     This looks like:
     87 
     88     localhost ~ # dbus-send --system --dest=org.chromium.flimflam \
     89             --print-reply --reply-timeout=2000 / \
     90             org.chromium.flimflam.Manager.GetProperties
     91     method return time=1490931987.170070 sender=org.chromium.flimflam -> \
     92         destination=:1.37 serial=6 reply_serial=2
     93        array [
     94           dict entry(
     95              string "ActiveProfile"
     96              variant             string "/profile/default"
     97           )
     98           dict entry(
     99              string "ArpGateway"
    100              variant             boolean true
    101           )
    102           ...
    103        ]
    104 
    105     @param dbus_send_output: string stdout from dbus-send
    106     @return a DBusSendResult.
    107 
    108     """
    109     lines = dbus_send_stdout.strip().splitlines()
    110     # The first line contains meta-information about the response
    111     header = lines[0]
    112     lines = lines[1:]
    113     dbus_address_pattern = r'[:\d\\.]+|[a-zA-Z.]+'
    114     # The header may or may not have a time= field.
    115     match = re.match(r'method return (time=[\d\\.]+ )?sender=(%s) -> '
    116                      r'destination=(%s) serial=\d+ reply_serial=\d+' %
    117                      (dbus_address_pattern, dbus_address_pattern), header)
    118 
    119     if match is None:
    120         raise error.TestError('Could not parse dbus-send header: %s' % header)
    121 
    122     sender = match.group(2)
    123     responder = match.group(3)
    124     token_stream = _build_token_stream(lines)
    125     ret_val = _parse_value(token_stream)
    126     # Note that DBus permits multiple response values, and this is not handled.
    127     logging.debug('Got DBus response: %r', ret_val)
    128     return DBusSendResult(sender=sender, responder=responder, response=ret_val)
    129 
    130 
    131 def _dbus2string(raw_arg):
    132     """Turn a dbus.* type object into a string that dbus-send expects.
    133 
    134     @param raw_dbus dbus.* type object to stringify.
    135     @return string suitable for dbus-send.
    136 
    137     """
    138     int_map = {
    139             dbus.Int16: 'int16:',
    140             dbus.Int32: 'int32:',
    141             dbus.Int64: 'int64:',
    142             dbus.UInt16: 'uint16:',
    143             dbus.UInt32: 'uint32:',
    144             dbus.UInt64: 'uint64:',
    145             dbus.Double: 'double:',
    146             dbus.Byte: 'byte:',
    147     }
    148 
    149     if isinstance(raw_arg, dbus.String):
    150         return pipes.quote('string:%s' % raw_arg.replace('"', r'\"'))
    151 
    152     if isinstance(raw_arg, dbus.Boolean):
    153         if raw_arg:
    154             return 'boolean:true'
    155         else:
    156             return 'boolean:false'
    157 
    158     for prim_type, prefix in int_map.iteritems():
    159         if isinstance(raw_arg, prim_type):
    160             return prefix + str(raw_arg)
    161 
    162     raise error.TestError('No support for serializing %r' % raw_arg)
    163 
    164 
    165 def _build_arg_string(raw_args):
    166     """Construct a string of arguments to a DBus method as dbus-send expects.
    167 
    168     @param raw_args list of dbus.* type objects to seriallize.
    169     @return string suitable for dbus-send.
    170 
    171     """
    172     return ' '.join([_dbus2string(arg) for arg in raw_args])
    173 
    174 
    175 def dbus_send(bus_name, interface, object_path, method_name, args=None,
    176               host=None, timeout_seconds=2, tolerate_failures=False, user=None):
    177     """Call dbus-send without arguments.
    178 
    179     @param bus_name: string identifier of DBus connection to send a message to.
    180     @param interface: string DBus interface of object to call method on.
    181     @param object_path: string DBus path of remote object to call method on.
    182     @param method_name: string name of method to call.
    183     @param args: optional list of arguments.  Arguments must be of types
    184             from the python dbus module.
    185     @param host: An optional host object if running against a remote host.
    186     @param timeout_seconds: number of seconds to wait for a response.
    187     @param tolerate_failures: boolean True to ignore problems receiving a
    188             response.
    189     @param user: An option argument to run dbus-send as a given user.
    190 
    191     """
    192     run = utils.run if host is None else host.run
    193     cmd = ('dbus-send --system --print-reply --reply-timeout=%d --dest=%s '
    194            '%s %s.%s' % (int(timeout_seconds * 1000), bus_name,
    195                          object_path, interface, method_name))
    196 
    197     if user is not None:
    198         cmd = ('sudo -u %s %s' % (user, cmd))
    199     if args is not None:
    200         cmd = cmd + ' ' + _build_arg_string(args)
    201     result = run(cmd, ignore_status=tolerate_failures)
    202     if result.exit_status != 0:
    203         logging.debug('%r', result.stdout)
    204         return None
    205     return _parse_dbus_send_output(result.stdout)
    206 
    207 
    208 def get_property(bus_name, interface, object_path, property_name, host=None):
    209     """A helpful wrapper that extracts the value of a DBus property.
    210 
    211     @param bus_name: string identifier of DBus connection to send a message to.
    212     @param interface: string DBus interface exposing the property.
    213     @param object_path: string DBus path of remote object to call method on.
    214     @param property_name: string name of property to get.
    215     @param host: An optional host object if running against a remote host.
    216 
    217     """
    218     return dbus_send(bus_name, dbus.PROPERTIES_IFACE, object_path, 'Get',
    219                      args=[dbus.String(interface), dbus.String(property_name)],
    220                      host=host)
    221