Patch Session Fixation (#273)

Author: panagiks

aiohttp_session/__init__.py

@@ -18,13 +18,14 @@ class Session(MutableMapping):
     def __init__(self, identity, *, data, new, max_age=None):
         self._changed = False
         self._mapping = {}
-        self._identity = identity
+        self._identity = identity if data != {} else None
         self._new = new
+        self._new = new if data != {} else True
         self._max_age = max_age
         created = data.get('created', None) if data else None
         session_data = data.get('session', None) if data else None
 
-        if new or created is None:
+        if self._new or created is None:
             self._created = int(time.time())
         else:
             self._created = created

tests/test_encrypted_cookie_storage.py

@@ -136,3 +136,27 @@ async def handler(request):
     assert '' == morsel.value
     assert not morsel['httponly']
     assert morsel['path'] == '/'
+
+
+async def test_encrypted_cookie_session_fixation(aiohttp_client, fernet, key):
+    async def login(request):
+        session = await get_session(request)
+        session['k'] = 'v'
+        return web.Response()
+
+    async def logout(request):
+        session = await get_session(request)
+        session.invalidate()
+        return web.Response()
+
+    app = create_app(login, key)
+    app.router.add_route('DELETE', '/', logout)
+    client = await aiohttp_client(app)
+    resp = await client.get('/')
+    assert 'AIOHTTP_SESSION' in resp.cookies
+    evil_cookie = resp.cookies['AIOHTTP_SESSION'].value
+    resp = await client.delete('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value == ""
+    client.session.cookie_jar.update_cookies({'AIOHTTP_SESSION': evil_cookie})
+    resp = await client.get('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value != evil_cookie

tests/test_memcached_storage.py

@@ -203,3 +203,27 @@ def key_factory():
     value = await load_cookie(client, memcached)
     assert 'key' in value['session']
     assert value['session']['key'] == 'value'
+
+
+async def test_memcached_session_fixation(aiohttp_client, memcached):
+    async def login(request):
+        session = await get_session(request)
+        session['k'] = 'v'
+        return web.Response()
+
+    async def logout(request):
+        session = await get_session(request)
+        session.invalidate()
+        return web.Response()
+
+    app = create_app(login, memcached)
+    app.router.add_route('DELETE', '/', logout)
+    client = await aiohttp_client(app)
+    resp = await client.get('/')
+    assert 'AIOHTTP_SESSION' in resp.cookies
+    evil_cookie = resp.cookies['AIOHTTP_SESSION'].value
+    resp = await client.delete('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value == ""
+    client.session.cookie_jar.update_cookies({'AIOHTTP_SESSION': evil_cookie})
+    resp = await client.get('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value != evil_cookie

tests/test_nacl_storage.py

@@ -129,3 +129,27 @@ async def handler(request):
     assert '' == morsel.value
     assert not morsel['httponly']
     assert morsel['path'] == '/'
+
+
+async def test_nacl_session_fixation(aiohttp_client, secretbox, key):
+    async def login(request):
+        session = await get_session(request)
+        session['k'] = 'v'
+        return web.Response()
+
+    async def logout(request):
+        session = await get_session(request)
+        session.invalidate()
+        return web.Response()
+
+    app = create_app(login, key)
+    app.router.add_route('DELETE', '/', logout)
+    client = await aiohttp_client(app)
+    resp = await client.get('/')
+    assert 'AIOHTTP_SESSION' in resp.cookies
+    evil_cookie = resp.cookies['AIOHTTP_SESSION'].value
+    resp = await client.delete('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value == ""
+    client.session.cookie_jar.update_cookies({'AIOHTTP_SESSION': evil_cookie})
+    resp = await client.get('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value != evil_cookie

tests/test_redis_storage.py

@@ -245,6 +245,30 @@ def key_factory():
     assert value['session']['key'] == 'value'
 
 
+async def test_redis_session_fixation(aiohttp_client, redis):
+    async def login(request):
+        session = await get_session(request)
+        session['k'] = 'v'
+        return web.Response()
+
+    async def logout(request):
+        session = await get_session(request)
+        session.invalidate()
+        return web.Response()
+
+    app = create_app(login, redis)
+    app.router.add_route('DELETE', '/', logout)
+    client = await aiohttp_client(app)
+    resp = await client.get('/')
+    assert 'AIOHTTP_SESSION' in resp.cookies
+    evil_cookie = resp.cookies['AIOHTTP_SESSION'].value
+    resp = await client.delete('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value == ""
+    client.session.cookie_jar.update_cookies({'AIOHTTP_SESSION': evil_cookie})
+    resp = await client.get('/')
+    assert resp.cookies['AIOHTTP_SESSION'].value != evil_cookie
+
+
 async def test_redis_from_create_pool(redis_params):
 
     async def handler(request):