components/sync/tools/testserver/xmppserver.py

# Copyright 2013 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.

"""A bare-bones and non-compliant XMPP server.

Just enough of the protocol is implemented to get it to work with
Chrome's sync notification system.
"""

import asynchat
import asyncore
import base64
import re
import socket
from xml.dom import minidom

# pychecker complains about the use of fileno(), which is implemented
# by asyncore by forwarding to an internal object via __getattr__.
__pychecker__ = 'no-classattr'


class Error(Exception):
  """Error class for this module."""
  pass


class UnexpectedXml(Error):
  """Raised when an unexpected XML element has been encountered."""

  def __init__(self, xml_element):
    xml_text = xml_element.toxml()
    Error.__init__(self, 'Unexpected XML element', xml_text)


def ParseXml(xml_string):
  """Parses the given string as XML and returns a minidom element
  object.
  """
  dom = minidom.parseString(xml_string)

  # minidom handles xmlns specially, but there's a bug where it sets
  # the attribute value to None, which causes toxml() or toprettyxml()
  # to break.
  def FixMinidomXmlnsBug(xml_element):
    if xml_element.getAttribute('xmlns') is None:
      xml_element.setAttribute('xmlns', '')

  def ApplyToAllDescendantElements(xml_element, fn):
    fn(xml_element)
    for node in xml_element.childNodes:
      if node.nodeType == node.ELEMENT_NODE:
        ApplyToAllDescendantElements(node, fn)

  root = dom.documentElement
  ApplyToAllDescendantElements(root, FixMinidomXmlnsBug)
  return root


def CloneXml(xml):
  """Returns a deep copy of the given XML element.

  Args:
    xml: The XML element, which should be something returned from
         ParseXml() (i.e., a root element).
  """
  return xml.ownerDocument.cloneNode(True).documentElement


class StanzaParser(object):
  """A hacky incremental XML parser.

  StanzaParser consumes data incrementally via FeedString() and feeds
  its delegate complete parsed stanzas (i.e., XML documents) via
  FeedStanza().  Any stanzas passed to FeedStanza() are unlinked after
  the callback is done.

  Use like so:

  class MyClass(object):
    ...
    def __init__(self, ...):
      ...
      self._parser = StanzaParser(self)
      ...

    def SomeFunction(self, ...):
      ...
      self._parser.FeedString(some_data)
      ...

    def FeedStanza(self, stanza):
      ...
      print stanza.toprettyxml()
      ...
  """

  # NOTE(akalin): The following regexps are naive, but necessary since
  # none of the existing Python 2.4/2.5 XML libraries support
  # incremental parsing.  This works well enough for our purposes.
  #
  # The regexps below assume that any present XML element starts at
  # the beginning of the string, but there may be trailing whitespace.

  # Matches an opening stream tag (e.g., '<stream:stream foo="bar">')
  # (assumes that the stream XML namespace is defined in the tag).
  _stream_re = re.compile(r'^(<stream:stream [^>]*>)\s*')

  # Matches an empty element tag (e.g., '<foo bar="baz"/>').
  _empty_element_re = re.compile(r'^(<[^>]*/>)\s*')

  # Matches a non-empty element (e.g., '<foo bar="baz">quux</foo>').
  # Does *not* handle nested elements.
  _non_empty_element_re = re.compile(r'^(<([^ >]*)[^>]*>.*?</\2>)\s*')

  # The closing tag for a stream tag.  We have to insert this
  # ourselves since all XML stanzas are children of the stream tag,
  # which is never closed until the connection is closed.
  _stream_suffix = '</stream:stream>'

  def __init__(self, delegate):
    self._buffer = ''
    self._delegate = delegate

  def FeedString(self, data):
    """Consumes the given string data, possibly feeding one or more
    stanzas to the delegate.
    """
    self._buffer += data
    while (self._ProcessBuffer(self._stream_re, self._stream_suffix) or
           self._ProcessBuffer(self._empty_element_re) or
           self._ProcessBuffer(self._non_empty_element_re)):
      pass

  def _ProcessBuffer(self, regexp, xml_suffix=''):
    """If the buffer matches the given regexp, removes the match from
    the buffer, appends the given suffix, parses it, and feeds it to
    the delegate.

    Returns:
      Whether or not the buffer matched the given regexp.
    """
    results = regexp.match(self._buffer)
    if not results:
      return False
    xml_text = self._buffer[:results.end()] + xml_suffix
    self._buffer = self._buffer[results.end():]
    stanza = ParseXml(xml_text)
    self._delegate.FeedStanza(stanza)
    # Needed because stanza may have cycles.
    stanza.unlink()
    return True


class Jid(object):
  """Simple struct for an XMPP jid (essentially an e-mail address with
  an optional resource string).
  """

  def __init__(self, username, domain, resource=''):
    self.username = username
    self.domain = domain
    self.resource = resource

  def __str__(self):
    jid_str = "%s@%s" % (self.username, self.domain)
    if self.resource:
      jid_str += '/' + self.resource
    return jid_str

  def GetBareJid(self):
    return Jid(self.username, self.domain)


class IdGenerator(object):
  """Simple class to generate unique IDs for XMPP messages."""

  def __init__(self, prefix):
    self._prefix = prefix
    self._id = 0

  def GetNextId(self):
    next_id = "%s.%s" % (self._prefix, self._id)
    self._id += 1
    return next_id


class HandshakeTask(object):
  """Class to handle the initial handshake with a connected XMPP
  client.
  """

  # The handshake states in order.
  (_INITIAL_STREAM_NEEDED,
   _AUTH_NEEDED,
   _AUTH_STREAM_NEEDED,
   _BIND_NEEDED,
   _SESSION_NEEDED,
   _FINISHED) = range(6)

  # Used when in the _INITIAL_STREAM_NEEDED and _AUTH_STREAM_NEEDED
  # states.  Not an XML object as it's only the opening tag.
  #
  # The from and id attributes are filled in later.
  _STREAM_DATA = (
    '<stream:stream from="%s" id="%s" '
    'version="1.0" xmlns:stream="http://etherx.jabber.org/streams" '
    'xmlns="jabber:client">')

  # Used when in the _INITIAL_STREAM_NEEDED state.
  _AUTH_STANZA = ParseXml(
    '<stream:features xmlns:stream="http://etherx.jabber.org/streams">'
    '  <mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl">'
    '    <mechanism>PLAIN</mechanism>'
    '    <mechanism>X-GOOGLE-TOKEN</mechanism>'
    '    <mechanism>X-OAUTH2</mechanism>'
    '  </mechanisms>'
    '</stream:features>')

  # Used when in the _AUTH_NEEDED state.
  _AUTH_SUCCESS_STANZA = ParseXml(
    '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>')

  # Used when in the _AUTH_NEEDED state.
  _AUTH_FAILURE_STANZA = ParseXml(
    '<failure xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>')

  # Used when in the _AUTH_STREAM_NEEDED state.
  _BIND_STANZA = ParseXml(
    '<stream:features xmlns:stream="http://etherx.jabber.org/streams">'
    '  <bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"/>'
    '  <session xmlns="urn:ietf:params:xml:ns:xmpp-session"/>'
    '</stream:features>')

  # Used when in the _BIND_NEEDED state.
  #
  # The id and jid attributes are filled in later.
  _BIND_RESULT_STANZA = ParseXml(
    '<iq id="" type="result">'
    '  <bind xmlns="urn:ietf:params:xml:ns:xmpp-bind">'
    '    <jid/>'
    '  </bind>'
    '</iq>')

  # Used when in the _SESSION_NEEDED state.
  #
  # The id attribute is filled in later.
  _IQ_RESPONSE_STANZA = ParseXml('<iq id="" type="result"/>')

  def __init__(self, connection, resource_prefix, authenticated):
    self._connection = connection
    self._id_generator = IdGenerator(resource_prefix)
    self._username = ''
    self._domain = ''
    self._jid = None
    self._authenticated = authenticated
    self._resource_prefix = resource_prefix
    self._state = self._INITIAL_STREAM_NEEDED

  def FeedStanza(self, stanza):
    """Inspects the given stanza and changes the handshake state if needed.

    Called when a stanza is received from the client.  Inspects the
    stanza to make sure it has the expected attributes given the
    current state, advances the state if needed, and sends a reply to
    the client if needed.
    """
    def ExpectStanza(stanza, name):
      if stanza.tagName != name:
        raise UnexpectedXml(stanza)

    def ExpectIq(stanza, type, name):
      ExpectStanza(stanza, 'iq')
      if (stanza.getAttribute('type') != type or
          stanza.firstChild.tagName != name):
        raise UnexpectedXml(stanza)

    def GetStanzaId(stanza):
      return stanza.getAttribute('id')

    def HandleStream(stanza):
      ExpectStanza(stanza, 'stream:stream')
      domain = stanza.getAttribute('to')
      if domain:
        self._domain = domain
      SendStreamData()

    def SendStreamData():
      next_id = self._id_generator.GetNextId()
      stream_data = self._STREAM_DATA % (self._domain, next_id)
      self._connection.SendData(stream_data)

    def GetUserDomain(stanza):
      encoded_username_password = stanza.firstChild.data
      username_password = base64.b64decode(encoded_username_password)
      (_, username_domain, _) = username_password.split('\0')
      # The domain may be omitted.
      #
      # If we were using python 2.5, we'd be able to do:
      #
      #   username, _, domain = username_domain.partition('@')
      #   if not domain:
      #     domain = self._domain
      at_pos = username_domain.find('@')
      if at_pos != -1:
        username = username_domain[:at_pos]
        domain = username_domain[at_pos+1:]
      else:
        username = username_domain
        domain = self._domain
      return (username, domain)

    def Finish():
      self._state = self._FINISHED
      self._connection.HandshakeDone(self._jid)

    if self._state == self._INITIAL_STREAM_NEEDED:
      HandleStream(stanza)
      self._connection.SendStanza(self._AUTH_STANZA, False)
      self._state = self._AUTH_NEEDED

    elif self._state == self._AUTH_NEEDED:
      ExpectStanza(stanza, 'auth')
      (self._username, self._domain) = GetUserDomain(stanza)
      if self._authenticated:
        self._connection.SendStanza(self._AUTH_SUCCESS_STANZA, False)
        self._state = self._AUTH_STREAM_NEEDED
      else:
        self._connection.SendStanza(self._AUTH_FAILURE_STANZA, False)
        Finish()

    elif self._state == self._AUTH_STREAM_NEEDED:
      HandleStream(stanza)
      self._connection.SendStanza(self._BIND_STANZA, False)
      self._state = self._BIND_NEEDED

    elif self._state == self._BIND_NEEDED:
      ExpectIq(stanza, 'set', 'bind')
      stanza_id = GetStanzaId(stanza)
      resource_element = stanza.getElementsByTagName('resource')[0]
      resource = resource_element.firstChild.data
      full_resource = '%s.%s' % (self._resource_prefix, resource)
      response = CloneXml(self._BIND_RESULT_STANZA)
      response.setAttribute('id', stanza_id)
      self._jid = Jid(self._username, self._domain, full_resource)
      jid_text = response.parentNode.createTextNode(str(self._jid))
      response.getElementsByTagName('jid')[0].appendChild(jid_text)
      self._connection.SendStanza(response)
      self._state = self._SESSION_NEEDED

    elif self._state == self._SESSION_NEEDED:
      ExpectIq(stanza, 'set', 'session')
      stanza_id = GetStanzaId(stanza)
      xml = CloneXml(self._IQ_RESPONSE_STANZA)
      xml.setAttribute('id', stanza_id)
      self._connection.SendStanza(xml)
      Finish()


def AddrString(addr):
  return '%s:%d' % addr


class XmppConnection(asynchat.async_chat):
  """A single XMPP client connection.

  This class handles the connection to a single XMPP client (via a
  socket).  It does the XMPP handshake and also implements the (old)
  Google notification protocol.
  """

  # Used for acknowledgements to the client.
  #
  # The from and id attributes are filled in later.
  _IQ_RESPONSE_STANZA = ParseXml('<iq from="" id="" type="result"/>')

  def __init__(self, sock, socket_map, delegate, addr, authenticated):
    """Starts up the xmpp connection.

    Args:
      sock: The socket to the client.
      socket_map: A map from sockets to their owning objects.
      delegate: The delegate, which is notified when the XMPP
        handshake is successful, when the connection is closed, and
        when a notification has to be broadcast.
      addr: The host/port of the client.
    """
    # We do this because in versions of python < 2.6,
    # async_chat.__init__ doesn't take a map argument nor pass it to
    # dispatcher.__init__.  We rely on the fact that
    # async_chat.__init__ calls dispatcher.__init__ as the last thing
    # it does, and that calling dispatcher.__init__ with socket=None
    # and map=None is essentially a no-op.
    asynchat.async_chat.__init__(self)
    asyncore.dispatcher.__init__(self, sock, socket_map)

    self.set_terminator(None)

    self._delegate = delegate
    self._parser = StanzaParser(self)
    self._jid = None

    self._addr = addr
    addr_str = AddrString(self._addr)
    self._handshake_task = HandshakeTask(self, addr_str, authenticated)
    print 'Starting connection to %s' % self

  def __str__(self):
    if self._jid:
      return str(self._jid)
    else:
      return AddrString(self._addr)

  # async_chat implementation.

  def collect_incoming_data(self, data):
    self._parser.FeedString(data)

  # This is only here to make pychecker happy.
  def found_terminator(self):
    asynchat.async_chat.found_terminator(self)

  def close(self):
    print "Closing connection to %s" % self
    self._delegate.OnXmppConnectionClosed(self)
    asynchat.async_chat.close(self)

  # Called by self._parser.FeedString().
  def FeedStanza(self, stanza):
    if self._handshake_task:
      self._handshake_task.FeedStanza(stanza)
    elif stanza.tagName == 'iq' and stanza.getAttribute('type') == 'result':
      # Ignore all client acks.
      pass
    elif (stanza.firstChild and
          stanza.firstChild.namespaceURI == 'google:push'):
      self._HandlePushCommand(stanza)
    else:
      raise UnexpectedXml(stanza)

  # Called by self._handshake_task.
  def HandshakeDone(self, jid):
    if jid:
      self._jid = jid
      self._handshake_task = None
      self._delegate.OnXmppHandshakeDone(self)
      print "Handshake done for %s" % self
    else:
      print "Handshake failed for %s" % self
      self.close()

  def _HandlePushCommand(self, stanza):
    if stanza.tagName == 'iq' and stanza.firstChild.tagName == 'subscribe':
      # Subscription request.
      self._SendIqResponseStanza(stanza)
    elif stanza.tagName == 'message' and stanza.firstChild.tagName == 'push':
      # Send notification request.
      self._delegate.ForwardNotification(self, stanza)
    else:
      raise UnexpectedXml(command_xml)

  def _SendIqResponseStanza(self, iq):
    stanza = CloneXml(self._IQ_RESPONSE_STANZA)
    stanza.setAttribute('from', str(self._jid.GetBareJid()))
    stanza.setAttribute('id', iq.getAttribute('id'))
    self.SendStanza(stanza)

  def SendStanza(self, stanza, unlink=True):
    """Sends a stanza to the client.

    Args:
      stanza: The stanza to send.
      unlink: Whether to unlink stanza after sending it. (Pass in
      False if stanza is a constant.)
    """
    self.SendData(stanza.toxml())
    if unlink:
      stanza.unlink()

  def SendData(self, data):
    """Sends raw data to the client.
    """
    # We explicitly encode to ascii as that is what the client expects
    # (some minidom library functions return unicode strings).
    self.push(data.encode('ascii'))

  def ForwardNotification(self, notification_stanza):
    """Forwards a notification to the client."""
    notification_stanza.setAttribute('from', str(self._jid.GetBareJid()))
    notification_stanza.setAttribute('to', str(self._jid))
    self.SendStanza(notification_stanza, False)


class XmppServer(asyncore.dispatcher):
  """The main XMPP server class.

  The XMPP server starts accepting connections on the given address
  and spawns off XmppConnection objects for each one.

  Use like so:

    socket_map = {}
    xmpp_server = xmppserver.XmppServer(socket_map, ('127.0.0.1', 5222))
    asyncore.loop(30.0, False, socket_map)
  """

  # Used when sending a notification.
  _NOTIFICATION_STANZA = ParseXml(
    '<message>'
    '  <push xmlns="google:push">'
    '    <data/>'
    '  </push>'
    '</message>')

  def __init__(self, socket_map, addr):
    asyncore.dispatcher.__init__(self, None, socket_map)
    self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
    self.set_reuse_addr()
    self.bind(addr)
    self.listen(5)
    self._socket_map = socket_map
    self._connections = set()
    self._handshake_done_connections = set()
    self._notifications_enabled = True
    self._authenticated = True

  def handle_accept(self):
    (sock, addr) = self.accept()
    xmpp_connection = XmppConnection(
      sock, self._socket_map, self, addr, self._authenticated)
    self._connections.add(xmpp_connection)
    # Return the new XmppConnection for testing.
    return xmpp_connection

  def close(self):
    # A copy is necessary since calling close on each connection
    # removes it from self._connections.
    for connection in self._connections.copy():
      connection.close()
    asyncore.dispatcher.close(self)

  def EnableNotifications(self):
    self._notifications_enabled = True

  def DisableNotifications(self):
    self._notifications_enabled = False

  def MakeNotification(self, channel, data):
    """Makes a notification from the given channel and encoded data.

    Args:
      channel: The channel on which to send the notification.
      data: The notification payload.
    """
    notification_stanza = CloneXml(self._NOTIFICATION_STANZA)
    push_element = notification_stanza.getElementsByTagName('push')[0]
    push_element.setAttribute('channel', channel)
    data_element = push_element.getElementsByTagName('data')[0]
    encoded_data = base64.b64encode(data)
    data_text = notification_stanza.parentNode.createTextNode(encoded_data)
    data_element.appendChild(data_text)
    return notification_stanza

  def SendNotification(self, channel, data):
    """Sends a notification to all connections.

    Args:
      channel: The channel on which to send the notification.
      data: The notification payload.
    """
    notification_stanza = self.MakeNotification(channel, data)
    self.ForwardNotification(None, notification_stanza)
    notification_stanza.unlink()

  def SetAuthenticated(self, auth_valid):
    self._authenticated = auth_valid

    # We check authentication only when establishing new connections.  We close
    # all existing connections here to make sure previously connected clients
    # pick up on the change.  It's a hack, but it works well enough for our
    # purposes.
    if not self._authenticated:
      for connection in self._handshake_done_connections:
        connection.close()

  def GetAuthenticated(self):
    return self._authenticated

  # XmppConnection delegate methods.
  def OnXmppHandshakeDone(self, xmpp_connection):
    self._handshake_done_connections.add(xmpp_connection)

  def OnXmppConnectionClosed(self, xmpp_connection):
    self._connections.discard(xmpp_connection)
    self._handshake_done_connections.discard(xmpp_connection)

  def ForwardNotification(self, unused_xmpp_connection, notification_stanza):
    if self._notifications_enabled:
      for connection in self._handshake_done_connections:
        print 'Sending notification to %s' % connection
        connection.ForwardNotification(notification_stanza)
    else:
      print 'Notifications disabled; dropping notification'