o ^[2h:@sddlZddlmZddlmZddlmZddlmZddlmZddlmZdd lm Z dd lm Z d d l m Z e eZd ZGdddee ZdS)N) JoseError)jwt) BaseGrant)InvalidClientError)InvalidGrantError)InvalidRequestError)TokenEndpointMixin)UnauthorizedClientErrorsign_jwt_bearer_assertionz+urn:ietf:params:oauth:grant-type:jwt-bearerc@seZdZeZddiddiddidZdZe    dddZdd Z d d Z d d Z ddZ ddZ ddZddZddZdS)JWTBearerGrant essentialT)issaudexp<NcKst|||||||fi|S)Nr )keyissueraudiencesubject issued_at expires_atclaimskwargsrr/home/skpark/git/infrasmart_work/infrasmart/venv/lib/python3.10/site-packages/authlib/oauth2/rfc7523/jwt_bearer.pysign!s  zJWTBearerGrant.signc Cs\ztj||j|jd}|j|jdW|Sty-}z td|t |j d|d}~ww)a#Extract JWT payload claims from request "assertion", per `Section 3.1`_. :param assertion: assertion string value in the request :return: JWTClaims :raise: InvalidGrantError .. _`Section 3.1`: https://tools.ietf.org/html/rfc7523#section-3.1 )claims_options)leewayzAssertion Error: %r descriptionN) rdecoderesolve_public_keyCLAIMS_OPTIONSvalidateLEEWAYrlogdebugrr")self assertionrerrrprocess_assertion_claims0s   z'JWTBearerGrant.process_assertion_claimscCs||d}||||S)Nr)resolve_issuer_clientresolve_client_key)r*headerspayloadclientrrrr$Dsz!JWTBearerGrant.resolve_public_keycCs|jjd}|s td||}||d}td|||j s.t d|j d||j_ | |d}|ra| |}|sItdd td |||||s[td d ||j_d Sd S) aThe client makes a request to the token endpoint by sending the following parameters using the "application/x-www-form-urlencoded" format per `Section 2.1`_: grant_type REQUIRED. Value MUST be set to "urn:ietf:params:oauth:grant-type:jwt-bearer". assertion REQUIRED. Value MUST contain a single JWT. scope OPTIONAL. The following example demonstrates an access token request with a JWT as an authorization grant: .. code-block:: http POST /token.oauth2 HTTP/1.1 Host: as.example.com Content-Type: application/x-www-form-urlencoded grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer &assertion=eyJhbGciOiJFUzI1NiIsImtpZCI6IjE2In0. eyJpc3Mi[...omitted for brevity...]. J9l-ZhwP[...omitted for brevity...] .. _`Section 2.1`: https://tools.ietf.org/html/rfc7523#section-2.1 r+zMissing 'assertion' in requestrzValidate token request of %sz0The client is not authorized to use 'grant_type='subz Invalid 'sub' value in assertionr!z'Check client(%s) permission to User(%s)z,Client has no permission to access user dataN)requestformgetrr-r.r(r)check_grant_type GRANT_TYPEr r2validate_requested_scopeauthenticate_userrhas_granted_permissionruser)r*r+rr2rr=rrrvalidate_token_requestHs0         z%JWTBearerGrant.validate_token_requestcCsB|j|jjj|jjdd}td||jj||d||j fS)zZIf valid and authorized, the authorization server issues an access token. F)scoper=include_refresh_tokenzIssue token %r to %r) generate_tokenr5r1r?r=r(r)r2 save_tokenTOKEN_RESPONSE_HEADER)r*tokenrrrcreate_token_responses  z$JWTBearerGrant.create_token_responsecCt)a1Fetch client via "iss" in assertion claims. Developers MUST implement this method in subclass, e.g.:: def resolve_issuer_client(self, issuer): return Client.query_by_iss(issuer) :param issuer: "iss" value in assertion :return: Client instance NotImplementedError)r*rrrrr. z$JWTBearerGrant.resolve_issuer_clientcCrG)auResolve client key to decode assertion data. Developers MUST implement this method in subclass. For instance, there is a "jwks" column on client table, e.g.:: def resolve_client_key(self, client, headers, payload): # from authlib.jose import JsonWebKey key_set = JsonWebKey.import_key_set(client.jwks) return key_set.find_by_kid(headers["kid"]) :param client: instance of OAuth client model :param headers: headers part of the JWT :param payload: payload part of the JWT :return: ``authlib.jose.Key`` instance rH)r*r2r0r1rrrr/sz!JWTBearerGrant.resolve_client_keycCrG)a%Authenticate user with the given assertion claims. Developers MUST implement it in subclass, e.g.:: def authenticate_user(self, subject): return User.get_by_sub(subject) :param subject: "sub" value in claims :return: User instance rH)r*rrrrr;rJz JWTBearerGrant.authenticate_usercCrG)aCheck if the client has permission to access the given user's resource. Developers MUST implement it in subclass, e.g.:: def has_granted_permission(self, client, user): permission = ClientUserGrant.query(client=client, user=user) return permission.granted :param client: instance of OAuth client model :param user: instance of User model :return: bool rH)r*r2r=rrrr<s z%JWTBearerGrant.has_granted_permission)NNNN)__name__ __module__ __qualname__JWT_BEARER_GRANT_TYPEr9r%r' staticmethodrr-r$r>rFr.r/r;r<rrrrrs* <   r)logging authlib.joserrrfc6749rrrrr r r+r getLoggerrKr(rNrrrrrs