Backport of OAuth login from dbbbd5d

Author: morguldir

README.rst

@@ -5,7 +5,7 @@ tidalapi
     :target: https://pypi.org/project/tidalapi
 
 .. image:: https://api.netlify.com/api/v1/badges/f05c0752-4565-4940-90df-d2b3fe91c84b/deploy-status
-    :target: https://tidalapi.netlify.com/
+    :target: https://0-6-x--tidalapi.netlify.app/
 
 Unofficial Python API for TIDAL music streaming service.
 
@@ -35,7 +35,8 @@ Example usage
     import tidalapi
 
     session = tidalapi.Session()
-    session.login('username', 'password')
+    # Will run until you visit the printed url and link your account
+    session.login_oauth_simple()
     tracks = session.get_album_tracks(album_id=16909093)
     for track in tracks:
         print(track.name)
@@ -44,4 +45,4 @@ Example usage
 Documentation
 -------------
 
-Documentation is available at https://tidalapi.netlify.com
+Documentation is available at https://0-6-x--tidalapi.netlify.app/

docs/requirements.txt

@@ -1 +1,2 @@
 sphinx>=1.8.5
+futures

setup.py

@@ -8,6 +8,9 @@
 if sys.version_info < (3,4):
     required.append('enum34')
 
+if sys.version_info < (3,0):
+    required.append('futures')
+
 long_description = ""
 with open('README.rst') as f:
     long_description += f.read()

tests/test_api.py

@@ -16,9 +16,9 @@
 # You should have received a copy of the GNU Lesser General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from __future__ import print_function
 from __future__ import unicode_literals
 import logging
-import os
 import pytest
 import requests
 import tidalapi
@@ -27,11 +27,14 @@
 
 
 @pytest.fixture(scope='session')
-def session():
+def session(request):
     session = tidalapi.Session()
-    username = os.getenv("TIDAL_USERNAME")
-    password = os.getenv("TIDAL_PASSWORD")
-    session.login(username, password)
+    login, future = session.login_oauth()
+    capmanager = request.config.pluginmanager.getplugin("capturemanager")
+    with capmanager.global_and_fixture_disabled():
+        print("Visit", login.verification_uri_complete, "to log in, the link expires in", login.expires_in, "seconds")
+    future.result()
+    assert session.check_login()
     return session
 
 
@@ -151,14 +154,36 @@ def test_get_video_url(session):
     assert session.get_video_url(video.id) != None
 
 
-def test_load_session(session):
+def test_load_oauth_session(session):
     """
     Test loading a session from a session id without supplying country code and user_id
     """
     user_id = session.user.id
     country_code = session.country_code
     session_id = session.session_id
+    token_type = session.token_type
+    access_token = session.access_token
+    refresh_token = session.refresh_token
     session = tidalapi.Session()
+    session.load_oauth_session(session_id, token_type, access_token, refresh_token)
+    assert session.check_login()
     session.load_session(session_id)
     assert user_id == session.user.id
     assert country_code == session.country_code
+
+
+def test_oauth_refresh(session):
+    access_token = session.access_token
+    expiry_time = session.expiry_time
+    refresh_token = session.refresh_token
+    assert session.token_refresh(refresh_token)
+    assert session.access_token != access_token
+    assert session.expiry_time != expiry_time
+    assert session.check_login()
+
+
+def test_simple_login(capsys):
+    session = tidalapi.Session()
+    with capsys.disabled():
+        session.login_oauth_simple()
+    assert session.check_login()

tidalapi/__init__.py

@@ -16,15 +16,19 @@
 # You should have received a copy of the GNU Lesser General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from __future__ import print_function
 from __future__ import unicode_literals
 from collections import namedtuple
 from enum import Enum
 
+import concurrent.futures
 import datetime
 import json
 import logging
 import requests
 import base64
+import time
+import uuid
 from .models import Artist, Album, Track, Video, Playlist, SearchResult, Category, Role
 
 try:
@@ -47,6 +51,26 @@ class VideoQuality(Enum):
     low = 'LOW'
 
 
+class LinkLogin(object):
+    """
+    The data required for logging in to TIDAL using a remote link, json is the data returned from TIDAL
+    """
+    #: Amount of seconds until the code expires
+    expires_in = None
+    #: The code the user should enter at the uri
+    user_code = None
+    #: The link the user has to visit
+    verification_uri = None
+    #: The link the user has to visit, with the code already included
+    verification_uri_complete = None
+
+    def __init__(self, json):
+        self.expires_in = json['expiresIn']
+        self.user_code = json['userCode']
+        self.verification_uri = json['verificationUri']
+        self.verification_uri_complete = json['verificationUriComplete']
+
+
 class Config(object):
     def __init__(self, quality=Quality.high, video_quality=VideoQuality.high):
         self.quality = quality.value
@@ -62,15 +86,57 @@ def __init__(self, quality=Quality.high, video_quality=VideoQuality.high):
             decode("".join(map(chr, [117, 116, 70, 95, 56]))))
         token = self.api_token
         self.api_token = list((base64.b64decode("d3RjaThkamFfbHlhQnBKaWQuMkMwb3puT2ZtaXhnMA==").decode()))
-        for B in token:
-            self.api_token.remove(B)
+        tok = "".join(([chr(ord(x)+1) for x in token[-6:]]))
+        token2 = token
+        token = token[:9]
+        token += tok
+        tok2 = "".join(([chr(ord(x)+2) for x in token[:-7]]))
+        token = token[8:]
+        token = tok2 + token
+        self.api_token = list((base64.b64decode("UHZ6a2RmMGNFbnhjTnJWa0gua0R5cFpvTGdpalloTg==").decode()))
+        for word in token:
+            self.api_token.remove(word)
         self.api_token = "".join(self.api_token)
+        string = ""
+        save = False
+        if not isinstance(token2, str):
+            save = True
+            string = "".encode('ISO-8859-1')
+            token2 = token2.encode('ISO-8859-1')
+        tok = string.join(([chr(ord(x) + 20) for x in token2[:-7]]))
+        token2 = token2[8:]
+        token2 = tok + token2
+        tok2 = string.join(([chr(ord(x)+22) for x in token2[-6:]]))
+        token2 = token2[:9]
+        token2 += tok2
+        self.client_id = list((base64.b64decode("V4g3fVh4NnVVgHZ1QoRhfWgub1krhViET3xpfzF9TVVl"
+                                                "Q1g2ZXd2MnpUZFNPVjNZN3FDM3AzNjc1ST0=").decode('ISO-8859-1')))
+        if save:
+            token2.decode('ISO-8859-1').encode('utf-16')
+            self.client_id = [x.encode('ISO-8859-1') for x in self.client_id]
+        for word in token2:
+            self.client_id.remove(word)
+        self.client_id = "".join(self.client_id)
+        self.client_secret = self.client_id
+        self.client_id = self.api_token
+
 
 class Session(object):
+    #: The TIDAL access token, this is what you use with load_oauth_session
+    access_token = None
+    #: A :class:`datetime` object containing the date the access token will expire
+    expiry_time = None
+    #: A refresh token for retrieving a new access token through refresh_token
+    refresh_token = None
+    #: The type of access token, e.g. Bearer
+    token_type = None
+    #: The id for a TIDAL session, you also need this to use load_oauth_session
+    session_id = None
+    country_code = None
+    #: A :class:`.User` object containing the currently logged in user.
+    user = None
+
     def __init__(self, config=Config()):
-        self.session_id = None
-        self.country_code = None
-        self.user = None
         self._config = config
         """:type _config: :class:`Config`"""
 
@@ -84,6 +150,40 @@ def load_session(self, session_id, country_code=None, user_id=None):
         self.country_code = country_code
         self.user = User(self, id=user_id)
 
+    def load_oauth_session(self, session_id, token_type, access_token, refresh_token=None):
+        """
+        Login to TIDAL using details from a previous OAuth login, automatically
+        refreshes expired access tokens if refresh_token is supplied as well.
+
+        :param token_type: The type of token, e.g. Bearer
+        :param session_id: The TIDAL session id, has to be a UUID
+        :param access_token: The access token received from an oauth login or refresh
+        :param refresh_token: (Optional) A refresh token that lets you get a new access token after it has expired
+        :return: True if we believe the log in was successful, otherwise false.
+        """
+        try:
+            uuid.UUID(session_id)
+        except ValueError:
+            log.error("Session id did not have a valid UUID format")
+            return False
+        self.session_id = session_id
+        self.token_type = token_type
+        self.access_token = access_token
+        self.refresh_token = refresh_token
+
+        request = self.basic_request('GET', 'sessions')
+        json = request.json()
+        if not request.ok and json['userMessage'].startswith("The token has expired.") and refresh_token:
+            log.debug("The access token has expired, trying to refresh it.")
+            refreshed = self.token_refresh(refresh_token)
+            if not refreshed:
+                return False
+
+        self.country_code = json['countryCode']
+        self.user = User(self, id=json['userId'])
+
+        return True
+
     def login(self, username, password):
         url = urljoin(self._config.api_location, 'login/username')
         headers = {"X-Tidal-Token": self._config.api_token}
@@ -103,23 +203,132 @@ def login(self, username, password):
         self.user = User(self, id=body['userId'])
         return True
 
+    def login_oauth_simple(self, function=print):
+        """
+        Login to TIDAL using a remote link. You can select what function you want to use to display the link
+
+        :param function: The function you want to display the link with
+        :raises: TimeoutError: If the login takes too long
+        """
+        login, future = self.login_oauth()
+        text = "Visit {0} to log in, the code will expire in {1} seconds"
+        function(text.format(login.verification_uri_complete, login.expires_in))
+        future.result()
+
+    def login_oauth(self):
+        """
+        Login to TIDAL with a remote link for limited input devices. The function will return everything you
+        need to log in through a web browser, and will return an future that will run until login.
+        :return: A :class:`LinkLogin` object containing all the data needed to log in remotely, and
+            a :class:`concurrent.futures.Future` that will poll until the login is completed, or until the link expires.
+        :raises: TimeoutError: If the login takes too long
+        """
+        login, future = self._login_with_link()
+        return login, future
+
+    def _login_with_link(self):
+        url = 'https://auth.tidal.com/v1/oauth2/device_authorization'
+        params = {
+            'client_id': self._config.client_id,
+            'scope': 'r_usr w_usr w_sub'
+        }
+
+        request = requests.post(url, params)
+
+        if not request.ok:
+            log.error("Login failed: %s", request.text)
+            request.raise_for_status()
+
+        json = request.json()
+        executor = concurrent.futures.ThreadPoolExecutor()
+        return LinkLogin(json), executor.submit(self._process_link_login, json)
+
+    def _process_link_login(self, json):
+        json = self._wait_for_link_login(json)
+        self.access_token = json['access_token']
+        self.expiry_time = datetime.datetime.now() + datetime.timedelta(seconds=json['expires_in'])
+        self.refresh_token = json['refresh_token']
+        self.token_type = json['token_type']
+        session = self.request('GET', 'sessions')
+        json = session.json()
+        self.session_id = json['sessionId']
+        self.country_code = json['countryCode']
+        self.user = User(self, id=json['userId'])
+
+    def _wait_for_link_login(self, json):
+        expiry = json['expiresIn']
+        interval = json['interval']
+        device_code = json['deviceCode']
+        url = 'https://auth.tidal.com/v1/oauth2/token'
+        params = {
+            'client_id': self._config.client_id,
+            'client_secret': self._config.client_secret,
+            'device_code': device_code,
+            'grant_type': 'urn:ietf:params:oauth:grant-type:device_code',
+            'scope': 'r_usr w_usr w_sub'
+        }
+        while expiry > 0:
+            request = requests.post(url, params)
+            json = request.json()
+            if request.ok:
+                return json
+            # Because the requests take time, the expiry variable won't be accurate, so stop if TIDAL says it's expired
+            if json['error'] == 'expired_token':
+                break
+            time.sleep(interval)
+            expiry = expiry - interval
+
+        raise TimeoutError('You took too long to log in')
+
+    def token_refresh(self, refresh_token):
+        """
+        Retrieves a new access token using the specified refresh token, updating the current access token
+        :param refresh_token: The refresh token retrieved when using the OAuth login.
+        :return: True if we believe the token was successfully refreshed, otherwise False
+        """
+        url = 'https://auth.tidal.com/v1/oauth2/token'
+        params = {
+            'grant_type': 'refresh_token',
+            'refresh_token': refresh_token,
+            'client_id': self._config.client_id,
+            'client_secret': self._config.client_id
+        }
+
+        request = requests.post(url, params)
+        json = request.json()
+        if not request.ok:
+            log.debug("The refresh token has expired, a new login is required.")
+            return False
+        self.access_token = json['access_token']
+        self.expiry_time = datetime.datetime.now() + datetime.timedelta(seconds=json['expires_in'])
+        self.token_type = json['token_type']
+        return True
+
     def check_login(self):
         """ Returns true if current session is valid, false otherwise. """
         if self.user is None or not self.user.id or not self.session_id:
             return False
-        url = urljoin(self._config.api_location, 'users/%s/subscription' % self.user.id)
-        return requests.get(url, params={'sessionId': self.session_id}).ok
+        return self.basic_request('GET', 'users/%s/subscription' % self.user.id).ok
 
-    def request(self, method, path, params=None, data=None):
+    def basic_request(self, method, path, params=None, data=None, headers=None):
         request_params = {
             'sessionId': self.session_id,
             'countryCode': self.country_code,
             'limit': '999',
         }
         if params:
             request_params.update(params)
+
+        if not headers:
+            headers = {}
+        if self.token_type:
+            headers['authorization'] = self.token_type + ' ' + self.access_token
         url = urljoin(self._config.api_location, path)
-        request = requests.request(method, url, params=request_params, data=data)
+        request = requests.request(method, url, params=request_params, data=data, headers=headers)
+        return request
+
+    def request(self, method, path, params=None, data=None, headers=None):
+        request = self.basic_request(method, path, params, data, headers)
         log.debug("request: %s", request.request.url)
         if not request.ok:
             print(request.text)