tools/usb_gadget/gadget.py

   1 # Copyright 2014 The Chromium 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 """Generic USB gadget functionality.
   6 """
   7
   8 import struct
   9
  10 import usb_constants
  11
  12
  13 class Gadget(object):
  14   """Basic functionality for a USB device.
  15
  16   Implements standard control requests assuming that a subclass will handle
  17   class- or vendor-specific requests.
  18   """
  19
  20   def __init__(self, device_desc, fs_config_desc, hs_config_desc):
  21     """Create a USB gadget device.
  22
  23     Args:
  24       device_desc: USB device descriptor.
  25       fs_config_desc: Low/full-speed device descriptor.
  26       hs_config_desc: High-speed device descriptor.
  27     """
  28     self._speed = usb_constants.Speed.UNKNOWN
  29     self._chip = None
  30     self._device_desc = device_desc
  31     self._fs_config_desc = fs_config_desc
  32     self._hs_config_desc = hs_config_desc
  33     # dict mapping language codes to a dict mapping indexes to strings
  34     self._strings = {}
  35     # dict mapping interface numbers to a set of endpoint addresses
  36     self._active_endpoints = {}
  37     # dict mapping endpoint addresses to interfaces
  38     self._endpoint_interface_map = {}
  39     self._ms_vendor_code = None
  40     self._ms_compat_ids = {}
  41
  42   def GetDeviceDescriptor(self):
  43     return self._device_desc
  44
  45   def GetFullSpeedConfigurationDescriptor(self):
  46     return self._fs_config_desc
  47
  48   def GetHighSpeedConfigurationDescriptor(self):
  49     return self._hs_config_desc
  50
  51   def GetConfigurationDescriptor(self):
  52     if self._speed == usb_constants.Speed.FULL:
  53       return self._fs_config_desc
  54     elif self._speed == usb_constants.Speed.HIGH:
  55       return self._hs_config_desc
  56     else:
  57       raise RuntimeError('Device is not connected.')
  58
  59   def GetSpeed(self):
  60     return self._speed
  61
  62   def AddStringDescriptor(self, index, value, lang=0x0409):
  63     """Add a string descriptor to this device.
  64
  65     Args:
  66       index: String descriptor index (matches 'i' fields in descriptors).
  67       value: The string.
  68       lang: Language code (default: English).
  69
  70     Raises:
  71       ValueError: The index or language code is invalid.
  72     """
  73     if index < 1 or index > 255:
  74       raise ValueError('String descriptor index out of range.')
  75     if lang < 0 or lang > 0xffff:
  76       raise ValueError('String descriptor language code out of range.')
  77
  78     lang_strings = self._strings.setdefault(lang, {})
  79     lang_strings[index] = value
  80
  81   def EnableMicrosoftOSDescriptorsV1(self, vendor_code=0x01):
  82     if vendor_code < 0 or vendor_code > 255:
  83       raise ValueError('Vendor code out of range.')
  84
  85     self._ms_vendor_code = vendor_code
  86
  87   def SetMicrosoftCompatId(self, interface_number, compat_id, sub_compat_id=''):
  88     self._ms_compat_ids[interface_number] = (compat_id, sub_compat_id)
  89
  90   def Connected(self, chip, speed):
  91     """The device has been connected to a USB host.
  92
  93     Args:
  94       chip: USB controller.
  95       speed: Connection speed.
  96     """
  97     self._speed = speed
  98     self._chip = chip
  99
 100   def Disconnected(self):
 101     """The device has been disconnected from the USB host."""
 102     self._speed = usb_constants.Speed.UNKNOWN
 103     self._chip = None
 104     self._active_endpoints.clear()
 105     self._endpoint_interface_map.clear()
 106
 107   def IsConnected(self):
 108     return self._chip is not None
 109
 110   def ControlRead(self, request_type, request, value, index, length):
 111     """Handle a read on the control pipe (endpoint zero).
 112
 113     Args:
 114       request_type: bmRequestType field of the setup packet.
 115       request: bRequest field of the setup packet.
 116       value: wValue field of the setup packet.
 117       index: wIndex field of the setup packet.
 118       length: Maximum amount of data the host expects the device to return.
 119
 120     Returns:
 121       A buffer to return to the USB host with len <= length on success or
 122       None to stall the pipe.
 123     """
 124     assert request_type & usb_constants.Dir.IN
 125     typ = request_type & usb_constants.Type.MASK
 126     recipient = request_type & usb_constants.Recipient.MASK
 127     if typ == usb_constants.Type.STANDARD:
 128       return self.StandardControlRead(
 129           recipient, request, value, index, length)
 130     elif typ == usb_constants.Type.CLASS:
 131       return self.ClassControlRead(
 132           recipient, request, value, index, length)
 133     elif typ == usb_constants.Type.VENDOR:
 134       return self.VendorControlRead(
 135           recipient, request, value, index, length)
 136
 137   def ControlWrite(self, request_type, request, value, index, data):
 138     """Handle a write to the control pipe (endpoint zero).
 139
 140     Args:
 141       request_type: bmRequestType field of the setup packet.
 142       request: bRequest field of the setup packet.
 143       value: wValue field of the setup packet.
 144       index: wIndex field of the setup packet.
 145       data: Data stage of the request.
 146
 147     Returns:
 148       True on success, None to stall the pipe.
 149     """
 150     assert not request_type & usb_constants.Dir.IN
 151     typ = request_type & usb_constants.Type.MASK
 152     recipient = request_type & usb_constants.Recipient.MASK
 153     if typ == usb_constants.Type.STANDARD:
 154       return self.StandardControlWrite(
 155           recipient, request, value, index, data)
 156     elif typ == usb_constants.Type.CLASS:
 157       return self.ClassControlWrite(
 158           recipient, request, value, index, data)
 159     elif typ == usb_constants.Type.VENDOR:
 160       return self.VendorControlWrite(
 161           recipient, request, value, index, data)
 162
 163   def SendPacket(self, endpoint, data):
 164     """Send a data packet on the given endpoint.
 165
 166     Args:
 167       endpoint: Endpoint address.
 168       data: Data buffer.
 169
 170     Raises:
 171       ValueError: If the endpoint address is not valid.
 172       RuntimeError: If the device is not connected.
 173     """
 174     if self._chip is None:
 175       raise RuntimeError('Device is not connected.')
 176     if not endpoint & usb_constants.Dir.IN:
 177       raise ValueError('Cannot write to non-input endpoint.')
 178     self._chip.SendPacket(endpoint, data)
 179
 180   def ReceivePacket(self, endpoint, data):
 181     """Handle an incoming data packet on one of the device's active endpoints.
 182
 183     This method should be overridden by a subclass implementing endpoint-based
 184     data transfers.
 185
 186     Args:
 187       endpoint: Endpoint address.
 188       data: Data buffer.
 189     """
 190     pass
 191
 192   def HaltEndpoint(self, endpoint):
 193     """Signals a STALL condition to the host on the given endpoint.
 194
 195     Args:
 196       endpoint: Endpoint address.
 197     """
 198     self._chip.HaltEndpoint(endpoint)
 199
 200   def StandardControlRead(self, recipient, request, value, index, length):
 201     """Handle standard control transfers.
 202
 203     Args:
 204       recipient: Request recipient (device, interface, endpoint, etc.)
 205       request: bRequest field of the setup packet.
 206       value: wValue field of the setup packet.
 207       index: wIndex field of the setup packet.
 208       length: Maximum amount of data the host expects the device to return.
 209
 210     Returns:
 211       A buffer to return to the USB host with len <= length on success or
 212       None to stall the pipe.
 213     """
 214     if recipient == usb_constants.Recipient.DEVICE:
 215       if request == usb_constants.Request.GET_DESCRIPTOR:
 216         desc_type = value >> 8
 217         desc_index = value & 0xff
 218         desc_lang = index
 219
 220         print 'GetDescriptor(recipient={}, type={}, index={}, lang={})'.format(
 221             recipient, desc_type, desc_index, desc_lang)
 222
 223         return self.GetDescriptor(recipient, desc_type, desc_index, desc_lang,
 224                                   length)
 225
 226   def GetDescriptor(self, recipient, typ, index, lang, length):
 227     """Handle a standard GET_DESCRIPTOR request.
 228
 229     See Universal Serial Bus Specification Revision 2.0 section 9.4.3.
 230
 231     Args:
 232       recipient: Request recipient (device, interface, endpoint, etc.)
 233       typ: Descriptor type.
 234       index: Descriptor index.
 235       lang: Descriptor language code.
 236       length: Maximum amount of data the host expects the device to return.
 237
 238     Returns:
 239       The value of the descriptor or None to stall the pipe.
 240     """
 241     if typ == usb_constants.DescriptorType.STRING:
 242       return self.GetStringDescriptor(index, lang, length)
 243
 244   def ClassControlRead(self, recipient, request, value, index, length):
 245     """Handle class-specific control transfers.
 246
 247     This function should be overridden by a subclass implementing a particular
 248     device class.
 249
 250     Args:
 251       recipient: Request recipient (device, interface, endpoint, etc.)
 252       request: bRequest field of the setup packet.
 253       value: wValue field of the setup packet.
 254       index: wIndex field of the setup packet.
 255       length: Maximum amount of data the host expects the device to return.
 256
 257     Returns:
 258       A buffer to return to the USB host with len <= length on success or
 259       None to stall the pipe.
 260     """
 261     _ = recipient, request, value, index, length
 262     return None
 263
 264   def VendorControlRead(self, recipient, request, value, index, length):
 265     """Handle vendor-specific control transfers.
 266
 267     This function should be overridden by a subclass if implementing a device
 268     that responds to vendor-specific requests.
 269
 270     Args:
 271       recipient: Request recipient (device, interface, endpoint, etc.)
 272       request: bRequest field of the setup packet.
 273       value: wValue field of the setup packet.
 274       index: wIndex field of the setup packet.
 275       length: Maximum amount of data the host expects the device to return.
 276
 277     Returns:
 278       A buffer to return to the USB host with len <= length on success or
 279       None to stall the pipe.
 280     """
 281     if (self._ms_vendor_code is not None and
 282         request == self._ms_vendor_code and
 283         (recipient == usb_constants.Recipient.DEVICE or
 284          recipient == usb_constants.Recipient.INTERFACE)):
 285       return self.GetMicrosoftOSDescriptorV1(recipient, value, index, length)
 286
 287     return None
 288
 289   def StandardControlWrite(self, recipient, request, value, index, data):
 290     """Handle standard control transfers.
 291
 292     Args:
 293       recipient: Request recipient (device, interface, endpoint, etc.)
 294       request: bRequest field of the setup packet.
 295       value: wValue field of the setup packet.
 296       index: wIndex field of the setup packet.
 297       data: Data stage of the request.
 298
 299     Returns:
 300       True on success, None to stall the pipe.
 301     """
 302     _ = data
 303
 304     if request == usb_constants.Request.SET_CONFIGURATION:
 305       if recipient == usb_constants.Recipient.DEVICE:
 306         return self.SetConfiguration(value)
 307     elif request == usb_constants.Request.SET_INTERFACE:
 308       if recipient == usb_constants.Recipient.INTERFACE:
 309         return self.SetInterface(index, value)
 310
 311   def ClassControlWrite(self, recipient, request, value, index, data):
 312     """Handle class-specific control transfers.
 313
 314     This function should be overridden by a subclass implementing a particular
 315     device class.
 316
 317     Args:
 318       recipient: Request recipient (device, interface, endpoint, etc.)
 319       request: bRequest field of the setup packet.
 320       value: wValue field of the setup packet.
 321       index: wIndex field of the setup packet.
 322       data: Data stage of the request.
 323
 324     Returns:
 325       True on success, None to stall the pipe.
 326     """
 327     _ = recipient, request, value, index, data
 328     return None
 329
 330   def VendorControlWrite(self, recipient, request, value, index, data):
 331     """Handle vendor-specific control transfers.
 332
 333     This function should be overridden by a subclass if implementing a device
 334     that responds to vendor-specific requests.
 335
 336     Args:
 337       recipient: Request recipient (device, interface, endpoint, etc.)
 338       request: bRequest field of the setup packet.
 339       value: wValue field of the setup packet.
 340       index: wIndex field of the setup packet.
 341       data: Data stage of the request.
 342
 343     Returns:
 344       True on success, None to stall the pipe.
 345     """
 346     _ = recipient, request, value, index, data
 347     return None
 348
 349   def GetStringDescriptor(self, index, lang, length):
 350     """Handle a GET_DESCRIPTOR(String) request from the host.
 351
 352     Descriptor index 0 returns the set of languages supported by the device.
 353     All other indices return the string descriptors registered with those
 354     indices.
 355
 356     See Universal Serial Bus Specification Revision 2.0 section 9.6.7.
 357
 358     Args:
 359       index: Descriptor index.
 360       lang: Descriptor language code.
 361       length: Maximum amount of data the host expects the device to return.
 362
 363     Returns:
 364       The string descriptor or None to stall the pipe if the descriptor is not
 365       found.
 366     """
 367     if index == 0:
 368       length = 2 + len(self._strings) * 2
 369       header = struct.pack('<BB', length, usb_constants.DescriptorType.STRING)
 370       lang_codes = [struct.pack('<H', lang)
 371                     for lang in self._strings.iterkeys()]
 372       buf = header + ''.join(lang_codes)
 373       assert len(buf) == length
 374       return buf[:length]
 375     if index == 0xEE and lang == 0 and self._ms_vendor_code is not None:
 376       # See https://msdn.microsoft.com/en-us/windows/hardware/gg463179 for the
 377       # definition of this special string descriptor.
 378       buf = (struct.pack('<BB', 18, usb_constants.DescriptorType.STRING) +
 379              'MSFT100'.encode('UTF-16LE') +
 380              struct.pack('<BB', self._ms_vendor_code, 0))
 381       assert len(buf) == 18
 382       return buf[:length]
 383     elif lang not in self._strings:
 384       return None
 385     elif index not in self._strings[lang]:
 386       return None
 387     else:
 388       string = self._strings[lang][index].encode('UTF-16LE')
 389       header = struct.pack(
 390           '<BB', 2 + len(string), usb_constants.DescriptorType.STRING)
 391       buf = header + string
 392       return buf[:length]
 393
 394   def GetMicrosoftOSDescriptorV1(self, recipient, value, index, length):
 395     """Handle a the Microsoft OS 1.0 Descriptor request from the host.
 396
 397     See https://msdn.microsoft.com/en-us/windows/hardware/gg463179 for the
 398     format of these descriptors.
 399
 400     Args:
 401       recipient: Request recipient (device or interface)
 402       value: wValue field of the setup packet.
 403       index: wIndex field of the setup packet.
 404       length: Maximum amount of data the host expects the device to return.
 405
 406     Returns:
 407       The descriptor or None to stall the pipe if the descriptor is not
 408       supported.
 409     """
 410     _ = recipient, value
 411     if index == 0x0004:
 412       return self.GetMicrosoftCompatIds(length)
 413
 414   def GetMicrosoftCompatIds(self, length):
 415     interfaces = self.GetConfigurationDescriptor().GetInterfaces()
 416     max_interface = max([iface.bInterfaceNumber for iface in interfaces])
 417
 418     header = struct.pack('<IHHBxxxxxxx',
 419                          16 + 24 * (max_interface + 1),
 420                          0x0100,
 421                          0x0004,
 422                          max_interface + 1)
 423     if length <= len(header):
 424       return header[:length]
 425
 426     buf = header
 427     for interface in xrange(max_interface + 1):
 428       compat_id, sub_compat_id = self._ms_compat_ids.get(interface, ('', ''))
 429       buf += struct.pack('<BB8s8sxxxxxx',
 430                          interface, 0x01, compat_id, sub_compat_id)
 431     return buf[:length]
 432
 433   def SetConfiguration(self, index):
 434     """Handle a SET_CONFIGURATION request from the host.
 435
 436     See Universal Serial Bus Specification Revision 2.0 section 9.4.7.
 437
 438     Args:
 439       index: Configuration index selected.
 440
 441     Returns:
 442       True on success, None on error to stall the pipe.
 443     """
 444     print 'SetConfiguration({})'.format(index)
 445
 446     for endpoint_addrs in self._active_endpoints.values():
 447       for endpoint_addr in endpoint_addrs:
 448         self._chip.StopEndpoint(endpoint_addr)
 449       endpoint_addrs.clear()
 450     self._endpoint_interface_map.clear();
 451
 452     if index == 0:
 453       # SET_CONFIGRATION(0) puts the device into the Address state which
 454       # Windows does before suspending the port.
 455       return True
 456     elif index != 1:
 457       return None
 458
 459     config_desc = self.GetConfigurationDescriptor()
 460     for interface_desc in config_desc.GetInterfaces():
 461       if interface_desc.bAlternateSetting != 0:
 462         continue
 463       endpoint_addrs = self._active_endpoints.setdefault(
 464           interface_desc.bInterfaceNumber, set())
 465       for endpoint_desc in interface_desc.GetEndpoints():
 466         self._chip.StartEndpoint(endpoint_desc)
 467         endpoint_addrs.add(endpoint_desc.bEndpointAddress)
 468         self._endpoint_interface_map[endpoint_desc.bEndpointAddress] = \
 469             interface_desc.bInterfaceNumber
 470     return True
 471
 472   def SetInterface(self, interface, alt_setting):
 473     """Handle a SET_INTERFACE request from the host.
 474
 475     See Universal Serial Bus Specification Revision 2.0 section 9.4.10.
 476
 477     Args:
 478       interface: Interface number to configure.
 479       alt_setting: Alternate setting to select.
 480
 481     Returns:
 482       True on success, None on error to stall the pipe.
 483     """
 484     print 'SetInterface({}, {})'.format(interface, alt_setting)
 485
 486     config_desc = self.GetConfigurationDescriptor()
 487     interface_desc = None
 488     for interface_option in config_desc.GetInterfaces():
 489       if (interface_option.bInterfaceNumber == interface and
 490           interface_option.bAlternateSetting == alt_setting):
 491         interface_desc = interface_option
 492     if interface_desc is None:
 493       return None
 494
 495     endpoint_addrs = self._active_endpoints.setdefault(interface, set())
 496     for endpoint_addr in endpoint_addrs:
 497       self._chip.StopEndpoint(endpoint_addr)
 498       del self._endpoint_interface_map[endpoint_addr]
 499     for endpoint_desc in interface_desc.GetEndpoints():
 500       self._chip.StartEndpoint(endpoint_desc)
 501       endpoint_addrs.add(endpoint_desc.bEndpointAddress)
 502       self._endpoint_interface_map[endpoint_desc.bEndpointAddress] = \
 503           interface_desc.bInterfaceNumber
 504     return True
 505
 506   def GetInterfaceForEndpoint(self, endpoint_addr):
 507     return self._endpoint_interface_map.get(endpoint_addr)