[IMP] session_redis: add absolute expiration option

ODOO_SESSION_REDIS_EXPIRATION_ABSOLUTE=1 makes authenticated sessions expire
ODOO_SESSION_REDIS_EXPIRATION seconds after the login, whatever the activity.

Default mode: SET … EX instead of SET then EXPIRE, so a failed EXPIRE can no
longer leave a key without TTL. rotate() also resets should_rotate, as
odoo.http.FilesystemSessionStore does.
This commit is contained in:
Vincent Van Rossem committed 2026-09-25 11:03:00 +02:00
1 parent 5108bfe728
commit ff8cd092bd
7 files changed
+186 -5

No files matched your search

+1 -1
View File
@@ -3,4 +3,4 @@ azure-identity
azure-storage-blob
prometheus_client
python-json-logger
redis
redis>=3.5
+4
View File
@@ -20,6 +20,10 @@ The storage of sessions in Redis is activated using environment variables.
the sessions (default is 7 days)
* ``ODOO_SESSION_REDIS_EXPIRATION_ANONYMOUS`` is the time in seconds before expiration of
the anonymous sessions (default is 3 hours)
* ``ODOO_SESSION_REDIS_EXPIRATION_ABSOLUTE``, when ``1`` or ``true``, the
authenticated sessions expire ``ODOO_SESSION_REDIS_EXPIRATION`` seconds
after the login, whatever the activity, instead of after the last request
(requires Redis >= 6.0 and redis-py >= 3.5)
The keys are set to ``session:<session id>``.
+2 -2
View File
@@ -5,7 +5,7 @@
{
"name": "Sessions in Redis",
"summary": "Store web sessions in Redis",
"version": "17.0.1.0.0",
"version": "17.0.1.1.0",
"author": "Camptocamp,Odoo Community Association (OCA)",
"license": "AGPL-3",
"category": "Extra Tools",
@@ -15,7 +15,7 @@
"auth_session_timeout",
],
"external_dependencies": {
"python": ["redis"],
"python": ["redis>=3.5"],
},
"website": "https://github.com/camptocamp/odoo-cloud-platform",
"data": [],
+2
View File
@@ -40,6 +40,7 @@ url = os.environ.get("ODOO_SESSION_REDIS_URL")
password = os.environ.get("ODOO_SESSION_REDIS_PASSWORD")
expiration = os.environ.get("ODOO_SESSION_REDIS_EXPIRATION")
anon_expiration = os.environ.get("ODOO_SESSION_REDIS_EXPIRATION_ANONYMOUS")
absolute_expiration = is_true(os.environ.get("ODOO_SESSION_REDIS_EXPIRATION_ABSOLUTE"))
@lazy_property
@@ -56,6 +57,7 @@ def session_store(self):
prefix=prefix,
expiration=expiration,
anon_expiration=anon_expiration,
absolute_expiration=absolute_expiration,
session_class=http.Session,
)
+19 -2
View File
@@ -27,6 +27,7 @@ class RedisSessionStore(SessionStore):
prefix="",
expiration=None,
anon_expiration=None,
absolute_expiration=False,
):
super().__init__(session_class=session_class)
self.redis = redis
@@ -38,6 +39,8 @@ class RedisSessionStore(SessionStore):
self.anon_expiration = DEFAULT_SESSION_TIMEOUT_ANONYMOUS
else:
self.anon_expiration = anon_expiration
# authenticated sessions expire after `expiration` since their login
self.absolute_expiration = absolute_expiration
self.prefix = "session:"
if prefix:
self.prefix = f"{self.prefix}:{prefix}:"
@@ -69,8 +72,19 @@ class RedisSessionStore(SessionStore):
data = json.dumps(dict(session), cls=json_encoding.SessionEncoder).encode(
"utf-8"
)
if self.redis.set(key, data):
return self.redis.expire(key, expiration)
if self.absolute_expiration and session.uid and not session.is_new:
# absolute mode: keep the TTL set at login (keepttl) and never
# recreate an expired session (xx); anonymous and new sessions
# get a full TTL below
if not self.redis.set(key, data, xx=True, keepttl=True):
return False
# keepttl keeps "no TTL": give one to a key left without
if self.redis.ttl(key) == -1:
self.redis.expire(key, expiration)
return True
# write the data and the TTL in one command: with SET then EXPIRE, a
# failed EXPIRE left a key without TTL, never expiring
return self.redis.set(key, data, ex=expiration)
def delete(self, session):
key = self.build_key(session.sid)
@@ -113,8 +127,11 @@ class RedisSessionStore(SessionStore):
def rotate(self, session, env):
self.delete(session)
session.sid = self.generate_key()
# new key: save it with a full expiration, in absolute mode too
session.is_new = True
if session.uid and env:
session.session_token = security.compute_session_token(session, env)
session.should_rotate = False
self.save(session)
def vacuum(self, *args, **kwargs):
+1
View File
@@ -0,0 +1 @@
from . import test_session_store
+157
View File
@@ -0,0 +1,157 @@
# Copyright 2026 Camptocamp SA
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
from odoo import http
from odoo.tests import BaseCase
from ..session import RedisSessionStore
DAY = 60 * 60 * 24
HOUR = 60 * 60
class FakeRedis:
"""In-memory Redis with a manual clock, for the commands used by the store"""
def __init__(self):
self.now = 0
self.data = {}
self.expire_at = {}
def advance(self, seconds):
self.now += seconds
def _purge(self, key):
expire_at = self.expire_at.get(key)
if expire_at is not None and expire_at <= self.now:
self.data.pop(key, None)
self.expire_at.pop(key, None)
def set(self, key, value, ex=None, xx=False, keepttl=False):
self._purge(key)
if xx and key not in self.data:
return None
self.data[key] = value
if ex is not None:
self.expire_at[key] = self.now + int(ex)
elif not keepttl:
self.expire_at.pop(key, None)
return True
def get(self, key):
self._purge(key)
return self.data.get(key)
def delete(self, key):
self.expire_at.pop(key, None)
return int(self.data.pop(key, None) is not None)
def expire(self, key, seconds):
self._purge(key)
if key not in self.data:
return False
self.expire_at[key] = self.now + int(seconds)
return True
def ttl(self, key):
self._purge(key)
if key not in self.data:
return -2
if key not in self.expire_at:
return -1
return self.expire_at[key] - self.now
class TestRedisSessionStore(BaseCase):
def _get_store(self, **kwargs):
self.redis = FakeRedis()
return RedisSessionStore(
redis=self.redis,
session_class=http.Session,
expiration=7 * DAY,
anon_expiration=3 * HOUR,
**kwargs,
)
def _login(self, store):
session = store.new()
session.update({"uid": 1, "login": "admin"})
store.save(session)
return session
def _ttl(self, store, session):
return self.redis.ttl(store.build_key(session.sid))
def _use(self, store, session):
"""Simulate a request of the user, which touches the session"""
session = store.get(session.sid)
session["context"] = {"lang": "en_US"}
store.save(session)
return session
def test_sliding_expiration(self):
"""By default, each save renews the expiration of the session"""
store = self._get_store()
session = self._login(store)
self.assertEqual(self._ttl(store, session), 7 * DAY)
for __ in range(10):
self.redis.advance(DAY)
session = self._use(store, session)
self.assertEqual(self._ttl(store, session), 7 * DAY)
# still logged in after 10 days of daily use
self.assertEqual(store.get(session.sid).uid, 1)
def test_absolute_expiration(self):
"""The session expires 7 days after the login, whatever the use"""
store = self._get_store(absolute_expiration=True)
session = self._login(store)
sid = session.sid
for day in range(1, 7):
self.redis.advance(DAY)
session = self._use(store, session)
self.assertEqual(session.uid, 1)
self.assertEqual(self._ttl(store, session), (7 - day) * DAY)
self.assertEqual(store.get(sid)["context"], {"lang": "en_US"})
self.redis.advance(DAY)
self.assertFalse(store.get(sid).uid)
def test_absolute_expiration_not_recreated(self):
"""A request ending after the expiration does not revive the session"""
store = self._get_store(absolute_expiration=True)
session = self._login(store)
session = store.get(session.sid)
self.redis.advance(7 * DAY)
session["context"] = {"lang": "en_US"}
store.save(session)
self.assertEqual(self._ttl(store, session), -2)
def test_absolute_expiration_rotation(self):
"""A rotated session (login, logout, MFA) gets a full expiration"""
store = self._get_store(absolute_expiration=True)
session = self._login(store)
old_sid = session.sid
self.redis.advance(3 * DAY)
store.rotate(session, None)
self.assertNotEqual(session.sid, old_sid)
self.assertIsNone(self.redis.get(store.build_key(old_sid)))
self.assertEqual(self._ttl(store, session), 7 * DAY)
def test_absolute_expiration_anonymous(self):
"""Anonymous sessions keep a sliding expiration"""
store = self._get_store(absolute_expiration=True)
session = store.new()
session["debug"] = ""
store.save(session)
self.assertEqual(self._ttl(store, session), 3 * HOUR)
self.redis.advance(2 * HOUR)
session = self._use(store, session)
self.assertEqual(self._ttl(store, session), 3 * HOUR)
def test_absolute_expiration_key_without_ttl(self):
"""A key without TTL (left by the former SET then EXPIRE) gets one"""
store = self._get_store(absolute_expiration=True)
session = self._login(store)
self.redis.expire_at.pop(store.build_key(session.sid))
self.assertEqual(self._ttl(store, session), -1)
session = self._use(store, session)
self.assertEqual(self._ttl(store, session), 7 * DAY)
Reference in new issue
Block a user
Blocking a user prevents them from interacting with repositories, such as opening or commenting on pull requests or issues. Learn more about blocking a user.