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
1"""This OAuth2 client implementation aims to be spec-compliant, and generic."""
2# OAuth2 spec https://tools.ietf.org/html/rfc6749
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
21from .authcode import AuthCodeReceiver as _AuthCodeReceiver
23try:
24 PermissionError # Available in Python 3
25except:
26 from socket import error as PermissionError # Workaround for Python 2
29string_types = (str,) if sys.version_info[0] >= 3 else (basestring, )
32class BrowserInteractionTimeoutError(RuntimeError):
33 pass
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
40 @staticmethod
41 def encode_saml_assertion(assertion):
42 return base64.urlsafe_b64encode(assertion).rstrip(b'=') # Per RFC 7522
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}
48 @property
49 def session(self):
50 warnings.warn("Will be gone in next major release", DeprecationWarning)
51 return self._http_client
53 @session.setter
54 def session(self, value):
55 warnings.warn("Will be gone in next major release", DeprecationWarning)
56 self._http_client = value
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.
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
87 http_client (http.HttpClient):
88 Your implementation of abstract class :class:`http.HttpClient`.
89 Defaults to a requests session instance.
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:
96 s = requests.Session()
97 s.request = functools.partial(s.request, timeout=3)
99 and then feed that patched session instance to this class.
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.
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).
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.
136 This does not apply if you have chosen to pass your own Http client.
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>`_.
143 This does not apply if you have chosen to pass your own Http client.
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>`_.
150 This does not apply if you have chosen to pass your own Http client.
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
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)
181 @staticmethod
182 def _accepts_context(func):
183 """Check if a callable requires at least one positional argument.
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
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()
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)
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
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}
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)
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
256 if _data.get('scope'):
257 _data['scope'] = self._stringify(_data['scope'])
259 _headers = {'Accept': 'application/json'}
260 _headers.update(self.default_headers)
261 _headers.update(headers or {})
263 # Quoted from https://tools.ietf.org/html/rfc6749#section-2.3.1
264 # Clients in possession of a client password MAY use the HTTP Basic
265 # authentication.
266 # Alternatively, (but NOT RECOMMENDED,)
267 # the authorization server MAY support including the
268 # client credentials in the request-body using the following
269 # parameters: client_id, client_secret.
270 if self.client_secret and self.client_id:
271 _headers["Authorization"] = "Basic " + base64.b64encode("{}:{}".format(
272 # Per https://tools.ietf.org/html/rfc6749#section-2.3.1
273 # client_id and client_secret needs to be encoded by
274 # "application/x-www-form-urlencoded"
275 # https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4.1
276 # BEFORE they are fed into HTTP Basic Authentication
277 quote_plus(self.client_id), quote_plus(self.client_secret)
278 ).encode("ascii")).decode("ascii")
280 if "token_endpoint" not in self.configuration:
281 raise ValueError("token_endpoint not found in configuration")
282 resp = (post or self._http_client.post)(
283 self.configuration["token_endpoint"],
284 headers=_headers, params=params, data=_data,
285 **kwargs)
286 if resp.status_code >= 500:
287 resp.raise_for_status() # TODO: Will probably retry here
288 try:
289 # The spec (https://tools.ietf.org/html/rfc6749#section-5.2) says
290 # even an error response will be a valid json structure,
291 # so we simply return it here, without needing to invent an exception.
292 return json.loads(resp.text)
293 except ValueError:
294 self.logger.exception(
295 "Token response is not in json format: %s", resp.text)
296 raise
298 def obtain_token_by_refresh_token(self, refresh_token, scope=None, **kwargs):
299 # type: (str, Union[str, list, set, tuple]) -> dict
300 """Obtain an access token via a refresh token.
302 :param refresh_token: The refresh token issued to the client
303 :param scope: If omitted, is treated as equal to the scope originally
304 granted by the resource owner,
305 according to https://tools.ietf.org/html/rfc6749#section-6
306 """
307 assert isinstance(refresh_token, string_types)
308 data = kwargs.pop('data', {})
309 data.update(refresh_token=refresh_token, scope=scope)
310 return self._obtain_token("refresh_token", data=data, **kwargs)
312 def _stringify(self, sequence):
313 if isinstance(sequence, (list, set, tuple)):
314 return ' '.join(sorted(sequence)) # normalizing it, ascendingly
315 return sequence # as-is
318def _scope_set(scope):
319 assert scope is None or isinstance(scope, (list, set, tuple))
320 return set(scope) if scope else set([])
323def _generate_pkce_code_verifier(length=43):
324 assert 43 <= length <= 128
325 alphabet = string.ascii_letters + string.digits + "-._~"
326 verifier = "".join( # https://tools.ietf.org/html/rfc7636#section-4.1
327 secrets.choice(alphabet) for _ in range(length))
328 code_challenge = (
329 # https://tools.ietf.org/html/rfc7636#section-4.2
330 base64.urlsafe_b64encode(hashlib.sha256(verifier.encode("ascii")).digest())
331 .rstrip(b"=")) # Required by https://tools.ietf.org/html/rfc7636#section-3
332 return {
333 "code_verifier": verifier,
334 "transformation": "S256", # In Python, sha256 is always available
335 "code_challenge": code_challenge,
336 }
339class Client(BaseClient): # We choose to implement all 4 grants in 1 class
340 """This is the main API for oauth2 client.
342 Its methods define and document parameters mentioned in OAUTH2 RFC 6749.
343 """
344 DEVICE_FLOW = { # consts for device flow, that can be customized by sub-class
345 "GRANT_TYPE": "urn:ietf:params:oauth:grant-type:device_code",
346 "DEVICE_CODE": "device_code",
347 }
348 DEVICE_FLOW_RETRIABLE_ERRORS = ("authorization_pending", "slow_down")
349 GRANT_TYPE_SAML2 = "urn:ietf:params:oauth:grant-type:saml2-bearer" # RFC7522
350 GRANT_TYPE_JWT = "urn:ietf:params:oauth:grant-type:jwt-bearer" # RFC7523
351 grant_assertion_encoders = {GRANT_TYPE_SAML2: BaseClient.encode_saml_assertion}
354 def initiate_device_flow(self, scope=None, *, data=None, **kwargs):
355 # type: (list, **dict) -> dict
356 # The naming of this method is following the wording of this specs
357 # https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.1
358 """Initiate a device flow.
360 Returns the data defined in Device Flow specs.
361 https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.2
363 You should then orchestrate the User Interaction as defined in here
364 https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.3
366 And possibly here
367 https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.3.1
368 """
369 DAE = "device_authorization_endpoint"
370 if not self.configuration.get(DAE):
371 raise ValueError("You need to provide device authorization endpoint")
372 _data = {"client_id": self.client_id, "scope": self._stringify(scope or [])}
373 if isinstance(data, dict):
374 _data.update(data)
375 resp = self._http_client.post(self.configuration[DAE],
376 data=_data,
377 headers=dict(self.default_headers, **kwargs.pop("headers", {})),
378 **kwargs)
379 flow = json.loads(resp.text)
380 flow["interval"] = int(flow.get("interval", 5)) # Some IdP returns string
381 flow["expires_in"] = int(flow.get("expires_in", 1800))
382 flow["expires_at"] = time.time() + flow["expires_in"] # We invent this
383 return flow
385 def _obtain_token_by_device_flow(self, flow, **kwargs):
386 # type: (dict, **dict) -> dict
387 # This method updates flow during each run. And it is non-blocking.
388 now = time.time()
389 skew = 1
390 if flow.get("latest_attempt_at", 0) + flow.get("interval", 5) - skew > now:
391 warnings.warn('Attempted too soon. Please do time.sleep(flow["interval"])')
392 data = kwargs.pop("data", {})
393 data.update({
394 "client_id": self.client_id,
395 self.DEVICE_FLOW["DEVICE_CODE"]: flow["device_code"],
396 })
397 result = self._obtain_token(
398 self.DEVICE_FLOW["GRANT_TYPE"], data=data, **kwargs)
399 if result.get("error") == "slow_down":
400 # Respecting https://tools.ietf.org/html/draft-ietf-oauth-device-flow-12#section-3.5
401 flow["interval"] = flow.get("interval", 5) + 5
402 flow["latest_attempt_at"] = now
403 return result
405 def obtain_token_by_device_flow(self,
406 flow,
407 exit_condition=lambda flow: flow.get("expires_at", 0) < time.time(),
408 **kwargs):
409 # type: (dict, Callable) -> dict
410 """Obtain token by a device flow object, with customizable polling effect.
412 Args:
413 flow (dict):
414 An object previously generated by initiate_device_flow(...).
415 Its content WILL BE CHANGED by this method during each run.
416 We share this object with you, so that you could implement
417 your own loop, should you choose to do so.
419 exit_condition (Callable):
420 This method implements a loop to provide polling effect.
421 The loop's exit condition is calculated by this callback.
423 The default callback makes the loop run until the flow expires.
424 Therefore, one of the ways to exit the polling early,
425 is to change the flow["expires_at"] to a small number such as 0.
427 In case you are doing async programming, you may want to
428 completely turn off the loop. You can do so by using a callback as:
430 exit_condition = lambda flow: True
432 to make the loop run only once, i.e. no polling, hence non-block.
433 """
434 while True:
435 result = self._obtain_token_by_device_flow(flow, **kwargs)
436 if result.get("error") not in self.DEVICE_FLOW_RETRIABLE_ERRORS:
437 return result
438 for i in range(flow.get("interval", 5)): # Wait interval seconds
439 if exit_condition(flow):
440 return result
441 time.sleep(1) # Shorten each round, to make exit more responsive
443 def _build_auth_request_uri(
444 self,
445 response_type,
446 *,
447 redirect_uri=None, scope=None, state=None, response_mode=None,
448 **kwargs):
449 if "authorization_endpoint" not in self.configuration:
450 raise ValueError("authorization_endpoint not found in configuration")
451 authorization_endpoint = self.configuration["authorization_endpoint"]
452 if response_mode != 'form_post':
453 warnings.warn(
454 "response_mode='form_post' is recommended for better security. "
455 "See https://www.rfc-editor.org/rfc/rfc9700.html#section-4.3.1"
456 )
457 params = self._build_auth_request_params(
458 response_type, redirect_uri=redirect_uri, scope=scope, state=state,
459 response_mode=response_mode,
460 **kwargs)
461 sep = '&' if '?' in authorization_endpoint else '?'
462 return "%s%s%s" % (authorization_endpoint, sep, urlencode(params))
464 def build_auth_request_uri(
465 self,
466 response_type, redirect_uri=None, scope=None, state=None, **kwargs):
467 # This method could be named build_authorization_request_uri() instead,
468 # but then there would be a build_authentication_request_uri() in the OIDC
469 # subclass doing almost the same thing. So we use a loose term "auth" here.
470 """Generate an authorization uri to be visited by resource owner.
472 Parameters are the same as another method :func:`initiate_auth_code_flow()`,
473 whose functionality is a superset of this method.
475 :return: The auth uri as a string.
476 """
477 warnings.warn("Use initiate_auth_code_flow() instead. ", DeprecationWarning)
478 return self._build_auth_request_uri(
479 response_type, redirect_uri=redirect_uri, scope=scope, state=state,
480 **kwargs)
482 def initiate_auth_code_flow(
483 # The name is influenced by OIDC
484 # https://openid.net/specs/openid-connect-core-1_0.html#CodeFlowAuth
485 self,
486 scope=None, redirect_uri=None, state=None,
487 **kwargs):
488 """Initiate an auth code flow.
490 Later when the response reaches your redirect_uri,
491 you can use :func:`~obtain_token_by_auth_code_flow()`
492 to complete the authentication/authorization.
494 This method also provides PKCE protection automatically.
496 :param list scope:
497 It is a list of case-sensitive strings.
498 Some ID provider can accept empty string to represent default scope.
499 :param str redirect_uri:
500 Optional. If not specified, server will use the pre-registered one.
501 :param str state:
502 An opaque value used by the client to
503 maintain state between the request and callback.
504 If absent, this library will automatically generate one internally.
505 :param kwargs: Other parameters, typically defined in OpenID Connect.
507 :return:
508 The auth code flow. It is a dict in this form::
510 {
511 "auth_uri": "https://...", // Guide user to visit this
512 "state": "...", // You may choose to verify it by yourself,
513 // or just let obtain_token_by_auth_code_flow()
514 // do that for you.
515 "...": "...", // Everything else are reserved and internal
516 }
518 The caller is expected to::
520 1. somehow store this content, typically inside the current session,
521 2. guide the end user (i.e. resource owner) to visit that auth_uri,
522 3. and then relay this dict and subsequent auth response to
523 :func:`~obtain_token_by_auth_code_flow()`.
524 """
525 response_type = kwargs.pop("response_type", "code") # Auth Code flow
526 # Must be "code" when you are using Authorization Code Grant.
527 # The "token" for Implicit Grant is not applicable thus not allowed.
528 # It could theoretically be other
529 # (possibly space-delimited) strings as registered extension value.
530 # See https://tools.ietf.org/html/rfc6749#section-3.1.1
531 if "token" in response_type:
532 # Implicit grant would cause auth response coming back in #fragment,
533 # but fragment won't reach a web service.
534 raise ValueError('response_type="token ..." is not allowed')
535 pkce = _generate_pkce_code_verifier()
536 flow = { # These data are required by obtain_token_by_auth_code_flow()
537 "state": state or secrets.token_urlsafe(16),
538 "redirect_uri": redirect_uri,
539 "scope": scope,
540 }
541 auth_uri = self._build_auth_request_uri(
542 response_type,
543 code_challenge=pkce["code_challenge"],
544 code_challenge_method=pkce["transformation"],
545 **dict(flow, **kwargs))
546 flow["auth_uri"] = auth_uri
547 flow["code_verifier"] = pkce["code_verifier"]
548 return flow
550 def obtain_token_by_auth_code_flow(
551 self,
552 auth_code_flow,
553 auth_response,
554 scope=None,
555 **kwargs):
556 """With the auth_response being redirected back,
557 validate it against auth_code_flow, and then obtain tokens.
559 Internally, it implements PKCE to mitigate the auth code interception attack.
561 :param dict auth_code_flow:
562 The same dict returned by :func:`~initiate_auth_code_flow()`.
563 :param dict auth_response:
564 A dict based on query string received from auth server.
566 :param scope:
567 You don't usually need to use scope parameter here.
568 Some Identity Provider allows you to provide
569 a subset of what you specified during :func:`~initiate_auth_code_flow`.
570 :type scope: collections.Iterable[str]
572 :return:
573 * A dict containing "access_token" and/or "id_token", among others,
574 depends on what scope was used.
575 (See https://tools.ietf.org/html/rfc6749#section-5.1)
576 * A dict containing "error", optionally "error_description", "error_uri".
577 (It is either `this <https://tools.ietf.org/html/rfc6749#section-4.1.2.1>`_
578 or `that <https://tools.ietf.org/html/rfc6749#section-5.2>`_
579 * Most client-side data error would result in ValueError exception.
580 So the usage pattern could be without any protocol details::
582 def authorize(): # A controller in a web app
583 try:
584 result = client.obtain_token_by_auth_code_flow(
585 session.get("flow", {}), auth_resp)
586 if "error" in result:
587 return render_template("error.html", result)
588 store_tokens()
589 except ValueError: # Usually caused by CSRF
590 pass # Simply ignore them
591 return redirect(url_for("index"))
592 """
593 assert isinstance(auth_code_flow, dict) and isinstance(auth_response, dict)
594 # This is app developer's error which we do NOT want to map to ValueError
595 if not auth_code_flow.get("state"):
596 # initiate_auth_code_flow() already guarantees a state to be available.
597 # This check will also allow a web app to blindly call this method with
598 # obtain_token_by_auth_code_flow(session.get("flow", {}), auth_resp)
599 # which further simplifies their usage.
600 raise ValueError("state missing from auth_code_flow")
601 if auth_code_flow.get("state") != auth_response.get("state"):
602 raise ValueError("state mismatch: {} vs {}".format(
603 auth_code_flow.get("state"), auth_response.get("state")))
604 if scope and set(scope) - set(auth_code_flow.get("scope", [])):
605 raise ValueError(
606 "scope must be None or a subset of %s" % auth_code_flow.get("scope"))
607 if auth_response.get("code"): # i.e. the first leg was successful
608 return self._obtain_token_by_authorization_code(
609 auth_response["code"],
610 redirect_uri=auth_code_flow.get("redirect_uri"),
611 # Required, if "redirect_uri" parameter was included in the
612 # authorization request, and their values MUST be identical.
613 scope=scope or auth_code_flow.get("scope"),
614 # It is both unnecessary and harmless, per RFC 6749.
615 # We use the same scope already used in auth request uri,
616 # thus token cache can know what scope the tokens are for.
617 data=dict( # Extract and update the data
618 kwargs.pop("data", {}),
619 code_verifier=auth_code_flow["code_verifier"],
620 ),
621 **kwargs)
622 if auth_response.get("error"): # It means the first leg encountered error
623 # Here we do NOT return original auth_response as-is, to prevent a
624 # potential {..., "access_token": "attacker's AT"} input being leaked
625 error = {"error": auth_response["error"]}
626 if auth_response.get("error_description"):
627 error["error_description"] = auth_response["error_description"]
628 if auth_response.get("error_uri"):
629 error["error_uri"] = auth_response["error_uri"]
630 return error
631 raise ValueError('auth_response must contain either "code" or "error"')
633 def obtain_token_by_browser(
634 # Name influenced by RFC 8252: "native apps should (use) ... user's browser"
635 self,
636 redirect_uri=None,
637 auth_code_receiver=None,
638 **kwargs):
639 """A native app can use this method to obtain token via a local browser.
641 Internally, it implements PKCE to mitigate the auth code interception attack.
643 :param scope: A list of scopes that you would like to obtain token for.
644 :type scope: collections.Iterable[str]
646 :param extra_scope_to_consent:
647 Some IdP allows you to include more scopes for end user to consent.
648 The access token returned by this method will NOT include those scopes,
649 but the refresh token would record those extra consent,
650 so that your future :func:`~obtain_token_by_refresh_token()` call
651 would be able to obtain token for those additional scopes, silently.
652 :type scope: collections.Iterable[str]
654 :param string redirect_uri:
655 The redirect_uri to be sent via auth request to Identity Provider (IdP),
656 to indicate where an auth response would come back to.
657 Such as ``http://127.0.0.1:0`` (default) or ``http://localhost:1234``.
659 If port 0 is specified, this method will choose a system-allocated port,
660 then the actual redirect_uri will contain that port.
661 To use this behavior, your IdP would need to accept such dynamic port.
663 Per HTTP convention, if port number is absent, it would mean port 80,
664 although you probably want to specify port 0 in this context.
666 :param dict auth_params:
667 These parameters will be sent to authorization_endpoint.
669 :param int timeout: In seconds. None means wait indefinitely.
671 :param str browser_name:
672 If you did
673 ``webbrowser.register("xyz", None, BackgroundBrowser("/path/to/browser"))``
674 beforehand, you can pass in the name "xyz" to use that browser.
675 The default value ``None`` means using default browser,
676 which is customizable by env var $BROWSER.
678 :return: Same as :func:`~obtain_token_by_auth_code_flow()`
679 """
680 if auth_code_receiver: # Then caller already knows the listen port
681 return self._obtain_token_by_browser( # Use all input param as-is
682 auth_code_receiver, redirect_uri=redirect_uri, **kwargs)
683 # Otherwise we will listen on _redirect_uri.port
684 _redirect_uri = urlparse(redirect_uri or "http://127.0.0.1:0")
685 if not _redirect_uri.hostname:
686 raise ValueError("redirect_uri should contain hostname")
687 listen_port = ( # Conventionally, port-less uri would mean port 80
688 80 if _redirect_uri.port is None else _redirect_uri.port)
689 try:
690 with _AuthCodeReceiver(port=listen_port) as receiver:
691 uri = redirect_uri if _redirect_uri.port != 0 else urlunparse((
692 _redirect_uri.scheme,
693 "{}:{}".format(_redirect_uri.hostname, receiver.get_port()),
694 _redirect_uri.path,
695 _redirect_uri.params,
696 _redirect_uri.query,
697 _redirect_uri.fragment,
698 )) # It could be slightly different than raw redirect_uri
699 self.logger.debug("Using {} as redirect_uri".format(uri))
700 return self._obtain_token_by_browser(
701 receiver, redirect_uri=uri, **kwargs)
702 except PermissionError:
703 raise ValueError(
704 "Can't listen on port %s. You may try port 0." % listen_port)
706 def _obtain_token_by_browser(
707 self,
708 auth_code_receiver,
709 scope=None,
710 extra_scope_to_consent=None,
711 redirect_uri=None,
712 timeout=None,
713 welcome_template=None,
714 success_template=None,
715 error_template=None,
716 auth_params=None,
717 auth_uri_callback=None,
718 browser_name=None,
719 **kwargs):
720 # Internally, it calls self.initiate_auth_code_flow() and
721 # self.obtain_token_by_auth_code_flow().
722 #
723 # Parameters are documented in public method obtain_token_by_browser().
724 flow = self.initiate_auth_code_flow(
725 redirect_uri=redirect_uri,
726 scope=_scope_set(scope) | _scope_set(extra_scope_to_consent),
727 response_mode='form_post', # The auth_code_receiver has been changed to require it
728 **(auth_params or {}))
729 auth_response = auth_code_receiver.get_auth_response(
730 auth_uri=flow["auth_uri"],
731 state=flow["state"], # So receiver can check it early
732 timeout=timeout,
733 welcome_template=welcome_template,
734 success_template=success_template,
735 error_template=error_template,
736 auth_uri_callback=auth_uri_callback,
737 browser_name=browser_name,
738 )
739 if auth_response is None:
740 raise BrowserInteractionTimeoutError("User did not complete the flow in time")
741 return self.obtain_token_by_auth_code_flow(
742 flow, auth_response, scope=scope, **kwargs)
744 @staticmethod
745 def parse_auth_response(params, state=None):
746 """Parse the authorization response being redirected back.
748 :param params: A string or dict of the query string
749 :param state: REQUIRED if the state parameter was present in the client
750 authorization request. This function will compare it with response.
751 """
752 warnings.warn(
753 "Use obtain_token_by_auth_code_flow() instead", DeprecationWarning)
754 if not isinstance(params, dict):
755 params = parse_qs(params)
756 if params.get('state') != state:
757 raise ValueError('state mismatch')
758 return params
760 def obtain_token_by_authorization_code(
761 self, code, redirect_uri=None, scope=None, **kwargs):
762 """Get a token via authorization code. a.k.a. Authorization Code Grant.
764 This is typically used by a server-side app (Confidential Client),
765 but it can also be used by a device-side native app (Public Client).
766 See more detail at https://tools.ietf.org/html/rfc6749#section-4.1.3
768 You are encouraged to use its higher level method
769 :func:`~obtain_token_by_auth_code_flow` instead.
771 :param code: The authorization code received from authorization server.
772 :param redirect_uri:
773 Required, if the "redirect_uri" parameter was included in the
774 authorization request, and their values MUST be identical.
775 :param scope:
776 It is both unnecessary and harmless to use scope here, per RFC 6749.
777 We suggest to use the same scope already used in auth request uri,
778 so that this library can link the obtained tokens with their scope.
779 """
780 warnings.warn(
781 "Use obtain_token_by_auth_code_flow() instead", DeprecationWarning)
782 return self._obtain_token_by_authorization_code(
783 code, redirect_uri=redirect_uri, scope=scope, **kwargs)
785 def _obtain_token_by_authorization_code(
786 self, code, redirect_uri=None, scope=None, **kwargs):
787 data = kwargs.pop("data", {})
788 data.update(code=code, redirect_uri=redirect_uri)
789 if scope:
790 data["scope"] = scope
791 if not self.client_secret:
792 # client_id is required, if the client is not authenticating itself.
793 # See https://tools.ietf.org/html/rfc6749#section-4.1.3
794 data["client_id"] = self.client_id
795 return self._obtain_token("authorization_code", data=data, **kwargs)
797 def obtain_token_by_username_password(
798 self, username, password, scope=None, **kwargs):
799 """The Resource Owner Password Credentials Grant, used by legacy app."""
800 data = kwargs.pop("data", {})
801 data.update(username=username, password=password, scope=scope)
802 return self._obtain_token("password", data=data, **kwargs)
804 def obtain_token_for_client(self, scope=None, **kwargs):
805 """Obtain token for this client (rather than for an end user),
806 a.k.a. the Client Credentials Grant, used by Backend Applications.
808 We don't name it obtain_token_by_client_credentials(...) because those
809 credentials are typically already provided in class constructor, not here.
810 You can still explicitly provide an optional client_secret parameter,
811 or you can provide such extra parameters as `default_body` during the
812 class initialization.
813 """
814 data = kwargs.pop("data", {})
815 data.update(scope=scope)
816 return self._obtain_token("client_credentials", data=data, **kwargs)
818 def obtain_token_by_user_fic(
819 self, scope, assertion, username=None, user_object_id=None,
820 **kwargs):
821 """Obtain token using the ``user_fic`` grant type.
823 This exchanges a federated identity credential (e.g. an agent
824 instance token) for a user-scoped access token.
826 :param scope: Scopes for the target resource (already decorated
827 with OIDC scopes by the caller).
828 :param str assertion: The federated identity credential token.
829 :param str username: The target user's UPN (mutually exclusive
830 with *user_object_id*).
831 :param str user_object_id: The target user's Object ID (mutually
832 exclusive with *username*).
833 """
834 data = kwargs.pop("data", {})
835 data.update(
836 scope=scope,
837 user_federated_identity_credential=assertion,
838 client_info="1",
839 )
840 if user_object_id:
841 data["user_id"] = str(user_object_id)
842 elif username:
843 data["username"] = username
844 return self._obtain_token("user_fic", data=data, **kwargs)
846 def __init__(self,
847 server_configuration, client_id,
848 on_obtaining_tokens=lambda event: None, # event is defined in _obtain_token(...)
849 on_removing_rt=lambda token_item: None,
850 on_updating_rt=lambda token_item, new_rt: None,
851 **kwargs):
852 super(Client, self).__init__(server_configuration, client_id, **kwargs)
853 self.on_obtaining_tokens = on_obtaining_tokens
854 self.on_removing_rt = on_removing_rt
855 self.on_updating_rt = on_updating_rt
857 def _obtain_token(
858 self, grant_type, params=None, data=None,
859 also_save_rt=False,
860 on_obtaining_tokens=None,
861 *args, **kwargs):
862 _data = data.copy() # to prevent side effect
863 resp = super(Client, self)._obtain_token(
864 grant_type, params, _data, *args, **kwargs)
865 if "error" not in resp:
866 _resp = resp.copy()
867 RT = "refresh_token"
868 if grant_type == RT and RT in _resp and not also_save_rt:
869 # Then we skip it from on_obtaining_tokens();
870 # Leave it to self.obtain_token_by_refresh_token()
871 _resp.pop(RT, None)
872 if "scope" in _resp:
873 scope = _resp["scope"].split() # It is conceptually a set,
874 # but we represent it as a list which can be persisted to JSON
875 else:
876 # Note: The scope will generally be absent in authorization grant,
877 # but our obtain_token_by_authorization_code(...) encourages
878 # app developer to still explicitly provide a scope here.
879 scope = _data.get("scope")
880 (on_obtaining_tokens or self.on_obtaining_tokens)({
881 "client_id": self.client_id,
882 "scope": scope,
883 "token_endpoint": self.configuration["token_endpoint"],
884 "grant_type": grant_type, # can be used to know an IdToken-less
885 # response is for an app or for a user
886 "response": _resp, "params": params, "data": _data,
887 })
888 return resp
890 def obtain_token_by_refresh_token(self, token_item, scope=None,
891 rt_getter=lambda token_item: token_item["refresh_token"],
892 on_removing_rt=None,
893 on_updating_rt=None,
894 **kwargs):
895 # type: (Union[str, dict], Union[str, list, set, tuple], Callable) -> dict
896 """This is an overload which will trigger token storage callbacks.
898 :param token_item:
899 A refresh token (RT) item, in flexible format. It can be a string,
900 or a whatever data structure containing RT string and its metadata,
901 in such case the `rt_getter` callable must be able to
902 extract the RT string out from the token item data structure.
904 Either way, this token_item will be passed into other callbacks as-is.
906 :param scope: If omitted, is treated as equal to the scope originally
907 granted by the resource owner,
908 according to https://tools.ietf.org/html/rfc6749#section-6
909 :param rt_getter: A callable to translate the token_item to a raw RT string
910 :param on_removing_rt: If absent, fall back to the one defined in initialization
912 :param on_updating_rt:
913 Default to None, it will fall back to the one defined in initialization.
914 This is the most common case.
916 As a special case, you can pass in a False,
917 then this function will NOT trigger on_updating_rt() for RT UPDATE,
918 instead it will allow the RT to be added by on_obtaining_tokens().
919 This behavior is useful when you are migrating RTs from elsewhere
920 into a token storage managed by this library.
921 """
922 resp = super(Client, self).obtain_token_by_refresh_token(
923 rt_getter(token_item)
924 if not isinstance(token_item, string_types) else token_item,
925 scope=scope,
926 also_save_rt=on_updating_rt is False,
927 **kwargs)
928 if resp.get('error') == 'invalid_grant':
929 (on_removing_rt or self.on_removing_rt)(token_item) # Discard old RT
930 RT = "refresh_token"
931 if on_updating_rt is not False and RT in resp:
932 (on_updating_rt or self.on_updating_rt)(token_item, resp[RT])
933 return resp
935 def obtain_token_by_assertion(
936 self, assertion, grant_type, scope=None, **kwargs):
937 # type: (bytes, Union[str, None], Union[str, list, set, tuple]) -> dict
938 """This method implements Assertion Framework for OAuth2 (RFC 7521).
939 See details at https://tools.ietf.org/html/rfc7521#section-4.1
941 :param assertion:
942 The assertion bytes can be a raw SAML2 assertion, or a JWT assertion.
943 :param grant_type:
944 It is typically either the value of :attr:`GRANT_TYPE_SAML2`,
945 or :attr:`GRANT_TYPE_JWT`, the only two profiles defined in RFC 7521.
946 :param scope: Optional. It must be a subset of previously granted scopes.
947 """
948 encoder = self.grant_assertion_encoders.get(grant_type, lambda a: a)
949 data = kwargs.pop("data", {})
950 data.update(scope=scope, assertion=encoder(assertion))
951 return self._obtain_token(grant_type, data=data, **kwargs)