Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def parse_iq(stanza):
if not nbxmpp.isResultNode(stanza):
log.error('Error on MAM query: %s', stanza.getError())
raise InvalidMamIQ
fin = stanza.getTag('fin')
if fin is None:
log.error('Malformed MAM query result received: %s', stanza)
raise InvalidMamIQ
set_ = fin.getTag('set', namespace=nbxmpp.NS_RSM)
if set_ is None:
log.error(
'Malformed MAM query result received (no "set" Node): %s',
stanza)
raise InvalidMamIQ
return fin, set_
def _received_result(self, _nbxmpp_client, stanza):
data = None
is_dataform = False
if nbxmpp.isResultNode(stanza):
self._log.info('Received result from %s', stanza.getFrom())
tag = stanza.getTag('query', namespace=Namespace.SEARCH)
if tag is None:
self._log.info('Invalid stanza: %s', stanza)
return
data = tag.getTag('x', namespace=Namespace.DATA)
if data is not None:
is_dataform = True
else:
data = []
for item in tag.getTags('item'):
# We also show attributes. jid is there
field = item.attrs
for i in item.getPayload():
field[i.getName()] = i.getData()
def _metacontacts_received(self, _nbxmpp_client, stanza):
if not nbxmpp.isResultNode(stanza):
self._log.info('Request error: %s', stanza.getError())
else:
self.available = True
meta_list = self._parse_metacontacts(stanza)
self._log.info('Received: %s', meta_list)
app.nec.push_incoming_event(NetworkEvent(
'metacontacts-received', conn=self._con, meta_list=meta_list))
self._con.connect_machine()
def _on_register_result(result):
if not nbxmpp.isResultNode(result):
reason = result.getErrorMsg() or result.getError()
app.nec.push_incoming_event(AccountNotCreatedEvent(
None, conn=self, reason=reason))
return
if app.is_installed('GPG'):
self.USE_GPG = True
self.gpg = gpg.GnuPG()
app.nec.push_incoming_event(
AccountCreatedEvent(None, conn=self,
account_info=self.new_account_info))
self.new_account_info = None
self.new_account_form = None
if self.connection:
self.connection.UnregisterDisconnectHandler(
self._on_new_account)
self.disconnect(on_purpose=True)
def _cancel_result_received(self, _nbxmpp_client, stanza):
if not nbxmpp.isResultNode(stanza):
self._log.warning('Error: %s', stanza.getError())
else:
self._log.info('Cancel successful')
def _privacy_lists_received(self, _nbxmpp_client, stanza, callback):
lists = []
new_default = None
result = nbxmpp.isResultNode(stanza)
if not result:
self._log.warning('List not available: %s', stanza.getError())
else:
for list_ in stanza.getQueryPayload():
name = list_.getAttr('name')
if list_.getName() == 'active':
self.active_list = name
elif list_.getName() == 'default':
new_default = name
else:
lists.append(name)
self._log.info('Received lists: %s', lists)
# Download default list if we don’t have it
if self.default_list != new_default:
def _default_callback(self, _con, stanza, *args, **kwargs):
if not nbxmpp.isResultNode(stanza):
self._log.warning('Error: %s', stanza.getError())
def _pong_received(self,
_nbxmpp_client: Any,
stanza: nbxmpp.Iq,
ping_time: int,
contact: ContactsT) -> None:
if not nbxmpp.isResultNode(stanza):
self._log.info('Error: %s', stanza.getError())
app.nec.push_incoming_event(
PingErrorEvent(None, conn=self._con, contact=contact))
return
diff = round(time.time() - ping_time, 2)
self._log.info('Received pong from %s after %s seconds',
stanza.getFrom(), diff)
app.nec.push_incoming_event(
PingReplyEvent(None, conn=self._con,
contact=contact,
seconds=diff))
def _del_privacy_list_result(_nbxmpp_client, stanza):
if not nbxmpp.isResultNode(stanza):
self._log.warning('List deletion failed: %s', stanza.getError())
app.nec.push_incoming_event(InformationEvent(
None, dialog_name='privacy-list-error', args=name))
else:
app.nec.push_incoming_event(PrivacyListRemovedEvent(
None, conn=self._con, list_name=name))
def _received_pb_configuration(self, _nbxmpp_client, stanza, node):
if not nbxmpp.isResultNode(stanza):
self._log.warning('Error: %s', stanza.getError())
return
pubsub = stanza.getTag('pubsub', namespace=Namespace.PUBSUB_OWNER)
if pubsub is None:
self._log.warning('Malformed PubSub configure '
'stanza (no pubsub node): %s', stanza)
return
configure = pubsub.getTag('configure')
if configure is None:
self._log.warning('Malformed PubSub configure '
'stanza (no configure node): %s', stanza)
return
if configure.getAttr('node') != node: