586 lines
19 KiB
Python
586 lines
19 KiB
Python
# -*- test-case-name: wokkel.test.test_mam -*-
|
|
#
|
|
# Copyright (c) Adrien Cossa.
|
|
# See LICENSE for details.
|
|
|
|
"""
|
|
XMPP Message Archive Management protocol.
|
|
|
|
This protocol is specified in
|
|
U{XEP-0313<http://xmpp.org/extensions/xep-0313.html>}.
|
|
"""
|
|
|
|
from dateutil.tz import tzutc
|
|
|
|
from zope.interface import Interface, implements
|
|
|
|
from twisted.words.protocols.jabber.xmlstream import IQ, toResponse
|
|
from twisted.words.xish import domish
|
|
from twisted.words.protocols.jabber import jid
|
|
from twisted.python import log
|
|
|
|
from wokkel.subprotocols import IQHandlerMixin, XMPPHandler
|
|
from wokkel import disco, data_form, delay
|
|
|
|
import rsm
|
|
|
|
NS_MAM = 'urn:xmpp:mam:0'
|
|
NS_FORWARD = 'urn:xmpp:forward:0'
|
|
|
|
FIELDS_REQUEST = "/iq[@type='get']/query[@xmlns='%s']" % NS_MAM
|
|
ARCHIVE_REQUEST = "/iq[@type='set']/query[@xmlns='%s']" % NS_MAM
|
|
PREFS_GET_REQUEST = "/iq[@type='get']/prefs[@xmlns='%s']" % NS_MAM
|
|
PREFS_SET_REQUEST = "/iq[@type='set']/prefs[@xmlns='%s']" % NS_MAM
|
|
|
|
# TODO: add the tests!
|
|
|
|
|
|
class MAMError(Exception):
|
|
"""
|
|
MAM error.
|
|
"""
|
|
|
|
|
|
class Unsupported(MAMError):
|
|
def __init__(self, feature, text=None):
|
|
self.feature = feature
|
|
MAMError.__init__(self, 'feature-not-implemented',
|
|
'unsupported',
|
|
feature,
|
|
text)
|
|
|
|
def __str__(self):
|
|
message = MAMError.__str__(self)
|
|
message += ', feature %r' % self.feature
|
|
return message
|
|
|
|
|
|
class MAMQueryRequest():
|
|
"""
|
|
A Message Archive Management <query/> request.
|
|
|
|
@ivar form: Data Form specifing the filters.
|
|
@itype form: L{data_form.Form}
|
|
|
|
@ivar rsm: RSM request instance.
|
|
@itype rsm: L{rsm.RSMRequest}
|
|
|
|
@ivar node: pubsub node id if querying a pubsub node, else None.
|
|
@itype form: C{unicode}
|
|
"""
|
|
|
|
form = None
|
|
rsm = None
|
|
node = None
|
|
|
|
def __init__(self, form=None, rsm=None, node=None):
|
|
if form is not None:
|
|
assert(form.formType == 'submit')
|
|
self.form = form
|
|
self.rsm = rsm
|
|
self.node = node
|
|
|
|
@classmethod
|
|
def parse(cls, element):
|
|
"""Parse the DOM representation of a MAM <query/> request.
|
|
|
|
@param element: MAM <query/> request element.
|
|
@type element: L{Element<twisted.words.xish.domish.Element>}
|
|
|
|
@return: MAMQueryRequest instance.
|
|
@rtype: L{MAMQueryRequest}
|
|
"""
|
|
if element.uri != NS_MAM or element.name != 'query':
|
|
raise MAMError('Element provided is not a MAM <query/> request')
|
|
form = data_form.findForm(element, NS_MAM)
|
|
try:
|
|
rsm_request = rsm.RSMRequest.parse(element)
|
|
except rsm.RSMNotFoundError:
|
|
rsm_request = None
|
|
node = element['node'] if element.hasAttribute('node') else None
|
|
return MAMQueryRequest(form, rsm_request, node)
|
|
|
|
def toElement(self):
|
|
"""
|
|
Return the DOM representation of this RSM <query/> request.
|
|
|
|
@rtype: L{Element<twisted.words.xish.domish.Element>}
|
|
"""
|
|
mam_elt = domish.Element((NS_MAM, 'query'))
|
|
if self.node is not None:
|
|
mam_elt['node'] = self.node
|
|
if self.form is not None:
|
|
mam_elt.addChild(self.form.toElement())
|
|
if self.rsm is not None:
|
|
mam_elt.addChild(self.rsm.toElement())
|
|
|
|
return mam_elt
|
|
|
|
def render(self, parent):
|
|
"""Embed the DOM representation of this MAM request in the given element.
|
|
|
|
@param parent: parent IQ element.
|
|
@type parent: L{Element<twisted.words.xish.domish.Element>}
|
|
|
|
@return: MAM request element.
|
|
@rtype: L{Element<twisted.words.xish.domish.Element>}
|
|
"""
|
|
assert(parent.name == 'iq')
|
|
mam_elt = self.toElement()
|
|
parent.addChild(mam_elt)
|
|
return mam_elt
|
|
|
|
|
|
class MAMPrefs():
|
|
"""
|
|
A Message Archive Management <prefs/> request.
|
|
|
|
@param default: A value in ('always', 'never', 'roster').
|
|
@type : C{unicode}
|
|
|
|
@param always (list): A list of JID instances.
|
|
@type always: C{list}
|
|
|
|
@param never (list): A list of JID instances.
|
|
@type never: C{list}
|
|
"""
|
|
|
|
default = None
|
|
always = None
|
|
never = None
|
|
|
|
def __init__(self, default=None, always=None, never=None):
|
|
if default:
|
|
assert(default in ('always', 'never', 'roster'))
|
|
self.default = default
|
|
if always:
|
|
assert(isinstance(always, list))
|
|
self.always = always
|
|
else:
|
|
self.always = []
|
|
if never:
|
|
assert(isinstance(never, list))
|
|
self.never = never
|
|
else:
|
|
self.never = []
|
|
|
|
@classmethod
|
|
def parse(cls, element):
|
|
"""Parse the DOM representation of a MAM <prefs/> request.
|
|
|
|
@param element: MAM <prefs/> request element.
|
|
@type element: L{Element<twisted.words.xish.domish.Element>}
|
|
|
|
@return: MAMPrefs instance.
|
|
@rtype: L{MAMPrefs}
|
|
"""
|
|
if element.uri != NS_MAM or element.name != 'prefs':
|
|
raise MAMError('Element provided is not a MAM <prefs/> request')
|
|
default = element['default'] if element.hasAttribute('default') else None
|
|
prefs = {}
|
|
for attr in ('always', 'never'):
|
|
prefs[attr] = []
|
|
try:
|
|
pref = domish.generateElementsNamed(element.elements(), attr).next()
|
|
for jid_s in domish.generateElementsNamed(pref.elements(), 'jid'):
|
|
prefs[attr].append(jid.JID(jid_s))
|
|
except StopIteration:
|
|
pass
|
|
return MAMPrefs(default, **prefs)
|
|
|
|
def toElement(self):
|
|
"""
|
|
Return the DOM representation of this RSM <prefs/>request.
|
|
|
|
@rtype: L{Element<twisted.words.xish.domish.Element>}
|
|
"""
|
|
mam_elt = domish.Element((NS_MAM, 'prefs'))
|
|
if self.default:
|
|
mam_elt['default'] = self.default
|
|
for attr in ('always', 'never'):
|
|
attr_elt = mam_elt.addElement(attr)
|
|
jids = getattr(self, attr)
|
|
for jid in jids:
|
|
attr_elt.addElement('jid', content=jid.full())
|
|
return mam_elt
|
|
|
|
def render(self, parent):
|
|
"""Embed the DOM representation of this MAM request in the given element.
|
|
|
|
@param parent: parent IQ element.
|
|
@type parent: L{Element<twisted.words.xish.domish.Element>}
|
|
|
|
@return: MAM request element.
|
|
@rtype: L{Element<twisted.words.xish.domish.Element>}
|
|
"""
|
|
assert(parent.name == 'iq')
|
|
mam_elt = self.toElement()
|
|
parent.addChild(mam_elt)
|
|
return mam_elt
|
|
|
|
|
|
class MAMClient(XMPPHandler):
|
|
"""
|
|
MAM client.
|
|
|
|
This handler implements the protocol for sending out MAM requests.
|
|
"""
|
|
|
|
def queryArchive(self, service=None, form=None, rsm=None, node=None, sender=None):
|
|
"""Query a user, MUC or pubsub archive.
|
|
|
|
@param service: Entity offering the MAM service (None for user archives).
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param form: Data Form to filter the request.
|
|
@type form: L{Form<wokkel.data_form.Form>}
|
|
|
|
@param rsm: RSM request instance.
|
|
@type rsm: L{RSMRequest<wokkel.rsm.RSMRequest>}
|
|
|
|
@param node: Pubsub node to query, or None if inappropriate.
|
|
@type node: C{unicode}
|
|
|
|
@param sender: Optional sender address.
|
|
@type sender: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: A deferred that fires upon receiving a response.
|
|
@rtype: L{Deferred<twisted.internet.defer.Deferred>}
|
|
"""
|
|
iq = IQ(self.xmlstream, 'set')
|
|
MAMQueryRequest(form, rsm, node).render(iq)
|
|
if sender is not None:
|
|
iq['from'] = unicode(sender)
|
|
return iq.send(to=service.full() if service else None)
|
|
|
|
def queryFields(self, service=None, sender=None):
|
|
"""Ask the server about additional supported fields.
|
|
|
|
@param service: Entity offering the MAM service (None for user archives).
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param sender: Optional sender address.
|
|
@type sender: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: A deferred that fires upon receiving a response.
|
|
@rtype: L{Deferred<twisted.internet.defer.Deferred>}
|
|
"""
|
|
# http://xmpp.org/extensions/xep-0313.html#query-form
|
|
iq = IQ(self.xmlstream, 'get')
|
|
MAMQueryRequest().render(iq)
|
|
if sender is not None:
|
|
iq['from'] = unicode(sender)
|
|
return iq.send(to=service.full() if service else None)
|
|
|
|
def queryPrefs(self, service=None, sender=None):
|
|
"""Retrieve the current user preferences.
|
|
|
|
@param service: Entity offering the MAM service (None for user archives).
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param sender: Optional sender address.
|
|
@type sender: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: A deferred that fires upon receiving a response.
|
|
@rtype: L{Deferred<twisted.internet.defer.Deferred>}
|
|
"""
|
|
# http://xmpp.org/extensions/xep-0313.html#prefs
|
|
iq = IQ(self.xmlstream, 'get')
|
|
MAMPrefs().render(iq)
|
|
if sender is not None:
|
|
iq['from'] = unicode(sender)
|
|
return iq.send(to=service.full() if service else None)
|
|
|
|
def setPrefs(self, service=None, default='roster', always=None, never=None, sender=None):
|
|
"""Set new user preferences.
|
|
|
|
@param service: Entity offering the MAM service (None for user archives).
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param default: A value in ('always', 'never', 'roster').
|
|
@type : C{unicode}
|
|
|
|
@param always (list): A list of JID instances.
|
|
@type always: C{list}
|
|
|
|
@param never (list): A list of JID instances.
|
|
@type never: C{list}
|
|
|
|
@param sender: Optional sender address.
|
|
@type sender: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: A deferred that fires upon receiving a response.
|
|
@rtype: L{Deferred<twisted.internet.defer.Deferred>}
|
|
"""
|
|
# http://xmpp.org/extensions/xep-0313.html#prefs
|
|
assert(default is not None)
|
|
iq = IQ(self.xmlstream, 'set')
|
|
MAMPrefs(default, always, never).render(iq)
|
|
if sender is not None:
|
|
iq['from'] = unicode(sender)
|
|
return iq.send(to=service.full() if service else None)
|
|
|
|
|
|
class IMAMResource(Interface):
|
|
|
|
def onArchiveRequest(self, mam, requestor):
|
|
"""
|
|
|
|
@param mam: The MAM <query/> request.
|
|
@type mam: L{MAMQueryReques<wokkel.mam.MAMQueryRequest>}
|
|
|
|
@param requestor: JID of the requestor.
|
|
@type requestor: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: The RSM answer.
|
|
@rtype: L{RSMResponse<wokkel.rsm.RSMResponse>}
|
|
"""
|
|
|
|
def onPrefsGetRequest(self, requestor):
|
|
"""
|
|
|
|
@param requestor: JID of the requestor.
|
|
@type requestor: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: The current settings.
|
|
@rtype: L{wokkel.mam.MAMPrefs}
|
|
"""
|
|
|
|
def onPrefsSetRequest(self, prefs, requestor):
|
|
"""
|
|
|
|
@param prefs: The new settings to set.
|
|
@type prefs: L{wokkel.mam.MAMPrefs}
|
|
|
|
@param requestor: JID of the requestor.
|
|
@type requestor: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@return: The new current settings.
|
|
@rtype: L{wokkel.mam.MAMPrefs}
|
|
"""
|
|
|
|
|
|
class MAMService(XMPPHandler, IQHandlerMixin):
|
|
"""
|
|
Protocol implementation for a MAM service.
|
|
|
|
This handler waits for XMPP Ping requests and sends a response.
|
|
"""
|
|
|
|
implements(disco.IDisco)
|
|
|
|
iqHandlers = {FIELDS_REQUEST: '_onFieldsRequest',
|
|
ARCHIVE_REQUEST: '_onArchiveRequest',
|
|
PREFS_GET_REQUEST: '_onPrefsGetRequest',
|
|
PREFS_SET_REQUEST: '_onPrefsSetRequest'
|
|
}
|
|
|
|
_legacyFilters = {'start': {'fieldType': 'text-single',
|
|
'var': 'start',
|
|
'label': 'Starting time',
|
|
'desc': 'Starting time a the result period.',
|
|
},
|
|
'end': {'fieldType': 'text-single',
|
|
'var': 'end',
|
|
'label': 'Ending time',
|
|
'desc': 'Ending time of the result period.',
|
|
},
|
|
'with': {'fieldType': 'jid-single',
|
|
'var': 'with',
|
|
'label': 'Entity',
|
|
'desc': 'Entity against which to match message.',
|
|
},
|
|
}
|
|
|
|
extra_filters = {}
|
|
|
|
def __init__(self, resource):
|
|
"""
|
|
@param resource: instance implementing IMAMResource
|
|
@type resource: L{object}
|
|
"""
|
|
self.resource = resource
|
|
|
|
def connectionInitialized(self):
|
|
"""
|
|
Called when the XML stream has been initialized.
|
|
|
|
This sets up an observer for incoming ping requests.
|
|
"""
|
|
self.xmlstream.addObserver(FIELDS_REQUEST, self.handleRequest)
|
|
self.xmlstream.addObserver(ARCHIVE_REQUEST, self.handleRequest)
|
|
self.xmlstream.addObserver(PREFS_GET_REQUEST, self.handleRequest)
|
|
self.xmlstream.addObserver(PREFS_SET_REQUEST, self.handleRequest)
|
|
|
|
def addFilter(self, field):
|
|
"""
|
|
Add a new filter for querying MAM archive.
|
|
|
|
@param field: dictionary specifying the attributes to build a
|
|
wokkel.data_form.Field.
|
|
@type field: C{dict}
|
|
"""
|
|
self.extra_filters[field.var] = field
|
|
|
|
def _onFieldsRequest(self, iq):
|
|
"""
|
|
Called when a fields request has been received.
|
|
|
|
This immediately replies with a result response.
|
|
"""
|
|
response = toResponse(iq, 'result')
|
|
query = response.addElement((NS_MAM, 'query'))
|
|
query.addChild(buildForm('form', extra=self.extra_filters).toElement())
|
|
self.xmlstream.send(response)
|
|
iq.handled = True
|
|
|
|
def _onArchiveRequest(self, iq):
|
|
"""
|
|
Called when a message archive request has been received.
|
|
|
|
This immediately replies with a result response, followed by the
|
|
list of archived message and the finally the <fin/> message.
|
|
"""
|
|
response = toResponse(iq, 'result')
|
|
self.xmlstream.send(response)
|
|
|
|
mam_ = MAMQueryRequest.parse(iq.query)
|
|
requestor = jid.JID(iq['from'])
|
|
|
|
# remove unsupported filters
|
|
unsupported_fields = []
|
|
if mam_.form:
|
|
for key, field in mam_.form.fields.iteritems():
|
|
if key not in self._legacyFilters and key not in self.extra_filters:
|
|
log.msg('Ignored unsupported MAM filter: %s' % field)
|
|
unsupported_fields.append(key)
|
|
for key in unsupported_fields:
|
|
del mam_.form.fields[key]
|
|
|
|
def forward_message(id_, elt, date):
|
|
msg = domish.Element((None, 'message'))
|
|
msg['to'] = iq['from']
|
|
result = msg.addElement('result', NS_MAM)
|
|
if iq.hasAttribute('queryid'):
|
|
result['queryid'] = iq.query['queryid']
|
|
result['id'] = id_
|
|
forward = result.addElement('forwarded', NS_FORWARD)
|
|
forward.addChild(delay.Delay(date).toElement())
|
|
forward.addChild(elt)
|
|
self.xmlstream.send(msg)
|
|
|
|
def cb(result):
|
|
msg_data, rsm_elt = result
|
|
for data in msg_data:
|
|
forward_message(*data)
|
|
msg = domish.Element((None, 'message'))
|
|
msg['to'] = iq['from']
|
|
fin = msg.addElement('fin', NS_MAM)
|
|
if iq.hasAttribute('queryid'):
|
|
fin['queryid'] = iq.query['queryid']
|
|
if rsm_elt is not None:
|
|
fin.addChild(rsm_elt)
|
|
self.xmlstream.send(msg)
|
|
|
|
self.resource.onArchiveRequest(mam_, requestor).addCallback(cb)
|
|
iq.handled = True
|
|
|
|
def _onPrefsGetRequest(self, iq):
|
|
"""
|
|
Called when a prefs get request has been received.
|
|
|
|
This immediately replies with a result response.
|
|
"""
|
|
response = toResponse(iq, 'result')
|
|
|
|
requestor = jid.JID(iq['from'])
|
|
|
|
def cb(prefs):
|
|
response.addChild(prefs.toElement())
|
|
self.xmlstream.send(response)
|
|
|
|
self.resource.onPrefsGetRequest(requestor).addCallback(cb)
|
|
iq.handled = True
|
|
|
|
def _onPrefsSetRequest(self, iq):
|
|
"""
|
|
Called when a prefs get request has been received.
|
|
|
|
This immediately replies with a result response.
|
|
"""
|
|
response = toResponse(iq, 'result')
|
|
|
|
prefs = MAMPrefs.parse(iq.prefs)
|
|
requestor = jid.JID(iq['from'])
|
|
|
|
def cb(prefs):
|
|
response.addChild(prefs.toElement())
|
|
self.xmlstream.send(response)
|
|
|
|
self.resource.onPrefsSetRequest(prefs, requestor).addCallback(cb)
|
|
iq.handled = True
|
|
|
|
def getDiscoInfo(self, requestor, target, nodeIdentifier=''):
|
|
return [disco.DiscoFeature(NS_MAM)]
|
|
|
|
def getDiscoItems(self, requestor, target, nodeIdentifier=''):
|
|
return []
|
|
|
|
|
|
def datetime2utc(datetime):
|
|
"""Convert a datetime to a XEP-0082 compliant UTC datetime.
|
|
|
|
@param datetime: Offset-aware timestamp to convert.
|
|
@type datetime: L{datetime<datetime.datetime>}
|
|
|
|
@return: The datetime converted to UTC.
|
|
@rtype: C{unicode}
|
|
"""
|
|
stampFormat = '%Y-%m-%dT%H:%M:%SZ'
|
|
return datetime.astimezone(tzutc()).strftime(stampFormat)
|
|
|
|
|
|
def buildForm(formType='submit', start=None, end=None, with_jid=None, extra=None):
|
|
"""Prepare a Data Form for MAM.
|
|
|
|
@param formType: The type of the Data Form ('submit' or 'form').
|
|
@type formType: C{unicode}
|
|
|
|
@param start: Offset-aware timestamp to filter out older messages.
|
|
@type start: L{datetime<datetime.datetime>}
|
|
|
|
@param end: Offset-aware timestamp to filter out later messages.
|
|
@type end: L{datetime<datetime.datetime>}
|
|
|
|
@param with_jid: JID against which to match messages.
|
|
@type with_jid: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param extra: list of extra fields that are not defined by the
|
|
specification. Each element of the list must be a dictionary
|
|
specifying the attributes to build a wokkel.data_form.Field.
|
|
@type: C{list}
|
|
|
|
@return: XEP-0004 Data Form object.
|
|
@rtype: L{Form<wokkel.data_form.Form>}
|
|
"""
|
|
form = data_form.Form(formType, formNamespace=NS_MAM)
|
|
filters = []
|
|
|
|
if formType == 'form':
|
|
filters.extend(MAMService._legacyFilters.values())
|
|
elif formType == 'submit':
|
|
if start:
|
|
filters.append({'var': 'start', 'value': datetime2utc(start)})
|
|
if end:
|
|
filters.append({'var': 'end', 'value': datetime2utc(end)})
|
|
if with_jid:
|
|
# must specify the field type to overwrite default value in Field.__init__
|
|
filters.append({'fieldType': 'jid-single', 'var': 'with', 'value': with_jid.full()})
|
|
|
|
if extra is not None:
|
|
filters.extend(extra)
|
|
|
|
for field in filters:
|
|
form.addField(data_form.Field(**field))
|
|
|
|
return form
|