~hJddlmZddlmZddlmZddlmZGddeZy) )default_json_headers)InvalidRequestError) TokenEndpoint)UnsupportedTokenTypeErrorc>eZdZdZdZdZdZdZdZdZ dZ d Z y ) IntrospectionEndpointzImplementation of introspection endpoint which is described in `RFC7662`_. .. _RFC7662: https://tools.ietf.org/html/rfc7662 introspectionc|j|||j|jd|jjd}|r|j |||r|Syy)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. tokentoken_type_hintN) check_params query_tokenformgetcheck_permission)selfrequestclientr s `/opt/mcp/mcp-sentiment/venv/lib/python3.12/site-packages/authlib/oauth2/rfc7662/introspection.pyauthenticate_tokenz(IntrospectionEndpoint.authenticate_tokense& '6*  LL !7<<#3#34E#F  T**5&'BLC5c|j}d|vr t|jd}|r||jvr t yy)Nr r )rrrSUPPORTED_TOKEN_TYPESr)rrrparamshints rrz"IntrospectionEndpoint.check_params,sK & %' 'zz+, D : ::+- -;4rc||j|}|j||}|j|}d|tfS)zpValidate introspection request and create the response. :returns: (status_code, body, headers) )authenticate_endpoint_clientrcreate_introspection_payloadr)rrrr bodys rcreate_endpoint_responsez.IntrospectionEndpoint.create_endpoint_response5sH 227;''8007D...rc|sddiS|js|jrddiS|j|}d|vrd|d<|S)NactiveFT) is_expired is_revokedintrospect_token)rr payloads rr z2IntrospectionEndpoint.create_introspection_payloadEs[ e$ $    !1!1!3e$ $''. 7 " $GH rct)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 rrs rrz&IntrospectionEndpoint.check_permissionSs "##rct)aGet 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 s rrz!IntrospectionEndpoint.query_token_s "##rct)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 s rr'z&IntrospectionEndpoint.introspect_tokenps ("##rN) __name__ __module__ __qualname____doc__ ENDPOINT_NAMErrr"r rrr'rrr r s1$M4./  $$"$rr N)authlib.constsrrfc6749rrrr r4rrr7s/)#/|$M|$r