Source code for aioxmpp.sasl

########################################################################
# File name: sasl.py
# This file is part of: aioxmpp
#
# LICENSE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program.  If not, see
# <http://www.gnu.org/licenses/>.
#
########################################################################
"""
:mod:`~aioxmpp.sasl` -- SASL helpers
####################################

This module is used to implement SASL in :mod:`aioxmpp.security_layer`. It
provides a state machine for use by the different SASL mechanisms and
implementations of some SASL mechansims.

It provides an XMPP adaptor for :mod:`aiosasl`.

.. autoclass:: SASLXMPPInterface

The XSOs for SASL authentication can be found in :mod:`aioxmpp.nonza`.

"""

import asyncio
import logging

import aiosasl

from . import errors, protocol, nonza

logger = logging.getLogger(__name__)


[docs]class SASLXMPPInterface(aiosasl.SASLInterface): def __init__(self, xmlstream): super().__init__() self.xmlstream = xmlstream self.timeout = None @asyncio.coroutine def _send_sasl_node_and_wait_for(self, node): node = yield from protocol.send_and_wait_for( self.xmlstream, [node], [ nonza.SASLChallenge, nonza.SASLFailure, nonza.SASLSuccess ], timeout=self.timeout ) state = node.TAG[1] if state == "failure": xmpp_error = node.condition[1] text = node.text raise aiosasl.SASLFailure(xmpp_error, text=text) if hasattr(node, "payload"): payload = node.payload else: payload = None return state, payload @asyncio.coroutine def initiate(self, mechanism, payload=None): return (yield from self._send_sasl_node_and_wait_for( nonza.SASLAuth(mechanism=mechanism, payload=payload))) @asyncio.coroutine def respond(self, payload): return (yield from self._send_sasl_node_and_wait_for( nonza.SASLResponse(payload=payload) )) @asyncio.coroutine def abort(self): try: next_state, payload = yield from self._send_sasl_node_and_wait_for( nonza.SASLAbort() ) except aiosasl.SASLFailure as err: self._state = "failure" if err.opaque_error != "aborted": raise return "failure", None else: raise aiosasl.SASLFailure( "aborted", text="unexpected non-failure after abort: " "{}".format(self._state) )