f5fed013ab
All checks were successful
Build and deploy / Validate source (push) Successful in 4s
Build and deploy / Integration suite on a real stack (push) Successful in 2m13s
Build and deploy / Secret scan and release gate (push) Successful in 5s
Build and deploy / Publish images (push) Successful in 1m9s
A credit card payment with the test credentials was refused with 10113
("the payment method is excluded by a rule"). Every card was sent with
three_d_secure_mode, which only debit needs, and with the order's CNPJ as
payer instead of the cardholder's document from the card form. Debit
methods keep 3-D Secure, and the card form's document is the payer.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
197 lines
10 KiB
Python
197 lines
10 KiB
Python
"""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=<unix>,v1=<hex>"): HMAC-SHA256, keyed with the
|
|
webhook secret from the integration panel, over the manifest
|
|
"id:<data.id>;request-id:<x-request-id>;ts:<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':
|
|
body.update(token=method['token'], payment_method_id=method['payment_method_id'],
|
|
installments=int(method.get('installments', 1)))
|
|
# 3-D Secure (the bank's confirmation) only for debit, which needs
|
|
# it; asking it of every card was refused as a rule (10113).
|
|
if method['payment_method_id'].startswith('deb'):
|
|
body['three_d_secure_mode'] = 'optional'
|
|
if method.get('payer_document_type') and method.get('payer_document'):
|
|
body['payer']['identification'] = {'type': method['payer_document_type'],
|
|
'number': method['payer_document']}
|
|
# 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})
|