o iT@sdZddlZddlZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z mZmZmZmZmZmZmZddlZddlZddlmZddlmZddlmZddlmZdd lm Z dd l!m"Z"ej#d krxdd l m$Z$dd l%m&Z&n dd l'm$Z$dd l m&Z&e(e)Z*edZ+e&dZ,GdddeZ-Gddde.Z/Gddde.Z0GdddeZ1Gdddej2Z3Gddde.Z4de$e,e+fde$e,e+ffddZ5Gd d!d!Z6dS)"zSSO login base dependency.N) TracebackType) AnyClassVarDictListLiteralOptionalType TypedDictTypeVarUnionoverload)WebApplicationClient) HTTPException)Request)RedirectResponse)get_pkce_challenge_pair)generate_random_state) )Callable) ParamSpecTPc@s*eZdZUdZeed<eed<eed<dS)DiscoveryDocumentzDiscovery document.authorization_endpointtoken_endpointuserinfo_endpointN)__name__ __module__ __qualname____doc__str__annotations__r$r$K/home/app/Keep/.python/lib/python3.10/site-packages/fastapi_sso/sso/base.pyr$s  rc@eZdZdZdS)UnsetStateWarningz$Warning about unset state parameter.Nrrr r!r$r$r$r%r',r'c@r&)ReusedOauthClientWarningz+Warning about reused oauth client instance.Nr(r$r$r$r%r*0r)r*c@r&) SSOLoginErrorzRaised when any login-related error ocurrs. Such as when user is not verified or if there was an attempt for fake login. Nr(r$r$r$r%r+4r)r+c@seZdZUdZdZeeed<dZee j ed<dZ eeed<dZ eeed<dZ eeed<dZeeed<dZeeed <dS) OpenIDzOClass (schema) to represent information got from sso provider in a common form.Nidemail first_name last_name display_namepictureprovider)rrr r!r-rr"r#r.pydanticZEmailStrr/r0r1r2r3r$r$r$r%r,;s r,c@r&)SecurityWarningz&Raised when insecure usage is detectedNr(r$r$r$r%r5Gr)r5funcreturncs"dtjdtjdtffdd }|S)Nargskwargsr7csH|r t|dts|i|S|djstjdtdd|i|S)NrzPlease make sure you are using SSO provider in an async context (using 'async with provider:'). See https://github.com/tomasvotava/fastapi-sso/issues/186 for more information.)category stacklevel) isinstanceSSOBase _in_stackwarningswarnr5)r8r9r6r$r%wrapperLs z'requires_async_context..wrapper)rr8r9r)r6rCr$rBr%requires_async_contextKs rDc@s<eZdZUdZeZeed<eZeed<eZ eed<eZ e e e jefed<gZeeeed<dZee eeefed<d Zeed <d Zeed <d Zeed <  dOdedede e e jefdedede eef ddZede efddZeedefddZeede efddZ eede efddZ!eede efddZ"dPde#de e$j%de&fdd Z'de(fd!d"Z)ede efd#d$Z*ede efd%d&Z+ede efd'd(Z,dddd)de e e jefd*e eeefd+e edefd,d-Z-dddd)de ed*e eeefd+e ede.fd.d/Z/e0dddd0d1d2e1d*e eeefd3e eeefde ed4e2d0de e&f d5d6Z3e0dddd7d2e1d*e eeefd3e eeefde ed4e2d de eeeff d8d6Z3edddd0d1d2e1d*e eeefd3e eeefde ed4e e2d0e2d fde e e&e eeefff d9d6Z3dQd:d;Z4dQde e6e7d?e e7d@e e8ddfdAdBZ9d>e e6e7d?e e7d@e e8ddfdCdDZ:edefdEdFZ;e0ddddd0dGdHed2e1d*e eeefde eeefde edIe ed4e2d0de e&fdJdKZz!Base class for all SSO providers.r3 client_id client_secret redirect_uriscopeNadditional_headersF uses_pkcerequires_state`_pkce_challenge_lengthallow_insecure_http use_statecCs||_||_||_||_t|_d|_d|_d|_ |jr*t d|j j dtjd<|r2tdt|p6|j|_d|_d|_d|_d|_d|_d|_dS)z)Base class (mixin) for all SSO providers.FNz-Initializing %s with allow_insecure_http=True1ZOAUTHLIB_INSECURE_TRANSPORTzArgument 'use_state' of SSOBase's constructor is deprecated and will be removed in future releases. Use 'state' argument of individual methods instead.ZS256)rErFrGrNasyncioLock _login_lockr? _oauth_client_generated_stateloggerdebug __class__rosenvironr@rADeprecationWarningrH_scope_refresh_token _id_token_state_pkce_code_challenge_pkce_code_verifier_pkce_challenge_method)selfrErFrGrNrOrHr$r$r%__init__is.     zSSOBase.__init__r7cCs|jdur tdt|jS)agRetrieves the state as it was returned from the server. Warning: This will emit a warning if the state is unset, implying either that the server didn't return a state or `verify_and_process` hasn't been called yet. Returns: Optional[str]: The state parameter returned from the server. Nz'state' parameter is unset. This means the server either didn't return state (was this expected?) or 'verify_and_process' hasn't been called yet.)r_r@rAr'rcr$r$r%states z SSOBase.statecCs8|jtkrtd|jd|jdurt|j|_|jS)aRetrieves the OAuth Client to aid in generating requests and parsing responses. Raises: NotImplementedError: If the provider is not supported or `client_id` is not set. Returns: WebApplicationClient: OAuth client instance. Provider  not supportedN)rENotImplementedNotImplementedErrorr3rTrrer$r$r% oauth_clients   zSSOBase.oauth_clientcCs|jjS)zRetrieves the access token from token endpoint. Returns: Optional[str]: The access token if available. )rk access_tokenrer$r$r%rlszSSOBase.access_tokencCs|jp|jjS)zRetrieves the refresh token if returned from provider. Returns: Optional[str]: The refresh token if available. )r]rk refresh_tokenrer$r$r%rmszSSOBase.refresh_tokencCs|jS)zRetrieves the id token if returned from provider. Returns: Optional[str]: The id token if available. )r^rer$r$r%id_tokenszSSOBase.id_tokenresponsesessionctd|jd)aConverts a response from the provider's user info endpoint to an OpenID object. Args: response (dict): The response from the user info endpoint. session (Optional[httpx.AsyncClient]): The HTTPX AsyncClient session. Raises: NotImplementedError: If the provider is not supported. Returns: OpenID: The user information in a standardized format. rgrhrjr3)rcrorpr$r$r%openid_from_responses zSSOBase.openid_from_responsecrq)aRetrieves the discovery document containing useful URLs. Raises: NotImplementedError: If the provider is not supported. Returns: DiscoveryDocument: A dictionary containing important endpoints like authorization, token and userinfo. rgrhrrrer$r$r%get_discovery_documents zSSOBase.get_discovery_documentc|IdH}|dS)z8Return `authorization_endpoint` from discovery document.NrrtgetrcZ discoveryr$r$r%r zSSOBase.authorization_endpointcru)z0Return `token_endpoint` from discovery document.Nrrvrxr$r$r%rryzSSOBase.token_endpointcru)z3Return `userinfo_endpoint` from discovery document.Nrrvrxr$r$r%rryzSSOBase.userinfo_endpointrGparamsrfr{rfcs|pi}|p |j}|durtd|jr't|j|jfs't|jj d|j r>|s>|j dur;t|jj d|j }|j j |jIdHf|||j|j|jd|}|S)aGenerates and returns the prepared login URL. Args: redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance. params (Optional[Dict[str, Any]]): Additional query parameters to add to the login request. state (Optional[str]): The state parameter for the OAuth 2.0 authorization request. Raises: ValueError: If `redirect_uri` is not provided either at construction or request time. Returns: str: The prepared login URL. NzEredirect_uri must be provided, either at construction or request timez| uses PKCE and no code was generated yet. Use SSO class as a context manager to get rid of this warning and possible errors.z requires state in the request but none was provided nor generated automatically. Use SSO as a context manager. The login process will most probably fail.)rGrfrHZcode_challengeZcode_challenge_method)rG ValueErrorrJallrar`r@rArXrrKrUrkZprepare_request_urirr\rb)rcrGr{rf request_urir$r$r% get_login_urls6       zSSOBase.get_login_urlcsN|jr |s |j}|j|||dIdH}t|d}|jr%|dt|j|S)aConstructs and returns a redirect response to the login page of OAuth SSO provider. Args: redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance. params (Optional[Dict[str, Any]]): Additional query parameters to add to the login request. state (Optional[str]): The state parameter for the OAuth 2.0 authorization request. Returns: RedirectResponse: A Starlette response directing to the login page of the OAuth SSO provider. rzNi/pkce_code_verifier)rKrUrrrJ set_cookier"ra)rcrGr{rfZ login_uriror$r$r%get_login_redirect/s  zSSOBase.get_login_redirectT)r{headersrGconvert_responserequestrrcdSNr$rcrr{rrGrr$r$r%verify_and_processH zSSOBase.verify_and_process)r{rrGcrrr$rr$r$r%rSrc s|pi}|jd}|durtd|j|j|jtdd|jd|_d}|jr:|j d}|dur:t d|j |||||||d IdHS) atProcesses the login given a FastAPI (Starlette) Request object. This should be used for the /callback path. Args: request (Request): FastAPI or Starlette request object. params (Optional[Dict[str, Any]]): Additional query parameters to pass to the provider. headers (Optional[Dict[str, Any]]): Additional headers to pass to the provider. redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance. convert_response (bool): If True, userinfo response is converted to OpenID object. Raises: SSOLoginError: If the 'code' parameter is not found in the callback request. Returns: Optional[OpenID]: User information as OpenID instance (if convert_response == True) Optional[Dict[str, Any]]: The original JSON response from the API. codeNz9Callback request: URI: %s Headers: %s Query params: %siz2'code' parameter was not found in callback requestrfrzaPKCE code verifier was not found in the request Cookie. This will probably lead to a login error.r{rIrGrr) Z query_paramsrwrVrWurlrr+r_rJcookiesr@rA process_login)rcrr{rrGrrrr$r$r%r^s8    cCsRtjdtddd|_d|_d|_d|_|jrt|_ |j r't |j \|_ |_|S)NzSSO Providers are supposed to be used in async context, please change 'with provider' to 'async with provider'. See https://github.com/tomasvotava/fastapi-sso/issues/186 for more information.r:)r<)r@rAr[rTr]r^r_rKrrUrJrrMrar`rer$r$r% __enter__szSSOBase.__enter__csZ|jIdHd|_d|_d|_d|_d|_|jrt|_ |j r+t |j \|_ |_|S)NT)rSacquirer?rTr]r^r_rKrrUrJrrMrar`rer$r$r% __aenter__szSSOBase.__aenter__ _exc_type_exc_val_exc_tbcsd|_|jdS)NF)r?rSreleasercrrrr$r$r% __aexit__szSSOBase.__aexit__cCsdSrr$rr$r$r%__exit__szSSOBase.__exit__cCsiSrr$rer$r$r%_extra_query_paramsszSSOBase._extra_query_paramsrrrcrrr$rcrrr{rIrGrrr$r$r%r zSSOBase.process_login)r{rIrGrcrrr$rr$r$r%rrc s |jdurd|_d|_d|_tdt|pi}||j|p"i}||jp)i|j }|j s?|j dkr?t | dd} nt |} |j d|j|j} |rX|d|i|jj|jIdHf| |ph|jph| |d|\} } } | durxdS| |t|j|j}t4IdHk}|j| | | |d IdH}|}|d |_|d |_|jt||j|jIdH\}} }| ||j | ||IdH}|}|r|!||IdHWdIdHS|WdIdHS1IdHswYdS) aProcesses login from the callback endpoint to verify the user and request user info endpoint. It's a lower-level method, typically, you should use `verify_and_process` instead. Args: code (str): The authorization code. request (Request): FastAPI or Starlette request object. params (Optional[Dict[str, Any]]): Additional query parameters to pass to the provider. additional_headers (Optional[Dict[str, Any]]): Additional headers to be added to all requests. redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance. pkce_code_verifier (Optional[str]): A PKCE code verifier sent to the server to verify the login request. convert_response (bool): If True, userinfo response is converted to OpenID object. Raises: ReusedOauthClientWarning: If the SSO object is reused, which is not safe and caused security issues. Returns: Optional[OpenID]: User information in OpenID format if the login was successful (convert_response == True). Optional[Dict[str, Any]]: Original userinfo API endpoint response. NzReusing the SSO object is not safe and caused a security issue in previous versions.To make sure you don't see this warning, please use the SSO object as a context manager.httpszhttp://zhttps://z://Z code_verifier)Zauthorization_responseZ redirect_urlr)rcontentauthrmrn)"rTr]r^r@rAr*updaterrIrrNschemer"replacenetlocpathrkZprepare_token_requestrrGhttpxZ BasicAuthrErF AsyncClientpostjsonrwZparse_request_body_responsedumpsZ add_tokenrrrs)rcrrr{rIrGrrrZ current_url current_pathZ token_urlrbodyrrproruri_r$r$r%rsb          0)NFFNr)r7r>)=rrr r!rir3r"r#rErFrGrr r4Z AnyHttpUrlrHrrrIrrrJboolrKrMintrdpropertyrfrDrrkrlrmrndictrrr,rsrrtrrrrrrr rrrrrr BaseExceptionrrrrrr$r$r$r%r>[s         (  2      5             r>)7r!rQrloggingrYsysr@typesrtypingrrrrrrr r r r r rr4Zoauthlib.oauth2rZstarlette.exceptionsrZstarlette.requestsrZstarlette.responsesrZfastapi_sso.pkcerZfastapi_sso.stater version_inforZtyping_extensionsrcollections.abc getLoggerrrVrrr UserWarningr'r*r+Z BaseModelr,r5rDr>r$r$r$r%sB 4            "