PARAF-503: Added acroform signature tags validation

.github/workflows/main.yml

@@ -26,7 +26,7 @@ jobs:
       - name: Install dependencies
         run: |
           sudo apt-get update
-          sudo apt-get install -y libjpeg8-dev libbz2-dev liblzma-dev
+          sudo apt-get install -y libjpeg8-dev libbz2-dev liblzma-dev poppler-utils
       - name: Set up pyenv and Python
         uses: "gabrielfalcao/pyenv-action@v18"
         with:
@@ -64,7 +64,7 @@ jobs:
       - name: Setup Env
         run: |
           sudo apt-get update -qqy
-          sudo apt-get install -y libjpeg8-dev libbz2-dev liblzma-dev libreadline-dev
+          sudo apt-get install -y libjpeg8-dev libbz2-dev liblzma-dev libreadline-dev poppler-utils
       - name: Set up pyenv and Python
         uses: "gabrielfalcao/pyenv-action@v18"
         with:

CHANGES.rst

@@ -11,6 +11,8 @@ Changelog
   [chris-adam]
 - When a session status is received as completed, mark each signer as signed.
   [sgeulette]
+- Added acroform signature and seal tags validation (PARAF-503).
+  [chris-adam]
 
 1.0b10 (2026-06-18)
 -------------------

setup.py

@@ -78,6 +78,7 @@
             "mock",
             "imio.annex",
             "mock",
+            "reportlab",
         ],
     },
     entry_points="""

src/imio/esign/acroform.py

@@ -0,0 +1,192 @@
+# -*- coding: utf-8 -*-
+"""Detection and validation of the acroform tags of a signable file.
+
+A signable file may carry one tag per signer, telling the external signing service where to
+paste the signature image, like {{#"ID":"Signer1","Size":{"Height":"70","Width":"200"}#}}
+and one seal tag, for the seal image, like
+{{#"ID":"SCEAU","Size":{"Height":"200","Width":"200"}#}}
+"""
+
+from collections import OrderedDict
+from imio.esign import _
+from imio.esign import logger
+from imio.esign.interfaces import IContextUidProvider
+from imio.esign.utils import get_session_info
+from imio.helpers.content import uuidToObject
+from io import BytesIO
+from Products.CMFPlone.utils import safe_unicode
+from xml.sax.saxutils import unescape
+from zope.component import getAdapter
+from zope.i18n import translate
+
+import re
+import subprocess
+import zipfile
+
+
+ACROFORM_TAG_RE = re.compile(r"\{\{#(.*?)#\}\}")
+SIGNER_ID_RE = re.compile(r'"ID":"Signer(0|[1-9]\d*)"')
+SEAL_ID_RE = re.compile(r'"ID":"SCEAU"')
+XML_TAG_RE = re.compile(r"<[^>]*>")
+WHITESPACE_RE = re.compile(r"\s+")
+PDFTOTEXT = "pdftotext"
+
+# content type -> path, inside the zipped document, of the xml part holding the text
+ZIPPED_XML_PARTS = {
+    "application/vnd.oasis.opendocument.text": "content.xml",
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": "word/document.xml",
+}
+
+
+def _pdf_text(data):
+    """Return the text of a pdf, read by the poppler pdftotext binary."""
+    process = subprocess.Popen(
+        [PDFTOTEXT, "-q", "-enc", "UTF-8", "-", "-"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    out, err = process.communicate(data)
+    if process.returncode != 0:
+        raise RuntimeError("pdftotext failed: {}".format(err))
+    return safe_unicode(out)
+
+
+def _zipped_xml_text(data, part_name):
+    """Return the text of the given xml part of a zipped document (odt, docx).
+
+    The entities have to be resolved: a document generated by appy/POD stores the double
+    quotes of a tag as &quot;, so the tag would never be recognised (PARAF-503).
+    """
+    zip_file = zipfile.ZipFile(BytesIO(data))
+    try:
+        xml = safe_unicode(zip_file.read(part_name))
+    finally:
+        zip_file.close()
+    return unescape(XML_TAG_RE.sub(u" ", xml), {u"&quot;": u'"', u"&apos;": u"'"})
+
+
+def extract_text(nbf):
+    """Return the text of a file field, with every whitespace character removed."""
+    if nbf is None or not getattr(nbf, "data", None):
+        return u""
+    try:
+        if nbf.contentType == "application/pdf":
+            text = _pdf_text(nbf.data)
+        elif nbf.contentType in ZIPPED_XML_PARTS:
+            text = _zipped_xml_text(nbf.data, ZIPPED_XML_PARTS[nbf.contentType])
+        else:
+            return u""
+    except OSError:
+        logger.error("Cannot run '%s': the signature tags of pdf files are not checked!", PDFTOTEXT)
+        return u""
+    except Exception:
+        logger.debug("Could not extract the text of file '%s'", nbf.filename, exc_info=True)
+        return u""
+    return WHITESPACE_RE.sub(u"", text)
+
+
+def get_tag_ids(nbf):
+    """Return the acroform tag ids of a file field, reading its text only once.
+
+    :return: the signer number of every signature tag, in document order, and the number
+             of seal tags
+    """
+    numbers = []
+    seals = 0
+    for payload in ACROFORM_TAG_RE.findall(extract_text(nbf)):
+        match = SIGNER_ID_RE.search(payload)
+        if match:
+            numbers.append(int(match.group(1)))
+        elif SEAL_ID_RE.search(payload):
+            seals += 1
+    return numbers, seals
+
+
+def validate_signer_numbers(numbers, nb_signers):
+    """Return the acroform tag error messages for one file.
+
+    A file is valid when it holds no tag at all, or exactly one tag per signer from Signer1
+    to Signer<nb_signers>
+    """
+    if not numbers:
+        return []
+    counts = {}
+    for nb in numbers:
+        counts[nb] = counts.get(nb, 0) + 1
+    errors = []
+    for nb in sorted(counts):
+        if counts[nb] > 1:
+            errors.append(_("The signature tag of Signer${nb} is present ${count} times.",
+                            mapping={"nb": nb, "count": counts[nb]}))
+        if nb < 1 or nb > nb_signers:
+            errors.append(_("There is a signature tag for Signer${nb} but ${count} signer(s) are defined.",
+                            mapping={"nb": nb, "count": nb_signers}))
+    for nb in range(1, nb_signers + 1):
+        if nb not in counts:
+            errors.append(_("The signature tag of Signer${nb} is missing.", mapping={"nb": nb}))
+    return errors
+
+
+def validate_seal_count(count, seal):
+    """Return the acroform seal tag error messages for one file.
+
+    A file is valid when it holds no seal tag, or exactly one while a seal is defined.
+    """
+    if not count:
+        return []
+    if not seal:
+        return [_("There is a seal tag but no seal is defined.")]
+    if count > 1:
+        return [_("The seal tag is present ${count} times.", mapping={"count": count})]
+    return []
+
+
+def check_file(obj, nb_signers, seal=False):
+    """Return the acroform tag error messages of a content object holding a file.
+
+    :param nb_signers: number of signers, or None when the signature tags are unused
+    :param seal: whether a seal is defined on the container
+    """
+    numbers, seal_count = get_tag_ids(getattr(obj, "file", None))
+    errors = [] if nb_signers is None else validate_signer_numbers(numbers, nb_signers)
+    return errors + validate_seal_count(seal_count, seal)
+
+
+def get_session_acroform_errors(session_id):
+    """Return the acroform tag errors of the files of a session, checked against its signers.
+
+    :param session_id: internal session id
+    :return: OrderedDict {file uid: (file object, context uid, [error messages])}, holding
+             only the files in error
+    """
+    res = OrderedDict()
+    session = get_session_info(session_id)
+    if not session:
+        return res
+    nb_signers = len(session.get("signers", []))
+    for file_info in session["files"]:
+        obj = uuidToObject(uuid=file_info["uid"], unrestricted=True)
+        if obj is None:
+            continue
+        errors = check_file(obj, nb_signers, session.get("seal", False))
+        if errors:
+            context_uid = getAdapter(obj, IContextUidProvider).get_context_uid()
+            res[file_info["uid"]] = (obj, context_uid, errors)
+    return res
+
+
+def format_errors(pairs, request):
+    """Return a one line readable summary of acroform tag errors.
+
+    :param pairs: iterable of (file object, [error messages])
+    :param request: used to translate the messages
+    """
+    return u" / ".join(
+        [
+            u"{}: {}".format(
+                safe_unicode(obj.Title()), u" ".join([translate(msg, context=request) for msg in messages])
+            )
+            for obj, messages in pairs
+        ]
+    )

src/imio/esign/browser/configure.zcml

@@ -4,6 +4,7 @@
         xmlns:i18n="http://namespaces.zope.org/i18n"
         xmlns:plone="http://namespaces.plone.org/plone"
         xmlns:browser="http://namespaces.zope.org/browser"
+        xmlns:zcml="http://namespaces.zope.org/zcml"
         i18n_domain="imio.esign">
 
     <include package="eea.facetednavigation" />
@@ -87,6 +88,16 @@
             permission="imio.esign.ManageSessions"
     /-->
 
+    <browser:viewlet
+            zcml:condition="installed imio.annex"
+            for="imio.annex.content.annex.IAnnex"
+            manager="plone.app.layout.viewlets.interfaces.IAboveContentTitle"
+            name="esign-acroform-errors"
+            template="templates/acroform_errors.pt"
+            class=".views.AcroformErrorsViewlet"
+            permission="zope2.View"
+    />
+
     <browser:view
             name="add-to-esign-session"
             for="*"

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

@@ -0,0 +1,17 @@
+<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en"
+      xmlns:tal="http://xml.zope.org/namespaces/tal"
+      xmlns:metal="http://xml.zope.org/namespaces/metal"
+      xmlns:i18n="http://xml.zope.org/namespaces/i18n"
+      lang="en"
+      i18n:domain="imio.esign">
+  <div class="portalMessage error acroform-errors">
+    <strong i18n:translate="">Wrong signature or seal tags</strong>
+    <ul>
+      <li tal:repeat="item python: view.errors.values()">
+        <a tal:attributes="href python: item[0].absolute_url()"
+           tal:content="python: item[0].Title()">file title</a>
+        <div tal:repeat="message python: item[1]" tal:content="message">error message</div>
+      </li>
+    </ul>
+  </div>
+</html>

src/imio/esign/browser/views.py

@@ -1,11 +1,14 @@
 # -*- coding: utf-8 -*-
 
 from AccessControl import Unauthorized
+from collections import OrderedDict
 from copy import deepcopy
 from datetime import datetime
 from datetime import timedelta
 from imio.esign import _
 from imio.esign import manage_session_perm
+from imio.esign.acroform import format_errors
+from imio.esign.acroform import get_session_acroform_errors
 from imio.esign.audit import audit
 from imio.esign.browser.table import external_session_link
 from imio.esign.browser.table import SessionsTable
@@ -166,6 +169,23 @@ def __call__(self, session_id=None):
         if session_id is None:
             api.portal.show_message(_("No session ID provided!"), request=self.request, type="error")
             return self.context.absolute_url() + "/@@parapheo"
+        acroform_errors = get_session_acroform_errors(int(session_id))
+        if acroform_errors:
+            audit("send_to_external_service", "session={} error=acroform".format(session_id))
+            api.portal.show_message(
+                _(
+                    "Session ${id} not sent because signature or seal tags are wrong: ${details}",
+                    mapping={
+                        "id": session_id,
+                        "details": format_errors(
+                            [(obj, messages) for obj, _cuid, messages in acroform_errors.values()], self.request
+                        ),
+                    },
+                ),
+                request=self.request,
+                type="error",
+            )
+            return self.context.absolute_url() + "/@@parapheo"
         resp = create_external_session(int(session_id))
         if resp == "_session_not_found_":
             audit("send_to_external_service", "session={} error=session_not_found".format(session_id))
@@ -296,6 +316,45 @@ def sessions(self):
         return get_sessions_for(self.context.UID())
 
 
+class AcroformErrorsViewlet(ViewletBase):
+    """Show the signature tag errors of the files of a draft session.
+
+    Works both on a file and on its container:
+    - a file reports its own errors
+    - a container reports the errors of the files it owns.
+    """
+
+    index = ViewPageTemplateFile("templates/acroform_errors.pt")
+    draft_states = ("draft", "draft_full")
+
+    @CachedProperty
+    def errors(self):
+        """Return OrderedDict {file uid: (file object, [translated error messages])}."""
+        uid = self.context.UID()
+        annot = get_session_annotation()
+        if uid in annot["uids"]:  # the context is a file itself
+            session_ids = [annot["uids"][uid]]
+        else:                     # the context is a container
+            session_ids = list(get_sessions_for(uid).keys())
+        res = OrderedDict()
+        for session_id in session_ids:
+            if get_session_info(session_id).get("state") not in self.draft_states:
+                continue
+            for f_uid, (obj, context_uid, messages) in get_session_acroform_errors(session_id).items():
+                if uid not in (f_uid, context_uid):
+                    continue
+                res[f_uid] = (obj, [translate(msg, context=self.request) for msg in messages])
+        return res
+
+    def available(self):
+        return bool(self.errors)
+
+    def render(self):
+        if not self.available():
+            return ""
+        return self.index()
+
+
 @implementer(IPublishTraverse)
 class DownloadFileView(BrowserView):
     """View to download a file based on an identifier passed in the URL path.