Auto delete old esign sessions

Author: chris-adam

CHANGES.rst

@@ -5,7 +5,8 @@ Changelog
 1.0b8 (unreleased)
 ------------------
 
-- Nothing changed yet.
+- Auto delete old esign sessions.
+  [chris-adam]
 
 
 1.0b7 (2026-04-02)
@@ -36,10 +37,12 @@ Changelog
 ------------------
 
 - Renamed imio.esign config functions.
-  [cadam]
+  [chris-adam]
 - Highlight `draft` session in table view and viewlet, use `Id` as
   column header instead `identifier` so it is more narrow and
   make `sessions` view columns sortable.
+  [chris-adam]
+- Highlight `draft` session and make `sessions` view columns sortable.
   [gbastien]
 
 1.0b3 (2026-03-26)
@@ -130,4 +133,4 @@ Changelog
 ------------------
 
 - Initial release.
-  [sgeulette, gbastien, aduchene, cadam]
+  [sgeulette, gbastien, aduchene, chris-adam]

src/imio/esign/__init__.py

@@ -13,6 +13,7 @@
 logger = logging.getLogger("imio.esign")
 PLONE_VERSION = int(api.env.plone_version()[0])
 API_ROOT_URL = os.getenv("API_ROOT_URL", "http://127.0.0.1:8000")
+CLEANUP_THROTTLE_HOURS = 24
 manage_session_perm = "imio.esign: Manage Sessions"
 
 

src/imio/esign/browser/settings.py

@@ -112,6 +112,17 @@ class IImioEsignSettings(Interface):
         required=False,
     )
 
+    auto_cleanup_days = schema.Int(
+        title=_("Auto-cleanup delay (days)"),
+        description=_(
+            "Number of days after a session enters 'to_sign' state before it is automatically "
+            "deleted (finalized sessions only). Set to 0 to disable auto-cleanup."
+        ),
+        default=100,
+        min=0,
+        required=False,
+    )
+
 
 class ImioEsignSettings(RegistryEditForm):
     schema = IImioEsignSettings

src/imio/esign/browser/table.py

@@ -5,6 +5,7 @@
 from imio.esign import _
 from imio.esign.config import get_esign_registry_seal_code
 from imio.esign.config import get_esign_registry_seal_email
+from imio.esign.utils import get_deletion_date_msg
 from imio.esign.utils import get_state_description
 from imio.helpers.security import check_zope_admin
 from imio.pyutils.utils import safe_encode
@@ -60,6 +61,9 @@ def renderCell(self, item):
         ))
         title = escape(translate(get_state_description(item.get("state", "")), context=self.request,
                                  domain="imio.esign"))
+        deletion_msg = escape(get_deletion_date_msg(item, self.request))
+        if deletion_msg:
+            title = title + u"\n\n" + deletion_msg
         return (u"<span class='state-title state-title-{state_title_value}' title='{title}'>{state} <span class='far fa-question-circle' />"
                 u"</span>".format(state=state, title=title, state_title_value=item.get("state")))
 

src/imio/esign/browser/templates/macros.pt

@@ -11,7 +11,7 @@
     <td class="table_widget_label"><label i18n:translate="">State</label></td>
     <td class="table_widget_value">
         <span tal:attributes="class string:state-title state-title-${session/state};
-                              title python:view.get_state_description(session['state'])">
+                              title python:view.get_state_title(session)">
           <tal:block content="python:session['state']" i18n:translate="">draft</tal:block>
           <span class='far fa-question-circle' />
         </span>

src/imio/esign/browser/views.py

@@ -12,7 +12,9 @@
 from imio.esign.config import get_esign_registry_enabled
 from imio.esign.config import get_esign_registry_parapheo_url
 from imio.esign.config import get_esign_registry_signing_users_email_content
+from imio.esign.utils import cleanup_expired_sessions
 from imio.esign.utils import create_external_session
+from imio.esign.utils import get_deletion_date_msg
 from imio.esign.utils import get_session_annotation
 from imio.esign.utils import get_session_info
 from imio.esign.utils import get_sessions_for
@@ -41,6 +43,7 @@
 
 import csv
 import json
+import logging
 import os
 
 
@@ -50,6 +53,9 @@
     from io import StringIO  # Python 3
 
 
+logger = logging.getLogger(__name__)
+
+
 class SessionsListingView(BrowserView):
     """View to list sessions."""
 
@@ -61,6 +67,10 @@ def __init__(self, context, request):
     def __call__(self):
         if not self.available():
             raise Unauthorized
+        try:
+            cleanup_expired_sessions()
+        except Exception:
+            logger.exception("Auto-cleanup failed in SessionsListingView.__call__")
         return super(SessionsListingView, self).__call__()
 
     def available(self):
@@ -262,6 +272,15 @@ def collapsible_content_css_default(self):
     def get_state_description(self, state):
         return translate(get_state_description(state), context=self.request, domain="imio.esign")
 
+    def get_state_title(self, session):
+        """Return state description with auto-deletion date appended for finalized sessions."""
+        state = session.get("state", "")
+        title = translate(get_state_description(state), context=self.request, domain="imio.esign")
+        deletion_msg = get_deletion_date_msg(session, self.request)
+        if deletion_msg:
+            title = title + u"\n\n" + deletion_msg
+        return title
+
 
 class ItemSessionInfoViewlet(FacetedSessionInfoViewlet):
     """Show session info for all sessions linked to a context item."""

src/imio/esign/config.py

@@ -46,6 +46,10 @@ def get_esign_registry_external_watchers():
     return [ew.strip() for ew in value.split(",") if ew.strip()]
 
 
+def get_esign_registry_auto_cleanup_days(default=100):
+    return api.portal.get_registry_record("imio.esign.auto_cleanup_days", default=default)
+
+
 def set_esign_registry_enabled(value):
     api.portal.set_registry_record("imio.esign.enabled", value)
 
@@ -86,6 +90,10 @@ def set_esign_registry_external_watchers(value):
     api.portal.set_registry_record("imio.esign.external_watchers", value)
 
 
+def set_esign_registry_auto_cleanup_days(value):
+    api.portal.set_registry_record("imio.esign.auto_cleanup_days", value)
+
+
 SIGNERS_EMAIL_CONTENT = u"""
 <meta charset="UTF-8"><tal:global>
 <p style="font-weight: bold;" tal:condition="nothing">!! Attention: ne pas modifier ceci directement mais passer

src/imio/esign/services/external_session_feedback.py

@@ -47,6 +47,7 @@ def reply(self):  # noqa C901
             if code == 21:
                 # sign_session_confirmed
                 session_update["state"] = "to_sign"
+                session_update["to_sign_date"] = datetime.now()
                 if value and "sign_session_url" in value and not session["sign_url"]:
                     session_update["sign_url"] = value["sign_session_url"]
             elif code == 22:

src/imio/esign/tests/test_utils.py

@@ -3,12 +3,15 @@
 from collections import OrderedDict
 from collective.iconifiedcategory.utils import calculate_category_id
 from datetime import date
+from datetime import datetime
 from datetime import timedelta
 from imio.esign.config import get_esign_registry_max_session_size
+from imio.esign.config import set_esign_registry_auto_cleanup_days
 from imio.esign.config import set_esign_registry_external_watchers
 from imio.esign.config import set_esign_registry_max_session_size
 from imio.esign.testing import IMIO_ESIGN_INTEGRATION_TESTING
 from imio.esign.utils import add_files_to_session
+from imio.esign.utils import cleanup_expired_sessions
 from imio.esign.utils import create_external_session
 from imio.esign.utils import get_file_download_url
 from imio.esign.utils import get_file_info
@@ -810,6 +813,108 @@ def test_create_external_session_both_payload(self):
             },
         )
 
+    def test_cleanup_expired_sessions(self):
+        """Auto-cleanup removes expired finalized sessions and respects the delay and throttle."""
+        signers = [("user1", "user1@sign.com", "User 1", "Position 1")]
+        annot = get_session_annotation()
+        self.addCleanup(set_esign_registry_auto_cleanup_days, 100)
+
+        # --- Eligibility: only finalized sessions are cleaned up ---
+
+        # Draft session with old last_update: never removed
+        sid_draft, session_draft = add_files_to_session(signers, (self.uids[0],), discriminators=("draft",))
+        session_draft["last_update"] = datetime.now() - timedelta(days=200)
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertIn(sid_draft, annot["sessions"])
+        self.assertIn("last_cleanup", annot)
+
+        # Non-finalized states (sent, to_sign, returned, etc.): never removed regardless of age
+        for state in ("sent", "to_sign", "returned", "refused", "signed", "errored"):
+            sid, session = add_files_to_session(
+                signers, (self.uids[1],), discriminators=(state,)
+            )
+            session["state"] = state
+            session["last_update"] = datetime.now() - timedelta(days=200)
+            annot["last_cleanup"] = datetime.min
+            cleanup_expired_sessions()
+            self.assertIn(sid, annot["sessions"], "State %s should not be cleaned up" % state)
+            del annot["sessions"][sid]  # clean up manually for next iteration
+
+        # --- Expiry based on last_update (fallback when to_sign_date absent) ---
+
+        # Finalized, last_update within window: not removed
+        sid_recent_fin, session_recent_fin = add_files_to_session(
+            signers, (self.uids[1],), discriminators=("recent_fin",)
+        )
+        session_recent_fin["state"] = "finalized"
+        session_recent_fin["last_update"] = datetime.now() - timedelta(days=50)
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertIn(sid_recent_fin, annot["sessions"])
+
+        # Finalized, last_update expired (> 100 days), no to_sign_date: removed
+        sid_old_fin, session_old_fin = add_files_to_session(
+            signers, (self.uids[2],), discriminators=("old_fin",)
+        )
+        session_old_fin["state"] = "finalized"
+        session_old_fin["last_update"] = datetime.now() - timedelta(days=101)
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertNotIn(sid_old_fin, annot["sessions"])
+        self.assertNotIn(self.uids[2], annot["uids"])
+        self.assertIn(sid_recent_fin, annot["sessions"])  # recent session untouched
+
+        # --- Expiry based on to_sign_date (takes priority over last_update) ---
+
+        # to_sign_date expired but last_update recent: to_sign_date wins → removed
+        sid_ts_old, session_ts_old = add_files_to_session(
+            signers, (self.uids[2],), discriminators=("ts_old",)
+        )
+        session_ts_old["state"] = "finalized"
+        session_ts_old["to_sign_date"] = datetime.now() - timedelta(days=101)
+        session_ts_old["last_update"] = datetime.now() - timedelta(days=1)
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertNotIn(sid_ts_old, annot["sessions"])
+
+        # to_sign_date within window but last_update old: to_sign_date wins → kept
+        sid_ts_recent, session_ts_recent = add_files_to_session(
+            signers, (self.uids[2],), discriminators=("ts_recent",)
+        )
+        session_ts_recent["state"] = "finalized"
+        session_ts_recent["to_sign_date"] = datetime.now() - timedelta(days=50)
+        session_ts_recent["last_update"] = datetime.now() - timedelta(days=200)
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertIn(sid_ts_recent, annot["sessions"])
+
+        # --- Throttle ---
+
+        sid_throttle, session_throttle = add_files_to_session(
+            signers, (self.uids[3],), discriminators=("throttle",)
+        )
+        session_throttle["state"] = "finalized"
+        session_throttle["last_update"] = datetime.now() - timedelta(days=101)
+        annot["last_cleanup"] = datetime.now() - timedelta(hours=1)  # ran 1h ago
+        cleanup_expired_sessions()
+        self.assertIn(sid_throttle, annot["sessions"])  # throttled: skipped
+
+        # --- Disabled (0 days) ---
+
+        set_esign_registry_auto_cleanup_days(0)
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertIn(sid_throttle, annot["sessions"])  # disabled: skipped
+
+        # --- Custom delay: 10 days ---
+
+        set_esign_registry_auto_cleanup_days(10)
+        # session_throttle has last_update 101 days ago → expired under 10-day rule
+        annot["last_cleanup"] = datetime.min
+        cleanup_expired_sessions()
+        self.assertNotIn(sid_throttle, annot["sessions"])
+
 
 # example of annotation content
 """

src/imio/esign/utils.py

@@ -6,8 +6,10 @@
 from datetime import timedelta
 from imio.esign import _tr as _
 from imio.esign import API_ROOT_URL
+from imio.esign import CLEANUP_THROTTLE_HOURS
 from imio.esign import logger
 from imio.esign.audit import audit
+from imio.esign.config import get_esign_registry_auto_cleanup_days
 from imio.esign.config import get_esign_registry_external_watchers
 from imio.esign.config import get_esign_registry_file_url
 from imio.esign.config import get_esign_registry_max_session_size
@@ -27,6 +29,7 @@
 from plone import api
 from zope.annotation import IAnnotations
 from zope.component import getAdapter
+from zope.i18n import translate
 
 import json
 import requests
@@ -469,6 +472,82 @@ def remove_session(session_id):
     # logger.info("Session %s removed", session_id)
 
 
+def get_session_deletion_date(session, cleanup_days=None):
+    """Return the expected auto-deletion date for a finalized session, or None.
+
+    Deletion is scheduled ``cleanup_days`` days after the session entered
+    the 'to_sign' state.  Falls back to ``last_update`` when ``to_sign_date``
+    was not recorded (sessions created before this feature).  Returns ``None``
+    when the session is not in 'finalized' state or auto-cleanup is disabled
+    (cleanup_days == 0).
+    """
+    if session.get("state") != "finalized":
+        return None
+    if cleanup_days is None:
+        cleanup_days = get_esign_registry_auto_cleanup_days()
+    if not cleanup_days:
+        return None
+    reference_date = session.get("to_sign_date") or session.get("last_update")
+    if not reference_date:
+        return None
+    deletion_date = reference_date + timedelta(days=cleanup_days)
+    annot = get_session_annotation()
+    return max(deletion_date, annot["last_cleanup"] + timedelta(hours=CLEANUP_THROTTLE_HOURS))
+
+
+def get_deletion_date_msg(session, request):
+    """Return translated deletion date message for a finalized session, or empty string."""
+    deletion_date = get_session_deletion_date(session)
+    if not deletion_date:
+        return u""
+    return translate(
+        _("The session will be automatically deleted on ${date}. "
+          "Files won't be deleted from the application, but the session won't be accessible anymore.",
+          mapping={"date": deletion_date.strftime("%d/%m/%Y %H:%M")}),
+        context=request, domain="imio.esign",
+    )
+
+
+def cleanup_expired_sessions():
+    """Remove finalized sessions that have exceeded the auto_cleanup_days delay.
+
+    The delay is counted from ``to_sign_date`` when available, falling back
+    to ``last_update`` for older sessions that predate this feature.
+
+    Throttled to run at most once every CLEANUP_THROTTLE_HOURS hours via a
+    'last_cleanup' timestamp stored in the portal annotation.
+    """
+    cleanup_days = get_esign_registry_auto_cleanup_days()
+    if not cleanup_days:
+        return
+
+    annot = get_session_annotation()
+    now = datetime.now()
+
+    last_cleanup = annot.get("last_cleanup", datetime.min)
+    if now - last_cleanup < timedelta(hours=CLEANUP_THROTTLE_HOURS):
+        return
+
+    expired = [
+        sid for sid, session in annot["sessions"].items()
+        if (get_session_deletion_date(session, cleanup_days=cleanup_days) or now) < now
+    ]
+
+    for session_id in expired:
+        session = annot["sessions"].get(session_id)
+        if session:
+            logger.info(
+                "Auto-cleanup: removing expired session %s (state=%s, to_sign_date=%s, sign_id=%s)",
+                session_id,
+                session.get("state"),
+                session.get("to_sign_date"),
+                session.get("sign_id"),
+            )
+        remove_session(session_id)
+
+    annot["last_cleanup"] = now
+
+
 def get_file_download_url(uid, root_url=None, short_uid=None):
     """Get the file download URL for a given file UID.