git.rot13.org Git - goodfet/blob - client/USBDevice.py

   1 # USBDevice.py
   2 #
   3 # Contains class definitions for USBDevice and USBDeviceRequest.
   4
   5 from USB import *
   6
   7 class USBDevice:
   8     name = "generic device"
   9
  10     def __init__(self, maxusb_app, device_class, device_subclass,
  11             protocol_rel_num, max_packet_size_ep0, vendor_id, product_id,
  12             device_rev, manufacturer_string, product_string,
  13             serial_number_string, configurations=[], descriptors={},
  14             verbose=0):
  15         self.maxusb_app = maxusb_app
  16         self.verbose = verbose
  17
  18         self.strings = [ ]
  19
  20         self.usb_spec_version           = 0x0001
  21         self.device_class               = device_class
  22         self.device_subclass            = device_subclass
  23         self.protocol_rel_num           = protocol_rel_num
  24         self.max_packet_size_ep0        = max_packet_size_ep0
  25         self.vendor_id                  = vendor_id
  26         self.product_id                 = product_id
  27         self.device_rev                 = device_rev
  28         self.manufacturer_string_id     = self.get_string_id(manufacturer_string)
  29         self.product_string_id          = self.get_string_id(product_string)
  30         self.serial_number_string_id    = self.get_string_id(serial_number_string)
  31
  32         # maps from USB.desc_type_* to bytearray OR callable
  33         self.descriptors = descriptors
  34         self.descriptors[USB.desc_type_device] = self.get_descriptor
  35         self.descriptors[USB.desc_type_configuration] = self.handle_get_configuration_descriptor_request
  36         self.descriptors[USB.desc_type_string] = self.handle_get_string_descriptor_request
  37
  38         self.config_num = -1
  39         self.configuration = None
  40         self.configurations = configurations
  41
  42         for c in self.configurations:
  43             csi = self.get_string_id(c.configuration_string)
  44             c.set_configuration_string_index(csi)
  45
  46             for i in c.interfaces:
  47                 i.device = self
  48
  49         self.state = USB.state_detached
  50         self.ready = False
  51
  52         self.address = 0
  53
  54         self.setup_request_handlers()
  55
  56     def get_string_id(self, s):
  57         try:
  58             i = self.strings.index(s)
  59         except ValueError:
  60             # string descriptors start at index 1
  61             self.strings.append(s)
  62             i = len(self.strings)
  63
  64         return i
  65
  66     def setup_request_handlers(self):
  67         # see table 9-4 of USB 2.0 spec, page 279
  68         self.request_handlers = {
  69              0 : self.handle_get_status_request,
  70              1 : self.handle_clear_feature_request,
  71              3 : self.handle_set_feature_request,
  72              5 : self.handle_set_address_request,
  73              6 : self.handle_get_descriptor_request,
  74              7 : self.handle_set_descriptor_request,
  75              8 : self.handle_get_configuration_request,
  76              9 : self.handle_set_configuration_request,
  77             10 : self.handle_get_interface_request,
  78             11 : self.handle_set_interface_request,
  79             12 : self.handle_synch_frame_request
  80         }
  81
  82     def connect(self):
  83         self.maxusb_app.connect(self)
  84
  85         # skipping USB.state_attached may not be strictly correct (9.1.1.{1,2})
  86         self.state = USB.state_powered
  87
  88     def disconnect(self):
  89         self.maxusb_app.disconnect()
  90
  91         self.state = USB.state_detached
  92
  93     def run(self):
  94         self.maxusb_app.service_irqs()
  95
  96     def ack_status_stage(self):
  97         self.maxusb_app.ack_status_stage()
  98
  99     def get_descriptor(self, n):
 100         d = bytearray([
 101             18,         # length of descriptor in bytes
 102             1,          # descriptor type 1 == device
 103             (self.usb_spec_version >> 8) & 0xff,
 104             self.usb_spec_version & 0xff,
 105             self.device_class,
 106             self.device_subclass,
 107             self.protocol_rel_num,
 108             self.max_packet_size_ep0,
 109             (self.vendor_id >> 8) & 0xff,
 110             self.vendor_id & 0xff,
 111             (self.product_id >> 8) & 0xff,
 112             self.product_id & 0xff,
 113             (self.device_rev >> 8) & 0xff,
 114             self.device_rev & 0xff,
 115             self.manufacturer_string_id,
 116             self.product_string_id,
 117             self.serial_number_string_id,
 118             len(self.configurations)
 119         ])
 120
 121         return d
 122
 123     # IRQ handlers
 124     #####################################################
 125
 126     def handle_request(self, req):
 127         if self.verbose > 3:
 128             print(self.name, "received request", req)
 129
 130         # figure out the intended recipient
 131         if req.get_recipient() == USB.request_recipient_device:
 132             recipient = self
 133         elif req.get_recipient() == USB.request_recipient_interface:
 134             recipient = self.configuration.interfaces[req.index]
 135         elif req.get_recipient() == USB.request_recipient_endpoint:
 136             recipient = self.configuration.endpoints[req.index]
 137
 138         # and then the type
 139         if req.get_type() == USB.request_type_standard:
 140             handler = recipient.request_handlers[req.request]
 141             handler(req)
 142         elif req.get_type() == USB.request_type_class:
 143             # HACK: evidently, FreeBSD doesn't pay attention to the device
 144             # until it sends a GET_STATUS(class) message
 145             self.ready = True
 146             self.maxusb_app.stall_ep0()
 147         elif req.get_type() == USB.request_type_vendor:
 148             self.maxusb_app.stall_ep0()
 149
 150     def handle_data_available(self, ep_num, data):
 151         if self.ready and ep_num in self.endpoints:
 152             endpoint = self.endpoints[ep_num]
 153             endpoint.handler(data)
 154
 155     def handle_buffer_available(self, ep_num):
 156         if self.ready and ep_num in self.endpoints:
 157             endpoint = self.endpoints[ep_num]
 158             endpoint.handler()
 159
 160     # standard request handlers
 161     #####################################################
 162
 163     # USB 2.0 specification, section 9.4.5 (p 282 of pdf)
 164     def handle_get_status_request(self, req):
 165         print(self.name, "received GET_STATUS request")
 166
 167         # self-powered and remote-wakeup (USB 2.0 Spec section 9.4.5)
 168         response = b'\x03\x00'
 169         self.maxusb_app.send_on_endpoint(0, response)
 170
 171     # USB 2.0 specification, section 9.4.1 (p 280 of pdf)
 172     def handle_clear_feature_request(self, req):
 173         print(self.name, "received CLEAR_FEATURE request with type 0x%02x and value 0x%02x" \
 174                 % (req.request_type, req.value))
 175
 176     # USB 2.0 specification, section 9.4.9 (p 286 of pdf)
 177     def handle_set_feature_request(self, req):
 178         print(self.name, "received SET_FEATURE request")
 179
 180     # USB 2.0 specification, section 9.4.6 (p 284 of pdf)
 181     def handle_set_address_request(self, req):
 182         self.address = req.value
 183         self.state = USB.state_address
 184         self.ack_status_stage()
 185
 186         if self.verbose > 2:
 187             print(self.name, "received SET_ADDRESS request for address",
 188                     self.address)
 189
 190     # USB 2.0 specification, section 9.4.3 (p 281 of pdf)
 191     def handle_get_descriptor_request(self, req):
 192         dtype  = (req.value >> 8) & 0xff
 193         dindex = req.value & 0xff
 194         lang   = req.index
 195         n      = req.length
 196
 197         response = None
 198
 199         if self.verbose > 2:
 200             print(self.name, ("received GET_DESCRIPTOR req %d, index %d, " \
 201                     + "language 0x%04x, length %d") \
 202                     % (dtype, dindex, lang, n))
 203
 204         # TODO: handle KeyError
 205         response = self.descriptors[dtype]
 206         if callable(response):
 207             response = response(dindex)
 208
 209         if response:
 210             n = min(n, len(response))
 211             self.maxusb_app.verbose += 1
 212             self.maxusb_app.send_on_endpoint(0, response[:n])
 213             self.maxusb_app.verbose -= 1
 214
 215             if self.verbose > 5:
 216                 print(self.name, "sent", n, "bytes in response")
 217
 218     def handle_get_configuration_descriptor_request(self, num):
 219         return self.configurations[num].get_descriptor()
 220
 221     def handle_get_string_descriptor_request(self, num):
 222         if num == 0:
 223             # HACK: hard-coding baaaaad
 224             d = bytes([
 225                     4,      # length of descriptor in bytes
 226                     3,      # descriptor type 3 == string
 227                     9,      # language code 0, byte 0
 228                     4       # language code 0, byte 1
 229             ])
 230         else:
 231             # string descriptors start at 1
 232             s = self.strings[num-1].encode('utf-16')
 233             d = bytearray([
 234                     len(s) + 2,     # length of descriptor in bytes
 235                     3               # descriptor type 3 == string
 236             ])
 237             d += s
 238
 239         return d
 240
 241     # USB 2.0 specification, section 9.4.8 (p 285 of pdf)
 242     def handle_set_descriptor_request(self, req):
 243         print(self.name, "received SET_DESCRIPTOR request")
 244
 245     # USB 2.0 specification, section 9.4.2 (p 281 of pdf)
 246     def handle_get_configuration_request(self, req):
 247         print(self.name, "received GET_CONFIGURATION request with data 0x%02x" \
 248                 % req.data)
 249
 250     # USB 2.0 specification, section 9.4.7 (p 285 of pdf)
 251     def handle_set_configuration_request(self, req):
 252         print(self.name, "received SET_CONFIGURATION request")
 253
 254         # configs are one-based
 255         self.config_num = req.value - 1
 256         self.configuration = self.configurations[self.config_num]
 257         self.state = USB.state_configured
 258
 259         # collate endpoint numbers
 260         self.endpoints = { }
 261         for i in self.configuration.interfaces:
 262             for e in i.endpoints:
 263                 self.endpoints[e.number] = e
 264
 265         # HACK: blindly acknowledge request
 266         self.ack_status_stage()
 267
 268     # USB 2.0 specification, section 9.4.4 (p 282 of pdf)
 269     def handle_get_interface_request(self, req):
 270         print(self.name, "received GET_INTERFACE request")
 271
 272         if req.index == 0:
 273             # HACK: currently only support one interface
 274             self.maxusb_app.send_on_endpoint(0, b'\x00')
 275         else:
 276             self.maxusb_app.stall_ep0()
 277
 278     # USB 2.0 specification, section 9.4.10 (p 288 of pdf)
 279     def handle_set_interface_request(self, req):
 280         print(self.name, "received SET_INTERFACE request")
 281
 282     # USB 2.0 specification, section 9.4.11 (p 288 of pdf)
 283     def handle_synch_frame_request(self, req):
 284         print(self.name, "received SYNCH_FRAME request")
 285
 286
 287 class USBDeviceRequest:
 288     def __init__(self, raw_bytes):
 289         """Expects raw 8-byte setup data request packet"""
 290
 291         self.request_type   = raw_bytes[0]
 292         self.request        = raw_bytes[1]
 293         self.value          = (raw_bytes[3] << 8) | raw_bytes[2]
 294         self.index          = (raw_bytes[5] << 8) | raw_bytes[4]
 295         self.length         = (raw_bytes[7] << 8) | raw_bytes[6]
 296
 297     def __str__(self):
 298         s = "dir=%d, type=%d, rec=%d, r=%d, v=%d, i=%d, l=%d" \
 299                 % (self.get_direction(), self.get_type(), self.get_recipient(),
 300                    self.request, self.value, self.index, self.length)
 301         return s
 302
 303     def raw(self):
 304         """returns request as bytes"""
 305         b = bytes([ self.request_type, self.request,
 306                     (self.value  >> 8) & 0xff, self.value  & 0xff,
 307                     (self.index  >> 8) & 0xff, self.index  & 0xff,
 308                     (self.length >> 8) & 0xff, self.length & 0xff
 309                   ])
 310         return b
 311
 312     def get_direction(self):
 313         return (self.request_type >> 7) & 0x01
 314
 315     def get_type(self):
 316         return (self.request_type >> 5) & 0x03
 317
 318     def get_recipient(self):
 319         return self.request_type & 0x1f
 320