codekingpro/portable-devtools

1from authlib.jose import jwt, JsonWebToken, JsonWebKey2from authlib.oidc.core import UserInfo, CodeIDToken, ImplicitIDToken3 4 5class OpenIDMixin:6    def fetch_jwk_set(self, force=False):7        metadata = self.load_server_metadata()8        jwk_set = metadata.get('jwks')9        if jwk_set and not force:10            return jwk_set11 12        uri = metadata.get('jwks_uri')13        if not uri:14            raise RuntimeError('Missing "jwks_uri" in metadata')15 16        with self.client_cls(**self.client_kwargs) as session:17            resp = session.request('GET', uri, withhold_token=True)18            resp.raise_for_status()19            jwk_set = resp.json()20 21        self.server_metadata['jwks'] = jwk_set22        return jwk_set23 24    def userinfo(self, **kwargs):25        """Fetch user info from ``userinfo_endpoint``."""26        metadata = self.load_server_metadata()27        resp = self.get(metadata['userinfo_endpoint'], **kwargs)28        resp.raise_for_status()29        data = resp.json()30        return UserInfo(data)31 32    def parse_id_token(self, token, nonce, claims_options=None, leeway=120):33        """Return an instance of UserInfo from token's ``id_token``."""34        if 'id_token' not in token:35            return None36 37        def load_key(header, _):38            jwk_set = JsonWebKey.import_key_set(self.fetch_jwk_set())39            try:40                return jwk_set.find_by_kid(header.get('kid'))41            except ValueError:42                # re-try with new jwk set43                jwk_set = JsonWebKey.import_key_set(self.fetch_jwk_set(force=True))44                return jwk_set.find_by_kid(header.get('kid'))45 46        claims_params = dict(47            nonce=nonce,48            client_id=self.client_id,49        )50        if 'access_token' in token:51            claims_params['access_token'] = token['access_token']52            claims_cls = CodeIDToken53        else:54            claims_cls = ImplicitIDToken55 56        metadata = self.load_server_metadata()57        if claims_options is None and 'issuer' in metadata:58            claims_options = {'iss': {'values': [metadata['issuer']]}}59 60        alg_values = metadata.get('id_token_signing_alg_values_supported')61        if alg_values:62            _jwt = JsonWebToken(alg_values)63        else:64            _jwt = jwt65 66        claims = _jwt.decode(67            token['id_token'], key=load_key,68            claims_cls=claims_cls,69            claims_options=claims_options,70            claims_params=claims_params,71        )72        # https://github.com/lepture/authlib/issues/25973        if claims.get('nonce_supported') is False:74            claims.params['nonce'] = None75 76        claims.validate(leeway=leeway)77        return UserInfo(claims)78