"""Mercado Pago: payment creation, webhook verification and status lookup. Written from the public API documentation and exercised only against a fake HTTP transport. It is not a verified integration until it has passed the sandbox flows in docs/PRODUCTION_INPUTS.md with the client's own account; until then the runtime refuses to select it without explicit credentials. The notification is only a pointer. Its body says "payment 123 changed" and nothing about amount or status, so nothing in it is trusted beyond the id: the payment is fetched from the API with our access token, and that response is what the order service compares against the approved quote. Signature (x-signature: "ts=,v1="): HMAC-SHA256, keyed with the webhook secret from the integration panel, over the manifest "id:;request-id:;ts:;", where data.id comes from the notification URL's query string (lower-cased when alphanumeric). A part whose value is absent from the notification is left out of the manifest. """ import hashlib import hmac import json import os import time from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from typing import Mapping import httpx from .adapters import PaymentEvent API = 'https://api.mercadopago.com' # How old a signed timestamp may be. Mercado Pago retries a failed delivery # every 15 minutes, re-signing each attempt, so a replayed old one is refused. MAX_SIGNATURE_AGE = 30 * 60 STATUSES = {'approved': 'approved', 'pending': 'pending', 'in_process': 'pending', 'authorized': 'pending', 'in_mediation': 'pending', 'rejected': 'rejected', 'cancelled': 'cancelled', 'refunded': 'refunded', 'charged_back': 'refunded'} # A PIX code stops working after this; Mercado Pago then cancels the payment. PIX_MINUTES = 30 BRASILIA = timezone(timedelta(hours=-3)) class MercadoPagoPayment: name = 'mercadopago' def __init__(self, access_token=None, webhook_secret=None, notification_url=None, transport=None, clock=time.time): self.access_token = access_token or os.environ.get('MP_ACCESS_TOKEN', '') self.webhook_secret = (webhook_secret or os.environ.get('MP_WEBHOOK_SECRET', '')).encode() self.notification_url = notification_url or os.environ.get('MP_NOTIFICATION_URL', '') if not self.access_token or not self.webhook_secret: raise RuntimeError('Mercado Pago needs MP_ACCESS_TOKEN and MP_WEBHOOK_SECRET') self.http = httpx.Client(base_url=API, transport=transport, timeout=15, headers={'Authorization': f'Bearer {self.access_token}'}) self.clock = clock # Payments ------------------------------------------------------------ def create(self, quote_id: str, total_cents: int, customer: dict, method: dict | None = None) -> dict: """Create a payment for an approved quote. The quote id is the idempotency key, so a retry after a timeout returns the payment already created rather than charging again. `method` is {'type': 'pix'} or {'type': 'card', 'token', 'payment_method_id', 'installments', 'issuer_id'} from Mercado Pago's card form: card data is tokenised in the customer's browser and never reaches this server. """ method = method or {'type': 'pix'} body = {'transaction_amount': float(Decimal(total_cents) / 100), 'description': f'DTF - cotação {quote_id[:8]}', 'external_reference': quote_id, 'payer': {'email': customer['mail'], 'identification': {'type': 'CNPJ', 'number': customer['cnpj']}}} if self.notification_url: body['notification_url'] = self.notification_url expires_at = None if method['type'] == 'pix': body['payment_method_id'] = 'pix' expires_at = (datetime.now(BRASILIA) + timedelta(minutes=PIX_MINUTES)).isoformat(timespec='milliseconds') body['date_of_expiration'] = expires_at elif method['type'] == 'card': # The issuer decides whether the cardholder must confirm in the # bank's app or page (3-D Secure); debit cards usually must. body.update(token=method['token'], payment_method_id=method['payment_method_id'], installments=int(method.get('installments', 1)), three_d_secure_mode='optional') # No issuer_id: Mercado Pago takes the issuer from the card number. # The form's suggestion was refused for its own test cards # (10111, "the issuer does not have the BIN configured"). else: raise ValueError('Unsupported payment method') # A PIX retry must return the same code; a new one, after the last # expired, is the next attempt. A card retry after a decline is a new # attempt with a new token, so the token is part of the key; the intent # route refuses new attempts once one is approved or in review. key = f"dtf-quote-{quote_id}-pix-{int(method.get('attempt', 1))}" if method['type'] == 'pix' else \ f"dtf-quote-{quote_id}-card-{hashlib.sha256(method['token'].encode()).hexdigest()[:24]}" response = self.http.post('/v1/payments', json=body, headers={'X-Idempotency-Key': key}) response.raise_for_status() payment = response.json() transaction = (payment.get('point_of_interaction') or {}).get('transaction_data') or {} # A payment waiting for 3-D Secure carries the bank's challenge page, # which the Site shows in a frame by posting `creq` to that address. three_ds = payment.get('three_ds_info') or {} challenge = ({'url': three_ds['external_resource_url'], 'creq': three_ds['creq']} if payment.get('status_detail') == 'pending_challenge' and three_ds.get('external_resource_url') and three_ds.get('creq') else None) return {'provider': self.name, 'id': str(payment['id']), 'status': STATUSES.get(payment.get('status'), 'pending'), 'status_detail': payment.get('status_detail'), 'total_cents': total_cents, 'pix_qr_code': transaction.get('qr_code'), 'pix_qr_code_base64': transaction.get('qr_code_base64'), 'ticket_url': transaction.get('ticket_url'), 'expires_at': payment.get('date_of_expiration') or expires_at, 'challenge': challenge} def lookup(self, payment_id: str) -> dict: response = self.http.get(f'/v1/payments/{payment_id}') response.raise_for_status() return response.json() # Webhooks ------------------------------------------------------------ def verify(self, headers: Mapping[str, str], body: bytes, query: Mapping[str, str] | None = None) -> bool: signature = headers.get('x-signature') or '' parts = dict(part.strip().split('=', 1) for part in signature.split(',') if '=' in part) ts, supplied = parts.get('ts', ''), parts.get('v1', '') if not ts.isdigit() or not supplied: return False if abs(self.clock() - int(ts[:10])) > MAX_SIGNATURE_AGE: return False data_id = (query or {}).get('data.id', '') if data_id.isalnum(): data_id = data_id.lower() manifest = '' if data_id: manifest += f'id:{data_id};' request_id = headers.get('x-request-id') or '' if request_id: manifest += f'request-id:{request_id};' manifest += f'ts:{ts};' expected = hmac.new(self.webhook_secret, manifest.encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(expected, supplied.lower()) def parse(self, body: bytes, query: Mapping[str, str] | None = None) -> PaymentEvent | None: try: data = json.loads(body) except ValueError: return None if not isinstance(data, dict) or data.get('type') != 'payment': return None payment_id = str((query or {}).get('data.id') or (data.get('data') or {}).get('id') or '') if not payment_id.isdigit(): return None try: payment = self.lookup(payment_id) except httpx.HTTPStatusError as error: # Signed by Mercado Pago but about no payment of ours, such as the # panel's "Simular notificação". Anything else is raised so that a # real notification is retried. if error.response.status_code == 404: return None raise return event_from_payment(payment, notification_id=str(data.get('id', ''))) def event_from_payment(payment: dict, notification_id: str = '') -> PaymentEvent: """Normalise a payment fetched from the API. Amount is None unless it is BRL and a whole number of centavos, so a foreign or malformed amount is refused.""" amount = None if payment.get('currency_id') == 'BRL': try: cents = Decimal(str(payment.get('transaction_amount'))) * 100 if cents == cents.to_integral_value(): amount = int(cents) except (InvalidOperation, TypeError, ValueError): amount = None status = STATUSES.get(payment.get('status'), 'pending') # One event per payment state: a notification id alone would let the same # approval be applied twice under two notifications, and would collapse a # later refund into the earlier approval. event_id = f"{payment.get('id')}:{payment.get('status')}" return PaymentEvent(event_id=event_id, reference=str(payment.get('external_reference') or ''), status=status, amount_cents=amount, raw={'provider': 'mercadopago', 'payment_id': str(payment.get('id')), 'status': payment.get('status'), 'status_detail': payment.get('status_detail'), 'currency_id': payment.get('currency_id'), 'transaction_amount': payment.get('transaction_amount'), 'date_approved': payment.get('date_approved'), 'notification_id': notification_id})