mirror of
https://github.com/google/earthengine-api.git
synced 2025-12-08 19:26:12 +00:00
86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
#!/usr/bin/env python
|
|
"""Earth Engine OAuth2 helper functions for generating client tokens.
|
|
|
|
Typical use-case consists of:
|
|
1. Calling 'get_authorization_url'
|
|
2. Using a browser to access the output URL and copy the generated OAuth2 code
|
|
3. Calling 'request_token' to request a token using that code and the OAuth API
|
|
4. Calling 'write_token' to save the token at the path given by
|
|
'get_credentials_path'
|
|
"""
|
|
|
|
|
|
import datetime
|
|
import errno
|
|
import json
|
|
import os
|
|
from six.moves.urllib import parse
|
|
from six.moves.urllib import request
|
|
from six.moves.urllib.error import HTTPError
|
|
|
|
|
|
CLIENT_ID = ('517222506229-vsmmajv00ul0bs7p89v5m89qs8eb9359.'
|
|
'apps.googleusercontent.com')
|
|
CLIENT_SECRET = 'RUP0RZ6e0pPhDzsqIJ7KlNd1'
|
|
SCOPES = [
|
|
'https://www.googleapis.com/auth/earthengine',
|
|
'https://www.googleapis.com/auth/cloud-platform'
|
|
]
|
|
REDIRECT_URI = 'urn:ietf:wg:oauth:2.0:oob' # Prompts user to copy-paste code
|
|
TOKEN_URI = 'https://accounts.google.com/o/oauth2/token'
|
|
|
|
|
|
def get_credentials_path():
|
|
cred_path = os.path.expanduser('~/.config/earthengine/credentials')
|
|
return cred_path
|
|
|
|
|
|
def get_authorization_url():
|
|
"""Returns a URL to generate an auth code."""
|
|
|
|
return 'https://accounts.google.com/o/oauth2/auth?' + parse.urlencode({
|
|
'client_id': CLIENT_ID,
|
|
'scope': ' '.join(SCOPES),
|
|
'redirect_uri': REDIRECT_URI,
|
|
'response_type': 'code',
|
|
})
|
|
|
|
|
|
def request_token(auth_code):
|
|
"""Uses authorization code to request tokens."""
|
|
|
|
request_args = {
|
|
'code': auth_code,
|
|
'client_id': CLIENT_ID,
|
|
'client_secret': CLIENT_SECRET,
|
|
'redirect_uri': REDIRECT_URI,
|
|
'grant_type': 'authorization_code',
|
|
}
|
|
|
|
refresh_token = None
|
|
|
|
try:
|
|
response = request.urlopen(
|
|
TOKEN_URI,
|
|
parse.urlencode(request_args).encode()).read().decode()
|
|
refresh_token = json.loads(response)['refresh_token']
|
|
except HTTPError as e:
|
|
raise Exception('Problem requesting tokens. Please try again. %s %s' %
|
|
(e, e.read()))
|
|
|
|
return refresh_token
|
|
|
|
|
|
def write_token(refresh_token):
|
|
"""Attempts to write the passed token to the given user directory."""
|
|
|
|
credentials_path = get_credentials_path()
|
|
dirname = os.path.dirname(credentials_path)
|
|
try:
|
|
os.makedirs(dirname)
|
|
except OSError as e:
|
|
if e.errno != errno.EEXIST:
|
|
raise Exception('Error creating directory %s: %s' % (dirname, e))
|
|
|
|
json.dump({'refresh_token': refresh_token}, open(credentials_path, 'w'))
|