[18.0] l10n_br_fiscal_certificate: refactor l10n_br_fiscal_certificate to reuse new certificate core module

l10n_br_fiscal_certificate/__manifest__.py

@@ -9,9 +9,10 @@
     "maintainers": ["renatonlima"],
     "website": "https://github.com/OCA/l10n-brazil",
     "development_status": "Production/Stable",
-    "version": "18.0.1.3.1",
+    "version": "18.0.2.0.0",
     "depends": [
         "l10n_br_fiscal",
+        "certificate",
     ],
     "data": [
         "security/ir.model.access.csv",

l10n_br_fiscal_certificate/hooks.py

@@ -24,8 +24,9 @@ def prepare_fake_certificate_vals(
         return {
             "type": cert_type,
             "subtype": "a1",
-            "password": passwd,
-            "file": misc.create_fake_certificate_file(
+            "scope": "l10n_br",
+            "pkcs12_password": passwd,
+            "content": misc.create_fake_certificate_file(
                 valid, passwd, issuer, country, subject
             ),
         }
@@ -41,14 +42,16 @@ def prepare_fake_certificate_vals(
             env.ref("l10n_br_base.empresa_lucro_real", raise_if_not_found=False),
         ]
         try:
+            certificate_model = env["certificate.certificate"]
             for company in companies:
-                l10n_br_fiscal_certificate_id = env["l10n_br_fiscal.certificate"]
-                company.certificate_nfe_id = l10n_br_fiscal_certificate_id.create(
-                    prepare_fake_certificate_vals()
-                )
-                company.certificate_ecnpj_id = l10n_br_fiscal_certificate_id.create(
-                    prepare_fake_certificate_vals(cert_type=CERTIFICATE_TYPE_ECNPJ)
-                )
+                if not company:
+                    continue
+                vals = prepare_fake_certificate_vals()
+                vals["company_id"] = company.id
+                company.certificate_nfe_id = certificate_model.create(vals)
+                vals = prepare_fake_certificate_vals(cert_type=CERTIFICATE_TYPE_ECNPJ)
+                vals["company_id"] = company.id
+                company.certificate_ecnpj_id = certificate_model.create(vals)
         except NameError:  # (means from erpbrasil.assinatura import misc failed)
             _logger.error(
                 _(

l10n_br_fiscal_certificate/migrations/18.0.2.0.0/post-migration.py

@@ -0,0 +1,81 @@
+# Copyright (C) 2026  Raphaël Valyi - Akretion
+# License AGPL-3 - See http://www.gnu.org/licenses/agpl-3.0.html
+
+import logging
+
+from openupgradelib import openupgrade
+
+_logger = logging.getLogger(__name__)
+
+_LEGACY_TABLE = "l10n_br_fiscal_certificate"
+_TMP_TABLE = "l10n_br_fiscal_certificate_migration"
+
+
+def _company_by_legacy_cert_id(env):
+    env.cr.execute(
+        "SELECT id, certificate_nfe_id, certificate_ecnpj_id FROM res_company"
+    )
+    mapping = {}
+    for company_id, nfe_id, ecnpj_id in env.cr.fetchall():
+        if nfe_id:
+            mapping.setdefault(nfe_id, company_id)
+        if ecnpj_id:
+            mapping.setdefault(ecnpj_id, company_id)
+    return mapping
+
+
+@openupgrade.migrate(use_env=True)
+def migrate(env, version):
+    if not openupgrade.table_exists(env.cr, _TMP_TABLE):
+        return
+
+    company_by_old = _company_by_legacy_cert_id(env)
+    main_company = env.ref("base.main_company", raise_if_not_found=False)
+    fallback_company_id = main_company.id if main_company else None
+
+    certificate_model = env["certificate.certificate"]
+    old_new = {}
+
+    env.cr.execute(
+        f"SELECT legacy_id, file, password, type, subtype, active "
+        f"FROM {_TMP_TABLE} ORDER BY legacy_id"
+    )
+    for old_id, file, password, ctype, subtype, active in env.cr.fetchall():
+        vals = {
+            "content": file,
+            "pkcs12_password": password,
+            "type": ctype,
+            "subtype": subtype,
+            "active": active,
+            "scope": "l10n_br",
+            "company_id": company_by_old.get(old_id, fallback_company_id),
+        }
+        try:
+            # Core re-parses ``content`` and derives pem_certificate, dates,
+            # subject_common_name, serial_number and the private key. Its
+            # ``_constrains_certificate_loaded`` rejects unparseable files, which
+            # the legacy module also rejected on create, so this is a no-op in
+            # practice.
+            old_new[old_id] = certificate_model.create(vals).id
+        except Exception:  # noqa: BLE001
+            _logger.exception(
+                "Skipping legacy certificate %s: could not be re-parsed", old_id
+            )
+
+    # Repoint the res.company foreign keys to the new certificate IDs.
+    for old_id, new_id in old_new.items():
+        env.cr.execute(
+            "UPDATE res_company SET certificate_nfe_id = %s "
+            "WHERE certificate_nfe_id = %s",
+            (new_id, old_id),
+        )
+        env.cr.execute(
+            "UPDATE res_company SET certificate_ecnpj_id = %s "
+            "WHERE certificate_ecnpj_id = %s",
+            (new_id, old_id),
+        )
+
+    # Cleanup.
+    env.cr.execute(f"DROP TABLE IF EXISTS {_TMP_TABLE}")
+    if openupgrade.table_exists(env.cr, _LEGACY_TABLE):
+        env.cr.execute(f"DROP TABLE IF EXISTS {_LEGACY_TABLE}")

l10n_br_fiscal_certificate/migrations/18.0.2.0.0/pre-migration.py

@@ -0,0 +1,57 @@
+# Copyright (C) 2026  Raphaël Valyi - Akretion
+# License AGPL-3 - See http://www.gnu.org/licenses/agpl-3.0.html
+
+from openupgradelib import openupgrade
+
+_TMP_TABLE = "l10n_br_fiscal_certificate_migration"
+
+
+@openupgrade.migrate(use_env=True)
+def migrate(env, version):
+    """Snapshot the legacy certificates before the model is replaced.
+
+    The old ``l10n_br_fiscal.certificate`` model is merged into the Odoo core
+    ``certificate.certificate`` model. We read it here (while the ORM still knows
+    the old model) and stash the payload into a temporary table, because the
+    ``file`` binary lives in ``ir.attachment`` and can no longer be read through
+    the ORM once the model is gone.
+    """
+    try:
+        legacy_model = env["l10n_br_fiscal.certificate"]
+    except KeyError:
+        return
+
+    legacy_certs = legacy_model.sudo().search([])
+    if not legacy_certs:
+        return
+
+    openupgrade.logged_query(
+        env.cr,
+        f"""
+        CREATE TABLE IF NOT EXISTS {_TMP_TABLE} (
+            legacy_id INTEGER PRIMARY KEY,
+            file TEXT,
+            password VARCHAR,
+            type VARCHAR,
+            subtype VARCHAR,
+            active BOOLEAN
+        )
+        """,
+    )
+    for cert in legacy_certs:
+        openupgrade.logged_query(
+            env.cr,
+            f"""
+            INSERT INTO {_TMP_TABLE}
+                (legacy_id, file, password, type, subtype, active)
+            VALUES (%s, %s, %s, %s, %s, %s)
+            """,
+            (
+                cert.id,
+                cert.with_context(bin_size=False).file or None,
+                cert.password,
+                cert.type,
+                cert.subtype,
+                cert.active,
+            ),
+        )

l10n_br_fiscal_certificate/models/certificate.py

@@ -1,146 +1,88 @@
 # Copyright (C) 2019  Renato Lima - Akretion
+# Copyright (C) 2024  Raphaël Valyi - Akretion
 # License AGPL-3 - See http://www.gnu.org/licenses/agpl-3.0.html
 
+import base64
+from contextlib import suppress
 
-from erpbrasil.assinatura import certificado
+from cryptography import x509
 
-from odoo import _, api, fields, models
-from odoo.exceptions import ValidationError
-from odoo.tools import DEFAULT_SERVER_DATETIME_FORMAT
+from odoo import api, fields, models
 from odoo.tools.misc import format_date
 
-from ..constants import (
-    CERTIFICATE_SUBTYPE,
-    CERTIFICATE_SUBTYPE_DEFAULT,
-    CERTIFICATE_TYPE,
-    CERTIFICATE_TYPE_DEFAULT,
-)
+from ..constants import CERTIFICATE_SUBTYPE, CERTIFICATE_TYPE
 
 
 class Certificate(models.Model):
-    _name = "l10n_br_fiscal.certificate"
-    _inherit = ["mail.thread", "mail.activity.mixin"]
-    _description = "Certificate"
-    _order = "date_expiration"
+    _inherit = "certificate.certificate"
 
-    name = fields.Char(compute="_compute_name", readonly=True)
-
-    active = fields.Boolean(default=True)
-
-    date_start = fields.Datetime(readonly=True, store=True)
-
-    date_expiration = fields.Datetime(readonly=True, store=True)
-
-    issuer_name = fields.Char(size=120, readonly=True, store=True)
-
-    owner_name = fields.Char(string="Owner", size=120, readonly=True, store=True)
-
-    owner_cnpj_cpf = fields.Char(string="CNPJ/CPF", size=18, readonly=True, store=True)
+    scope = fields.Selection(
+        selection_add=[("l10n_br", "Brazilian Fiscal")],
+    )
 
     type = fields.Selection(
         selection=CERTIFICATE_TYPE,
         string="Certificate Type",
-        default=CERTIFICATE_TYPE_DEFAULT,
-        required=True,
     )
 
     subtype = fields.Selection(
         selection=CERTIFICATE_SUBTYPE,
         string="Document SubType",
-        default=CERTIFICATE_SUBTYPE_DEFAULT,
-        required=True,
     )
 
-    file = fields.Binary(string="file", prefetch=True, required=True)
-
-    file_name = fields.Char(compute="_compute_name", size=255)
-
-    password = fields.Char(required=True)
-
-    is_valid = fields.Boolean(compute="_compute_is_valid", string="Is Valid?")
+    name = fields.Char(
+        compute="_compute_name",
+        store=True,
+    )
 
-    @api.model
-    def _certificate_data(self, cert_file, cert_password):
-        values = {}
-        if cert_file and cert_password:
-            try:
-                cert = certificado.Certificado(cert_file, cert_password)
-            except Exception as e:
-                raise ValidationError(
-                    _("Cannot load Certificate ! \n\n {}").format(e)
-                ) from e
+    owner_cnpj_cpf = fields.Char(
+        string="CNPJ/CPF",
+        compute="_compute_owner_cnpj_cpf",
+        store=True,
+    )
 
-            if cert:
-                values["issuer_name"] = cert.emissor
-                values["owner_name"] = cert.proprietario
-                values["owner_cnpj_cpf"] = cert.cnpj_cpf
-                if cert.fim_validade:
-                    values["date_expiration"] = cert.fim_validade.strftime(
-                        DEFAULT_SERVER_DATETIME_FORMAT
-                    )
+    issuer_name = fields.Char(
+        string="Issuer",
+        compute="_compute_issuer_name",
+        store=True,
+    )
 
-                if cert.inicio_validade:
-                    values["date_start"] = cert.inicio_validade.strftime(
-                        DEFAULT_SERVER_DATETIME_FORMAT
+    @api.depends("subject_common_name")
+    def _compute_owner_cnpj_cpf(self):
+        for certificate in self:
+            cnpj_cpf = ""
+            subject = certificate.subject_common_name or ""
+            if ":" in subject:
+                # Brazilian certificates carry the CNPJ/CPF in the subject CN
+                # after the last colon, e.g. "NOME DA EMPRESA:12345678000190".
+                cnpj_cpf = subject.rsplit(":", 1)[1]
+            certificate.owner_cnpj_cpf = cnpj_cpf
+
+    @api.depends("pem_certificate")
+    def _compute_issuer_name(self):
+        for certificate in self:
+            issuer_name = ""
+            pem_certificate = certificate.with_context(bin_size=False).pem_certificate
+            if pem_certificate:
+                with suppress(ValueError, TypeError):
+                    x509_cert = x509.load_pem_x509_certificate(
+                        base64.b64decode(pem_certificate)
                     )
+                    issuer_name = self._get_common_name(x509_cert, issuer=True) or ""
+            certificate.issuer_name = issuer_name
 
-        return values
-
-    @api.constrains("file", "password")
-    def _check_certificate(self):
-        for c in self:
-            cert_values = c._certificate_data(c.file, c.password)
-            if not cert_values:
-                raise ValidationError(_("Cannot load Certificate !"))
-
-    @api.depends("file", "password")
+    @api.depends("type", "subtype", "subject_common_name", "date_end")
     def _compute_name(self):
-        for cert in self:
-            name = False
-            file_name = False
-            if cert.file and cert.password:
-                name = "{} - {} - {} - Valid: {}".format(
-                    cert.type and cert.type.upper() or "",
-                    cert.subtype and cert.subtype.upper() or "",
-                    cert.owner_name or "",
-                    format_date(self.env, cert.date_expiration.date())
-                    if cert.date_expiration
-                    else "",
-                )
-                file_name = name + ".p12"
-
-            cert.name = name
-            cert.file_name = file_name
-
-    def update_certificate_data(self, values):
-        cert_file = values.get("file")
-        if isinstance(cert_file, str):
-            cert_file = cert_file.encode()
-        values.update(self._certificate_data(cert_file, values.get("password")))
-        return values
-
-    @api.depends("date_expiration")
-    def _compute_is_valid(self):
-        for c in self:
-            c.is_valid = False
-            if c.date_expiration:
-                c.is_valid = c.date_expiration >= fields.Datetime.now()
-
-    @api.model_create_multi
-    def create(self, vals_list):
-        for vals in vals_list:
-            self.update_certificate_data(vals)
-        return super().create(vals_list)
-
-    def write(self, values):
-        values = self.update_certificate_data(values)
-        return super().write(values)
-
-    @api.onchange("file", "password")
-    def _onchange_file_password(self):
-        if self.file and self.password:
-            self.update(
-                self.update_certificate_data(
-                    {"file": self.file, "password": self.password}
+        for certificate in self:
+            parts = []
+            if certificate.type:
+                parts.append(certificate.type.upper())
+            if certificate.subtype:
+                parts.append(certificate.subtype.upper())
+            if certificate.subject_common_name:
+                parts.append(certificate.subject_common_name)
+            if certificate.date_end:
+                parts.append(
+                    f"Valid: {format_date(self.env, certificate.date_end.date())}"
                 )
-            )
+            certificate.name = " - ".join(parts)