Skip to content

Try to reauthenticate on a http status code 403 #59

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pnbecker wants to merge 4 commits into
base: main
Choose a base branch
from
+62 −98
Open

Try to reauthenticate on a http status code 403 #59

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
28d7e72
Use decorator to update csrf token instead of parameter retry
pnbecker Feb 3, 2026
427dd64
Try to reauthenticate on a http status code 403
pnbecker Feb 3, 2026
3b9176a
handle create_bitsream as special case and try reauthentication manually
pnbecker Feb 3, 2026
344bccf
Fix reauthenticate in create_bitstream()
pnbecker Feb 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 62 additions & 98 deletions dspace_rest_client/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,9 +156,44 @@ def do_paginate(url, params):
return fun(do_paginate, self, *args, **kwargs)

return decorated

return decorator

@staticmethod
def reauthenticate(func):
@functools.wraps(func)
def decorated(self, *args, reauthenticate=True, **kwargs):
r = func(self, *args, **kwargs)
if r is None:
return r
if r.status_code == 401 and reauthenticate:
if self.authenticate():
r = func(self, *args, **kwargs)
return r
return decorated

@staticmethod
def refresh_csrf(func):
@functools.wraps(func)
def decorated(self, *args, refresh_csrf=True, **kwargs):
r = func(self, *args, **kwargs)
if r is None:
return r
if r.status_code == 403 and refresh_csrf:
logging.debug("Retrying request with updated CSRF token")
r = func(self, *args, **kwargs)
r_json = None
try:
r_json = r.json()
except ValueError:
logging.warning("Tried to refresh CSRF token after getting a 403, got a non json-response.")
return r
if r is not None and "message" in r_json and "CSRF token" in r_json["message"]:
logging.warning(
"Too many retries updating token: %s: %s", r.status_code, r.text
)
return r
return decorated

def __init__(
self,
api_endpoint=API_ENDPOINT,
Expand Down Expand Up @@ -292,147 +327,82 @@ def api_get(self, url, params=None, data=None, headers=None):
self.update_token(r)
return r

def api_post(self, url, params, json, retry=False):
@reauthenticate
@refresh_csrf
def api_post(self, url, params, json):
"""
Perform a POST request. Refresh XSRF token if necessary.
POSTs are typically used to create objects.
@param url: DSpace REST API URL
@param params: Any parameters to include (eg ?parent=abbc-....)
@param json: Data in json-ready form (dict) to send as POST body (eg. item.as_dict())
@param retry: Has this method already been retried? Used if we need to refresh XSRF.
@return: Response from API
"""
r = self.session.post(
url, json=json, params=params, headers=self.request_headers,
proxies=self.proxies
)
self.update_token(r)

if r.status_code == 403:
# 403 Forbidden
# If we had a CSRF failure, retry the request with the updated token
# After speaking in #dev it seems that these do need occasional refreshes but I suspect
# it's happening too often for me, so check for accidentally triggering it
r_json = parse_json(r)
if "message" in r_json and "CSRF token" in r_json["message"]:
if retry:
logging.warning(
"Too many retries updating token: %s: %s", r.status_code, r.text
)
else:
logging.debug("Retrying request with updated CSRF token")
return self.api_post(url, params=params, json=json, retry=True)

return r

def api_post_uri(self, url, params, uri_list, retry=False):
@reauthenticate
@refresh_csrf
def api_post_uri(self, url, params, uri_list):
"""
Perform a POST request. Refresh XSRF token if necessary.
POSTs are typically used to create objects.
@param url: DSpace REST API URL
@param params: Any parameters to include (eg ?parent=abbc-....)
@param uri_list: One or more URIs referencing objects
@param retry: Has this method already been retried? Used if we need to refresh XSRF.
@return: Response from API
"""
r = self.session.post(
url, data=uri_list, params=params, headers=self.list_request_headers,
proxies=self.proxies
)
self.update_token(r)

if r.status_code == 403:
# 403 Forbidden
# If we had a CSRF failure, retry the request with the updated token
# After speaking in #dev it seems that these do need occasional refreshes but I suspect
# it's happening too often for me, so check for accidentally triggering it
r_json = r.json()
if "message" in r_json and "CSRF token" in r_json["message"]:
if retry:
logging.warning(
"Too many retries updating token: %s: %s", r.status_code, r.text
)
else:
logging.debug("Retrying request with updated CSRF token")
return self.api_post_uri(
url, params=params, uri_list=uri_list, retry=True
)

return r

def api_put(self, url, params, json, retry=False):
@reauthenticate
@refresh_csrf
def api_put(self, url, params, json):
"""
Perform a PUT request. Refresh XSRF token if necessary.
PUTs are typically used to update objects.
@param url: DSpace REST API URL
@param params: Any parameters to include (eg ?parent=abbc-....)
@param json: Data in json-ready form (dict) to send as PUT body (eg. item.as_dict())
@param retry: Has this method already been retried? Used if we need to refresh XSRF.
@return: Response from API
"""
r = self.session.put(
url, params=params, json=json, headers=self.request_headers,
proxies=self.proxies
)
self.update_token(r)

if r.status_code == 403:
# 403 Forbidden
# If we had a CSRF failure, retry the request with the updated token
# After speaking in #dev it seems that these do need occasional refreshes but I suspect
# it's happening too often for me, so check for accidentally triggering it
logging.debug(r.text)
# Parse response
r_json = parse_json(r)
if "message" in r_json and "CSRF token" in r_json["message"]:
if retry:
logging.warning(
"Too many retries updating token: %s: %s", r.status_code, r.text
)
else:
logging.debug("Retrying request with updated CSRF token")
return self.api_put(url, params=params, json=json, retry=True)

return r

def api_delete(self, url, params, retry=False):
@reauthenticate
@refresh_csrf
def api_delete(self, url, params):
"""
Perform a DELETE request. Refresh XSRF token if necessary.
DELETES are typically used to update objects.
@param url: DSpace REST API URL
@param params: Any parameters to include (eg ?parent=abbc-....)
@param retry: Has this method already been retried? Used if we need to refresh XSRF.
@return: Response from API
"""
r = self.session.delete(url, params=params, headers=self.request_headers)
self.update_token(r)

if r.status_code == 403:
# 403 Forbidden
# If we had a CSRF failure, retry the request with the updated token
# After speaking in #dev it seems that these do need occasional refreshes but I suspect
# it's happening too often for me, so check for accidentally triggering it
logging.debug(r.text)
# Parse response
r_json = parse_json(r)
if "message" in r_json and "CSRF token" in r_json["message"]:
if retry:
logging.warning(
"Too many retries updating token: %s: %s", r.status_code, r.text
)
else:
logging.debug("Retrying request with updated CSRF token")
return self.api_delete(url, params=params, retry=True)

return r

def api_patch(self, url, operation, path, value, params=None, retry=False):
@reauthenticate
@refresh_csrf
def api_patch(self, url, operation, path, value, params=None):
"""
@param url: DSpace REST API URL
@param operation: 'add', 'remove', 'replace', or 'move' (see PatchOperation enumeration)
@param path: path to perform operation - eg, metadata, withdrawn, etc.
@param value: new value for add or replace operations, or 'original' path for move operations
@param retry: Has this method already been retried? Used if we need to refresh XSRF.
@return:
@see https://github.com/DSpace/RestContract/blob/main/metadata-patch.md
"""
Expand Down Expand Up @@ -470,22 +440,7 @@ def api_patch(self, url, operation, path, value, params=None, retry=False):
)
self.update_token(r)

if r.status_code == 403:
# 403 Forbidden
# If we had a CSRF failure, retry the request with the updated token
# After speaking in #dev it seems that these do need occasional refreshes but I suspect
# it's happening too often for me, so check for accidentally triggering it
logging.debug(r.text)
r_json = parse_json(r)
if "message" in r_json and "CSRF token" in r_json["message"]:
if retry:
logging.warning(
"Too many retries updating token: %s: %s", r.status_code, r.text
)
else:
logging.debug("Retrying request with updated CSRF token")
return self.api_patch(url, operation, path, value, params, True)
elif r.status_code == 200:
if r.status_code == 200:
# 200 Success
logging.info(
"successful patch update to %s %s", r.json()["type"], r.json()["id"]
Expand Down Expand Up @@ -903,6 +858,7 @@ def create_bitstream(
metadata=None,
embeds=None,
retry=False,
reauthenticated=False,
):
"""
Upload a file and create a bitstream for a specified parent bundle, from the uploaded file and
Expand All @@ -921,6 +877,7 @@ def create_bitstream(
@param metadata: Full metadata JSON
@param retry: A 'retried' indicator. If the first attempt fails due to an expired or missing auth
token, the request will retry once, after the token is refreshed. (default: False)
@param reauthenticated An indicator, if we tried to reauthenticate in case of a http 403 status.
@return: constructed Bitstream object from the API response, or None if the operation failed.
"""
# TODO: It is probably wise to allow the bundle UUID to be simply passed as an alternative to having the full
Expand Down Expand Up @@ -956,6 +913,8 @@ def create_bitstream(
logging.debug("Updating token to %s", t)
self.session.headers.update({"X-XSRF-Token": t})
self.session.cookies.update({"X-XSRF-Token": t})
# as this method doesn't return the request, we cannot use our @refresh_csft decorator
# we should enhance self.api_post to be able to send files and use our decorators
if r.status_code == 403:
r_json = parse_json(r)
if "message" in r_json and "CSRF token" in r_json["message"]:
Expand All @@ -966,7 +925,12 @@ def create_bitstream(
return self.create_bitstream(
bundle, name, path, mime, metadata, embeds, True
)

# as this method doesn't return the request, we cannot use our @reauthenticate decorator
# we should enhance self.api_post to be able to send files and use our decorators
if r.status_code == 401 and not reauthenticated:
self.authenticate()
prepared_req = self.session.prepare_request(req)
r = self.session.send(prepared_req)
if r.status_code == 201 or r.status_code == 200:
# Success
return Bitstream(api_resource=parse_json(r))
Expand Down