codekingpro/portable-devtools

1from authlib.jose import JsonWebToken, JsonWebKey2from authlib.oidc.core import UserInfo, CodeIDToken, ImplicitIDToken3 4__all__ = ['AsyncOpenIDMixin']5 6 7class AsyncOpenIDMixin:8    async def fetch_jwk_set(self, force=False):9        metadata = await self.load_server_metadata()10        jwk_set = metadata.get('jwks')11        if jwk_set and not force:12            return jwk_set13 14        uri = metadata.get('jwks_uri')15        if not uri:16            raise RuntimeError('Missing "jwks_uri" in metadata')17 18        async with self.client_cls(**self.client_kwargs) as client:19            resp = await client.request('GET', uri, withhold_token=True)20            resp.raise_for_status()21            jwk_set = resp.json()22 23        self.server_metadata['jwks'] = jwk_set24        return jwk_set25 26    async def userinfo(self, **kwargs):27        """Fetch user info from ``userinfo_endpoint``."""28        metadata = await self.load_server_metadata()29        resp = await self.get(metadata['userinfo_endpoint'], **kwargs)30        resp.raise_for_status()31        data = resp.json()32        return UserInfo(data)33 34    async def parse_id_token(self, token, nonce, claims_options=None):35        """Return an instance of UserInfo from token's ``id_token``."""36        claims_params = dict(37            nonce=nonce,38            client_id=self.client_id,39        )40        if 'access_token' in token:41            claims_params['access_token'] = token['access_token']42            claims_cls = CodeIDToken43        else:44            claims_cls = ImplicitIDToken45 46        metadata = await self.load_server_metadata()47        if claims_options is None and 'issuer' in metadata:48            claims_options = {'iss': {'values': [metadata['issuer']]}}49 50        alg_values = metadata.get('id_token_signing_alg_values_supported')51        if not alg_values:52            alg_values = ['RS256']53 54        jwt = JsonWebToken(alg_values)55 56        jwk_set = await self.fetch_jwk_set()57        try:58            claims = jwt.decode(59                token['id_token'],60                key=JsonWebKey.import_key_set(jwk_set),61                claims_cls=claims_cls,62                claims_options=claims_options,63                claims_params=claims_params,64            )65        except ValueError:66            jwk_set = await self.fetch_jwk_set(force=True)67            claims = jwt.decode(68                token['id_token'],69                key=JsonWebKey.import_key_set(jwk_set),70                claims_cls=claims_cls,71                claims_options=claims_options,72                claims_params=claims_params,73            )74 75        # https://github.com/lepture/authlib/issues/25976        if claims.get('nonce_supported') is False:77            claims.params['nonce'] = None78        claims.validate(leeway=120)79        return UserInfo(claims)80