[IMP] certificate, l10n_es_edi_*: Factor patched HTTPAdapter for certificates models

We generalize the CertificateAdapter that is needed to make zeep read
our own Certificate model for the session.

task-4477745

closes odoo#210572

Signed-off-by: Ricardo Gomes Rodrigues (rigr) <[email protected]>

Author: lordkrandel

addons/certificate/tools/__init__.py

@@ -0,0 +1 @@
+from .certificate_adapter import CertificateAdapter

addons/certificate/tools/certificate_adapter.py

@@ -0,0 +1,50 @@
+from base64 import b64decode
+
+import requests
+from OpenSSL.crypto import FILETYPE_PEM, load_certificate, load_privatekey
+from urllib3.contrib.pyopenssl import inject_into_urllib3
+from urllib3.util.ssl_ import create_urllib3_context
+
+
+class CertificateAdapter(requests.adapters.HTTPAdapter):
+
+    def __init__(self, *args, ciphers=None, **kwargs):
+        self._context_args = {}
+        if ciphers:
+            self._context_args['ciphers'] = ciphers
+        super().__init__(*args, **kwargs)
+
+    def init_poolmanager(self, *args, **kwargs):
+        """ We need inject_into_urllib3 as it forces the adapter to use PyOpenSSL.
+            With PyOpenSSL, we can further patch the code to make it do what we want
+            (with the use of SSLContext)
+        """
+        # OVERRIDE
+        inject_into_urllib3()
+        kwargs['ssl_context'] = create_urllib3_context(**self._context_args)
+        return super().init_poolmanager(*args, **kwargs)
+
+    def cert_verify(self, conn, url, verify, cert):
+        """ The original method wants to check for an existing file
+            at the cert location. As we use in-memory objects,
+            we skip the check and assign it manually.
+        """
+        # OVERRIDE
+        super().cert_verify(conn, url, verify, None)
+        conn.cert_file = cert
+        conn.key_file = None
+
+    def get_connection(self, url, proxies=None):
+        """ Reads the certificate from a certificate.certificate rather than from the filesystem """
+        # OVERRIDE
+        conn = super().get_connection(url, proxies=proxies)
+        context = conn.conn_kw['ssl_context']
+
+        def patched_load_cert_chain(certificate, keyfile=None, password=None):
+            certificate = certificate.sudo()
+            pem, key = map(b64decode, (certificate.pem_certificate, certificate.private_key_id.pem_key))
+            context._ctx.use_certificate(load_certificate(FILETYPE_PEM, pem))
+            context._ctx.use_privatekey(load_privatekey(FILETYPE_PEM, key))
+
+        context.load_cert_chain = patched_load_cert_chain
+        return conn

addons/l10n_es_edi_sii/models/account_edi_format.py

@@ -1,64 +1,21 @@
-# -*- coding: utf-8 -*-
-# Part of Odoo. See LICENSE file for full copyright and licensing details.
+import json
+import math
 from collections import defaultdict
-from urllib3.util.ssl_ import create_urllib3_context
-from urllib3.contrib.pyopenssl import inject_into_urllib3
-from OpenSSL.crypto import load_certificate, load_privatekey, FILETYPE_PEM
 
-from odoo import fields, models, _
+import requests
+
+from odoo import _, fields, models
 from odoo.exceptions import UserError
 from odoo.tools import html_escape, zeep
 from odoo.tools.float_utils import float_round
 
-import base64
-import math
-import json
-import requests
-
+from odoo.addons.certificate.tools import CertificateAdapter
 
 # Custom patches to perform the WSDL requests.
 # Avoid failure on servers where the DH key is too small
 EUSKADI_CIPHERS = "DEFAULT:!DH"
 
 
-class PatchedHTTPAdapter(requests.adapters.HTTPAdapter):
-    """ An adapter to block DH ciphers which may not work for the tax agencies called"""
-
-    def init_poolmanager(self, *args, **kwargs):
-        # OVERRIDE
-        inject_into_urllib3()
-        kwargs['ssl_context'] = create_urllib3_context(ciphers=EUSKADI_CIPHERS)
-        return super().init_poolmanager(*args, **kwargs)
-
-    def cert_verify(self, conn, url, verify, cert):
-        # OVERRIDE
-        # The last parameter is only used by the super method to check if the file exists.
-        # In our case, cert is an odoo record 'certificate.certificate' so not a path to a file.
-        # By putting 'None' as last parameter, we ensure the check about TLS configuration is
-        # still made without checking temporary files exist.
-        super().cert_verify(conn, url, verify, None)
-        conn.cert_file = cert
-        conn.key_file = None
-
-    def get_connection(self, url, proxies=None):
-        # OVERRIDE
-        # Patch the OpenSSLContext to decode the certificate in-memory.
-        conn = super().get_connection(url, proxies=proxies)
-        context = conn.conn_kw['ssl_context']
-
-        def patched_load_cert_chain(l10n_es_odoo_certificate, keyfile=None, password=None):
-            certificate = l10n_es_odoo_certificate
-            cert_obj = load_certificate(FILETYPE_PEM, base64.b64decode(certificate.sudo().pem_certificate))
-            pkey_obj = load_privatekey(FILETYPE_PEM, base64.b64decode(certificate.sudo().private_key_id.pem_key))
-
-            context._ctx.use_certificate(cert_obj)
-            context._ctx.use_privatekey(pkey_obj)
-
-        context.load_cert_chain = patched_load_cert_chain
-
-        return conn
-
-
 class AccountEdiFormat(models.Model):
     _inherit = 'account.edi.format'
 
@@ -511,7 +468,7 @@ def _l10n_es_edi_call_web_service_sign_common(self, invoices, info_list, cancel=
 
         session = requests.Session()
         session.cert = company.l10n_es_sii_certificate_id
-        session.mount('https://', PatchedHTTPAdapter())
+        session.mount('https://', CertificateAdapter(ciphers=EUSKADI_CIPHERS))
 
         client = zeep.Client(connection_vals['url'], operation_timeout=60, timeout=60, session=session)
 

addons/l10n_es_edi_tbai/models/l10n_es_edi_tbai_document.py

@@ -11,7 +11,7 @@
 from requests.exceptions import RequestException
 
 from odoo import _, api, fields, models, release
-from odoo.addons.l10n_es_edi_sii.models.account_edi_format import PatchedHTTPAdapter
+from odoo.addons.certificate.tools import CertificateAdapter
 from odoo.addons.l10n_es_edi_tbai.models.l10n_es_edi_tbai_agencies import get_key
 from odoo.addons.l10n_es_edi_tbai.models.xml_utils import (
     NS_MAP,
@@ -190,7 +190,7 @@ def _post_to_agency(self, env, is_sale):
         def _send_request_to_agency(*args, **kwargs):
             session = requests.Session()
             session.cert = kwargs.pop('pkcs12_data')
-            session.mount("https://", PatchedHTTPAdapter())
+            session.mount("https://", CertificateAdapter())
             response = session.request('post', *args, **kwargs)
             response.raise_for_status()
             response_xml = None