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

Commits: 63889e95 by Philipp Hörist at 2023-04-20T22:42:59+02:00 feat: Add method to generate XMPP IRIs

- - - - -

3 changed files:

- nbxmpp/protocol.py - + nbxmpp/xmppiri.py - test/unit/test_jid_parsing.py

Changes:

===================================== nbxmpp/protocol.py ===================================== @@ -38,6 +38,12 @@ from dataclasses import asdict from gi.repository import GLib

import idna +from nbxmpp.xmppiri import escape_ifragment +from nbxmpp.xmppiri import escape_inode +from nbxmpp.xmppiri import escape_ires +from nbxmpp.xmppiri import escape_ivalue +from nbxmpp.xmppiri import validate_ikey +from nbxmpp.xmppiri import validate_querytype from nbxmpp.simplexml import Node from nbxmpp.namespaces import Namespace from nbxmpp.stringprep import nodeprep @@ -315,6 +321,7 @@ ERR_REDIRECT = 'urn:ietf:params:xml:ns:xmpp-stanzas redirect' STREAM_UNSUPPORTED_STANZA_TYPE = 'urn:ietf:params:xml:ns:xmpp-streams unsupported-stanza-type' ERR_FORBIDDEN = 'urn:ietf:params:xml:ns:xmpp-stanzas forbidden'

+ def isResultNode(node): """ Return true if the node is a positive reply @@ -768,6 +775,45 @@ class JID: return f'{localpart}@{self.domain}{domain_encoded}' return f'{localpart}@{self.domain}/{self.resource}{domain_encoded}'

+ def to_iri(self, + query: Optional[str | tuple[str, list[tuple[str, str]]]] = None, + fragment: Optional[str] = None + ) -> str: + + if self.localpart: + inode = escape_inode(self.localpart) + ipathxmpp = f'{inode}@{self.domain}' + else: + ipathxmpp = f'{self.domain}' + + if self.resource is not None: + ires = escape_ires(self.resource) + ipathxmpp = f'{ipathxmpp}/{ires}' + + iri = f'xmpp:{ipathxmpp}' + + if query is not None: + if isinstance(query, str): + querytype = query + queryparams = None + else: + querytype, queryparams = query + + iquerytype = validate_querytype(querytype) + iri += f'?{iquerytype}' + + if queryparams is not None: + for ikey, ivalue in queryparams: + ivalue = escape_ivalue(ivalue) + ikey = validate_ikey(ikey) + iri += f';{ikey}={ivalue}' + + if fragment is not None: + ifragment = escape_ifragment(fragment) + iri += f'#{ifragment}' + + return iri + def copy(self) -> JID: deprecation_warning('copy() is not needed, JID is immutable') return self

===================================== nbxmpp/xmppiri.py ===================================== @@ -0,0 +1,57 @@ + +import re +from gi.repository import GLib + + +# https://www.rfc-editor.org/rfc/rfc3987 + +ucschar = r'\xA0-\uD7FF\uF900-\uFDCF\uFDF0-\uFFEF'\ + r'\U00010000-\U0001FFFD\U00020000-\U0002FFFD\U00030000-\U0003FFFD'\ + r'\U00040000-\U0004FFFD\U00050000-\U0005FFFD\U00060000-\U0006FFFD'\ + r'\U00070000-\U0007FFFD\U00080000-\U0008FFFD\U00090000-\U0009FFFD'\ + r'\U000A0000-\U000AFFFD\U000B0000-\U000BFFFD\U000C0000-\U000CFFFD'\ + r'\U000D0000-\U000DFFFD\U000E1000-\U000EFFFD' +unreserved = r'A-Za-z0-9-._~' +iunreserved = fr'{unreserved}{ucschar}' +subdelims = r"!$&'()*+,;=" + +# https://www.rfc-editor.org/rfc/rfc5122.html#section-2.2 +nodeallow = r"!$()*+,;=" +resallow = r"!$&'()*+,:;=" + +# ifragment without iunreserved and pct-encoded +reserved_chars_allowed_in_ifragment = subdelims + ":@" + "/?" + +rx_ikey = f'[{iunreserved}]*' +rx_iquerytype = f'[{iunreserved}]*' + + +def validate_ikey(ikey: str) -> str: + res = re.fullmatch(rx_ikey, ikey) + if res is None: + raise ValueError('Not allowed characters in key') + return ikey + + +def validate_querytype(querytype: str) -> str: + res = re.fullmatch(rx_iquerytype, querytype) + if res is None: + raise ValueError('Not allowed characters in querytype') + return querytype + + +def escape_ifragment(ifragment: str) -> str: + return GLib.Uri.escape_string( + ifragment, reserved_chars_allowed_in_ifragment, True) + + +def escape_ivalue(ivalue: str) -> str: + return GLib.Uri.escape_string(ivalue, None, True) + + +def escape_inode(inode: str) -> str: + return GLib.Uri.escape_string(inode, nodeallow, True) + + +def escape_ires(ires: str) -> str: + return GLib.Uri.escape_string(ires, resallow, True)

===================================== test/unit/test_jid_parsing.py ===================================== @@ -189,3 +189,26 @@ class JIDParsing(unittest.TestCase): # from_user_input does only support bare jids with self.assertRaises(Exception): JID.from_user_input(user_input) + + def test_jid_to_iri(self): + tests = [ + ('nasty!#$%()*+,-.;=?[\]^_`{|}~node@example.com', 'xmpp:nasty!%23$%25()*+,-.;=%3F%5B%5C%5D%5E_%60%7B%7C%7D~node@example.com'), + ('node@example.com/repulsive !#"$%&'()*+,-./:;<=>?@[\]^_`{|}~resource', 'xmpp:node@example.com/repulsive%20!%23%22$%25&'()*+,-.%2F:;%3C=%3E%3F%40%5B%5C%5D%5E_%60%7B%7C%7D~resource'), + ] + + for jid, iri in tests: + jid = JID.from_string(jid) + self.assertEqual(jid.to_iri(), iri) + + jid = JID.from_string('example-node@example.com') + iri = jid.to_iri(('message', [('subject', 'Hello World')]), 'frag') + self.assertEqual(iri, 'xmpp:example-node@example.com?message;subject=Hello%20World#frag') + + iri = jid.to_iri('message') + self.assertEqual(iri, 'xmpp:example-node@example.com?message') + + iri = jid.to_iri(fragment='onlyfragment') + self.assertEqual(iri, 'xmpp:example-node@example.com#onlyfragment') + + jid = JID.from_user_input('call me "ishmael"@example.com') + self.assertEqual(jid.to_iri(), 'xmpp:call%5C20me%5C20%5C22ishmael%5C22@example.com')

View it on GitLab: https://dev.gajim.org/gajim/python-nbxmpp/-/commit/63889e9543c56b9e0172dc6be...