我正在使用Google OAuth2流程，如下所述:https://developers.google.com/identity/protocols/oauth2/web-server.
首先检查用户凭证是否已经在DB中，如果没有，则启动验证流。在将代码交换为访问令牌之后，凭证将保存在DB中。除了访问令牌和刷新令牌外，还保存了令牌过期日期，即1小时(凭据.过期)。
然后我想检查如何刷新访问令牌。所以我在令牌过期后显式地请求API。以前保存的凭据在DB中，所以我从那里获取它们。检查其有效性，并更新访问令牌。然而，由于某种原因，令牌是有效的，并且通过向Google API发出测试请求，我有点可以发出请求，不会引发错误。
我不明白为什么令牌是有效的，当它明显是过期的。我还检查了令牌的有效性:https://www.googleapis.com/oauth2/v3/tokeninfo?access_token=AT并得到{"error_description": "无效值"} .
我代码:
def google_oauth_init_flow(request):
# get user id
user_id = request.GET.get('user_id', None)
print(user_id)
user = User.objects.get(id=user_id)
# set id to a state parameter
request.session['state'] = str(user.id)
flow = google_auth_oauthlib.flow.Flow.from_client_secrets_file(
os.path.abspath(os.path.join(os.path.dirname(__file__), KEY_FILE_LOCATION)),
scopes=SCOPES)
flow.redirect_uri = REDIRECT_URI
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',
# Enable incremental authorization. Recommended as a best practice.
include_granted_scopes='true',
state=str(user.id)
)
return HttpResponseRedirect(authorization_url)
def google_oauth_exchange_token(request):
state = request.session.get('state', 'No state')
print(state)
flow = google_auth_oauthlib.flow.Flow.from_client_secrets_file(
os.path.abspath(os.path.join(os.path.dirname(__file__), KEY_FILE_LOCATION)),
scopes=SCOPES,
state=state)
flow.redirect_uri = REDIRECT_URI
authorization_response = request.build_absolute_uri()
flow.fetch_token(authorization_response=authorization_response)
credentials = flow.credentials
# save creds in DB
try:
save_credentials(credentials, state)
except IntegrityError as e:
if 'unique constraint' in e.message:
# duplicate detected
return HttpResponse('Violation of unique constraint')
def google_oauth_check_token(credentials, user_id):
# convert dict into credentials instance
if type(credentials) == dict:
credentials = init_creds_instance(credentials)
if credentials.expired:
print('Token is expired getting new one...')
# refresh credentials
request = google.auth.transport.requests.Request()
credentials.refresh(request)
# alternative method
# credentials.refresh(httplib2.Http())
# update token in DB
SystemServiceTokens.objects.filter(id=user_id).update(name='token', key=credentials.token)
return credentials
else: # always Token is valid
print('TOKEN', credentials.token)
print('EXPIRY', credentials.expiry)
print('REFRESH TOKEN', credentials.refresh_token)
print('Token is valid')
return credentials
def get_user_info(credentials):
user_info = build(serviceName='oauth2', version='v2', credentials=credentials)
return user_info.userinfo().get().execute()
def save_credentials(credentials, state):
user = User.objects.get(id=state)
google_oauth_check_token(credentials, user.id)
# get email
user_info = get_user_info(credentials)
model = SystemServiceTokens()
# check duplicate values
if not model.objects.filter(user_id=user.id, email=user_info['email'], service_id='ga').exists():
print("Entry contained in queryset")
model.token = credentials.token
model.refresh_token = credentials.refresh_token
model.expires_at = datetime.datetime.strftime(credentials.expiry, '%Y-%m-%d %H:%M:%S')
model.user = user
model.email = user_info['email']
model.service_id = 'ga'
return model.save()
else:
return HttpResponse('Unique constraint violation')
# No errors
def test_request_google(request):
credentials = SystemServiceTokens.objects.filter(user_id=9).first() # DEBUG ID
print(model_to_dict(credentials))
credentials = google_oauth_check_token(model_to_dict(credentials), 9)
# test
user_info = build(serviceName='oauth2', version='v2', credentials=credentials)
print(user_info.userinfo().get().execute())
drive = build('drive', 'v2', credentials=credentials)
如果python客户端库可以检测到刷新令牌，它将为您刷新它。假设您的代码工作正常，您应该永远不会有访问令牌过期-库将在到期前五分钟刷新它。
我的Python不是最好的，但它的源代码可能是这样的。http.py # L1559