python3-pysaml2 3.0.0-3ubuntu1 / usr / lib / python3 / dist-packages / saml2 / response.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#

import calendar
import logging
import six
from saml2.samlp import STATUS_VERSION_MISMATCH
from saml2.samlp import STATUS_AUTHN_FAILED
from saml2.samlp import STATUS_INVALID_ATTR_NAME_OR_VALUE
from saml2.samlp import STATUS_INVALID_NAMEID_POLICY
from saml2.samlp import STATUS_NO_AUTHN_CONTEXT
from saml2.samlp import STATUS_NO_AVAILABLE_IDP
from saml2.samlp import STATUS_NO_PASSIVE
from saml2.samlp import STATUS_NO_SUPPORTED_IDP
from saml2.samlp import STATUS_PARTIAL_LOGOUT
from saml2.samlp import STATUS_PROXY_COUNT_EXCEEDED
from saml2.samlp import STATUS_REQUEST_DENIED
from saml2.samlp import STATUS_REQUEST_UNSUPPORTED
from saml2.samlp import STATUS_REQUEST_VERSION_DEPRECATED
from saml2.samlp import STATUS_REQUEST_VERSION_TOO_HIGH
from saml2.samlp import STATUS_REQUEST_VERSION_TOO_LOW
from saml2.samlp import STATUS_RESOURCE_NOT_RECOGNIZED
from saml2.samlp import STATUS_TOO_MANY_RESPONSES
from saml2.samlp import STATUS_UNKNOWN_ATTR_PROFILE
from saml2.samlp import STATUS_UNKNOWN_PRINCIPAL
from saml2.samlp import STATUS_UNSUPPORTED_BINDING
from saml2.samlp import STATUS_RESPONDER

from saml2 import xmldsig as ds
from saml2 import xmlenc as xenc

from saml2 import samlp
from saml2 import class_name
from saml2 import saml
from saml2 import extension_elements_to_elements
from saml2 import SAMLError
from saml2 import time_util

from saml2.s_utils import RequestVersionTooLow
from saml2.s_utils import RequestVersionTooHigh
from saml2.saml import attribute_from_string, XSI_TYPE
from saml2.saml import SCM_BEARER
from saml2.saml import SCM_HOLDER_OF_KEY
from saml2.saml import SCM_SENDER_VOUCHES
from saml2.saml import encrypted_attribute_from_string
from saml2.sigver import security_context
from saml2.sigver import SignatureError
from saml2.sigver import signed
from saml2.attribute_converter import to_local
from saml2.time_util import str_to_time, later_than

from saml2.validate import validate_on_or_after
from saml2.validate import validate_before
from saml2.validate import valid_instance
from saml2.validate import valid_address
from saml2.validate import NotValid

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------


class IncorrectlySigned(SAMLError):
    pass


class DecryptionFailed(SAMLError):
    pass


class VerificationError(SAMLError):
    pass


class StatusError(SAMLError):
    pass


class UnsolicitedResponse(SAMLError):
    pass


class StatusVersionMismatch(StatusError):
    pass


class StatusAuthnFailed(StatusError):
    pass


class StatusInvalidAttrNameOrValue(StatusError):
    pass


class StatusInvalidNameidPolicy(StatusError):
    pass


class StatusNoAuthnContext(StatusError):
    pass


class StatusNoAvailableIdp(StatusError):
    pass


class StatusNoPassive(StatusError):
    pass


class StatusNoSupportedIdp(StatusError):
    pass


class StatusPartialLogout(StatusError):
    pass


class StatusProxyCountExceeded(StatusError):
    pass


class StatusRequestDenied(StatusError):
    pass


class StatusRequestUnsupported(StatusError):
    pass


class StatusRequestVersionDeprecated(StatusError):
    pass


class StatusRequestVersionTooHigh(StatusError):
    pass


class StatusRequestVersionTooLow(StatusError):
    pass


class StatusResourceNotRecognized(StatusError):
    pass


class StatusTooManyResponses(StatusError):
    pass


class StatusUnknownAttrProfile(StatusError):
    pass


class StatusUnknownPrincipal(StatusError):
    pass


class StatusUnsupportedBinding(StatusError):
    pass

class StatusResponder(StatusError):
    pass

STATUSCODE2EXCEPTION = {
    STATUS_VERSION_MISMATCH: StatusVersionMismatch,
    STATUS_AUTHN_FAILED: StatusAuthnFailed,
    STATUS_INVALID_ATTR_NAME_OR_VALUE: StatusInvalidAttrNameOrValue,
    STATUS_INVALID_NAMEID_POLICY: StatusInvalidNameidPolicy,
    STATUS_NO_AUTHN_CONTEXT: StatusNoAuthnContext,
    STATUS_NO_AVAILABLE_IDP: StatusNoAvailableIdp,
    STATUS_NO_PASSIVE: StatusNoPassive,
    STATUS_NO_SUPPORTED_IDP: StatusNoSupportedIdp,
    STATUS_PARTIAL_LOGOUT: StatusPartialLogout,
    STATUS_PROXY_COUNT_EXCEEDED: StatusProxyCountExceeded,
    STATUS_REQUEST_DENIED: StatusRequestDenied,
    STATUS_REQUEST_UNSUPPORTED: StatusRequestUnsupported,
    STATUS_REQUEST_VERSION_DEPRECATED: StatusRequestVersionDeprecated,
    STATUS_REQUEST_VERSION_TOO_HIGH: StatusRequestVersionTooHigh,
    STATUS_REQUEST_VERSION_TOO_LOW: StatusRequestVersionTooLow,
    STATUS_RESOURCE_NOT_RECOGNIZED: StatusResourceNotRecognized,
    STATUS_TOO_MANY_RESPONSES: StatusTooManyResponses,
    STATUS_UNKNOWN_ATTR_PROFILE: StatusUnknownAttrProfile,
    STATUS_UNKNOWN_PRINCIPAL: StatusUnknownPrincipal,
    STATUS_UNSUPPORTED_BINDING: StatusUnsupportedBinding,
    STATUS_RESPONDER: StatusResponder,
}
# ---------------------------------------------------------------------------


def _dummy(_):
    return None


def for_me(conditions, myself):
    """ Am I among the intended audiences """

    if not conditions.audience_restriction:  # No audience restriction
        return True

    for restriction in conditions.audience_restriction:
        if not restriction.audience:
            continue
        for audience in restriction.audience:
            if audience.text.strip() == myself:
                return True
            else:
                #print("Not for me: %s != %s" % (audience.text.strip(), myself))
                pass

    return False


def authn_response(conf, return_addrs, outstanding_queries=None, timeslack=0,
                   asynchop=True, allow_unsolicited=False,
                   want_assertions_signed=False):
    sec = security_context(conf)
    if not timeslack:
        try:
            timeslack = int(conf.accepted_time_diff)
        except TypeError:
            timeslack = 0

    return AuthnResponse(sec, conf.attribute_converters, conf.entityid,
                         return_addrs, outstanding_queries, timeslack,
                         asynchop=asynchop, allow_unsolicited=allow_unsolicited,
                         want_assertions_signed=want_assertions_signed)


# comes in over SOAP so synchronous
def attribute_response(conf, return_addrs, timeslack=0, asynchop=False,
                       test=False):
    sec = security_context(conf)
    if not timeslack:
        try:
            timeslack = int(conf.accepted_time_diff)
        except TypeError:
            timeslack = 0

    return AttributeResponse(sec, conf.attribute_converters, conf.entityid,
                             return_addrs, timeslack, asynchop=asynchop,
                             test=test)


class StatusResponse(object):
    msgtype = "status_response"

    def __init__(self, sec_context, return_addrs=None, timeslack=0,
                 request_id=0, asynchop=True):
        self.sec = sec_context
        self.return_addrs = return_addrs

        self.timeslack = timeslack
        self.request_id = request_id

        self.xmlstr = ""
        self.origxml = ""
        self.name_id = None
        self.response = None
        self.not_on_or_after = 0
        self.in_response_to = None
        self.signature_check = self.sec.correctly_signed_response
        self.require_signature = False
        self.require_response_signature = False
        self.not_signed = False
        self.asynchop = asynchop

    def _clear(self):
        self.xmlstr = ""
        self.name_id = None
        self.response = None
        self.not_on_or_after = 0

    def _postamble(self):
        if not self.response:
            logger.error("Response was not correctly signed")
            if self.xmlstr:
                logger.info(self.xmlstr)
            raise IncorrectlySigned()

        logger.debug("response: %s" % (self.response,))

        try:
            valid_instance(self.response)
        except NotValid as exc:
            logger.error("Not valid response: %s" % exc.args[0])
            self._clear()
            return self

        self.in_response_to = self.response.in_response_to
        return self

    def load_instance(self, instance):
        if signed(instance):
            # This will check signature on Assertion which is the default
            try:
                self.response = self.sec.check_signature(instance)
            except SignatureError:
                # The response as a whole might be signed or not
                self.response = self.sec.check_signature(
                    instance, samlp.NAMESPACE + ":Response")
        else:
            self.not_signed = True
            self.response = instance

        return self._postamble()

    def _loads(self, xmldata, decode=True, origxml=None):

        # own copy
        self.xmlstr = xmldata[:]
        logger.debug("xmlstr: %s" % (self.xmlstr,))
        if origxml:
            self.origxml = origxml
        else:
            self.origxml = self.xmlstr

        try:
            self.response = self.signature_check(
                xmldata, origdoc=origxml, must=self.require_signature,
                require_response_signature=self.require_response_signature)

        except TypeError:
            raise
        except SignatureError:
            raise
        except Exception as excp:
            logger.exception("EXCEPTION: %s", excp)
            raise

        #print("<", self.response)

        return self._postamble()

    def status_ok(self):
        if self.response.status:
            status = self.response.status
            logger.info("status: %s" % (status,))
            if status.status_code.value != samlp.STATUS_SUCCESS:
                logger.info("Not successful operation: %s" % status)
                if status.status_code.status_code:
                    excep = STATUSCODE2EXCEPTION[
                        status.status_code.status_code.value]
                else:
                    excep = StatusError
                if status.status_message:
                    msg = status.status_message.text
                else:
                    try:
                        msg = status.status_code.status_code.value
                    except Exception:
                        msg = "Unknown error"
                raise excep(
                    "%s from %s" % (msg, status.status_code.value,))
        return True

    def issue_instant_ok(self):
        """ Check that the response was issued at a reasonable time """
        upper = time_util.shift_time(time_util.time_in_a_while(days=1),
                                     self.timeslack).timetuple()
        lower = time_util.shift_time(time_util.time_a_while_ago(days=1),
                                     -self.timeslack).timetuple()
        # print("issue_instant: %s" % self.response.issue_instant)
        # print("%s < x < %s" % (lower, upper))
        issued_at = str_to_time(self.response.issue_instant)
        return lower < issued_at < upper

    def _verify(self):
        if self.request_id and self.in_response_to and \
                        self.in_response_to != self.request_id:
            logger.error("Not the id I expected: %s != %s" % (
                self.in_response_to, self.request_id))
            return None

        try:
            assert self.response.version == "2.0"
        except AssertionError:
            _ver = float(self.response.version)
            if _ver < 2.0:
                raise RequestVersionTooLow()
            else:
                raise RequestVersionTooHigh()

        if self.asynchop:
            if self.response.destination and \
                    self.response.destination not in self.return_addrs:
                logger.error("%s not in %s" % (self.response.destination,
                                               self.return_addrs))
                return None

        assert self.issue_instant_ok()
        assert self.status_ok()
        return self

    def loads(self, xmldata, decode=True, origxml=None):
        return self._loads(xmldata, decode, origxml)

    def verify(self, keys=None):
        try:
            return self._verify()
        except AssertionError:
            logger.exception("verify")
            return None

    def update(self, mold):
        self.xmlstr = mold.xmlstr
        self.in_response_to = mold.in_response_to
        self.response = mold.response

    def issuer(self):
        return self.response.issuer.text.strip()


class LogoutResponse(StatusResponse):
    msgtype = "logout_response"

    def __init__(self, sec_context, return_addrs=None, timeslack=0,
                 asynchop=True):
        StatusResponse.__init__(self, sec_context, return_addrs, timeslack,
                                asynchop=asynchop)
        self.signature_check = self.sec.correctly_signed_logout_response


class NameIDMappingResponse(StatusResponse):
    msgtype = "name_id_mapping_response"

    def __init__(self, sec_context, return_addrs=None, timeslack=0,
                 request_id=0, asynchop=True):
        StatusResponse.__init__(self, sec_context, return_addrs, timeslack,
                                request_id, asynchop)
        self.signature_check = self.sec\
            .correctly_signed_name_id_mapping_response


class ManageNameIDResponse(StatusResponse):
    msgtype = "manage_name_id_response"

    def __init__(self, sec_context, return_addrs=None, timeslack=0,
                 request_id=0, asynchop=True):
        StatusResponse.__init__(self, sec_context, return_addrs, timeslack,
                                request_id, asynchop)
        self.signature_check = self.sec.correctly_signed_manage_name_id_response


# ----------------------------------------------------------------------------


class AuthnResponse(StatusResponse):
    """ This is where all the profile compliance is checked.
    This one does saml2int compliance. """
    msgtype = "authn_response"

    def __init__(self, sec_context, attribute_converters, entity_id,
                 return_addrs=None, outstanding_queries=None,
                 timeslack=0, asynchop=True, allow_unsolicited=False,
                 test=False, allow_unknown_attributes=False,
                 want_assertions_signed=False, want_response_signed=False,
                 **kwargs):

        StatusResponse.__init__(self, sec_context, return_addrs, timeslack,
                                asynchop=asynchop)
        self.entity_id = entity_id
        self.attribute_converters = attribute_converters
        if outstanding_queries:
            self.outstanding_queries = outstanding_queries
        else:
            self.outstanding_queries = {}
        self.context = "AuthnReq"
        self.came_from = None
        self.ava = None
        self.assertion = None
        self.assertions = []
        self.session_not_on_or_after = 0
        self.allow_unsolicited = allow_unsolicited
        self.require_signature = want_assertions_signed
        self.require_response_signature = want_response_signed
        self.test = test
        self.allow_unknown_attributes = allow_unknown_attributes
        #
        try:
            self.extension_schema = kwargs["extension_schema"]
        except KeyError:
            self.extension_schema = {}

    def check_subject_confirmation_in_response_to(self, irp):
        for assertion in self.response.assertion:
            for _sc in assertion.subject.subject_confirmation:
                try:
                    assert _sc.subject_confirmation_data.in_response_to == irp
                except AssertionError:
                    return False

        return True

    def loads(self, xmldata, decode=True, origxml=None):
        self._loads(xmldata, decode, origxml)

        if self.asynchop:
            if self.in_response_to in self.outstanding_queries:
                self.came_from = self.outstanding_queries[self.in_response_to]
                #del self.outstanding_queries[self.in_response_to]
                try:
                    if not self.check_subject_confirmation_in_response_to(
                            self.in_response_to):
                        logger.exception(
                            "Unsolicited response %s" % self.in_response_to)
                        raise UnsolicitedResponse(
                            "Unsolicited response: %s" % self.in_response_to)
                except AttributeError:
                    pass
            elif self.allow_unsolicited:
                pass
            else:
                logger.exception(
                    "Unsolicited response %s" % self.in_response_to)
                raise UnsolicitedResponse(
                    "Unsolicited response: %s" % self.in_response_to)

        return self

    def clear(self):
        self._clear()
        self.came_from = None
        self.ava = None
        self.assertion = None

    def authn_statement_ok(self, optional=False):
        try:
            # the assertion MUST contain one AuthNStatement
            assert len(self.assertion.authn_statement) == 1
        except AssertionError:
            if optional:
                return True
            else:
                logger.error("No AuthnStatement")
                raise

        authn_statement = self.assertion.authn_statement[0]
        if authn_statement.session_not_on_or_after:
            if validate_on_or_after(authn_statement.session_not_on_or_after,
                                    self.timeslack):
                self.session_not_on_or_after = calendar.timegm(
                    time_util.str_to_time(
                        authn_statement.session_not_on_or_after))
            else:
                return False
        return True
        # check authn_statement.session_index

    def condition_ok(self, lax=False):
        if self.test:
            lax = True

        # The Identity Provider MUST include a <saml:Conditions> element
        assert self.assertion.conditions
        conditions = self.assertion.conditions

        logger.debug("conditions: %s" % conditions)

        # if no sub-elements or elements are supplied, then the
        # assertion is considered to be valid.
        if not conditions.keyswv():
            return True

        # if both are present NotBefore must be earlier than NotOnOrAfter
        if conditions.not_before and conditions.not_on_or_after:
            if not later_than(conditions.not_on_or_after,
                              conditions.not_before):
                return False

        try:
            if conditions.not_on_or_after:
                self.not_on_or_after = validate_on_or_after(
                    conditions.not_on_or_after, self.timeslack)
            if conditions.not_before:
                validate_before(conditions.not_before, self.timeslack)
        except Exception as excp:
            logger.error("Exception on conditions: %s" % (excp,))
            if not lax:
                raise
            else:
                self.not_on_or_after = 0

        if not self.allow_unsolicited:
            if not for_me(conditions, self.entity_id):
                if not lax:
                    raise Exception("Not for me!!!")

        if conditions.condition:  # extra conditions
            for cond in conditions.condition:
                try:
                    if cond.extension_attributes[
                        XSI_TYPE] in self.extension_schema:
                        pass
                    else:
                        raise Exception("Unknown condition")
                except KeyError:
                    raise Exception("Missing xsi:type specification")

        return True

    def decrypt_attributes(self, attribute_statement):
        """
        Decrypts possible encrypted attributes and adds the decrypts to the
        list of attributes.

        :param attribute_statement: A SAML.AttributeStatement which might
            contain both encrypted attributes and attributes.
        """
        #        _node_name = [
        #            "urn:oasis:names:tc:SAML:2.0:assertion:EncryptedData",
        #            "urn:oasis:names:tc:SAML:2.0:assertion:EncryptedAttribute"]

        for encattr in attribute_statement.encrypted_attribute:
            if not encattr.encrypted_key:
                _decr = self.sec.decrypt(encattr.encrypted_data)
                _attr = attribute_from_string(_decr)
                attribute_statement.attribute.append(_attr)
            else:
                _decr = self.sec.decrypt(encattr)
                enc_attr = encrypted_attribute_from_string(_decr)
                attrlist = enc_attr.extensions_as_elements("Attribute", saml)
                attribute_statement.attribute.extend(attrlist)

    def read_attribute_statement(self, attr_statem):
        logger.debug("Attribute Statement: %s" % (attr_statem,))
        for aconv in self.attribute_converters:
            logger.debug("Converts name format: %s" % (aconv.name_format,))

        self.decrypt_attributes(attr_statem)
        return to_local(self.attribute_converters, attr_statem,
                       self.allow_unknown_attributes)

    def get_identity(self):
        """ The assertion can contain zero or one attributeStatements

        """
        ava = {}
        for _assertion in self.assertions:
            if _assertion.advice:
                if _assertion.advice.assertion:
                    for tmp_assertion in _assertion.advice.assertion:
                        if tmp_assertion.attribute_statement:
                            assert len(tmp_assertion.attribute_statement) == 1
                            ava.update(self.read_attribute_statement(tmp_assertion.attribute_statement[0]))
            if _assertion.attribute_statement:
                assert len(_assertion.attribute_statement) == 1
                _attr_statem = _assertion.attribute_statement[0]
                ava.update(self.read_attribute_statement(_attr_statem))
            if not ava:
                logger.error("Missing Attribute Statement")
        return ava

    def _bearer_confirmed(self, data):
        if not data:
            return False

        if data.address:
            if not valid_address(data.address):
                return False
                # verify that I got it from the correct sender

        # These two will raise exception if untrue
        validate_on_or_after(data.not_on_or_after, self.timeslack)
        validate_before(data.not_before, self.timeslack)

        # not_before must be < not_on_or_after
        if not later_than(data.not_on_or_after, data.not_before):
            return False

        if self.asynchop and self.came_from is None:
            if data.in_response_to:
                if data.in_response_to in self.outstanding_queries:
                    self.came_from = self.outstanding_queries[
                        data.in_response_to]
                    #del self.outstanding_queries[data.in_response_to]
                elif self.allow_unsolicited:
                    pass
                else:
                    # This is where I don't allow unsolicited reponses
                    # Either in_response_to == None or has a value I don't
                    # recognize
                    logger.debug("in response to: '%s'" % data.in_response_to)
                    logger.info("outstanding queries: %s" % (
                        self.outstanding_queries.keys(),))
                    raise Exception(
                        "Combination of session id and requestURI I don't "
                        "recall")
        return True

    def _holder_of_key_confirmed(self, data):
        if not data:
            return False

        has_keyinfo = False
        for element in extension_elements_to_elements(data,
                                                      [samlp, saml, xenc, ds]):
            if isinstance(element, ds.KeyInfo):
                has_keyinfo = True

        return has_keyinfo

    def get_subject(self):
        """ The assertion must contain a Subject
        """
        assert self.assertion.subject
        subject = self.assertion.subject
        subjconf = []
        for subject_confirmation in subject.subject_confirmation:
            _data = subject_confirmation.subject_confirmation_data

            if subject_confirmation.method == SCM_BEARER:
                if not self._bearer_confirmed(_data):
                    continue
            elif subject_confirmation.method == SCM_HOLDER_OF_KEY:
                if not self._holder_of_key_confirmed(_data):
                    continue
            elif subject_confirmation.method == SCM_SENDER_VOUCHES:
                pass
            else:
                raise ValueError("Unknown subject confirmation method: %s" % (
                    subject_confirmation.method,))

            subjconf.append(subject_confirmation)

        if not subjconf:
            raise VerificationError("No valid subject confirmation")

        subject.subject_confirmation = subjconf

        # The subject must contain a name_id
        try:
            assert subject.name_id
            self.name_id = subject.name_id
        except AssertionError:
            if subject.encrypted_id:
                # decrypt encrypted ID
                _name_id_str = self.sec.decrypt(
                    subject.encrypted_id.encrypted_data.to_string())
                _name_id = saml.name_id_from_string(_name_id_str)
                self.name_id = _name_id
            else:
                raise VerificationError("Missing NameID")

        logger.info("Subject NameID: %s" % self.name_id)
        return self.name_id

    def _assertion(self, assertion, verified=False):
        """
        Check the assertion
        :param assertion:
        :return: True/False depending on if the assertion is sane or not
        """

        if not hasattr(assertion, 'signature') or not assertion.signature:
            logger.debug("unsigned")
            if self.require_signature:
                raise SignatureError("Signature missing for assertion")
        else:
            logger.debug("signed")
            if not verified:
                try:
                    self.sec.check_signature(assertion, class_name(assertion),self.xmlstr)
                except Exception as exc:
                    logger.error("correctly_signed_response: %s" % exc)
                    raise

        self.assertion = assertion
        logger.debug("assertion context: %s" % (self.context,))
        logger.debug("assertion keys: %s" % (assertion.keyswv()))
        logger.debug("outstanding_queries: %s" % (self.outstanding_queries,))

        #if self.context == "AuthnReq" or self.context == "AttrQuery":
        if self.context == "AuthnReq":
            self.authn_statement_ok()
        #        elif self.context == "AttrQuery":
        #            self.authn_statement_ok(True)

        if not self.condition_ok():
            raise VerificationError("Condition not OK")

        logger.debug("--- Getting Identity ---")

        #if self.context == "AuthnReq" or self.context == "AttrQuery":
        #    self.ava = self.get_identity()
        #    logger.debug("--- AVA: %s" % (self.ava,))

        try:
            self.get_subject()
            if self.asynchop:
                if self.allow_unsolicited:
                    pass
                elif self.came_from is None:
                    raise VerificationError("Came from")
            return True
        except Exception:
            logger.exception("get subject")
            raise

    def decrypt_assertions(self, encrypted_assertions, decr_txt, issuer=None, verified=False):
        """ Moves the decrypted assertion from the encrypted assertion to a list.

        :param encrypted_assertions: A list of encrypted assertions.
        :param decr_txt: The string representation containing the decrypted data. Used when verifying signatures.
        :param issuer: The issuer of the response.
        :param verified: If True do not verify signatures, otherwise verify the signature if it exists.
        :return: A list of decrypted assertions.
        """
        res = []
        for encrypted_assertion in encrypted_assertions:
            if encrypted_assertion.extension_elements:
                assertions = extension_elements_to_elements(
                    encrypted_assertion.extension_elements, [saml, samlp])
                for assertion in assertions:
                    if assertion.signature and not verified:
                        if not self.sec.check_signature(
                                assertion, origdoc=decr_txt,
                                node_name=class_name(assertion), issuer=issuer):
                            logger.error("Failed to verify signature on '%s'" % assertion)
                            raise SignatureError()
                    res.append(assertion)
        return res

    def find_encrypt_data_assertion(self, enc_assertions):
        """ Verifies if a list of encrypted assertions contains encrypted data.

        :param enc_assertions: A list of encrypted assertions.
        :return: True encrypted data exists otherwise false.
        """
        for _assertion in enc_assertions:
                if _assertion.encrypted_data is not None:
                    return True

    def find_encrypt_data_assertion_list(self, _assertions):
        """ Verifies if a list of assertions contains encrypted data in the advice element.

        :param _assertions: A list of assertions.
        :return: True encrypted data exists otherwise false.
        """
        for _assertion in _assertions:
            if _assertion.advice:
                if _assertion.advice.encrypted_assertion:
                    res = self.find_encrypt_data_assertion(_assertion.advice.encrypted_assertion)
                    if res:
                        return True

    def find_encrypt_data(self, resp):
        """ Verifies if a saml response contains encrypted assertions with encrypted data.

        :param resp: A saml response.
        :return: True encrypted data exists otherwise false.
        """
        _has_encrypt_data = False
        if resp.encrypted_assertion:
            res = self.find_encrypt_data_assertion(resp.encrypted_assertion)
            if res:
                return True
        if resp.assertion:
            for tmp_assertion in resp.assertion:
                if tmp_assertion.advice:
                    if tmp_assertion.advice.encrypted_assertion:
                        res = self.find_encrypt_data_assertion(tmp_assertion.advice.encrypted_assertion)
                        if res:
                            return True
        return False

    def parse_assertion(self, keys=None):
        """ Parse the assertions for a saml response.

        :param keys: A string representing a RSA key or a list of strings containing RSA keys.
        :return: True if the assertions are parsed otherwise False.
        """
        if self.context == "AuthnQuery":
            # can contain one or more assertions
            pass
        else:  # This is a saml2int limitation
            try:
                assert len(self.response.assertion) == 1 or \
                    len(self.response.encrypted_assertion) == 1
            except AssertionError:
                raise Exception("No assertion part")

        has_encrypted_assertions = self.find_encrypt_data(self.response) #self.response.encrypted_assertion
        #if not has_encrypted_assertions and self.response.assertion:
        #    for tmp_assertion in self.response.assertion:
        #        if tmp_assertion.advice:
        #            if tmp_assertion.advice.encrypted_assertion:
        #                has_encrypted_assertions = True
        #                break

        if self.response.assertion:
            logger.debug("***Unencrypted assertion***")
            for assertion in self.response.assertion:
                if not self._assertion(assertion, False):
                    return False

        if has_encrypted_assertions:
            _enc_assertions = []
            logger.debug("***Encrypted assertion/-s***")
            decr_text = "%s" % self.response
            resp = self.response
            decr_text_old = None
            while self.find_encrypt_data(resp) and decr_text_old != decr_text:
                decr_text_old = decr_text
                decr_text = self.sec.decrypt_keys(decr_text, keys)
                resp = samlp.response_from_string(decr_text)
            _enc_assertions = self.decrypt_assertions(resp.encrypted_assertion, decr_text)
            decr_text_old = None
            while (self.find_encrypt_data(resp) or self.find_encrypt_data_assertion_list(_enc_assertions)) and \
                            decr_text_old != decr_text:
                decr_text_old = decr_text
                decr_text = self.sec.decrypt_keys(decr_text, keys)
                resp = samlp.response_from_string(decr_text)
                _enc_assertions = self.decrypt_assertions(resp.encrypted_assertion, decr_text, verified=True)
            #_enc_assertions = self.decrypt_assertions(resp.encrypted_assertion, decr_text, verified=True)
            all_assertions = _enc_assertions
            if resp.assertion:
                all_assertions = all_assertions + resp.assertion
            if len(all_assertions) > 0:
                for tmp_ass in all_assertions:
                    if tmp_ass.advice and tmp_ass.advice.encrypted_assertion:

                        advice_res = self.decrypt_assertions(tmp_ass.advice.encrypted_assertion,
                                                             decr_text,
                                                             tmp_ass.issuer)
                        if tmp_ass.advice.assertion:
                            tmp_ass.advice.assertion.extend(advice_res)
                        else:
                            tmp_ass.advice.assertion = advice_res
                        if len(advice_res) > 0:
                            tmp_ass.advice.encrypted_assertion = []
            self.response.assertion = resp.assertion
            for assertion in _enc_assertions:
                if not self._assertion(assertion, True):
                    return False
                else:
                    self.assertions.append(assertion)

            self.xmlstr = decr_text
            if len(_enc_assertions) > 0:
                self.response.encrypted_assertion = []

        if self.response.assertion:
            for assertion in self.response.assertion:
                self.assertions.append(assertion)

        if self.assertions and len(self.assertions) > 0:
            self.assertion = self.assertions[0]

        if self.context == "AuthnReq" or self.context == "AttrQuery":
            self.ava = self.get_identity()
            logger.debug("--- AVA: %s" % (self.ava,))

        return True

    def verify(self, keys=None):
        """ Verify that the assertion is syntactically correct and
        the signature is correct if present.
        :param key_file: If not the default key file should be used this is it.
        """

        try:
            res = self._verify()
        except AssertionError as err:
            logger.error("Verification error on the response: %s" % err)
            raise
        else:
            if res is None:
                return None

        if not isinstance(self.response, samlp.Response):
            return self

        if self.parse_assertion(keys):
            return self
        else:
            logger.error("Could not parse the assertion")
            return None

    def session_id(self):
        """ Returns the SessionID of the response """
        return self.response.in_response_to

    def id(self):
        """ Return the ID of the response """
        return self.response.id

    def authn_info(self):
        res = []
        for astat in self.assertion.authn_statement:
            context = astat.authn_context
            if context:
                try:
                    aclass = context.authn_context_class_ref.text
                except AttributeError:
                    aclass = ""
                try:
                    authn_auth = [a.text for a in
                                  context.authenticating_authority]
                except AttributeError:
                    authn_auth = []
                res.append((aclass, authn_auth))
        return res

    def authz_decision_info(self):
        res = {"permit": [], "deny": [], "indeterminate": []}
        for adstat in self.assertion.authz_decision_statement:
            # one of 'Permit', 'Deny', 'Indeterminate'
            res[adstat.decision.text.lower()] = adstat
        return res

    def session_info(self):
        """ Returns a predefined set of information gleened from the
        response.
        :returns: Dictionary with information
        """
        if self.session_not_on_or_after > 0:
            nooa = self.session_not_on_or_after
        else:
            nooa = self.not_on_or_after

        if self.context == "AuthzQuery":
            return {"name_id": self.name_id, "came_from": self.came_from,
                    "issuer": self.issuer(), "not_on_or_after": nooa,
                    "authz_decision_info": self.authz_decision_info()}
        else:
            return {"ava": self.ava, "name_id": self.name_id,
                    "came_from": self.came_from, "issuer": self.issuer(),
                    "not_on_or_after": nooa, "authn_info": self.authn_info()}

    def __str__(self):
        if not isinstance(self.xmlstr, six.string_types):
            return "%s" % self.xmlstr.decode("utf-8")
        return "%s" % self.xmlstr

    def verify_attesting_entity(self, address):
        """
        Assumes one assertion. At least one address specification has to be
        correct.

        :param address: IP address of attesting entity
        :return: True/False
        """

        correct = 0
        for subject_conf in self.assertion.subject.subject_confirmation:
            if subject_conf.subject_confirmation_data is None:
                correct += 1  # In reality undefined
            elif subject_conf.subject_confirmation_data.address:
                if subject_conf.subject_confirmation_data.address == address:
                    correct += 1
            else:
                correct += 1

        if correct:
            return True
        else:
            return False


class AuthnQueryResponse(AuthnResponse):
    msgtype = "authn_query_response"

    def __init__(self, sec_context, attribute_converters, entity_id,
                 return_addrs=None, timeslack=0, asynchop=False, test=False):
        AuthnResponse.__init__(self, sec_context, attribute_converters,
                               entity_id, return_addrs, timeslack=timeslack,
                               asynchop=asynchop, test=test)
        self.entity_id = entity_id
        self.attribute_converters = attribute_converters
        self.assertion = None
        self.context = "AuthnQuery"

    def condition_ok(self, lax=False):  # Should I care about conditions ?
        return True


class AttributeResponse(AuthnResponse):
    msgtype = "attribute_response"

    def __init__(self, sec_context, attribute_converters, entity_id,
                 return_addrs=None, timeslack=0, asynchop=False, test=False):
        AuthnResponse.__init__(self, sec_context, attribute_converters,
                               entity_id, return_addrs, timeslack=timeslack,
                               asynchop=asynchop, test=test)
        self.entity_id = entity_id
        self.attribute_converters = attribute_converters
        self.assertion = None
        self.context = "AttrQuery"


class AuthzResponse(AuthnResponse):
    """ A successful response will be in the form of assertions containing
    authorization decision statements."""
    msgtype = "authz_decision_response"

    def __init__(self, sec_context, attribute_converters, entity_id,
                 return_addrs=None, timeslack=0, asynchop=False):
        AuthnResponse.__init__(self, sec_context, attribute_converters,
                               entity_id, return_addrs, timeslack=timeslack,
                               asynchop=asynchop)
        self.entity_id = entity_id
        self.attribute_converters = attribute_converters
        self.assertion = None
        self.context = "AuthzQuery"


class ArtifactResponse(AuthnResponse):
    msgtype = "artifact_response"

    def __init__(self, sec_context, attribute_converters, entity_id,
                 return_addrs=None, timeslack=0, asynchop=False, test=False):
        AuthnResponse.__init__(self, sec_context, attribute_converters,
                               entity_id, return_addrs, timeslack=timeslack,
                               asynchop=asynchop, test=test)
        self.entity_id = entity_id
        self.attribute_converters = attribute_converters
        self.assertion = None
        self.context = "ArtifactResolve"


def response_factory(xmlstr, conf, return_addrs=None, outstanding_queries=None,
                     timeslack=0, decode=True, request_id=0, origxml=None,
                     asynchop=True, allow_unsolicited=False,
                     want_assertions_signed=False):
    sec_context = security_context(conf)
    if not timeslack:
        try:
            timeslack = int(conf.accepted_time_diff)
        except TypeError:
            timeslack = 0

    attribute_converters = conf.attribute_converters
    entity_id = conf.entityid
    extension_schema = conf.extension_schema

    response = StatusResponse(sec_context, return_addrs, timeslack, request_id,
                              asynchop)
    try:
        response.loads(xmlstr, decode, origxml)
        if response.response.assertion or response.response.encrypted_assertion:
            authnresp = AuthnResponse(sec_context, attribute_converters,
                                      entity_id, return_addrs,
                                      outstanding_queries, timeslack, asynchop,
                                      allow_unsolicited,
                                      extension_schema=extension_schema,
                                      want_assertions_signed=want_assertions_signed)
            authnresp.update(response)
            return authnresp
    except TypeError:
        response.signature_check = sec_context.correctly_signed_logout_response
        response.loads(xmlstr, decode, origxml)
        logoutresp = LogoutResponse(sec_context, return_addrs, timeslack,
                                    asynchop=asynchop)
        logoutresp.update(response)
        return logoutresp

    return response


# ===========================================================================
# A class of it's own


class AssertionIDResponse(object):
    msgtype = "assertion_id_response"

    def __init__(self, sec_context, attribute_converters, timeslack=0,
                 **kwargs):

        self.sec = sec_context
        self.timeslack = timeslack
        self.xmlstr = ""
        self.origxml = ""
        self.name_id = ""
        self.response = None
        self.not_signed = False
        self.attribute_converters = attribute_converters
        self.assertion = None
        self.context = "AssertionIdResponse"
        self.signature_check = self.sec.correctly_signed_assertion_id_response

    def loads(self, xmldata, decode=True, origxml=None):
        # own copy
        self.xmlstr = xmldata[:]
        logger.debug("xmlstr: %s" % (self.xmlstr,))
        self.origxml = origxml

        try:
            self.response = self.signature_check(xmldata, origdoc=origxml)
            self.assertion = self.response
        except TypeError:
            raise
        except SignatureError:
            raise
        except Exception as excp:
            logger.exception("EXCEPTION: %s", excp)
            raise

        #print("<", self.response)

        return self._postamble()

    def verify(self, keys=None):
        try:
            valid_instance(self.response)
        except NotValid as exc:
            logger.error("Not valid response: %s" % exc.args[0])
            raise
        return self

    def _postamble(self):
        if not self.response:
            logger.error("Response was not correctly signed")
            if self.xmlstr:
                logger.info(self.xmlstr)
            raise IncorrectlySigned()

        logger.debug("response: %s" % (self.response,))

        return self