o ^[2h¬ã@sDddlmZddlmZddlmZddlmZGdd„deƒZdS) é)Údefault_json_headersé)ÚInvalidRequestError)Ú TokenEndpoint)ÚUnsupportedTokenTypeErrorc@sLeZdZdZdZdd„Zdd„Zdd„Zd d „Zd d „Z d d„Z dd„Z dS)ÚIntrospectionEndpointz‰Implementation of introspection endpoint which is described in `RFC7662`_. .. _RFC7662: https://tools.ietf.org/html/rfc7662 Ú introspectioncCsD| ||¡| |jd|j d¡¡}|r| |||¡r |SdSdS)awThe protected resource calls the introspection endpoint using an HTTP ``POST`` request with parameters sent as "application/x-www-form-urlencoded" data. The protected resource sends a parameter representing the token along with optional parameters representing additional context that is known by the protected resource to aid the authorization server in its response. token **REQUIRED** The string value of the token. For access tokens, this is the ``access_token`` value returned from the token endpoint defined in OAuth 2.0. For refresh tokens, this is the ``refresh_token`` value returned from the token endpoint as defined in OAuth 2.0. token_type_hint **OPTIONAL** A hint about the type of the token submitted for introspection. ÚtokenÚtoken_type_hintN)Ú check_paramsÚ query_tokenÚformÚgetÚcheck_permission)ÚselfÚrequestÚclientr ©rúu/home/skpark/git/infrasmart_work/infrasmart/venv/lib/python3.10/site-packages/authlib/oauth2/rfc7662/introspection.pyÚauthenticate_tokens ÿÿz(IntrospectionEndpoint.authenticate_tokencCs:|j}d|vr tƒ‚| d¡}|r||jvrtƒ‚dSdS)Nr r )r rrÚSUPPORTED_TOKEN_TYPESr)rrrÚparamsÚhintrrrr ,s ÿz"IntrospectionEndpoint.check_paramscCs*| |¡}| ||¡}| |¡}d|tfS)zpValidate introspection request and create the response. :returns: (status_code, body, headers) éÈ)Úauthenticate_endpoint_clientrÚcreate_introspection_payloadr)rrrr ÚbodyrrrÚcreate_endpoint_response5s    z.IntrospectionEndpoint.create_endpoint_responsecCsB|sddiS| ¡s| ¡rddiS| |¡}d|vrd|d<|S)NÚactiveFT)Ú is_expiredÚ is_revokedÚintrospect_token)rr ÚpayloadrrrrEs z2IntrospectionEndpoint.create_introspection_payloadcCótƒ‚)aUCheck if the request has permission to introspect the token. Developers MUST implement this method:: def check_permission(self, token, client, request): # only allow a special client to introspect the token return client.client_id == "introspection_client" :return: bool ©ÚNotImplementedError)rr rrrrrrSs z&IntrospectionEndpoint.check_permissioncCr#)a’Get the token from database/storage by the given token string. Developers should implement this method:: def query_token(self, token_string, token_type_hint): if token_type_hint == "access_token": tok = Token.query_by_access_token(token_string) elif token_type_hint == "refresh_token": tok = Token.query_by_refresh_token(token_string) else: tok = Token.query_by_access_token(token_string) if not tok: tok = Token.query_by_refresh_token(token_string) return tok r$)rÚ token_stringr rrrr _sz!IntrospectionEndpoint.query_tokencCr#)aRead given token and return its introspection metadata as a dictionary following `Section 2.2`_:: def introspect_token(self, token): return { "active": True, "client_id": token.client_id, "token_type": token.token_type, "username": get_token_username(token), "scope": token.get_scope(), "sub": get_token_user_sub(token), "aud": token.client_id, "iss": "https://server.example.com/", "exp": token.expires_at, "iat": token.issued_at, } .. _`Section 2.2`: https://tools.ietf.org/html/rfc7662#section-2.2 r$)rr rrrr!psz&IntrospectionEndpoint.introspect_tokenN) Ú__name__Ú __module__Ú __qualname__Ú__doc__Ú ENDPOINT_NAMErr rrrr r!rrrrrs  rN)Úauthlib.constsrÚrfc6749rrrrrrrrÚs