Running this code:
from google.oauth2 import credentials
from google.cloud import storage
from oauthlib.oauth2 import DeviceClient
from requests_oauthlib import OAuth2Session
from google.auth.transport import requests
import json
import time
import os
# I'm trying to build a kiosk type appliance in tkinter in Python,
# where I can log in once, and then use the refresh token each time
# the program starts, until the refresh token expires and I need to
# re-authenticate (every 30 days?)
#
# To reproduce the problem I'm seeing, create a client for DeviceFlow
# authentication and configure the parameters below here.
CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")
PROJECT = os.getenv("PROJECT")
SCOPE = ['https://www.googleapis.com/auth/devstorage.read_write']
DEVICE_AUTH_URL = 'https://oauth2.googleapis.com/device/code'
TOKEN_URL = 'https://oauth2.googleapis.com/token'
# start a device client flow
client = DeviceClient(client_id=CLIENT_ID)
oauth = OAuth2Session(client=client)
device_auth_response = oauth.post(DEVICE_AUTH_URL, data={
'client_id': CLIENT_ID,
'scope': ' '.join(SCOPE)
})
darj = device_auth_response.json()
print(f"oauth response: {json.dumps(darj)}", flush=True)
device_code = darj['device_code']
user_code = darj['user_code']
verification_url = darj['verification_url']
print(f"Please visit {verification_url} and paste the code: {user_code}", flush=True)
# wait for the user to go through the flow
input("Press return when you are done:")
while True:
token_response = oauth.post(TOKEN_URL, data={
'client_id': CLIENT_ID,
'client_secret': CLIENT_SECRET,
'device_code': device_code,
'grant_type': 'urn:ietf:params:oauth:grant-type:device_code'
})
if token_response.status_code == 200:
# Successfully retrieved the token
token = token_response.json()
break
if token_response.json().get('error') == 'authorization_pending':
print("waiting for authorization ...", flush=True)
time.sleep(5.0)
else:
raise Exception(f"Error in token request: {token_response.json()['error']}")
print(f"got token: {json.dumps(token)}", flush=True)
# save credentials
with open("/tmp/creds.json", "w") as f:
json.dump(token, f)
# The first full login works
print("\n\ncase 1", flush=True)
creds1 = credentials.Credentials(token['access_token'],
refresh_token=token['refresh_token'],
token_uri=TOKEN_URL,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
scopes=SCOPE)
storage1 = storage.Client(credentials=creds1, project=PROJECT)
buckets1 = [b for b in storage1.list_buckets()]
print(f"buckets1: {len(buckets1)}")
# A login without the access token doesn't work, even though the
# docs for credentials.Credentials() says it should.
print("\n\ncase 2", flush=True)
try:
creds2 = credentials.Credentials(None,
refresh_token=token['refresh_token'],
token_uri=TOKEN_URL,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
scopes=SCOPE)
storage2 = storage.Client(credentials=creds2, project=PROJECT)
buckets2 = [b for b in storage2.list_buckets()]
print(f"buckets2: {len(buckets2)}")
except Exception as e:
print(f"case 2 didn't work: {e}", flush=True)
# Refreshing using the documented way to refresh also doesn't work
print("\n\ncase 3", flush=True)
try:
creds1.refresh(requests.Request())
print(f"refresh worked, access token {creds1.token}")
except Exception as e:
print(f"case 3 didn't work: {e}")
# wait for authtoken to expire
print("Waiting for authtoken to expire (takes 1 hour by default)", flush=True)
time.sleep(3601)
# read back credentials
with open("/tmp/creds.json", "r") as f:
token = json.load(f)
# The second login doesn't work, so something is being consumed
# in the first login.
print("\n\ncase 4", flush=True)
try:
creds4 = credentials.Credentials(token['access_token'],
refresh_token=token['refresh_token'],
token_uri=TOKEN_URL,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
scopes=SCOPE)
storage4 = storage.Client(credentials=creds4, project=PROJECT)
buckets4 = [b for b in storage4.list_buckets()]
print(f"buckets4: {len(buckets4)}")
except Exception as e:
print(f"case 4 didn't work: {e}", flush=True)Note: It takes over an hour because of the sleep before case 4. It demonstrates a bunch of cases I think should work based on documentation, but which don't.
(venv) jwatte@Jons-MacBook-Pro videoripper % python bug.py
oauth response: {"device_code": "AH-1Nxxxxx", "user_code": "PJV-xxxxx", "expires_in": 1800, "interval": 5, "verification_url": "https://www.google.com/device"}
Please visit https://www.google.com/device and paste the code: PJV-QQQ-TNT
Press return when you are done:
got token: {"access_token": "ya29.a0AXoxxxxx", "expires_in": 3599, "refresh_token": "1//06_xxxxx", "scope": "https://www.googleapis.com/auth/devstorage.read_write", "token_type": "Bearer"}
case 1
buckets1: 273
case 2
case 2 didn't work: Reauthentication is needed. Please run `gcloud auth application-default login` to reauthenticate.
case 3
case 3 didn't work: Reauthentication is needed. Please run `gcloud auth application-default login` to reauthenticate.
case 4
case 4 didn't work: Reauthentication is needed. Please run `gcloud auth application-default login` to reauthenticate.
Any reason you aren't using a service account with Cloud Storage and avoiding the whole OAuth dance drama?