357 lines
16 KiB
Python
357 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Página pública de pago de un enlace de cobro (sin sesión).
|
|
|
|
Endpoints anónimos bajo `/skeen/frontend/v1/public/cobro`:
|
|
|
|
GET /cobro?token=... datos mínimos y no sensibles del enlace
|
|
POST /cobro/pagar confirma el pago (simulado o pasarela real)
|
|
POST /cobro/webhook/stripe notificación de Stripe (firma verificada)
|
|
POST /cobro/webhook/mercadopago notificación de MercadoPago (secreto/token)
|
|
|
|
Reglas de oro:
|
|
- El token es HMAC firmado con caducidad; sin token válido no se expone nada.
|
|
- El pago en modo `simulado` SOLO se permite si el ajuste correspondiente
|
|
(`reservas_online.anticipo.permitir_simulado` o
|
|
`membresias.cobro_automatico.permitir_simulado`) está encendido. Nunca se
|
|
simula un pago en producción con la configuración por defecto.
|
|
- Límite de peticiones por IP y trampa antiautómatas (honeypot) como en el
|
|
portal de reservas.
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import threading
|
|
import time
|
|
|
|
from odoo import http
|
|
from odoo.http import request, Response
|
|
|
|
from odoo.addons.skeen_citas.clinic_settings import (
|
|
get_settings as get_clinic_settings,
|
|
get_anticipo,
|
|
get_cobro_automatico,
|
|
brand_dict,
|
|
)
|
|
from odoo.addons.skeen_citas.glosario import resolve as resolver_glosario
|
|
|
|
PREFIX = '/skeen/frontend/v1/public/cobro'
|
|
|
|
RATE_LIMITS = {'get': (60, 60), 'pagar': (10, 60), 'webhook': (120, 60)}
|
|
_RATE_LOCK = threading.Lock()
|
|
_RATE_HITS = {}
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
_ERROR_INTERNO = 'No se pudo procesar la solicitud. Intenta de nuevo.'
|
|
|
|
|
|
def _json(data, status=200):
|
|
return Response(
|
|
json.dumps(data, default=str),
|
|
status=status,
|
|
mimetype='application/json',
|
|
headers={
|
|
'Access-Control-Allow-Origin': '*',
|
|
'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
|
|
'Access-Control-Allow-Headers': 'Content-Type, Stripe-Signature',
|
|
},
|
|
)
|
|
|
|
|
|
def _client_ip():
|
|
xff = request.httprequest.headers.get('X-Forwarded-For') or ''
|
|
if xff:
|
|
return xff.split(',')[0].strip()
|
|
return request.httprequest.remote_addr or 'desconocida'
|
|
|
|
|
|
def _rate_limited(bucket):
|
|
limit, window = RATE_LIMITS.get(bucket, (30, 60))
|
|
key = (_client_ip(), bucket)
|
|
ahora = time.time()
|
|
with _RATE_LOCK:
|
|
hits = [t for t in _RATE_HITS.get(key, []) if ahora - t < window]
|
|
if len(hits) >= limit:
|
|
_RATE_HITS[key] = hits
|
|
return True
|
|
hits.append(ahora)
|
|
_RATE_HITS[key] = hits
|
|
return False
|
|
|
|
|
|
class SkeenPublicCobroController(http.Controller):
|
|
"""Página pública de pago de un enlace."""
|
|
|
|
def _service(self):
|
|
return request.env['skeen.cobro.service'].sudo()
|
|
|
|
def _settings(self):
|
|
return get_clinic_settings(request.env)
|
|
|
|
def _brand(self):
|
|
guardado = brand_dict(request.env)
|
|
out = {
|
|
'nombre': self._settings().get('nombre') or 'Clínica',
|
|
'logo_claro': '/brand/logo-blanco.svg',
|
|
'logo_oscuro': '/brand/logo-negro.svg',
|
|
'colores': {'primario': '', 'acento': ''},
|
|
}
|
|
if isinstance(guardado, dict):
|
|
if guardado.get('nombre'):
|
|
out['nombre'] = str(guardado['nombre'])
|
|
for k in ('logo_claro', 'logo_oscuro'):
|
|
if guardado.get(k):
|
|
out[k] = str(guardado[k])
|
|
colores = guardado.get('colores')
|
|
if isinstance(colores, dict):
|
|
for c in ('primario', 'acento'):
|
|
if colores.get(c):
|
|
out['colores'][c] = str(colores[c])
|
|
return out
|
|
|
|
def _simulado_permitido(self, enlace):
|
|
try:
|
|
if enlace.origen == 'reserva':
|
|
return bool(get_anticipo(request.env).get('permitir_simulado'))
|
|
if enlace.origen == 'membresia':
|
|
return bool(get_cobro_automatico(request.env).get('permitir_simulado'))
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
@staticmethod
|
|
def _estado_label(enlace):
|
|
return dict(enlace._fields['estado'].selection).get(enlace.estado, enlace.estado)
|
|
|
|
def _enlace_de_token(self, token):
|
|
return self._service().verificar_token(token)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 1. Datos del enlace (mínimos, no sensibles)
|
|
# ------------------------------------------------------------------
|
|
@http.route(PREFIX, type='http', auth='none', methods=['GET', 'OPTIONS'], csrf=False)
|
|
def cobro_get(self, **kw):
|
|
if request.httprequest.method == 'OPTIONS':
|
|
return _json({}, 200)
|
|
if _rate_limited('get'):
|
|
return _json({'status': 'error',
|
|
'message': 'Demasiadas solicitudes. Intenta más tarde.'}, 429)
|
|
try:
|
|
enlace = self._enlace_de_token(kw.get('token'))
|
|
if not enlace:
|
|
return _json({'status': 'error',
|
|
'message': 'Enlace no válido o expirado'}, 404)
|
|
settings = self._settings()
|
|
_pack, terms = resolver_glosario(request.env)
|
|
proveedor = self._service().proveedor_disponible()
|
|
puede_pagar = (enlace.estado == 'pendiente' and
|
|
(proveedor != 'simulado' or self._simulado_permitido(enlace)))
|
|
return _json({
|
|
'status': 'success',
|
|
'folio': enlace.name,
|
|
'concepto': enlace.concepto,
|
|
'monto': enlace.monto,
|
|
'moneda': enlace.currency_id.name or 'MXN',
|
|
'estado': enlace.estado,
|
|
'estado_label': self._estado_label(enlace),
|
|
'origen': enlace.origen,
|
|
'proveedor': enlace.proveedor,
|
|
'expira': enlace.token_expira,
|
|
'fecha_pago': enlace.fecha_pago,
|
|
'paciente': (enlace.partner_id.name or '').split(' ')[0] if enlace.partner_id else '',
|
|
'puede_pagar': bool(puede_pagar),
|
|
'simulado': enlace.proveedor == 'simulado',
|
|
'url_externa': enlace.url_externa or '',
|
|
'brand': self._brand(),
|
|
'terms': terms,
|
|
'zona_horaria': settings.get('zona_horaria') or 'America/Tijuana',
|
|
})
|
|
except Exception as exc:
|
|
_logger.exception('Error consultando enlace de cobro: %s', exc)
|
|
return _json({'status': 'error', 'message': _ERROR_INTERNO}, 500)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 2. Pagar
|
|
# ------------------------------------------------------------------
|
|
@http.route(PREFIX + '/pagar', type='http', auth='none',
|
|
methods=['POST', 'OPTIONS'], csrf=False)
|
|
def cobro_pagar(self, **kw):
|
|
if request.httprequest.method == 'OPTIONS':
|
|
return _json({}, 200)
|
|
if _rate_limited('pagar'):
|
|
return _json({'status': 'error',
|
|
'message': 'Demasiadas solicitudes. Intenta más tarde.'}, 429)
|
|
try:
|
|
data = json.loads(request.httprequest.get_data(as_text=True) or '{}')
|
|
except Exception:
|
|
data = {}
|
|
if str(data.get('hp') or '').strip():
|
|
return _json({'status': 'error', 'message': 'Solicitud rechazada'}, 400)
|
|
try:
|
|
enlace = self._enlace_de_token(data.get('token') or kw.get('token'))
|
|
if not enlace:
|
|
return _json({'status': 'error',
|
|
'message': 'Enlace no válido o expirado'}, 404)
|
|
if enlace.estado == 'pagado':
|
|
return _json({'status': 'error',
|
|
'message': 'Este enlace ya fue pagado.'}, 409)
|
|
if enlace.estado == 'expirado':
|
|
return _json({'status': 'error',
|
|
'message': 'Este enlace expiró.'}, 410)
|
|
if enlace.estado == 'cancelado':
|
|
return _json({'status': 'error',
|
|
'message': 'Este enlace fue cancelado.'}, 409)
|
|
|
|
proveedor = self._service().proveedor_disponible()
|
|
if proveedor != 'simulado':
|
|
# Camino real: se crea/recupera el checkout de la pasarela.
|
|
if not enlace.url_externa:
|
|
self._service().intentar_checkout(enlace)
|
|
if enlace.url_externa:
|
|
return _json({'status': 'success', 'requiere_pasarela': True,
|
|
'url': enlace.url_externa, 'proveedor': proveedor})
|
|
return _json({'status': 'error',
|
|
'message': 'La pasarela no está disponible. Contacta a la clínica.'}, 502)
|
|
|
|
# Camino simulado (sin pasarela).
|
|
if not self._simulado_permitido(enlace):
|
|
return _json({'status': 'error',
|
|
'message': 'El pago en línea no está disponible. '
|
|
'Contacta a la clínica.'}, 403)
|
|
pago = self._service().confirmar(
|
|
enlace, proveedor='simulado', referencia='SIMULADO-%s' % enlace.id,
|
|
response='Pago simulado desde la página pública', user='Portal en línea')
|
|
if not pago:
|
|
return _json({'status': 'error',
|
|
'message': 'No se pudo confirmar el pago.'}, 409)
|
|
return _json({
|
|
'status': 'success',
|
|
'ok': True,
|
|
'estado': 'pagado',
|
|
'estado_label': self._estado_label(enlace),
|
|
'comprobante': {
|
|
'folio': enlace.name,
|
|
'concepto': enlace.concepto,
|
|
'monto': enlace.monto,
|
|
'moneda': enlace.currency_id.name or 'MXN',
|
|
'fecha': enlace.fecha_pago,
|
|
'proveedor': 'simulado',
|
|
'pago': pago.name,
|
|
},
|
|
})
|
|
except Exception as exc:
|
|
request.env.cr.rollback()
|
|
_logger.exception('Error pagando enlace de cobro: %s', exc)
|
|
return _json({'status': 'error', 'message': _ERROR_INTERNO}, 500)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 3. Webhooks (notificación de la pasarela)
|
|
# ------------------------------------------------------------------
|
|
@http.route(PREFIX + '/webhook/stripe', type='http', auth='none',
|
|
methods=['POST', 'OPTIONS'], csrf=False)
|
|
def webhook_stripe(self, **kw):
|
|
if request.httprequest.method == 'OPTIONS':
|
|
return _json({}, 200)
|
|
secret = request.env['ir.config_parameter'].sudo().get_param(
|
|
'skeen.stripe_webhook_secret')
|
|
raw = request.httprequest.get_data() or b''
|
|
if not secret:
|
|
return _json({'status': 'error',
|
|
'message': 'Webhook de Stripe no configurado'}, 400)
|
|
firma = request.httprequest.headers.get('Stripe-Signature') or ''
|
|
if not self._stripe_firma_valida(raw, firma, secret):
|
|
return _json({'status': 'error', 'message': 'Firma inválida'}, 400)
|
|
try:
|
|
evento = json.loads(raw.decode('utf-8') or '{}')
|
|
except Exception:
|
|
return _json({'status': 'error', 'message': 'Cuerpo inválido'}, 400)
|
|
tipo = evento.get('type') or ''
|
|
obj = ((evento.get('data') or {}).get('object') or {})
|
|
if tipo in ('checkout.session.completed', 'payment_intent.succeeded'):
|
|
meta = obj.get('metadata') or {}
|
|
enlace_id = int(meta.get('enlace_id') or 0)
|
|
enlace = request.env['skeen.cobro.enlace'].sudo().browse(enlace_id)
|
|
if enlace.exists():
|
|
self._service().confirmar(
|
|
enlace, proveedor='stripe',
|
|
referencia=obj.get('id') or obj.get('payment_intent') or '',
|
|
response=json.dumps(obj)[:2000], user='Webhook Stripe')
|
|
return _json({'status': 'success'})
|
|
|
|
@staticmethod
|
|
def _stripe_firma_valida(raw, header, secret):
|
|
partes = {}
|
|
for item in (header or '').split(','):
|
|
if '=' in item:
|
|
k, v = item.split('=', 1)
|
|
partes.setdefault(k.strip(), v.strip())
|
|
t = partes.get('t')
|
|
v1 = partes.get('v1')
|
|
if not t or not v1:
|
|
return False
|
|
firmado = ('%s.%s' % (t, raw.decode('utf-8'))).encode('utf-8')
|
|
esperada = hmac.new(secret.encode('utf-8'), firmado, hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(esperada, v1)
|
|
|
|
@http.route(PREFIX + '/webhook/mercadopago', type='http', auth='none',
|
|
methods=['POST', 'OPTIONS'], csrf=False)
|
|
def webhook_mercadopago(self, **kw):
|
|
if request.httprequest.method == 'OPTIONS':
|
|
return _json({}, 200)
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
token = ICP.get_param('skeen.mercadopago_access_token')
|
|
secret = ICP.get_param('skeen.mercadopago_webhook_secret')
|
|
raw = request.httprequest.get_data() or b''
|
|
try:
|
|
cuerpo = json.loads(raw.decode('utf-8') or '{}')
|
|
except Exception:
|
|
cuerpo = {}
|
|
data_id = (cuerpo.get('data') or {}).get('id') or request.httprequest.args.get('data.id') or ''
|
|
tipo = cuerpo.get('type') or request.httprequest.args.get('type') or ''
|
|
if not token and not secret:
|
|
return _json({'status': 'error',
|
|
'message': 'Webhook de MercadoPago no configurado'}, 400)
|
|
# Verificación por secreto (x-signature) cuando está configurado.
|
|
if secret:
|
|
firma = request.httprequest.headers.get('x-signature') or ''
|
|
if not self._mp_firma_valida(request.httprequest.args.get('data.id') or '', firma, secret):
|
|
return _json({'status': 'error', 'message': 'Firma inválida'}, 400)
|
|
elif tipo != 'payment' or not data_id:
|
|
return _json({'status': 'error', 'message': 'Notificación inválida'}, 400)
|
|
# Con token, se confirma el pago real consultando la API de MercadoPago.
|
|
if tipo == 'payment' and data_id and token:
|
|
try:
|
|
import requests
|
|
resp = requests.get('https://api.mercadopago.com/v1/payments/%s' % data_id,
|
|
headers={'Authorization': 'Bearer %s' % token}, timeout=30)
|
|
pago = resp.json() if resp.status_code == 200 else {}
|
|
estado = pago.get('status')
|
|
ext = (pago.get('metadata') or {}).get('enlace_id')
|
|
if estado == 'approved' and ext:
|
|
enlace = request.env['skeen.cobro.enlace'].sudo().browse(int(ext))
|
|
if enlace.exists():
|
|
self._service().confirmar(
|
|
enlace, proveedor='mercadopago', referencia=str(data_id),
|
|
response=json.dumps(pago)[:2000], user='Webhook MercadoPago')
|
|
except Exception as exc:
|
|
_logger.warning('Webhook MercadoPago: %s', exc)
|
|
return _json({'status': 'error', 'message': 'No se pudo verificar'}, 502)
|
|
return _json({'status': 'success'})
|
|
|
|
@staticmethod
|
|
def _mp_firma_valida(data_id, header, secret):
|
|
partes = {}
|
|
for item in (header or '').split(','):
|
|
if '=' in item:
|
|
k, v = item.split('=', 1)
|
|
partes[k.strip()] = v.strip()
|
|
ts = partes.get('ts')
|
|
v1 = partes.get('v1')
|
|
if not ts or not v1:
|
|
return False
|
|
manifiesto = 'id:%s;ts:%s;' % (data_id, ts)
|
|
esperada = hmac.new(secret.encode('utf-8'), manifiesto.encode('utf-8'),
|
|
hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(esperada, v1)
|