From 62e49c1030ef4f68ddc4d5d71543a4f9e6fb631c Mon Sep 17 00:00:00 2001 From: Brent Bumann Date: Wed, 19 Jan 2022 10:38:07 -0800 Subject: [PATCH] Ensure Dropbox error is thrown in refresh access token (#407) Ensure Dropbox error is thrown in refresh access token --- dropbox/dropbox_client.py | 87 ++++++++++++++++++-------------- test/integration/test_dropbox.py | 2 +- 2 files changed, 51 insertions(+), 38 deletions(-) diff --git a/dropbox/dropbox_client.py b/dropbox/dropbox_client.py index afe664a6..b1fdec89 100644 --- a/dropbox/dropbox_client.py +++ b/dropbox/dropbox_client.py @@ -376,7 +376,6 @@ def refresh_access_token(self, host=API_HOST, scope=None): :param scope: list of permission scopes for access token :return: """ - if scope is not None and (len(scope) == 0 or not isinstance(scope, list)): raise BadInputException("Scope list must be of type list") @@ -401,12 +400,7 @@ def refresh_access_token(self, host=API_HOST, scope=None): if self._timeout: timeout = self._timeout res = self._session.post(url, data=body, timeout=timeout) - if res.status_code == 400 and res.json()['error'] == 'invalid_grant': - request_id = res.headers.get('x-dropbox-request-id') - err = stone_serializers.json_compat_obj_decode( - AuthError_validator, 'invalid_access_token') - raise AuthError(request_id, err) - res.raise_for_status() + self.raise_dropbox_error_for_resp(res) token_content = res.json() self._oauth2_access_token = token_content["access_token"] @@ -596,53 +590,72 @@ def request_json_string(self, verify=True, timeout=timeout, ) - + self.raise_dropbox_error_for_resp(r) request_id = r.headers.get('x-dropbox-request-id') - if r.status_code >= 500: - raise InternalServerError(request_id, r.status_code, r.text) - elif r.status_code == 400: - raise BadInputError(request_id, r.text) - elif r.status_code == 401: + if r.status_code in (403, 404, 409): + raw_resp = r.content.decode('utf-8') + return RouteErrorResult(request_id, raw_resp) + + if route_style == self._ROUTE_STYLE_DOWNLOAD: + raw_resp = r.headers['dropbox-api-result'] + else: assert r.headers.get('content-type') == 'application/json', ( 'Expected content-type to be application/json, got %r' % r.headers.get('content-type')) + raw_resp = r.content.decode('utf-8') + if route_style == self._ROUTE_STYLE_DOWNLOAD: + return RouteResult(raw_resp, r) + else: + return RouteResult(raw_resp) + + def raise_dropbox_error_for_resp(self, res): + """Checks for errors from a res and handles appropiately. + + :param res: Response of an api request. + """ + request_id = res.headers.get('x-dropbox-request-id') + if res.status_code >= 500: + raise InternalServerError(request_id, res.status_code, res.text) + elif res.status_code == 400: + try: + if res.json()['error'] == 'invalid_grant': + request_id = res.headers.get('x-dropbox-request-id') + err = stone_serializers.json_compat_obj_decode( + AuthError_validator, 'invalid_access_token') + raise AuthError(request_id, err) + else: + raise BadInputError(request_id, res.text) + except ValueError: + raise BadInputError(request_id, res.text) + elif res.status_code == 401: + assert res.headers.get('content-type') == 'application/json', ( + 'Expected content-type to be application/json, got %r' % + res.headers.get('content-type')) err = stone_serializers.json_compat_obj_decode( - AuthError_validator, r.json()['error']) + AuthError_validator, res.json()['error']) raise AuthError(request_id, err) - elif r.status_code == HTTP_STATUS_INVALID_PATH_ROOT: + elif res.status_code == HTTP_STATUS_INVALID_PATH_ROOT: err = stone_serializers.json_compat_obj_decode( - PathRootError_validator, r.json()['error']) + PathRootError_validator, res.json()['error']) raise PathRootError(request_id, err) - elif r.status_code == 429: + elif res.status_code == 429: err = None - if r.headers.get('content-type') == 'application/json': + if res.headers.get('content-type') == 'application/json': err = stone_serializers.json_compat_obj_decode( - RateLimitError_validator, r.json()['error']) + RateLimitError_validator, res.json()['error']) retry_after = err.retry_after else: - retry_after_str = r.headers.get('retry-after') + retry_after_str = res.headers.get('retry-after') if retry_after_str is not None: retry_after = int(retry_after_str) else: retry_after = None raise RateLimitError(request_id, err, retry_after) - elif 200 <= r.status_code <= 299: - if route_style == self._ROUTE_STYLE_DOWNLOAD: - raw_resp = r.headers['dropbox-api-result'] - else: - assert r.headers.get('content-type') == 'application/json', ( - 'Expected content-type to be application/json, got %r' % - r.headers.get('content-type')) - raw_resp = r.content.decode('utf-8') - if route_style == self._ROUTE_STYLE_DOWNLOAD: - return RouteResult(raw_resp, r) - else: - return RouteResult(raw_resp) - elif r.status_code in (403, 404, 409): - raw_resp = r.content.decode('utf-8') - return RouteErrorResult(request_id, raw_resp) - else: - raise HttpError(request_id, r.status_code, r.text) + elif res.status_code in (403, 404, 409): + # special case handled by requester + return + elif not (200 <= res.status_code <= 299): + raise HttpError(request_id, res.status_code, res.text) def _get_route_url(self, hostname, route_name): """Returns the URL of the route. diff --git a/test/integration/test_dropbox.py b/test/integration/test_dropbox.py index a7824e6e..5aeb3232 100644 --- a/test/integration/test_dropbox.py +++ b/test/integration/test_dropbox.py @@ -153,7 +153,7 @@ def test_default_oauth2_urls(self): def test_bad_auth(self): # Test malformed token malformed_token_dbx = Dropbox(MALFORMED_TOKEN) - with pytest.raises(BadInputError) as cm: + with pytest.raises(BadInputError,) as cm: malformed_token_dbx.files_list_folder('') assert 'token is malformed' in cm.value.message