Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions ibmcloudant/couchdb_session_token_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,8 @@ def request_token(self):

return response

_DEFAULT_SESSION_COOKIE_TTL_SECONDS = 24 * 60 * 60

def _save_token_info(self, token_response) -> None:
"""
Decode the access token and save the response from the service to the object's state
Expand All @@ -100,6 +102,8 @@ def _save_token_info(self, token_response) -> None:
cookie = next(x for x in self.access_token if x.name == 'AuthSession')
exp = cookie.expires
iat = self._get_current_time()
if exp is None:
exp = iat + self._DEFAULT_SESSION_COOKIE_TTL_SECONDS
self.expire_time = exp
ttl = exp - iat
buffer = ttl * 0.2
Expand Down
43 changes: 40 additions & 3 deletions test/unit/test_couchdb_session_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

from ibmcloudant import CouchDbSessionAuthenticator
from ibmcloudant.cloudant_v1 import CloudantV1
from ibmcloudant.couchdb_session_token_manager import CouchDbSessionTokenManager

request = requests.request
request_args = None
Expand Down Expand Up @@ -55,9 +56,12 @@ def setUp(self) -> None:

def prepare_for_url(self, url):
def post_session(request):
return (200, {
"Set-Cookie": "AuthSession=" + self.cookie_value + "; Version=1; Expires=" +
self.cookie_expire_time + "; Max-Age=600; Path=/; HttpOnly"},
if self.cookie_expire_time is None:
Comment thread
ricellis marked this conversation as resolved.
set_cookie = "AuthSession=" + self.cookie_value + "; Version=1; Path=/; HttpOnly"
else:
set_cookie = ("AuthSession=" + self.cookie_value + "; Version=1; Expires=" +
self.cookie_expire_time + "; Max-Age=600; Path=/; HttpOnly")
return (200, {"Set-Cookie": set_cookie},
json.dumps({"ok": True, "name": "adm", "roles": ["_admin"]}))

responses.add_callback(responses.POST, url + '/_session', post_session)
Expand All @@ -70,6 +74,7 @@ def post_session(request):

@responses.activate
def test_header_passing(self):
self.cookie_expire_time = None
self.client.set_default_headers({"yes": "works"})
response = self.client.get_session_information(headers={"foo": "bar"})
self.assertIsNotNone(response)
Expand All @@ -86,6 +91,7 @@ def test_header_passing(self):

@responses.activate
def test_disable_ssl_verification_on(self):
self.cookie_expire_time = None
original_http_client = self.client.get_http_client()
mock_session = MockSession()
try:
Expand All @@ -98,6 +104,7 @@ def test_disable_ssl_verification_on(self):

@responses.activate
def test_disable_ssl_verification_off(self):
self.cookie_expire_time = None
original_http_client = self.client.get_http_client()
mock_session = MockSession()
try:
Expand Down Expand Up @@ -136,6 +143,7 @@ def test_invalid_disable_ssl_verification_type(self):

@responses.activate
def test_cookie_refresh(self):
self.cookie_expire_time = None
self.client.get_session_information()
self.assertEqual(responses.calls[-1].request.headers["Cookie"], "AuthSession=foobar")
self.cookie_value = "bar"
Expand All @@ -151,6 +159,7 @@ def test_cookie_refresh(self):

@responses.activate
def test_cookie_expired(self):
self.cookie_expire_time = None
self.client.get_session_information()
self.assertEqual(responses.calls[-1].request.headers["Cookie"], "AuthSession=foobar")
self.cookie_value = "bar"
Expand All @@ -161,6 +170,7 @@ def test_cookie_expired(self):

@responses.activate
def test_cookie_not_yet_expired(self):
self.cookie_expire_time = None
self.client.get_session_information()
self.assertEqual(responses.calls[-1].request.headers["Cookie"], "AuthSession=foobar")
self.cookie_value = "bar"
Expand All @@ -177,8 +187,35 @@ def test_refresh_time_calculation(self):
self.assertAlmostEqual(self.authenticator.token_manager.refresh_time,
self.authenticator.token_manager._get_current_time() + 8 * 60, delta=3)

@responses.activate
def test_refresh_time_calculation_no_expiry(self):
self.cookie_expire_time = None
self.client.get_session_information()
token_manager = self.authenticator.token_manager
default_ttl = CouchDbSessionTokenManager._DEFAULT_SESSION_COOKIE_TTL_SECONDS
self.assertAlmostEqual(token_manager.expire_time,
token_manager._get_current_time() + default_ttl, delta=3)
self.assertAlmostEqual(token_manager.refresh_time,
token_manager._get_current_time() + default_ttl * 0.8, delta=3)

@responses.activate
def test_cookie_without_expiry(self):
self.cookie_expire_time = None
self.client.get_session_information()
self.assertEqual(responses.calls[-1].request.headers["Cookie"], "AuthSession=foobar")
self.cookie_value = "bar"
self.authenticator.token_manager.refresh_time = 0
self.client.get_session_information()
self.assertEqual(responses.calls[-1].request.headers["Cookie"], "AuthSession=bar")
token_manager = self.authenticator.token_manager
self.assertIsNotNone(token_manager.expire_time)
self.assertIsNotNone(token_manager.refresh_time)
self.assertFalse(token_manager._is_token_expired())
self.assertGreater(token_manager.expire_time, token_manager._get_current_time())

@responses.activate
def test_set_service_url(self):
self.cookie_expire_time = None
self.client.get_session_information()
self.assertEqual(responses.calls[-2].request.url, "http://cloudant.example/_session")
self.assertEqual(responses.calls[-2].request.method, "POST")
Expand Down