(app: RequestCaptureAndRedirectApp)
| 461 | success_uri = self._infer_leap_success_uri(self.leap_api_endpoint) |
| 462 | |
| 463 | def exchange_code(app: RequestCaptureAndRedirectApp) -> str: |
| 464 | # error responses redirect immediately |
| 465 | if 'error' in app.query: |
| 466 | app.set_exception( |
| 467 | OAuthError(error=app.query['error'], |
| 468 | description=app.query.get('error_description'))) |
| 469 | return add_params_to_uri(error_uri, app.query) |
| 470 | |
| 471 | # when code received, exchange it for token |
| 472 | try: |
| 473 | # verify state (Cross-Site Request Forgery protection) |
| 474 | state = app.query.get('state') |
| 475 | if state != self.state: |
| 476 | raise ValueError('State mismatch') |
| 477 | |
| 478 | self.fetch_token(code=app.query.get('code'), state=state) |
| 479 | |
| 480 | except OAuthError as exc: |
| 481 | # store for main thread |
| 482 | app.set_exception(exc) |
| 483 | |
| 484 | # redirect to leap error page |
| 485 | query = dict(error=exc.error, error_description=exc.description, |
| 486 | state=app.query.get('state')) |
| 487 | return add_params_to_uri(error_uri, query) |
| 488 | |
| 489 | except Exception as exc: |
| 490 | # store for main thread |
| 491 | app.set_exception(exc) |
| 492 | |
| 493 | # redirect to leap error page |
| 494 | query = dict(error=type(exc).__name__, error_description=str(exc), |
| 495 | state=app.query.get('state')) |
| 496 | return add_params_to_uri(error_uri, query) |
| 497 | |
| 498 | # redirect to leap success page |
| 499 | return success_uri |
| 500 | |
| 501 | app = RequestCaptureAndRedirectApp( |
| 502 | message=self._REDIRECT_DONE_MSG, |
nothing calls this directly
no test coverage detected