codekingpro/portable-devtools

1#------------------------------------------------------------------------------2#3# Copyright (c) Microsoft Corporation.4# All rights reserved.5#6# This code is licensed under the MIT License.7#8# Permission is hereby granted, free of charge, to any person obtaining a copy9# of this software and associated documentation files(the "Software"), to deal10# in the Software without restriction, including without limitation the rights11# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell12# copies of the Software, and to permit persons to whom the Software is13# furnished to do so, subject to the following conditions :14#15# The above copyright notice and this permission notice shall be included in16# all copies or substantial portions of the Software.17#18# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR19# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,20# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE21# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER22# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,23# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN24# THE SOFTWARE.25#26#------------------------------------------------------------------------------27 28try:29    from urllib.parse import urlparse30except ImportError:31    from urlparse import urlparse32try:33    from xml.etree import cElementTree as ET34except ImportError:35    from xml.etree import ElementTree as ET36import logging37 38 39logger = logging.getLogger(__name__)40 41def _xpath_of_root(route_to_leaf):42    # Construct an xpath suitable to find a root node which has a specified leaf43    return '/'.join(route_to_leaf + ['..'] * (len(route_to_leaf)-1))44 45 46def send_request(mex_endpoint, http_client, **kwargs):47    mex_resp = http_client.get(mex_endpoint, **kwargs)48    mex_resp.raise_for_status()49    try:50        return Mex(mex_resp.text).get_wstrust_username_password_endpoint()51    except ET.ParseError:52        logger.exception(53            "Malformed MEX document: %s, %s", mex_resp.status_code, mex_resp.text)54        raise55 56 57class Mex(object):58 59    NS = {  # Also used by wstrust_*.py60        'wsdl': 'http://schemas.xmlsoap.org/wsdl/',61        'sp': 'http://docs.oasis-open.org/ws-sx/ws-securitypolicy/200702',62        'sp2005': 'http://schemas.xmlsoap.org/ws/2005/07/securitypolicy',63        'wsu': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd',64        'wsa': 'http://www.w3.org/2005/08/addressing',  # Duplicate?65        'wsa10': 'http://www.w3.org/2005/08/addressing',66        'http': 'http://schemas.microsoft.com/ws/06/2004/policy/http',67        'soap12': 'http://schemas.xmlsoap.org/wsdl/soap12/',68        'wsp': 'http://schemas.xmlsoap.org/ws/2004/09/policy',69        's': 'http://www.w3.org/2003/05/soap-envelope',70        'wst': 'http://docs.oasis-open.org/ws-sx/ws-trust/200512',71        'trust': "http://docs.oasis-open.org/ws-sx/ws-trust/200512",  # Duplicate?72        'saml': "urn:oasis:names:tc:SAML:1.0:assertion",73        'wst2005': 'http://schemas.xmlsoap.org/ws/2005/02/trust',  # was named "t"74        }75    ACTION_13 = 'http://docs.oasis-open.org/ws-sx/ws-trust/200512/RST/Issue'76    ACTION_2005 = 'http://schemas.xmlsoap.org/ws/2005/02/trust/RST/Issue'77 78    def __init__(self, mex_document):79        self.dom = ET.fromstring(mex_document)80 81    def _get_policy_ids(self, components_to_leaf, binding_xpath):82        id_attr = '{%s}Id' % self.NS['wsu']83        return set(["#{}".format(policy.get(id_attr))84            for policy in self.dom.findall(_xpath_of_root(components_to_leaf), self.NS)85            # If we did not find any binding, this is potentially bad.86            if policy.find(binding_xpath, self.NS) is not None])87 88    def _get_username_password_policy_ids(self):89        path = ['wsp:Policy', 'wsp:ExactlyOne', 'wsp:All',90            'sp:SignedEncryptedSupportingTokens', 'wsp:Policy',91            'sp:UsernameToken', 'wsp:Policy', 'sp:WssUsernameToken10']92        policies = self._get_policy_ids(path, './/sp:TransportBinding')93        path2005 = ['wsp:Policy', 'wsp:ExactlyOne', 'wsp:All',94            'sp2005:SignedSupportingTokens', 'wsp:Policy',95            'sp2005:UsernameToken', 'wsp:Policy', 'sp2005:WssUsernameToken10']96        policies.update(self._get_policy_ids(path2005, './/sp2005:TransportBinding'))97        return policies98 99    def _get_iwa_policy_ids(self):100        return self._get_policy_ids(101            ['wsp:Policy', 'wsp:ExactlyOne', 'wsp:All', 'http:NegotiateAuthentication'],102            './/sp2005:TransportBinding')103 104    def _get_bindings(self):105        bindings = {}  # {binding_name: {"policy_uri": "...", "version": "..."}}106        for binding in self.dom.findall("wsdl:binding", self.NS):107            if (binding.find('soap12:binding', self.NS).get("transport") !=108                    'http://schemas.xmlsoap.org/soap/http'):109                continue110            action = binding.find(111                'wsdl:operation/soap12:operation', self.NS).get("soapAction")112            for pr in binding.findall("wsp:PolicyReference", self.NS):113                bindings[binding.get("name")] = {114                    "policy_uri": pr.get("URI"), "action": action}115        return bindings116 117    def _get_endpoints(self, bindings, policy_ids):118        endpoints = []119        for port in self.dom.findall('wsdl:service/wsdl:port', self.NS):120            binding_name = port.get("binding").split(':')[-1]  # Should have 2 parts121            binding = bindings.get(binding_name)122            if binding and binding["policy_uri"] in policy_ids:123                address = port.find('wsa10:EndpointReference/wsa10:Address', self.NS)124                if address is not None and address.text.lower().startswith("https://"):125                    endpoints.append(126                        {"address": address.text, "action": binding["action"]})127        return endpoints128 129    def get_wstrust_username_password_endpoint(self):130        """Returns {"address": "https://...", "action": "the soapAction value"}"""131        endpoints = self._get_endpoints(132                self._get_bindings(), self._get_username_password_policy_ids())133        for e in endpoints:134            if e["action"] == self.ACTION_13:135                return e  # Historically, we prefer ACTION_13 a.k.a. WsTrust13136        return endpoints[0] if endpoints else None137 138