Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/msal/mex.py: 31%

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 copy 

9# of this software and associated documentation files(the "Software"), to deal 

10# in the Software without restriction, including without limitation the rights 

11# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell 

12# copies of the Software, and to permit persons to whom the Software is 

13# furnished to do so, subject to the following conditions : 

14# 

15# The above copyright notice and this permission notice shall be included in 

16# all copies or substantial portions of the Software. 

17# 

18# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

19# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

20# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE 

21# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

22# 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 IN 

24# THE SOFTWARE. 

25# 

26#------------------------------------------------------------------------------ 

27 

28try: 

29 from urllib.parse import urlparse 

30except: 

31 from urlparse import urlparse 

32try: 

33 from xml.etree import cElementTree as ET 

34except ImportError: 

35 from xml.etree import ElementTree as ET 

36import logging 

37 

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 leaf 

43 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 raise 

55 

56 

57class Mex(object): 

58 

59 NS = { # Also used by wstrust_*.py 

60 '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 policies 

98 

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 continue 

110 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 bindings 

116 

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 parts 

121 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 endpoints 

128 

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

136 return endpoints[0] if endpoints else None 

137