libervia-pubsub: sat_pubsub/backend.py comparison

comparison sat_pubsub/backend.py @ 353:7c5d85c6fb3a

backend: enforce schema on get/publish and notifications

author	Goffi <goffi@goffi.org>
date	Fri, 08 Sep 2017 08:02:05 +0200
parents	efbdca10f0fb
children	18b983fe9e1b

comparison

equal deleted inserted replaced

-:efbdca10f0fb
+:7c5d85c6fb3a
 def __init__(self, storage):
 utility.EventDispatcher.__init__(self)
 self.storage = storage
 self._callbackList = []
 def supportsPublisherAffiliation(self):
 return True
 def supportsGroupBlog(self):
 return True
 def supportsOutcastAffiliation(self):
 return True
 def supportsPersistentItems(self):
 return True
 def supportsPublishModel(self):
 return True
 def getNodeType(self, nodeIdentifier, pep, recipient=None):
 # FIXME: manage pep and recipient
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(lambda node: node.getType())
 d = self.privilege.isSubscribedFrom(requestor, recipient)
 d.addCallback(self._getNodesIds, pep, recipient)
 return d
 return self.storage.getNodeIds(pep, recipient)
 def getNodeMetaData(self, nodeIdentifier, pep, recipient=None):
 # FIXME: manage pep and recipient
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(lambda node: node.getMetaData())
 d.addCallback(self._makeMetaData)
 return d
 def _makeMetaData(self, metaData):
 options = []
 for key, value in metaData.iteritems():
 if key in self.nodeOptions:
 option.update(self.nodeOptions[key])
 option["value"] = value
 options.append(option)
 return options
 def _checkAuth(self, node, requestor):
 """ Check authorisation of publishing in node for requestor """
 def check(affiliation):
 if category:
 categories.append(category)
 return categories
+def enforceSchema(self, item_elt, schema, affiliation):
+"""modifify item according to element, or refuse publishing
+@param item_elt(domish.Element): item to check/modify
+@param schema(domish.Eement): schema to enfore
+@param affiliation(unicode): affiliation of the publisher
+"""
+try:
+x_elt = next(item_elt.elements(data_form.NS_X_DATA, 'x'))
+item_form = data_form.Form.fromElement(x_elt)
+except (StopIteration, data_form.Error):
+raise pubsub.BadRequest(text="node has a schema but item has no form")
+else:
+item_elt.children.remove(x_elt)
+schema_form = data_form.Form.fromElement(schema)
+# we enforce restrictions
+for field_elt in schema.elements(data_form.NS_X_DATA, 'field'):
+var = field_elt['var']
+for restrict_elt in field_elt.elements(const.NS_SCHEMA_RESTRICT, 'restrict'):
+write_restriction = restrict_elt.attributes.get('write')
+if write_restriction is not None:
+if write_restriction == 'owner':
+if affiliation != 'owner':
+# write is not allowed on this field, we use default value
+# we can safely use Field from schema_form because
+# we have created this instance only for this method
+try:
+item_form.removeField(item_form.fields[var])
+except KeyError:
+pass
+item_form.addField(schema_form.fields[var])
+else:
+raise StanzaError('feature-not-implemented', text='unknown write restriction {}'.format(write_restriction))
+# we now remove every field which is not in data schema
+to_remove = set()
+for item_var, item_field in item_form.fields.iteritems():
+if item_var not in schema_form.fields:
+to_remove.add(item_field)
+for field in to_remove:
+item_form.removeField(field)
+item_elt.addChild(item_form.toElement())
 def _checkOverwrite(self, node, itemIdentifiers, publisher):
 """Check that the itemIdentifiers correspond to items published
 by the current publisher"""
 def doCheck(item_pub_map):
 for item_publisher in item_pub_map.itervalues():
 d = node.getItemsPublishers(itemIdentifiers)
 d.addCallback(doCheck)
 return d
 def publish(self, nodeIdentifier, items, requestor, pep, recipient):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(self._checkAuth, requestor)
 #FIXME: owner and publisher are not necessarly the same. So far we use only owner to get roster.
 #FIXME: in addition, there can be several owners: that is not managed yet
 d.addCallback(self._doPublish, items, requestor, pep, recipient)
 return d
 def _doPublish(self, result, items, requestor, pep, recipient):
 affiliation, node = result
 if node.nodeType == 'collection':
 raise error.NoPublishing()
 item["id"] = str(uuid.uuid4())
 else:
 check_overwrite = True
 access_model, item_config = self.parseItemConfig(item)
 categories = self.parseCategories(item)
+schema = node.getSchema()
+if schema is not None:
+self.enforceSchema(item, schema, affiliation)
 items_data.append(container.ItemData(item, access_model, item_config, categories))
 if persistItems:
 if check_overwrite and affiliation != 'owner':
 # we don't want a publisher to overwrite the item
 d.addCallback(self._doNotify, node, items_data,
 deliverPayloads, pep, recipient)
 return d
 def _doNotify(self, result, node, items_data, deliverPayloads, pep, recipient):
 if items_data and not deliverPayloads:
 for item_data in items_data:
 item_data.item.children = []
 self.dispatch({'items_data': items_data, 'node': node, 'pep': pep, 'recipient': recipient},
 '//event/pubsub/notify')
 def getNotifications(self, node, items_data):
 """Build a list of subscriber to the node
 subscribers will be associated with subscribed items,
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, subscriberEntity)
 d.addCallback(self._doSubscribe, subscriber)
 return d
 def _doSubscribe(self, result, subscriber):
 # TODO: implement other access models
 node, affiliation = result
 if affiliation == 'outcast':
 d = node.addSubscription(subscriber, 'subscribed', {})
 d.addCallbacks(lambda _: True, trapExists)
 d.addCallback(cb)
 return d
 def _sendLastPublished(self, subscription, node):
 def notifyItem(items):
 if items:
 d.addCallback(notifyItem)
 d.addErrback(log.err)
 return subscription
 def unsubscribe(self, nodeIdentifier, subscriber, requestor, pep, recipient):
 if subscriber.userhostJID() != requestor.userhostJID():
 return defer.fail(error.Forbidden())
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(lambda node: node.removeSubscription(subscriber))
 return d
 def getSubscriptions(self, entity):
 return self.storage.getSubscriptions(entity)
 def supportsAutoCreate(self):
 return True
 def supportsCreatorCheck(self):
 return True
 def supportsInstantNodes(self):
 return True
 def createNode(self, nodeIdentifier, requestor, options = None, pep=False, recipient=None):
 if not nodeIdentifier:
 nodeIdentifier = 'generic/%s' % uuid.uuid4()
 # TODO: handle schema on creation
 d = self.storage.createNode(nodeIdentifier, requestor, config, None, pep, recipient)
 d.addCallback(lambda _: nodeIdentifier)
 return d
 def getDefaultConfiguration(self, nodeType):
 d = defer.succeed(self.storage.getDefaultConfiguration(nodeType))
 return d
 def getNodeConfiguration(self, nodeIdentifier, pep, recipient):
 if not nodeIdentifier:
 return defer.fail(error.NoRootNode())
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(lambda node: node.getConfiguration())
 return d
 def setNodeConfiguration(self, nodeIdentifier, options, requestor, pep, recipient):
 if not nodeIdentifier:
 return defer.fail(error.NoRootNode())
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doSetNodeConfiguration, options)
 return d
 def _doSetNodeConfiguration(self, result, options):
 node, affiliation = result
 if affiliation != 'owner':
 raise error.Forbidden()
 return node.setConfiguration(options)
 def getNodeSchema(self, nodeIdentifier, pep, recipient):
 if not nodeIdentifier:
 return defer.fail(error.NoRootNode())
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doSetNodeSchema, schema)
 return d
 def _doSetNodeSchema(self, result, schema):
 node, affiliation = result
 if affiliation != 'owner':
 raise error.Forbidden()
 return node.setSchema(schema)
 def getAffiliations(self, entity, nodeIdentifier, pep, recipient):
 return self.storage.getAffiliations(entity, nodeIdentifier, pep, recipient)
 def getAffiliationsOwner(self, nodeIdentifier, requestor, pep, recipient):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doGetAffiliationsOwner)
 return d
 def _doGetAffiliationsOwner(self, result):
 node, affiliation = result
 if affiliation != 'owner':
 raise error.Forbidden()
 return node.getAffiliations()
 def setAffiliationsOwner(self, nodeIdentifier, requestor, affiliations, pep, recipient):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doSetAffiliationsOwner, requestor, affiliations)
 return d
 def _doSetAffiliationsOwner(self, result, requestor, affiliations):
 # Check that requestor is allowed to set affiliations, and delete entities
 # with "none" affiliation
 else:
 d = node.setAffiliations(affiliations)
 return d
 def getSubscriptionsOwner(self, nodeIdentifier, requestor, pep, recipient):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doGetSubscriptionsOwner)
 return d
 def _doGetSubscriptionsOwner(self, result):
 node, affiliation = result
 if affiliation != 'owner':
 raise error.Forbidden()
 return node.getSubscriptions()
 def setSubscriptionsOwner(self, nodeIdentifier, requestor, subscriptions, pep, recipient):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doSetSubscriptionsOwner, requestor, subscriptions)
 d = defer.gatherResults(d_list, consumeErrors=True)
 d.addCallback(lambda _: None)
 d.addErrback(self.unwrapFirstError)
 return d
+def filterItemsWithSchema(self, items_data, schema, owner):
+"""check schema restriction and remove fields/items if they don't comply
+@param items_data(list[ItemData]): items to filter
+items in this list will be modified
+@param schema(domish.Element): node schema
+@param owner(bool): True is requestor is a owner of the node
+"""
+fields_to_remove = set()
+for field_elt in schema.elements(data_form.NS_X_DATA, 'field'):
+for restrict_elt in field_elt.elements(const.NS_SCHEMA_RESTRICT, 'restrict'):
+read_restriction = restrict_elt.attributes.get('read')
+if read_restriction is not None:
+if read_restriction == 'owner':
+if not owner:
+fields_to_remove.add(field_elt['var'])
+else:
+raise StanzaError('feature-not-implemented', text='unknown read restriction {}'.format(read_restriction))
+items_to_remove = []
+for idx, item_data in enumerate(items_data):
+item_elt = item_data.item
+try:
+x_elt = next(item_elt.elements(data_form.NS_X_DATA, 'x'))
+except StopIteration:
+log.msg("WARNING, item {id} has a schema but no form, ignoring it")
+items_to_remove.append(item_data)
+continue
+form = data_form.Form.fromElement(x_elt)
+# we remove fields which are not visible for this user
+for field in fields_to_remove:
+try:
+form.removeField(form.fields[field])
+except KeyError:
+continue
+item_elt.children.remove(x_elt)
+item_elt.addChild(form.toElement())
+for item_data in items_to_remove:
+items_data.remove(item_data)
 @defer.inlineCallbacks
 def checkNodeAccess(self, node, requestor):
 """check if a requestor can access data of a node
 #FIXME
 raise NotImplementedError
 else:
 raise error.BadAccessTypeError(access_model)
+schema = node.getSchema()
+if schema is not None:
+self.filterItemsWithSchema(items_data, schema, owner)
 yield self._items_rsm(items_data, node, requestor_groups, owner, itemIdentifiers, ext_data)
 defer.returnValue(items_data)
 def _setCount(self, value, response):
 response.count = value
 if notify:
 d.addCallback(self._doNotifyRetraction, node, pep, recipient)
 return d
 def _doNotifyRetraction(self, items_data, node, pep, recipient):
 self.dispatch({'items_data': items_data,
 'node': node,
 'pep': pep,
 'recipient': recipient},
 '//event/pubsub/retract')
 def purgeNode(self, nodeIdentifier, requestor, pep, recipient):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doPurge)
 return d
 def _doPurge(self, result):
 node, affiliation = result
 persistItems = node.getConfiguration()[const.OPT_PERSIST_ITEMS]
 if affiliation != 'owner':
 d = node.purge()
 d.addCallback(self._doNotifyPurge, node.nodeIdentifier)
 return d
 def _doNotifyPurge(self, result, nodeIdentifier):
 self.dispatch(nodeIdentifier, '//event/pubsub/purge')
 def registerPreDelete(self, preDeleteFn):
 self._callbackList.append(preDeleteFn)
 def getSubscribers(self, nodeIdentifier, pep, recipient):
 def cb(subscriptions):
 return [subscription.subscriber for subscription in subscriptions]
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(lambda node: node.getSubscriptions('subscribed'))
 d.addCallback(cb)
 return d
 def deleteNode(self, nodeIdentifier, requestor, pep, recipient, redirectURI=None):
 d = self.storage.getNode(nodeIdentifier, pep, recipient)
 d.addCallback(_getAffiliation, requestor)
 d.addCallback(self._doPreDelete, redirectURI, pep, recipient)
 return d
 def _doPreDelete(self, result, redirectURI, pep, recipient):
 node, affiliation = result
 if affiliation != 'owner':
 raise error.Forbidden()
 d = defer.DeferredList([cb(data, pep, recipient)
 for cb in self._callbackList],
 consumeErrors=1)
 d.addCallback(self._doDelete, node.nodeDbId)
 def _doDelete(self, result, nodeDbId):
 dl = []
 for succeeded, r in result:
 if succeeded and r:
 d = self.storage.deleteNodeByDbId(nodeDbId)
 d.addCallback(self._doNotifyDelete, dl)
 return d
 def _doNotifyDelete(self, result, dl):
 for d in dl:
 d.callback(None)
 class PubSubResourceFromBackend(pubsub.PubSubResource):
 """
 Adapts a backend to an xmpp publish-subscribe service.
 self.features.append("groupblog")
 # if self.backend.supportsPublishModel():       #XXX: this feature is not really described in XEP-0060, we just can see it in examples
 #     self.features.append("publish_model")     #     but it's necessary for microblogging comments (see XEP-0277)
 def getFullItem(self, item_data):
 """ Attach item configuration to this item
 Used to give item configuration back to node's owner (and *only* to owner)
 """
 notifications_filtered)
 d = self._prepareNotify(items_data, node, data.get('subscription'), pep, recipient)
 d.addCallback(afterPrepare)
 return d
 @defer.inlineCallbacks
 def _prepareNotify(self, items_data, node, subscription=None, pep=None, recipient=None):
 """Do a bunch of permissions check and filter notifications
 # now we check access of subscriber for each item, and keep only allowed ones
 #we filter items not allowed for the subscribers
 notifications_filtered = []
+schema = node.getSchema()
 for subscriber, subscriptions, items_data in notifications:
 subscriber_bare = subscriber.userhostJID()
 if subscriber_bare in owners:
 # as notification is always sent to owner,
 # we ignore owner if he is here
 continue
 allowed_items = [] #we keep only item which subscriber can access
+if schema is not None:
+# we have to deepcopy items because different subscribers may receive
+# different items (e.g. read restriction in schema)
+items_data = deepcopy(items_data)
+self.backend.filterItemsWithSchema(items_data, schema, False)
 for item_data in items_data:
 item, access_model = item_data.item, item_data.access_model
 access_list = item_data.config
 if access_model == const.VAL_AMODEL_OPEN:
 d = self.backend.getNodes(requestor.userhostJID(),
 pep,
 service)
 return d.addErrback(self._mapErrors)
 def getConfigurationOptions(self):
 return self.backend.nodeOptions
 def _publish_errb(self, failure, request):
 if failure.type == error.NodeNotFound and self.backend.supportsAutoCreate():
 self._isPep(request),
 request.recipient)
 d.addErrback(self._publish_errb, request)
 return d.addErrback(self._mapErrors)
 def subscribe(self, request):
 d = self.backend.subscribe(request.nodeIdentifier,
 request.subscriber,
 request.sender,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def unsubscribe(self, request):
 d = self.backend.unsubscribe(request.nodeIdentifier,
 request.subscriber,
 request.sender,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def subscriptions(self, request):
 d = self.backend.getSubscriptions(self._isPep(request),
 request.sender)
 return d.addErrback(self._mapErrors)
 def affiliations(self, request):
 """Retrieve affiliation for normal entity (cf. XEP-0060 §5.7)
 retrieve all node where this jid is affiliated
 request.nodeIdentifier,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def create(self, request):
 d = self.backend.createNode(request.nodeIdentifier,
 request.sender, request.options,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def default(self, request):
 d = self.backend.getDefaultConfiguration(request.nodeType,
 self._isPep(request),
 request.sender)
 return d.addErrback(self._mapErrors)
 def configureGet(self, request):
 d = self.backend.getNodeConfiguration(request.nodeIdentifier,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def configureSet(self, request):
 d = self.backend.setNodeConfiguration(request.nodeIdentifier,
 request.options,
 request.sender,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def affiliationsGet(self, request):
 """Retrieve affiliations for owner (cf. XEP-0060 §8.9.1)
 retrieve all affiliations for a node
 """
 request.affiliations,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def subscriptionsGet(self, request):
 """Retrieve subscriptions for owner (cf. XEP-0060 §8.8.1)
 retrieve all affiliations for a node
 """
 request.sender,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def subscriptionsSet(self, request):
 d = self.backend.setSubscriptionsOwner(request.nodeIdentifier,
 request.sender,
 request.subscriptions,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def items(self, request):
 ext_data = {}
 if const.FLAG_ENABLE_RSM and request.rsm is not None:
 ext_data['rsm'] = request.rsm
 request.notify,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def purge(self, request):
 d = self.backend.purgeNode(request.nodeIdentifier,
 request.sender,
 self._isPep(request),
 request.recipient)
 return d.addErrback(self._mapErrors)
 def delete(self, request):
 d = self.backend.deleteNode(request.nodeIdentifier,
 request.sender,
 self._isPep(request),
 request.recipient)

Mercurial > libervia-pubsub

comparison sat_pubsub/backend.py @ 353:7c5d85c6fb3a