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

1"""This OAuth2 client implementation aims to be spec-compliant, and generic.""" 

2# OAuth2 spec https://tools.ietf.org/html/rfc6749 

3 

4import json 

5try: 

6 from urllib.parse import urlencode, parse_qs, quote_plus, urlparse, urlunparse 

7except ImportError: 

8 from urlparse import parse_qs, urlparse, urlunparse 

9 from urllib import urlencode, quote_plus 

10import inspect 

11import logging 

12import warnings 

13import time 

14import base64 

15import sys 

16import functools 

17import secrets 

18import string 

19import hashlib 

20 

21from .authcode import AuthCodeReceiver as _AuthCodeReceiver 

22 

23try: 

24 PermissionError # Available in Python 3 

25except: 

26 from socket import error as PermissionError # Workaround for Python 2 

27 

28 

29string_types = (str,) if sys.version_info[0] >= 3 else (basestring, ) 

30 

31 

32class BrowserInteractionTimeoutError(RuntimeError): 

33 pass 

34 

35class BaseClient(object): 

36 # This low-level interface works. Yet you'll find its sub-class 

37 # more friendly to remind you what parameters are needed in each scenario. 

38 # More on Client Types at https://tools.ietf.org/html/rfc6749#section-2.1 

39 

40 @staticmethod 

41 def encode_saml_assertion(assertion): 

42 return base64.urlsafe_b64encode(assertion).rstrip(b'=') # Per RFC 7522 

43 

44 CLIENT_ASSERTION_TYPE_JWT = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" 

45 CLIENT_ASSERTION_TYPE_SAML2 = "urn:ietf:params:oauth:client-assertion-type:saml2-bearer" 

46 client_assertion_encoders = {CLIENT_ASSERTION_TYPE_SAML2: encode_saml_assertion} 

47 

48 @property 

49 def session(self): 

50 warnings.warn("Will be gone in next major release", DeprecationWarning) 

51 return self._http_client 

52 

53 @session.setter 

54 def session(self, value): 

55 warnings.warn("Will be gone in next major release", DeprecationWarning) 

56 self._http_client = value 

57 

58 

59 def __init__( 

60 self, 

61 server_configuration, # type: dict 

62 client_id, # type: str 

63 http_client=None, # We insert it here to match the upcoming async API 

64 client_secret=None, # type: Optional[str] 

65 client_assertion=None, # type: Union[bytes, callable, None] 

66 client_assertion_type=None, # type: Optional[str] 

67 default_headers=None, # type: Optional[dict] 

68 default_body=None, # type: Optional[dict] 

69 verify=None, # type: Union[str, True, False, None] 

70 proxies=None, # type: Optional[dict] 

71 timeout=None, # type: Union[tuple, float, None] 

72 ): 

73 """Initialize a client object to talk all the OAuth2 grants to the server. 

74 

75 Args: 

76 server_configuration (dict): 

77 It contains the configuration (i.e. metadata) of the auth server. 

78 The actual content typically contains keys like 

79 "authorization_endpoint", "token_endpoint", etc.. 

80 Based on RFC 8414 (https://tools.ietf.org/html/rfc8414), 

81 you can probably fetch it online from either 

82 https://example.com/.../.well-known/oauth-authorization-server 

83 or 

84 https://example.com/.../.well-known/openid-configuration 

85 client_id (str): The client's id, issued by the authorization server 

86 

87 http_client (http.HttpClient): 

88 Your implementation of abstract class :class:`http.HttpClient`. 

89 Defaults to a requests session instance. 

90 

91 There is no session-wide `timeout` parameter defined here. 

92 Timeout behavior is determined by the actual http client you use. 

93 If you happen to use Requests, it disallows session-wide timeout 

94 (https://github.com/psf/requests/issues/3341). The workaround is: 

95 

96 s = requests.Session() 

97 s.request = functools.partial(s.request, timeout=3) 

98 

99 and then feed that patched session instance to this class. 

100 

101 client_secret (str): Triggers HTTP AUTH for Confidential Client 

102 client_assertion (bytes, callable): 

103 The client assertion to authenticate this client, per RFC 7521. 

104 It can be a raw SAML2 assertion (we will base64 encode it for you), 

105 or a raw JWT assertion in bytes (which we will relay to http layer). 

106 It can also be a callable (recommended), 

107 so that we will do lazy creation of an assertion. 

108 

109 The callable may accept zero arguments (legacy) or one 

110 required positional argument. Callables whose positional 

111 parameters all have default values (e.g. 

112 ``lambda token=token: token``) are treated as zero-arg. 

113 When the callable declares a required positional parameter, 

114 it will receive a dict containing ``"client_id"``, 

115 ``"token_endpoint"``, and optionally ``"fmi_path"`` 

116 (when an FMI path is set on the current request). 

117 client_assertion_type (str): 

118 The type of your :attr:`client_assertion` parameter. 

119 It is typically the value of :attr:`CLIENT_ASSERTION_TYPE_SAML2` or 

120 :attr:`CLIENT_ASSERTION_TYPE_JWT`, the only two defined in RFC 7521. 

121 default_headers (dict): 

122 A dict to be sent in each request header. 

123 It is not required by OAuth2 specs, but you may use it for telemetry. 

124 default_body (dict): 

125 A dict to be sent in each token request body. For example, 

126 you could choose to set this as {"client_secret": "your secret"} 

127 if your authorization server wants it to be in the request body 

128 (rather than in the request header). 

129 

130 verify (boolean): 

131 It will be passed to the 

132 `verify parameter in the underlying requests library 

133 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#ssl-cert-verification>`_. 

134 When leaving it with default value (None), we will use True instead. 

135 

136 This does not apply if you have chosen to pass your own Http client. 

137 

138 proxies (dict): 

139 It will be passed to the 

140 `proxies parameter in the underlying requests library 

141 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#proxies>`_. 

142 

143 This does not apply if you have chosen to pass your own Http client. 

144 

145 timeout (object): 

146 It will be passed to the 

147 `timeout parameter in the underlying requests library 

148 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#timeouts>`_. 

149 

150 This does not apply if you have chosen to pass your own Http client. 

151 

152 """ 

153 if not server_configuration: 

154 raise ValueError("Missing input parameter server_configuration") 

155 # Generally we should have client_id, but we tolerate its absence 

156 self.configuration = server_configuration 

157 self.client_id = client_id 

158 self.client_secret = client_secret 

159 self.client_assertion = client_assertion 

160 self.default_headers = default_headers or {} 

161 self.default_body = default_body or {} 

162 if client_assertion_type is not None: 

163 self.default_body["client_assertion_type"] = client_assertion_type 

164 self.logger = logging.getLogger(__name__) 

165 if http_client: 

166 if verify is not None or proxies is not None or timeout is not None: 

167 raise ValueError( 

168 "verify, proxies, or timeout is not allowed " 

169 "when http_client is in use") 

170 self._http_client = http_client 

171 else: 

172 import requests # Lazy loading 

173 

174 self._http_client = requests.Session() 

175 self._http_client.verify = True if verify is None else verify 

176 self._http_client.proxies = proxies 

177 self._http_client.request = functools.partial( 

178 # A workaround for requests not supporting session-wide timeout 

179 self._http_client.request, timeout=timeout) 

180 

181 @staticmethod 

182 def _accepts_context(func): 

183 """Check if a callable requires at least one positional argument. 

184 

185 Returns True only when the callable has a positional parameter 

186 **without** a default value. This ensures that legacy zero-arg 

187 callables — including ``lambda token=token: token`` patterns 

188 where every positional param has a default — are still invoked 

189 with no arguments. 

190 """ 

191 try: 

192 sig = inspect.signature(func) 

193 for p in sig.parameters.values(): 

194 if p.kind in ( 

195 inspect.Parameter.POSITIONAL_ONLY, 

196 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

197 ) and p.default is inspect.Parameter.empty: 

198 return True 

199 return False 

200 except (ValueError, TypeError): 

201 return False # Signature not inspectable; treat as zero-arg 

202 

203 def _invoke_assertion_callable(self, assertion_callable, data=None): 

204 """Invoke an assertion callable, passing context if it accepts one.""" 

205 if self._accepts_context(assertion_callable): 

206 context = { 

207 "client_id": self.client_id, 

208 "token_endpoint": self.configuration.get( 

209 "token_endpoint", ""), 

210 } 

211 if data and data.get("fmi_path"): 

212 context["fmi_path"] = data["fmi_path"] 

213 return assertion_callable(context) 

214 return assertion_callable() 

215 

216 def _build_auth_request_params(self, response_type, **kwargs): 

217 # response_type is a string defined in 

218 # https://tools.ietf.org/html/rfc6749#section-3.1.1 

219 # or it can be a space-delimited string as defined in 

220 # https://tools.ietf.org/html/rfc6749#section-8.4 

221 response_type = self._stringify(response_type) 

222 

223 params = {'client_id': self.client_id, 'response_type': response_type} 

224 params.update(kwargs) # Note: None values will override params 

225 params = {k: v for k, v in params.items() if v is not None} # clean up 

226 if params.get('scope'): 

227 params['scope'] = self._stringify(params['scope']) 

228 return params # A dict suitable to be used in http request 

229 

230 def _obtain_token( # The verb "obtain" is influenced by OAUTH2 RFC 6749 

231 self, grant_type, 

232 params=None, # a dict to be sent as query string to the endpoint 

233 data=None, # All relevant data, which will go into the http body 

234 headers=None, # a dict to be sent as request headers 

235 post=None, # A callable to replace requests.post(), for testing. 

236 # Such as: lambda url, **kwargs: 

237 # Mock(status_code=200, text='{}') 

238 **kwargs # Relay all extra parameters to underlying requests 

239 ): # Returns the json object came from the OAUTH2 response 

240 _data = {'client_id': self.client_id, 'grant_type': grant_type} 

241 

242 if self.default_body.get("client_assertion_type") and self.client_assertion: 

243 # See https://tools.ietf.org/html/rfc7521#section-4.2 

244 encoder = self.client_assertion_encoders.get( 

245 self.default_body["client_assertion_type"], lambda a: a) 

246 if callable(self.client_assertion): 

247 raw = self._invoke_assertion_callable(self.client_assertion, data) 

248 else: 

249 raw = self.client_assertion 

250 _data["client_assertion"] = encoder(raw) 

251 

252 _data.update(self.default_body) # It may contain authen parameters 

253 _data.update(data or {}) # So the content in data param prevails 

254 _data = {k: v for k, v in _data.items() if v} # Clean up None values 

255 

256 # "client_claims" is a cache-key-only pseudo-parameter: callers merge its 

257 # value into the standard "claims" body parameter upstream, and it is kept 

258 # in the request data solely so it contributes to the extended cache key. 

259 # It must not be sent on the wire. Popping it here (from this method's own 

260 # local copy) keeps the wire body clean while the caller's data dict — used 

261 # for the cache-add event — still carries it. 

262 _data.pop("client_claims", None) 

263 

264 if _data.get('scope'): 

265 _data['scope'] = self._stringify(_data['scope']) 

266 

267 _headers = {'Accept': 'application/json'} 

268 _headers.update(self.default_headers) 

269 _headers.update(headers or {}) 

270 

271 # Quoted from https://tools.ietf.org/html/rfc6749#section-2.3.1 

272 # Clients in possession of a client password MAY use the HTTP Basic 

273 # authentication. 

274 # Alternatively, (but NOT RECOMMENDED,) 

275 # the authorization server MAY support including the 

276 # client credentials in the request-body using the following 

277 # parameters: client_id, client_secret. 

278 if self.client_secret and self.client_id: 

279 _headers["Authorization"] = "Basic " + base64.b64encode("{}:{}".format( 

280 # Per https://tools.ietf.org/html/rfc6749#section-2.3.1 

281 # client_id and client_secret needs to be encoded by 

282 # "application/x-www-form-urlencoded" 

283 # https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4.1 

284 # BEFORE they are fed into HTTP Basic Authentication 

285 quote_plus(self.client_id), quote_plus(self.client_secret) 

286 ).encode("ascii")).decode("ascii") 

287 

288 if "token_endpoint" not in self.configuration: 

289 raise ValueError("token_endpoint not found in configuration") 

290 resp = (post or self._http_client.post)( 

291 self.configuration["token_endpoint"], 

292 headers=_headers, params=params, data=_data, 

293 **kwargs) 

294 if resp.status_code >= 500: 

295 resp.raise_for_status() # TODO: Will probably retry here 

296 try: 

297 # The spec (https://tools.ietf.org/html/rfc6749#section-5.2) says 

298 # even an error response will be a valid json structure, 

299 # so we simply return it here, without needing to invent an exception. 

300 return json.loads(resp.text) 

301 except ValueError: 

302 self.logger.exception( 

303 "Token response is not in json format: %s", resp.text) 

304 raise 

305 

306 def obtain_token_by_refresh_token(self, refresh_token, scope=None, **kwargs): 

307 # type: (str, Union[str, list, set, tuple]) -> dict 

308 """Obtain an access token via a refresh token. 

309 

310 :param refresh_token: The refresh token issued to the client 

311 :param scope: If omitted, is treated as equal to the scope originally 

312 granted by the resource owner, 

313 according to https://tools.ietf.org/html/rfc6749#section-6 

314 """ 

315 assert isinstance(refresh_token, string_types) 

316 data = kwargs.pop('data', {}) 

317 data.update(refresh_token=refresh_token, scope=scope) 

318 return self._obtain_token("refresh_token", data=data, **kwargs) 

319 

320 def _stringify(self, sequence): 

321 if isinstance(sequence, (list, set, tuple)): 

322 return ' '.join(sorted(sequence)) # normalizing it, ascendingly 

323 return sequence # as-is 

324 

325 

326def _scope_set(scope): 

327 assert scope is None or isinstance(scope, (list, set, tuple)) 

328 return set(scope) if scope else set([]) 

329 

330 

331def _generate_pkce_code_verifier(length=43): 

332 assert 43 <= length <= 128 

333 alphabet = string.ascii_letters + string.digits + "-._~" 

334 verifier = "".join( # https://tools.ietf.org/html/rfc7636#section-4.1 

335 secrets.choice(alphabet) for _ in range(length)) 

336 code_challenge = ( 

337 # https://tools.ietf.org/html/rfc7636#section-4.2 

338 base64.urlsafe_b64encode(hashlib.sha256(verifier.encode("ascii")).digest()) 

339 .rstrip(b"=")) # Required by https://tools.ietf.org/html/rfc7636#section-3 

340 return { 

341 "code_verifier": verifier, 

342 "transformation": "S256", # In Python, sha256 is always available 

343 "code_challenge": code_challenge, 

344 } 

345 

346 

347class Client(BaseClient): # We choose to implement all 4 grants in 1 class 

348 """This is the main API for oauth2 client. 

349 

350 Its methods define and document parameters mentioned in OAUTH2 RFC 6749. 

351 """ 

352 DEVICE_FLOW = { # consts for device flow, that can be customized by sub-class 

353 "GRANT_TYPE": "urn:ietf:params:oauth:grant-type:device_code", 

354 "DEVICE_CODE": "device_code", 

355 } 

356 DEVICE_FLOW_RETRIABLE_ERRORS = ("authorization_pending", "slow_down") 

357 GRANT_TYPE_SAML2 = "urn:ietf:params:oauth:grant-type:saml2-bearer" # RFC7522 

358 GRANT_TYPE_JWT = "urn:ietf:params:oauth:grant-type:jwt-bearer" # RFC7523 

359 grant_assertion_encoders = {GRANT_TYPE_SAML2: BaseClient.encode_saml_assertion} 

360 

361 

362 def initiate_device_flow(self, scope=None, *, data=None, **kwargs): 

363 # type: (list, **dict) -> dict 

364 # The naming of this method is following the wording of this specs 

365 # https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.1 

366 """Initiate a device flow. 

367 

368 Returns the data defined in Device Flow specs. 

369 https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.2 

370 

371 You should then orchestrate the User Interaction as defined in here 

372 https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.3 

373 

374 And possibly here 

375 https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.3.1 

376 """ 

377 DAE = "device_authorization_endpoint" 

378 if not self.configuration.get(DAE): 

379 raise ValueError("You need to provide device authorization endpoint") 

380 _data = {"client_id": self.client_id, "scope": self._stringify(scope or [])} 

381 if isinstance(data, dict): 

382 _data.update(data) 

383 resp = self._http_client.post(self.configuration[DAE], 

384 data=_data, 

385 headers=dict(self.default_headers, **kwargs.pop("headers", {})), 

386 **kwargs) 

387 flow = json.loads(resp.text) 

388 flow["interval"] = int(flow.get("interval", 5)) # Some IdP returns string 

389 flow["expires_in"] = int(flow.get("expires_in", 1800)) 

390 flow["expires_at"] = time.time() + flow["expires_in"] # We invent this 

391 return flow 

392 

393 def _obtain_token_by_device_flow(self, flow, **kwargs): 

394 # type: (dict, **dict) -> dict 

395 # This method updates flow during each run. And it is non-blocking. 

396 now = time.time() 

397 skew = 1 

398 if flow.get("latest_attempt_at", 0) + flow.get("interval", 5) - skew > now: 

399 warnings.warn('Attempted too soon. Please do time.sleep(flow["interval"])') 

400 data = kwargs.pop("data", {}) 

401 data.update({ 

402 "client_id": self.client_id, 

403 self.DEVICE_FLOW["DEVICE_CODE"]: flow["device_code"], 

404 }) 

405 result = self._obtain_token( 

406 self.DEVICE_FLOW["GRANT_TYPE"], data=data, **kwargs) 

407 if result.get("error") == "slow_down": 

408 # Respecting https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.5 

409 flow["interval"] = flow.get("interval", 5) + 5 

410 flow["latest_attempt_at"] = now 

411 return result 

412 

413 def obtain_token_by_device_flow(self, 

414 flow, 

415 exit_condition=lambda flow: flow.get("expires_at", 0) < time.time(), 

416 **kwargs): 

417 # type: (dict, Callable) -> dict 

418 """Obtain token by a device flow object, with customizable polling effect. 

419 

420 Args: 

421 flow (dict): 

422 An object previously generated by initiate_device_flow(...). 

423 Its content WILL BE CHANGED by this method during each run. 

424 We share this object with you, so that you could implement 

425 your own loop, should you choose to do so. 

426 

427 exit_condition (Callable): 

428 This method implements a loop to provide polling effect. 

429 The loop's exit condition is calculated by this callback. 

430 

431 The default callback makes the loop run until the flow expires. 

432 Therefore, one of the ways to exit the polling early, 

433 is to change the flow["expires_at"] to a small number such as 0. 

434 

435 In case you are doing async programming, you may want to 

436 completely turn off the loop. You can do so by using a callback as: 

437 

438 exit_condition = lambda flow: True 

439 

440 to make the loop run only once, i.e. no polling, hence non-block. 

441 """ 

442 while True: 

443 result = self._obtain_token_by_device_flow(flow, **kwargs) 

444 if result.get("error") not in self.DEVICE_FLOW_RETRIABLE_ERRORS: 

445 return result 

446 for i in range(flow.get("interval", 5)): # Wait interval seconds 

447 if exit_condition(flow): 

448 return result 

449 time.sleep(1) # Shorten each round, to make exit more responsive 

450 

451 def _build_auth_request_uri( 

452 self, 

453 response_type, 

454 *, 

455 redirect_uri=None, scope=None, state=None, response_mode=None, 

456 **kwargs): 

457 if "authorization_endpoint" not in self.configuration: 

458 raise ValueError("authorization_endpoint not found in configuration") 

459 authorization_endpoint = self.configuration["authorization_endpoint"] 

460 if response_mode != 'form_post': 

461 warnings.warn( 

462 "response_mode='form_post' is recommended for better security. " 

463 "See https://www.rfc-editor.org/rfc/rfc9700.html#section-4.3.1" 

464 ) 

465 params = self._build_auth_request_params( 

466 response_type, redirect_uri=redirect_uri, scope=scope, state=state, 

467 response_mode=response_mode, 

468 **kwargs) 

469 sep = '&' if '?' in authorization_endpoint else '?' 

470 return "%s%s%s" % (authorization_endpoint, sep, urlencode(params)) 

471 

472 def build_auth_request_uri( 

473 self, 

474 response_type, redirect_uri=None, scope=None, state=None, **kwargs): 

475 # This method could be named build_authorization_request_uri() instead, 

476 # but then there would be a build_authentication_request_uri() in the OIDC 

477 # subclass doing almost the same thing. So we use a loose term "auth" here. 

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

479 

480 Parameters are the same as another method :func:`initiate_auth_code_flow()`, 

481 whose functionality is a superset of this method. 

482 

483 :return: The auth uri as a string. 

484 """ 

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

486 return self._build_auth_request_uri( 

487 response_type, redirect_uri=redirect_uri, scope=scope, state=state, 

488 **kwargs) 

489 

490 def initiate_auth_code_flow( 

491 # The name is influenced by OIDC 

492 # https://openid.net/specs/openid-connect-core-1_0.html#CodeFlowAuth 

493 self, 

494 scope=None, redirect_uri=None, state=None, 

495 **kwargs): 

496 """Initiate an auth code flow. 

497 

498 Later when the response reaches your redirect_uri, 

499 you can use :func:`~obtain_token_by_auth_code_flow()` 

500 to complete the authentication/authorization. 

501 

502 This method also provides PKCE protection automatically. 

503 

504 :param list scope: 

505 It is a list of case-sensitive strings. 

506 Some ID provider can accept empty string to represent default scope. 

507 :param str redirect_uri: 

508 Optional. If not specified, server will use the pre-registered one. 

509 :param str state: 

510 An opaque value used by the client to 

511 maintain state between the request and callback. 

512 If absent, this library will automatically generate one internally. 

513 :param kwargs: Other parameters, typically defined in OpenID Connect. 

514 

515 :return: 

516 The auth code flow. It is a dict in this form:: 

517 

518 { 

519 "auth_uri": "https://...", // Guide user to visit this 

520 "state": "...", // You may choose to verify it by yourself, 

521 // or just let obtain_token_by_auth_code_flow() 

522 // do that for you. 

523 "...": "...", // Everything else are reserved and internal 

524 } 

525 

526 The caller is expected to:: 

527 

528 1. somehow store this content, typically inside the current session, 

529 2. guide the end user (i.e. resource owner) to visit that auth_uri, 

530 3. and then relay this dict and subsequent auth response to 

531 :func:`~obtain_token_by_auth_code_flow()`. 

532 """ 

533 response_type = kwargs.pop("response_type", "code") # Auth Code flow 

534 # Must be "code" when you are using Authorization Code Grant. 

535 # The "token" for Implicit Grant is not applicable thus not allowed. 

536 # It could theoretically be other 

537 # (possibly space-delimited) strings as registered extension value. 

538 # See https://tools.ietf.org/html/rfc6749#section-3.1.1 

539 if "token" in response_type: 

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

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

542 raise ValueError('response_type="token ..." is not allowed') 

543 pkce = _generate_pkce_code_verifier() 

544 flow = { # These data are required by obtain_token_by_auth_code_flow() 

545 "state": state or secrets.token_urlsafe(16), 

546 "redirect_uri": redirect_uri, 

547 "scope": scope, 

548 } 

549 auth_uri = self._build_auth_request_uri( 

550 response_type, 

551 code_challenge=pkce["code_challenge"], 

552 code_challenge_method=pkce["transformation"], 

553 **dict(flow, **kwargs)) 

554 flow["auth_uri"] = auth_uri 

555 flow["code_verifier"] = pkce["code_verifier"] 

556 return flow 

557 

558 def obtain_token_by_auth_code_flow( 

559 self, 

560 auth_code_flow, 

561 auth_response, 

562 scope=None, 

563 **kwargs): 

564 """With the auth_response being redirected back, 

565 validate it against auth_code_flow, and then obtain tokens. 

566 

567 Internally, it implements PKCE to mitigate the auth code interception attack. 

568 

569 :param dict auth_code_flow: 

570 The same dict returned by :func:`~initiate_auth_code_flow()`. 

571 :param dict auth_response: 

572 A dict based on query string received from auth server. 

573 

574 :param scope: 

575 You don't usually need to use scope parameter here. 

576 Some Identity Provider allows you to provide 

577 a subset of what you specified during :func:`~initiate_auth_code_flow`. 

578 :type scope: collections.Iterable[str] 

579 

580 :return: 

581 * A dict containing "access_token" and/or "id_token", among others, 

582 depends on what scope was used. 

583 (See https://tools.ietf.org/html/rfc6749#section-5.1) 

584 * A dict containing "error", optionally "error_description", "error_uri". 

585 (It is either `this <https://tools.ietf.org/html/rfc6749#section-4.1.2.1>`_ 

586 or `that <https://tools.ietf.org/html/rfc6749#section-5.2>`_ 

587 * Most client-side data error would result in ValueError exception. 

588 So the usage pattern could be without any protocol details:: 

589 

590 def authorize(): # A controller in a web app 

591 try: 

592 result = client.obtain_token_by_auth_code_flow( 

593 session.get("flow", {}), auth_resp) 

594 if "error" in result: 

595 return render_template("error.html", result) 

596 store_tokens() 

597 except ValueError: # Usually caused by CSRF 

598 pass # Simply ignore them 

599 return redirect(url_for("index")) 

600 """ 

601 assert isinstance(auth_code_flow, dict) and isinstance(auth_response, dict) 

602 # This is app developer's error which we do NOT want to map to ValueError 

603 if not auth_code_flow.get("state"): 

604 # initiate_auth_code_flow() already guarantees a state to be available. 

605 # This check will also allow a web app to blindly call this method with 

606 # obtain_token_by_auth_code_flow(session.get("flow", {}), auth_resp) 

607 # which further simplifies their usage. 

608 raise ValueError("state missing from auth_code_flow") 

609 if auth_code_flow.get("state") != auth_response.get("state"): 

610 raise ValueError("state mismatch: {} vs {}".format( 

611 auth_code_flow.get("state"), auth_response.get("state"))) 

612 if scope and set(scope) - set(auth_code_flow.get("scope", [])): 

613 raise ValueError( 

614 "scope must be None or a subset of %s" % auth_code_flow.get("scope")) 

615 if auth_response.get("code"): # i.e. the first leg was successful 

616 return self._obtain_token_by_authorization_code( 

617 auth_response["code"], 

618 redirect_uri=auth_code_flow.get("redirect_uri"), 

619 # Required, if "redirect_uri" parameter was included in the 

620 # authorization request, and their values MUST be identical. 

621 scope=scope or auth_code_flow.get("scope"), 

622 # It is both unnecessary and harmless, per RFC 6749. 

623 # We use the same scope already used in auth request uri, 

624 # thus token cache can know what scope the tokens are for. 

625 data=dict( # Extract and update the data 

626 kwargs.pop("data", {}), 

627 code_verifier=auth_code_flow["code_verifier"], 

628 ), 

629 **kwargs) 

630 if auth_response.get("error"): # It means the first leg encountered error 

631 # Here we do NOT return original auth_response as-is, to prevent a 

632 # potential {..., "access_token": "attacker's AT"} input being leaked 

633 error = {"error": auth_response["error"]} 

634 if auth_response.get("error_description"): 

635 error["error_description"] = auth_response["error_description"] 

636 if auth_response.get("error_uri"): 

637 error["error_uri"] = auth_response["error_uri"] 

638 return error 

639 raise ValueError('auth_response must contain either "code" or "error"') 

640 

641 def obtain_token_by_browser( 

642 # Name influenced by RFC 8252: "native apps should (use) ... user's browser" 

643 self, 

644 redirect_uri=None, 

645 auth_code_receiver=None, 

646 **kwargs): 

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

648 

649 Internally, it implements PKCE to mitigate the auth code interception attack. 

650 

651 :param scope: A list of scopes that you would like to obtain token for. 

652 :type scope: collections.Iterable[str] 

653 

654 :param extra_scope_to_consent: 

655 Some IdP allows you to include more scopes for end user to consent. 

656 The access token returned by this method will NOT include those scopes, 

657 but the refresh token would record those extra consent, 

658 so that your future :func:`~obtain_token_by_refresh_token()` call 

659 would be able to obtain token for those additional scopes, silently. 

660 :type scope: collections.Iterable[str] 

661 

662 :param string redirect_uri: 

663 The redirect_uri to be sent via auth request to Identity Provider (IdP), 

664 to indicate where an auth response would come back to. 

665 Such as ``http://127.0.0.1:0`` (default) or ``http://localhost:1234``. 

666 

667 If port 0 is specified, this method will choose a system-allocated port, 

668 then the actual redirect_uri will contain that port. 

669 To use this behavior, your IdP would need to accept such dynamic port. 

670 

671 Per HTTP convention, if port number is absent, it would mean port 80, 

672 although you probably want to specify port 0 in this context. 

673 

674 :param dict auth_params: 

675 These parameters will be sent to authorization_endpoint. 

676 

677 :param int timeout: In seconds. None means wait indefinitely. 

678 

679 :param str browser_name: 

680 If you did 

681 ``webbrowser.register("xyz", None, BackgroundBrowser("/path/to/browser"))`` 

682 beforehand, you can pass in the name "xyz" to use that browser. 

683 The default value ``None`` means using default browser, 

684 which is customizable by env var $BROWSER. 

685 

686 :return: Same as :func:`~obtain_token_by_auth_code_flow()` 

687 """ 

688 if auth_code_receiver: # Then caller already knows the listen port 

689 return self._obtain_token_by_browser( # Use all input param as-is 

690 auth_code_receiver, redirect_uri=redirect_uri, **kwargs) 

691 # Otherwise we will listen on _redirect_uri.port 

692 _redirect_uri = urlparse(redirect_uri or "http://127.0.0.1:0") 

693 if not _redirect_uri.hostname: 

694 raise ValueError("redirect_uri should contain hostname") 

695 listen_port = ( # Conventionally, port-less uri would mean port 80 

696 80 if _redirect_uri.port is None else _redirect_uri.port) 

697 try: 

698 with _AuthCodeReceiver(port=listen_port) as receiver: 

699 uri = redirect_uri if _redirect_uri.port != 0 else urlunparse(( 

700 _redirect_uri.scheme, 

701 "{}:{}".format(_redirect_uri.hostname, receiver.get_port()), 

702 _redirect_uri.path, 

703 _redirect_uri.params, 

704 _redirect_uri.query, 

705 _redirect_uri.fragment, 

706 )) # It could be slightly different than raw redirect_uri 

707 self.logger.debug("Using {} as redirect_uri".format(uri)) 

708 return self._obtain_token_by_browser( 

709 receiver, redirect_uri=uri, **kwargs) 

710 except PermissionError: 

711 raise ValueError( 

712 "Can't listen on port %s. You may try port 0." % listen_port) 

713 

714 def _obtain_token_by_browser( 

715 self, 

716 auth_code_receiver, 

717 scope=None, 

718 extra_scope_to_consent=None, 

719 redirect_uri=None, 

720 timeout=None, 

721 welcome_template=None, 

722 success_template=None, 

723 error_template=None, 

724 auth_params=None, 

725 auth_uri_callback=None, 

726 browser_name=None, 

727 **kwargs): 

728 # Internally, it calls self.initiate_auth_code_flow() and 

729 # self.obtain_token_by_auth_code_flow(). 

730 # 

731 # Parameters are documented in public method obtain_token_by_browser(). 

732 flow = self.initiate_auth_code_flow( 

733 redirect_uri=redirect_uri, 

734 scope=_scope_set(scope) | _scope_set(extra_scope_to_consent), 

735 response_mode='form_post', # The auth_code_receiver has been changed to require it 

736 **(auth_params or {})) 

737 auth_response = auth_code_receiver.get_auth_response( 

738 auth_uri=flow["auth_uri"], 

739 state=flow["state"], # So receiver can check it early 

740 timeout=timeout, 

741 welcome_template=welcome_template, 

742 success_template=success_template, 

743 error_template=error_template, 

744 auth_uri_callback=auth_uri_callback, 

745 browser_name=browser_name, 

746 ) 

747 if auth_response is None: 

748 raise BrowserInteractionTimeoutError("User did not complete the flow in time") 

749 return self.obtain_token_by_auth_code_flow( 

750 flow, auth_response, scope=scope, **kwargs) 

751 

752 @staticmethod 

753 def parse_auth_response(params, state=None): 

754 """Parse the authorization response being redirected back. 

755 

756 :param params: A string or dict of the query string 

757 :param state: REQUIRED if the state parameter was present in the client 

758 authorization request. This function will compare it with response. 

759 """ 

760 warnings.warn( 

761 "Use obtain_token_by_auth_code_flow() instead", DeprecationWarning) 

762 if not isinstance(params, dict): 

763 params = parse_qs(params) 

764 if params.get('state') != state: 

765 raise ValueError('state mismatch') 

766 return params 

767 

768 def obtain_token_by_authorization_code( 

769 self, code, redirect_uri=None, scope=None, **kwargs): 

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

771 

772 This is typically used by a server-side app (Confidential Client), 

773 but it can also be used by a device-side native app (Public Client). 

774 See more detail at https://tools.ietf.org/html/rfc6749#section-4.1.3 

775 

776 You are encouraged to use its higher level method 

777 :func:`~obtain_token_by_auth_code_flow` instead. 

778 

779 :param code: The authorization code received from authorization server. 

780 :param redirect_uri: 

781 Required, if the "redirect_uri" parameter was included in the 

782 authorization request, and their values MUST be identical. 

783 :param scope: 

784 It is both unnecessary and harmless to use scope here, per RFC 6749. 

785 We suggest to use the same scope already used in auth request uri, 

786 so that this library can link the obtained tokens with their scope. 

787 """ 

788 warnings.warn( 

789 "Use obtain_token_by_auth_code_flow() instead", DeprecationWarning) 

790 return self._obtain_token_by_authorization_code( 

791 code, redirect_uri=redirect_uri, scope=scope, **kwargs) 

792 

793 def _obtain_token_by_authorization_code( 

794 self, code, redirect_uri=None, scope=None, **kwargs): 

795 data = kwargs.pop("data", {}) 

796 data.update(code=code, redirect_uri=redirect_uri) 

797 if scope: 

798 data["scope"] = scope 

799 if not self.client_secret: 

800 # client_id is required, if the client is not authenticating itself. 

801 # See https://tools.ietf.org/html/rfc6749#section-4.1.3 

802 data["client_id"] = self.client_id 

803 return self._obtain_token("authorization_code", data=data, **kwargs) 

804 

805 def obtain_token_by_username_password( 

806 self, username, password, scope=None, **kwargs): 

807 """The Resource Owner Password Credentials Grant, used by legacy app.""" 

808 data = kwargs.pop("data", {}) 

809 data.update(username=username, password=password, scope=scope) 

810 return self._obtain_token("password", data=data, **kwargs) 

811 

812 def obtain_token_for_client(self, scope=None, **kwargs): 

813 """Obtain token for this client (rather than for an end user), 

814 a.k.a. the Client Credentials Grant, used by Backend Applications. 

815 

816 We don't name it obtain_token_by_client_credentials(...) because those 

817 credentials are typically already provided in class constructor, not here. 

818 You can still explicitly provide an optional client_secret parameter, 

819 or you can provide such extra parameters as `default_body` during the 

820 class initialization. 

821 """ 

822 data = kwargs.pop("data", {}) 

823 data.update(scope=scope) 

824 return self._obtain_token("client_credentials", data=data, **kwargs) 

825 

826 def obtain_token_by_user_fic( 

827 self, scope, assertion, username=None, user_object_id=None, 

828 **kwargs): 

829 """Obtain token using the ``user_fic`` grant type. 

830 

831 This exchanges a federated identity credential (e.g. an agent 

832 instance token) for a user-scoped access token. 

833 

834 :param scope: Scopes for the target resource (already decorated 

835 with OIDC scopes by the caller). 

836 :param str assertion: The federated identity credential token. 

837 :param str username: The target user's UPN (mutually exclusive 

838 with *user_object_id*). 

839 :param str user_object_id: The target user's Object ID (mutually 

840 exclusive with *username*). 

841 """ 

842 data = kwargs.pop("data", {}) 

843 data.update( 

844 scope=scope, 

845 user_federated_identity_credential=assertion, 

846 client_info="1", 

847 ) 

848 if user_object_id: 

849 data["user_id"] = str(user_object_id) 

850 elif username: 

851 data["username"] = username 

852 return self._obtain_token("user_fic", data=data, **kwargs) 

853 

854 def __init__(self, 

855 server_configuration, client_id, 

856 on_obtaining_tokens=lambda event: None, # event is defined in _obtain_token(...) 

857 on_removing_rt=lambda token_item: None, 

858 on_updating_rt=lambda token_item, new_rt: None, 

859 **kwargs): 

860 super(Client, self).__init__(server_configuration, client_id, **kwargs) 

861 self.on_obtaining_tokens = on_obtaining_tokens 

862 self.on_removing_rt = on_removing_rt 

863 self.on_updating_rt = on_updating_rt 

864 

865 def _obtain_token( 

866 self, grant_type, params=None, data=None, 

867 also_save_rt=False, 

868 on_obtaining_tokens=None, 

869 *args, **kwargs): 

870 _data = data.copy() # to prevent side effect 

871 resp = super(Client, self)._obtain_token( 

872 grant_type, params, _data, *args, **kwargs) 

873 if "error" not in resp: 

874 _resp = resp.copy() 

875 RT = "refresh_token" 

876 if grant_type == RT and RT in _resp and not also_save_rt: 

877 # Then we skip it from on_obtaining_tokens(); 

878 # Leave it to self.obtain_token_by_refresh_token() 

879 _resp.pop(RT, None) 

880 if "scope" in _resp: 

881 scope = _resp["scope"].split() # It is conceptually a set, 

882 # but we represent it as a list which can be persisted to JSON 

883 else: 

884 # Note: The scope will generally be absent in authorization grant, 

885 # but our obtain_token_by_authorization_code(...) encourages 

886 # app developer to still explicitly provide a scope here. 

887 scope = _data.get("scope") 

888 (on_obtaining_tokens or self.on_obtaining_tokens)({ 

889 "client_id": self.client_id, 

890 "scope": scope, 

891 "token_endpoint": self.configuration["token_endpoint"], 

892 "grant_type": grant_type, # can be used to know an IdToken-less 

893 # response is for an app or for a user 

894 "response": _resp, "params": params, "data": _data, 

895 }) 

896 return resp 

897 

898 def obtain_token_by_refresh_token(self, token_item, scope=None, 

899 rt_getter=lambda token_item: token_item["refresh_token"], 

900 on_removing_rt=None, 

901 on_updating_rt=None, 

902 **kwargs): 

903 # type: (Union[str, dict], Union[str, list, set, tuple], Callable) -> dict 

904 """This is an overload which will trigger token storage callbacks. 

905 

906 :param token_item: