SKEEN-Proyecto/odoo-addons/skeen_whatsapp/controllers/auth.py

# -*- coding: utf-8 -*-
"""Autenticación ligera (token HMAC) para el frontend React de SKEEN.

Token = base64url(payload_json) + "." + firma HMAC-SHA256, sin tabla de sesiones.
El secreto vive en ir.config_parameter 'skeen.frontend.secret' (se genera al primer uso).
"""

import base64
import functools
import hashlib
import hmac
import json
import secrets
import time

from odoo.http import request, Response

from ..roles import ROLE_RANKS, get_menu_roles, AGENDA_PERMS, resolve_agenda_perms

TTL_SECONDS = 60 * 60 * 12  # 12 horas
SECRET_KEY = 'skeen.frontend.secret'

RANK = ROLE_RANKS

# Módulos de dinero globales vetados para el rol 'medico' por API: su menú ya
# los oculta (MENU_ROLES) y aquí se cierra la vía directa con token. Su propio
# reporte (/my/sales) no pasa por esta lista.
MEDICO_BLOCKED_PREFIXES = (
    '/skeen/frontend/v1/reports/',
    '/skeen/frontend/v1/sales',
    '/skeen/frontend/v1/payments',
    '/skeen/frontend/v1/wallets',
    '/skeen/frontend/v1/pos/',
    '/skeen/frontend/v1/cash_closings',
    '/skeen/frontend/v1/doctors/',
)


def _medico_reporte_permitido(env, path):
    """True si la matriz de menus concede al medico ESA pestana de reportes.

    El veto de 'medico' sobre /reports/ es por diseno (no ve dinero), pero la
    matriz permite habilitar pestañas puntuales (/reportes?tab=horas-agenda).
    Sin este puente el menu mostraria la opcion y la API la rechazaria con 403.
    Las demas rutas vetadas (sales, payments, wallets...) siguen cerradas.
    """
    prefijo = '/skeen/frontend/v1/reports/'
    if not path.startswith(prefijo):
        return False
    tab = path[len(prefijo):].strip('/')
    if not tab or '/' in tab:
        return False
    matriz = get_menu_roles(env)
    if 'medico' in (matriz.get('/reportes') or []):
        return True
    return 'medico' in (matriz.get('/reportes?tab=%s' % tab) or [])


def _cors_headers():
    return {
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
        'Access-Control-Allow-Headers': 'Content-Type, Authorization',
    }


def _json(data, status=200):
    return Response(
        json.dumps(data, default=str),
        status=status,
        mimetype='application/json',
        headers=_cors_headers(),
    )


def _b64url_encode(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b'=').decode('ascii')


def _b64url_decode(s: str) -> bytes:
    pad = '=' * (-len(s) % 4)
    return base64.urlsafe_b64decode(s + pad)


def _get_secret(env) -> str:
    ICP = env['ir.config_parameter'].sudo()
    secret = ICP.get_param(SECRET_KEY)
    if not secret:
        secret = secrets.token_urlsafe(48)
        ICP.set_param(SECRET_KEY, secret)
    return secret


def sign_token(env, user) -> str:
    payload = {
        'uid': user.id,
        'login': user.login,
        'role': user.role,
        'exp': int(time.time()) + TTL_SECONDS,
    }
    body = _b64url_encode(json.dumps(payload, separators=(',', ':')).encode('utf-8'))
    sig = hmac.new(_get_secret(env).encode('utf-8'), body.encode('ascii'), hashlib.sha256).hexdigest()
    return f'{body}.{sig}'


def verify_token(env, token: str):
    if not token or '.' not in token:
        return None
    body, sig = token.rsplit('.', 1)
    expected = hmac.new(_get_secret(env).encode('utf-8'), body.encode('ascii'), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        return None
    try:
        payload = json.loads(_b64url_decode(body).decode('utf-8'))
    except Exception:
        return None
    if int(payload.get('exp', 0)) < int(time.time()):
        return None
    return payload


def current_user(env):
    """Devuelve el usuario autenticado (record) o None."""
    auth = request.httprequest.headers.get('Authorization', '') or ''
    if not auth.lower().startswith('bearer '):
        return None
    token = auth[7:].strip()
    payload = verify_token(env, token)
    if not payload:
        return None
    user = env['skeen.frontend.user'].sudo().browse(payload.get('uid'))
    if not user.exists() or not user.active:
        return None
    return user


def forbid_roles(*roles):
    """Decorator: niega el acceso a una lista explícita de roles.

    Se usa para cerrar operaciones de dinero a roles con rango de escritura
    que no deben tocarlas (p. ej. marketing, que comparte rango con recepción
    para poder agendar, pero no debe cobrar).
    """

    def decorator(func):
        @functools.wraps(func)
        def wrapper(self, *args, **kw):
            if request.httprequest.method == 'OPTIONS':
                return _json({}, 200)
            user = current_user(request.env)
            if user and user.role in roles:
                return _json({'status': 'error', 'message': 'Sin permiso'}, 403)
            return func(self, *args, **kw)
        return wrapper
    return decorator


def require_role(min_role='lectura'):
    """Decorator: exige token válido y un rol con rango >= min_role.

    Deja pasar OPTIONS (preflight CORS) sin autenticación.
    """
    required = RANK.get(min_role, 10)

    def decorator(func):
        @functools.wraps(func)
        def wrapper(self, *args, **kw):
            if request.httprequest.method == 'OPTIONS':
                return _json({}, 200)
            user = current_user(request.env)
            if not user:
                return _json({'status': 'error', 'message': 'No autenticado'}, 401)
            if user.role_rank() < required:
                return _json({'status': 'error', 'message': 'Sin permiso'}, 403)
            if user.role == 'medico' and any(
                request.httprequest.path == p or request.httprequest.path.startswith(
                    p if p.endswith('/') else p + '/')
                for p in MEDICO_BLOCKED_PREFIXES
            ) and not _medico_reporte_permitido(request.env, request.httprequest.path):
                return _json({'status': 'error', 'message': 'Sin permiso'}, 403)
            request.frontend_user = user
            return func(self, *args, **kw)
        return wrapper
    return decorator


# ---------------------------------------------------------------------------
# Permisos de ACCION dentro de la Agenda (selector por usuario).
#
# require_role sigue siendo el guard de rango: el token no lleva permisos,
# asi que no hay listas cacheadas que se puedan desincronizar. Este decorador
# REFINA dentro de la agenda y solo puede QUITAR acceso, nunca darlo por
# encima del rango del rol.
# ---------------------------------------------------------------------------
AGENDA_PERM_LABELS = dict(AGENDA_PERMS)


def agenda_perm_error(user, perm):
    """Response 403 si la accion no esta permitida; None si si lo esta."""
    if user is None or perm not in resolve_agenda_perms(user):
        label = AGENDA_PERM_LABELS.get(perm, perm)
        return _json({'status': 'error', 'message': 'Sin permiso: %s' % label}, 403)
    return None


def require_agenda(perm):
    """Decorator: exige token valido y la accion de agenda indicada.

    Se apila por DEBAJO de require_role (decorador interno), de modo que el
    control de rango del rol ya corrió y solo falta refinar la acción concreta.
    """

    def decorator(func):
        @functools.wraps(func)
        def wrapper(self, *args, **kw):
            if request.httprequest.method == 'OPTIONS':
                return _json({}, 200)
            user = current_user(request.env)
            if not user:
                return _json({'status': 'error', 'message': 'No autenticado'}, 401)
            denied = agenda_perm_error(user, perm)
            if denied is not None:
                return denied
            request.frontend_user = user
            return func(self, *args, **kw)
        return wrapper
    return decorator