sleekxmpp/plugins/xep_0050/adhoc.py

   1 """
   2     SleekXMPP: The Sleek XMPP Library
   3     Copyright (C) 2011 Nathanael C. Fritz, Lance J.T. Stout
   4     This file is part of SleekXMPP.
   5
   6     See the file LICENSE for copying permission.
   7 """
   8
   9 import logging
  10 import time
  11
  12 from sleekxmpp import Iq
  13 from sleekxmpp.xmlstream.handler import Callback
  14 from sleekxmpp.xmlstream.matcher import StanzaPath
  15 from sleekxmpp.xmlstream import register_stanza_plugin, JID
  16 from sleekxmpp.plugins.base import base_plugin
  17 from sleekxmpp.plugins.xep_0050 import stanza
  18 from sleekxmpp.plugins.xep_0050 import Command
  19
  20
  21 log = logging.getLogger(__name__)
  22
  23
  24 class xep_0050(base_plugin):
  25
  26     """
  27     XEP-0050: Ad-Hoc Commands
  28
  29     XMPP's Adhoc Commands provides a generic workflow mechanism for
  30     interacting with applications. The result is similar to menu selections
  31     and multi-step dialogs in normal desktop applications. Clients do not
  32     need to know in advance what commands are provided by any particular
  33     application or agent. While adhoc commands provide similar functionality
  34     to Jabber-RPC, adhoc commands are used primarily for human interaction.
  35
  36     Also see <http://xmpp.org/extensions/xep-0050.html>
  37
  38     Configuration Values:
  39         threaded -- Indicates if command events should be threaded.
  40                     Defaults to True.
  41
  42     Events:
  43         command_execute  -- Received a command with action="execute"
  44         command_next     -- Received a command with action="next"
  45         command_complete -- Received a command with action="complete"
  46         command_cancel   -- Received a command with action="cancel"
  47
  48     Attributes:
  49         threaded -- Indicates if command events should be threaded.
  50                     Defaults to True.
  51         commands -- A dictionary mapping JID/node pairs to command
  52                     names and handlers.
  53         sessions -- A dictionary or equivalent backend mapping
  54                     session IDs to dictionaries containing data
  55                     relevant to a command's session.
  56
  57     Methods:
  58         plugin_init       -- Overrides base_plugin.plugin_init
  59         post_init         -- Overrides base_plugin.post_init
  60         new_session       -- Return a new session ID.
  61         prep_handlers     -- Placeholder. May call with a list of handlers
  62                              to prepare them for use with the session storage
  63                              backend, if needed.
  64         set_backend       -- Replace the default session storage with some
  65                              external storage mechanism, such as a database.
  66                              The provided backend wrapper must be able to
  67                              act using the same syntax as a dictionary.
  68         add_command       -- Add a command for use by external entitites.
  69         get_commands      -- Retrieve a list of commands provided by a
  70                              remote agent.
  71         send_command      -- Send a command request to a remote agent.
  72         start_command     -- Command user API: initiate a command session
  73         continue_command  -- Command user API: proceed to the next step
  74         cancel_command    -- Command user API: cancel a command
  75         complete_command  -- Command user API: finish a command
  76         terminate_command -- Command user API: delete a command's session
  77     """
  78
  79     def plugin_init(self):
  80         """Start the XEP-0050 plugin."""
  81         self.xep = '0050'
  82         self.description = 'Ad-Hoc Commands'
  83         self.stanza = stanza
  84
  85         self.threaded = self.config.get('threaded', True)
  86         self.commands = {}
  87         self.sessions = self.config.get('session_db', {})
  88
  89         self.xmpp.register_handler(
  90                 Callback("Ad-Hoc Execute",
  91                          StanzaPath('iq@type=set/command'),
  92                          self._handle_command))
  93
  94         self.xmpp.register_handler(
  95                 Callback("Ad-Hoc Result",
  96                          StanzaPath('iq@type=result/command'),
  97                          self._handle_command_result))
  98
  99         self.xmpp.register_handler(
 100                 Callback("Ad-Hoc Error",
 101                          StanzaPath('iq@type=error/command'),
 102                          self._handle_command_result))
 103
 104         register_stanza_plugin(Iq, stanza.Command)
 105
 106         self.xmpp.add_event_handler('command_execute',
 107                                     self._handle_command_start,
 108                                     threaded=self.threaded)
 109         self.xmpp.add_event_handler('command_next',
 110                                     self._handle_command_next,
 111                                     threaded=self.threaded)
 112         self.xmpp.add_event_handler('command_cancel',
 113                                     self._handle_command_cancel,
 114                                     threaded=self.threaded)
 115         self.xmpp.add_event_handler('command_complete',
 116                                     self._handle_command_complete,
 117                                     threaded=self.threaded)
 118
 119     def post_init(self):
 120         """Handle cross-plugin interactions."""
 121         base_plugin.post_init(self)
 122         self.xmpp['xep_0030'].add_feature(Command.namespace)
 123
 124     def set_backend(self, db):
 125         """
 126         Replace the default session storage dictionary with
 127         a generic, external data storage mechanism.
 128
 129         The replacement backend must be able to interact through
 130         the same syntax and interfaces as a normal dictionary.
 131
 132         Arguments:
 133             db -- The new session storage mechanism.
 134         """
 135         self.sessions = db
 136
 137     def prep_handlers(self, handlers, **kwargs):
 138         """
 139         Prepare a list of functions for use by the backend service.
 140
 141         Intended to be replaced by the backend service as needed.
 142
 143         Arguments:
 144             handlers -- A list of function pointers
 145             **kwargs -- Any additional parameters required by the backend.
 146         """
 147         pass
 148
 149     # =================================================================
 150     # Server side (command provider) API
 151
 152     def add_command(self, jid=None, node=None, name='', handler=None):
 153         """
 154         Make a new command available to external entities.
 155
 156         Access control may be implemented in the provided handler.
 157
 158         Command workflow is done across a sequence of command handlers. The
 159         first handler is given the intial Iq stanza of the request in order
 160         to support access control. Subsequent handlers are given only the
 161         payload items of the command. All handlers will receive the command's
 162         session data.
 163
 164         Arguments:
 165             jid     -- The JID that will expose the command.
 166             node    -- The node associated with the command.
 167             name    -- A human readable name for the command.
 168             handler -- A function that will generate the response to the
 169                        initial command request, as well as enforcing any
 170                        access control policies.
 171         """
 172         if jid is None:
 173             jid = self.xmpp.boundjid
 174         elif not isinstance(jid, JID):
 175             jid = JID(jid)
 176         item_jid = jid.full
 177
 178         # Client disco uses only the bare JID
 179         if self.xmpp.is_component:
 180             jid = jid.full
 181         else:
 182             jid = jid.bare
 183
 184         self.xmpp['xep_0030'].add_identity(category='automation',
 185                                            itype='command-list',
 186                                            name='Ad-Hoc commands',
 187                                            node=Command.namespace,
 188                                            jid=jid)
 189         self.xmpp['xep_0030'].add_item(jid=item_jid,
 190                                        name=name,
 191                                        node=Command.namespace,
 192                                        subnode=node,
 193                                        ijid=jid)
 194         self.xmpp['xep_0030'].add_identity(category='automation',
 195                                            itype='command-node',
 196                                            name=name,
 197                                            node=node,
 198                                            jid=jid)
 199         self.xmpp['xep_0030'].add_feature(Command.namespace, None, jid)
 200
 201         self.commands[(item_jid, node)] = (name, handler)
 202
 203     def new_session(self):
 204         """Return a new session ID."""
 205         return str(time.time()) + '-' + self.xmpp.new_id()
 206
 207     def _handle_command(self, iq):
 208         """Raise command events based on the command action."""
 209         self.xmpp.event('command_%s' % iq['command']['action'], iq)
 210
 211     def _handle_command_start(self, iq):
 212         """
 213         Process an initial request to execute a command.
 214
 215         Arguments:
 216             iq -- The command execution request.
 217         """
 218         sessionid = self.new_session()
 219         node = iq['command']['node']
 220         key = (iq['to'].full, node)
 221         name, handler = self.commands.get(key, ('Not found', None))
 222         if not handler:
 223             log.debug('Command not found: %s, %s' % (key, self.commands))
 224
 225         initial_session = {'id': sessionid,
 226                            'from': iq['from'],
 227                            'to': iq['to'],
 228                            'node': node,
 229                            'payload': None,
 230                            'interfaces': '',
 231                            'payload_classes': None,
 232                            'notes': None,
 233                            'has_next': False,
 234                            'allow_complete': False,
 235                            'allow_prev': False,
 236                            'past': [],
 237                            'next': None,
 238                            'prev': None,
 239                            'cancel': None}
 240
 241         session = handler(iq, initial_session)
 242
 243         self._process_command_response(iq, session)
 244
 245     def _handle_command_next(self, iq):
 246         """
 247         Process a request for the next step in the workflow
 248         for a command with multiple steps.
 249
 250         Arguments:
 251             iq -- The command continuation request.
 252         """
 253         sessionid = iq['command']['sessionid']
 254         session = self.sessions[sessionid]
 255
 256         handler = session['next']
 257         interfaces = session['interfaces']
 258         results = []
 259         for stanza in iq['command']['substanzas']:
 260             if stanza.plugin_attrib in interfaces:
 261                 results.append(stanza)
 262         if len(results) == 1:
 263             results = results[0]
 264
 265         session = handler(results, session)
 266
 267         self._process_command_response(iq, session)
 268
 269     def _process_command_response(self, iq, session):
 270         """
 271         Generate a command reply stanza based on the
 272         provided session data.
 273
 274         Arguments:
 275             iq      -- The command request stanza.
 276             session -- A dictionary of relevant session data.
 277         """
 278         sessionid = session['id']
 279
 280         payload = session['payload']
 281         if not isinstance(payload, list):
 282             payload = [payload]
 283
 284         session['interfaces'] = [item.plugin_attrib for item in payload]
 285         session['payload_classes'] = [item.__class__ for item in payload]
 286
 287         self.sessions[sessionid] = session
 288
 289         for item in payload:
 290             register_stanza_plugin(Command, item.__class__, iterable=True)
 291
 292         iq.reply()
 293         iq['command']['node'] = session['node']
 294         iq['command']['sessionid'] = session['id']
 295
 296         if session['next'] is None:
 297             iq['command']['actions'] = []
 298             iq['command']['status'] = 'completed'
 299         elif session['has_next']:
 300             actions = ['next']
 301             if session['allow_complete']:
 302                 actions.append('complete')
 303             if session['allow_prev']:
 304                 actions.append('prev')
 305             iq['command']['actions'] = actions
 306             iq['command']['status'] = 'executing'
 307         else:
 308             iq['command']['actions'] = ['complete']
 309             iq['command']['status'] = 'executing'
 310
 311         iq['command']['notes'] = session['notes']
 312
 313         for item in payload:
 314             iq['command'].append(item)
 315
 316         iq.send()
 317
 318     def _handle_command_cancel(self, iq):
 319         """
 320         Process a request to cancel a command's execution.
 321
 322         Arguments:
 323             iq -- The command cancellation request.
 324         """
 325         node = iq['command']['node']
 326         sessionid = iq['command']['sessionid']
 327         session = self.sessions[sessionid]
 328         handler = session['cancel']
 329
 330         if handler:
 331             handler(iq, session)
 332
 333         try:
 334             del self.sessions[sessionid]
 335         except:
 336             pass
 337
 338         iq.reply()
 339         iq['command']['node'] = node
 340         iq['command']['sessionid'] = sessionid
 341         iq['command']['status'] = 'canceled'
 342         iq['command']['notes'] = session['notes']
 343         iq.send()
 344
 345     def _handle_command_complete(self, iq):
 346         """
 347         Process a request to finish the execution of command
 348         and terminate the workflow.
 349
 350         All data related to the command session will be removed.
 351
 352         Arguments:
 353             iq -- The command completion request.
 354         """
 355         node = iq['command']['node']
 356         sessionid = iq['command']['sessionid']
 357         session = self.sessions[sessionid]
 358         handler = session['next']
 359         interfaces = session['interfaces']
 360         results = []
 361         for stanza in iq['command']['substanzas']:
 362             if stanza.plugin_attrib in interfaces:
 363                 results.append(stanza)
 364         if len(results) == 1:
 365             results = results[0]
 366
 367         if handler:
 368             handler(results, session)
 369
 370         iq.reply()
 371         iq['command']['node'] = node
 372         iq['command']['sessionid'] = sessionid
 373         iq['command']['actions'] = []
 374         iq['command']['status'] = 'completed'
 375         iq['command']['notes'] = session['notes']
 376         iq.send()
 377
 378         del self.sessions[sessionid]
 379
 380
 381     # =================================================================
 382     # Client side (command user) API
 383
 384     def get_commands(self, jid, **kwargs):
 385         """
 386         Return a list of commands provided by a given JID.
 387
 388         Arguments:
 389             jid      -- The JID to query for commands.
 390             local    -- If true, then the query is for a JID/node
 391                         combination handled by this Sleek instance and
 392                         no stanzas need to be sent.
 393                         Otherwise, a disco stanza must be sent to the
 394                         remove JID to retrieve the items.
 395             ifrom    -- Specifiy the sender's JID.
 396             block    -- If true, block and wait for the stanzas' reply.
 397             timeout  -- The time in seconds to block while waiting for
 398                         a reply. If None, then wait indefinitely.
 399             callback -- Optional callback to execute when a reply is
 400                         received instead of blocking and waiting for
 401                         the reply.
 402             iterator -- If True, return a result set iterator using
 403                         the XEP-0059 plugin, if the plugin is loaded.
 404                         Otherwise the parameter is ignored.
 405         """
 406         return self.xmpp['xep_0030'].get_items(jid=jid,
 407                                                node=Command.namespace,
 408                                                **kwargs)
 409
 410     def send_command(self, jid, node, ifrom=None, action='execute',
 411                     payload=None, sessionid=None, **kwargs):
 412         """
 413         Create and send a command stanza, without using the provided
 414         workflow management APIs.
 415
 416         Arguments:
 417             jid       -- The JID to send the command request or result.
 418             node      -- The node for the command.
 419             ifrom     -- Specify the sender's JID.
 420             action    -- May be one of: execute, cancel, complete,
 421                          or cancel.
 422             payload   -- Either a list of payload items, or a single
 423                          payload item such as a data form.
 424             sessionid -- The current session's ID value.
 425             block     -- Specify if the send call will block until a
 426                          response is received, or a timeout occurs.
 427                          Defaults to True.
 428             timeout   -- The length of time (in seconds) to wait for a
 429                          response before exiting the send call
 430                          if blocking is used. Defaults to
 431                          sleekxmpp.xmlstream.RESPONSE_TIMEOUT
 432             callback  -- Optional reference to a stream handler
 433                          function. Will be executed when a reply
 434                          stanza is received.
 435         """
 436         iq = self.xmpp.Iq()
 437         iq['type'] = 'set'
 438         iq['to'] = jid
 439         if ifrom:
 440             iq['from'] = ifrom
 441         iq['command']['node'] = node
 442         iq['command']['action'] = action
 443         if sessionid is not None:
 444             iq['command']['sessionid'] = sessionid
 445         if payload is not None:
 446             if not isinstance(payload, list):
 447                 payload = [payload]
 448             for item in payload:
 449                 iq['command'].append(item)
 450         return iq.send(**kwargs)
 451
 452     def start_command(self, jid, node, session, ifrom=None):
 453         """
 454         Initiate executing a command provided by a remote agent.
 455
 456         The workflow provided is always non-blocking.
 457
 458         The provided session dictionary should contain:
 459             next  -- A handler for processing the command result.
 460             error -- A handler for processing any error stanzas
 461                      generated by the request.
 462
 463         Arguments:
 464             jid     -- The JID to send the command request.
 465             node    -- The node for the desired command.
 466             session -- A dictionary of relevant session data.
 467             ifrom   -- Optionally specify the sender's JID.
 468         """
 469         session['jid'] = jid
 470         session['node'] = node
 471         session['timestamp'] = time.time()
 472         session['payload'] = None
 473         iq = self.xmpp.Iq()
 474         iq['type'] = 'set'
 475         iq['to'] = jid
 476         if ifrom:
 477             iq['from'] = ifrom
 478             session['from'] = ifrom
 479         iq['command']['node'] = node
 480         iq['command']['action'] = 'execute'
 481         sessionid = 'client:pending_' + iq['id']
 482         session['id'] = sessionid
 483         self.sessions[sessionid] = session
 484         iq.send(block=False)
 485
 486     def continue_command(self, session):
 487         """
 488         Execute the next action of the command.
 489
 490         Arguments:
 491             session -- All stored data relevant to the current
 492                        command session.
 493         """
 494         sessionid = 'client:' + session['id']
 495         self.sessions[sessionid] = session
 496
 497         self.send_command(session['jid'],
 498                           session['node'],
 499                           ifrom=session.get('from', None),
 500                           action='next',
 501                           payload=session.get('payload', None),
 502                           sessionid=session['id'])
 503
 504     def cancel_command(self, session):
 505         """
 506         Cancel the execution of a command.
 507
 508         Arguments:
 509             session -- All stored data relevant to the current
 510                        command session.
 511         """
 512         sessionid = 'client:' + session['id']
 513         self.sessions[sessionid] = session
 514
 515         self.send_command(session['jid'],
 516                           session['node'],
 517                           ifrom=session.get('from', None),
 518                           action='cancel',
 519                           payload=session.get('payload', None),
 520                           sessionid=session['id'])
 521
 522     def complete_command(self, session):
 523         """
 524         Finish the execution of a command workflow.
 525
 526         Arguments:
 527             session -- All stored data relevant to the current
 528                        command session.
 529         """
 530         sessionid = 'client:' + session['id']
 531         self.sessions[sessionid] = session
 532
 533         self.send_command(session['jid'],
 534                           session['node'],
 535                           ifrom=session.get('from', None),
 536                           action='complete',
 537                           payload=session.get('payload', None),
 538                           sessionid=session['id'])
 539
 540     def terminate_command(self, session):
 541         """
 542         Delete a command's session after a command has completed
 543         or an error has occured.
 544
 545         Arguments:
 546             session -- All stored data relevant to the current
 547                        command session.
 548         """
 549         try:
 550             del self.sessions[session['id']]
 551         except:
 552             pass
 553
 554     def _handle_command_result(self, iq):
 555         """
 556         Process the results of a command request.
 557
 558         Will execute the 'next' handler stored in the session
 559         data, or the 'error' handler depending on the Iq's type.
 560
 561         Arguments:
 562             iq -- The command response.
 563         """
 564         sessionid = 'client:' + iq['command']['sessionid']
 565         pending = False
 566
 567         if sessionid not in self.sessions:
 568             pending = True
 569             pendingid = 'client:pending_' + iq['id']
 570             if pendingid not in self.sessions:
 571                 return
 572             sessionid = pendingid
 573
 574         session = self.sessions[sessionid]
 575         sessionid = 'client:' + iq['command']['sessionid']
 576         session['id'] = iq['command']['sessionid']
 577
 578         self.sessions[sessionid] = session
 579
 580         if pending:
 581             del self.sessions[pendingid]
 582
 583         handler_type = 'next'
 584         if iq['type'] == 'error':
 585             handler_type = 'error'
 586         handler = session.get(handler_type, None)
 587         if handler:
 588             handler(iq, session)
 589         elif iq['type'] == 'error':
 590             self.terminate_command(session)
 591
 592         if iq['command']['status'] == 'completed':
 593             self.terminate_command(session)