path: root/chromium/tools/usb_gadget/gadget.py

# Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Generic USB gadget functionality.
"""

import struct

import msos20_descriptors
import usb_constants
import usb_descriptors


class Gadget(object):
  """Basic functionality for a USB device.

  Implements standard control requests assuming that a subclass will handle
  class- or vendor-specific requests.
  """

  def __init__(self, device_desc, fs_config_desc, hs_config_desc):
    """Create a USB gadget device.

    Args:
      device_desc: USB device descriptor.
      fs_config_desc: Low/full-speed device descriptor.
      hs_config_desc: High-speed device descriptor.
    """
    self._speed = usb_constants.Speed.UNKNOWN
    self._chip = None
    self._device_desc = device_desc
    self._fs_config_desc = fs_config_desc
    self._hs_config_desc = hs_config_desc
    # dict mapping language codes to a dict mapping indexes to strings
    self._strings = {}
    self._bos_descriptor = None
    # dict mapping interface numbers to a set of endpoint addresses
    self._active_endpoints = {}
    # dict mapping endpoint addresses to interfaces
    self._endpoint_interface_map = {}
    self._ms_vendor_code_v1 = None
    self._ms_vendor_code_v2 = None
    self._ms_compat_ids = {}
    self._ms_os20_config_subset = None

  def GetDeviceDescriptor(self):
    return self._device_desc

  def GetFullSpeedConfigurationDescriptor(self):
    return self._fs_config_desc

  def GetHighSpeedConfigurationDescriptor(self):
    return self._hs_config_desc

  def GetConfigurationDescriptor(self):
    if self._speed == usb_constants.Speed.FULL:
      return self._fs_config_desc
    elif self._speed == usb_constants.Speed.HIGH:
      return self._hs_config_desc
    else:
      raise RuntimeError('Device is not connected.')

  def GetSpeed(self):
    return self._speed

  def AddStringDescriptor(self, index, value, lang=0x0409):
    """Add a string descriptor to this device.

    Args:
      index: String descriptor index (matches 'i' fields in descriptors).
      value: The string.
      lang: Language code (default: English).

    Raises:
      ValueError: The index or language code is invalid.
    """
    if index < 1 or index > 255:
      raise ValueError('String descriptor index out of range.')
    if lang < 0 or lang > 0xffff:
      raise ValueError('String descriptor language code out of range.')

    lang_strings = self._strings.setdefault(lang, {})
    lang_strings[index] = value

  def EnableMicrosoftOSDescriptorsV1(self, vendor_code=0x01):
    if vendor_code < 0 or vendor_code > 255:
      raise ValueError('Vendor code out of range.')
    if vendor_code == self._ms_vendor_code_v1:
      raise ValueError('OS Descriptor v1 vendor code conflicts with v2.')

    self._ms_vendor_code_v1 = vendor_code

  def EnableMicrosoftOSDescriptorsV2(self, vendor_code=0x02):
    if vendor_code < 0 or vendor_code > 255:
      raise ValueError('Vendor code out of range.')
    if vendor_code == self._ms_vendor_code_v1:
      raise ValueError('OS Descriptor v2 vendor code conflicts with v1.')

    self._ms_vendor_code_v2 = vendor_code
    self._ms_os20_descriptor_set = \
        msos20_descriptors.DescriptorSetHeader(dwWindowsVersion=0x06030000)
    # Gadget devices currently only support one configuration. Contrary to
    # Microsoft's documentation the bConfigurationValue field should be set to
    # the index passed to GET_DESCRIPTOR that returned the configuration instead
    # of the configuration's bConfigurationValue field. (i.e. 0 instead of 1).
    #
    # https://social.msdn.microsoft.com/Forums/windowsdesktop/en-US/ae64282c-3bc3-49af-8391-4d174479d9e7/microsoft-os-20-descriptors-not-working-on-an-interface-of-a-composite-usb-device
    self._ms_os20_config_subset = msos20_descriptors.ConfigurationSubsetHeader(
        bConfigurationValue=0)
    self._ms_os20_descriptor_set.Add(self._ms_os20_config_subset)
    self._ms_os20_platform_descriptor = \
        msos20_descriptors.PlatformCapabilityDescriptor(
            dwWindowsVersion=0x06030000,
            bMS_VendorCode=self._ms_vendor_code_v2)
    self._ms_os20_platform_descriptor.SetDescriptorSet(
        self._ms_os20_descriptor_set)
    self.AddDeviceCapabilityDescriptor(self._ms_os20_platform_descriptor)

  def SetMicrosoftCompatId(self, interface_number, compat_id, sub_compat_id=''):
    self._ms_compat_ids[interface_number] = (compat_id, sub_compat_id)
    if self._ms_os20_config_subset is not None:
      function_header = msos20_descriptors.FunctionSubsetHeader(
          bFirstInterface=interface_number)
      function_header.Add(msos20_descriptors.CompatibleId(
          CompatibleID=compat_id, SubCompatibleID=sub_compat_id))
      self._ms_os20_config_subset.Add(function_header)

  def AddDeviceCapabilityDescriptor(self, device_capability):
    """Add a device capability descriptor to this device.

    Args:
      device_capability: The Descriptor object.
    """
    if self._bos_descriptor is None:
      self._bos_descriptor = usb_descriptors.BosDescriptor()
    self._bos_descriptor.AddDeviceCapability(device_capability)

  def Connected(self, chip, speed):
    """The device has been connected to a USB host.

    Args:
      chip: USB controller.
      speed: Connection speed.
    """
    self._speed = speed
    self._chip = chip

  def Disconnected(self):
    """The device has been disconnected from the USB host."""
    self._speed = usb_constants.Speed.UNKNOWN
    self._chip = None
    self._active_endpoints.clear()
    self._endpoint_interface_map.clear()

  def IsConnected(self):
    return self._chip is not None

  def ControlRead(self, request_type, request, value, index, length):
    """Handle a read on the control pipe (endpoint zero).

    Args:
      request_type: bmRequestType field of the setup packet.
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      A buffer to return to the USB host with len <= length on success or
      None to stall the pipe.
    """
    assert request_type & usb_constants.Dir.IN
    typ = request_type & usb_constants.Type.MASK
    recipient = request_type & usb_constants.Recipient.MASK
    if typ == usb_constants.Type.STANDARD:
      return self.StandardControlRead(
          recipient, request, value, index, length)
    elif typ == usb_constants.Type.CLASS:
      return self.ClassControlRead(
          recipient, request, value, index, length)
    elif typ == usb_constants.Type.VENDOR:
      return self.VendorControlRead(
          recipient, request, value, index, length)

  def ControlWrite(self, request_type, request, value, index, data):
    """Handle a write to the control pipe (endpoint zero).

    Args:
      request_type: bmRequestType field of the setup packet.
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      data: Data stage of the request.

    Returns:
      True on success, None to stall the pipe.
    """
    assert not request_type & usb_constants.Dir.IN
    typ = request_type & usb_constants.Type.MASK
    recipient = request_type & usb_constants.Recipient.MASK
    if typ == usb_constants.Type.STANDARD:
      return self.StandardControlWrite(
          recipient, request, value, index, data)
    elif typ == usb_constants.Type.CLASS:
      return self.ClassControlWrite(
          recipient, request, value, index, data)
    elif typ == usb_constants.Type.VENDOR:
      return self.VendorControlWrite(
          recipient, request, value, index, data)

  def SendPacket(self, endpoint, data):
    """Send a data packet on the given endpoint.

    Args:
      endpoint: Endpoint address.
      data: Data buffer.

    Raises:
      ValueError: If the endpoint address is not valid.
      RuntimeError: If the device is not connected.
    """
    if self._chip is None:
      raise RuntimeError('Device is not connected.')
    if not endpoint & usb_constants.Dir.IN:
      raise ValueError('Cannot write to non-input endpoint.')
    self._chip.SendPacket(endpoint, data)

  def ReceivePacket(self, endpoint, data):
    """Handle an incoming data packet on one of the device's active endpoints.

    This method should be overridden by a subclass implementing endpoint-based
    data transfers.

    Args:
      endpoint: Endpoint address.
      data: Data buffer.
    """
    pass

  def HaltEndpoint(self, endpoint):
    """Signals a STALL condition to the host on the given endpoint.

    Args:
      endpoint: Endpoint address.
    """
    self._chip.HaltEndpoint(endpoint)

  def StandardControlRead(self, recipient, request, value, index, length):
    """Handle standard control transfers.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      A buffer to return to the USB host with len <= length on success or
      None to stall the pipe.
    """
    if recipient == usb_constants.Recipient.DEVICE:
      if request == usb_constants.Request.GET_DESCRIPTOR:
        desc_type = value >> 8
        desc_index = value & 0xff
        desc_lang = index

        print 'GetDescriptor(recipient={}, type={}, index={}, lang={})'.format(
            recipient, desc_type, desc_index, desc_lang)

        return self.GetDescriptor(recipient, desc_type, desc_index, desc_lang,
                                  length)

  def GetDescriptor(self, recipient, typ, index, lang, length):
    """Handle a standard GET_DESCRIPTOR request.

    See Universal Serial Bus Specification Revision 2.0 section 9.4.3.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      typ: Descriptor type.
      index: Descriptor index.
      lang: Descriptor language code.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      The value of the descriptor or None to stall the pipe.
    """
    if typ == usb_constants.DescriptorType.STRING:
      return self.GetStringDescriptor(index, lang, length)
    elif typ == usb_constants.DescriptorType.BOS:
      return self.GetBosDescriptor(length)

  def ClassControlRead(self, recipient, request, value, index, length):
    """Handle class-specific control transfers.

    This function should be overridden by a subclass implementing a particular
    device class.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      A buffer to return to the USB host with len <= length on success or
      None to stall the pipe.
    """
    _ = recipient, request, value, index, length
    return None

  def VendorControlRead(self, recipient, request, value, index, length):
    """Handle vendor-specific control transfers.

    This function should be overridden by a subclass if implementing a device
    that responds to vendor-specific requests.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      A buffer to return to the USB host with len <= length on success or
      None to stall the pipe.
    """
    if (self._ms_vendor_code_v1 is not None and
        request == self._ms_vendor_code_v1 and
        (recipient == usb_constants.Recipient.DEVICE or
         recipient == usb_constants.Recipient.INTERFACE)):
      return self.GetMicrosoftOSDescriptorV1(recipient, value, index, length)
    if (self._ms_vendor_code_v2 is not None and
        request == self._ms_vendor_code_v2 and
        recipient == usb_constants.Recipient.DEVICE and
        value == 0x0000 and
        index == 0x0007):
      return self.GetMicrosoftOSDescriptorV2(length)

    return None

  def StandardControlWrite(self, recipient, request, value, index, data):
    """Handle standard control transfers.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      data: Data stage of the request.

    Returns:
      True on success, None to stall the pipe.
    """
    _ = data

    if request == usb_constants.Request.SET_CONFIGURATION:
      if recipient == usb_constants.Recipient.DEVICE:
        return self.SetConfiguration(value)
    elif request == usb_constants.Request.SET_INTERFACE:
      if recipient == usb_constants.Recipient.INTERFACE:
        return self.SetInterface(index, value)

  def ClassControlWrite(self, recipient, request, value, index, data):
    """Handle class-specific control transfers.

    This function should be overridden by a subclass implementing a particular
    device class.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      data: Data stage of the request.

    Returns:
      True on success, None to stall the pipe.
    """
    _ = recipient, request, value, index, data
    return None

  def VendorControlWrite(self, recipient, request, value, index, data):
    """Handle vendor-specific control transfers.

    This function should be overridden by a subclass if implementing a device
    that responds to vendor-specific requests.

    Args:
      recipient: Request recipient (device, interface, endpoint, etc.)
      request: bRequest field of the setup packet.
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      data: Data stage of the request.

    Returns:
      True on success, None to stall the pipe.
    """
    _ = recipient, request, value, index, data
    return None

  def GetStringDescriptor(self, index, lang, length):
    """Handle a GET_DESCRIPTOR(String) request from the host.

    Descriptor index 0 returns the set of languages supported by the device.
    All other indices return the string descriptors registered with those
    indices.

    See Universal Serial Bus Specification Revision 2.0 section 9.6.7.

    Args:
      index: Descriptor index.
      lang: Descriptor language code.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      The string descriptor or None to stall the pipe if the descriptor is not
      found.
    """
    if index == 0:
      length = 2 + len(self._strings) * 2
      header = struct.pack('<BB', length, usb_constants.DescriptorType.STRING)
      lang_codes = [struct.pack('<H', lang)
                    for lang in self._strings.iterkeys()]
      buf = header + ''.join(lang_codes)
      assert len(buf) == length
      return buf[:length]
    if index == 0xEE and lang == 0 and self._ms_vendor_code_v1 is not None:
      # See https://msdn.microsoft.com/en-us/windows/hardware/gg463179 for the
      # definition of this special string descriptor.
      buf = (struct.pack('<BB', 18, usb_constants.DescriptorType.STRING) +
             'MSFT100'.encode('UTF-16LE') +
             struct.pack('<BB', self._ms_vendor_code_v1, 0))
      assert len(buf) == 18
      return buf[:length]
    elif lang not in self._strings:
      return None
    elif index not in self._strings[lang]:
      return None
    else:
      descriptor = usb_descriptors.StringDescriptor(
          bString=self._strings[lang][index])
      return descriptor.Encode()[:length]

  def GetMicrosoftOSDescriptorV1(self, recipient, value, index, length):
    """Handle a the Microsoft OS 1.0 Descriptor request from the host.

    See https://msdn.microsoft.com/en-us/windows/hardware/gg463179 for the
    format of these descriptors.

    Args:
      recipient: Request recipient (device or interface)
      value: wValue field of the setup packet.
      index: wIndex field of the setup packet.
      length: Maximum amount of data the host expects the device to return.

    Returns:
      The descriptor or None to stall the pipe if the descriptor is not
      supported.
    """
    _ = recipient, value
    if index == 0x0004:
      return self.GetMicrosoftCompatIds(length)

  def GetMicrosoftCompatIds(self, length):
    interfaces = self.GetConfigurationDescriptor().GetInterfaces()
    max_interface = max([iface.bInterfaceNumber for iface in interfaces])

    header = struct.pack('<IHHBxxxxxxx',
                         16 + 24 * (max_interface + 1),
                         0x0100,
                         0x0004,
                         max_interface + 1)
    if length <= len(header):
      return header[:length]

    buf = header
    for interface in xrange(max_interface + 1):
      compat_id, sub_compat_id = self._ms_compat_ids.get(interface, ('', ''))
      buf += struct.pack('<BB8s8sxxxxxx',
                         interface, 0x01, compat_id, sub_compat_id)
    return buf[:length]

  def GetMicrosoftOSDescriptorV2(self, length):
    return self._ms_os20_descriptor_set.Encode()[:length]

  def GetBosDescriptor(self, length):
    """Handle a GET_DESCRIPTOR(BOS) request from the host.

    Device capability descriptors can be added to the Binary Device Object Store
    returned by this method by calling AddDeviceCapabilityDescriptor.

    See Universal Serial Bus 3.1 Specification, Revision 1.0 section 9.6.2.

    Args:
      length: Maximum amount of data the host expects the device to return.

    Returns:
      The device's binary object store descriptor or None to stall the pipe if
      no device capability descriptors have been configured.
    """
    if self._bos_descriptor is None:
      return None

    return self._bos_descriptor.Encode()[:length]

  def SetConfiguration(self, index):
    """Handle a SET_CONFIGURATION request from the host.

    See Universal Serial Bus Specification Revision 2.0 section 9.4.7.

    Args:
      index: Configuration index selected.

    Returns:
      True on success, None on error to stall the pipe.
    """
    print 'SetConfiguration({})'.format(index)

    for endpoint_addrs in self._active_endpoints.values():
      for endpoint_addr in endpoint_addrs:
        self._chip.StopEndpoint(endpoint_addr)
      endpoint_addrs.clear()
    self._endpoint_interface_map.clear();

    if index == 0:
      # SET_CONFIGRATION(0) puts the device into the Address state which
      # Windows does before suspending the port.
      return True
    elif index != 1:
      return None

    config_desc = self.GetConfigurationDescriptor()
    for interface_desc in config_desc.GetInterfaces():
      if interface_desc.bAlternateSetting != 0:
        continue
      endpoint_addrs = self._active_endpoints.setdefault(
          interface_desc.bInterfaceNumber, set())
      for endpoint_desc in interface_desc.GetEndpoints():
        self._chip.StartEndpoint(endpoint_desc)
        endpoint_addrs.add(endpoint_desc.bEndpointAddress)
        self._endpoint_interface_map[endpoint_desc.bEndpointAddress] = \
            interface_desc.bInterfaceNumber
    return True

  def SetInterface(self, interface, alt_setting):
    """Handle a SET_INTERFACE request from the host.

    See Universal Serial Bus Specification Revision 2.0 section 9.4.10.

    Args:
      interface: Interface number to configure.
      alt_setting: Alternate setting to select.

    Returns:
      True on success, None on error to stall the pipe.
    """
    print 'SetInterface({}, {})'.format(interface, alt_setting)

    config_desc = self.GetConfigurationDescriptor()
    interface_desc = None
    for interface_option in config_desc.GetInterfaces():
      if (interface_option.bInterfaceNumber == interface and
          interface_option.bAlternateSetting == alt_setting):
        interface_desc = interface_option
    if interface_desc is None:
      return None

    endpoint_addrs = self._active_endpoints.setdefault(interface, set())
    for endpoint_addr in endpoint_addrs:
      self._chip.StopEndpoint(endpoint_addr)
      del self._endpoint_interface_map[endpoint_addr]
    for endpoint_desc in interface_desc.GetEndpoints():
      self._chip.StartEndpoint(endpoint_desc)
      endpoint_addrs.add(endpoint_desc.bEndpointAddress)
      self._endpoint_interface_map[endpoint_desc.bEndpointAddress] = \
          interface_desc.bInterfaceNumber
    return True

  def GetInterfaceForEndpoint(self, endpoint_addr):
    return self._endpoint_interface_map.get(endpoint_addr)