From 70f6b45eb85435f5a5ee7a7f0a06bdca3616e056 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Tue, 11 Aug 2026 17:52:57 -0400 Subject: [PATCH] Keep the refresh token current when Intuit rotates it (#397) Intuit returns a refresh token with every token response and replaces the value roughly 24 hours after it was issued. From that moment the previous value is rejected with "Incorrect or invalid refresh token", and the replacement appears in that one response and nowhere else. The client treated the refresh token as static configuration. _start_session() returned auth_client.refresh_token and __new__ wrote it over the caller's value -- which is None in the documented flow, where an access token is supplied and no token call is made. So the credential the application had just stored was discarded, the rotated token never reached the application, and a 401 raised AuthorizationException instead of refreshing. Applications worked around that by refreshing by hand with their stored token, replaying a superseded value and ending the connection. - track refresh_token/access_token as client state that only moves forward - add refresh_token_callback, called after every token response so the caller can persist the value the moment it changes - add public refresh_access_token(), which always presents the newest token - refresh and retry once on a 401 in make_request()/download_pdf(); auto_refresh=False keeps the previous behaviour - raise a readable QuickbooksException when no refresh token is available Co-Authored-By: Claude Opus 5 (1M context) --- quickbooks/client.py | 137 +++++++++- tests/unit/test_token_refresh.py | 361 ++++++++++++++++++++++++++ tests/unit/test_token_rotation_e2e.py | 267 +++++++++++++++++++ 3 files changed, 754 insertions(+), 11 deletions(-) create mode 100644 tests/unit/test_token_refresh.py create mode 100644 tests/unit/test_token_rotation_e2e.py diff --git a/quickbooks/client.py b/quickbooks/client.py index a428044..c356664 100644 --- a/quickbooks/client.py +++ b/quickbooks/client.py @@ -30,6 +30,20 @@ class QuickBooks(object): invoice_link = False use_decimal = False + # The refresh token currently issued for this connection. Intuit rotates + # this value roughly every 24 hours and the previous value stops working + # the moment it does, so it is kept up to date after every token response + # rather than treated as static configuration. + refresh_token = None + access_token = None + + # Called with the token payload after every token response so the caller + # can persist the current refresh token. See README, "Storing tokens". + refresh_token_callback = None + + # Refresh the access token and retry once when the API answers 401. + auto_refresh = True + sandbox_api_url_v3 = "https://sandbox-quickbooks.api.intuit.com/v3" api_url_v3 = "https://quickbooks.api.intuit.com/v3" current_user_url = "https://appcenter.intuit.com/api/v1/user/current" @@ -62,6 +76,12 @@ def __new__(cls, **kwargs): if 'refresh_token' in kwargs: instance.refresh_token = kwargs['refresh_token'] + if 'refresh_token_callback' in kwargs: + instance.refresh_token_callback = kwargs['refresh_token_callback'] + + if 'auto_refresh' in kwargs: + instance.auto_refresh = kwargs['auto_refresh'] + if 'auth_client' in kwargs: instance.auth_client = kwargs['auth_client'] @@ -70,8 +90,7 @@ def __new__(cls, **kwargs): else: instance.sandbox = False - refresh_token = instance._start_session() - instance.refresh_token = refresh_token + instance._start_session() if 'company_id' in kwargs: instance.company_id = kwargs['company_id'] @@ -101,18 +120,101 @@ def __new__(cls, **kwargs): return instance def _start_session(self): + """Prepare the HTTP session, getting an access token if there isn't one. + + Returns the refresh token this client will use from here on: the most + recent one the token service issued, or the caller's if no token call + was needed. Never None, and never an older value than either. + """ if self.auth_client.access_token is None: - self.auth_client.refresh(refresh_token=self.refresh_token) + self.refresh_access_token() + else: + # An access token was supplied, so no token call is needed. Read + # whatever the auth client holds without discarding what we were + # given: it may have no refresh token at all. + self._store_tokens(notify=False) + + self.session = OAuth2Session(self.auth_client.client_id, token=self._session_token()) + + return self.refresh_token - self.session = OAuth2Session( - self.auth_client.client_id, - token={ - 'access_token': self.auth_client.access_token, - 'refresh_token': self.auth_client.refresh_token, - } - ) + def refresh_access_token(self): + """Get a new access token, keeping the refresh token current. - return self.auth_client.refresh_token + Intuit hands back a refresh token with every token response. Most of the + time it is the value that was sent; roughly 24 hours after a value is + issued the service rotates it, and from that moment the previous value + is rejected with "Incorrect or invalid refresh token". The replacement + appears in that one response and nowhere else, so it is captured here + and passed to ``refresh_token_callback`` for the caller to persist. + """ + presented = self.refresh_token or getattr(self.auth_client, 'refresh_token', None) + + if not presented: + raise exceptions.QuickbooksException( + 'No refresh token available. Pass refresh_token to QuickBooks() ' + '(or to AuthClient()) with the value stored for this company.', + 10000) + + self.auth_client.refresh(refresh_token=presented) + self._store_tokens(presented=presented) + + if self.session is not None: + self.session.token = self._session_token() + + return self.refresh_token + + def _store_tokens(self, presented=None, notify=True): + """Take the tokens off the auth client and tell the caller about them.""" + issued = getattr(self.auth_client, 'refresh_token', None) + + # Only ever move forward: a token response always carries a refresh + # token, but an auth client that has not made one yet carries None. + if issued: + self.refresh_token = issued + + if self.auth_client.access_token: + self.access_token = self.auth_client.access_token + + if notify and self.refresh_token_callback is not None: + self.refresh_token_callback({ + 'refresh_token': self.refresh_token, + 'access_token': self.access_token, + 'expires_in': getattr(self.auth_client, 'expires_in', None), + 'x_refresh_token_expires_in': getattr( + self.auth_client, 'x_refresh_token_expires_in', None), + 'realm_id': getattr(self.auth_client, 'realm_id', None), + 'rotated': bool(presented and issued and issued != presented), + }) + + def _session_token(self): + return { + 'access_token': self.access_token, + 'refresh_token': self.refresh_token, + } + + def _refresh_and_retry(self, request_type, url, headers, params, data): + """Answer to a 401: get a new access token and send the request again. + + Returns None when refreshing is not possible or did not help. + """ + if self.auth_client is None or not self.auto_refresh: + return None + + try: + self.refresh_access_token() + except exceptions.QuickbooksException: + return None + except Exception as error: + # The refresh token is gone as far as the service is concerned -- + # the connection has to be authorized again. + raise exceptions.AuthorizationException( + 'Failed to refresh the access token. The connection must be ' + 'authorized again.', error_code=httplib.UNAUTHORIZED, + detail=str(error)) + + return self.process_request( + request_type, url, headers=headers, params=params, data=data) def _drop(self): QuickBooks.__instance = None @@ -219,6 +321,14 @@ def make_request(self, request_type, url, request_body=None, content_type='appli req = self.process_request(request_type, url, headers=headers, params=params, data=request_body) + if req.status_code == httplib.UNAUTHORIZED: + # Access tokens last an hour. Get a fresh one and send the request + # once more before giving up on the connection. + retried = self._refresh_and_retry( + request_type, url, headers, params, request_body) + if retried is not None: + req = retried + if req.status_code == httplib.UNAUTHORIZED: raise exceptions.AuthorizationException( "Application authentication failed", error_code=req.status_code, detail=req.text) @@ -363,6 +473,11 @@ def download_pdf(self, qbbo, item_id): response = self.process_request("GET", url, headers=headers) + if response.status_code == httplib.UNAUTHORIZED: + retried = self._refresh_and_retry("GET", url, headers, "", "") + if retried is not None: + response = retried + if response.status_code != httplib.OK: if response.status_code == httplib.UNAUTHORIZED: diff --git a/tests/unit/test_token_refresh.py b/tests/unit/test_token_refresh.py new file mode 100644 index 0000000..4680799 --- /dev/null +++ b/tests/unit/test_token_refresh.py @@ -0,0 +1,361 @@ +"""Regression coverage for refresh-token handling (issue #397). + +Intuit rotates the refresh token roughly every 24 hours. The rotated value +arrives in a single token response and the previous value is rejected from that +moment on, so the client has to keep its own copy current and hand the new one +to the caller for storage. +""" + +import http.client as httplib +from unittest import TestCase +from unittest.mock import patch + +from quickbooks import client +from quickbooks.exceptions import AuthorizationException, QuickbooksException + + +class FakeAuthClient(object): + """Stands in for ``intuitlib.client.AuthClient``. + + ``responses`` scripts what the token service returns, in order. The default + is the vendor's usual answer: the same refresh token that was presented. + """ + + def __init__(self, access_token=None, refresh_token=None, responses=None, + environment='sandbox', error=None, service=None): + self.client_id = 'CLIENT_ID' + self.environment = environment + self.access_token = access_token + self.refresh_token = refresh_token + self.expires_in = None + self.x_refresh_token_expires_in = None + self.realm_id = 'REALM' + self.responses = list(responses or []) + self.error = error + self.service = service + self.presented = [] + + def refresh(self, refresh_token=None): + token = refresh_token or self.refresh_token + if token is None: + raise ValueError('Refresh token not specified') + + self.presented.append(token) + + if self.error is not None: + raise self.error + + if self.service is not None: + response = self.service.refresh(token) + elif self.responses: + response = self.responses.pop(0) + else: + response = {'access_token': 'ACCESS_{0}'.format(len(self.presented)), + 'refresh_token': token} + + self.access_token = response['access_token'] + self.refresh_token = response['refresh_token'] + self.expires_in = 3600 + self.x_refresh_token_expires_in = 8726400 + + +class InvalidGrant(Exception): + """What intuitlib raises when the token service rejects a refresh token.""" + + +class FakeTokenService(object): + """Intuit's token service, as measured against the vendor. + + A refresh token is stable for 24 hours after it is issued: every refresh + returns the same value. The first refresh after that replaces it, and the + previous value is rejected from that moment on -- there is no grace period, + and the replacement is only in that one response. + """ + + ROTATION_INTERVAL = 24 * 60 * 60 + ACCESS_TOKEN_LIFETIME = 60 * 60 + + def __init__(self, refresh_token='RT0'): + self.clock = 0 + self.refresh_token = refresh_token + self.issued_at = 0 + self.superseded = set() + self.rotations = 0 + self.access_tokens = {} + + def issue_access_token(self): + access_token = 'AT{0}'.format(len(self.access_tokens)) + self.access_tokens[access_token] = self.clock + self.ACCESS_TOKEN_LIFETIME + return access_token + + def access_token_is_valid(self, access_token): + return self.access_tokens.get(access_token, 0) > self.clock + + def refresh(self, presented): + if presented != self.refresh_token: + raise InvalidGrant( + 'HTTP status 400, error message: {"error":"invalid_grant",' + '"error_description":"Incorrect or invalid refresh token"}') + + if self.clock - self.issued_at >= self.ROTATION_INTERVAL: + self.superseded.add(self.refresh_token) + self.rotations += 1 + self.refresh_token = 'RT{0}'.format(self.rotations) + self.issued_at = self.clock + + return {'access_token': self.issue_access_token(), + 'refresh_token': self.refresh_token} + + +class MockResponse(object): + def __init__(self, status_code=httplib.OK, text='{"QueryResponse": {}}'): + self.status_code = status_code + self.text = text + self.content = b'pdf' + + +class RefreshTokenTestCase(TestCase): + def build(self, auth_client, **kwargs): + kwargs.setdefault('company_id', 'COMPANY_ID') + kwargs.setdefault('minorversion', 75) + return client.QuickBooks(auth_client=auth_client, **kwargs) + + # -- the caller's refresh token --------------------------------------- + + def test_supplied_refresh_token_survives_a_supplied_access_token(self): + # The README's example passes an access token, so no token call is + # made. The client used to report the auth client's empty refresh + # token, destroying the credential the caller had just stored. + auth_client = FakeAuthClient(access_token='ACCESS_TOKEN') + + qb_client = self.build(auth_client, refresh_token='REFRESH_TOKEN') + + self.assertEqual(qb_client.refresh_token, 'REFRESH_TOKEN') + self.assertEqual(auth_client.presented, []) + + def test_refresh_token_is_read_from_the_auth_client_when_not_supplied(self): + auth_client = FakeAuthClient(access_token='ACCESS_TOKEN', + refresh_token='REFRESH_TOKEN') + + qb_client = self.build(auth_client) + + self.assertEqual(qb_client.refresh_token, 'REFRESH_TOKEN') + + def test_missing_refresh_token_raises_a_readable_error(self): + auth_client = FakeAuthClient() + + with self.assertRaises(QuickbooksException) as caught: + self.build(auth_client) + + self.assertIn('No refresh token available', str(caught.exception.message)) + + # -- rotation ---------------------------------------------------------- + + def test_rotated_refresh_token_replaces_the_one_that_was_presented(self): + auth_client = FakeAuthClient(responses=[ + {'access_token': 'ACCESS_1', 'refresh_token': 'ROTATED'}]) + + qb_client = self.build(auth_client, refresh_token='ORIGINAL') + + self.assertEqual(auth_client.presented, ['ORIGINAL']) + self.assertEqual(qb_client.refresh_token, 'ROTATED') + self.assertEqual(qb_client.access_token, 'ACCESS_1') + + def test_superseded_refresh_token_is_never_presented_again(self): + auth_client = FakeAuthClient(responses=[ + {'access_token': 'ACCESS_1', 'refresh_token': 'ROTATED'}, + {'access_token': 'ACCESS_2', 'refresh_token': 'ROTATED'}]) + + qb_client = self.build(auth_client, refresh_token='ORIGINAL') + qb_client.refresh_access_token() + + self.assertEqual(auth_client.presented, ['ORIGINAL', 'ROTATED']) + + def test_callback_receives_every_token_response(self): + stored = [] + auth_client = FakeAuthClient(responses=[ + {'access_token': 'ACCESS_1', 'refresh_token': 'ORIGINAL'}, + {'access_token': 'ACCESS_2', 'refresh_token': 'ROTATED'}]) + + qb_client = self.build(auth_client, refresh_token='ORIGINAL', + refresh_token_callback=stored.append) + qb_client.refresh_access_token() + + self.assertEqual([token['refresh_token'] for token in stored], + ['ORIGINAL', 'ROTATED']) + self.assertEqual([token['rotated'] for token in stored], [False, True]) + self.assertEqual(stored[-1]['access_token'], 'ACCESS_2') + self.assertEqual(stored[-1]['x_refresh_token_expires_in'], 8726400) + self.assertEqual(stored[-1]['realm_id'], 'REALM') + + def test_callback_is_not_called_when_no_token_call_is_made(self): + stored = [] + auth_client = FakeAuthClient(access_token='ACCESS_TOKEN') + + self.build(auth_client, refresh_token='REFRESH_TOKEN', + refresh_token_callback=stored.append) + + self.assertEqual(stored, []) + + def test_refreshing_updates_the_session_access_token(self): + auth_client = FakeAuthClient(responses=[ + {'access_token': 'ACCESS_1', 'refresh_token': 'ORIGINAL'}, + {'access_token': 'ACCESS_2', 'refresh_token': 'ROTATED'}]) + + qb_client = self.build(auth_client, refresh_token='ORIGINAL') + self.assertEqual(qb_client.session.access_token, 'ACCESS_1') + + qb_client.refresh_access_token() + + self.assertEqual(qb_client.session.access_token, 'ACCESS_2') + self.assertEqual(qb_client.session.token['refresh_token'], 'ROTATED') + + # -- expired access tokens -------------------------------------------- + + @patch('quickbooks.client.QuickBooks.process_request') + def test_expired_access_token_is_refreshed_and_the_request_retried(self, process_request): + process_request.side_effect = [ + MockResponse(status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED'), + MockResponse(), + ] + auth_client = FakeAuthClient(access_token='EXPIRED') + qb_client = self.build(auth_client, refresh_token='REFRESH_TOKEN') + + result = qb_client.get('https://example.com/v3/company/1/companyinfo/1') + + self.assertEqual(result, {'QueryResponse': {}}) + self.assertEqual(auth_client.presented, ['REFRESH_TOKEN']) + self.assertEqual(process_request.call_count, 2) + + @patch('quickbooks.client.QuickBooks.process_request') + def test_a_second_401_still_raises(self, process_request): + process_request.side_effect = [ + MockResponse(status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED'), + MockResponse(status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED'), + ] + auth_client = FakeAuthClient(access_token='EXPIRED') + qb_client = self.build(auth_client, refresh_token='REFRESH_TOKEN') + + self.assertRaises(AuthorizationException, qb_client.get, 'https://example.com/') + + @patch('quickbooks.client.QuickBooks.process_request') + def test_auto_refresh_can_be_turned_off(self, process_request): + process_request.return_value = MockResponse( + status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED') + auth_client = FakeAuthClient(access_token='EXPIRED') + qb_client = self.build(auth_client, refresh_token='REFRESH_TOKEN', + auto_refresh=False) + + self.assertRaises(AuthorizationException, qb_client.get, 'https://example.com/') + self.assertEqual(auth_client.presented, []) + self.assertEqual(process_request.call_count, 1) + + @patch('quickbooks.client.QuickBooks.process_request') + def test_a_rejected_refresh_token_reports_that_reauthorization_is_needed(self, process_request): + process_request.return_value = MockResponse( + status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED') + auth_client = FakeAuthClient( + access_token='EXPIRED', + error=Exception('HTTP status 400, error message: ' + '{"error":"invalid_grant","error_description":' + '"Incorrect or invalid refresh token"}')) + qb_client = self.build(auth_client, refresh_token='SUPERSEDED') + + with self.assertRaises(AuthorizationException) as caught: + qb_client.get('https://example.com/') + + self.assertIn('authorized again', str(caught.exception.message)) + self.assertIn('invalid_grant', str(caught.exception.detail)) + + @patch('quickbooks.client.QuickBooks.process_request') + def test_download_pdf_refreshes_on_401(self, process_request): + process_request.side_effect = [ + MockResponse(status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED'), + MockResponse(), + ] + auth_client = FakeAuthClient(access_token='EXPIRED') + qb_client = self.build(auth_client, refresh_token='REFRESH_TOKEN') + + self.assertEqual(qb_client.download_pdf('SalesReceipt', 1), b'pdf') + self.assertEqual(auth_client.presented, ['REFRESH_TOKEN']) + + +class Issue397CycleTestCase(TestCase): + """The sequence from issue #397, start to finish. + + An application connects, then works through the day rebuilding its client + whenever the hour-old access token has expired, storing the refresh token + the client reports. Somewhere past the 24 hour mark the service replaces + the value; if the application never hears about it, the next refresh is + rejected and the connection is over until someone authorizes it again. + """ + + HOUR = 60 * 60 + URL = 'https://sandbox-quickbooks.api.intuit.com/v3/company/1/companyinfo/1' + + def setUp(self): + self.service = FakeTokenService() + self.original_refresh_token = self.service.refresh_token + self.stored = {'refresh_token': self.service.refresh_token, + 'access_token': self.service.issue_access_token()} + + # The API answers 401 once the access token the session carries has + # expired, which it has by the time each hourly request comes around. + patcher = patch.object(client.QuickBooks, 'process_request', autospec=True) + self.addCleanup(patcher.stop) + patcher.start().side_effect = self.api_call + + def api_call(self, qb_client, request_type, url, headers="", params="", data=""): + if self.service.access_token_is_valid(qb_client.session.access_token): + return MockResponse() + return MockResponse(status_code=httplib.UNAUTHORIZED, text='UNAUTHORIZED') + + def save(self, token): + self.stored['refresh_token'] = token['refresh_token'] + self.stored['access_token'] = token['access_token'] + + def app_request(self): + """One request from an application written the way the README shows.""" + auth_client = FakeAuthClient(access_token=self.stored['access_token'], + service=self.service) + qb_client = client.QuickBooks( + auth_client=auth_client, + refresh_token=self.stored['refresh_token'], + company_id='COMPANY_ID', + minorversion=75, + refresh_token_callback=self.save, + ) + + try: + qb_client.get(self.URL) + except AuthorizationException: + # The workaround applications reach for, and the call in #397's + # traceback: refresh by hand with the token that was stored. + auth_client.refresh(refresh_token=self.stored['refresh_token']) + + # Store what the client reports, which is all an application can see. + if qb_client.refresh_token: + self.stored['refresh_token'] = qb_client.refresh_token + self.stored['access_token'] = auth_client.access_token + + def test_the_connection_survives_the_rotation(self): + for hour in [0, 1, 2, 3, 12, 24, 25, 26]: + self.service.clock = hour * self.HOUR + self.app_request() + + self.assertEqual(self.service.rotations, 1) + self.assertNotEqual(self.stored['refresh_token'], self.original_refresh_token) + self.assertEqual(self.stored['refresh_token'], self.service.refresh_token, + 'the application is holding a token the service replaced') + self.assertNotIn(self.stored['refresh_token'], self.service.superseded) + + def test_the_replaced_token_is_rejected_from_then_on(self): + """The premise of the bug: rotation is immediate and final.""" + self.service.clock = 25 * self.HOUR + self.app_request() + + self.assertIn(self.original_refresh_token, self.service.superseded) + with self.assertRaises(InvalidGrant) as caught: + self.service.refresh(self.original_refresh_token) + + self.assertIn('Incorrect or invalid refresh token', str(caught.exception)) diff --git a/tests/unit/test_token_rotation_e2e.py b/tests/unit/test_token_rotation_e2e.py new file mode 100644 index 0000000..3655052 --- /dev/null +++ b/tests/unit/test_token_rotation_e2e.py @@ -0,0 +1,267 @@ +"""End-to-end coverage for issue #397, over HTTP. + +`test_token_refresh.py` pins the client's logic with a fake auth client. This +covers the part that cannot reach: the real `intuitlib` `AuthClient` and a real +`OAuth2Session`, driven through a connect flow and a day of API calls past both +boundaries that end connections in production -- the access token expiring after +an hour and the refresh token being replaced after a day. + +The token service below answers the way Intuit's does: a refresh token is stable +for 24 hours after it is issued, the first refresh after that replaces it, and +the previous value is rejected from that moment on. It listens on localhost, so +no credentials and no outside network are involved. +""" + +import base64 +import http.client as httplib +import json +import os +import secrets +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest import TestCase +from urllib.parse import parse_qs, urlencode, urlparse + +# The token service speaks plain HTTP. +os.environ.setdefault('OAUTHLIB_INSECURE_TRANSPORT', '1') + +import urllib.error # noqa: E402 +import urllib.request # noqa: E402 + +from intuitlib.client import AuthClient # noqa: E402 +from intuitlib.exceptions import AuthClientError # noqa: E402 + +from quickbooks import QuickBooks # noqa: E402 + +CLIENT_ID = 'CLIENT_ID' +CLIENT_SECRET = 'CLIENT_SECRET' +REALM_ID = '9130344291180042' +REDIRECT_URI = 'http://localhost:8484/callback' + +HOUR = 60 * 60 +ACCESS_TOKEN_LIFETIME = HOUR +ROTATION_INTERVAL = 24 * HOUR + + +class IntuitTokenService(object): + """Intuit's token service, plus the sliver of the Accounting API a client + needs in order to notice that its access token went stale.""" + + def __init__(self): + self.clock_offset = 0 + self.refresh_token = None + self.refresh_issued_at = 0 + self.access_tokens = {} + self.codes = {} + self.rotations = 0 + self.lock = threading.Lock() + + service = self + + class Handler(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + + def log_message(self, *args): # keep the test output readable + pass + + def do_GET(self): + service.handle(self) + + do_POST = do_GET + + self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + self.base_url = 'http://{0}:{1}'.format(*self.server.server_address[:2]) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + + # -- test controls ----------------------------------------------------- + + @property + def discovery_url(self): + return self.base_url + '/.well-known/openid_configuration' + + @property + def api_base_url(self): + return self.base_url + '/v3' + + def now(self): + return int(time.time()) + self.clock_offset + + def stop(self): + self.server.shutdown() + self.server.server_close() + + # -- routing ----------------------------------------------------------- + + def handle(self, handler): + parsed = urlparse(handler.path) + length = int(handler.headers.get('Content-Length') or 0) + body = handler.rfile.read(length).decode('utf-8') if length else '' + + if parsed.path == '/.well-known/openid_configuration': + return self.send(handler, 200, { + 'issuer': self.base_url, + 'authorization_endpoint': self.base_url + '/oauth2/authorize', + 'token_endpoint': self.base_url + '/oauth2/v1/tokens/bearer', + 'revocation_endpoint': self.base_url + '/oauth2/v1/tokens/revoke', + 'userinfo_endpoint': self.base_url + '/openid_connect/userinfo', + 'jwks_uri': self.base_url + '/v1/openid_connect/jwks', + }) + if parsed.path == '/oauth2/authorize': + return self.authorize(handler, parse_qs(parsed.query)) + if parsed.path == '/oauth2/v1/tokens/bearer': + return self.token(handler, parse_qs(body)) + return self.api(handler) + + def authorize(self, handler, query): + code = secrets.token_hex(16) + self.codes[code] = True + handler.send_response(302) + handler.send_header('Location', '{0}?{1}'.format( + query['redirect_uri'][0], urlencode({ + 'code': code, + 'state': query.get('state', [''])[0], + 'realmId': REALM_ID, + }))) + handler.send_header('Content-Length', '0') + handler.end_headers() + + def token(self, handler, form): + credentials = base64.b64decode( + handler.headers.get('Authorization', 'Basic ')[6:] or b'').decode('utf-8') + if credentials != '{0}:{1}'.format(CLIENT_ID, CLIENT_SECRET): + return self.send(handler, 401, {'error': 'invalid_client'}) + + with self.lock: + if form.get('grant_type', [''])[0] == 'authorization_code': + if not self.codes.pop(form.get('code', [''])[0], None): + return self.send(handler, 400, {'error': 'invalid_grant'}) + self.refresh_token = secrets.token_hex(32) + self.refresh_issued_at = self.now() + return self.send(handler, 200, self.issue_access_token()) + + presented = form.get('refresh_token', [''])[0] + if presented != self.refresh_token: + # Replaced, revoked, or never issued -- all answer alike. + return self.send(handler, 400, { + 'error': 'invalid_grant', + 'error_description': 'Incorrect or invalid refresh token'}) + + # Rotation is anchored to when the value was issued, not to when it + # was last used: it survives any number of refreshes inside the + # window and is replaced on the first one after it. + if self.now() - self.refresh_issued_at >= ROTATION_INTERVAL: + self.refresh_token = secrets.token_hex(32) + self.refresh_issued_at = self.now() + self.rotations += 1 + + return self.send(handler, 200, self.issue_access_token()) + + def issue_access_token(self): + access_token = secrets.token_hex(32) + self.access_tokens[access_token] = self.now() + ACCESS_TOKEN_LIFETIME + return { + 'access_token': access_token, + 'token_type': 'bearer', + 'expires_in': ACCESS_TOKEN_LIFETIME, + 'refresh_token': self.refresh_token, + 'x_refresh_token_expires_in': 100 * 24 * HOUR, + } + + def api(self, handler): + presented = handler.headers.get('Authorization', '')[7:] + if self.access_tokens.get(presented, 0) > self.now(): + return self.send(handler, 200, { + 'CompanyInfo': {'Id': '1', 'CompanyName': 'Lakeshore Outfitters'}}) + return self.send(handler, 401, {'fault': { + 'error': [{'message': 'message=AuthenticationFailed', 'code': '3200'}], + 'type': 'AUTHENTICATION'}}) + + def send(self, handler, status, payload): + body = json.dumps(payload).encode('utf-8') + handler.send_response(status) + handler.send_header('Content-Type', 'application/json') + handler.send_header('Content-Length', str(len(body))) + handler.end_headers() + handler.wfile.write(body) + + +class TokenRotationTestCase(TestCase): + def setUp(self): + self.service = IntuitTokenService() + self.addCleanup(self.service.stop) + + auth_client = self.connect() + self.original_refresh_token = auth_client.refresh_token + self.stored = {'refresh_token': auth_client.refresh_token, + 'access_token': auth_client.access_token} + + def auth_client(self, **kwargs): + return AuthClient(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, + environment=self.service.discovery_url, + redirect_uri=REDIRECT_URI, **kwargs) + + def connect(self): + """Run the connect flow the way an application's callback view does.""" + auth_client = self.auth_client() + url = auth_client.get_authorization_url([]) + + class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *args): + return None + + try: + location = urllib.request.build_opener(NoRedirect).open(url).headers['Location'] + except urllib.error.HTTPError as error: + location = error.headers['Location'] + + query = dict(part.split('=', 1) for part in location.split('?', 1)[1].split('&')) + auth_client.get_bearer_token(query['code'], realm_id=REALM_ID) + return auth_client + + def app_request(self): + """One request from an application written the way the README shows.""" + auth_client = self.auth_client(access_token=self.stored['access_token']) + qb_client = QuickBooks( + auth_client=auth_client, + refresh_token=self.stored['refresh_token'], + company_id=REALM_ID, + minorversion=75, + refresh_token_callback=self.save, + ) + qb_client.api_url_v3 = qb_client.sandbox_api_url_v3 = self.service.api_base_url + + result = qb_client.get('{0}/company/{1}/companyinfo/1'.format( + self.service.api_base_url, REALM_ID)) + + # Store what the client reports, which is all an application can see. + if qb_client.refresh_token: + self.stored['refresh_token'] = qb_client.refresh_token + self.stored['access_token'] = auth_client.access_token + return result + + def save(self, token): + self.stored['refresh_token'] = token['refresh_token'] + self.stored['access_token'] = token['access_token'] + + def test_the_connection_survives_the_day(self): + for hour in [0, 1, 2, 3, 12, 24, 25, 26]: + self.service.clock_offset = hour * HOUR + self.assertIn('CompanyInfo', self.app_request(), + 'the connection died at t+{0}h'.format(hour)) + + self.assertEqual(self.service.rotations, 1) + self.assertNotEqual(self.stored['refresh_token'], self.original_refresh_token) + self.assertEqual(self.stored['refresh_token'], self.service.refresh_token, + 'the application is holding a token the service replaced') + + def test_a_replaced_refresh_token_is_rejected(self): + """The premise of the bug, through intuitlib: replacement is final.""" + self.service.clock_offset = 25 * HOUR + self.app_request() + + with self.assertRaises(AuthClientError) as caught: + self.auth_client().refresh(refresh_token=self.original_refresh_token) + + self.assertEqual(caught.exception.status_code, httplib.BAD_REQUEST) + self.assertIn('Incorrect or invalid refresh token', str(caught.exception))