Made cache_path and username optional (#567)

* Made cache_path and username optional

* Update readme and example

* Lint

* Lint

* Feedback / add warning
This commit is contained in:
Stéphane Bruckert 2020-08-30 23:59:38 +01:00 committed by GitHub
parent d448d33704
commit d0ca977647
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 126 additions and 137 deletions

View File

@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `SpotifyImplicitGrant` warns of security considerations and recommends
`SpotifyPKCE`
### Changed
- Specifying a cache_path or username is now optional
### Fixed
- Using `SpotifyPKCE.get_authorization_url` will now generate a code

View File

@ -50,7 +50,6 @@ from spotipy.oauth2 import SpotifyOAuth
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(client_id="YOUR_APP_CLIENT_ID",
client_secret="YOUR_APP_CLIENT_SECRET",
redirect_uri="YOUR_APP_REDIRECT_URI",
username="YOUR_SPOTIFY_USERNAME",
scope="user-library-read"))
results = sp.current_user_saved_tracks()

View File

@ -1,16 +1,8 @@
# Shows the top tracks for a user
import sys
import spotipy
from spotipy.oauth2 import SpotifyOAuth
if len(sys.argv) > 1:
username = sys.argv[1]
else:
print("Usage: %s username" % (sys.argv[0],))
sys.exit()
scope = 'user-top-read'
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(scope=scope))

View File

@ -15,7 +15,7 @@ if len(sys.argv) > 2:
track_ids.append({"uri": tid, "positions": [int(pos)]})
else:
print(
"Usage: %s username playlist_id track_id,pos track_id,pos ..." %
"Usage: %s playlist_id track_id,pos track_id,pos ..." %
(sys.argv[0],))
sys.exit()

View File

@ -10,7 +10,7 @@ if len(sys.argv) > 3:
playlist_id = sys.argv[1]
track_ids = sys.argv[2:]
else:
print("Usage: %s username playlist_id track_id ..." % (sys.argv[0],))
print("Usage: %s playlist_id track_id ..." % (sys.argv[0],))
sys.exit()
scope = 'playlist-modify-public'

View File

@ -1,5 +1,4 @@
# shows artist info for a URN or URL
# Shows artist info for a URN or URL
from spotipy.oauth2 import SpotifyClientCredentials
import spotipy

View File

@ -78,6 +78,17 @@ def _ensure_value(value, env_key):
return _val
def _get_cache_path(cache_path, username):
if cache_path:
return cache_path
cache_path = ".cache"
if username:
cache_path += "-" + str(username)
return cache_path
class SpotifyAuthBase(object):
def __init__(self, requests_session):
if isinstance(requests_session, requests.Session):
@ -219,7 +230,6 @@ class SpotifyOAuth(SpotifyAuthBase):
"""
Implements Authorization Code Flow for Spotify's OAuth implementation.
"""
OAUTH_AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
OAUTH_TOKEN_URL = "https://accounts.spotify.com/api/token"
@ -238,18 +248,22 @@ class SpotifyOAuth(SpotifyAuthBase):
requests_timeout=None
):
"""
Creates a SpotifyOAuth object
Creates a SpotifyOAuth object
Parameters:
- client_id - the client id of your app
- client_secret - the client secret of your app
- redirect_uri - the redirect URI of your app
- state - security state
- scope - the desired scope of the request
- cache_path - path to location to save tokens
- requests_timeout - tell Requests to stop waiting for a response
after a given number of seconds
- username - username of current client
Parameters:
* client_id: Must be supplied or set as environment variable
* client_secret: Must be supplied or set as environment variable
* redirect_uri: Must be supplied or set as environment variable
* state: May be supplied, no verification is performed
* scope: May be supplied, intuitively converted to proper format
* cache_path: May be supplied, will otherwise be generated
(takes precedence over `username`)
* username: May be supplied or set as environment variable
(will set `cache_path` to `.cache-{username}`)
* proxies: Proxy for the requests library to route through
* show_dialog: Interpreted as boolean
* requests_timeout: Tell Requests to stop waiting for a response after a given number
of seconds
"""
super(SpotifyOAuth, self).__init__(requests_session)
@ -258,10 +272,10 @@ class SpotifyOAuth(SpotifyAuthBase):
self.client_secret = client_secret
self.redirect_uri = redirect_uri
self.state = state
self.cache_path = cache_path
self.username = username or os.getenv(
CLIENT_CREDS_ENV_VARS["client_username"]
)
self.cache_path = _get_cache_path(cache_path, self.username)
self.scope = self._normalize_scope(scope)
self.proxies = proxies
self.requests_timeout = requests_timeout
@ -272,33 +286,26 @@ class SpotifyOAuth(SpotifyAuthBase):
"""
token_info = None
if not self.cache_path and self.username:
self.cache_path = ".cache-" + str(self.username)
elif not self.cache_path and not self.username:
raise SpotifyOauthError(
"You must either set a cache_path or a username."
)
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
if self.cache_path:
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
# if scopes don't match, then bail
if "scope" not in token_info or not self._is_scope_subset(
self.scope, token_info["scope"]
):
return None
# if scopes don't match, then bail
if "scope" not in token_info or not self._is_scope_subset(
self.scope, token_info["scope"]
):
return None
if self.is_token_expired(token_info):
token_info = self.refresh_access_token(
token_info["refresh_token"]
)
if self.is_token_expired(token_info):
token_info = self.refresh_access_token(
token_info["refresh_token"]
)
except IOError:
logger.warning("Couldn't read cache at: %s", self.cache_path)
except IOError:
pass
return token_info
def _save_token_info(self, token_info):
@ -579,28 +586,33 @@ class SpotifyPKCE(SpotifyAuthBase):
requests_timeout=None,
requests_session=True,):
"""
Creates Auth Manager with the PKCE Auth flow.
Creates Auth Manager with the PKCE Auth flow.
Parameters:
- client_id - the client id of your app
- redirect_uri - the redirect URI of your app
- state - security state
- scope - the desired scope of the request
- cache_path - path to location to save tokens
- username - username of current client
- proxies - proxy for the requests library to route through
- requests_timeout - tell Requests to stop waiting for a response
after a given number of seconds
Parameters:
* client_id: Must be supplied or set as environment variable
* client_secret: Must be supplied or set as environment variable
* redirect_uri: Must be supplied or set as environment variable
* state: May be supplied, no verification is performed
* scope: May be supplied, intuitively converted to proper format
* cache_path: May be supplied, will otherwise be generated
(takes precedence over `username`)
* username: May be supplied or set as environment variable
(will set `cache_path` to `.cache-{username}`)
* show_dialog: Interpreted as boolean
* proxies: Proxy for the requests library to route through
* requests_timeout: Tell Requests to stop waiting for a response after a given number
of seconds
"""
super(SpotifyPKCE, self).__init__(requests_session)
self.client_id = client_id
self.redirect_uri = redirect_uri
self.state = state
self.scope = self._normalize_scope(scope)
self.cache_path = cache_path
self.username = username or os.getenv(
CLIENT_CREDS_ENV_VARS["client_username"]
)
self.cache_path = _get_cache_path(cache_path, self.username)
self.proxies = proxies
self.requests_timeout = requests_timeout
@ -617,10 +629,10 @@ class SpotifyPKCE(SpotifyAuthBase):
return None
def _get_code_verifier(self):
''' Spotify PCKE code verifier - See step 1 of the reference guide below
""" Spotify PCKE code verifier - See step 1 of the reference guide below
Reference:
https://developer.spotify.com/documentation/general/guides/authorization-guide/#authorization-code-flow-with-proof-key-for-code-exchange-pkce
'''
"""
# Range (33,96) is used to select between 44-128 base64 characters for the
# next operation. The range looks weird because base64 is 6 bytes
import random
@ -638,10 +650,10 @@ class SpotifyPKCE(SpotifyAuthBase):
return verifier
def _get_code_challenge(self):
''' Spotify PCKE code challenge - See step 1 of the reference guide below
""" Spotify PCKE code challenge - See step 1 of the reference guide below
Reference:
https://developer.spotify.com/documentation/general/guides/authorization-guide/#authorization-code-flow-with-proof-key-for-code-exchange-pkce
'''
"""
import hashlib
import base64
code_challenge_digest = hashlib.sha256(self.code_verifier.encode('utf-8')).digest()
@ -744,33 +756,26 @@ class SpotifyPKCE(SpotifyAuthBase):
"""
token_info = None
if not self.cache_path and self.username:
self.cache_path = ".cache-" + str(self.username)
elif not self.cache_path and not self.username:
raise SpotifyOauthError(
"You must either set a cache_path or a username."
)
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
if self.cache_path:
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
# if scopes don't match, then bail
if "scope" not in token_info or not self._is_scope_subset(
self.scope, token_info["scope"]
):
return None
# if scopes don't match, then bail
if "scope" not in token_info or not self._is_scope_subset(
self.scope, token_info["scope"]
):
return None
if self.is_token_expired(token_info):
token_info = self.refresh_access_token(
token_info["refresh_token"]
)
if self.is_token_expired(token_info):
token_info = self.refresh_access_token(
token_info["refresh_token"]
)
except IOError:
logger.warning("Couldn't read cache at: %s", self.cache_path)
except IOError:
pass
return token_info
def _is_scope_subset(self, needle_scope, haystack_scope):
@ -853,9 +858,9 @@ class SpotifyPKCE(SpotifyAuthBase):
raise SpotifyOauthError('error: {0}, error_descr: {1}'.format(error_payload['error'],
error_payload[
'error_description'
]),
error=error_payload['error'],
error_description=error_payload['error_description'])
]),
error=error_payload['error'],
error_description=error_payload['error_description'])
token_info = response.json()
token_info = self._add_custom_values_to_token_info(token_info)
self._save_token_info(token_info)
@ -971,7 +976,9 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
* state: May be supplied, no verification is performed
* scope: May be supplied, intuitively converted to proper format
* cache_path: May be supplied, will otherwise be generated
* username: Must be supplied or set as environment variable
(takes precedence over `username`)
* username: May be supplied or set as environment variable
(will set `cache_path` to `.cache-{username}`)
* show_dialog: Interpreted as boolean
"""
logger.warning("The OAuth standard no longer recommends the Implicit "
@ -983,10 +990,10 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
self.client_id = client_id
self.redirect_uri = redirect_uri
self.state = state
self.cache_path = cache_path
self.username = username or os.getenv(
CLIENT_CREDS_ENV_VARS["client_username"]
)
self.cache_path = _get_cache_path(cache_path, self.username)
self.scope = self._normalize_scope(scope)
self.show_dialog = show_dialog
self._session = None # As to not break inherited __del__
@ -996,44 +1003,33 @@ class SpotifyImplicitGrant(SpotifyAuthBase):
"""
token_info = None
if not self.cache_path and self.username:
self.cache_path = ".cache-" + str(self.username)
elif not self.cache_path and not self.username:
raise SpotifyOauthError(
"You must either set a cache_path or a username."
)
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
if self.cache_path:
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
# if scopes don't match, then bail
if "scope" not in token_info or not self._is_scope_subset(
self.scope, token_info["scope"]
):
return None
# if scopes don't match, then bail
if "scope" not in token_info or not self._is_scope_subset(
self.scope, token_info["scope"]
):
return None
if self.is_token_expired(token_info):
return None
if self.is_token_expired(token_info):
return None
except IOError:
logger.warning("Couldn't read cache at: %s", self.cache_path)
except IOError:
pass
return token_info
def _save_token_info(self, token_info):
if not self.cache_path and self.username:
self.cache_path = ".cache-" + str(self.username)
if self.cache_path:
try:
f = open(self.cache_path, "w")
f.write(json.dumps(token_info))
f.close()
except IOError:
logger.warning('Couldn\'t write token to cache at: %s',
self.cache_path)
try:
f = open(self.cache_path, "w")
f.write(json.dumps(token_info))
f.close()
except IOError:
logger.warning("Couldn't write token to cache at: %s", self.cache_path)
def _is_scope_subset(self, needle_scope, haystack_scope):
needle_scope = set(needle_scope.split()) if needle_scope else set()

View File

@ -21,7 +21,7 @@ CLIENT_CREDS_ENV_VARS = {
def prompt_for_user_token(
username,
username=None,
scope=None,
client_id=None,
client_secret=None,
@ -43,14 +43,14 @@ def prompt_for_user_token(
Parameters:
- username - the Spotify username
- scope - the desired scope of the request
- client_id - the client id of your app
- client_secret - the client secret of your app
- redirect_uri - the redirect URI of your app
- cache_path - path to location to save tokens
- oauth_manager - Oauth manager object.
- show_dialog - If true, a login prompt always shows
- username - the Spotify username (optional)
- scope - the desired scope of the request (optional)
- client_id - the client id of your app (required)
- client_secret - the client secret of your app (required)
- redirect_uri - the redirect URI of your app (required)
- cache_path - path to location to save tokens (optional)
- oauth_manager - Oauth manager object (optional)
- show_dialog - If true, a login prompt always shows (optional, defaults to False)
"""
if not oauth_manager:
@ -79,14 +79,13 @@ def prompt_for_user_token(
)
raise spotipy.SpotifyException(550, -1, "no credentials set")
cache_path = cache_path or ".cache-" + username
sp_oauth = oauth_manager or spotipy.SpotifyOAuth(
client_id,
client_secret,
redirect_uri,
scope=scope,
cache_path=cache_path,
username=username,
show_dialog=show_dialog
)

View File

@ -436,8 +436,8 @@ class TestSpotifyPKCE(unittest.TestCase):
auth.get_pkce_handshake_parameters()
self.assertEqual(auth.code_challenge,
base64.urlsafe_b64encode(
hashlib.sha256(auth.code_verifier.encode('utf-8'))
.digest())
hashlib.sha256(auth.code_verifier.encode('utf-8'))
.digest())
.decode('utf-8')
.replace('=', ''))