diff --git a/impersonate_login/__init__.py b/impersonate_login/__init__.py index 6d58305f5d..6728c198e3 100644 --- a/impersonate_login/__init__.py +++ b/impersonate_login/__init__.py @@ -1,2 +1,3 @@ +from . import controllers from . import models from .hooks import pre_init_hook diff --git a/impersonate_login/controllers/__init__.py b/impersonate_login/controllers/__init__.py new file mode 100644 index 0000000000..0384bae6fc --- /dev/null +++ b/impersonate_login/controllers/__init__.py @@ -0,0 +1 @@ +from . import session diff --git a/impersonate_login/controllers/session.py b/impersonate_login/controllers/session.py new file mode 100644 index 0000000000..a119507aa7 --- /dev/null +++ b/impersonate_login/controllers/session.py @@ -0,0 +1,33 @@ +# Copyright 2026 360ERP () +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo import http +from odoo.http import request + +from odoo.addons.web.controllers.session import Session + + +class Session(Session): + def _impersonate_readonly(self): + """Serve these routes with a read/write cursor. + + The web module declares both routes as readonly, but the impersonate + log has to be closed before the session is destroyed. Letting Odoo + replay the request on a read/write cursor is not an option: the + session is already logged out by then, so the log id would be lost. + + A callable is used, like the web module does for /web/dataset/call_kw, + because a plain ``readonly=False`` makes Odoo log a warning about an + override changing the read/write mode of the route. + """ + return False + + @http.route(readonly=_impersonate_readonly) + def logout(self, redirect="/odoo"): + request.env["impersonate.log"]._close_session_log() + return super().logout(redirect=redirect) + + @http.route(readonly=_impersonate_readonly) + def destroy(self): + request.env["impersonate.log"]._close_session_log() + return super().destroy() diff --git a/impersonate_login/models/impersonate_log.py b/impersonate_login/models/impersonate_log.py index 83089611a6..ab08226418 100644 --- a/impersonate_login/models/impersonate_log.py +++ b/impersonate_login/models/impersonate_log.py @@ -2,8 +2,21 @@ # @author Kévin Roche # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +import logging +import time +from datetime import timedelta -from odoo import fields, models +from odoo import SUPERUSER_ID, api, fields, models +from odoo.http import get_session_max_inactivity, request + +_logger = logging.getLogger(__name__) + +# Minimum delay, in seconds, between two touches of an open log. Low enough to +# keep the end date accurate, high enough to avoid a write on every request. +ACTIVITY_REFRESH_DELAY = 60 + +# Key used in the session to throttle those touches. +ACTIVITY_SESSION_KEY = "impersonate_activity_ts" class ImpersonateLog(models.Model): @@ -23,3 +36,70 @@ class ImpersonateLog(models.Model): date_end = fields.Datetime( string="End Date", ) + + @api.model + def _close_session_log(self): + """Set the end date of the impersonation running in the current session. + + The logout route runs with ``auth="none"``, so the request may have no + user at all. Writing as the superuser keeps the log writable in every + case. + """ + if not request or not request.session.impersonate_log_id: + return + log = ( + request.env(user=SUPERUSER_ID, su=True)["impersonate.log"] + .browse(request.session.impersonate_log_id) + .exists() + ) + if log and not log.date_end: + log.date_end = fields.Datetime.now() + + @api.model + def _touch_session_log(self): + """Record that the impersonation of the current session is still alive. + + Touching the log refreshes its ``write_date``, which is then used as + the last known activity of the impersonation. Called on every request, + but writes at most once every ``ACTIVITY_REFRESH_DELAY`` seconds. + """ + session = request.session + log_id = session.impersonate_log_id + if not log_id: + return + if self.env.cr.readonly: + # A readonly request cannot write. Skipping is on purpose: Odoo + # would otherwise replay the whole request on a read/write cursor, + # which is way too expensive for a heartbeat. The next read/write + # request of the session refreshes the activity instead. + return + now = time.time() + if now - (session.get(ACTIVITY_SESSION_KEY) or 0) < ACTIVITY_REFRESH_DELAY: + return + session[ACTIVITY_SESSION_KEY] = now + request.env(user=SUPERUSER_ID, su=True)["impersonate.log"].browse( + log_id + ).exists().write({}) + + @api.autovacuum + def _gc_expired_logs(self): + """Close the logs whose session can no longer be used. + + Odoo reaps a session that has been inactive for longer than + ``sessions.max_inactivity_seconds``, so a log left open with no + activity since then belongs to a session that is definitely gone. Its + last activity is recorded as the end date, which is the closest we can + get to the real end of the impersonation. + + This runs with the daily vacuum, next to the garbage collection of the + sessions themselves. + """ + threshold = fields.Datetime.now() - timedelta( + seconds=get_session_max_inactivity(self.env) + ) + logs = self.search([("date_end", "=", False), ("write_date", "<", threshold)]) + for log in logs: + log.date_end = log.write_date + if logs: + _logger.info("Closed %s expired impersonate log(s).", len(logs)) + return len(logs) diff --git a/impersonate_login/models/ir_http.py b/impersonate_login/models/ir_http.py index f01aa613d2..2f19d6764b 100644 --- a/impersonate_login/models/ir_http.py +++ b/impersonate_login/models/ir_http.py @@ -9,6 +9,14 @@ class Http(models.AbstractModel): _inherit = "ir.http" + @classmethod + def _pre_dispatch(cls, rule, args): + res = super()._pre_dispatch(rule, args) + # Keep track of the activity of an impersonated session, so that its + # log can still be closed if the session ends without a logout. + request.env["impersonate.log"]._touch_session_log() + return res + def session_info(self): session_info = super().session_info() session_info.update( diff --git a/impersonate_login/readme/CONFIGURE.md b/impersonate_login/readme/CONFIGURE.md index 649fb31c76..6a103190b9 100644 --- a/impersonate_login/readme/CONFIGURE.md +++ b/impersonate_login/readme/CONFIGURE.md @@ -3,3 +3,10 @@ The impersonating user must belong to group "Impersonate Users". If you want to prevent impersonation of users with the *Administration: Settings* rights, enable the *Restrict Impersonation of "Administration: Settings" Users* option in the settings. + +A log is closed as soon as the impersonating user goes back to their own user +or logs out. When a session is simply abandoned, the log is closed by the daily +"Vacuum" scheduled action, once the session can no longer be used. The last +known activity is then recorded as the end date. That delay follows the +standard Odoo system parameter `sessions.max_inactivity_seconds` (7 days by +default); lower it if you want dangling logs to be closed sooner. diff --git a/impersonate_login/tests/__init__.py b/impersonate_login/tests/__init__.py index d2f02c982b..acca89e5fc 100644 --- a/impersonate_login/tests/__init__.py +++ b/impersonate_login/tests/__init__.py @@ -1 +1,2 @@ from . import test_impersonate_login +from . import test_impersonate_log_end diff --git a/impersonate_login/tests/test_impersonate_log_end.py b/impersonate_login/tests/test_impersonate_log_end.py new file mode 100644 index 0000000000..5c264fca25 --- /dev/null +++ b/impersonate_login/tests/test_impersonate_log_end.py @@ -0,0 +1,196 @@ +# Copyright 2026 360ERP () +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl) + +import json +from datetime import timedelta +from uuid import uuid4 + +from odoo import fields, http +from odoo.tests import HttpCase, tagged + +from odoo.addons.impersonate_login.models.impersonate_log import ACTIVITY_SESSION_KEY + + +@tagged("post_install", "-at_install") +class TestImpersonateLogEnd(HttpCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.admin_user = cls.env.ref("base.user_admin") + cls.demo_user = cls.env.ref("base.user_demo") + cls.log_model = cls.env["impersonate.log"] + + # Helpers + + def _call_button(self, model, method, args): + response = self.url_open( + "/web/dataset/call_button", + data=json.dumps( + { + "params": { + "model": model, + "method": method, + "args": args, + "kwargs": {}, + }, + } + ), + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(response.status_code, 200) + return response.json() + + def _impersonate_user(self, user): + return self._call_button("res.users", "impersonate_login", [user.id]) + + def _read_write_request(self): + """Send a request served by a read/write cursor.""" + return self._call_button("res.users", "action_impersonate_login", []) + + def _get_session_info(self): + """Send a request served by a readonly cursor.""" + response = self.url_open( + "/web/session/get_session_info", + data=json.dumps({"jsonrpc": "2.0", "method": "call", "id": str(uuid4())}), + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(response.status_code, 200) + return response.json() + + def _logout(self): + response = self.url_open("/web/session/logout", allow_redirects=False) + self.assertEqual(response.status_code, 303) + return response + + def _destroy_session(self): + response = self.url_open( + "/web/session/destroy", + data=json.dumps({"jsonrpc": "2.0", "method": "call", "id": str(uuid4())}), + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(response.status_code, 200) + return response.json() + + def _last_log(self): + return self.log_model.search([], order="id desc", limit=1) + + def _new_log(self): + return self.log_model.create( + { + "user_id": self.admin_user.id, + "impersonated_partner_id": self.demo_user.partner_id.id, + "date_start": fields.Datetime.now(), + } + ) + + def _set_write_date(self, log, value): + """Move write_date in the past, which the ORM does not allow.""" + self.env.cr.execute( + "UPDATE impersonate_log SET write_date = %s WHERE id = %s", + (value, log.id), + ) + log.invalidate_recordset() + + def _reset_activity_throttle(self): + """Make the session believe its activity was never recorded.""" + session = http.root.session_store.get(self.session.sid) + session[ACTIVITY_SESSION_KEY] = 0.0 + http.root.session_store.save(session) + + # Tests + + def test_10_logout_closes_the_log(self): + """Logging out while impersonating closes the log""" + self.authenticate(user="admin", password="admin") + self._impersonate_user(self.demo_user) + + log = self._last_log() + self.assertEqual(log.user_id, self.admin_user) + self.assertTrue(log.date_start) + self.assertFalse(log.date_end) + + self._logout() + + log.invalidate_recordset() + self.assertTrue(log.date_end) + + def test_20_destroy_session_closes_the_log(self): + """Destroying the session while impersonating closes the log""" + self.authenticate(user="admin", password="admin") + self._impersonate_user(self.demo_user) + log = self._last_log() + self.assertFalse(log.date_end) + + self._destroy_session() + + log.invalidate_recordset() + self.assertTrue(log.date_end) + + def test_30_logout_without_impersonation(self): + """Logging out without impersonating anybody does not touch any log""" + self.authenticate(user="admin", password="admin") + log = self._new_log() + self.env.flush_all() + + self._logout() + + log.invalidate_recordset() + self.assertFalse(log.date_end) + + def test_40_activity_is_recorded(self): + """A read/write request of an impersonated session refreshes the activity""" + self.authenticate(user="admin", password="admin") + self._impersonate_user(self.demo_user) + log = self._last_log() + + past = fields.Datetime.now() - timedelta(hours=2) + self._set_write_date(log, past) + + # Reset the throttle, otherwise the next request is too close in time + # to the previous one to trigger a write. + self._reset_activity_throttle() + self.env.flush_all() + + self._read_write_request() + + log.invalidate_recordset() + self.assertGreater(log.write_date, past) + self.assertFalse(log.date_end) + + def test_45_readonly_request_does_not_write(self): + """A readonly request leaves the log untouched""" + self.authenticate(user="admin", password="admin") + self._impersonate_user(self.demo_user) + log = self._last_log() + + past = fields.Datetime.now() - timedelta(hours=2) + self._set_write_date(log, past) + self._reset_activity_throttle() + self.env.flush_all() + + self._get_session_info() + + log.invalidate_recordset() + self.assertEqual(log.write_date, past) + + def test_50_vacuum_closes_an_expired_log(self): + """The vacuum closes a log whose session cannot be used anymore""" + log = self._new_log() + self.env.flush_all() + max_inactivity = http.get_session_max_inactivity(self.env) + long_ago = fields.Datetime.now() - timedelta(seconds=max_inactivity + 3600) + self._set_write_date(log, long_ago) + + self.log_model._gc_expired_logs() + + self.assertEqual(log.date_end, long_ago) + + def test_60_vacuum_keeps_a_live_log_open(self): + """The vacuum leaves a log with recent activity alone""" + log = self._new_log() + self.env.flush_all() + + self.log_model._gc_expired_logs() + + log.invalidate_recordset() + self.assertFalse(log.date_end)