import base64 import json import unittest from pyramid import testing from pyramid.compat import pickle class SharedCookieSessionTests(object): def test_ctor_no_cookie(self): request = testing.DummyRequest() session = self._makeOne(request) self.assertEqual(dict(session), {}) def test_instance_conforms(self): from zope.interface.verify import verifyObject from pyramid.interfaces import ISession request = testing.DummyRequest() session = self._makeOne(request) verifyObject(ISession, session) def test_ctor_with_cookie_still_valid(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time(), 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request) self.assertEqual(dict(session), {'state': 1}) def test_ctor_with_cookie_expired(self): request = testing.DummyRequest() cookieval = self._serialize((0, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request) self.assertEqual(dict(session), {}) def test_ctor_with_bad_cookie_cannot_deserialize(self): request = testing.DummyRequest() request.cookies['session'] = 'abc' session = self._makeOne(request) self.assertEqual(dict(session), {}) def test_ctor_with_bad_cookie_not_tuple(self): request = testing.DummyRequest() cookieval = self._serialize('abc') request.cookies['session'] = cookieval session = self._makeOne(request) self.assertEqual(dict(session), {}) def test_timeout(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time() - 5, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, timeout=1) self.assertEqual(dict(session), {}) def test_timeout_never(self): import time request = testing.DummyRequest() LONG_TIME = 31536000 cookieval = self._serialize((time.time() + LONG_TIME, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, timeout=None) self.assertEqual(dict(session), {'state': 1}) def test_timeout_str(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time() - 5, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, timeout='1') self.assertEqual(dict(session), {}) def test_timeout_invalid(self): request = testing.DummyRequest() self.assertRaises( ValueError, self._makeOne, request, timeout='Invalid value' ) def test_changed(self): request = testing.DummyRequest() session = self._makeOne(request) self.assertEqual(session.changed(), None) self.assertTrue(session._dirty) def test_invalidate(self): request = testing.DummyRequest() session = self._makeOne(request) session['a'] = 1 self.assertEqual(session.invalidate(), None) self.assertFalse('a' in session) def test_reissue_triggered(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time() - 2, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request) self.assertEqual(session['state'], 1) self.assertTrue(session._dirty) def test__set_cookie_on_exception(self): request = testing.DummyRequest() request.exception = True session = self._makeOne(request) session._cookie_on_exception = False response = DummyResponse() self.assertEqual(session._set_cookie(response), False) def test__set_cookie_on_exception_no_request_exception(self): import webob request = testing.DummyRequest() request.exception = None session = self._makeOne(request) session._cookie_on_exception = False response = webob.Response() self.assertEqual(session._set_cookie(response), True) self.assertEqual(response.headerlist[-1][0], 'Set-Cookie') def test__set_cookie_cookieval_too_long(self): request = testing.DummyRequest() session = self._makeOne(request) session['abc'] = 'x' * 100000 response = DummyResponse() self.assertRaises(ValueError, session._set_cookie, response) def test__set_cookie_real_webob_response(self): import webob request = testing.DummyRequest() session = self._makeOne(request) session['abc'] = 'x' response = webob.Response() self.assertEqual(session._set_cookie(response), True) self.assertEqual(response.headerlist[-1][0], 'Set-Cookie') def test__set_cookie_options(self): from pyramid.response import Response request = testing.DummyRequest() request.exception = None session = self._makeOne( request, cookie_name='abc', path='/foo', domain='localhost', secure=True, httponly=True, ) session['abc'] = 'x' response = Response() self.assertEqual(session._set_cookie(response), True) cookieval = response.headerlist[-1][1] val, domain, path, secure, httponly, samesite = [ x.strip() for x in cookieval.split(';') ] self.assertTrue(val.startswith('abc=')) self.assertEqual(domain, 'Domain=localhost') self.assertEqual(path, 'Path=/foo') self.assertEqual(secure, 'secure') self.assertEqual(httponly, 'HttpOnly') self.assertEqual(samesite, 'SameSite=Lax') def test_flash_default(self): request = testing.DummyRequest() session = self._makeOne(request) session.flash('msg1') session.flash('msg2') self.assertEqual(session['_f_'], ['msg1', 'msg2']) def test_flash_allow_duplicate_false(self): request = testing.DummyRequest() session = self._makeOne(request) session.flash('msg1') session.flash('msg1', allow_duplicate=False) self.assertEqual(session['_f_'], ['msg1']) def test_flash_allow_duplicate_true_and_msg_not_in_storage(self): request = testing.DummyRequest() session = self._makeOne(request) session.flash('msg1', allow_duplicate=True) self.assertEqual(session['_f_'], ['msg1']) def test_flash_allow_duplicate_false_and_msg_not_in_storage(self): request = testing.DummyRequest() session = self._makeOne(request) session.flash('msg1', allow_duplicate=False) self.assertEqual(session['_f_'], ['msg1']) def test_flash_mixed(self): request = testing.DummyRequest() session = self._makeOne(request) session.flash('warn1', 'warn') session.flash('warn2', 'warn') session.flash('err1', 'error') session.flash('err2', 'error') self.assertEqual(session['_f_warn'], ['warn1', 'warn2']) def test_pop_flash_default_queue(self): request = testing.DummyRequest() session = self._makeOne(request) queue = ['one', 'two'] session['_f_'] = queue result = session.pop_flash() self.assertEqual(result, queue) self.assertEqual(session.get('_f_'), None) def test_pop_flash_nodefault_queue(self): request = testing.DummyRequest() session = self._makeOne(request) queue = ['one', 'two'] session['_f_error'] = queue result = session.pop_flash('error') self.assertEqual(result, queue) self.assertEqual(session.get('_f_error'), None) def test_peek_flash_default_queue(self): request = testing.DummyRequest() session = self._makeOne(request) queue = ['one', 'two'] session['_f_'] = queue result = session.peek_flash() self.assertEqual(result, queue) self.assertEqual(session.get('_f_'), queue) def test_peek_flash_nodefault_queue(self): request = testing.DummyRequest() session = self._makeOne(request) queue = ['one', 'two'] session['_f_error'] = queue result = session.peek_flash('error') self.assertEqual(result, queue) self.assertEqual(session.get('_f_error'), queue) def test_new_csrf_token(self): request = testing.DummyRequest() session = self._makeOne(request) token = session.new_csrf_token() self.assertEqual(token, session['_csrft_']) def test_get_csrf_token(self): request = testing.DummyRequest() session = self._makeOne(request) session['_csrft_'] = 'token' token = session.get_csrf_token() self.assertEqual(token, 'token') self.assertTrue('_csrft_' in session) def test_get_csrf_token_new(self): request = testing.DummyRequest() session = self._makeOne(request) token = session.get_csrf_token() self.assertTrue(token) self.assertTrue('_csrft_' in session) def test_no_set_cookie_with_exception(self): import webob request = testing.DummyRequest() request.exception = True session = self._makeOne(request, set_on_exception=False) session['a'] = 1 callbacks = request.response_callbacks self.assertEqual(len(callbacks), 1) response = webob.Response() result = callbacks[0](request, response) self.assertEqual(result, None) self.assertFalse('Set-Cookie' in dict(response.headerlist)) def test_set_cookie_with_exception(self): import webob request = testing.DummyRequest() request.exception = True session = self._makeOne(request) session['a'] = 1 callbacks = request.response_callbacks self.assertEqual(len(callbacks), 1) response = webob.Response() result = callbacks[0](request, response) self.assertEqual(result, None) self.assertTrue('Set-Cookie' in dict(response.headerlist)) def test_cookie_is_set(self): import webob request = testing.DummyRequest() session = self._makeOne(request) session['a'] = 1 callbacks = request.response_callbacks self.assertEqual(len(callbacks), 1) response = webob.Response() result = callbacks[0](request, response) self.assertEqual(result, None) self.assertTrue('Set-Cookie' in dict(response.headerlist)) class TestBaseCookieSession(SharedCookieSessionTests, unittest.TestCase): def _makeOne(self, request, **kw): from pyramid.session import BaseCookieSessionFactory serializer = DummySerializer() return BaseCookieSessionFactory(serializer, **kw)(request) def _serialize(self, value): return base64.b64encode(json.dumps(value).encode('utf-8')) def test_reissue_not_triggered(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time(), 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, reissue_time=1) self.assertEqual(session['state'], 1) self.assertFalse(session._dirty) def test_reissue_never(self): request = testing.DummyRequest() cookieval = self._serialize((0, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, reissue_time=None, timeout=None) self.assertEqual(session['state'], 1) self.assertFalse(session._dirty) def test_reissue_str_triggered(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time() - 2, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, reissue_time='0') self.assertEqual(session['state'], 1) self.assertTrue(session._dirty) def test_reissue_invalid(self): request = testing.DummyRequest() self.assertRaises( ValueError, self._makeOne, request, reissue_time='invalid value' ) def test_cookie_max_age_invalid(self): request = testing.DummyRequest() self.assertRaises( ValueError, self._makeOne, request, max_age='invalid value' ) class TestSignedCookieSession(SharedCookieSessionTests, unittest.TestCase): def _makeOne(self, request, **kw): from pyramid.session import SignedCookieSessionFactory kw.setdefault('secret', 'secret') return SignedCookieSessionFactory(**kw)(request) def _serialize(self, value, salt=b'pyramid.session.', hashalg='sha512'): import base64 import hashlib import hmac import pickle digestmod = lambda: hashlib.new(hashalg) cstruct = pickle.dumps(value, pickle.HIGHEST_PROTOCOL) sig = hmac.new(salt + b'secret', cstruct, digestmod).digest() return base64.urlsafe_b64encode(sig + cstruct).rstrip(b'=') def test_reissue_not_triggered(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time(), 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, reissue_time=1) self.assertEqual(session['state'], 1) self.assertFalse(session._dirty) def test_reissue_never(self): request = testing.DummyRequest() cookieval = self._serialize((0, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, reissue_time=None, timeout=None) self.assertEqual(session['state'], 1) self.assertFalse(session._dirty) def test_reissue_str_triggered(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time() - 2, 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, reissue_time='0') self.assertEqual(session['state'], 1) self.assertTrue(session._dirty) def test_reissue_invalid(self): request = testing.DummyRequest() self.assertRaises( ValueError, self._makeOne, request, reissue_time='invalid value' ) def test_cookie_max_age_invalid(self): request = testing.DummyRequest() self.assertRaises( ValueError, self._makeOne, request, max_age='invalid value' ) def test_custom_salt(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time(), 0, {'state': 1}), salt=b'f.') request.cookies['session'] = cookieval session = self._makeOne(request, salt=b'f.') self.assertEqual(session['state'], 1) def test_salt_mismatch(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time(), 0, {'state': 1}), salt=b'f.') request.cookies['session'] = cookieval session = self._makeOne(request, salt=b'g.') self.assertEqual(session, {}) def test_custom_hashalg(self): import time request = testing.DummyRequest() cookieval = self._serialize( (time.time(), 0, {'state': 1}), hashalg='sha1' ) request.cookies['session'] = cookieval session = self._makeOne(request, hashalg='sha1') self.assertEqual(session['state'], 1) def test_hashalg_mismatch(self): import time request = testing.DummyRequest() cookieval = self._serialize( (time.time(), 0, {'state': 1}), hashalg='sha1' ) request.cookies['session'] = cookieval session = self._makeOne(request, hashalg='sha256') self.assertEqual(session, {}) def test_secret_mismatch(self): import time request = testing.DummyRequest() cookieval = self._serialize((time.time(), 0, {'state': 1})) request.cookies['session'] = cookieval session = self._makeOne(request, secret='evilsecret') self.assertEqual(session, {}) def test_custom_serializer(self): import base64 from hashlib import sha512 import hmac import time request = testing.DummyRequest() serializer = DummySerializer() cstruct = serializer.dumps((time.time(), 0, {'state': 1})) sig = hmac.new(b'pyramid.session.secret', cstruct, sha512).digest() cookieval = base64.urlsafe_b64encode(sig + cstruct).rstrip(b'=') request.cookies['session'] = cookieval session = self._makeOne(request, serializer=serializer) self.assertEqual(session['state'], 1) def test_invalid_data_size(self): from hashlib import sha512 import base64 request = testing.DummyRequest() num_bytes = sha512().digest_size - 1 cookieval = base64.b64encode(b' ' * num_bytes) request.cookies['session'] = cookieval session = self._makeOne(request) self.assertEqual(session, {}) def test_very_long_key(self): verylongkey = b'a' * 1024 import webob request = testing.DummyRequest() session = self._makeOne(request, secret=verylongkey) session['a'] = 1 callbacks = request.response_callbacks self.assertEqual(len(callbacks), 1) response = webob.Response() try: result = callbacks[0](request, response) except TypeError: # pragma: no cover self.fail('HMAC failed to initialize due to key length.') self.assertEqual(result, None) self.assertTrue('Set-Cookie' in dict(response.headerlist)) def test_bad_pickle(self): import base64 import hashlib import hmac digestmod = lambda: hashlib.new('sha512') # generated from dumping an object that cannot be found anymore, eg: # class Foo: pass # print(pickle.dumps(Foo())) cstruct = b'(i__main__\nFoo\np0\n(dp1\nb.' sig = hmac.new(b'pyramid.session.secret', cstruct, digestmod).digest() cookieval = base64.urlsafe_b64encode(sig + cstruct).rstrip(b'=') request = testing.DummyRequest() request.cookies['session'] = cookieval session = self._makeOne(request, secret='secret') self.assertEqual(session, {}) class TestUnencryptedCookieSession( SharedCookieSessionTests, unittest.TestCase ): def setUp(self): super(TestUnencryptedCookieSession, self).setUp() from zope.deprecation import __show__ __show__.off() def tearDown(self): super(TestUnencryptedCookieSession, self).tearDown() from zope.deprecation import __show__ __show__.on() def _makeOne(self, request, **kw): from pyramid.session import UnencryptedCookieSessionFactoryConfig self._rename_cookie_var(kw, 'path', 'cookie_path') self._rename_cookie_var(kw, 'domain', 'cookie_domain') self._rename_cookie_var(kw, 'secure', 'cookie_secure') self._rename_cookie_var(kw, 'httponly', 'cookie_httponly') self._rename_cookie_var(kw, 'set_on_exception', 'cookie_on_exception') return UnencryptedCookieSessionFactoryConfig('secret', **kw)(request) def _rename_cookie_var(self, kw, src, dest): if src in kw: kw.setdefault(dest, kw.pop(src)) def _serialize(self, value): from pyramid.compat import bytes_ from pyramid.session import signed_serialize return bytes_(signed_serialize(value, 'secret')) def test_serialize_option(self): from pyramid.response import Response secret = 'secret' request = testing.DummyRequest() session = self._makeOne( request, signed_serialize=dummy_signed_serialize ) session['key'] = 'value' response = Response() self.assertEqual(session._set_cookie(response), True) cookie = response.headerlist[-1][1] expected_cookieval = dummy_signed_serialize( (session.accessed, session.created, {'key': 'value'}), secret ) response = Response() response.set_cookie('session', expected_cookieval, samesite='Lax') expected_cookie = response.headerlist[-1][1] self.assertEqual(cookie, expected_cookie) def test_deserialize_option(self): import time secret = 'secret' request = testing.DummyRequest() accessed = time.time() state = {'key': 'value'} cookieval = dummy_signed_serialize((accessed, accessed, state), secret) request.cookies['session'] = cookieval session = self._makeOne( request, signed_deserialize=dummy_signed_deserialize ) self.assertEqual(dict(session), state) def dummy_signed_serialize(data, secret): import base64 from pyramid.compat import pickle, bytes_ pickled = pickle.dumps(data) return base64.b64encode(bytes_(secret)) + base64.b64encode(pickled) def dummy_signed_deserialize(serialized, secret): import base64 from pyramid.compat import pickle, bytes_ serialized_data = base64.b64decode( serialized[len(base64.b64encode(bytes_(secret))) :] ) return pickle.loads(serialized_data) class Test_manage_accessed(unittest.TestCase): def _makeOne(self, wrapped): from pyramid.session import manage_accessed return manage_accessed(wrapped) def test_accessed_set(self): request = testing.DummyRequest() session = DummySessionFactory(request) session.renewed = 0 wrapper = self._makeOne(session.__class__.get) wrapper(session, 'a') self.assertNotEqual(session.accessed, None) self.assertTrue(session._dirty) def test_accessed_without_renew(self): import time request = testing.DummyRequest() session = DummySessionFactory(request) session._reissue_time = 5 session.renewed = time.time() wrapper = self._makeOne(session.__class__.get) wrapper(session, 'a') self.assertNotEqual(session.accessed, None) self.assertFalse(session._dirty) def test_already_dirty(self): request = testing.DummyRequest() session = DummySessionFactory(request) session.renewed = 0 session._dirty = True session['a'] = 1 wrapper = self._makeOne(session.__class__.get) self.assertEqual(wrapper.__doc__, session.get.__doc__) result = wrapper(session, 'a') self.assertEqual(result, 1) callbacks = request.response_callbacks if callbacks is not None: self.assertEqual(len(callbacks), 0) class Test_manage_changed(unittest.TestCase): def _makeOne(self, wrapped): from pyramid.session import manage_changed return manage_changed(wrapped) def test_it(self): request = testing.DummyRequest() session = DummySessionFactory(request) wrapper = self._makeOne(session.__class__.__setitem__) wrapper(session, 'a', 1) self.assertNotEqual(session.accessed, None) self.assertTrue(session._dirty) def serialize(data, secret): import hmac import base64 from hashlib import sha1 from pyramid.compat import bytes_ from pyramid.compat import native_ from pyramid.compat import pickle pickled = pickle.dumps(data, pickle.HIGHEST_PROTOCOL) sig = hmac.new(bytes_(secret, 'utf-8'), pickled, sha1).hexdigest() return sig + native_(base64.b64encode(pickled)) class Test_signed_serialize(unittest.TestCase): def _callFUT(self, data, secret): from pyramid.session import signed_serialize return signed_serialize(data, secret) def test_it(self): expected = serialize('123', 'secret') result = self._callFUT('123', 'secret') self.assertEqual(result, expected) def test_it_with_highorder_secret(self): secret = b'\xce\xb1\xce\xb2\xce\xb3\xce\xb4'.decode('utf-8') expected = serialize('123', secret) result = self._callFUT('123', secret) self.assertEqual(result, expected) def test_it_with_latin1_secret(self): secret = b'La Pe\xc3\xb1a' expected = serialize('123', secret) result = self._callFUT('123', secret.decode('latin-1')) self.assertEqual(result, expected) class Test_signed_deserialize(unittest.TestCase): def _callFUT(self, serialized, secret, hmac=None): if hmac is None: import hmac from pyramid.session import signed_deserialize return signed_deserialize(serialized, secret, hmac=hmac) def test_it(self): serialized = serialize('123', 'secret') result = self._callFUT(serialized, 'secret') self.assertEqual(result, '123') def test_invalid_bits(self): serialized = serialize('123', 'secret') self.assertRaises(ValueError, self._callFUT, serialized, 'seekrit') def test_invalid_len(self): class hmac(object): def new(self, *arg): return self def hexdigest(self): return '1234' serialized = serialize('123', 'secret123') self.assertRaises( ValueError, self._callFUT, serialized, 'secret', hmac=hmac() ) def test_it_bad_encoding(self): serialized = 'bad' + serialize('123', 'secret') self.assertRaises(ValueError, self._callFUT, serialized, 'secret') def test_it_with_highorder_secret(self): secret = b'\xce\xb1\xce\xb2\xce\xb3\xce\xb4'.decode('utf-8') serialized = serialize('123', secret) result = self._callFUT(serialized, secret) self.assertEqual(result, '123') # bwcompat with pyramid <= 1.5b1 where latin1 is the default def test_it_with_latin1_secret(self): secret = b'La Pe\xc3\xb1a' serialized = serialize('123', secret) result = self._callFUT(serialized, secret.decode('latin-1')) self.assertEqual(result, '123') class TestPickleSerializer(unittest.TestCase): def _makeOne(self): from pyramid.session import PickleSerializer return PickleSerializer() def test_loads(self): # generated from dumping Dummy() using protocol=2 cstruct = b'\x80\x02ctests.test_session\nDummy\nq\x00)\x81q\x01.' serializer = self._makeOne() result = serializer.loads(cstruct) self.assertIsInstance(result, Dummy) def test_loads_raises_ValueError_on_invalid_data(self): cstruct = b'not pickled' serializer = self._makeOne() self.assertRaises(ValueError, serializer.loads, cstruct) def test_loads_raises_ValueError_on_bad_import(self): # generated from dumping an object that cannot be found anymore, eg: # class Foo: pass # print(pickle.dumps(Foo())) cstruct = b'(i__main__\nFoo\np0\n(dp1\nb.' serializer = self._makeOne() self.assertRaises(ValueError, serializer.loads, cstruct) def test_dumps(self): obj = Dummy() serializer = self._makeOne() result = serializer.dumps(obj) expected_result = pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL) self.assertEqual(result, expected_result) self.assertIsInstance(result, bytes) class Dummy(object): pass class DummySerializer(object): def dumps(self, value): return base64.b64encode(json.dumps(value).encode('utf-8')) def loads(self, value): try: return json.loads(base64.b64decode(value).decode('utf-8')) # base64.b64decode raises a TypeError on py2 instead of a ValueError # and a ValueError is required for the session to handle it properly except TypeError: raise ValueError class DummySessionFactory(dict): _dirty = False _cookie_name = 'session' _cookie_max_age = None _cookie_path = '/' _cookie_domain = None _cookie_secure = False _cookie_httponly = False _timeout = 1200 _reissue_time = 0 def __init__(self, request): self.request = request dict.__init__(self, {}) def changed(self): self._dirty = True class DummyResponse(object): def __init__(self): self.headerlist = []