# -*- 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)