CRM_AGENTES_CARGA/backend/core/local_session.py

"""
Sesión local del CRM (patrón SIWEB).

Emite y valida un JWT de sesión propio (HS256, firmado con SECRET_KEY) que
transporta la identidad YA verificada por Keycloak/Hub. Desacopla la sesión de la
app del token KC de 60s: la app valida esta sesión local (sin ir al Hub) durante
su ventana de inactividad, de modo que el refresh del token KC solo se intenta al
expirar la sesión local — no cada ~60s. Esto elimina el bucle de login.

Se RESPETA la revocación central: si el Hub rechaza el refresh, la sesión termina
(no hay re-emisión de fallback).

Marcadores del token:
  - source: "local" + crm_session: True  → distingue de tokens KC (RS256) y del
    token dev-local (dev_local: True).
  - sst (session start time, epoch seg)   → fija la vida ABSOLUTA máxima (cap).
  - exp                                    → sliding por inactividad (idle); se
    re-emite en cada refresh mientras no se supere el cap.

Seguridad: es un desacople CONSCIENTE de la revocación central de KC (OWASP A07).
Se acota con idle corto (= ssoSessionIdleTimeout) y cap absoluto
(= ssoSessionMaxLifespan); el logout elimina la sesión de valkey.
"""

from datetime import datetime, timezone
from typing import Any, Dict, Optional

from jose import JWTError, jwt

from core.config import settings

# Claims de identidad que se propagan del token KC a la sesión local.
_IDENTITY_CLAIMS = (
    "sub", "email", "preferred_username", "username", "name",
    "given_name", "family_name", "first_name", "last_name",
    "tenant_id", "tenant_slug", "roles", "permissions",
    "is_hub_admin", "avatar_url",
)


def _now_epoch() -> int:
    return int(datetime.now(timezone.utc).timestamp())


def mint_session_token(claims: Dict[str, Any], session_start: Optional[int] = None) -> str:
    """
    Emite un JWT de sesión local a partir de los claims (verificados) del usuario.
    `session_start` (epoch seg) fija el inicio de sesión para el cap absoluto; si
    no se provee, se usa el momento actual (sesión nueva).
    """
    now = _now_epoch()
    sst = int(session_start) if session_start else now

    payload: Dict[str, Any] = {
        k: claims[k] for k in _IDENTITY_CLAIMS if claims.get(k) is not None
    }
    payload.update({
        "source": "local",
        "crm_session": True,
        "sst": sst,
        "iat": now,
        "exp": now + settings.SESSION_IDLE_MINUTES * 60,
    })
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")


def verify_session_token(token: str) -> Optional[Dict[str, Any]]:
    """
    Valida un JWT de sesión local. Retorna los claims si es válido, no expiró por
    inactividad y no superó el cap absoluto de vida; None en cualquier otro caso.
    Nunca lanza (para poder encadenar con la validación contra el Hub).
    """
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None

    # Solo aceptamos tokens de sesión local del CRM (no KC, no dev-local).
    if not payload.get("crm_session") or payload.get("source") != "local":
        return None

    # Cap absoluto de vida de sesión (independiente del sliding por idle).
    sst = payload.get("sst")
    if isinstance(sst, (int, float)):
        if _now_epoch() - int(sst) > settings.SESSION_MAX_HOURS * 3600:
            return None

    return payload


def session_start_of(payload: Dict[str, Any]) -> Optional[int]:
    """Extrae el epoch de inicio de sesión (sst) de un payload de sesión local."""
    sst = payload.get("sst")
    return int(sst) if isinstance(sst, (int, float)) else None