[18.0][FIX] impersonate_login: close the log on logout

impersonate_login/__init__.py

@@ -1,2 +1,3 @@
+from . import controllers
 from . import models
 from .hooks import pre_init_hook

impersonate_login/controllers/__init__.py

@@ -0,0 +1 @@
+from . import session

impersonate_login/controllers/session.py

@@ -0,0 +1,33 @@
+# Copyright 2026 360ERP (<https://www.360erp.com>)
+# 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()

impersonate_login/models/impersonate_log.py

@@ -2,8 +2,21 @@
 # @author Kévin Roche <kevin.roche@akretion.com>
 # 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)

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(

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.

impersonate_login/tests/__init__.py

@@ -1 +1,2 @@
 from . import test_impersonate_login
+from . import test_impersonate_log_end

impersonate_login/tests/test_impersonate_log_end.py

@@ -0,0 +1,196 @@
+# Copyright 2026 360ERP (<https://www.360erp.com>)
+# 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)