brisa.upnp.ssdp

34 """ Implementation of a SSDP server. 35 36 The notify_received and search_received methods are called when the 37 appropriate type of datagram is received by the server. 38 """ 39 40 msg_already_started = 'tried to start() SSDPServer when already started' 41 msg_already_stopped = 'tried to stop() SSDPServer when already stopped' 42

43 - def __init__(self, server_name, xml_description_filename, max_age=1800, 44 receive_notify=True):

45 """ Constructor for the SSDPServer class. 46 47 @param server_name: server name 48 @param xml_description_filename: XML description filename 49 @param max_age: max age parameter, default 1800. 50 @param receive_notify: if False, ignores notify messages 51 52 @type server_name: string 53 @type xml_description_filename: 54 @type max_age: integer 55 @type receive_notify: boolean 56 """ 57 self.server_name = server_name 58 self.xml_description_filename = xml_description_filename 59 self.max_age = max_age 60 self.receive_notify = receive_notify 61 self.running = False 62 self.known_device = {} 63 self.advertised = {} 64 self._callbacks = {} 65 self.udp_transport = UDPTransport() 66 self.udp_listener = UDPListener(SSDP_ADDR, SSDP_PORT, 67 data_callback=self._datagram_received) 68 self.renew_loop = LoopingCall(self._renew_notifications) 69 self.renew_loop.start(0.8 * self.max_age, now=True)

70

71 - def is_running(self):

72 """ Returns True if the SSDPServer is running, False otherwise. 73 """ 74 return self.running

75

76 - def start(self):

77 """ Starts the SSDPServer. 78 """ 79 if not self.is_running(): 80 self.udp_listener.start() 81 self.running = True 82 else: 83 log.warning(self.msg_already_started)

84

85 - def stop(self):

86 """ Sends bye bye notifications and stops the SSDPServer. 87 """ 88 if self.is_running(): 89 # Avoid racing conditions 90 own_temp = self.advertised.copy() 91 for usn in own_temp: 92 self._do_byebye(usn) 93 94 self.renew_loop.stop() 95 self.udp_listener.stop() 96 self.running = False 97 else: 98 log.warning(self.msg_already_stopped)

99

100 - def destroy(self):

101 """ Destroys the SSDPServer. 102 """ 103 if self.is_running(): 104 self.stop() 105 self.renew_loop.destroy() 106 self.udp_listener.destroy() 107 self._cleanup()

108

109 - def clear_device_list(self):

110 """ Clears the device list. 111 """ 112 self.known_device.clear()

113

114 - def discovered_device_failed(self, dev):

115 """ Device could not be fully built, so forget it. 116 """ 117 usn = dev['USN'] 118 if usn in self.known_device: 119 self.known_device.pop(usn)

120

121 - def is_known_device(self, usn):

122 """ Returns if the device with the passed usn is already known. 123 124 @param usn: device's usn 125 @type usn: string 126 127 @return: True if it is known 128 @rtype: boolean 129 """ 130 return usn in self.known_device

131 143

144 - def unsubscribe(self, name, callback):

145 """ Unsubscribes a callback for an event. 146 147 @param name: name of the event 148 @param callback: callback 149 150 @type name: string 151 @type callback: callable 152 """ 153 callbacks = self._callbacks.get(name, []) 154 [callbacks.remove(c) for c in callbacks] 155 self._callbacks[name] = callbacks

156

157 - def announce_device(self):

158 """ Announces the device. 159 """ 160 [self._do_notify(usn) for usn in self.advertised]

161

162 - def register_device(self, device):

163 """ Registers a device on the SSDP server. 164 165 @param device: device to be registered 166 @type device: Device 167 """ 168 self._register_device(device) 169 if device.is_root_device(): 170 [self._register_device(d) for d in device.devices]

171 172 # Messaging 173

174 - def _datagram_received(self, data, (host, port)):

175 """ Handles a received multicast datagram. 176 177 @param data: raw data 178 @param host: datagram source host 179 @param port: datagram source port 180 181 @type data: string 182 @type host: string 183 @type port: integer 184 """ 185 try: 186 header, payload = data.split('\r\n\r\n') 187 except ValueError, err: 188 log.error('Error while receiving datagram packet: %s', str(err)) 189 190 lines = header.split('\r\n') 191 cmd = lines[0].split(' ') 192 lines = map(lambda x: x.replace(': ', ':', 1), lines[1:]) 193 lines = filter(lambda x: len(x) > 0, lines) 194 195 headers = [x.split(':', 1) for x in lines] 196 headers = dict(map(lambda x: (x[0].lower(), x[1]), headers)) 197 198 if cmd[0] == 'M-SEARCH' and cmd[1] == '*': 199 # SSDP discovery 200 log.debug('Received M-Search command from %s:%s', host, port) 201 self._discovery_request(headers, (host, port)) 202 elif cmd[0] == 'NOTIFY' and cmd[1] == '*': 203 if not self.receive_notify: 204 # Ignore notify 205 log.debug('Received NOTIFY command from %s:%s (ignored '\ 206 'because of SSDPServer.receive_notify is False)', 207 host, port) 208 return 209 # SSDP presence 210 self._notify_received(headers, (host, port)) 211 else: 212 log.warning('Received unknown SSDP command %s with headers %s '\ 213 'from %s:%s', cmd, str(headers), host, port)

214

215 - def _discovery_request(self, headers, (host, port)):

216 """ Processes discovery requests and responds accordingly. 217 218 @param headers: discovery headers 219 @param host: discovery source host 220 @param port: discovery source port 221 222 @type headers: dictionary 223 @type host: string 224 @type port integer 225 """ 226 right_key = 0 227 228 # Do we know about this service? 229 if headers['st'] == 'ssdp:all': 230 for i in self.known_device.keys(): 231 hcopy = dict(headers.iteritems()) 232 hcopy['st'] = self.known_device[i]['ST'] 233 self._discovery_request(hcopy, (host, port)) 234 return 235 236 for key in self.known_device.keys(): 237 if self.known_device[key]['ST'].split()[0] == (headers['st']).split()[0]: 238 right_key = key 239 break 240 else: 241 log.debug('Discovery request ST %s not found', headers['st']) 242 return 243 244 if right_key == 0: 245 log.error('Unknown error in DiscoveryRequest for %s', 246 headers['st']) 247 return 248 249 # Generate a response 250 response = [] 251 response.append('HTTP/1.1 200 OK') 252 append = response.append 253 254 [append('%s: %s' % (k, v)) for k, v in self.known_device[right_key].items()] 255 response.extend(('', '')) 256 delay = random.randint(0, int(headers['mx'])) 257 # Avoid using a timer with delay 0 :) 258 if delay: 259 self.udp_transport.send_delayed(delay, '\r\n'.join(response), 260 host, port) 261 else: 262 self.udp_transport.send_data('\r\n'.join(response), host, port) 263 log.debug('Discovery request response sent to (%s, %d)', host, port)

264

265 - def _notify_received(self, headers, (host, port)):

266 """ Processes a presence announcement. 267 268 @param headers: notify headers 269 @param host: notify source host 270 @param port: notify source port 271 272 @type headers: dictionary 273 @type host: string 274 @type port: integer 275 """ 276 if headers['nts'] == 'ssdp:alive': 277 if 'cache-control' not in headers: 278 headers['cache-control'] = 'max-age=1800' 279 try: 280 self.known_device[headers['usn']] 281 except KeyError: 282 self._register(headers['usn'], headers['nt'], 283 headers['location'], headers['server'], 284 headers['cache-control']) 285 elif headers['nts'] == 'ssdp:byebye': 286 if self.is_known_device(headers['usn']): 287 self._unregister(headers['usn']) 288 else: 289 log.warning('Unknown subtype %s for notification type %s', 290 headers['nts'], headers['nt'])

291 292 # Registering 293

294 - def _register(self, usn, st, location, server, cache_control, 295 where='remote'):

296 """ Registers a service or device. 297 298 @param usn: usn 299 @param st: st 300 @param location: location 301 @param server: server 302 @param cache_control: cache control 303 304 @type usn: string 305 @type location: string 306 @type st: string 307 @type server: string 308 @type cache_control: string 309 310 @note: these parameters are part of the UPnP Specification. Even though 311 they're abstracted by the framework (devices and services messages 312 already contain these parameters), if you want to understand it please 313 refer to the UPnP specification. Links can be found at our developer 314 documentation homepage. 315 """ 316 if where == 'remote': 317 d = self.known_device 318 elif where == 'local': 319 d = self.advertised 320 321 d[usn] = {'USN': usn, 322 'LOCATION': location, 323 'ST': st, 324 'EXT': '', 325 'SERVER': server, 326 'CACHE-CONTROL': cache_control} 327 328 if st == 'upnp:rootdevice' and where == 'remote': 329 self._callback("new_device_event", st, self.known_device[usn])

330

331 - def _local_register(self, usn, st, location, server, cache_control):

332 """ Registers locally a new service or device. 333 """ 334 log.debug('Registering locally %s (%s)', st, location) 335 self._register(usn, st, location, server, cache_control, 'local') 336 self._do_notify(usn)

337

338 - def _register_device(self, device):

339 device_id = device.udn 340 device_type = device.device_type 341 device_server = "BRisa Webserver UPnP/1.0 %s" % self.server_name 342 device_location = "%s/%s" % (device.location, 343 self.xml_description_filename) 344 age = 'max-age=%d' % self.max_age 345 346 # uuid:device-UUID::upnp:rootdevice 347 self._local_register('%s::upnp:rootdevice' % device_id, 348 'upnp:rootdevice', 349 device_location, 350 device_server, age) 351 352 # uuid:device-UUID 353 self._local_register(device_id, 354 device_id, 355 device_location, 356 device_server, age) 357 358 359 # urn:schemas-upnp-org:device:deviceType:v 360 self._local_register('%s::%s' % (device_id, device_type), 361 device_type, device_location, 362 device_server, age) 363 364 for serv_type, service in device.services.items(): 365 # urn:schemas-upnp-org:service:serviceType:v 366 self._local_register('%s::%s' % (device_id, service.service_type), 367 service.service_type, 368 device_location, device_server, age)

369

370 - def _renew_notifications(self):

371 """ Renew notifications (sends a notify 372 """ 373 own_temp = self.advertised.copy() 374 for usn in own_temp: 375 log.debug('Renew notification for %s ', own_temp[usn]['USN']) 376 self._do_notify(own_temp[usn]['USN'])

377

378 - def _unregister(self, usn):

379 log.debug("Unregistering %s", usn) 380 381 try: 382 self._callback("removed_device_event", self.known_device[usn]) 383 if usn in self.known_device: 384 del self.known_device[usn] 385 except: 386 pass

387 388 # Notify and byebye 389

390 - def _do_notify(self, usn):

391 """ Do a notification for the usn specified. 392 393 @param usn: st 394 @type usn: string 395 """ 396 log.debug('Sending alive notification for %s', usn) 397 resp = ['NOTIFY * HTTP/1.1', 'HOST: %s:%d' % (SSDP_ADDR, SSDP_PORT), 398 'NTS: ssdp:alive', ] 399 stcpy = dict(self.advertised[usn].iteritems()) 400 stcpy['NT'] = stcpy['ST'] 401 del stcpy['EXT'] 402 del stcpy['ST'] 403 resp.extend(map(lambda x: ': '.join(x), stcpy.iteritems())) 404 resp.extend(('', '')) 405 self.udp_transport.send_data('\r\n'.join(resp), SSDP_ADDR, SSDP_PORT) 406 self.udp_transport.send_data('\r\n'.join(resp), SSDP_ADDR, SSDP_PORT)

407

408 - def _do_byebye(self, usn):

409 """ Do byebye notification for the usn specified. 410 411 @param usn: usn 412 @type usn: string 413 """ 414 log.debug('Sending byebye notification for %s', usn) 415 resp = ['NOTIFY * HTTP/1.1', 'HOST: %s:%d' % (SSDP_ADDR, SSDP_PORT), 416 'NTS: ssdp:byebye', ] 417 stcpy = dict(self.advertised[usn].iteritems()) 418 stcpy['NT'] = stcpy['ST'] 419 del stcpy['ST'] 420 del stcpy['EXT'] 421 resp.extend(map(lambda x: ': '.join(x), stcpy.iteritems())) 422 resp.extend(('', '')) 423 self.udp_transport.send_data('\r\n'.join(resp), SSDP_ADDR, SSDP_PORT) 424 self.udp_transport.send_data('\r\n'.join(resp), SSDP_ADDR, SSDP_PORT)

425 426 # Eventing 427

428 - def _callback(self, name, *args):

429 """ Performs callbacks for events. 430 """ 431 for callback in self._callbacks.get(name, []): 432 callback(*args)

433 434 # Cleanup 435

436 - def _cleanup(self):

437 """ Cleans the SSDPServer by removing known devices and internal cache. 438 """ 439 self.clear_device_list()

Source Code for Module brisa.upnp.ssdp