summaryrefslogtreecommitdiff
path: root/sleekxmpp/jid.py
diff options
context:
space:
mode:
Diffstat (limited to 'sleekxmpp/jid.py')
-rw-r--r--sleekxmpp/jid.py541
1 files changed, 541 insertions, 0 deletions
diff --git a/sleekxmpp/jid.py b/sleekxmpp/jid.py
new file mode 100644
index 00000000..9e9c0d0b
--- /dev/null
+++ b/sleekxmpp/jid.py
@@ -0,0 +1,541 @@
+# -*- coding: utf-8 -*-
+"""
+ sleekxmpp.jid
+ ~~~~~~~~~~~~~~~~~~~~~~~
+
+ This module allows for working with Jabber IDs (JIDs).
+
+ Part of SleekXMPP: The Sleek XMPP Library
+
+ :copyright: (c) 2011 Nathanael C. Fritz
+ :license: MIT, see LICENSE for more details
+"""
+
+from __future__ import unicode_literals
+
+import re
+import socket
+import stringprep
+import encodings.idna
+
+from sleekxmpp.util import stringprep_profiles
+
+#: These characters are not allowed to appear in a JID.
+ILLEGAL_CHARS = '\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r' + \
+ '\x0e\x0f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19' + \
+ '\x1a\x1b\x1c\x1d\x1e\x1f' + \
+ ' !"#$%&\'()*+,./:;<=>?@[\\]^_`{|}~\x7f'
+
+#: The basic regex pattern that a JID must match in order to determine
+#: the local, domain, and resource parts. This regex does NOT do any
+#: validation, which requires application of nodeprep, resourceprep, etc.
+JID_PATTERN = "^(?:([^\"&'/:<>@]{1,1023})@)?([^/@]{1,1023})(?:/(.{1,1023}))?$"
+
+#: The set of escape sequences for the characters not allowed by nodeprep.
+JID_ESCAPE_SEQUENCES = set(['\\20', '\\22', '\\26', '\\27', '\\2f',
+ '\\3a', '\\3c', '\\3e', '\\40', '\\5c'])
+
+#: A mapping of unallowed characters to their escape sequences. An escape
+#: sequence for '\' is also included since it must also be escaped in
+#: certain situations.
+JID_ESCAPE_TRANSFORMATIONS = {' ': '\\20',
+ '"': '\\22',
+ '&': '\\26',
+ "'": '\\27',
+ '/': '\\2f',
+ ':': '\\3a',
+ '<': '\\3c',
+ '>': '\\3e',
+ '@': '\\40',
+ '\\': '\\5c'}
+
+#: The reverse mapping of escape sequences to their original forms.
+JID_UNESCAPE_TRANSFORMATIONS = {'\\20': ' ',
+ '\\22': '"',
+ '\\26': '&',
+ '\\27': "'",
+ '\\2f': '/',
+ '\\3a': ':',
+ '\\3c': '<',
+ '\\3e': '>',
+ '\\40': '@',
+ '\\5c': '\\'}
+
+
+# pylint: disable=c0103
+#: The nodeprep profile of stringprep used to validate the local,
+#: or username, portion of a JID.
+nodeprep = stringprep_profiles.create(
+ nfkc=True,
+ bidi=True,
+ mappings=[
+ stringprep_profiles.b1_mapping,
+ stringprep_profiles.c12_mapping],
+ prohibited=[
+ stringprep.in_table_c11,
+ stringprep.in_table_c12,
+ stringprep.in_table_c21,
+ stringprep.in_table_c22,
+ stringprep.in_table_c3,
+ stringprep.in_table_c4,
+ stringprep.in_table_c5,
+ stringprep.in_table_c6,
+ stringprep.in_table_c7,
+ stringprep.in_table_c8,
+ stringprep.in_table_c9,
+ lambda c: c in ' \'"&/:<>@'],
+ unassigned=[stringprep.in_table_a1])
+
+# pylint: disable=c0103
+#: The resourceprep profile of stringprep, which is used to validate
+#: the resource portion of a JID.
+resourceprep = stringprep_profiles.create(
+ nfkc=True,
+ bidi=True,
+ mappings=[stringprep_profiles.b1_mapping],
+ prohibited=[
+ stringprep.in_table_c12,
+ stringprep.in_table_c21,
+ stringprep.in_table_c22,
+ stringprep.in_table_c3,
+ stringprep.in_table_c4,
+ stringprep.in_table_c5,
+ stringprep.in_table_c6,
+ stringprep.in_table_c7,
+ stringprep.in_table_c8,
+ stringprep.in_table_c9],
+ unassigned=[stringprep.in_table_a1])
+
+
+def _parse_jid(data):
+ """
+ Parse string data into the node, domain, and resource
+ components of a JID, if possible.
+
+ :param string data: A string that is potentially a JID.
+
+ :raises InvalidJID:
+
+ :returns: tuple of the validated local, domain, and resource strings
+ """
+ match = re.match(JID_PATTERN, data)
+ if not match:
+ raise InvalidJID('JID could not be parsed')
+
+ (node, domain, resource) = match.groups()
+
+ node = _validate_node(node)
+ domain = _validate_domain(domain)
+ resource = _validate_resource(resource)
+
+ return node, domain, resource
+
+
+def _validate_node(node):
+ """Validate the local, or username, portion of a JID.
+
+ :raises InvalidJID:
+
+ :returns: The local portion of a JID, as validated by nodeprep.
+ """
+ try:
+ if node is not None:
+ node = nodeprep(node)
+
+ if not node:
+ raise InvalidJID('Localpart must not be 0 bytes')
+ if len(node) > 1023:
+ raise InvalidJID('Localpart must be less than 1024 bytes')
+ return node
+ except stringprep_profiles.StringPrepError:
+ raise InvalidJID('Invalid local part')
+
+
+def _validate_domain(domain):
+ """Validate the domain portion of a JID.
+
+ IP literal addresses are left as-is, if valid. Domain names
+ are stripped of any trailing label separators (`.`), and are
+ checked with the nameprep profile of stringprep. If the given
+ domain is actually a punyencoded version of a domain name, it
+ is converted back into its original Unicode form. Domains must
+ also not start or end with a dash (`-`).
+
+ :raises InvalidJID:
+
+ :returns: The validated domain name
+ """
+ ip_addr = False
+
+ # First, check if this is an IPv4 address
+ try:
+ socket.inet_aton(domain)
+ ip_addr = True
+ except socket.error:
+ pass
+
+ # Check if this is an IPv6 address
+ if not ip_addr and hasattr(socket, 'inet_pton'):
+ try:
+ socket.inet_pton(socket.AF_INET6, domain.strip('[]'))
+ domain = '[%s]' % domain.strip('[]')
+ ip_addr = True
+ except socket.error:
+ pass
+
+ if not ip_addr:
+ # This is a domain name, which must be checked further
+
+ if domain and domain[-1] == '.':
+ domain = domain[:-1]
+
+ domain_parts = []
+ for label in domain.split('.'):
+ try:
+ label = encodings.idna.nameprep(label)
+ encodings.idna.ToASCII(label)
+ pass_nameprep = True
+ except UnicodeError:
+ pass_nameprep = False
+
+ if not pass_nameprep:
+ raise InvalidJID('Could not encode domain as ASCII')
+
+ if label.startswith('xn--'):
+ label = encodings.idna.ToUnicode(label)
+
+ for char in label:
+ if char in ILLEGAL_CHARS:
+ raise InvalidJID('Domain contains illegar characters')
+
+ if '-' in (label[0], label[-1]):
+ raise InvalidJID('Domain started or ended with -')
+
+ domain_parts.append(label)
+ domain = '.'.join(domain_parts)
+
+ if not domain:
+ raise InvalidJID('Domain must not be 0 bytes')
+ if len(domain) > 1023:
+ raise InvalidJID('Domain must be less than 1024 bytes')
+
+ return domain
+
+
+def _validate_resource(resource):
+ """Validate the resource portion of a JID.
+
+ :raises InvalidJID:
+
+ :returns: The local portion of a JID, as validated by resourceprep.
+ """
+ try:
+ if resource is not None:
+ resource = resourceprep(resource)
+
+ if not resource:
+ raise InvalidJID('Resource must not be 0 bytes')
+ if len(resource) > 1023:
+ raise InvalidJID('Resource must be less than 1024 bytes')
+ return resource
+ except stringprep_profiles.StringPrepError:
+ raise InvalidJID('Invalid resource')
+
+
+def _escape_node(node):
+ """Escape the local portion of a JID."""
+ result = []
+
+ for i, char in enumerate(node):
+ if char == '\\':
+ if ''.join((node[i:i+3])) in JID_ESCAPE_SEQUENCES:
+ result.append('\\5c')
+ continue
+ result.append(char)
+
+ for i, char in enumerate(result):
+ if char != '\\':
+ result[i] = JID_ESCAPE_TRANSFORMATIONS.get(char, char)
+
+ escaped = ''.join(result)
+
+ if escaped.startswith('\\20') or escaped.endswith('\\20'):
+ raise InvalidJID('Escaped local part starts or ends with "\\20"')
+
+ _validate_node(escaped)
+
+ return escaped
+
+
+def _unescape_node(node):
+ """Unescape a local portion of a JID.
+
+ .. note::
+ The unescaped local portion is meant ONLY for presentation,
+ and should not be used for other purposes.
+ """
+ unescaped = []
+ seq = ''
+ for i, char in enumerate(node):
+ if char == '\\':
+ seq = node[i:i+3]
+ if seq not in JID_ESCAPE_SEQUENCES:
+ seq = ''
+ if seq:
+ if len(seq) == 3:
+ unescaped.append(JID_UNESCAPE_TRANSFORMATIONS.get(seq, char))
+
+ # Pop character off the escape sequence, and ignore it
+ seq = seq[1:]
+ else:
+ unescaped.append(char)
+ unescaped = ''.join(unescaped)
+
+ return unescaped
+
+
+def _format_jid(local=None, domain=None, resource=None):
+ """Format the given JID components into a full or bare JID.
+
+ :param string local: Optional. The local portion of the JID.
+ :param string domain: Required. The domain name portion of the JID.
+ :param strin resource: Optional. The resource portion of the JID.
+
+ :return: A full or bare JID string.
+ """
+ result = []
+ if local:
+ result.append(local)
+ result.append('@')
+ if domain:
+ result.append(domain)
+ if resource:
+ result.append('/')
+ result.append(resource)
+ return ''.join(result)
+
+
+class InvalidJID(ValueError):
+ """
+ Raised when attempting to create a JID that does not pass validation.
+
+ It can also be raised if modifying an existing JID in such a way as
+ to make it invalid, such trying to remove the domain from an existing
+ full JID while the local and resource portions still exist.
+ """
+
+# pylint: disable=R0903
+class UnescapedJID(object):
+
+ """
+ .. versionadded:: 1.1.10
+ """
+
+ def __init__(self, local, domain, resource):
+ self._jid = (local, domain, resource)
+
+ # pylint: disable=R0911
+ def __getattr__(self, name):
+ """Retrieve the given JID component.
+
+ :param name: one of: user, server, domain, resource,
+ full, or bare.
+ """
+ if name == 'resource':
+ return self._jid[2] or ''
+ elif name in ('user', 'username', 'local', 'node'):
+ return self._jid[0] or ''
+ elif name in ('server', 'domain', 'host'):
+ return self._jid[1] or ''
+ elif name in ('full', 'jid'):
+ return _format_jid(*self._jid)
+ elif name == 'bare':
+ return _format_jid(self._jid[0], self._jid[1])
+ elif name == '_jid':
+ return getattr(super(JID, self), '_jid')
+ else:
+ return None
+
+ def __str__(self):
+ """Use the full JID as the string value."""
+ return _format_jid(*self._jid)
+
+ def __repr__(self):
+ """Use the full JID as the representation."""
+ return self.__str__()
+
+
+class JID(object):
+
+ """
+ A representation of a Jabber ID, or JID.
+
+ Each JID may have three components: a user, a domain, and an optional
+ resource. For example: user@domain/resource
+
+ When a resource is not used, the JID is called a bare JID.
+ The JID is a full JID otherwise.
+
+ **JID Properties:**
+ :jid: Alias for ``full``.
+ :full: The string value of the full JID.
+ :bare: The string value of the bare JID.
+ :user: The username portion of the JID.
+ :username: Alias for ``user``.
+ :local: Alias for ``user``.
+ :node: Alias for ``user``.
+ :domain: The domain name portion of the JID.
+ :server: Alias for ``domain``.
+ :host: Alias for ``domain``.
+ :resource: The resource portion of the JID.
+
+ :param string jid:
+ A string of the form ``'[user@]domain[/resource]'``.
+ :param string local:
+ Optional. Specify the local, or username, portion
+ of the JID. If provided, it will override the local
+ value provided by the `jid` parameter. The given
+ local value will also be escaped if necessary.
+ :param string domain:
+ Optional. Specify the domain of the JID. If
+ provided, it will override the domain given by
+ the `jid` parameter.
+ :param string resource:
+ Optional. Specify the resource value of the JID.
+ If provided, it will override the domain given
+ by the `jid` parameter.
+
+ :raises InvalidJID:
+ """
+
+ # pylint: disable=W0212
+ def __init__(self, jid=None, **kwargs):
+ self._jid = (None, None, None)
+
+ if jid is None or jid == '':
+ jid = (None, None, None)
+ elif not isinstance(jid, JID):
+ jid = _parse_jid(jid)
+ else:
+ jid = jid._jid
+
+ local, domain, resource = jid
+
+ local = kwargs.get('local', local)
+ domain = kwargs.get('domain', domain)
+ resource = kwargs.get('resource', resource)
+
+ if 'local' in kwargs:
+ local = _escape_node(local)
+ if 'domain' in kwargs:
+ domain = _validate_domain(domain)
+ if 'resource' in kwargs:
+ resource = _validate_resource(resource)
+
+ self._jid = (local, domain, resource)
+
+ def unescape(self):
+ """Return an unescaped JID object.
+
+ Using an unescaped JID is preferred for displaying JIDs
+ to humans, and they should NOT be used for any other
+ purposes than for presentation.
+
+ :return: :class:`UnescapedJID`
+
+ .. versionadded:: 1.1.10
+ """
+ return UnescapedJID(_unescape_node(self._jid[0]),
+ self._jid[1],
+ self._jid[2])
+
+ def regenerate(self):
+ """No-op
+
+ .. deprecated:: 1.1.10
+ """
+ pass
+
+ def reset(self, data):
+ """Start fresh from a new JID string.
+
+ :param string data: A string of the form ``'[user@]domain[/resource]'``.
+
+ .. deprecated:: 1.1.10
+ """
+ self._jid = JID(data)._jid
+
+ # pylint: disable=R0911
+ def __getattr__(self, name):
+ """Retrieve the given JID component.
+
+ :param name: one of: user, server, domain, resource,
+ full, or bare.
+ """
+ if name == 'resource':
+ return self._jid[2] or ''
+ elif name in ('user', 'username', 'local', 'node'):
+ return self._jid[0] or ''
+ elif name in ('server', 'domain', 'host'):
+ return self._jid[1] or ''
+ elif name in ('full', 'jid'):
+ return _format_jid(*self._jid)
+ elif name == 'bare':
+ return _format_jid(self._jid[0], self._jid[1])
+ elif name == '_jid':
+ return getattr(super(JID, self), '_jid')
+ else:
+ return None
+
+ # pylint: disable=W0212
+ def __setattr__(self, name, value):
+ """Update the given JID component.
+
+ :param name: one of: ``user``, ``username``, ``local``,
+ ``node``, ``server``, ``domain``, ``host``,
+ ``resource``, ``full``, ``jid``, or ``bare``.
+ :param value: The new string value of the JID component.
+ """
+ if name == 'resource':
+ self._jid = JID(self, resource=value)._jid
+ elif name in ('user', 'username', 'local', 'node'):
+ self._jid = JID(self, local=value)._jid
+ elif name in ('server', 'domain', 'host'):
+ self._jid = JID(self, domain=value)._jid
+ elif name in ('full', 'jid'):
+ self._jid = JID(value)._jid
+ elif name == 'bare':
+ parsed = JID(value)._jid
+ self._jid = (parsed[0], parsed[1], self._jid[2])
+ elif name == '_jid':
+ super(JID, self).__setattr__('_jid', value)
+
+ def __str__(self):
+ """Use the full JID as the string value."""
+ return _format_jid(*self._jid)
+
+ def __repr__(self):
+ """Use the full JID as the representation."""
+ return self.__str__()
+
+ # pylint: disable=W0212
+ def __eq__(self, other):
+ """Two JIDs are equal if they have the same full JID value."""
+ if isinstance(other, UnescapedJID):
+ return False
+
+ other = JID(other)
+ return self._jid == other._jid
+
+ # pylint: disable=W0212
+ def __ne__(self, other):
+ """Two JIDs are considered unequal if they are not equal."""
+ return not self == other
+
+ def __hash__(self):
+ """Hash a JID based on the string version of its full JID."""
+ return hash(self.__str__())
+
+ def __copy__(self):
+ """Generate a duplicate JID."""
+ return JID(self)