class _Namespaces:
    ACTIVITY: str = 'http://jabber.org/protocol/activity'
    ADDRESS: str = 'http://jabber.org/protocol/address'
    AGENTS: str = 'jabber:iq:agents'
    ATTENTION: str = 'urn:xmpp:attention:0'
    AUTH: str = 'jabber:iq:auth'
    AVATAR_METADATA: str = 'urn:xmpp:avatar:metadata'
    AVATAR_DATA: str = 'urn:xmpp:avatar:data'
    BIND: str = 'urn:ietf:params:xml:ns:xmpp-bind'
    BLOCKING: str = 'urn:xmpp:blocking'
    BOB: str = 'urn:xmpp:bob'
    BOOKMARKS: str = 'storage:bookmarks'
    BOOKMARKS_1: str = 'urn:xmpp:bookmarks:1'
    BOOKMARKS_COMPAT: str = 'urn:xmpp:bookmarks:1#compat'
    BOOKMARKS_COMPAT_PEP: str = 'urn:xmpp:bookmarks:1#compat-pep'
    BOOKMARK_CONVERSION: str = 'urn:xmpp:bookmarks-conversion:0'
    BROWSE: str = 'jabber:iq:browse'
    BYTESTREAM: str = 'http://jabber.org/protocol/bytestreams'
    CAPS: str = 'http://jabber.org/protocol/caps'
    CAPTCHA: str = 'urn:xmpp:captcha'
    CARBONS: str = 'urn:xmpp:carbons:2'
    CHATMARKERS: str = 'urn:xmpp:chat-markers:0'
    CHATSTATES: str = 'http://jabber.org/protocol/chatstates'
    CLIENT: str = 'jabber:client'
    COMMANDS: str = 'http://jabber.org/protocol/commands'
    CONFERENCE: str = 'jabber:x:conference'
    CORRECT: str = 'urn:xmpp:message-correct:0'
    DATA: str = 'jabber:x:data'
    DATA_LAYOUT: str = 'http://jabber.org/protocol/xdata-layout'
    DATA_MEDIA: str = 'urn:xmpp:media-element'
    DATA_VALIDATE: str = 'http://jabber.org/protocol/xdata-validate'
    DELAY: str = 'jabber:x:delay'
    DELAY2: str = 'urn:xmpp:delay'
    DELIMITER: str = 'roster:delimiter'
    DISCO: str = 'http://jabber.org/protocol/disco'
    DISCO_INFO: str = 'http://jabber.org/protocol/disco#info'
    DISCO_ITEMS: str = 'http://jabber.org/protocol/disco#items'
    DOMAIN_BASED_NAME: str = 'urn:xmpp:domain-based-name:1'
    EME: str = 'urn:xmpp:eme:0'
    ENCRYPTED: str = 'jabber:x:encrypted'
    FASTEN: str = 'urn:xmpp:fasten:0'
    FILE_METADATA: str = 'urn:xmpp:file:metadata:0'
    FORWARD: str = 'urn:xmpp:forward:0'
    FRAMING: str = 'urn:ietf:params:xml:ns:xmpp-framing'
    GATEWAY: str = 'jabber:iq:gateway'
    GEOLOC: str = 'http://jabber.org/protocol/geoloc'
    HASHES: str = 'urn:xmpp:hashes:1'
    HASHES_2: str = 'urn:xmpp:hashes:2'
    HASHES_BLAKE2B_256: str = 'urn:xmpp:hash-function-text-names:id-blake2b256'
    HASHES_BLAKE2B_512: str = 'urn:xmpp:hash-function-text-names:id-blake2b512'
    HASHES_MD5: str = 'urn:xmpp:hash-function-text-names:md5'
    HASHES_SHA1: str = 'urn:xmpp:hash-function-text-names:sha-1'
    HASHES_SHA256: str = 'urn:xmpp:hash-function-text-names:sha-256'
    HASHES_SHA512: str = 'urn:xmpp:hash-function-text-names:sha-512'
    HASHES_SHA3_256: str = 'urn:xmpp:hash-function-text-names:sha3-256'
    HASHES_SHA3_512: str = 'urn:xmpp:hash-function-text-names:sha3-512'
    HINTS: str = 'urn:xmpp:hints'
    HTTPUPLOAD_0: str = 'urn:xmpp:http:upload:0'
    HTTP_AUTH: str = 'http://jabber.org/protocol/http-auth'
    IBB: str = 'http://jabber.org/protocol/ibb'
    IDLE: str = 'urn:xmpp:idle:1'
    JINGLE: str = 'urn:xmpp:jingle:1'
    JINGLE_BYTESTREAM: str = 'urn:xmpp:jingle:transports:s5b:1'
    JINGLE_DTLS: str = 'urn:xmpp:jingle:apps:dtls:0'
    JINGLE_ERRORS: str = 'urn:xmpp:jingle:errors:1'
    JINGLE_FILE_TRANSFER: str = 'urn:xmpp:jingle:apps:file-transfer:3'
    JINGLE_FILE_TRANSFER_5: str = 'urn:xmpp:jingle:apps:file-transfer:5'
    JINGLE_IBB: str = 'urn:xmpp:jingle:transports:ibb:1'
    JINGLE_ICE_UDP: str = 'urn:xmpp:jingle:transports:ice-udp:1'
    JINGLE_RAW_UDP: str = 'urn:xmpp:jingle:transports:raw-udp:1'
    JINGLE_RTP: str = 'urn:xmpp:jingle:apps:rtp:1'
    JINGLE_RTP_AUDIO: str = 'urn:xmpp:jingle:apps:rtp:audio'
    JINGLE_RTP_VIDEO: str = 'urn:xmpp:jingle:apps:rtp:video'
    JINGLE_XTLS: str = 'urn:xmpp:jingle:security:xtls:0'
    LAST: str = 'jabber:iq:last'
    LOCATION: str = 'http://jabber.org/protocol/geoloc'
    MAM_1: str = 'urn:xmpp:mam:1'
    MAM_2: str = 'urn:xmpp:mam:2'
    MESSAGE_MODERATE: str = 'urn:xmpp:message-moderate:0'
    MESSAGE_RETRACT: str = 'urn:xmpp:message-retract:0'
    MOOD: str = 'http://jabber.org/protocol/mood'
    MSG_HINTS: str = 'urn:xmpp:hints'
    MUCLUMBUS: str = 'https://xmlns.zombofant.net/muclumbus/search/1.0'
    MUC: str = 'http://jabber.org/protocol/muc'
    MUC_USER: str = 'http://jabber.org/protocol/muc#user'
    MUC_ADMIN: str = 'http://jabber.org/protocol/muc#admin'
    MUC_OWNER: str = 'http://jabber.org/protocol/muc#owner'
    MUC_UNIQUE: str = 'http://jabber.org/protocol/muc#unique'
    MUC_CONFIG: str = 'http://jabber.org/protocol/muc#roomconfig'
    MUC_REQUEST: str = 'http://jabber.org/protocol/muc#request'
    MUC_INFO: str = 'http://jabber.org/protocol/muc#roominfo'
    NICK: str = 'http://jabber.org/protocol/nick'
    OMEMO_TEMP: str = 'eu.siacs.conversations.axolotl'
    OMEMO_TEMP_BUNDLE: str = 'eu.siacs.conversations.axolotl.bundles'
    OMEMO_TEMP_DL: str = 'eu.siacs.conversations.axolotl.devicelist'
    OPENPGP: str = 'urn:xmpp:openpgp:0'
    OPENPGP_PK: str = 'urn:xmpp:openpgp:0:public-keys'
    OPENPGP_SK: str = 'urn:xmpp:openpgp:0:secret-key'
    PING: str = 'urn:xmpp:ping'
    PRIVACY: str = 'jabber:iq:privacy'
    PRIVATE: str = 'jabber:iq:private'
    PUBKEY_ATTEST: str = 'urn:xmpp:attest:2'
    PUBKEY_PUBKEY: str = 'urn:xmpp:pubkey:2'
    PUBKEY_REVOKE: str = 'urn:xmpp:revoke:2'
    PUBSUB: str = 'http://jabber.org/protocol/pubsub'
    PUBSUB_ERROR: str = 'http://jabber.org/protocol/pubsub#errors'
    PUBSUB_CONFIG: str = 'http://jabber.org/protocol/pubsub#node_config'
    PUBSUB_EVENT: str = 'http://jabber.org/protocol/pubsub#event'
    PUBSUB_OWNER: str = 'http://jabber.org/protocol/pubsub#owner'
    PUBSUB_PUBLISH_OPTIONS: str = 'http://jabber.org/protocol/pubsub#publish-options'
    PUBSUB_NODE_MAX: str = 'http://jabber.org/protocol/pubsub#config-node-max'
    RECEIPTS: str = 'urn:xmpp:receipts'
    REGISTER: str = 'jabber:iq:register'
    REGISTER_FEATURE: str = 'http://jabber.org/features/iq-register'
    REPORTING: str = 'urn:xmpp:reporting:0'
    ROSTER: str = 'jabber:iq:roster'
    ROSTERNOTES: str = 'storage:rosternotes'
    ROSTERX: str = 'http://jabber.org/protocol/rosterx'
    ROSTER_VER: str = 'urn:xmpp:features:rosterver'
    RSM: str = 'http://jabber.org/protocol/rsm'
    SASL: str = 'urn:ietf:params:xml:ns:xmpp-sasl'
    SEARCH: str = 'jabber:iq:search'
    SECLABEL: str = 'urn:xmpp:sec-label:0'
    SECLABEL_CATALOG: str = 'urn:xmpp:sec-label:catalog:2'
    SESSION: str = 'urn:ietf:params:xml:ns:xmpp-session'
    SFS: str = 'urn:xmpp:sfs:0'
    SID: str = 'urn:xmpp:sid:0'
    SIGNED: str = 'jabber:x:signed'
    SIMS: str = 'urn:xmpp:sims:1'
    STANZAS: str = 'urn:ietf:params:xml:ns:xmpp-stanzas'
    STICKERS: str = 'urn:xmpp:stickers:0'
    STREAM: str = 'http://affinix.com/jabber/stream'
    STREAMS: str = 'http://etherx.jabber.org/streams'
    STREAM_MGMT: str = 'urn:xmpp:sm:3'
    STYLING: str = 'urn:xmpp:styling:0'
    TIME_REVISED: str = 'urn:xmpp:time'
    TIME: str = 'urn:xmpp:time'
    TLS: str = 'urn:ietf:params:xml:ns:xmpp-tls'
    TUNE: str = 'http://jabber.org/protocol/tune'
    URL_DATA: str = 'http://jabber.org/protocol/url-data'
    VCARD: str = 'vcard-temp'
    VCARD_UPDATE: str = 'vcard-temp:x:update'
    VCARD_CONVERSION: str = 'urn:xmpp:pep-vcard-conversion:0'
    VCARD4: str = 'urn:ietf:params:xml:ns:vcard-4.0'
    VCARD4_PUBSUB: str = 'urn:xmpp:vcard4'
    VERSION: str = 'jabber:iq:version'
    XHTML_IM: str = 'http://jabber.org/protocol/xhtml-im'
    XHTML: str = 'http://www.w3.org/1999/xhtml'
    XMPP_STREAMS: str = 'urn:ietf:params:xml:ns:xmpp-streams'
    X_OOB: str = 'jabber:x:oob' class Features(Node):
    def __init__(self, node: Node):
        Node.__init__(self, node=node)

    def has_starttls(self):
        tls = self.getTag('starttls', namespace=Namespace.TLS)
        if tls is not None:
            required = tls.getTag('required') is not None
            return True, required
        return False, False

    def has_sasl(self):
        return self.getTag('mechanisms', namespace=Namespace.SASL) is not None

    def get_mechs(self) -> set[str]:
        mechanisms = self.getTag('mechanisms', namespace=Namespace.SASL)
        if mechanisms is None:
            return set()
        mechanisms = mechanisms.getTags('mechanism')
        return set(mech.getData() for mech in mechanisms)

    def get_domain_based_name(self):
        hostname = self.getTag('hostname',
                               namespace=Namespace.DOMAIN_BASED_NAME)
        if hostname is not None:
            return hostname.getData()
        return None

    def has_bind(self):
        return self.getTag('bind', namespace=Namespace.BIND) is not None

    def session_required(self):
        session = self.getTag('session', namespace=Namespace.SESSION)
        if session is not None:
            optional = session.getTag('optional') is not None
            return not optional
        return False

    def has_sm(self):
        return self.getTag('sm', namespace=Namespace.STREAM_MGMT) is not None

    def has_roster_version(self):
        return self.getTag('ver', namespace=Namespace.ROSTER_VER) is not None

    def has_register(self):
        return self.getTag(
            'register', namespace=Namespace.REGISTER_FEATURE) is not None

    def has_anonymous(self):
        return 'ANONYMOUS' in self.get_mechs() class StanzaMalformed(Exception):
    pass class SessionRequest(Iq):
    def __init__(self):
        Iq.__init__(self, typ='set')
        self.addChild(node=Node('session', attrs={'xmlns': Namespace.SESSION})) class BindRequest(Iq):
    def __init__(self, resource: Optional[str]):
        if resource is None:
            res = resource
        else:
            res = Node('resource', payload=resource)
        Iq.__init__(self, typ='set')
        self.addChild(node=Node('bind',
                                {'xmlns': Namespace.BIND},
                                payload=res)) class TLSRequest(Node):
    def __init__(self):
        Node.__init__(self, tag='starttls', attrs={'xmlns': Namespace.TLS}) def isResultNode(node):
    """
    Return true if the node is a positive reply
    """
    return node and node.getType() == 'result' class JID:
    localpart: Optional[str] = None
    domain: Optional[str] = None
    resource: Optional[str] = None

    def __init__(self,
                 localpart: Optional[str] = None,
                 domain: Optional[str] = None,
                 resource: Optional[str] = None):

        if localpart is not None:
            localpart = validate_localpart(localpart)
            object.__setattr__(self, "localpart", localpart)

        domain = validate_domainpart(domain)
        object.__setattr__(self, "domain", domain)

        if resource is not None:
            resource = validate_resourcepart(resource)
            object.__setattr__(self, "resource", resource)

    @classmethod
    @functools.lru_cache(maxsize=None)
    def from_string(cls, jid_string: str, force_bare: bool = False) -> 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_string.find('/') != -1:
            rest, resourcepart = jid_string.split('/', 1)
        else:
            rest, resourcepart = jid_string, None

        if rest.find('@') != -1:
            localpart, domainpart = rest.split('@', 1)
        else:
            localpart, domainpart = None, rest

        if force_bare:
            if localpart is None:
                raise LocalpartByteLimit
            resourcepart = None

        return cls(localpart=localpart,
                   domain=domainpart,
                   resource=resourcepart)

    @classmethod
    @functools.lru_cache(maxsize=None)
    def from_user_input(cls, user_input: str, escaped: bool = False) -> JID:
        # 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.

        # Only user_input which after escaping result in a bare JID can be
        # successfully parsed.

        # 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.

        if escaped:
            # for convenience
            return cls.from_string(user_input)

        if '@' in user_input:
            localpart, domainpart = user_input.rsplit('@', 1)
            if localpart.startswith(' ') or localpart.endswith(' '):
                raise LocalpartNotAllowedChar

            localpart = escape_localpart(localpart)

        else:
            localpart = None
            domainpart = user_input

        return cls(localpart=localpart,
                   domain=domainpart,
                   resource=None)

    def __str__(self) -> str:
        if self.localpart:
            jid = f'{self.localpart}@{self.domain}'
        else:
            jid = cast(str, self.domain)

        if self.resource is not None:
            return f'{jid}/{self.resource}'
        return jid

    def __hash__(self):
        return hash(str(self))

    def __eq__(self, other: Union[str, JID]) -> bool:
        if isinstance(other, str):
            try:
                return JID.from_string(other) == self
            except Exception:
                return False

        return (self.localpart == other.localpart and
                self.domain == other.domain and
                self.resource == other.resource)

    def __ne__(self, other: Union[str, JID]) -> bool:
        return not self.__eq__(other)

    def domain_to_ascii(self) -> str:
        return idna_encode(self.domain)

    @property
    def bare(self) -> str:
        if self.localpart is not None:
            return f'{self.localpart}@{self.domain}'
        return self.domain

    @property
    def is_bare(self) -> bool:
        return self.resource is None

    def new_as_bare(self) -> JID:
        if self.resource is None:
            return self
        new = asdict(self)
        new.pop('resource')
        return JID(**new)

    def bare_match(self, other: Union[str, JID]) -> bool:
        if isinstance(other, str):
            other = JID.from_string(other)
        return self.bare == other.bare

    @property
    def is_domain(self) -> bool:
        return self.localpart is None and self.resource is None

    @property
    def is_full(self) -> bool:
        return (self.localpart is not None and
                self.domain is not None and
                self.resource is not None)

    def new_with(self, **kwargs: dict[str, str]) -> JID:
        new = asdict(self)
        new.update(kwargs)
        return JID(**new)

    def to_user_string(self, show_punycode: bool = True) -> str:
        domain = self.domain_to_ascii()
        if domain.startswith('xn--') and show_punycode:
            domain_encoded = f' ({domain})'
        else:
            domain_encoded = ''

        if self.localpart is None:
            return f'{self}{domain_encoded}'

        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 copy(self) -> JID:
        deprecation_warning('copy() is not needed, JID is immutable')
        return self class Protocol(Node):
    """
    A "stanza" object class. Contains methods that are common for presences, iqs
    and messages
    """

    def __init__(self,
                 name=None,
                 to=None,
                 typ=None,
                 frm=None,
                 attrs=None,
                 payload=None,
                 timestamp=None,
                 xmlns=None,
                 node=None):
        """
        Constructor, name is the name of the stanza
        i.e. 'message' or 'presence'or 'iq'

        to is the value of 'to' attribure, 'typ' - 'type' attribute
        frn - from attribure, attrs - other attributes mapping,
        payload - same meaning as for simplexml payload definition
        timestamp - the time value that needs to be stamped over stanza
        xmlns - namespace of top stanza node
        node - parsed or unparsed stana to be taken as prototype.
        """
        if not attrs:
            attrs = {}
        if to:
            attrs['to'] = to
        if frm:
            attrs['from'] = frm
        if typ:
            attrs['type'] = typ
        Node.__init__(self, tag=name, attrs=attrs, payload=payload, node=node)
        if not node and xmlns:
            self.setNamespace(xmlns)
        if self['to']:
            self.setTo(self['to'])
        if self['from']:
            self.setFrom(self['from'])
        if (node and
                isinstance(node, Protocol) and
                self.__class__ == node.__class__
                and 'id' in self.attrs):
            del self.attrs['id']
        self.timestamp = None
        for d in self.getTags('delay', namespace=Namespace.DELAY2):
            try:
                if d.getAttr('stamp') < self.getTimestamp2():
                    self.setTimestamp(d.getAttr('stamp'))
            except Exception:
                pass
        if not self.timestamp:
            for x in self.getTags('x', namespace=Namespace.DELAY):
                try:
                    if x.getAttr('stamp') < self.getTimestamp():
                        self.setTimestamp(x.getAttr('stamp'))
                except Exception:
                    pass
        if timestamp is not None:
            self.setTimestamp(timestamp)

    def isError(self):
        return self.getAttr('type') == 'error'

    def isResult(self):
        return self.getAttr('type') == 'result'

    def getTo(self):
        """
        Return value of the 'to' attribute
        """
        try:
            return self['to']
        except Exception:
            pass
        return None

    def getFrom(self):
        """
        Return value of the 'from' attribute
        """
        try:
            return self['from']
        except Exception:
            pass
        return None

    def getTimestamp(self):
        """
        Return the timestamp in the 'yyyymmddThhmmss' format
        """
        if self.timestamp:
            return self.timestamp
        return time.strftime('%Y%m%dT%H:%M:%S', time.gmtime())

    def getTimestamp2(self):
        """
        Return the timestamp in the 'yyyymmddThhmmss' format
        """
        if self.timestamp:
            return self.timestamp
        return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())

    def getJid(self):
        """
        Return the value of the 'jid' attribute
        """
        attr = self.getAttr('jid')
        if attr:
            return JID.from_string(attr)
        return attr

    def getID(self) -> Optional[str]:
        """
        Return the value of the 'id' attribute
        """
        return self.getAttr('id')

    def setTo(self, val: Union[str, JID]):
        """
        Set the value of the 'to' attribute
        """
        if not isinstance(val, JID):
            val = JID.from_string(val)
        self.setAttr('to', val)

    def getType(self) -> Optional[str]:
        """
        Return the value of the 'type' attribute
        """
        return self.getAttr('type')

    def setFrom(self, val: Union[str, JID]):
        """
        Set the value of the 'from' attribute
        """
        if not isinstance(val, JID):
            val = JID.from_string(val)
        self.setAttr('from', val)

    def setType(self, val: str):
        """
        Set the value of the 'type' attribute
        """
        self.setAttr('type', val)

    def setID(self, val: str):
        """
        Set the value of the 'id' attribute
        """
        self.setAttr('id', val)

    def getError(self) -> Optional[str]:
        """
        Return the error-condition (if present) or the textual description
        of the error (otherwise)
        """
        errtag = self.getTag('error')
        if errtag is None:
            return None
        for tag in errtag.getChildren():
            if (tag.getName() != 'text' and
                    tag.getNamespace() == Namespace.STANZAS):
                return tag.getName()
        return None

    def getAppError(self):
        errtag = self.getTag('error')
        if errtag is None:
            return None
        for tag in errtag.getChildren():
            if (tag.getName() != 'text' and
                    tag.getNamespace() != Namespace.STANZAS):
                return tag.getName()
        return None

    def getAppErrorNamespace(self):
        errtag = self.getTag('error')
        if errtag is None:
            return None
        for tag in errtag.getChildren():
            if (tag.getName() != 'text' and
                    tag.getNamespace() != Namespace.STANZAS):
                return tag.getNamespace()
        return None

    def getErrorMsg(self):
        """
        Return the textual description of the error (if present)
        or the error condition
        """
        errtag = self.getTag('error')
        if errtag:
            for tag in errtag.getChildren():
                if tag.getName() == 'text':
                    return tag.getData()
            return self.getError()
        return None

    def getErrorCode(self):
        """
        Return the error code. Obsolete.
        """
        return self.getTagAttr('error', 'code')

    def getErrorType(self):
        """
        Return the error code. Obsolete.
        """
        return self.getTagAttr('error', 'type')

    def getStatusConditions(self, as_code=False):
        """
        Return the status conditions list as defined in XEP-0306.
        """
        result = set()
        status_tags = self.getTags('status')
        for status in status_tags:
            if as_code:
                code = status.getAttr('code')
                if code is not None:
                    result.add(code)
            else:
                for condition in status.getChildren():
                    result.add(condition.getName())
        return list(result)

    def setError(self, error, code=None):
        """
        Set the error code. Obsolete. Use error-conditions instead
        """
        if code:
            if str(code) in _errorcodes.keys():
                error = ErrorNode(_errorcodes[str(code)], text=error)
            else:
                error = ErrorNode(ERR_UNDEFINED_CONDITION, code=code,
                                  typ='cancel', text=error)
        elif isinstance(error, str):
            error = ErrorNode(error)
        self.setType('error')
        self.addChild(node=error)

    def setTimestamp(self, val=None):
        """
        Set the timestamp. timestamp should be the yyyymmddThhmmss string
        """
        if not val:
            val = time.strftime('%Y%m%dT%H:%M:%S', time.gmtime())
        self.timestamp=val
        self.setTag('x', {'stamp': self.timestamp}, namespace=Namespace.DELAY)

    def getProperties(self):
        """
        Return the list of namespaces to which belongs the
        direct childs of element
        """
        props = []
        for child in self.getChildren():
            prop = child.getNamespace()
            if prop not in props:
                props.append(prop)
        return props

    def getTag(self,
               name: str,
               attrs: Optional[dict[str, Any]] = None,
               namespace: Optional[str] = None,
               protocol: bool = False) -> Optional[Node]:
        """
        Return the Node instance for the tag.
        If protocol is True convert to a new Protocol/Message instance.
        """
        tag = Node.getTag(self, name, attrs, namespace)
        if protocol and tag:
            if name == 'message':
                return Message(node=tag)
            return Protocol(node=tag)
        return tag

    def __setitem__(self, item: str, val: Union[str, JID]):
        """
        Set the item 'item' to the value 'val'
        """
        if item in ['to', 'from']:
            if not isinstance(val, JID):
                val = JID.from_string(val)
        self.setAttr(item, val) class WebsocketCloseHeader(Node):
    def __init__(self):
        Node.__init__(self, tag='close', attrs={'xmlns': Namespace.FRAMING}) class TimeoutStanzaError(BaseError):

    log_level = logging.INFO

    def __init__(self):
        BaseError.__init__(self)
        self.text = 'Timeout reached' class StanzaError(BaseError):

    log_level = logging.INFO
    app_namespace = None

    def __init__(self, stanza: Protocol):
        BaseError.__init__(self)
        self.stanza = stanza
        self._stanza_name = stanza.getName()
        self._error_node = stanza.getTag('error')
        self.condition: Optional[str] = stanza.getError()
        self.condition_data = self._error_node.getTagData(self.condition)
        self.app_condition = self._get_app_condition()
        self.type = stanza.getErrorType()
        self.jid = stanza.getFrom()
        self.id = stanza.getID()
        self._text: dict[str, str] = {}

        text_elements = self._error_node.getTags('text',
                                                 namespace=Namespace.STANZAS)
        for element in text_elements:
            lang = element.getXmlLang()
            text = element.getData()
            self._text[lang] = text

    def _get_app_condition(self) -> Optional[str]:
        if self.app_namespace is None:
            return None

        for node in self._error_node.getChildren():
            if node.getNamespace() == self.app_namespace:
                return node.getName()
        return None

    def get_text(self, pref_lang: Optional[str] = None) -> str:
        if pref_lang is not None:
            text = self._text.get(pref_lang)
            if text is not None:
                return text

        if self._text:
            text = self._text.get('en')
            if text is not None:
                return text

            text = self._text.get(None)
            if text is not None:
                return text
            return self._text.popitem()[1]
        return ''

    def set_text(self, lang: str, text: str) -> None:
        self._text[lang] = text

    def __str__(self) -> str:
        condition = self.condition
        if self.app_condition is not None:
            condition = '%s (%s)' % (self.condition, self.app_condition)
        text = self.get_text('en') or ''
        if text:
            text = ' - %s' % text
        return 'Error from %s: %s%s' % (self.jid, condition, text) class CancelledError(BaseError):

    log_level = logging.INFO

    def __init__(self):
        BaseError.__init__(self, is_fatal=True)
        self.text = 'Task has been cancelled' class ServerAddresses(Observable):
    '''
    Signals:

        resolved

    '''

    def __init__(self, domain):
        Observable.__init__(self, log)

        self._domain = domain
        self._custom_host = None
        self._proxy = None
        self._is_resolved = False

        self._addresses = [
            ServerAddress(domain=self._domain,
                          service='xmpps-client',
                          host=None,
                          uri=None,
                          protocol=ConnectionProtocol.TCP,
                          type=ConnectionType.DIRECT_TLS,
                          proxy=None),

            ServerAddress(domain=self._domain,
                          service='xmpp-client',
                          host=None,
                          uri=None,
                          protocol=ConnectionProtocol.TCP,
                          type=ConnectionType.START_TLS,
                          proxy=None),

            ServerAddress(domain=self._domain,
                          service='xmpp-client',
                          host=None,
                          uri=None,
                          protocol=ConnectionProtocol.TCP,
                          type=ConnectionType.PLAIN,
                          proxy=None)
        ]

        self._fallback_addresses = [
            ServerAddress(domain=self._domain,
                          service=None,
                          host='%s:%s' % (self._domain, 5222),
                          uri=None,
                          protocol=ConnectionProtocol.TCP,
                          type=ConnectionType.START_TLS,
                          proxy=None),

            ServerAddress(domain=self._domain,
                          service=None,
                          host='%s:%s' % (self._domain, 5222),
                          uri=None,
                          protocol=ConnectionProtocol.TCP,
                          type=ConnectionType.PLAIN,
                          proxy=None)
        ]

    @property
    def domain(self):
        return self._domain

    @property
    def is_resolved(self):
        return self._is_resolved

    def resolve(self):
        if self._is_resolved:
            self._on_request_resolved()
            return

        if self._proxy is not None:
            # Let the proxy resolve the domain
            self._on_request_resolved()
            return

        if self._custom_host is not None:
            self._on_request_resolved()
            return

        GioResolver().resolve_alternatives(self._domain,
                                           self._on_alternatives_result)

    def cancel_resolve(self):
        self.remove_subscriptions()

    def set_custom_host(self, address):
        # Set a custom host, overwrites all other addresses
        self._custom_host = address
        if address is None:
            return

        host_or_uri, protocol, type_ = address
        if protocol == ConnectionProtocol.WEBSOCKET:
            host, uri = None, host_or_uri
        else:
            host, uri = host_or_uri, None

        self._fallback_addresses = []
        self._addresses = [
            ServerAddress(domain=self._domain,
                          service=None,
                          host=host,
                          uri=uri,
                          protocol=protocol,
                          type=type_,
                          proxy=None)]

    def set_proxy(self, proxy):
        self._proxy = proxy

    def _on_alternatives_result(self, uri):
        if uri is None:
            self._on_request_resolved()
            return

        if uri.startswith('wss'):
            type_ = ConnectionType.DIRECT_TLS
        elif uri.startswith('ws'):
            type_ = ConnectionType.PLAIN
        else:
            log.warning('Invalid websocket uri: %s', uri)
            self._on_request_resolved()
            return

        addr = ServerAddress(domain=self._domain,
                             service=None,
                             host=None,
                             uri=uri,
                             protocol=ConnectionProtocol.WEBSOCKET,
                             type=type_,
                             proxy=None)
        self._addresses.append(addr)

        self._on_request_resolved()

    def _on_request_resolved(self):
        self._is_resolved = True
        self.notify('resolved')
        self.remove_subscriptions()

    def get_next_address(self,
                         allowed_types,
                         allowed_protocols):
        '''
        Selects next address
        '''

        for addr in self._filter_allowed(self._addresses,
                                         allowed_types,
                                         allowed_protocols):
            yield self._assure_proxy(addr)

        for addr in self._filter_allowed(self._fallback_addresses,
                                         allowed_types,
                                         allowed_protocols):
            yield self._assure_proxy(addr)

        raise NoMoreAddresses

    def _assure_proxy(self, addr):
        if self._proxy is None:
            return addr

        if addr.protocol == ConnectionProtocol.TCP:
            return addr._replace(proxy=self._proxy)

        return addr

    def _filter_allowed(self, addresses, allowed_types, allowed_protocols):
        if self._proxy is not None:
            addresses = filter(lambda addr: addr.host is not None, addresses)

        addresses = filter(lambda addr: addr.type in allowed_types,
                           addresses)
        addresses = filter(lambda addr: addr.protocol in allowed_protocols,
                           addresses)
        return addresses

    def __str__(self):
        addresses = self._addresses + self._fallback_addresses
        return '\n'.join([str(addr) for addr in addresses]) class NoMoreAddresses(Exception):
    pass class TCPConnection(Connection):
    def __init__(self, *args, **kwargs):
        Connection.__init__(self, *args, **kwargs)

        self._client = Gio.SocketClient.new()
        self._client.set_timeout(7)

        if self._address.proxy is not None:
            self._proxy_resolver = self._address.proxy.get_resolver()
            self._client.set_proxy_resolver(self._proxy_resolver)

        GObject.Object.connect(self._client, 'event', self._on_event)

        self._con = None

        self._read_buffer = b''

        self._write_queue = deque([])
        self._write_stanza_buffer = None

        self._connect_cancellable = Gio.Cancellable()
        self._read_cancellable = Gio.Cancellable()

        self._tls_handshake_in_progress = False
        self._input_closed = False
        self._output_closed = False

        self._keepalive_id = None

    def connect(self):
        self.state = TCPState.CONNECTING

        if self._address.is_service:
            self._client.connect_to_service_async(self._address.domain,
                                                  self._address.service,
                                                  self._connect_cancellable,
                                                  self._on_connect_finished,
                                                  None)
        elif self._address.is_host:
            self._client.connect_to_host_async(self._address.host,
                                               0,
                                               self._connect_cancellable,
                                               self._on_connect_finished,
                                               None)

        else:
            raise ValueError('Invalid Address')

    def _on_event(self, _socket_client, event, _connectable, connection):
        if event == Gio.SocketClientEvent.CONNECTING:
            self._remote_address = connection.get_remote_address().to_string()
            use_proxy = self._address.proxy is not None
            target = 'proxy' if use_proxy else self._address.domain
            self._log.info('Connecting to %s (%s)',
                           target,
                           self._remote_address)

    def _check_certificate(self, _connection, certificate, errors):
        self._peer_certificate = certificate
        self._peer_certificate_errors = convert_tls_error_flags(errors)

        if self._accept_certificate():
            return True

        self.notify('bad-certificate')
        return False

    def _on_certificate_set(self, connection, _param):
        self._peer_certificate = connection.props.peer_certificate
        self._peer_certificate_errors = convert_tls_error_flags(
            connection.props.peer_certificate_errors)
        self._tls_handshake_in_progress = False
        self.notify('certificate-set')

    def _on_connect_finished(self, client, result, _user_data):
        try:
            if self._address.proxy is not None:
                self._con = client.connect_to_host_finish(result)
            elif self._address.is_service:
                self._con = client.connect_to_service_finish(result)
            elif self._address.is_host:
                self._con = client.connect_to_host_finish(result)
            else:
                raise ValueError('Address must be a service or host')
        except GLib.Error as error:
            self._log.info('Connect Error: %s', error)
            self._finalize('connection-failed')
            return

        # We use the timeout only for connecting
        self._con.get_socket().set_timeout(0)
        self._con.set_graceful_disconnect(True)
        self._con.get_socket().set_keepalive(True)

        self._local_address = self._con.get_local_address()

        self.state = TCPState.CONNECTED

        use_proxy = self._address.proxy is not None
        target = 'proxy' if use_proxy else self._address.domain
        self._log.info('Connected to %s (%s)',
                       target,
                       self._con.get_remote_address().to_string())

        self._on_connected()

    def _on_connected(self):
        self.notify('connected')
        self._read_async()

    def _remove_keepalive_timer(self):
        if self._keepalive_id is not None:
            self._log.info('Remove keepalive timer')
            GLib.source_remove(self._keepalive_id)
            self._keepalive_id = None

    def _renew_keepalive_timer(self):
        if self._con is None:
            return
        self._remove_keepalive_timer()
        self._log.info('Add keepalive timer')
        self._keepalive_id = GLib.timeout_add_seconds(5, self._send_keepalive)

    def _send_keepalive(self):
        self._log.info('Send keepalive')
        self._keepalive_id = None
        if not self._con.get_output_stream().has_pending():
            self._write_all_async(' '.encode())

    def start_tls_negotiation(self):
        self._log.info('Start TLS negotiation')
        self._tls_handshake_in_progress = True
        remote_address = self._con.get_remote_address()
        identity = Gio.NetworkAddress.new(self._address.domain,
                                          remote_address.props.port)

        tls_client = Gio.TlsClientConnection.new(self._con, identity)

        if self._address.type == ConnectionType.DIRECT_TLS:
            tls_client.set_advertised_protocols(['xmpp-client'])
        tls_client.set_validation_flags(Gio.TlsCertificateFlags.VALIDATE_ALL)
        tls_client.connect('accept-certificate', self._check_certificate)
        tls_client.connect('notify::peer-certificate', self._on_certificate_set)

        # This Wraps the Gio.TlsClientConnection and the Gio.Socket together
        # so we get back a Gio.SocketConnection
        self._con = Gio.TcpWrapperConnection.new(tls_client,
                                                 self._con.get_socket())

    def _read_async(self):
        if self._input_closed:
            return

        self._con.get_input_stream().read_bytes_async(
            READ_BUFFER_SIZE,
            GLib.PRIORITY_DEFAULT,
            self._read_cancellable,
            self._on_read_async_finish,
            None)

    def _on_read_async_finish(self, stream, result, _user_data):
        try:
            data = stream.read_bytes_finish(result)
        except GLib.Error as error:
            quark = GLib.quark_try_string('g-io-error-quark')
            if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
                if self._input_closed:
                    return

            quark = GLib.quark_try_string('g-tls-error-quark')
            if error.matches(quark, Gio.TlsError.MISC):
                if self._tls_handshake_in_progress:
                    self._log.error('Handshake failed: %s', error)
                    self._finalize('connection-failed')
                    return

            if error.matches(quark, Gio.TlsError.EOF):
                self._log.info('Incoming stream closed: TLS EOF')
                self._finalize('disconnected')
                return

            if error.matches(quark, Gio.TlsError.BAD_CERTIFICATE):
                self._log.info('Certificate Error: %s', error)
                self._finalize('disconnected')
                return

            self._log.error('Read Error: %s', error)

            if self._state not in (TCPState.DISCONNECTING,
                                   TCPState.DISCONNECTED):
                self._finalize('disconnected')
            return

        except RuntimeError as error:
            # PyGObject raises a RuntimeError when it fails to convert the
            # GError. Why it failed is printed by PyGObject
            self._log.error(error)
            return

        data = data.get_data()
        if not data:
            if self._state == TCPState.DISCONNECTING:
                self._log.info('Reveived zero data on _read_async()')
                self._finalize('disconnected')
            else:
                self._log.warning('Reveived zero data on _read_async()')
            return

        self._renew_keepalive_timer()

        self._read_buffer += data

        try:
            data, self._read_buffer = utf8_decode(self._read_buffer)
        except UnicodeDecodeError as error:
            self._log.warning(error)
            self._log.warning('read buffer: "%s"', self._read_buffer)
            self._log.warning('data: "%s"', data)
            self._finalize('disconnected')
            return

        self._log_stanza(data, received=True)

        try:
            self.notify('data-received', data)
        except Exception:
            self._log.exception('Error while executing data-received:')

        # Call next async read only after the received data is processed
        # otherwise this can lead to problems if we call
        # start_tls_negotiation() while we have a pending read which is
        # the case for START TLS because its triggered by <proceed>
        self._read_async()

    def _write_stanzas(self):
        self._write_stanza_buffer = self._write_queue
        self._write_queue = deque([])
        data = ''.join(map(str, self._write_stanza_buffer)).encode()
        self._write_all_async(data)

    def _write_all_async(self, data):
        # We have to pass data to the callback, because GLib takes no
        # reference on the passed data and python would gc collect it
        # bevor GLib has a chance to write it to the stream
        self._con.get_output_stream().write_all_async(
            data,
            GLib.PRIORITY_DEFAULT,
            None,
            self._on_write_all_async_finished,
            data)

    def _on_write_all_async_finished(self, stream, result, data):
        try:
            stream.write_all_finish(result)
        except GLib.Error as error:
            quark = GLib.quark_try_string('g-tls-error-quark')
            if error.matches(quark, Gio.TlsError.BAD_CERTIFICATE):
                self._write_stanza_buffer = None
                return

            if self._output_closed:
                self._check_for_shutdown()
                return

            self._log.error('Write Error: %s', error)
            return

        except RuntimeError as error:
            # PyGObject raises a RuntimeError when it fails to convert the
            # GError. Why it failed is printed by PyGObject
            self._log.error(error)
            return

        data = data.decode()
        self._log_stanza(data, received=False)

        if data == ' ':
            # keepalive whitespace
            self._renew_keepalive_timer()

        else:
            for stanza in self._write_stanza_buffer:
                try:
                    self.notify('data-sent', stanza)
                except Exception:
                    self._log.exception('Error while executing data-sent:')

        if self._output_closed and not self._write_queue:
            self._check_for_shutdown()
            return

        if self._write_queue:
            self._write_stanzas()

    def send(self, stanza, now=False):
        if self._state in (TCPState.DISCONNECTED, TCPState.DISCONNECTING):
            self._log.warning('send() not possible in state: %s', self._state)
            return

        if now:
            self._write_queue.appendleft(stanza)
        else:
            self._write_queue.append(stanza)

        if not self._con.get_output_stream().has_pending():
            self._write_stanzas()

    def disconnect(self):
        self._remove_keepalive_timer()
        if self.state == TCPState.CONNECTING:
            self.state = TCPState.DISCONNECTING
            self._connect_cancellable.cancel()
            return

        if self._state in (TCPState.DISCONNECTED, TCPState.DISCONNECTING):
            self._log.warning('Called disconnect on state: %s', self._state)
            return

        self.state = TCPState.DISCONNECTING
        self._finalize('disconnected')

    def _check_for_shutdown(self):
        if self._input_closed and self._output_closed:
            self._finalize('disconnected')

    def shutdown_input(self):
        self._remove_keepalive_timer()
        self._log.info('Shutdown input')
        self._input_closed = True
        self._read_cancellable.cancel()
        self._check_for_shutdown()

    def shutdown_output(self):
        self._remove_keepalive_timer()
        self.state = TCPState.DISCONNECTING
        self._log.info('Shutdown output')
        self._output_closed = True

    def _finalize(self, signal_name):
        self._remove_keepalive_timer()
        if self._con is not None:
            try:
                self._con.get_socket().shutdown(True, True)
            except GLib.Error as error:
                self._log.info(error)
        self._input_closed = True
        self._output_closed = True
        self.state = TCPState.DISCONNECTED
        self.notify(signal_name)
        self.destroy()

    def destroy(self):
        super().destroy()
        self._con = None
        self._client = None
        self._write_queue = None class WebsocketConnection(Connection):
    def __init__(self, *args, **kwargs):
        Connection.__init__(self, *args, **kwargs)

        self._session = Soup.Session()
        self._session.props.ssl_strict = False

        if self._log.getEffectiveLevel() == logging.INFO:
            self._session.add_feature(
                Soup.Logger.new(Soup.LoggerLogLevel.BODY, -1))

        self._websocket = None
        self._cancellable = Gio.Cancellable()

        self._input_closed = False
        self._output_closed = False

    def connect(self):
        self._log.info('Try to connect to %s', self._address.uri)

        self.state = TCPState.CONNECTING

        message = Soup.Message.new('GET', self._address.uri)
        message.connect('starting', self._check_certificate)
        message.set_flags(Soup.MessageFlags.NO_REDIRECT)
        self._session.websocket_connect_async(message,
                                              None,
                                              ['xmpp'],
                                              self._cancellable,
                                              self._on_connect,
                                              None)

    def _on_connect(self, session, result, _user_data):
        # TODO: check if protocol 'xmpp' is set
        try:
            self._websocket = session.websocket_connect_finish(result)
        except GLib.Error as error:
            quark = GLib.quark_try_string('g-io-error-quark')
            if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
                self._finalize('disconnected')
                return

            self._log.info('Connection Error: %s', error)
            self._finalize('connection-failed')
            return

        self._websocket.set_keepalive_interval(5)
        self._websocket.set_max_incoming_payload_size(1048576)
        self._websocket.connect('message', self._on_websocket_message)
        self._websocket.connect('closed', self._on_websocket_closed)
        self._websocket.connect('closing', self._on_websocket_closing)
        self._websocket.connect('error', self._on_websocket_error)
        self._websocket.connect('pong', self._on_websocket_pong)

        self.state = TCPState.CONNECTED
        self.notify('connected')

    def start_tls_negotiation(self):
        # Soup.Session does this automatically
        raise NotImplementedError

    def _check_certificate(self, message):
        https_used, certificate, errors = message.get_https_status()
        if not https_used and self._address.type == ConnectionType.PLAIN:
            return

        self._peer_certificate = certificate
        self._peer_certificate_errors = convert_tls_error_flags(errors)

        self.notify('certificate-set')

        if self._accept_certificate():
            return

        self.notify('bad-certificate')
        self._cancellable.cancel()

    def _on_websocket_message(self, _websocket, _type, message):
        data = message.get_data().decode()
        self._log_stanza(data)

        if self._input_closed:
            self._log.warning('Received data after stream closed')
            return

        self.notify('data-received', data)

    def _on_websocket_pong(self, _websocket, _message):
        self._log.info('Pong received')

    def _on_websocket_closed(self, websocket):
        self._log.info('Closed %s', get_websocket_close_string(websocket))
        self._finalize('disconnected')

    def _on_websocket_closing(self, _websocket):
        self._log.info('Closing')

    def _on_websocket_error(self, _websocket, error):
        self._log.error(error)
        if self._state not in (TCPState.DISCONNECTED, TCPState.DISCONNECTING):
            self._finalize('disconnected')

    def send(self, stanza, now=False):
        if self._state in (TCPState.DISCONNECTED, TCPState.DISCONNECTING):
            self._log.warning('send() not possible in state: %s', self._state)
            return

        data = str(stanza)
        self._websocket.send_text(data)
        self._log_stanza(data, received=False)
        self.notify('data-sent', stanza)

    def disconnect(self):
        if self._state == TCPState.CONNECTING:
            self.state = TCPState.DISCONNECTING
            self._cancellable.cancel()
            return

        if self._state in (TCPState.DISCONNECTED, TCPState.DISCONNECTING):
            self._log.warning('Called disconnect on state: %s', self._state)
            return

        self._websocket.close(Soup.WebsocketCloseCode.NORMAL, None)
        self.state = TCPState.DISCONNECTING

    def _check_for_shutdown(self):
        if self._input_closed and self._output_closed:
            self._websocket.close(Soup.WebsocketCloseCode.NORMAL, None)

    def shutdown_input(self):
        self._log.info('Shutdown input')
        self._input_closed = True
        self._check_for_shutdown()

    def shutdown_output(self):
        self.state = TCPState.DISCONNECTING
        self._log.info('Shutdown output')
        self._output_closed = True

    def _finalize(self, signal_name):
        self._input_closed = True
        self._output_closed = True
        self.state = TCPState.DISCONNECTED
        self.notify(signal_name)
        self.destroy()

    def destroy(self):
        super().destroy()
        self._session.abort()
        self._session = None
        self._websocket = None class Smacks:
    """
    This is Smacks is the Stream Management class. It takes care of requesting
    and sending acks. Also, it keeps track of the unhandled outgoing stanzas.

    The dispatcher has to be able to access this class to increment the
    number of handled stanzas
    """

    def __init__(self, client):
        self._client = client
        self._out_h = 0  # Outgoing stanzas handled
        self._in_h = 0  # Incoming stanzas handled
        self._acked_h = 0  # Last acked stanza

        self._uqueue = []  # Unhandled stanzas queue
        self._old_uqueue = []  # Unhandled stanzas queue of the last session

        # Max number of stanzas in queue before making a request
        self.max_queue = 0

        self._sm_supported = False
        self.enabled = False  # If SM is enabled
        self._enable_sent = False  # If we sent 'enable'
        self.resumed = False  # If the session was resumed
        self.resume_in_progress = False
        self.resume_supported = False  # Does the session support resume

        self._session_id = None
        self._location = None

        self._log = LogAdapter(log, {'context': client.log_context})

        self.register_handlers()

    @property
    def sm_supported(self):
        return self._sm_supported

    @sm_supported.setter
    def sm_supported(self, value):
        self._log.info('Server supports detected: %s', value)
        self._sm_supported = value

    @property
    def resumeable(self):
        return self._session_id is not None and self.resume_supported

    def delegate(self, stanza):
        if stanza.getNamespace() != Namespace.STREAM_MGMT:
            return
        if stanza.getName() == 'resumed':
            self._on_resumed(stanza)
        elif stanza.getName() == 'failed':
            self._on_failed(None, stanza, None)

    def register_handlers(self):
        handlers = [
            StanzaHandler(name='enabled',
                          callback=self._on_enabled,
                          xmlns=Namespace.STREAM_MGMT),
            StanzaHandler(name='failed',
                          callback=self._on_failed,
                          xmlns=Namespace.STREAM_MGMT),
            StanzaHandler(name='r',
                          callback=self._send_ack,
                          xmlns=Namespace.STREAM_MGMT),
            StanzaHandler(name='a',
                          callback=self._on_ack,
                          xmlns=Namespace.STREAM_MGMT)
        ]

        for handler in handlers:
            self._client.register_handler(handler)

    def send_enable(self):
        if not self.sm_supported:
            return

        if self._client.sm_disabled:
            return

        enable = Node(Namespace.STREAM_MGMT + ' enable', attrs={'resume': 'true'})
        self._client.send_nonza(enable, now=False)
        self._log.debug('Send enable')
        self._enable_sent = True

    def _on_enabled(self, _con, stanza, _properties):
        if self.enabled:
            self._log.error('Received "enabled", but SM is already enabled')
            return
        resume = stanza.getAttr('resume')
        if resume in ('true', 'True', '1'):
            self.resume_supported = True
            self._session_id = stanza.getAttr('id')

        self._location = stanza.getAttr('location')
        self.enabled = True
        self._log.info(
            'Received enabled, location: %s, resume supported: %s, '
            'session-id: %s', self._location, resume, self._session_id)

    def count_incoming(self, name):
        if not self.enabled:
            # Dont count while we didnt receive 'enabled'
            return
        if name in ('a', 'r', 'resumed', 'enabled'):
            return
        self._log.debug('IN, %s', name)
        self._in_h += 1

    def save_in_queue(self, stanza):
        if not self._enable_sent and not self.resumed:
            # We did not yet sent 'enable' so the server
            # will not count our stanzas
            return

        # Make a full copy so we dont run into problems when
        # the stanza is modified after sending for some reason
        # TODO: Make also copies of Protocol.Error objects
        if not isinstance(stanza, Error):
            stanza = type(stanza)(node=str(stanza))

        self._add_delay(stanza)

        self._uqueue.append(stanza)
        self._log.debug('OUT, %s', stanza.getName())
        self._out_h += 1

        if len(self._uqueue) > self.max_queue:
            self._request_ack()
        # Send an ack after 100 unacked messages
        if (self._in_h - self._acked_h) > 100:
            self._send_ack()

    def _add_delay(self, stanza):
        if stanza.getName() != 'message':
            return

        if stanza.getType() not in ('chat', 'groupchat'):
            return

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

        if stanza.getType() != 'groupchat':
            # Dont leak our JID to Groupchats
            attrs['from'] = str(self._client.get_bound_jid())

        stanza.addChild('delay', namespace=Namespace.DELAY2, attrs=attrs)

    def _resend_queue(self):
        """
        Resends unsent stanzas when a new session is established.
        This way there won't be any lost outgoing messages even on failed
        smacks resumes (but message duplicates are possible).
        """
        if not self._old_uqueue:
            return
        self._log.info('Resend %s stanzas', len(self._old_uqueue))
        for stanza in self._old_uqueue:
            # Use dispatcher so we increment the counter
            self._client.send_stanza(stanza)
        self._old_uqueue = []

    def resume_request(self):
        if self._session_id is None:
            self._log.error('Attempted to resume without a valid session id')
            return

        # Save old messages in an extra "queue" to avoid race conditions
        # and to make it possible to replay stanzas even when resuming fails
        # Add messages here (instead of overwriting) so that repeated
        # connection errors don't delete unacked stanzas
        # (uqueue should be empty in this case anyways)
        self._old_uqueue += self._uqueue
        self._uqueue = []

        resume = Node(Namespace.STREAM_MGMT + ' resume',
                      attrs={'h': self._in_h, 'previd': self._session_id})

        self._acked_h = self._in_h
        self.resume_in_progress = True
        self._client.send_nonza(resume, now=False)

    def _on_resumed(self, stanza):
        """
        Checks if the number of stanzas sent are the same as the
        number of stanzas received by the server. Resends stanzas not received
        by the server in the last session.
        """
        self._log.info('Session resumption succeeded, session-id: %s',
                       self._session_id)
        self._validate_ack(stanza, self._old_uqueue)
        # Set our out h to the h we received
        self._out_h = int(stanza.getAttr('h'))
        self.enabled = True
        self.resumed = True
        self.resume_in_progress = False
        self._client.set_state(StreamState.RESUME_SUCCESSFUL)
        self._resend_queue()

    def _send_ack(self, *args):
        ack = Node(Namespace.STREAM_MGMT + ' a', attrs={'h': self._in_h})
        self._acked_h = self._in_h
        self._log.debug('Send ack, h: %s', self._in_h)
        self._client.send_nonza(ack, now=False)

    def close_session(self):
        # We end the connection deliberately
        # Reset the state -> no resume
        self._log.info('Close session')
        self._reset_state()

    def _request_ack(self):
        request = Node(Namespace.STREAM_MGMT + ' r')
        self._log.debug('Request ack')
        self._client.send_nonza(request, now=False)

    def _on_ack(self, _stream, stanza, _properties):
        if not self.enabled:
            return
        self._log.debug('Ack received, h: %s', stanza.getAttr('h'))
        self._validate_ack(stanza, self._uqueue)

    def _validate_ack(self, stanza, queue):
        """
        Checks if the number of stanzas sent are the same as the
        number of stanzas received by the server. Pops stanzas that were
        handled by the server from the queue.
        """
        count_server = stanza.getAttr('h')
        if count_server is None:
            self._log.error('Server did not send h attribute')
            return

        count_server = int(count_server)
        diff = self._out_h - count_server
        queue_size = len(queue)
        if diff < 0:
            self._log.error(
                'Mismatch detected, our h: %d, server h: %d, queue: %d',
                self._out_h, count_server, queue_size)
            # Don't accumulate all messages in this case
            # (they would otherwise all be resent on the next reconnect)
            queue = []

        elif queue_size < diff:
            self._log.error(
                'Mismatch detected, our h: %d, server h: %d, queue: %d',
                self._out_h, count_server, queue_size)
        else:
            self._log.debug('Validate ack, our h: %d, server h: %d, queue: %d',
                            self._out_h, count_server, queue_size)
            self._log.debug('Removing %d stanzas from queue', queue_size - diff)

            while len(queue) > diff:
                queue.pop(0)

    def _on_failed(self, _stream, stanza, _properties):
        '''
        This can be called after 'enable' and 'resume'
        '''

        self._log.info('Negotiation failed')
        error_text = stanza.getTagData('text')
        if error_text is not None:
            self._log.info(error_text)

        if stanza.getTag('item-not-found') is not None:
            self._log.info('Session timed out, last server h: %s',
                           stanza.getAttr('h'))
            self._validate_ack(stanza, self._old_uqueue)
        else:
            for tag in stanza.getChildren():
                if tag.getName() != 'text':
                    self._log.info(tag.getName())

        if self.resume_in_progress:
            # Reset state before sending Bind, because otherwise stanza
            # will be counted and ack will be requested.
            # _reset_state() also resets resume_in_progress
            self._reset_state()
            self._client.set_state(StreamState.RESUME_FAILED)

        self._reset_state()

    def _reset_state(self):
        # Reset all values to default
        self._out_h = 0
        self._in_h = 0
        self._acked_h = 0

        self._uqueue = []
        self._old_uqueue = []

        self.enabled = False
        self._enable_sent = False
        self.resumed = False
        self.resume_in_progress = False
        self.resume_supported = False

        self._session_id = None
        self._location = None
        self._enable_sent = False class SASL:
    """
    Implements SASL authentication.
    """
    def __init__(self, client):
        self._client = client

        self._password = None

        self._allowed_mechs = None
        self._enabled_mechs = None
        self._method = None
        self._error = None

        self._log = LogAdapter(log, {'context': client.log_context})

    @property
    def error(self):
        return self._error

    def set_password(self, password):
        self._password = password

    @property
    def password(self):
        return self._password

    def delegate(self, stanza):
        if stanza.getNamespace() != Namespace.SASL:
            return
        if stanza.getName() == 'challenge':
            self._on_challenge(stanza)
        elif stanza.getName() == 'failure':
            self._on_failure(stanza)
        elif stanza.getName() == 'success':
            self._on_success(stanza)

    def start_auth(self, features):
        self._allowed_mechs = self._client.mechs
        self._enabled_mechs = self._allowed_mechs
        self._method = None
        self._error = None

        # -PLUS variants need TLS channel binding data
        # This is currently not supported via GLib
        self._enabled_mechs.discard('SCRAM-SHA-1-PLUS')
        self._enabled_mechs.discard('SCRAM-SHA-256-PLUS')
        # channel_binding_data = None

        if not GSSAPI_AVAILABLE:
            self._enabled_mechs.discard('GSSAPI')

        available_mechs = features.get_mechs() & self._enabled_mechs
        self._log.info('Available mechanisms: %s', available_mechs)

        domain_based_name = features.get_domain_based_name()
        if domain_based_name is not None:
            self._log.info('Found domain based name: %s', domain_based_name)

        if not available_mechs:
            self._log.error('No available auth mechanisms found')
            self._abort_auth('invalid-mechanism')
            return

        chosen_mechanism = None
        for mech in SASL_AUTH_MECHS:
            if mech in available_mechs:
                chosen_mechanism = mech
                break

        if chosen_mechanism is None:
            self._log.error('No available auth mechanisms found')
            self._abort_auth('invalid-mechanism')
            return

        self._log.info('Chosen auth mechanism: %s', chosen_mechanism)

        if chosen_mechanism in ('SCRAM-SHA-256', 'SCRAM-SHA-1', 'PLAIN'):
            if not self._password:
                self._on_sasl_finished(False, 'no-password')
                return

        # if chosen_mechanism == 'SCRAM-SHA-256-PLUS':
        #     self._method = SCRAM_SHA_256_PLUS(self._client,
        #                                       channel_binding_data)
        #     self._method.initiate(self._client.username, self._password)

        # elif chosen_mechanism == 'SCRAM-SHA-1-PLUS':
        #     self._method = SCRAM_SHA_1_PLUS(self._client,
        #                                     channel_binding_data)
        #     self._method.initiate(self._client.username, self._password)

        if chosen_mechanism == 'SCRAM-SHA-256':
            self._method = SCRAM_SHA_256(self._client, None)
            self._method.initiate(self._client.username, self._password)

        elif chosen_mechanism == 'SCRAM-SHA-1':
            self._method = SCRAM_SHA_1(self._client, None)
            self._method.initiate(self._client.username, self._password)

        elif chosen_mechanism == 'PLAIN':
            self._method = PLAIN(self._client)
            self._method.initiate(self._client.username, self._password)

        elif chosen_mechanism == 'ANONYMOUS':
            self._method = ANONYMOUS(self._client)
            self._method.initiate()

        elif chosen_mechanism == 'EXTERNAL':
            self._method = EXTERNAL(self._client)
            self._method.initiate(self._client.username, self._client.Server)

        elif chosen_mechanism == 'GSSAPI':
            self._method = GSSAPI(self._client)
            if domain_based_name:
                hostname = domain_based_name
            else:
                hostname = self._client.domain
            try:
                self._method.initiate(hostname)
            except AuthFail as error:
                self._log.error(error)
                self._abort_auth()
                return
        else:
            self._log.error('Unknown auth mech')

    def _on_challenge(self, stanza):
        try:
            self._method.response(stanza.getData())
        except AttributeError:
            self._log.info('Mechanism has no response method')
            self._abort_auth()
        except AuthFail as error:
            self._log.error(error)
            self._abort_auth()

    def _on_success(self, stanza):
        self._log.info('Successfully authenticated with remote server')
        try:
            self._method.success(stanza.getData())
        except AttributeError:
            pass
        except AuthFail as error:
            self._log.error(error)
            self._abort_auth()
            return

        self._on_sasl_finished(True, None, None)

    def _on_failure(self, stanza):
        text = stanza.getTagData('text')
        reason = 'not-authorized'
        childs = stanza.getChildren()
        for child in childs:
            name = child.getName()
            if name == 'text':
                continue
            if name in SASL_ERROR_CONDITIONS:
                reason = name
                break

        self._log.info('Failed SASL authentification: %s %s', reason, text)
        self._abort_auth(reason, text)

    def _abort_auth(self, reason='malformed-request', text=None):
        node = Node('abort', attrs={'xmlns': Namespace.SASL})
        self._client.send_nonza(node)
        self._on_sasl_finished(False, reason, text)

    def _on_sasl_finished(self, successful, reason, text=None):
        if not successful:
            self._error = (reason, text)
            self._client.set_state(StreamState.AUTH_FAILED)
        else:
            self._client.set_state(StreamState.AUTH_SUCCESSFUL) class StreamState(Enum):
    RESOLVE = 'resolve'
    RESOLVED = 'resolved'
    CONNECTING = 'connecting'
    CONNECTED = 'connected'
    DISCONNECTED = 'disconnected'
    DISCONNECTING = 'disconnecting'
    STREAM_START = 'stream start'
    WAIT_FOR_STREAM_START = 'wait for stream start'
    WAIT_FOR_FEATURES = 'wait for features'
    WAIT_FOR_TLS_PROCEED = 'wait for tls proceed'
    TLS_START_SUCCESSFUL = 'tls start successful'
    PROCEED_WITH_AUTH = 'proceed with auth'
    AUTH_SUCCESSFUL = 'auth successful'
    AUTH_FAILED = 'auth failed'
    WAIT_FOR_RESUMED = 'wait for resumed'
    RESUME_FAILED = 'resume failed'
    RESUME_SUCCESSFUL = 'resume successful'
    PROCEED_WITH_BIND = 'proceed with bind'
    BIND_SUCCESSFUL = 'bind successful'
    WAIT_FOR_BIND = 'wait for bind'
    WAIT_FOR_SESSION = 'wait for session'
    ACTIVE = 'active' class StreamError(Enum):
    PARSING = 0
    CONNECTION_FAILED = 1
    SESSION = 2
    BIND = 3
    TLS = 4
    BAD_CERTIFICATE = 5
    STREAM = 6
    SASL = 7
    REGISTER = 8
    END = 9 class ConnectionType(Enum):
    DIRECT_TLS = 'DIRECT TLS'
    START_TLS = 'START TLS'
    PLAIN = 'PLAIN'

    @property
    def is_direct_tls(self):
        return self == ConnectionType.DIRECT_TLS

    @property
    def is_start_tls(self):
        return self == ConnectionType.START_TLS

    @property
    def is_plain(self):
        return self == ConnectionType.PLAIN class ConnectionProtocol(IntEnum):
    TCP = 0
    WEBSOCKET = 1 class Mode(IntEnum):
    CLIENT = 0
    REGISTER = 1
    LOGIN_TEST = 2
    ANONYMOUS_TEST = 3

    @property
    def is_client(self):
        return self == Mode.CLIENT

    @property
    def is_register(self):
        return self == Mode.REGISTER

    @property
    def is_login_test(self):
        return self == Mode.LOGIN_TEST

    @property
    def is_anonymous_test(self):
        return self == Mode.ANONYMOUS_TEST class StanzaDispatcher(Observable):
    """
    Dispatches stanzas to handlers

    Signals:
        before-dispatch
        parsing-error
        stream-end

    """

    def __init__(self, client):
        Observable.__init__(self, log)
        self._client = client
        self._modules = {}
        self._parser = None
        self._websocket_stream_error = None

        self._log = LogAdapter(log, {'context': client.log_context})

        self._handlers = {}

        self._id_callbacks = {}
        self._dispatch_callback = None
        self._timeout_id = None

        self._stanza_types = {
            'iq': Iq,
            'message': Message,
            'presence': Presence,
            'error': StreamErrorNode,
        }

        self.invalid_chars_re = get_invalid_xml_regex()

        self._register_namespace('unknown')
        self._register_namespace(Namespace.STREAMS)
        self._register_namespace(Namespace.CLIENT)
        self._register_protocol('iq', Iq)
        self._register_protocol('presence', Presence)
        self._register_protocol('message', Message)

        self._register_modules()

    def set_dispatch_callback(self, callback):
        self._log.info('Set dispatch callback: %s', callback)
        self._dispatch_callback = callback

    def get_module(self, name):
        return self._modules[name]

    def _register_modules(self):
        self._modules['BasePresence'] = BasePresence(self._client)
        self._modules['BaseMessage'] = BaseMessage(self._client)
        self._modules['BaseIq'] = BaseIq(self._client)
        self._modules['EME'] = EME(self._client)
        self._modules['HTTPAuth'] = HTTPAuth(self._client)
        self._modules['Nickname'] = Nickname(self._client)
        self._modules['MUC'] = MUC(self._client)
        self._modules['Moderation'] = Moderation(self._client)
        self._modules['Delay'] = Delay(self._client)
        self._modules['Captcha'] = Captcha(self._client)
        self._modules['Idle'] = Idle(self._client)
        self._modules['PGPLegacy'] = PGPLegacy(self._client)
        self._modules['VCardAvatar'] = VCardAvatar(self._client)
        self._modules['EntityCaps'] = EntityCaps(self._client)
        self._modules['Blocking'] = Blocking(self._client)
        self._modules['PubSub'] = PubSub(self._client)
        self._modules['Mood'] = Mood(self._client)
        self._modules['Activity'] = Activity(self._client)
        self._modules['Tune'] = Tune(self._client)
        self._modules['Location'] = Location(self._client)
        self._modules['UserAvatar'] = UserAvatar(self._client)
        self._modules['PrivateBookmarks'] = PrivateBookmarks(self._client)
        self._modules['PEPBookmarks'] = PEPBookmarks(self._client)
        self._modules['NativeBookmarks'] = NativeBookmarks(self._client)
        self._modules['OpenPGP'] = OpenPGP(self._client)
        self._modules['OMEMO'] = OMEMO(self._client)
        self._modules['Annotations'] = Annotations(self._client)
        self._modules['Muclumbus'] = Muclumbus(self._client)
        self._modules['SoftwareVersion'] = SoftwareVersion(self._client)
        self._modules['AdHoc'] = AdHoc(self._client)
        self._modules['IBB'] = IBB(self._client)
        self._modules['Discovery'] = Discovery(self._client)
        self._modules['ChatMarkers'] = ChatMarkers(self._client)
        self._modules['Receipts'] = Receipts(self._client)
        self._modules['OOB'] = OOB(self._client)
        self._modules['Correction'] = Correction(self._client)
        self._modules['Attention'] = Attention(self._client)
        self._modules['SecurityLabels'] = SecurityLabels(self._client)
        self._modules['Chatstates'] = Chatstates(self._client)
        self._modules['Register'] = Register(self._client)
        self._modules['HTTPUpload'] = HTTPUpload(self._client)
        self._modules['MAM'] = MAM(self._client)
        self._modules['VCardTemp'] = VCardTemp(self._client)
        self._modules['VCard4'] = VCard4(self._client)
        self._modules['Ping'] = Ping(self._client)
        self._modules['Delimiter'] = Delimiter(self._client)
        self._modules['Roster'] = Roster(self._client)
        self._modules['LastActivity'] = LastActivity(self._client)
        self._modules['EntityTime'] = EntityTime(self._client)

        for instance in self._modules.values():
            for handler in instance.handlers:
                self.register_handler(handler)

    def reset_parser(self):
        if self._parser is not None:
            self._parser.dispatch = None
            self._parser.destroy()
            self._parser = None

        self._parser = NodeBuilder(dispatch_depth=2,
                                   finished=False)
        self._parser.dispatch = self.dispatch

    def replace_non_character(self, data):
        return re.sub(self.invalid_chars_re, '\ufffd', data)

    def process_data(self, data):
        # Parse incoming data

        data = self.replace_non_character(data)

        if self._client.is_websocket:
            stanza = Node(node=data)
            if is_websocket_stream_error(stanza):
                for tag in stanza.getChildren():
                    name = tag.getName()
                    if (name != 'text' and
                            tag.getNamespace() == Namespace.XMPP_STREAMS):
                        self._websocket_stream_error = name

            elif is_websocket_close(stanza):
                self._log.info('Stream <close> received')
                self.notify('stream-end', self._websocket_stream_error)
                return

            self.dispatch(stanza)
            return

        try:
            self._parser.Parse(data)
        except (ExpatError, ValueError) as error:
            self._log.error('XML parsing error: %s', error)
            self.notify('parsing-error', str(error))
            return

        # end stream:stream tag received
        if self._parser.has_received_endtag():
            self._log.info('End of stream: %s', self._parser.streamError)
            self.notify('stream-end', self._parser.streamError)
            return

    def _register_namespace(self, xmlns):
        """
        Setup handler structure for namespace
        """
        self._log.debug('Register namespace "%s"', xmlns)
        self._handlers[xmlns] = {}
        self._register_protocol('error', Protocol, xmlns=xmlns)
        self._register_protocol('unknown', Protocol, xmlns=xmlns)
        self._register_protocol('default', Protocol, xmlns=xmlns)

    def _register_protocol(self, tag_name, protocol, xmlns=None):
        """
        Register protocol for top level tag names
        """
        if xmlns is None:
            xmlns = Namespace.CLIENT
        self._log.debug('Register protocol "%s (%s)" as %s',
                        tag_name, xmlns, protocol)
        self._handlers[xmlns][tag_name] = {'type': protocol, 'default': []}

    def register_handler(self, handler):
        """
        Register handler
        """

        xmlns = handler.xmlns or Namespace.CLIENT

        typ = handler.typ
        if not typ and not handler.ns:
            typ = 'default'

        self._log.debug(
            'Register handler %s for "%s" type->%s ns->%s(%s) priority->%s',
            handler.callback, handler.name, typ, handler.ns,
            xmlns, handler.priority
        )

        if xmlns not in self._handlers:
            self._register_namespace(xmlns)
        if handler.name not in self._handlers[xmlns]:
            self._register_protocol(handler.name, Protocol, xmlns)

        specific = typ + handler.ns
        if specific not in self._handlers[xmlns][handler.name]:
            self._handlers[xmlns][handler.name][specific] = []

        self._handlers[xmlns][handler.name][specific].append(
            {'func': handler.callback,
             'priority': handler.priority,
             'specific': specific})

    def unregister_handler(self, handler):
        """
        Unregister handler
        """

        xmlns = handler.xmlns or Namespace.CLIENT

        typ = handler.typ
        if not typ and not handler.ns:
            typ = 'default'

        specific = typ + handler.ns
        try:
            self._handlers[xmlns][handler.name][specific]
        except KeyError:
            return

        for handler_dict in self._handlers[xmlns][handler.name][specific]:
            if handler_dict['func'] != handler.callback:
                continue

            try:
                self._handlers[xmlns][handler.name][specific].remove(
                    handler_dict)
            except ValueError:
                self._log.warning(
                    'Unregister failed: %s for "%s" type->%s ns->%s(%s)',
                    handler.callback, handler.name, typ, handler.ns, xmlns)
            else:
                self._log.debug(
                    'Unregister handler %s for "%s" type->%s ns->%s(%s)',
                    handler.callback, handler.name, typ, handler.ns, xmlns)

    def _default_handler(self, stanza):
        """
        Return stanza back to the sender with <feature-not-implemented/> error
        """
        if stanza.getType() in ('get', 'set'):
            self._client.send_stanza(Error(stanza, ERR_FEATURE_NOT_IMPLEMENTED))

    def dispatch(self, stanza):
        self.notify('before-dispatch', stanza)

        if self._dispatch_callback is not None:
            name = stanza.getName()
            protocol_class = self._stanza_types.get(name)
            if protocol_class is not None:
                stanza = protocol_class(node=stanza)
            self._dispatch_callback(stanza)
            return

        # Count stanza
        self._client._smacks.count_incoming(stanza.getName())

        name = stanza.getName()
        xmlns = stanza.getNamespace()

        if xmlns not in self._handlers:
            self._log.warning('Unknown namespace: %s', xmlns)
            xmlns = 'unknown'

        if name not in self._handlers[xmlns]:
            self._log.warning('Unknown stanza: %s', stanza)
            name = 'unknown'

        # Convert simplexml to Protocol object
        try:
            stanza = self._handlers[xmlns][name]['type'](node=stanza)
        except InvalidJid:
            self._log.warning('Invalid JID, ignoring stanza')
            self._log.warning(stanza)
            return

        own_jid = self._client.get_bound_jid()
        properties = get_properties_struct(name, own_jid)

        if name == 'iq':
            if stanza.getFrom() is None and own_jid is not None:
                stanza.setFrom(own_jid.bare)

        if name == 'message':
            # https://tools.ietf.org/html/rfc6120#section-8.1.1.1
            # If the stanza does not include a 'to' address then the client MUST
            # treat it as if the 'to' address were included with a value of the
            # client's full JID.

            to = stanza.getTo()
            if to is None:
                stanza.setTo(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.bare)

            # Unwrap carbon
            try:
                stanza, properties.carbon = unwrap_carbon(stanza, own_jid)
            except (InvalidFrom, InvalidJid) as exc:
                self._log.warning(exc)
                self._log.warning(stanza)
                return
            except NodeProcessed as exc:
                self._log.info(exc)
                return

            # Unwrap mam
            try:
                stanza, properties.mam = unwrap_mam(stanza, own_jid)
            except (InvalidStanza, InvalidJid) as exc:
                self._log.warning(exc)
                self._log.warning(stanza)
                return

        typ = stanza.getType()
        if name == 'message' and not typ:
            typ = 'normal'
        elif not typ:
            typ = ''

        stanza.props = stanza.getProperties()
        self._log.debug('type: %s, properties: %s', typ, stanza.props)

        # Process callbacks
        _id = stanza.getID()
        func, _timeout, user_data = self._id_callbacks.pop(
            _id, (None, None, {}))
        if user_data is None:
            user_data = {}

        if func is not None:
            try:
                func(self._client, stanza, **user_data)
            except Exception:
                self._log.exception('Error while handling stanza')
            return

        # Gather specifics depending on stanza properties
        specifics = ['default']
        if typ and typ in self._handlers[xmlns][name]:
            specifics.append(typ)
        for prop in stanza.props:
            if prop in self._handlers[xmlns][name]:
                specifics.append(prop)
            if typ and typ + prop in self._handlers[xmlns][name]:
                specifics.append(typ + prop)

        # Create the handler chain
        chain = []
        chain += self._handlers[xmlns]['default']['default']
        for specific in specifics:
            chain += self._handlers[xmlns][name][specific]

        # Sort chain with priority
        chain.sort(key=lambda x: x['priority'])

        for handler in chain:
            self._log.info('Call handler: %s', handler['func'].__qualname__)
            try:
                handler['func'](self._client, stanza, properties)
            except NodeProcessed:
                return
            except Exception:
                self._log.exception('Handler exception:')
                return

        # Stanza was not processed call default handler
        self._default_handler(stanza)

    def add_callback_for_id(self, id_, func, timeout, user_data):
        if timeout is not None and self._timeout_id is None:
            self._log.info('Add timeout check')
            self._timeout_id = GLib.timeout_add_seconds(
                1, self._timeout_check)
            timeout = time.monotonic() + timeout
        self._id_callbacks[id_] = (func, timeout, user_data)

    def _timeout_check(self):
        self._log.info('Run timeout check')
        timeouts = {}
        for id_, data in self._id_callbacks.items():
            if data[1] is not None:
                timeouts[id_] = data

        if not timeouts:
            self._log.info('Remove timeout check, no timeouts scheduled')
            self._timeout_id = None
            return False

        for id_, data in timeouts.items():
            func, timeout, user_data = data

            if user_data is None:
                user_data = {}

            if timeout < time.monotonic():
                self._id_callbacks.pop(id_)
                func(self._client, None, **user_data)
        return True

    def _remove_timeout_source(self):
        if self._timeout_id is not None:
            GLib.source_remove(self._timeout_id)
            self._timeout_id = None

    def remove_iq_callback(self, id_):
        self._id_callbacks.pop(id_, None)

    def clear_iq_callbacks(self):
        self._log.info('Clear IQ callbacks')
        self._id_callbacks.clear()

    def cleanup(self):
        self._client = None
        self._modules = {}
        self._parser = None
        self.clear_iq_callbacks()
        self._dispatch_callback = None
        self._handlers.clear()
        self._remove_timeout_source()
        self.remove_subscriptions() def get_stream_header(domain: str, lang: str, is_websocket: bool) -> str:
    if is_websocket:
        return str(WebsocketOpenHeader(domain, lang))
    header = StreamHeader(domain, lang)
    return "<?xml version='1.0'?>%s>" % str(header)[:-3] def generate_id() -> str:
    return str(uuid.uuid4()) class Observable:
    def __init__(self, log_: logging.Logger):
        self._log = log_
        self._frozen = False
        self._callbacks: defaultdict[str, list[Callable[..., Any]]] = defaultdict(list)

    def remove_subscriptions(self):
        self._callbacks = defaultdict(list)

    def subscribe(self, signal_name: str, func: Callable[..., Any]):
        self._callbacks[signal_name].append(func)

    def notify(self, signal_name: str, *args: Any, **kwargs: dict[str, Any]):
        if self._frozen:
            self._frozen = False
            return

        self._log.info('Signal: %s', signal_name)
        callbacks = self._callbacks.get(signal_name, [])
        for func in callbacks:
            func(self, signal_name, *args, **kwargs) def validate_stream_header(stanza: Node, domain: str, is_websocket: bool) -> str:
    attrs = stanza.getAttrs()
    if attrs.get('from') != domain:
        raise StanzaMalformed('Invalid from attr in stream header')

    if is_websocket:
        if attrs.get('xmlns') != Namespace.FRAMING:
            raise StanzaMalformed('Invalid namespace in stream header')
    else:
        if attrs.get('xmlns:stream') != Namespace.STREAMS:
            raise StanzaMalformed('Invalid stream namespace in stream header')
        if attrs.get('xmlns') != Namespace.CLIENT:
            raise StanzaMalformed('Invalid namespace in stream header')

    if attrs.get('version') != '1.0':
        raise StanzaMalformed('Invalid stream version in stream header')
    stream_id = attrs.get('id')
    if stream_id is None:
        raise StanzaMalformed('No stream id found in stream header')
    return stream_id class LogAdapter(LoggerAdapter):

    def set_context(self, context):
        self.extra['context'] = context

    def process(self, msg, kwargs):
        return '(%s) %s' % (self.extra['context'], msg), kwargs
import logging
from gi.repository import GLib
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import Features
from nbxmpp.protocol import StanzaMalformed
from nbxmpp.protocol import SessionRequest
from nbxmpp.protocol import BindRequest
from nbxmpp.protocol import TLSRequest
from nbxmpp.protocol import isResultNode
from nbxmpp.protocol import JID
from nbxmpp.protocol import Protocol
from nbxmpp.protocol import WebsocketCloseHeader
from nbxmpp.errors import TimeoutStanzaError
from nbxmpp.errors import StanzaError
from nbxmpp.errors import CancelledError
from nbxmpp.addresses import ServerAddresses
from nbxmpp.addresses import NoMoreAddresses
from nbxmpp.tcp import TCPConnection
from nbxmpp.websocket import WebsocketConnection
from nbxmpp.smacks import Smacks
from nbxmpp.auth import SASL
from nbxmpp.const import StreamState
from nbxmpp.const import StreamError
from nbxmpp.const import ConnectionType
from nbxmpp.const import ConnectionProtocol
from nbxmpp.const import Mode
from nbxmpp.dispatcher import StanzaDispatcher
from nbxmpp.util import get_stream_header
from nbxmpp.util import generate_id
from nbxmpp.util import Observable
from nbxmpp.util import validate_stream_header
from nbxmpp.util import LogAdapter
        self._con = None
        self._local_address = None
        self._remote_address = None
        self._mode = Mode.CLIENT

        self._ping_source_id = None
        self._tasks = []

        self._dispatcher = StanzaDispatcher(self)
        self._dispatcher.subscribe('before-dispatch', self._on_before_dispatch)
        self._dispatcher.subscribe('parsing-error', self._on_parsing_error)
        self._dispatcher.subscribe('stream-end', self._on_stream_end)

        self._smacks = Smacks(self)
        self._sasl = SASL(self)

        self._state = StreamState.DISCONNECTED

    def add_task(self, task):
        self._tasks.append(task)

    def remove_task(self, task, _context):
        try:
            self._tasks.remove(task)
        except Exception:
            pass

    @property
    def log_context(self):
        return self._log_context

    @property
    def features(self):
        return self._stream_features

    @property
    def resumeable(self):
        return self._smacks.resumeable

    @property
    def sm_supported(self):
        return self._smacks.sm_supported

    @property
    def lang(self):
        return self._lang

    @property
    def username(self):
        return self._username

    @property
    def domain(self):
        return self._domain

    @property
    def resource(self):
        return self._resource

    def set_username(self, username):
        self._username = username

    def set_domain(self, domain):
        self._domain = domain

    def set_resource(self, resource):
        self._resource = resource

    def set_mode(self, mode):
        self._mode = mode

    @property
    def custom_host(self):
        return self._custom_host

    def set_custom_host(self, host_or_uri, protocol, type_):
        if self._domain is None:
            raise ValueError('Call set_domain() first before set_custom_host()')
        self._custom_host = (host_or_uri, protocol, type_)

    def set_accepted_certificates(self, certificates):
        self._accepted_certificates = certificates

    @property
    def ignored_tls_errors(self):
        return self._ignored_tls_errors

    def set_ignored_tls_errors(self, errors):
        if errors is None:
            errors = set()
        self._ignored_tls_errors = errors

    @property
    def ignore_tls_errors(self):
        return self._ignore_tls_errors

    def set_ignore_tls_errors(self, ignore):
        self._ignore_tls_errors = ignore

    def set_password(self, password):
        self._sasl.set_password(password)

    @property
    def password(self):
        return self._sasl.password

    @property
    def peer_certificate(self):
        return self._peer_certificate, self._peer_certificate_errors

    @property
    def current_address(self):
        return self._current_address

    @property
    def current_connection_type(self):
        return self._current_address.type

    @property
    def is_websocket(self):

        return self._current_address.protocol == ConnectionProtocol.WEBSOCKET