GitLab

at 2020-08-26T21:39:15+02:00

Make JID object a namedtuple

Use new JID attributes/methods

Add JID Escaping (XEP-0106) support

Fixes #82

JID: Show punycode when using to_user_string()

         return self._jid

 

     def _set_bound_jid(self, jid):

-        self._jid = JID(jid)

+        self._jid = JID.from_string(jid)

 

     @property

     def has_error(self):

 

         if name == 'iq':

             if stanza.getFrom() is None and own_jid is not None:

-                stanza.setFrom(own_jid.getBare())

+                stanza.setFrom(own_jid.bare)

 

         if name == 'message':

             # https://tools.ietf.org/html/rfc6120#section-8.1.1.1

             to = stanza.getTo()

             if to is None:

                 stanza.setTo(own_jid)

-            elif not to.bareMatch(own_jid):

+

+            elif not to.bare_match(own_jid):

                 self._log.warning('Message addressed to someone else: %s',

                                   stanza)

                 return

 

             if stanza.getFrom() is None:

-                stanza.setFrom(own_jid.getBare())

+                stanza.setFrom(own_jid.bare)

 

             # Unwrap carbon

             try:

 

     def _create_client(self):

         self._client = Client(log_context='TEST')

-        self._client.set_domain(self.address.getDomain())

-        self._client.set_username(self.address.getNode())

+        self._client.set_domain(self.address.domain)

+        self._client.set_username(self.address.localpart)

         self._client.set_resource('test')

 

         proxy_ip = self._builder.proxy_ip.get_text()

 

     @property

     def address(self):

-        return JID(self._builder.address.get_text())

+        return JID.from_string(self._builder.address.get_text())

 

     @property

     def xml_box(self):

             if data.text:

                 item.addChild('text', payload=data.text)

 

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.ACTIVITY, item, id_='current')
     @call_on_response('_command_list_received')

     def request_command_list(self, jid=None):

         if jid is None:

-            jid = self._client.get_bound_jid().getBare()

+            jid = self._client.get_bound_jid().bare

         return get_disco_request(Namespace.DISCO_ITEMS,

                                  jid,

                                  node=Namespace.COMMANDS)

 

     @property

     def domain(self):

-        return self._client.get_bound_jid().getDomain()

+        return self._client.get_bound_jid().domain

 

     @call_on_response('_annotations_received')

     def request_annotations(self):

         notes = []

         for note in storage.getTags('note'):

             try:

-                jid = JID(note.getAttr('jid'))

+                jid = JID.from_string(note.getAttr('jid'))

             except Exception as error:

                 self._log.warning('Invalid JID: %s, %s',

                                   note.getAttr('jid'), error)

         confs = storage.getTags('conference')

         for conf in confs:

             try:

-                jid = JID(conf.getAttr('jid'))

+                jid = JID.from_string(conf.getAttr('jid'))

             except Exception as error:

                 self._log.warning('Invalid JID: "%s", %s',

                                   conf.getAttr('jid'),

             return None

 

         try:

-            jid = JID(item.getAttr('id'))

+            jid = JID.from_string(item.getAttr('id'))

         except Exception as error:

             self._log.warning('Invalid JID: "%s", %s',

                               item.getAttr('id'),

 

     @call_on_response('_bookmarks_received')

     def request_bookmarks(self, type_):

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         if type_ == BookmarkStoreType.PUBSUB_BOOKMARK_2:

             self._log.info('Request bookmarks 2 (PubSub)')

             request = get_pubsub_request(jid, Namespace.BOOKMARKS_2)

 

         from_ = stanza.getFrom()

         if from_ is None:

-            from_ = self._client.get_bound_jid().getBare()

+            from_ = self._client.get_bound_jid().bare

         self._log.info('Received bookmarks from: %s', from_)

         for bookmark in bookmarks:

             self._log.info(bookmark)

 

     def retract_bookmark(self, bookmark_jid):

         self._log.info('Retract Bookmark: %s', bookmark_jid)

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').retract(jid,

                                                   Namespace.BOOKMARKS_2,

                                                   str(bookmark_jid))

 

     def _store_bookmark_1(self, bookmarks):

         self._log.info('Store Bookmarks 1 (PubSub)')

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._bookmark_1_queue = bookmarks

         item = self._build_storage_node(bookmarks)

         options = get_publish_options(BOOKMARK_1_OPTIONS)

             return

 

         self._log.info('Store Bookmarks 2 (PubSub)')

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         for bookmark in bookmarks:

             self._bookmark_2_queue[bookmark.jid] = bookmark

             item = self._build_conference_node(bookmark)

                 return

 

             self._node_configuration_in_progress = True

-            jid = self._client.get_bound_jid().getBare()

+            jid = self._client.get_bound_jid().bare

             self._client.get_module('PubSub').get_node_configuration(

                 jid,

                 node,

     def is_valid(self):

         if self.value:

             try:

-                JID(self.value)

+                JID.from_string(self.value)

                 return True, ''

             except Exception as error:

                 return False, error

         if self.values:

             for value in self.values:

                 try:

-                    JID(value)

+                    JID.from_string(value)

                 except Exception as error:

                     return False, error

             return True, ''

             # to indicate when the user has set the subject,

             # the 'from' attr on these delays is the MUC server

             # but we treat it as user timestamp

-            jids = [properties.jid.getBare(),

-                    properties.jid.getDomain()]

+            jids = [properties.jid.bare,

+                    properties.jid.domain]

 

             properties.user_timestamp = parse_delay(stanza, from_=jids)

 

         else:

             if properties.from_muc:

                 # Some servers use the MUC JID, others the domain

-                jids = [properties.jid.getBare(),

-                        properties.jid.getDomain()]

+                jids = [properties.jid.bare,

+                        properties.jid.domain]

             else:

-                jids = [self._client.get_bound_jid().getDomain()]

+                jids = [self._client.get_bound_jid().domain]

 

             server_delay = parse_delay(stanza, from_=jids)

             if server_delay is not None:

                 if value is not None:

                     item.addChild(tag, payload=value)

 

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.LOCATION, item, id_='current')
                 continue

 

             try:

-                jid = JID(jid)

+                jid = JID.from_string(jid)

             except Exception:

                 self._log.warning('Invalid jid found in preferences: %s',

                                   jid)

 

         elif properties.is_mam_message and not properties.type.is_groupchat:

             own_jid = self._client.get_bound_jid()

-            if own_jid.bareMatch(stanza.getFrom()):

+            if own_jid.bare_match(stanza.getFrom()):

                 properties.jid = stanza.getTo()

             else:

                 properties.jid = stanza.getFrom()

     def _parse_self_message(stanza, properties):

         if properties.type.is_groupchat:

             return False

-        return stanza.getFrom().bareMatch(stanza.getTo())
+        return stanza.getFrom().bare_match(stanza.getTo())
             return stanza, None

 

     # Carbon must be from our bare jid

-    if not stanza.getFrom() == own_jid.getBare():

+    if stanza.getFrom() != own_jid.new_as_bare():

         raise InvalidFrom('Invalid from: %s' % stanza.getAttr('from'))

 

     forwarded = carbon.getTag('forwarded', namespace=Namespace.FORWARD)

     # Fill missing to/from

     to = message.getTo()

     if to is None:

-        message.setTo(own_jid.getBare())

+        message.setTo(own_jid.bare)

 

     frm = message.getFrom()

     if frm is None:

-        message.setFrom(own_jid.getBare())

+        message.setFrom(own_jid.bare)

 

     if type_ == 'received':

-        if message.getFrom().bareMatch(own_jid):

+        if message.getFrom().bare_match(own_jid):

             # Drop 'received' Carbons from ourself, we already

             # got the message with the 'sent' Carbon or via the

             # message itself

     # Fill missing to/from

     to = message.getTo()

     if to is None:

-        message.setTo(own_jid.getBare())

+        message.setTo(own_jid.bare)

 

     frm = message.getFrom()

     if frm is None:

-        message.setFrom(own_jid.getBare())

+        message.setFrom(own_jid.bare)

 

     # Timestamp parsing

     # Most servers dont set the 'from' attr, so we cant check for it

             if data.text:

                 item.addChild('text', payload=data.text)

 

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.MOOD, item, id_='current')
         if muc is None:

             return

         properties.from_muc = True

-        properties.muc_jid = properties.jid.copy()

-        properties.muc_jid.setBare()

-        properties.muc_nickname = properties.jid.getResource()

+        properties.muc_jid = properties.jid.new_as_bare()

+        properties.muc_nickname = properties.jid.resource

 

     def _process_muc_user_presence(self, _client, stanza, properties):

         muc_user = stanza.getTag('x', namespace=Namespace.MUC_USER)

         if muc_user is None:

             return

         properties.from_muc = True

-        properties.muc_jid = properties.jid.copy()

-        properties.muc_jid.setBare()

+        properties.muc_jid = properties.jid.new_as_bare()

 

         destroy = muc_user.getTag('destroy')

         if destroy is not None:

             alternate = destroy.getAttr('jid')

             if alternate is not None:

                 try:

-                    alternate = JID(alternate)

+                    alternate = JID.from_string(alternate)

                 except Exception as error:

                     self._log.warning('Invalid alternate JID provided: %s',

                                       error)

                 password=muc_user.getTagData('password'))

             return

 

-        properties.muc_nickname = properties.jid.getResource()

+        properties.muc_nickname = properties.jid.resource

 

         # https://xmpp.org/extensions/xep-0045.html#registrar-statuscodes

         message_status_codes = [

 

     def _process_groupchat_message(self, _client, stanza, properties):

         properties.from_muc = True

-        properties.muc_jid = properties.jid.copy()

-        properties.muc_jid.setBare()

-        properties.muc_nickname = properties.jid.getResource() or None

+        properties.muc_jid = properties.jid.new_as_bare()

+        properties.muc_nickname = properties.jid.resource

 

         muc_user = stanza.getTag('x', namespace=Namespace.MUC_USER)

         if muc_user is not None:

         if addresses is not None:

             address = addresses.getTag('address', attrs={'type': 'ofrom'})

             if address is not None:

-                properties.muc_ofrom = JID(address.getAttr('jid'))

+                properties.muc_ofrom = JID.from_string(address.getAttr('jid'))

 

     def _process_message(self, _client, stanza, properties):

         muc_user = stanza.getTag('x', namespace=Namespace.MUC_USER)

             return

 

         properties.from_muc = True

-        properties.muc_jid = properties.jid.copy()

-        properties.muc_jid.setBare()

+        properties.muc_jid = properties.jid.new_as_bare()

 

-        if not properties.jid.isBare:

+        if not properties.jid.is_bare:

             return

 

         # MUC Config change

             return

 

         data = {}

-        data['muc'] = JID(direct.getAttr('jid'))

+        data['muc'] = JID.from_string(direct.getAttr('jid'))

         data['from_'] = properties.jid

         data['reason'] = direct.getAttr('reason')

         data['password'] = direct.getAttr('password')

             return

 

         properties.from_muc = True

-        properties.muc_jid = properties.jid.copy()

-        properties.muc_jid.setBare()

+        properties.muc_jid = properties.jid.new_as_bare()

 

         data = {}

 

         invite = muc_user.getTag('invite')

         if invite is not None:

-            data['muc'] = JID(properties.jid.getBare())

-            data['from_'] = JID(invite.getAttr('from'))

+            data['muc'] = properties.jid.new_as_bare()

+            data['from_'] = JID.from_string(invite.getAttr('from'))

             data['reason'] = invite.getTagData('reason')

             data['password'] = muc_user.getTagData('password')

             data['type'] = InviteType.MEDIATED

 

         decline = muc_user.getTag('decline')

         if decline is not None:

-            data['muc'] = JID(properties.jid.getBare())

-            data['from_'] = JID(decline.getAttr('from'))

+            data['muc'] = properties.jid.new_as_bare()

+            data['from_'] = JID.from_string(decline.getAttr('from'))

             data['reason'] = decline.getTagData('reason')

             properties.muc_decline = DeclineData(**data)

             return

         nick = data_form['muc#roomnick'].value

 

         try:

-            jid = JID(data_form['muc#jid'].value)

+            jid = JID.from_string(data_form['muc#jid'].value)

         except Exception:

             self._log.warning('Invalid JID on voice request')

             self._log.warning(stanza)

                                                 nick=nick,

                                                 form=data_form)

         properties.from_muc = True

-        properties.muc_jid = properties.jid.copy()

-        properties.muc_jid.setBare()

+        properties.muc_jid = properties.jid.new_as_bare()

 

     def approve_voice_request(self, muc_jid, voice_request):

         form = voice_request.form

         users_dict = {}

         for item in items:

             try:

-                jid = JID(item.getAttr('jid'))

+                jid = JID.from_string(item.getAttr('jid'))

             except Exception as error:

                 self._log.warning('Invalid JID: %s, %s',

                                   item.getAttr('jid'), error)

         jid = item_dict.get('jid')

         if jid is not None:

             try:

-                jid = JID(jid)

+                jid = JID.from_string(jid)

             except InvalidJid as error:

                 raise StanzaMalformed('invalid jid %s, %s' % (jid, error))

 

         item = Node('nick', {'xmlns': Namespace.NICK})

         if nickname is not None:

             item.addData(nickname)

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.NICK, item, id_='current')
             item.addChild('device').setAttr('id', device)

 

         self._log.info('Set devicelist: %s', devicelist)

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.OMEMO_TEMP_DL, item, id_='current')

 

     @call_on_response('_devicelist_received')

     def request_devicelist(self, jid=None):

         if jid is None:

-            jid = self._client.get_bound_jid().getBare()

+            jid = self._client.get_bound_jid().bare

         self._log.info('Request devicelist from: %s', jid)

         return get_pubsub_request(jid, Namespace.OMEMO_TEMP_DL, max_items=1)

 

         self._log.info('Set bundle')

 

         node = '%s:%s' % (Namespace.OMEMO_TEMP_BUNDLE, device_id)

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, node, item, id_='current')

 

                 item.addChild('pubkey-metadata', attrs=attrs)

 

         self._log.info('Set keylist: %s', keylist)

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.OPENPGP_PK, item, id_='current')

 

         node = '%s:%s' % (Namespace.OPENPGP_PK, fingerprint)

 

         self._log.info('Set public key')

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, node, item, id_='current')

 

 

     @callback

     def _public_key_received(self, stanza):

-        jid = JID(stanza.getFrom().getBare())

+        jid = stanza.getFrom().bare

 

         if not isResultNode(stanza):

             return raise_error(self._log.info, stanza)

 

     @callback

     def _keylist_received(self, stanza):

-        jid = JID(stanza.getFrom().getBare())

+        jid = stanza.getFrom().bare

 

         if not isResultNode(stanza):

             return raise_error(self._log.info, stanza)

     @call_on_response('_secret_key_received')

     def request_secret_key(self):

         self._log.info('Request secret key')

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         return get_pubsub_request(jid, Namespace.OPENPGP_SK, max_items=1)

 

     @callback

             item.setData(b64encode(secret_key))

 

         self._log.info('Set secret key')

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.OPENPGP_SK, item, id_='current')

 

             stanza.delChild(node)

 

     signcrypt = Node('signcrypt', attrs={'xmlns': Namespace.OPENPGP})

-    signcrypt.addChild('to', attrs={'jid': stanza.getTo().getBare()})

+    signcrypt.addChild('to', attrs={'jid': str(stanza.getTo().bare)})

 

     timestamp = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())

     signcrypt.addChild('time', attrs={'stamp': timestamp})

 

         own_jid = self._client.get_bound_jid()

         properties.self_presence = own_jid == properties.jid

-        properties.self_bare = properties.jid.bareMatch(own_jid)

+        properties.self_bare = properties.jid.bare_match(own_jid)

 

     def _parse_priority(self, stanza):

         priority = stanza.getPriority()

 

     is_muc_pm = isMucPM(stanza)

 

-    jid = stanza.getFrom().copy()

+    jid = stanza.getFrom()

     typ = stanza.getType()

     if typ == 'groupchat' or not is_muc_pm:

-        jid.setBare()

+        jid = jid.new_as_bare()

 

     message = Message(to=jid, typ=typ)

     if is_muc_pm:

 

     @call_on_response('_on_password_change')

     def change_password(self, password):

-        domain = self._client.get_bound_jid().getDomain()

-        username = self._client.get_bound_jid().getNode()

+        domain = self._client.get_bound_jid().domain

+        username = self._client.get_bound_jid().localpart

         iq = Iq('set', Namespace.REGISTER, to=domain)

         query = iq.getQuery()

         query.setTagData('username', username)

 

     @call_on_response('_default_response')

     def change_password_with_form(self, form):

-        domain = self._client.get_bound_jid().getDomain()

+        domain = self._client.get_bound_jid().domain

         iq = Iq('set', Namespace.REGISTER, to=domain)

         iq.setQueryPayload(form)

         return iq

                 if value is not None:

                     item.addChild(tag, payload=value)

 

-        jid = self._client.get_bound_jid().getBare()

+        jid = self._client.get_bound_jid().bare

         self._client.get_module('PubSub').publish(

             jid, Namespace.TUNE, item, id_='current')
     def _avatar_data_received(self, stanza):

         jid = stanza.getFrom()

         if jid is None:

-            jid = JID(self._client.get_bound_jid().getBare())

+            jid = self._client.get_bound_jid().bare

 

         if not isResultNode(stanza):

             return raise_error(self._log.warning, stanza)

 

         if name == 'iq':

             if stanza.getFrom() is None and own_jid is not None:

-                stanza.setFrom(own_jid.getBare())

+                stanza.setFrom(own_jid.bare)

 

         if name == 'message':

             # https://tools.ietf.org/html/rfc6120#section-8.1.1.1

             to = stanza.getTo()

             if to is None:

                 stanza.setTo(own_jid)

-            elif not to.bareMatch(own_jid):

+

+            elif not to.bare_match(own_jid):

                 log.warning('Message addressed to someone else: %s', stanza)

                 return

 

             if stanza.getFrom() is None:

-                stanza.setFrom(own_jid.getBare())

+                stanza.setFrom(own_jid.bare)

 

             # Unwrap carbon

             try:

 import time

 import hashlib

 import functools

+import warnings

 from base64 import b64encode

+from collections import namedtuple

 

 from gi.repository import GLib

 

 }

 

 _localpart_disallowed_chars = set('"&\'/:<>@')

+_localpart_escape_chars = ' "&\'/:<>@'

 

 

 STREAM_NOT_AUTHORIZED = 'urn:ietf:params:xml:ns:xmpp-streams not-authorized'

                      'xml-not-well-formed': XMLNotWellFormed}

 

 

-@functools.lru_cache(maxsize=None)

-def parse_jid(jid):

-    # https://tools.ietf.org/html/rfc7622#section-3.2

-

-    # Remove any portion from the first '/' character to the end of the

-    # string (if there is a '/' character present).

-

-    # Remove any portion from the beginning of the string to the first

-    # '@' character (if there is an '@' character present).

-

-    if jid.find('/') != -1:

-        rest, resourcepart = jid.split('/', 1)

-        resourcepart = validate_resourcepart(resourcepart)

-    else:

-        rest, resourcepart = jid, ''

-

-    if rest.find('@') != -1:

-        localpart, domainpart = rest.split('@', 1)

-        localpart = validate_localpart(localpart)

-    else:

-        localpart, domainpart = '', rest

+def deprecation_warning(message):

+    warnings.warn(message, DeprecationWarning)

 

-    domainpart = validate_domainpart(domainpart)

-    return localpart, domainpart, resourcepart

 

-

-@functools.lru_cache(maxsize=32768)

+@functools.lru_cache(maxsize=None)

 def validate_localpart(localpart):

     if not localpart or len(localpart.encode()) > 1023:

         raise LocalpartByteLimit

         raise LocalpartNotAllowedChar

 

 

-@functools.lru_cache(maxsize=32768)

+@functools.lru_cache(maxsize=None)

 def validate_resourcepart(resourcepart):

     if not resourcepart or len(resourcepart.encode()) > 1023:

         raise ResourcepartByteLimit

         raise ResourcepartNotAllowedChar

 

 

-@functools.lru_cache(maxsize=4096)

+@functools.lru_cache(maxsize=None)

 def validate_domainpart(domainpart):

     if not domainpart:

         raise DomainpartByteLimit

         domainpart = domainpart[:-1]

 

     try:

-        idna.encode(domainpart, uts46=True)

+        idna_encode(domainpart)

     except Exception:

         raise DomainpartNotAllowedChar

 

     return domainpart

 

 

-class JID:

-    """

-    JID can be built from string, modified, compared, serialised into string

-    """

+@functools.lru_cache(maxsize=None)

+def idna_encode(domain):

+    return idna.encode(domain, uts46=True).decode()

 

-    def __init__(self, jid=None, node='', domain='', resource=''):

-        """

-        JID can be specified as string (jid argument) or as separate parts

 

-        Examples:

-        JID('node@domain/resource')

-        JID(node='node',domain='domain.org')

-        """

-        if not jid and not domain:

-            raise ValueError('JID must contain at least domain name')

+@functools.lru_cache(maxsize=None)

+def escape_localpart(localpart):

+    # https://xmpp.org/extensions/xep-0106.html#bizrules-algorithm

+    #

+    # If there are any instances of character sequences that correspond

+    # to escapings of the disallowed characters

+    # (e.g., the character sequence "\27") or the escaping character

+    # (i.e., the character sequence "\5c") in the source address,

+    # the leading backslash character MUST be escaped to the character

+    # sequence "\5c"

+

+    for char in '\\' + _localpart_escape_chars:

+        seq = "\\{:02x}".format(ord(char))

+        localpart = localpart.replace(seq, "\\5c{:02x}".format(ord(char)))

+

+    # Escape all other chars

+    for char in _localpart_escape_chars:

+        localpart = localpart.replace(char, "\\{:02x}".format(ord(char)))

+

+    return localpart

+

+

+@functools.lru_cache(maxsize=None)

+def unescape_localpart(localpart):

+    if localpart.startswith('\\20') or localpart.endswith('\\20'):

+        # Escaped JIDs are not allowed to start or end with \20

+        # so this localpart must be already unescaped

+        return localpart

+

+    for char in _localpart_escape_chars:

+        seq = "\\{:02x}".format(ord(char))

+        localpart = localpart.replace(seq, char)

+

+    for char in _localpart_escape_chars + "\\":

+        seq = "\\5c{:02x}".format(ord(char))

+        localpart = localpart.replace(seq, "\\{:02x}".format(ord(char)))

+

+    return localpart

+

+

+class JID(namedtuple('JID',

+                     ['jid', 'localpart', 'domain', 'resource'])):

+

+    __slots__ = []

+

+    def __new__(cls, jid=None, localpart=None, domain=None, resource=None):

+        if jid is not None:

+            deprecation_warning('JID(jid) is deprecated, use from_string()')

+            return JID.from_string(str(jid))

+

+        if localpart is not None:

+            localpart = validate_localpart(localpart)

+

+        domain = validate_domainpart(domain)

+

+        if resource is not None:

+            resource = validate_resourcepart(resource)

+

+        return super().__new__(cls, None, localpart, domain, resource)

 

-        if isinstance(jid, JID):

-            self.node, self.domain = jid.node, jid.domain

-            self.resource = jid.resource

-        elif domain:

-            self.node, self.domain, self.resource = node, domain, resource

+    @classmethod

+    @functools.lru_cache(maxsize=None)

+    def from_string(cls, jid_string):

+        # https://tools.ietf.org/html/rfc7622#section-3.2

+

+        # Remove any portion from the first '/' character to the end of the

+        # string (if there is a '/' character present).

+

+        # Remove any portion from the beginning of the string to the first

+        # '@' character (if there is an '@' character present).

+

+        if jid_string.find('/') != -1:

+            rest, resourcepart = jid_string.split('/', 1)

         else:

-            self.node, self.domain, self.resource = parse_jid(jid)

+            rest, resourcepart = jid_string, None

 

-    def copy(self):

-        return JID(self)

+        if rest.find('@') != -1:

+            localpart, domainpart = rest.split('@', 1)

+        else:

+            localpart, domainpart = None, rest

 

-    def getNode(self):

-        """

-        Return the node part of the JID

-        """

-        return self.node

+        return cls(jid=None,

+                   localpart=localpart,

+                   domain=domainpart,

+                   resource=resourcepart)

 

-    def setNode(self, node):

-        """

-        Set the node part of the JID to new value. Specify None to remove

-        the node part

-        """

-        self.node = node.lower()

+    @classmethod

+    @functools.lru_cache(maxsize=None)

+    def from_user_input(cls, user_input, escaped=False):

+        # Use this if we want JIDs to be escaped according to XEP-0106

+        # The standard JID parsing cannot be applied because user_input is

+        # not a valid JID.

 

-    def getDomain(self):

-        """

-        Return the domain part of the JID

-        """

-        return self.domain

+        # Only user_input which after escaping result in a bare JID can be

+        # successfully parsed.

 

-    def setDomain(self, domain):

-        """

-        Set the domain part of the JID to new value

-        """

-        self.domain = domain.lower()

+        # The assumpution is user_input is a bare JID so we start with an

+        # rsplit on @ because we assume there is no resource, so the char @

+        # in the localpart can later be escaped.

 

-    def getResource(self):

-        """

-        Return the resource part of the JID

-        """

-        return self.resource

+        if escaped:

+            # for convenience

+            return cls.from_string(user_input)

 

-    def setResource(self, resource):

-        """

-        Set the resource part of the JID to new value.

-        Specify None to remove the resource part

-        """

-        self.resource = resource

+        if '@' in user_input:

+            localpart, domainpart = user_input.rsplit('@', 1)

+            if localpart.startswith(' ') or localpart.endswith(' '):

+                raise LocalpartNotAllowedChar

 

-    def getStripped(self):

-        """

-        Return the bare representation of JID. I.e. string value w/o resource

-        """

-        if not self.node:

-            return self.domain

-        return '%s@%s' % (self.node, self.domain)

+            localpart = escape_localpart(localpart)

 

-    def getBare(self):

-        """

-        Return the bare representation of JID. I.e. string value w/o resource

-        """

-        if not self.node:

-            return self.domain

-        return '%s@%s' % (self.node, self.domain)

+        else:

+            localpart = None

+            domainpart = user_input

 

-    def setBare(self):

-        """

-        Deletes the resource which makes the JID bare

-        """

-        self.resource = ''

+        return cls(jid=None,

+                   localpart=localpart,

+                   domain=domainpart,

+                   resource=None)

 

-    @property

-    def isBare(self):

-        return self.node and self.domain and not self.resource

+    def __str__(self):

+        if self.localpart:

+            jid = f'{self.localpart}@{self.domain}'

+        else:

+            jid = self.domain

 

-    @property

-    def isDomain(self):

-        return self.domain and not self.node and not self.resource

+        if self.resource is not None:

+            return f'{jid}/{self.resource}'

+        return jid

 

-    @property

-    def isFull(self):

-        return bool(self.domain and self.node and self.resource)

+    def __hash__(self):

+        return hash(str(self))

 

     def __eq__(self, other):

-        """

-        Compare the JID to another instance or to string for equality

-        """

-        if not isinstance(other, JID):

+        if isinstance(other, str):

+            deprecation_warning('comparing string with JID is deprected')

             try:

-                other = JID(other)

+                return JID.from_string(other) == self

             except Exception:

                 return False

-

-        return (self.node == other.node and

-                self.domain == other.domain and

-                self.resource == other.resource)

+        return super().__eq__(other)

 

     def __ne__(self, other):

-        """

-        Compare the JID to another instance or to string for non-equality

-        """

         return not self.__eq__(other)

 

-    def bareMatch(self, other):

-        """

-        Compare the node and domain parts of the JID's for equality

-        """

-        if not isinstance(other, JID):

-            try:

-                other = JID(other)

-            except Exception:

-                return False

+    def domain_to_ascii(self):

+        return idna_encode(self.domain)

 

-        return (self.node == other.node and

-                self.domain == other.domain)

+    @property

+    def bare(self):

+        if self.localpart is not None:

+            return f'{self.localpart}@{self.domain}'

+        return self.domain

 

-    def __str__(self):

-        """

-        Serialise JID into string

-        """

-        if self.node:

-            jid = '%s@%s' % (self.node, self.domain)

+    @property

+    def is_bare(self):

+        return self.resource is None

+

+    def new_as_bare(self):

+        if self.resource is None:

+            return self

+        return self._replace(resource=None)

+

+    def bare_match(self, other):

+        if isinstance(other, str):

+            other = JID.from_string(other)

+        return self.bare == other.bare

+

+    @property

+    def is_domain(self):

+        return self.localpart is None and self.resource is None

+

+    @property

+    def is_full(self):

+        return (self.localpart is not None and

+                self.domain is not None and

+                self.resource is not None)

+

+    def new_with(self, **kwargs):

+        return self._replace(**kwargs)

+

+    def to_user_string(self, show_punycode=True):

+        domain = self.domain_to_ascii()

+        if domain.startswith('xn--') and show_punycode:

+            domain_encoded = f' ({domain})'

         else:

-            jid = self.domain

+            domain_encoded = ''

 

-        if self.resource:

-            return '%s/%s' % (jid, self.resource)

-        return jid

+        if self.localpart is None:

+            return f'{self}{domain_encoded}'

 

-    def __hash__(self):

-        """

-        Produce hash of the JID, Allows to use JID objects as keys of the

-        dictionary

-        """

-        return hash(str(self))

+        localpart = unescape_localpart(self.localpart)

+

+        if self.resource is None:

+            return f'{localpart}@{self.domain}{domain_encoded}'

+        return f'{localpart}@{self.domain}/{self.resource}{domain_encoded}'

+

+    def bareMatch(self, other):

+        deprecation_warning('bareMatch() is deprected use bare_match()')

+        return self.bare_match(other)

+

+    @property

+    def isBare(self):

+        deprecation_warning('isBare() is deprected use '

+                            'the attribute is_bare')

+        return self.is_bare

+

+    @property

+    def isDomain(self):

+        deprecation_warning('isDomain() is deprected use '

+                            'the attribute is_domain')

+        return self.is_domain

 

-    def __repr__(self):

-        return str(self)

+    @property

+    def isFull(self):

+        deprecation_warning('isFull() is deprected use '

+                            'the attribute is_full')

+        return self.is_full

+

+    def copy(self):

+        deprecation_warning('copy() is not needed, JID is immutable')

+        return self

+

+    def getNode(self):

+        deprecation_warning('getNode() is deprected use '

+                            'the attribute localpart')

+        return self.localpart

+

+    def getDomain(self):

+        deprecation_warning('getDomain() is deprected use '

+                            'the attribute domain')

+        return self.domain

+

+    def getResource(self):

+        deprecation_warning('getResource() is deprected use '

+                            'the attribute resource')

+        return self.resource

+

+    def getStripped(self):

+        deprecation_warning('getStripped() is deprected use '

+                            'the attribute bare')

+        return self.bare

+

+    def getBare(self):

+        deprecation_warning('getBare() is deprected use '

+                            'the attribute bare')

+        return self.bare

 

 

 class StreamErrorNode(Node):

         """

         attr = self.getAttr('jid')

         if attr:

-            return JID(attr)

+            return JID.from_string(attr)

         return attr

 

     def getID(self):

         """

         Set the value of the 'to' attribute

         """

-        self.setAttr('to', JID(val))

+        if not isinstance(val, JID):

+            val = JID.from_string(val)

+        self.setAttr('to', val)

 

     def getType(self):

         """

         """

         Set the value of the 'from' attribute

         """

-        self.setAttr('from', JID(val))

+        if not isinstance(val, JID):

+            val = JID.from_string(val)

+        self.setAttr('from', val)

 

     def setType(self, val):

         """

         Set the item 'item' to the value 'val'

         """

         if item in ['to', 'from']:

-            val = JID(val)

+            if not isinstance(val, JID):

+                val = JID.from_string(val)

         return self.setAttr(item, val)

 

 

             return self.muc_identity_name

 

         if self.jid is not None:

-            return self.jid.getNode()

+            return self.jid.localpart

         return None

 

     @property

     def new_jid(self):

         if not self.is_nickname_changed:

             raise ValueError('This is not a nickname change')

-        jid = JID(self.jid)

-        jid.setResource(self.muc_user.nick)

-        return jid

+        return self.jid.new_with(resource=self.muc_user.nick)

 

     @property

     def is_kicked(self):

         self.client.is_websocket = False

         self.dispatcher = StanzaDispatcher(self.client)

 

-        self.client.get_bound_jid.return_value = JID('test@test.test')

+        self.client.get_bound_jid.return_value = JID.from_string('test@test.test')

 

         self.dispatcher.reset_parser()

         self.dispatcher.process_data(STREAM_START)
         def _on_message(_con, _stanza, properties):

 

             data = [

-                BookmarkData(jid=JID('theplay@conference.shakespeare.lit'),

+                BookmarkData(jid=JID.from_string('theplay@conference.shakespeare.lit'),

                              name='The Play\'s the Thing',

                              autojoin=True,

                              password='pass',

                              nick='JC'),

-                BookmarkData(jid=JID('second@conference.shakespeare.lit'),

+                BookmarkData(jid=JID.from_string('second@conference.shakespeare.lit'),

                              name='Second room',

                              autojoin=False,

                              password=None,

     def test_bookmark_2_parsing(self):

         def _on_message(_con, _stanza, properties):

 

-            data = BookmarkData(jid=JID('theplay@conference.shakespeare.lit'),

+            data = BookmarkData(jid=JID.from_string('theplay@conference.shakespeare.lit'),

                                 name='The Play\'s the Thing',

                                 autojoin=True,

                                 password=None,

 import unittest

 

 from nbxmpp.protocol import Iq

+from nbxmpp.protocol import JID

 from nbxmpp.util import error_factory

 

 

         self.assertEqual(error.get_text('de'), 'File zu groß. Erlaubt sind 20000 bytes')

         self.assertEqual(error.type, 'modify')

         self.assertEqual(error.id, 'step_03')

-        self.assertEqual(error.jid, 'upload.montague.tld')
+        self.assertEqual(error.jid, JID.from_string('upload.montague.tld'))
         ]

 

         for jid in tests:

-            JID(jid)

+            JID.from_string(jid)

 

     def test_invalid_jids(self):

         tests = [

 

         for jid, exception in tests:

             with self.assertRaises(exception):

-                JID(jid)

+                JID.from_string(jid)

 

     def test_ip_literals(self):

         tests = [

         ]

 

         for jid in tests:

-            JID(jid)

+            JID.from_string(jid)

 

     def test_jid_equality(self):

         tests = [

         ]

 

         for jid in tests:

-            self.assertTrue(JID(jid) == jid)
+            self.assertTrue(JID.from_string(jid) == JID.from_string(jid))

+

+    def test_jid_escaping(self):

+        # (user input, escaped)

+        tests = [

+            (r'space cadet@example.com',

+             r'space\20cadet@example.com'),

+

+            (r'call me "ishmael"@example.com',

+             r'call\20me\20\22ishmael\22@example.com'),

+

+            (r'at&t guy@example.com',

+             r'at\26t\20guy@example.com'),

+

+            ('d\'artagnan@example.com',

+             r'd\27artagnan@example.com'),

+

+            (r'/.fanboy@example.com',

+             r'\2f.fanboy@example.com'),

+

+            (r'::foo::@example.com',

+             r'\3a\3afoo\3a\3a@example.com'),

+

+            (r'<foo>@example.com',

+             r'\3cfoo\3e@example.com'),

+

+            (r'user@host@example.com',

+             r'user\40host@example.com'),

+

+            (r'c:\net@example.com',

+             r'c\3a\net@example.com'),

+

+            (r'c:\\net@example.com',

+             r'c\3a\\net@example.com'),

+

+            (r'c:\cool stuff@example.com',

+             r'c\3a\cool\20stuff@example.com'),

+

+            (r'c:\5commas@example.com',

+             r'c\3a\5c5commas@example.com'),

+

+            (r'call me\20@example.com',

+             r'call\20me\5c20@example.com'),

+        ]

+

+        tests2 = [

+            'juliet@example.com',

+            'juliet@example.com',

+            'juliet@example.com',

+            'juliet@example.com',

+            'fussball@example.com',

+            'fu\U000000DFball@example.com',

+            '\U000003C0@example.com',

+            '\U000003A3@example.com',

+            '\U000003C3@example.com',

+            '\U000003C2@example.com',

+            'example.com',

+        ]

+

+        test3 = [

+            '\\20callme\\20@example.com',

+            '\\20callme@example.com',

+            'callme\\20@example.com',

+        ]

+

+        test4 = [

+            ('call\\20me@example.com', 'call me@example.com',)

+        ]

+

+        fail_tests = [

+            r'c:\5commas@example.com/asd',

+            r'juliet@example.com/test'

+        ]

+

+        for user_input, escaped in tests:

+            # Parse user input and escape it

+            jid = JID.from_user_input(user_input)

+

+            self.assertTrue(jid.domain == 'example.com')

+            self.assertTrue(str(jid) == escaped)

+            self.assertTrue(jid.to_user_string() == user_input)

+

+            # We must fail on invalid JIDs

+            with self.assertRaises(Exception):

+                JID.from_string(user_input)

+

+            # Parse escaped JIDs

+            jid = JID.from_string(escaped)

+            self.assertTrue(str(jid) == escaped)

+            self.assertTrue(jid.domain == 'example.com')

+

+        for jid in tests2:

+            # show that from_string() and from_user_input() produce the same

+            # result for valid bare JIDs

+            self.assertTrue(JID.from_string(jid) == JID.from_user_input(jid))

+

+        for jid in test3:

+            # JIDs starting or ending with \20 are not escaped

+            self.assertTrue(JID.from_string(jid).to_user_string() == jid)

+

+        for user_input, user_string in test4:

+            # Test escaped keyword argument

+            self.assertTrue(JID.from_user_input(user_input, escaped=True) != JID.from_user_input(user_input))

+            self.assertTrue(JID.from_user_input(user_input, escaped=True).to_user_string() == user_string)

+

+        for user_input in fail_tests:

+            # from_user_input does only support bare jids

+            with self.assertRaises(Exception):

+                JID.from_user_input(user_input)

Philipp Hörist pushed to branch master at gajim / python-nbxmpp

Commits:

30 changed files:

Changes: