Run the OAuth2 Authorization Code flow with PKCE.
()
| 144 | |
| 145 | |
| 146 | def authorize(): |
| 147 | """Run the OAuth2 Authorization Code flow with PKCE.""" |
| 148 | discovery = _discover_endpoints() |
| 149 | |
| 150 | session = OAuth2Session( |
| 151 | client_id=AUTH_CLIENT_ID, |
| 152 | redirect_uri=_REDIRECT_URI, |
| 153 | scope="openid offline_access", |
| 154 | token_endpoint_auth_method="none", |
| 155 | ) |
| 156 | |
| 157 | code_verifier = generate_token(48) |
| 158 | code_challenge = create_s256_code_challenge(code_verifier) |
| 159 | |
| 160 | uri, state = session.create_authorization_url( |
| 161 | discovery["authorization_endpoint"], |
| 162 | code_challenge=code_challenge, |
| 163 | code_challenge_method="S256", |
| 164 | ) |
| 165 | |
| 166 | _CallbackHandler.callback_url = None |
| 167 | _CallbackHandler.error = None |
| 168 | |
| 169 | server = HTTPServer(("localhost", _REDIRECT_PORT), _CallbackHandler) |
| 170 | |
| 171 | print("Opening browser for authentication...") |
| 172 | webbrowser.open(uri) |
| 173 | |
| 174 | server.handle_request() |
| 175 | server.server_close() |
| 176 | |
| 177 | if _CallbackHandler.error: |
| 178 | raise ValueError(f"Authorization failed: {_CallbackHandler.error}") |
| 179 | |
| 180 | if not _CallbackHandler.callback_url: |
| 181 | raise ValueError("Authorization failed: no callback received") |
| 182 | |
| 183 | token = session.fetch_token( |
| 184 | discovery["token_endpoint"], |
| 185 | authorization_response=_CallbackHandler.callback_url, |
| 186 | code_verifier=code_verifier, |
| 187 | ) |
| 188 | |
| 189 | _save_credentials(token) |
| 190 | return token |
| 191 | |
| 192 | |
| 193 | def get_access_token() -> str: |
no test coverage detected
searching dependent graphs…