restore cookie auth (#1201)

* fix cookie auth * add a test * fix RecursionError * add a log message so we know what's going on

Author: adehad

jira/client.py

@@ -210,59 +210,76 @@ class JiraCookieAuth(AuthBase):
  """Jira Cookie Authentication
 
  Allows using cookie authentication as described by
- https://developer.atlassian.com/jiradev/jira-apis/jira-rest-apis/jira-rest-api-tutorials/jira-rest-api-example-cookie-based-authentication
-
+ https://developer.atlassian.com/server/jira/platform/cookie-based-authentication/
  """
 
  def __init__(
- self, session: ResilientSession, _get_session: Callable, auth: Tuple[str, str]
+ self, session: ResilientSession, session_api_url: str, auth: Tuple[str, str]
  ):
  """Cookie Based Authentication
 
  Args:
  session (ResilientSession): The Session object to communicate with the API.
- _get_session (Callable): The function that returns a :py_class:``User``
- auth (Tuple[str, str]): The username, password tuple
+ session_api_url (str): The session api url to use.
+ auth (Tuple[str, str]): The username, password tuple.
  """
+
  self._session = session
- self._get_session = _get_session
+ self._session_api_url = session_api_url # e.g ."/rest/auth/1/session"
  self.__auth = auth
+ self._retry_counter_401 = 0
+ self._max_allowed_401_retries = 1 # 401 aren't recoverable with retries really
+
+ @property
+ def cookies(self):
+ return self._session.cookies
+
+ def _increment_401_retry_counter(self):
+ self._retry_counter_401 += 1
+
+ def _reset_401_retry_counter(self):
+ self._retry_counter_401 = 0
 
- def handle_401(self, response, **kwargs):
- if response.status_code != 401:
- return response
- self.init_session()
- response = self.process_original_request(response.request.copy())
+ def __call__(self, request: requests.PreparedRequest):
+ request.register_hook("response", self.handle_401)
+ return request
+
+ def init_session(self):
+ """Initialise the Session object's cookies, so we can use the session cookie."""
+ username, password = self.__auth
+ authentication_data = {"username": username, "password": password}
+ r = self._session.post( # this also goes through the handle_401() hook
+ self._session_api_url, data=json.dumps(authentication_data)
+ )
+ r.raise_for_status()
+
+ def handle_401(self, response: requests.Response, **kwargs):
+ """Refresh cookies if the session cookie has expired. Then retry the request."""
+ if (
+ response.status_code == 401
+ and self._retry_counter_401 < self._max_allowed_401_retries
+ ):
+ LOG.info("Trying to refresh the cookie auth session...")
+ self._increment_401_retry_counter()
+ self.init_session()
+ response = self.process_original_request(response.request.copy())
+ self._reset_401_retry_counter()
  return response
 
- def process_original_request(self, original_request):
+ def process_original_request(self, original_request: requests.PreparedRequest):
  self.update_cookies(original_request)
  return self.send_request(original_request)
 
- def update_cookies(self, original_request):
+ def update_cookies(self, original_request: requests.PreparedRequest):
  # Cookie header needs first to be deleted for the header to be updated using
  # the prepare_cookies method. See request.PrepareRequest.prepare_cookies
  if "Cookie" in original_request.headers:
  del original_request.headers["Cookie"]
  original_request.prepare_cookies(self.cookies)
 
- def init_session(self):
- self.start_session()
-
- def __call__(self, request):
- request.register_hook("response", self.handle_401)
- return request
-
- def send_request(self, request):
+ def send_request(self, request: requests.PreparedRequest):
  return self._session.send(request)
 
- @property
- def cookies(self):
- return self._session.cookies
-
- def start_session(self):
- self._get_session(self.__auth)
-
 
 class TokenAuth(AuthBase):
  """Bearer Token Authentication"""
@@ -571,8 +588,17 @@ def _create_cookie_auth(
  auth: Tuple[str, str],
  timeout: Optional[Union[Union[float, int], Tuple[float, float]]],
  ):
+ warnings.warn(
+ "Use OAuth or Token based authentication "
+ + "instead of Cookie based Authentication.",
+ DeprecationWarning,
+ )
  self._session = ResilientSession(timeout=timeout)
- self._session.auth = JiraCookieAuth(self._session, self.session, auth)
+ self._session.auth = JiraCookieAuth(
+ session=self._session,
+ session_api_url="{server}{auth_url}".format(**self._options),
+ auth=auth,
+ )
 
  def _check_update_(self):
  """Check if the current version of the library is outdated."""

tests/test_client.py

@@ -1,14 +1,12 @@
 import getpass
+from unittest import mock
 
 import pytest
 
 import jira.client
 from jira.exceptions import JIRAError
 from tests.conftest import JiraTestManager, get_unique_project_name
 
-# from tenacity import retry
-# from tenacity import wait_incrementing
-
 
 @pytest.fixture()
 def prep():
@@ -215,3 +213,39 @@ def test_token_auth(cl_admin: jira.client.JIRA):
  # THEN: The reported authenticated user of the token
  # matches the original token creator user.
  assert cl_admin.myself() == new_jira_client.myself()
+
+
+def test_cookie_auth(test_manager: JiraTestManager):
+ """Test Cookie based authentication works.
+
+ NOTE: this is deprecated in Cloud and is not recommended in Server.
+ https://developer.atlassian.com/cloud/jira/platform/jira-rest-api-cookie-based-authentication/
+ https://developer.atlassian.com/server/jira/platform/cookie-based-authentication/
+ """
+ # GIVEN: the username and password
+ # WHEN: We create a session with cookie auth for the same server
+ cookie_auth_jira = jira.client.JIRA(
+ server=test_manager.CI_JIRA_URL,
+ auth=(test_manager.CI_JIRA_ADMIN, test_manager.CI_JIRA_ADMIN_PASSWORD),
+ )
+ # THEN: We get the same result from the API
+ assert test_manager.jira_admin.myself() == cookie_auth_jira.myself()
+
+
+def test_cookie_auth_retry():
+ """Test Cookie based authentication retry logic works."""
+ # GIVEN: arguments that will cause a 401 error
+ auth_class = jira.client.JiraCookieAuth
+ reset_func = jira.client.JiraCookieAuth._reset_401_retry_counter
+ new_options = jira.client.JIRA.DEFAULT_OPTIONS.copy()
+ new_options["auth_url"] = "/401"
+ with pytest.raises(JIRAError):
+ with mock.patch.object(auth_class, reset_func.__name__) as mock_reset_func:
+ # WHEN: We create a session with cookie auth
+ jira.client.JIRA(
+ server="https://httpstat.us",
+ options=new_options,
+ auth=("user", "pass"),
+ )
+ # THEN: We don't get a RecursionError and only call the reset_function once
+ mock_reset_func.assert_called_once()