如何使用 oauth2 并通过 google api 刷新令牌？

Posted 2023-03-25

【中文标题】如何使用 oauth2 并通过 google api 刷新令牌？

【英文标题】：How do I use oauth2 and refresh tokens with the google api?

【发布时间】：2018-06-22 21:21:29

【问题描述】：

所以我最近几天才试图解决这个问题并提出这个问题，以便我可以为其他遇到问题的人回答这个问题。

首先，谷歌文档很糟糕，并且根据您正在查看的众多谷歌 API 示例中的哪一个使用不同的 oauth2 库。它通常是自相矛盾的，有时直接包含不起作用的代码。

哦，好吧。

所以我的问题基本上是：

如何使用 google api 库让我的用户授予我访问他们的 google 帐户的权限？ 如何存储 google 返回的 oauth2 访问令牌，以便我可以在几天后使用它们？ 我如何实际使用 refresh_token 并刷新它？

有关完整功能的授权流程，从获取初始令牌到保存、稍后加载、刷新和使用它，请参阅下面的答案。

干杯。

【参考方案1】：

首先，关于如何使用其 API 的 google 文档非常糟糕且自相矛盾。

这是我的解决方案（使用他们的库）使用 oauth2 来使用我存储在数据库中并定期刷新的令牌。我正在使用 django 2.0 和 python 3.6。这些都在我的“views.py”文件中。

首先，导入和其他脚本范围的设置：

import google.oauth2.credentials
import google.auth.transport.requests
import google_auth_oauthlib.flow
from googleapiclient.discovery import build

import os
import json
import datetime

API_SCOPE = ['https://mail.google.com/',]
JSON_FILE = "test_server_client_json.json"
JSON_PATH = os.path.join(os.getcwd(),"<folder_name>",JSON_FILE)
if settings.TEST_SERVER:
    REDIRECT_URI = "http://localhost:5000/oauth2/gmail/"
    #we don't have ssl on local host
    os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
else:
    REDIRECT_URL = "https://www.example.com/oauth2/gmail/"

好的，这是我们发送给用户以启动身份验证过程的第一个服务器端点/页面

@login_required
def connect_gmail_to_manager_page_1(request):
    #this is the function that a new user uses to set up their gmail account
    #and connect it to our system.
    #this particular page is used to:
    #1) have the user enter their email address so we know what is going on
    #2) explain the process
    #=====================    
    #basically we get their email address, and thats it, on this page. then     we send them 
    #to google to grant us access.
    if request.method == "POST":
        form = admin.getEmailAddress(request.POST)
        if form.is_valid():
            #first, get their email address. this is optional.
            #i'm using django and their forms to get it.
            new_email = form.cleaned_data.get("email")
            #-----
            #we are going to create the flow object using <redacted>'s keys and such
            flow = google_auth_oauthlib.flow.Flow.from_client_secrets_file(
                JSON_PATH,
                scopes=API_SCOPE)

            flow.redirect_uri = REDIRECT_URI

            # Generate URL for request to Google's OAuth 2.0 server.
            # Use kwargs to set optional request parameters.
            authorization_url, state = flow.authorization_url(
                # Enable offline access so that you can refresh an access     token without
                # re-prompting the user for permission. Recommended for web     server apps.
                access_type='offline',
                #which email is trying to login?
                login_hint=new_email,
                # Enable incremental authorization. Recommended as a         best     practice.
                include_granted_scopes='true')

            #and finally, we send them off to google for them to provide     access:
            return HttpResponseRedirect(authorization_url)
    else:    
        form = admin.getEmailAddress()
    token = 
    token.update(csrf(request))
    token['form'] = form
    return render(request,'connect_gmail_to_manager_page_1.html',token)

这会将用户发送到 google 以授予我们授权。在他们授予它之后，用户将被重定向到我们服务器上的授权端点。这是我的授权端点（我在这里删除了一些特定于项目的代码）

@login_required
def g_auth_endpoint(request):
    #this is the endpoint that the logged in token is sent to
    #here we are basically exchanging the auth code provided by gmail for an     access token.
    #the access token allows us to send emails.
    #it is a passthrough endpoint: we want to redirect to the next stage of 
    #whatever process they are doing here on completion.
    #===============================================
    #first we need to get the paramater 'state' from the url
    #NOTE that you should do some error handling here incase its not a valid token. I've removed that for brevity on stack overflow
    state = request.GET.get('state',None)

    flow = google_auth_oauthlib.flow.Flow.from_client_secrets_file(
        JSON_PATH,
        scopes=API_SCOPE,
        state=state)
    flow.redirect_uri = REDIRECT_URI

    #get the full URL that we are on, including all the "?param1=token&param2=key" parameters that google has sent us.
    authorization_response = request.build_absolute_uri()        

    #now turn those parameters into a token.
    flow.fetch_token(authorization_response=authorization_response)

    credentials = flow.credentials

    #now  we build the API service object    
    service = build('gmail', 'v1', credentials=credentials)
    #ok. awesome!
    #what email did they use? (this is just an example of how to use the api - you can skip this part if you want)
    profile = service.users().getProfile(userId="me").execute()
    email_address = profile['emailAddress']
    #ok. now we get the active manager
    manager = get_active_manager(request.user)
    #<lots of project specific code removed>
    #NOTE: 'manager' object is a project-specific type of object.
    #I store the auth token in it.
    #alright, if we get to here we have a valid manager object.

    #now lets create/update the credentials object in the DB.
    temp = save_credentials(manager,credentials)
    #now send them on their merry way that you've got access
    return HttpResponse("http://www.example.com")

这是我正在使用的保存/加载功能。请注意，“经理”和“Gmail_Connection_Token”对象是我保存令牌的项目特定对象。

def save_credentials(manager,credentials,valid=True):
    #this is the function that should be called to save the various tokens.
    #credentials is a google.oauth2.credentials.Credentials() object.
    #this saves it in a format that is easy to turn back 
    #into the same type of object in load_credentials(manager).
    #valid is, for the most part, always going to be true, but if for some reason its not
    #make sure to set that flag.
    #this returns the credentials as a dict (ignores the valid flag)
    #---------------------------------------
    #first we get or create the correct DB object
    try:
        creds = Gmail_Connection_Token.objects.get(manager=manager)
    except Gmail_Connection_Token.DoesNotExist:
        creds = Gmail_Connection_Token()
        creds.manager = manager
    #now we turn the passed in credentials obj into a dicts obj
    #note the expiry formatting
    temp = 
        'token': credentials.token,
        'refresh_token': credentials.refresh_token,
        'id_token':credentials.id_token,
        'token_uri': credentials.token_uri,
        'client_id': credentials.client_id,
        'client_secret': credentials.client_secret,
        'scopes': credentials.scopes,
        'expiry':datetime.datetime.strftime(credentials.expiry,'%Y-%m-%d %H:%M:%S')
    
    #now we save it as a json_string into the creds DB obj
    creds.json_string = json.dumps(temp)
    #update the valid flag.
    creds.valid = valid
    #and save everythign in the DB
    creds.save()
    #and finally, return the dict we just created.
    return temp

以下是我在需要时加载令牌的方式：

def load_credentials(manager,ignore_valid=False):
    #this is the function that should be called to load a credentials object     from the database.
    #it loads, refreshes, and returns a     google.oauth2.credentials.Credentials() object.
    #raises a value error if valid = False 
    #------
    #NOTE: if 'ignore_valid' is True:
    #will NOT raise a value error if valid == False
    #returns a Tuple formated as (Credentails(),valid_boolean)
    #======================================
    try:
        creds = Gmail_Connection_Token.objects.get(manager=manager)
    except: 
        #if something goes wrong here, we want to just raise the error
        #and pass it to the calling function.
        raise #yes, this is proper syntax! (don't want to lose the stack     trace)
    #is it valid? do we raise an error?
    if not ignore_valid and not creds.valid:
        raise ValueError('Credentials are not valid.')
    #ok, if we get to here we load/create the Credentials obj()
    temp = json.loads(creds.json_string)
    credentials = google.oauth2.credentials.Credentials(
        temp['token'],
        refresh_token=temp['refresh_token'],
        id_token=temp['id_token'],
        token_uri=temp['token_uri'],
        client_id=temp['client_id'],
        client_secret=temp['client_secret'],
        scopes=temp['scopes'],
    )
    expiry = temp['expiry']
    expiry_datetime = datetime.datetime.strptime(expiry,'%Y-%m-%d %H:%M:%S')
    credentials.expiry = expiry_datetime
    #and now we refresh the token   
    #but not if we know that its not a valid token.
    if creds.valid:
        request = google.auth.transport.requests.Request()
        if credentials.expired:
            credentials.refresh(request)
    #and finally, we return this whole deal
    if ignore_valid:
        return (credentials,creds.valid)
    else:
        return credentials

这几乎就是一切。这是一个示例端点，展示了在您需要访问 Gmail api 时如何使用这些功能

@login_required
def test_endpoint(request):
    #get the project-specific manager database object we are using to store the tokens
    manager = get_active_manager(request.user)
    #and convert that manager object into the google credentials object
    credentials = load_credentials(manager)

    #do whatever you need the gmail api for here:
    msg = send_test_email(credentials)

    #when you're done, make sure to save/update the credentials in the DB for future use.
    save_credentials(manager,credentials)

    #then send your user on their merry way.
    return HttpResponse(msg)

【讨论】：

很好的例子。请小心使用用户凭据存储您的客户端 ID 和客户端密码。

当访问令牌因刷新而改变时，google API wrapper如何调用save_credentials？