+
    P(i                     L    ^ RI Ht ^RIHt ^RIHt ^RIHt  ! R R]4      tR# )    )default_json_headers)InvalidRequestError)TokenEndpoint)UnsupportedTokenTypeErrorc                   R   a  ] tR t^t o RtRtR tR tR tR t	R t
R tR	 tR
tV tR# )IntrospectionEndpointz}Implementation of introspection endpoint which is described in
`RFC7662`_.

.. _RFC7662: https://tools.ietf.org/html/rfc7662
introspectionc                    V P                  W4       V P                  VP                  R,          VP                  P                  R4      4      pV'       d   V P	                  W2V4      '       d   V# R# R# )a  The 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   s   &&& j/Users/agent/.openclaw/workspace/venv/lib/python3.14/site-packages/authlib/oauth2/rfc7662/introspection.pyauthenticate_token(IntrospectionEndpoint.authenticate_token   sd    & 	'*  LL!7<<#3#34E#F
 T**5'BBL C5    c                    VP                   pR V9  d   \        4       hVP                  R4      pV'       d   W@P                  9  d   \	        4       hR# R# )r   r   N)r   r   r   SUPPORTED_TOKEN_TYPESr   )r   r   r   paramshints   &&&  r   r   "IntrospectionEndpoint.check_params,   sI    & %''zz+,D : ::+-- ;4r   c                z    V P                  V4      pV P                  W4      pV P                  V4      p^V\        3# )z`Validate introspection request and create the response.

:returns: (status_code, body, headers)
)authenticate_endpoint_clientr   create_introspection_payloadr   )r   r   r   r   bodys   &&   r   create_endpoint_response.IntrospectionEndpoint.create_endpoint_response5   sF     227; ''8 007D...r   c                    V'       g   R R/# VP                  4       '       g   VP                  4       '       d   R R/# V P                  V4      pR V9  d   RVR &   V# )activeFT)
is_expired
is_revokedintrospect_token)r   r   payloads   && r   r    2IntrospectionEndpoint.create_introspection_payloadE   sa    
 e$$!1!1!3!3e$$''.7" $GHr   c                    \        4       h)a%  Check 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)r   r   r   r   s   &&&&r   r   &IntrospectionEndpoint.check_permissionS   s     "##r   c                    \        4       h)a2  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   s   &&&r   r   !IntrospectionEndpoint.query_token_   s     "##r   c                    \        4       h)a  Read 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,   )r   r   s   &&r   r(   &IntrospectionEndpoint.introspect_tokenp   s    ( "##r    N)__name__
__module____qualname____firstlineno____doc__ENDPOINT_NAMEr   r   r"   r    r   r   r(   __static_attributes____classdictcell__)__classdict__s   @r   r   r      s9      $M4./ 
$$"$ $r   r   N)authlib.constsr   rfc6749r   r   r   r   r4   r   r   <module>r@      s    / ) # /|$M |$r   