Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.10/site-packages/msal/oauth2cli/oidc.py: 37%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

111 statements  

1import json 

2import base64 

3import time 

4import secrets 

5import warnings 

6import hashlib 

7import logging 

8 

9from . import oauth2 

10 

11 

12logger = logging.getLogger(__name__) 

13 

14def decode_part(raw, encoding="utf-8"): 

15 """Decode a part of the JWT. 

16 

17 JWT is encoded by padding-less base64url, 

18 based on `JWS specs <https://tools.ietf.org/html/rfc7515#appendix-C>`_. 

19 

20 :param encoding: 

21 If you are going to decode the first 2 parts of a JWT, i.e. the header 

22 or the payload, the default value "utf-8" would work fine. 

23 If you are going to decode the last part i.e. the signature part, 

24 it is a binary string so you should use `None` as encoding here. 

25 """ 

26 raw += '=' * (-len(raw) % 4) # https://stackoverflow.com/a/32517907/728675 

27 raw = str( 

28 # On Python 2.7, argument of urlsafe_b64decode must be str, not unicode. 

29 # This is not required on Python 3. 

30 raw) 

31 output = base64.urlsafe_b64decode(raw) 

32 if encoding: 

33 output = output.decode(encoding) 

34 return output 

35 

36base64decode = decode_part # Obsolete. For backward compatibility only. 

37 

38def _epoch_to_local(epoch): 

39 return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(epoch)) 

40 

41class IdTokenError(RuntimeError): # We waised RuntimeError before, so keep it 

42 """In unlikely event of an ID token is malformed, this exception will be raised.""" 

43 def __init__(self, reason, now, claims): 

44 super(IdTokenError, self).__init__( 

45 "%s Current epoch = %s. The id_token was approximately: %s" % ( 

46 reason, _epoch_to_local(now), json.dumps(dict( 

47 claims, 

48 iat=_epoch_to_local(claims["iat"]) if claims.get("iat") else None, 

49 exp=_epoch_to_local(claims["exp"]) if claims.get("exp") else None, 

50 ), indent=2))) 

51 

52class _IdTokenTimeError(IdTokenError): # This is not intended to be raised and caught 

53 _SUGGESTION = "Make sure your computer's time and time zone are both correct." 

54 def __init__(self, reason, now, claims): 

55 super(_IdTokenTimeError, self).__init__(reason+ " " + self._SUGGESTION, now, claims) 

56 def log(self): 

57 # Influenced by JWT specs https://tools.ietf.org/html/rfc7519#section-4.1.5 

58 # and OIDC specs https://openid.net/specs/openid-connect-core-1_0.html#IDTokenValidation 

59 # We used to raise this error, but now we just log it as warning, because: 

60 # 1. If it is caused by incorrect local machine time, 

61 # then the token(s) are still correct and probably functioning, 

62 # so, there is no point to error out. 

63 # 2. If it is caused by incorrect IdP time, then it is IdP's fault, 

64 # There is not much a client can do, so, we might as well return the token(s) 

65 # and let downstream components to decide what to do. 

66 logger.warning(str(self)) 

67 

68class IdTokenIssuerError(IdTokenError): 

69 pass 

70 

71class IdTokenAudienceError(IdTokenError): 

72 pass 

73 

74class IdTokenNonceError(IdTokenError): 

75 pass 

76 

77def _decode_id_token_claims(id_token): 

78 """Decode an id_token and return its claims as a dictionary, WITHOUT validation. 

79 

80 MSAL does not validate the ID token. Per OpenID Connect, an ID token that is 

81 obtained via direct communication between the client and the token endpoint 

82 (which is how MSAL retrieves tokens) does not need to be validated by the 

83 client. Validation, if needed, is the responsibility of the application, 

84 which can perform it at the appropriate time (see the issue below). 

85 https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/911 

86 

87 ID token claims would at least contain: "iss", "sub", "aud", "exp", "iat", 

88 per `specs <https://openid.net/specs/openid-connect-core-1_0.html#IDToken>`_ 

89 and it may contain other optional content such as "preferred_username", 

90 `maybe more <https://openid.net/specs/openid-connect-core-1_0.html#Claims>`_ 

91 """ 

92 return json.loads(decode_part(id_token.split('.')[1])) 

93 

94 

95def decode_id_token(id_token, client_id=None, issuer=None, nonce=None, now=None): 

96 """Decodes and validates an id_token and returns its claims as a dictionary. 

97 

98 .. deprecated:: 1.38.0 

99 MSAL's token acquisition no longer validates the ID token, because the 

100 SDK should not perform any ID token validation 

101 (`issue #911 <https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/911>`_). 

102 This standalone helper is kept only for backward compatibility and it 

103 still performs the legacy validations described below. Prefer the 

104 ``id_token_claims`` that MSAL already returns alongside each token, or 

105 decode the token yourself, and perform any validation your app requires. 

106 

107 ID token claims would at least contain: "iss", "sub", "aud", "exp", "iat", 

108 per `specs <https://openid.net/specs/openid-connect-core-1_0.html#IDToken>`_ 

109 and it may contain other optional content such as "preferred_username", 

110 `maybe more <https://openid.net/specs/openid-connect-core-1_0.html#Claims>`_ 

111 """ 

112 warnings.warn( 

113 "decode_id_token() is deprecated. MSAL's token acquisition no longer " 

114 "validates the ID token; validation is the application's " 

115 "responsibility. This legacy helper still performs some validation. " 

116 "Prefer the id_token_claims returned alongside the token.", 

117 DeprecationWarning, stacklevel=2) 

118 decoded = json.loads(decode_part(id_token.split('.')[1])) 

119 # Based on https://openid.net/specs/openid-connect-core-1_0.html#IDTokenValidation 

120 _now = int(now or time.time()) 

121 skew = 120 # 2 minutes 

122 

123 if _now + skew < decoded.get("nbf", _now - 1): # nbf is optional per JWT specs 

124 # This is not an ID token validation, but a JWT validation 

125 # https://tools.ietf.org/html/rfc7519#section-4.1.5 

126 _IdTokenTimeError("0. The ID token is not yet valid.", _now, decoded).log() 

127 

128 if issuer and issuer != decoded["iss"]: 

129 # https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderConfigurationResponse 

130 raise IdTokenIssuerError( 

131 '2. The Issuer Identifier for the OpenID Provider, "%s", ' 

132 "(which is typically obtained during Discovery), " 

133 "MUST exactly match the value of the iss (issuer) Claim." % issuer, 

134 _now, 

135 decoded) 

136 

137 if client_id: 

138 valid_aud = client_id in decoded["aud"] if isinstance( 

139 decoded["aud"], list) else client_id == decoded["aud"] 

140 if not valid_aud: 

141 raise IdTokenAudienceError( 

142 "3. The aud (audience) claim must contain this client's client_id " 

143 '"%s", case-sensitively. Was your client_id in wrong casing?' 

144 # Some IdP accepts wrong casing request but issues right casing IDT 

145 % client_id, 

146 _now, 

147 decoded) 

148 

149 # Per specs: 

150 # 6. If the ID Token is received via direct communication between 

151 # the Client and the Token Endpoint (which it is during _obtain_token()), 

152 # the TLS server validation MAY be used to validate the issuer 

153 # in place of checking the token signature. 

154 

155 if _now - skew > decoded["exp"]: 

156 _IdTokenTimeError("9. The ID token already expires.", _now, decoded).log() 

157 

158 if nonce and nonce != decoded.get("nonce"): 

159 raise IdTokenNonceError( 

160 "11. Nonce must be the same value " 

161 "as the one that was sent in the Authentication Request.", 

162 _now, 

163 decoded) 

164 

165 return decoded 

166 

167 

168def _nonce_hash(nonce): 

169 # https://openid.net/specs/openid-connect-core-1_0.html#NonceNotes 

170 return hashlib.sha256(nonce.encode("ascii")).hexdigest() 

171 

172 

173class Prompt(object): 

174 """This class defines the constant strings for prompt parameter. 

175 

176 The values are based on 

177 https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest 

178 """ 

179 NONE = "none" 

180 LOGIN = "login" 

181 CONSENT = "consent" 

182 SELECT_ACCOUNT = "select_account" 

183 CREATE = "create" # Defined in https://openid.net/specs/openid-connect-prompt-create-1_0.html#PromptParameter 

184 

185 

186class Client(oauth2.Client): 

187 """OpenID Connect is a layer on top of the OAuth2. 

188 

189 See its specs at https://openid.net/connect/ 

190 """ 

191 

192 def decode_id_token(self, id_token, nonce=None): 

193 """See :func:`~decode_id_token`. 

194 

195 .. deprecated:: 1.38.0 

196 MSAL's token acquisition no longer validates the ID token. This 

197 method is kept for backward compatibility and still performs the 

198 legacy validations. See :func:`~decode_id_token`. 

199 """ 

200 return decode_id_token( 

201 id_token, nonce=nonce, 

202 client_id=self.client_id, issuer=self.configuration.get("issuer")) 

203 

204 def _obtain_token(self, grant_type, *args, **kwargs): 

205 """The result will also contain one more key "id_token_claims", 

206 whose value is a dictionary of the (non-validated) ID token claims. 

207 """ 

208 ret = super(Client, self)._obtain_token(grant_type, *args, **kwargs) 

209 if "id_token" in ret: 

210 # MSAL does not validate the ID token. It only decodes the claims, 

211 # so that downstream components can build accounts, etc. 

212 # https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/911 

213 ret["id_token_claims"] = _decode_id_token_claims(ret["id_token"]) 

214 return ret 

215 

216 def build_auth_request_uri(self, response_type, nonce=None, **kwargs): 

217 """Generate an authorization uri to be visited by resource owner. 

218 

219 Return value and all other parameters are the same as 

220 :func:`oauth2.Client.build_auth_request_uri`, plus new parameter(s): 

221 

222 :param nonce: 

223 A hard-to-guess string used to mitigate replay attacks. See also 

224 `OIDC specs <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

225 """ 

226 warnings.warn("Use initiate_auth_code_flow() instead", DeprecationWarning) 

227 return super(Client, self).build_auth_request_uri( 

228 response_type, nonce=nonce, **kwargs) 

229 

230 def obtain_token_by_authorization_code(self, code, nonce=None, **kwargs): 

231 """Get a token via authorization code. a.k.a. Authorization Code Grant. 

232 

233 Return value and all other parameters are the same as 

234 :func:`oauth2.Client.obtain_token_by_authorization_code`, 

235 plus new parameter(s): 

236 

237 :param nonce: 

238 If you provided a nonce when calling :func:`build_auth_request_uri`, 

239 same nonce should also be provided here, so that we'll validate it. 

240 An exception will be raised if the nonce in id token mismatches. 

241 """ 

242 warnings.warn( 

243 "Use obtain_token_by_auth_code_flow() instead", DeprecationWarning) 

244 result = super(Client, self).obtain_token_by_authorization_code( 

245 code, **kwargs) 

246 nonce_in_id_token = result.get("id_token_claims", {}).get("nonce") 

247 if "id_token_claims" in result and nonce and nonce != nonce_in_id_token: 

248 raise ValueError( 

249 'The nonce in id token ("%s") should match your nonce ("%s")' % 

250 (nonce_in_id_token, nonce)) 

251 return result 

252 

253 def initiate_auth_code_flow( 

254 self, 

255 scope=None, 

256 **kwargs): 

257 """Initiate an auth code flow. 

258 

259 It provides nonce protection automatically. 

260 

261 :param list scope: 

262 A list of strings, e.g. ["profile", "email", ...]. 

263 This method will automatically send ["openid"] to the wire, 

264 although it won't modify your input list. 

265 

266 See :func:`oauth2.Client.initiate_auth_code_flow` in parent class 

267 for descriptions on other parameters and return value. 

268 """ 

269 if "id_token" in kwargs.get("response_type", ""): 

270 # Implicit grant would cause auth response coming back in #fragment, 

271 # but fragment won't reach a web service. 

272 raise ValueError('response_type="id_token ..." is not allowed') 

273 _scope = list(scope) if scope else [] # We won't modify input parameter 

274 if "openid" not in _scope: 

275 # "If no openid scope value is present, 

276 # the request may still be a valid OAuth 2.0 request, 

277 # but is not an OpenID Connect request." -- OIDC Core Specs, 3.1.2.2 

278 # https://openid.net/specs/openid-connect-core-1_0.html#AuthRequestValidation 

279 # Here we just automatically add it. If the caller do not want id_token, 

280 # they should simply go with oauth2.Client. 

281 _scope.append("openid") 

282 nonce = secrets.token_urlsafe(16) 

283 flow = super(Client, self).initiate_auth_code_flow( 

284 scope=_scope, nonce=_nonce_hash(nonce), **kwargs) 

285 flow["nonce"] = nonce 

286 if kwargs.get("max_age") is not None: 

287 flow["max_age"] = kwargs["max_age"] 

288 return flow 

289 

290 def obtain_token_by_auth_code_flow(self, auth_code_flow, auth_response, **kwargs): 

291 """Validate the auth_response being redirected back, and then obtain tokens, 

292 including ID token which can be used for user sign in. 

293 

294 Internally, it implements nonce to mitigate replay attack. 

295 It also implements PKCE to mitigate the auth code interception attack. 

296 

297 See :func:`oauth2.Client.obtain_token_by_auth_code_flow` in parent class 

298 for descriptions on other parameters and return value. 

299 """ 

300 result = super(Client, self).obtain_token_by_auth_code_flow( 

301 auth_code_flow, auth_response, **kwargs) 

302 if "id_token_claims" in result: 

303 nonce_in_id_token = result.get("id_token_claims", {}).get("nonce") 

304 expected_hash = _nonce_hash(auth_code_flow["nonce"]) 

305 if nonce_in_id_token != expected_hash: 

306 raise RuntimeError( 

307 'The nonce in id token ("%s") should match our nonce ("%s")' % 

308 (nonce_in_id_token, expected_hash)) 

309 

310 if auth_code_flow.get("max_age") is not None: 

311 auth_time = result.get("id_token_claims", {}).get("auth_time") 

312 if not auth_time: 

313 raise RuntimeError( 

314 "13. max_age was requested, ID token should contain auth_time") 

315 now = int(time.time()) 

316 skew = 120 # 2 minutes. Hardcoded, for now 

317 if now - skew > auth_time + auth_code_flow["max_age"]: 

318 raise RuntimeError( 

319 "13. auth_time ({auth_time}) was requested, " 

320 "by using max_age ({max_age}) parameter, " 

321 "and now ({now}) too much time has elasped " 

322 "since last end-user authentication. " 

323 "The ID token was: {id_token}".format( 

324 auth_time=auth_time, 

325 max_age=auth_code_flow["max_age"], 

326 now=now, 

327 id_token=json.dumps(result["id_token_claims"], indent=2), 

328 )) 

329 return result 

330 

331 def obtain_token_by_browser( 

332 self, 

333 display=None, 

334 prompt=None, 

335 max_age=None, 

336 ui_locales=None, 

337 id_token_hint=None, # It is relevant, 

338 # because this library exposes raw ID token 

339 login_hint=None, 

340 acr_values=None, 

341 **kwargs): 

342 """A native app can use this method to obtain token via a local browser. 

343 

344 Internally, it implements nonce to mitigate replay attack. 

345 It also implements PKCE to mitigate the auth code interception attack. 

346 

347 :param string display: Defined in 

348 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

349 :param string prompt: Defined in 

350 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

351 You can find the valid string values defined in :class:`oidc.Prompt`. 

352 

353 :param int max_age: Defined in 

354 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

355 :param string ui_locales: Defined in 

356 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

357 :param string id_token_hint: Defined in 

358 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

359 :param string login_hint: Defined in 

360 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

361 :param string acr_values: Defined in 

362 `OIDC <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_. 

363 

364 See :func:`oauth2.Client.obtain_token_by_browser` in parent class 

365 for descriptions on other parameters and return value. 

366 """ 

367 filtered_params = {k:v for k, v in dict( 

368 prompt=" ".join(prompt) if isinstance(prompt, (list, tuple)) else prompt, 

369 display=display, 

370 max_age=max_age, 

371 ui_locales=ui_locales, 

372 id_token_hint=id_token_hint, 

373 login_hint=login_hint, 

374 acr_values=acr_values, 

375 ).items() if v is not None} # Filter out None values 

376 return super(Client, self).obtain_token_by_browser( 

377 auth_params=dict(kwargs.pop("auth_params", {}), **filtered_params), 

378 **kwargs) 

379