[ADD] pos_mercado_pago: integration of Mercado Pago payment terminal

This PR add the Mercado Pago "Smart Point" terminal
dedicated to the LATAM (Latin America) region

task-3350386

mool

closes #154962

Signed-off-by: Quentin Lejeune (qle) <qle@odoo.com>

addons/pos_mercado_pago/__init__.py

@@ -0,0 +1,2 @@
+from . import models
+from . import controllers

addons/pos_mercado_pago/__manifest__.py

@@ -0,0 +1,18 @@
+{
+    'name': 'POS Mercado Pago',
+    'version': '1.0',
+    'category': 'Sales/Point of Sale',
+    'sequence': 6,
+    'summary': 'Integrate your POS with the Mercado Pago Smart Point terminal',
+    'data': [
+        'views/pos_payment_method_views.xml',
+    ],
+    'depends': ['point_of_sale'],
+    'installable': True,
+    'assets': {
+        'point_of_sale._assets_pos': [
+            'pos_mercado_pago/static/**/*',
+        ],
+    },
+    'license': 'LGPL-3',
+}

addons/pos_mercado_pago/controllers/__init__.py

@@ -0,0 +1,3 @@
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+from . import main

addons/pos_mercado_pago/controllers/main.py

@@ -0,0 +1,84 @@
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+import hashlib
+import hmac
+import logging
+import re
+
+from odoo import http
+from odoo.http import request
+
+_logger = logging.getLogger(__name__)
+
+
+class PosMercadoPagoWebhook(http.Controller):
+    @http.route('/pos_mercado_pago/notification', methods=['POST'], type="http", auth="none", csrf=False)
+    def notification(self):
+        """ Process the notification sent by Mercado Pago
+
+        Notification format is always json
+        """
+        _logger.debug('POST message received on the end point')
+
+        # Check for mandatory keys in header
+        x_request_id = request.httprequest.headers.get('X-Request-Id')
+        if not x_request_id:
+            _logger.debug('POST message received with no X-Request-Id in header')
+            return http.Response(status=400)
+
+        x_signature = request.httprequest.headers.get('X-Signature')
+        if not x_signature:
+            _logger.debug('POST message received with no X-Signature in header')
+            return http.Response(status=400)
+
+        ts_m = re.search(r"ts=(\d+)", x_signature)
+        v1_m = re.search(r"v1=([a-f0-9]+)", x_signature)
+        ts = ts_m.group(1) if ts_m else None
+        v1 = v1_m.group(1) if v1_m else None
+        if not ts or not v1:
+            _logger.debug('Webhook bad X-Signature, ts: %s, v1: %s', ts, v1)
+            return http.Response(status=400)
+
+        # Check for payload
+        data = request.httprequest.get_json(silent=True)
+        if not data:
+            _logger.debug('POST message received with no data')
+            return http.Response(status=400)
+
+        # If and only if this webhook is related with a payment intend (see payment_mercado_pago.js)
+        # then the field data['additional_info']['external_reference'] contains a string
+        # formated like "XXX_YYY" where "XXX" is the session_id and "YYY" is the payment_method_id
+        external_reference = data.get('additional_info', {}).get('external_reference')
+
+        if not external_reference or not re.fullmatch(r'\d+_\d+', external_reference):
+            _logger.debug('POST message received with no or malformed "external_reference" key')
+            return http.Response(status=400)
+
+        session_id, payment_method_id = external_reference.split('_')
+
+        pos_session_sudo = request.env['pos.session'].sudo().browse(int(session_id))
+        if not pos_session_sudo or pos_session_sudo.state != 'opened':
+            _logger.error("Invalid session id: %s", session_id)
+            # This error is not related with Mercado Pago, simply acknowledge Mercado Pago message
+            return http.Response('OK', status=200)
+
+        payment_method_sudo = pos_session_sudo.config_id.payment_method_ids.filtered(lambda p: p.id == int(payment_method_id))
+        if not payment_method_sudo or payment_method_sudo.use_payment_terminal != 'mercado_pago':
+            _logger.error("Invalid payment method id: %s", payment_method_id)
+            # This error is not related with Mercado Pago, simply acknowledge Mercado Pago message
+            return http.Response('OK', status=200)
+
+        # We have to check if this comes from Mercado Pago with the secret key
+        secret_key = payment_method_sudo.mp_webhook_secret_key
+        signed_template = f"id:{data['id']};request-id:{x_request_id};ts:{ts};"
+        cyphed_signature = hmac.new(secret_key.encode(), signed_template.encode(), hashlib.sha256).hexdigest()
+        if not hmac.compare_digest(cyphed_signature, v1):
+            _logger.error('Webhook authenticating failure, ts: %s, v1: %s', ts, v1)
+            return http.Response(status=401)
+
+        _logger.debug('Webhook authenticated, POST message: %s', data)
+
+        # Notify the frontend that we received a message from Mercado Pago
+        request.env['bus.bus']._sendone(pos_session_sudo._get_bus_channel_name(), 'MERCADO_PAGO_LATEST_MESSAGE', {})
+
+        # Acknowledge Mercado Pago message
+        return http.Response('OK', status=200)

addons/pos_mercado_pago/models/__init__.py

@@ -0,0 +1,3 @@
+from . import mercado_pago_pos_request
+from . import pos_payment_method
+from . import pos_session

addons/pos_mercado_pago/models/mercado_pago_pos_request.py

@@ -0,0 +1,33 @@
+import logging
+import requests
+
+_logger = logging.getLogger(__name__)
+
+
+REQUEST_TIMEOUT = 10
+MERCADO_PAGO_API_ENDPOINT = 'https://api.mercadopago.com'
+
+
+class MercadoPagoPosRequest:
+    def __init__(self, mp_bearer_token):
+        self.mercado_pago_bearer_token = mp_bearer_token
+
+    def call_mercado_pago(self, method, endpoint, payload):
+        """ Make a request to Mercado Pago POS API.
+
+        :param method: "GET", "POST", ...
+        :param endpoint: The endpoint to be reached by the request.
+        :param payload: The payload of the request.
+        :return The JSON-formatted content of the response.
+        """
+        endpoint = MERCADO_PAGO_API_ENDPOINT + endpoint
+        header = {'Authorization': f"Bearer {self.mercado_pago_bearer_token}"}
+        try:
+            response = requests.request(method, endpoint, headers=header, json=payload, timeout=REQUEST_TIMEOUT)
+            return response.json()
+        except requests.exceptions.RequestException as error:
+            _logger.warning("Cannot connect with Mercado Pago POS. Error: %s", error)
+            return {'errorMessage': str(error)}
+        except ValueError as error:
+            _logger.warning("Cannot decode response json. Error: %s", error)
+            return {'errorMessage': f"Cannot decode Mercado Pago POS response. Error: {error}"}

addons/pos_mercado_pago/models/pos_payment_method.py

@@ -0,0 +1,116 @@
+import logging
+
+from odoo import fields, models, _
+from odoo.exceptions import AccessError, UserError
+
+from .mercado_pago_pos_request import MercadoPagoPosRequest
+
+_logger = logging.getLogger(__name__)
+
+
+class PosPaymentMethod(models.Model):
+    _inherit = 'pos.payment.method'
+
+    mp_bearer_token = fields.Char(
+        string="Production user token",
+        help='Mercado Pago customer production user token: https://www.mercadopago.com.mx/developers/en/reference',
+        groups="point_of_sale.group_pos_manager")
+    mp_webhook_secret_key = fields.Char(
+        string="Production secret key",
+        help='Mercado Pago production secret key from integration application: https://www.mercadopago.com.mx/developers/panel/app',
+        groups="point_of_sale.group_pos_manager")
+    mp_id_point_smart = fields.Char(
+        string="Terminal S/N",
+        help="Enter your Point Smart terminal serial number written on the back of your terminal (after the S/N:)")
+    mp_id_point_smart_complet = fields.Char()
+
+    def _get_payment_terminal_selection(self):
+        return super()._get_payment_terminal_selection() + [('mercado_pago', 'Mercado Pago')]
+
+    def force_pdv(self):
+        """
+        Triggered in debug mode when the user wants to force the "PDV" mode.
+        It calls the Mercado Pago API to set the terminal mode to "PDV".
+        """
+        if not self.env.user.has_group('point_of_sale.group_pos_user'):
+            raise AccessError(_("Do not have access to fetch token from Mercado Pago"))
+
+        mercado_pago = MercadoPagoPosRequest(self.sudo().mp_bearer_token)
+        _logger.info('Calling Mercado Pago to force the terminal mode to "PDV"')
+
+        mode = {"operating_mode": "PDV"}
+        resp = mercado_pago.call_mercado_pago("patch", f"/point/integration-api/devices/{self.mp_id_point_smart_complet}", mode)
+        if resp.get("operating_mode") != "PDV":
+            raise UserError(_("Unexpected Mercado Pago response: %s", resp))
+        _logger.debug("Successfully set the terminal mode to 'PDV'.")
+        return None
+
+    def mp_payment_intent_create(self, infos):
+        """
+        Called from frontend for creating a payment intent in Mercado Pago
+        """
+        if not self.env.user.has_group('point_of_sale.group_pos_user'):
+            raise AccessError(_("Do not have access to fetch token from Mercado Pago"))
+
+        mercado_pago = MercadoPagoPosRequest(self.sudo().mp_bearer_token)
+        # Call Mercado Pago for payment intend creation
+        resp = mercado_pago.call_mercado_pago("post", f"/point/integration-api/devices/{self.mp_id_point_smart_complet}/payment-intents", infos)
+        _logger.debug("mp_payment_intent_create(), response from Mercado Pago: %s", resp)
+        return resp
+
+    def mp_payment_intent_get(self, payment_intent_id):
+        """
+        Called from frontend to get the last payment intend from Mercado Pago
+        """
+        if not self.env.user.has_group('point_of_sale.group_pos_user'):
+            raise AccessError(_("Do not have access to fetch token from Mercado Pago"))
+
+        mercado_pago = MercadoPagoPosRequest(self.sudo().mp_bearer_token)
+        # Call Mercado Pago for payment intend status
+        resp = mercado_pago.call_mercado_pago("get", f"/point/integration-api/payment-intents/{payment_intent_id}", {})
+        _logger.debug("mp_payment_intent_get(), response from Mercado Pago: %s", resp)
+        return resp
+
+    def mp_payment_intent_cancel(self, payment_intent_id):
+        """
+        Called from frontend to cancel a payment intent in Mercado Pago
+        """
+        if not self.env.user.has_group('point_of_sale.group_pos_user'):
+            raise AccessError(_("Do not have access to fetch token from Mercado Pago"))
+
+        mercado_pago = MercadoPagoPosRequest(self.sudo().mp_bearer_token)
+        # Call Mercado Pago for payment intend cancelation
+        resp = mercado_pago.call_mercado_pago("delete", f"/point/integration-api/devices/{self.mp_id_point_smart_complet}/payment-intents/{payment_intent_id}", {})
+        _logger.debug("mp_payment_intent_cancel(), response from Mercado Pago: %s", resp)
+        return resp
+
+    def _find_terminal(self, token, point_smart):
+        mercado_pago = MercadoPagoPosRequest(token)
+        data = mercado_pago.call_mercado_pago("get", "/point/integration-api/devices", {})
+        if 'devices' in data:
+            # Search for a device id that contains the serial number entered by the user
+            found_device = next((device for device in data['devices'] if point_smart in device['id']), None)
+
+            if not found_device:
+                raise UserError(_("The terminal serial number is not registered on Mercado Pago"))
+
+            return found_device.get('id', '')
+        else:
+            raise UserError(_("Please verify your production user token as it was rejected"))
+
+    def write(self, vals):
+        records = super().write(vals)
+
+        if 'mp_id_point_smart' in vals or 'mp_bearer_token' in vals:
+            self.mp_id_point_smart_complet = self._find_terminal(self.mp_bearer_token, self.mp_id_point_smart)
+
+        return records
+
+    def create(self, vals):
+        records = super().create(vals)
+
+        for record in records:
+            if record.mp_bearer_token:
+                record.mp_id_point_smart_complet = record._find_terminal(record.mp_bearer_token, record.mp_id_point_smart)
+
+        return records

addons/pos_mercado_pago/models/pos_session.py

@@ -0,0 +1,11 @@
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+from odoo import models
+
+
+class PosSession(models.Model):
+    _inherit = 'pos.session'
+
+    def _loader_params_pos_payment_method(self):
+        result = super()._loader_params_pos_payment_method()
+        result['search_params']['fields'].append('mp_id_point_smart')
+        return result