
    lKj                        d dl mZ d dlZd dlmZ d dlmZ d dlmZ d dlmZ d dl	m
Z
 d dlmZ d d	lmZ d d
lmZ ddlmZ dZ ej(                  e      Z G d d      Zy)    )annotationsN)jwk)jws)jwt)	JoseError)to_bytes)import_any_key)
json_loads)	deprecate   )InvalidClientErrorz6urn:ietf:params:oauth:client-assertion-type:jwt-bearerc                  V    e Zd ZdZeZdZddZd ZddZ	d Z
d Zd	 Zdd
Zd ZddZy)JWTBearerClientAssertionz]Implementation of Using JWTs for Client Authentication, which is
    defined by RFC7523.
    client_assertion_jwtNc                L    |t        dd       || _        || _        || _        y )Nz<'token_url' is deprecated. Override 'get_audiences' instead.1.8version)r   	token_url_validate_jtileeway)selfr   validate_jtir   s       n/Users/ahmed/devFolder/Ultron/claude-voice/.venv/lib/python3.12/site-packages/authlib/oauth2/rfc7523/client.py__init__z!JWTBearerClientAssertion.__init__   s0     N #)     c                    |j                   }|j                  d      }|j                  d      }|t        k(  r|r~| j                  |      \  }}|d   } ||      }	|	st	        d      	 t        | j                  |	            }
|	|_	        | j                  ||
       | j                  |j                        S t        j                  d	| j                         y # t        $ r+ t        | j                  |	|            }
t        dd       Y w xY w)
Nclient_assertion_typeclient_assertionsubz)The client does not exist on this server.descriptionz8resolve_client_public_key takes only 'client' parameter.r   r   zAuthenticate via %r failed)formgetASSERTION_TYPEextract_assertionr   r	   resolve_client_public_key	TypeErrorr   clientprocess_assertion_claimsauthenticate_clientlogdebugCLIENT_AUTH_METHOD)r   query_clientrequestdataassertion_type	assertionheadersclaims	client_idr)   keys              r   __call__z!JWTBearerClientAssertion.__call__+   s    ||"9:HH/0	^+	"44Y?OGVuI!),F( K $T%C%CF%KL $GN)))S9++GNN;;		.0G0GH  $T%C%CFG%TUN!s   *C 1DDc                   ddiddid| j                         dddid}t        j                  dd| j                  i|}	 |j	                  |       |d   |d	   k7  rt        d
      | j                  r2d|vrt        d      | j                  ||d         st        d      y y # t
        $ r2}t        j                  d|       t        |j                        |d }~ww xY w)N	essentialT)r:   values)issr    audexpr   Assertion Error: %rr!   r    r<   zIssuer and Subject MUST match.jtizMissing JWT ID.zJWT ID is used before. )get_audiencesr   JWTClaimsRegistryr   validater   r,   r-   r   r"   r   r   )r   r5   optionsclaims_requestses        r   verify_claimsz&JWTBearerClientAssertion.verify_claimsF   s      &&!%1C1C1EF&	
 //Nt{{NgN	G$$V,
 %=F5M)$1QRRF"(5FGG$$VVE];(5MNN <	   	GII+Q/$?QF	Gs   B( (	C#1-CC#c                H    | j                   | j                   gS t               )a  Return a list of valid audience identifiers for this authorization
        server. Per RFC 7523 Section 3, the audience identifies the
        authorization server as an intended audience.

        Developers MUST implement this method::

            def get_audiences(self):
                return ["https://example.com/oauth/token", "https://example.com"]

        :return: list of valid audience strings
        )r   NotImplementedError)r   s    r   rB   z&JWTBearerClientAssertion.get_audiences`   s$     >>%NN##!##r   c                    	 t        j                  ||      }| j                  |j                         |j                  S # t        $ r2}t        j	                  d|       t        |j                        |d}~ww xY w)aa  Extract JWT payload claims from request "assertion", per
        `Section 3.1`_.

        :param assertion: assertion string value in the request
        :param resolve_key: function to resolve the sign key
        :return: JWTClaims
        :raise: InvalidClientError

        .. _`Section 3.1`: https://tools.ietf.org/html/rfc7523#section-3.1
        r?   r!   N)	r   decoder   r,   r-   r   r"   rH   r5   )r   r3   resolve_keytokenrG   s        r   r*   z1JWTBearerClientAssertion.process_assertion_claimsp   sj    	GJJy+6E
 	5<<(||  	GII+Q/$?QF	Gs   ? 	A:-A55A:c                p    |j                  | j                  d      r|S t        d| j                         )NrN   z,The client cannot authenticate with method: r!   )check_endpoint_auth_methodr.   r   r   r)   s     r   r+   z,JWTBearerClientAssertion.authenticate_client   s;    ,,T-D-DgNM FtG^G^F_`
 	
r   c                    t        j                  t        |            }	 t        |j                        }|j                         |fS # t
        $ r t        d      d w xY w)NzInvalid JWT payload.r!   )r   extract_compactr   r
   payload
ValueErrorr   r4   )r   r3   objr5   s       r   r&   z*JWTBearerClientAssertion.extract_assertion   s_    !!(9"56	S,F {{}f$$  	S$1GHdR	Ss   A Ac                    t               )af  Validate if the given ``jti`` value is used before. Developers
        MUST implement this method::

            def validate_jti(self, claims, jti):
                key = "jti:{}-{}".format(claims["sub"], jti)
                if redis.get(key):
                    return False
                redis.set(key, 1, ex=3600)
                return True
        rJ   )r   r5   r@   s      r   r   z%JWTBearerClientAssertion.validate_jti   s     "##r   c                    t               )a  Resolve the client public key for verifying the JWT signature.
        Developers MUST implement this method::

            from joserfc.jwk import KeySet


            def resolve_client_public_key(self, client):
                return KeySet.import_key_set(client.public_jwks)
        rX   rQ   s     r   r'   z2JWTBearerClientAssertion.resolve_client_public_key   s     "##r   )NT<   )r5   z
jwt.Claims)r3   str)returnzjwk.Key | jwk.KeySet)__name__
__module____qualname____doc__r%   CLIENT_ASSERTION_TYPEr.   r   r8   rH   rB   r*   r+   r&   r   r'   rA   r   r   r   r      sD    
 +/
I6O4$ (
%$
$r   r   )
__future__r   loggingjoserfcr   r   r   joserfc.errorsr   joserfc.utilr   authlib._joserfc_helpersr	   authlib.common.encodingr
   authlib.deprecater   rfc6749r   r%   	getLoggerr]   r,   r   rA   r   r   <module>rl      sI    "     $ ! 3 . ' (Ig!U$ U$r   