codekingpro/portable-devtools

1from authlib.oauth2.rfc7009 import RevocationEndpoint as _RevocationEndpoint2 3 4class RevocationEndpoint(_RevocationEndpoint):5    """The revocation endpoint for OAuth authorization servers allows clients6    to notify the authorization server that a previously obtained refresh or7    access token is no longer needed.8 9    Register it into authorization server, and create token endpoint response10    for token revocation::11 12        from django.views.decorators.http import require_http_methods13 14        # see register into authorization server instance15        server.register_endpoint(RevocationEndpoint)16 17        @require_http_methods(["POST"])18        def revoke_token(request):19            return server.create_endpoint_response(20                RevocationEndpoint.ENDPOINT_NAME,21                request22            )23    """24 25    def query_token(self, token, token_type_hint):26        """Query requested token from database."""27        token_model = self.server.token_model28        if token_type_hint == 'access_token':29            rv = _query_access_token(token_model, token)30        elif token_type_hint == 'refresh_token':31            rv = _query_refresh_token(token_model, token)32        else:33            rv = _query_access_token(token_model, token)34            if not rv:35                rv = _query_refresh_token(token_model, token)36 37        return rv38 39    def revoke_token(self, token, request):40        """Mark the give token as revoked."""41        token.revoked = True42        token.save()43 44 45def _query_access_token(token_model, token):46    try:47        return token_model.objects.get(access_token=token)48    except token_model.DoesNotExist:49        return None50 51 52def _query_refresh_token(token_model, token):53    try:54        return token_model.objects.get(refresh_token=token)55    except token_model.DoesNotExist:56        return None57