ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'development' into feature/reporte-bak

Browse Source
This commit is contained in:
hreyes
2026-04-30 16:30:14 -06:00
parent 143bfded29 2fbd476145
commit 259d67ebb2
45 changed files with 1923 additions and 2297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
backend/.env.example
View File

@@ -4,29 +4,18 @@ APP_VERSION=1.0.0
DEBUG=True
ENVIRONMENT=development
# Database - Core (Shared)
# Database - Core
CORE_DB_HOST=localhost
CORE_DB_PORT=5432
CORE_DB_NAME=anexo76_core
CORE_DB_USER=postgres
CORE_DB_PASSWORD=postgres
# Keycloak
KEYCLOAK_SERVER_URL=http://localhost:8080/kcauth
KEYCLOAK_REALM=master
KEYCLOAK_CLIENT_ID=anexo76-backend
KEYCLOAK_CLIENT_SECRET=your-client-secret
# Security
SECRET_KEY=your-secret-key-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
# CORS
CORS_ORIGINS=http://localhost:5173,http://localhost:3000
# License Service
LICENSE_CHECK_ENABLED=True
# Hub de Aduanasoft — requerido siempre (SaaS y self-hosted)
HUB_URL=https://hub.aduanasoft.com
# Factura COVE / VUCEM / DODA / API Ventanilla Única
# Llave y IV AES-256-CBC para cifrar la clave FIEL.
@@ -37,7 +26,6 @@ COVE_API_URL=https://api.vu.aduanasoft.com
# Verificación SSL para el API de VU (False en redes internas / dev, True en producción).
COVE_API_VERIFY_SSL=False
# Synchronization (Hub & Spoke)
# Sincronización (Hub & Spoke)
SYNC_SECRET_TOKEN=change-this-sync-token-in-production
# Only for spokes/clients. Leave empty if this is the Hub.
CENTRAL_SERVER_URL=http://localhost:8000/api/v1/core/help-center/sync/
CENTRAL_SERVER_URL=

53
backend/alembic/versions/g1b2c3d4e5f6_invoice_header_user_field_length.py Normal file
View File

@@ -0,0 +1,53 @@
"""increase invoice_header who_processed and capture_user to 100 chars
Revision ID: g1b2c3d4e5f6
Revises: f7a8b9c0d1e2
Create Date: 2026-04-30 00:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'g1b2c3d4e5f6'
down_revision = 'f7a8b9c0d1e2'
branch_labels = None
depends_on = None
def upgrade() -> None:
op.alter_column(
'invoice_header',
'who_processed',
existing_type=sa.String(length=20),
type_=sa.String(length=100),
existing_nullable=True,
schema='a76',
)
op.alter_column(
'invoice_header',
'capture_user',
existing_type=sa.String(length=20),
type_=sa.String(length=100),
existing_nullable=True,
schema='a76',
)
def downgrade() -> None:
op.alter_column(
'invoice_header',
'capture_user',
existing_type=sa.String(length=100),
type_=sa.String(length=20),
existing_nullable=True,
schema='a76',
)
op.alter_column(
'invoice_header',
'who_processed',
existing_type=sa.String(length=100),
type_=sa.String(length=20),
existing_nullable=True,
schema='a76',
)

26
backend/alembic/versions/h1a2b3c4d5e6_merge_multiple_heads.py Normal file
View File

@@ -0,0 +1,26 @@
"""merge alembic heads for deployment
Revision ID: h1a2b3c4d5e6
Revises: c3d4e5f6a7b8, g1b2c3d4e5f6
Create Date: 2026-04-30 12:00:00.000000
"""
from typing import Sequence, Union
# revision identifiers, used by Alembic.
revision: str = "h1a2b3c4d5e6"
down_revision: Union[str, Sequence[str], None] = (
"c3d4e5f6a7b8",
"g1b2c3d4e5f6",
)
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass

104
backend/alembic/versions/i7j8k9l0m1n2_fix_varchar20_truncation_errors.py Normal file
View File

@@ -0,0 +1,104 @@
"""Fix VARCHAR(20) truncation errors in audit_log and discharges
Revision ID: i7j8k9l0m1n2
Revises: h1a2b3c4d5e6
Create Date: 2026-04-30 13:00:00.000000
Issues fixed:
- audit_log.system: String(20) → String(50)
- audit_log.operation_type: String(20) → String(50)
- discharges.cancelled_by: String(20) → String(100)
- octave_balance.octave_permit: String(20) → String(100)
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'i7j8k9l0m1n2'
down_revision = 'h1a2b3c4d5e6'
branch_labels = None
depends_on = None
def upgrade() -> None:
# Fix audit_log.system
op.alter_column(
'audit_logs',
'system',
existing_type=sa.String(length=20),
type_=sa.String(length=50),
existing_nullable=False,
schema='a76',
)
# Fix audit_log.operation_type
op.alter_column(
'audit_logs',
'operation_type',
existing_type=sa.String(length=20),
type_=sa.String(length=50),
existing_nullable=True,
schema='a76',
)
# Fix discharges.cancelled_by
op.alter_column(
'discharge_header',
'cancelled_by',
existing_type=sa.String(length=20),
type_=sa.String(length=100),
existing_nullable=True,
schema='a24',
)
# Fix octave_balance.octave_permit
op.alter_column(
'octave_balance',
'octave_permit',
existing_type=sa.String(length=20),
type_=sa.String(length=100),
existing_nullable=False,
schema='a76',
)
def downgrade() -> None:
# Revert octave_balance.octave_permit
op.alter_column(
'octave_balance',
'octave_permit',
existing_type=sa.String(length=100),
type_=sa.String(length=20),
existing_nullable=False,
schema='a76',
)
# Revert discharges.cancelled_by
op.alter_column(
'discharge_header',
'cancelled_by',
existing_type=sa.String(length=100),
type_=sa.String(length=20),
existing_nullable=True,
schema='a24',
)
# Revert audit_log.operation_type
op.alter_column(
'audit_logs',
'operation_type',
existing_type=sa.String(length=50),
type_=sa.String(length=20),
existing_nullable=True,
schema='a76',
)
# Revert audit_log.system
op.alter_column(
'audit_logs',
'system',
existing_type=sa.String(length=50),
type_=sa.String(length=20),
existing_nullable=False,
schema='a76',
)

2
backend/api/v1/modules/a24/discharges/models.py
View File

@@ -132,7 +132,7 @@ class DischargeHeader(Base, TenantScopedMixin, TimestampMixin):
)
# ── Cancellation trail ────────────────────────────────────────────────
cancelled_by: Mapped[Optional[str]] = mapped_column(String(20))
cancelled_by: Mapped[Optional[str]] = mapped_column(String(100))
cancellation_reason: Mapped[Optional[str]] = mapped_column(String(300))
# ── Relationships ─────────────────────────────────────────────────────

2
backend/api/v1/modules/a76/audit_log/middleware.py
View File

@@ -12,7 +12,7 @@ class UserContextMiddleware(BaseHTTPMiddleware):
try:
# verify_token might raise exception if invalid, we catch it to not block request
# but we won't have user context
user_info = verify_token(token)
user_info = await verify_token(token)
set_user_context(user_info)
except Exception:
# Log error or ignore

4
backend/api/v1/modules/a76/audit_log/models.py
View File

@@ -30,12 +30,12 @@ class AuditLog(Base, TenantScopedMixin, TimestampMixin):
# Technical Columns
timestamp = Column(DateTime(timezone=True), nullable=False, index=True) # Combined for queries
system = Column(String(20), nullable=False, index=True, default="fixed_asset")
system = Column(String(50), nullable=False, index=True, default="fixed_asset")
# Traceability
table_name = Column(String(100), nullable=True, index=True)
record_id = Column(String(255), nullable=True, index=True)
operation_type = Column(String(20), nullable=True, index=True) # CREATE, UPDATE, DELETE, LOGIN
operation_type = Column(String(50), nullable=True, index=True) # CREATE, UPDATE, DELETE, LOGIN
# Data Changes
old_values = Column(JSONB, nullable=True)

4
backend/api/v1/modules/a76/invoices/models.py
View File

@@ -124,10 +124,10 @@ class InvoiceHeader(Base, TenantScopedMixin, TimestampMixin):
TIMESTAMP(timezone=False)
) # FECHAACTUALIZACION / FECHAACTUAL
who_processed: Mapped[Optional[str]] = mapped_column(
String(20)
String(100)
) # USUARIOACT / Quien actualizó
capture_user: Mapped[Optional[str]] = mapped_column(
String(20)
String(100)
) # USUARIOCAP / Usuario que capturó
traffic_light_status: Mapped[Optional[str]] = mapped_column(

11
backend/api/v1/modules/a76/invoices/routes.py
View File

@@ -1,6 +1,6 @@
from typing import Dict, Any, Optional
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from core.security import collect_user_role_names, get_current_user, validate_access_to_resource
from fastapi import APIRouter, Depends, HTTPException, Query, Path
from sqlalchemy import func, or_, and_
from sqlalchemy.orm import Session
@@ -201,8 +201,9 @@ def list_invoices(
List invoices with optional filters and granular permission enforcement.
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
user_roles = current_user.get("realm_access", {}).get("roles", [])
allowed_filters = []
# Roles vienen del Hub (/auth/me), no de realm_access del JWT crudo.
is_hub_admin = "admin" in collect_user_role_names(current_user)
allowed_filters = []
# Información del usuario para debugging (se ve en los logs del servidor)
user_name = current_user.get('preferred_username') or current_user.get('email', 'Desconocido')
@@ -246,8 +247,8 @@ def list_invoices(
if not allowed_filters:
# Si no tiene ningún permiso de factura, bloqueamos
# Excepto si es un admin de Keycloak, le damos el beneficio de la duda pero logeamos
if "admin" in user_roles:
# Excepto si es admin (Hub roles / Keycloak), fallback a ver todo.
if is_hub_admin:
print(f"[AUTH] Keycloak Admin {user_name} has no app permissions. Granting view_all as fallback.")
allowed_filters = None
else:

4
backend/api/v1/modules/a76/invoices/schemas.py
View File

@@ -73,8 +73,8 @@ class InvoiceHeaderBase(BaseModel):
return InvoiceStatus.PROCESSED.value
return v
processed_date: Optional[datetime] = Field(None, description="Update date")
who_processed: Optional[str] = Field(None, max_length=20, description="Who processed")
capture_user: Optional[str] = Field(None, max_length=20, description="Capture user")
who_processed: Optional[str] = Field(None, max_length=100, description="Who processed")
capture_user: Optional[str] = Field(None, max_length=100, description="Capture user")
traffic_light_status: Optional[str] = Field(
None, max_length=50, description="Traffic light status"
)

2
backend/api/v1/modules/a76/rule_octave/balances/models.py
View File

@@ -42,7 +42,7 @@ class OctaveBalance(Base, TenantScopedMixin, TimestampMixin):
origin_country: Mapped[str] = mapped_column(String(3)) # PAISORIGEN
fraction_type: Mapped[str] = mapped_column(String(7)) # TIPOFRACIMPO
sector: Mapped[str] = mapped_column(String(8)) # SECTOR
octave_permit: Mapped[str] = mapped_column(String(20)) # PERMISOROCTAVA
octave_permit: Mapped[str] = mapped_column(String(100)) # PERMISOROCTAVA
origin: Mapped[str] = mapped_column(String(3)) # PROCEDENCIA ('TEM')
system: Mapped[str] = mapped_column(String(5)) # SISTEMA ('fixed_asset | inventory')
line: Mapped[int] = mapped_column(Integer) # LINEA

9
backend/api/v1/modules/core/auth/dto.py
View File

@@ -33,6 +33,9 @@ class TokenResponseDTO(BaseModel):
refresh_token: str
token_type: str = "bearer"
expires_in: int
tenant: Optional["TenantInfoDTO"] = None
tenant_id: Optional[int] = None
tenant_slug: Optional[str] = None
class Config:
json_schema_extra = {
@@ -197,3 +200,9 @@ class LoginChoiceResponseDTO(BaseModel):
status: str = "choose_tenant"
tenants: list[TenantInfoDTO]
class SSOExchangeRequestDTO(BaseModel):
"""DTO para canjear el relay token por KC tokens."""
relay_token: str = Field(..., description="Relay token recibido en la URL")

55
backend/api/v1/modules/core/auth/routes.py
View File

@@ -17,6 +17,7 @@ from .dto import (
RegisterRequestDTO,
RegisterResponseDTO,
SetCookieRequestDTO,
SSOExchangeRequestDTO,
SwitchTenantRequestDTO,
TokenResponseDTO,
UserInfoResponseDTO,
@@ -48,7 +49,7 @@ async def register(
- Atributos de tenant
"""
service = AuthService(db)
return service.register(register_data)
return await service.register(register_data)
@router.post("/login", response_model=None)
@@ -68,7 +69,7 @@ async def login(
service = AuthService(db)
import logging
logger = logging.getLogger(__name__)
return service.login(
return await service.login(
login_data=login_data,
ip_address=request.client.host,
user_agent=request.headers.get("user-agent")
@@ -90,7 +91,7 @@ async def switch_tenant(
"""
service = AuthService(db)
# Obtener info del usuario desde el access token actual
user_info = service.get_user_info(credentials.credentials)
user_info = await service.get_user_info(credentials.credentials)
keycloak_user_id = user_info.sub
# El realm se puede inferir del token; usamos el campo tenant_id para buscar el realm actual,
@@ -103,7 +104,7 @@ async def switch_tenant(
if not tenant:
raise HTTPException(status_code=403, detail="Access denied")
return service.switch_tenant(
return await service.switch_tenant(
keycloak_user_id=keycloak_user_id,
keycloak_realm=tenant.keycloak_realm,
tenant_slug=data.tenant_slug,
@@ -119,7 +120,7 @@ async def refresh_token(
Refresca el access token usando el refresh token
"""
service = AuthService(db)
return service.refresh_token(refresh_data)
return await service.refresh_token(refresh_data)
@router.get("/me", response_model=UserInfoResponseDTO)
@@ -131,7 +132,7 @@ async def get_current_user_info(
Obtiene información del usuario actual desde el token
"""
service = AuthService(db)
return service.get_user_info(credentials.credentials)
return await service.get_user_info(credentials.credentials)
@router.post("/logout")
@@ -155,7 +156,7 @@ async def logout(
# I'll keep it simple.
service = AuthService(db)
return service.logout(logout_data)
return await service.logout(logout_data)
@router.post("/exchange-code", response_model=TokenResponseDTO)
@@ -172,7 +173,7 @@ async def exchange_code(
externo y Keycloak lo redirige al frontend con el código en los query params.
"""
service = AuthService(db)
return service.exchange_code(exchange_data)
return await service.exchange_code(exchange_data)
@router.post("/set-cookie")
@@ -198,7 +199,7 @@ async def set_cookie(
service = AuthService(db)
try:
# Validar el access token
user_info = service.get_user_info(cookie_data.access_token)
user_info = await service.get_user_info(cookie_data.access_token)
# Establecer las cookies
# Access token cookie
@@ -231,3 +232,39 @@ async def set_cookie(
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error validando tokens: {str(e)}")
@router.post("/sso-exchange", response_model=TokenResponseDTO)
async def sso_exchange(
body: SSOExchangeRequestDTO,
response: Response,
db: Session = Depends(get_core_db),
):
"""
Canjea un relay token de un solo uso (generado por el Hub) por KC tokens.
Llamado server-side desde la página /auth/sso del frontend de Anexo76.
Establece cookies HttpOnly con los tokens y devuelve el resultado.
"""
service = AuthService(db)
tokens = await service.sso_exchange(body.relay_token)
_is_prod = False # TODO: leer de settings.ENVIRONMENT == "production"
response.set_cookie(
key="access_token",
value=tokens.access_token,
httponly=True,
secure=_is_prod,
samesite="lax",
max_age=3600,
path="/",
)
response.set_cookie(
key="refresh_token",
value=tokens.refresh_token,
httponly=True,
secure=_is_prod,
samesite="lax",
max_age=86400,
path="/",
)
return tokens

822
backend/api/v1/modules/core/auth/service.py
View File

@@ -1,24 +1,15 @@
"""
Servicio de autenticación con Keycloak
"""
import logging
from datetime import datetime
import httpx
from typing import Any, Dict
from api.v1.modules.core.tenants.service import TenantService
from api.v1.modules.core.user_tenant.service import UserTenantService
from core.config import settings
from fastapi import HTTPException
from keycloak import KeycloakAdmin, KeycloakOpenID
from keycloak.exceptions import KeycloakError
from sqlalchemy.orm import Session
from .dto import (
LoginRequestDTO,
LogoutRequestDTO,
RefreshTokenRequestDTO,
RegisterRequestDTO,
RegisterResponseDTO,
TokenResponseDTO,
UserInfoResponseDTO,
)
@@ -27,738 +18,195 @@ logger = logging.getLogger(__name__)
class AuthService:
"""Servicio de autenticación"""
"""Servicio de autenticación centralizado vía Hub"""
def __init__(self, db: Session):
self.db = db
self.keycloak_openid = KeycloakOpenID(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=settings.KEYCLOAK_REALM,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
def login(
async def login(
self,
login_data: LoginRequestDTO,
ip_address: str = None,
user_agent: str = None
):
"""
Autentica usuario y obtiene tokens.
Si se omite tenant_slug, verifica credenciales primero y devuelve
la lista de tenants disponibles (LoginChoiceResponseDTO) en lugar de tokens.
Args:
login_data: Credenciales de login (tenant_slug es opcional)
ip_address: Dirección IP del cliente
user_agent: User Agent del cliente
Returns:
TokenResponseDTO si tenant_slug fue provisto,
LoginChoiceResponseDTO si no se proveyó tenant_slug.
Raises:
HTTPException: Si las credenciales son inválidas
Autentica usuario a través del Hub y obtiene tokens.
"""
# PRIMER PASO: sin tenant_slug → verificar creds y devolver lista de orgs
if not login_data.tenant_slug:
from .dto import LoginChoiceResponseDTO, TenantInfoDTO
tenants = self._verify_credentials_and_list_tenants(
login_data.username, login_data.password
)
# Siempre devolver LoginChoiceResponseDTO; el frontend decide si
# auto-seleccionar (1 tenant) o mostrar selector (>1 tenants).
return LoginChoiceResponseDTO(
tenants=[TenantInfoDTO(**t) for t in tenants]
)
try:
# Verificar que el tenant existe
tenant_service = TenantService(self.db)
user_tenant_service = UserTenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(login_data.tenant_slug)
if not tenant or not tenant.is_active:
raise HTTPException(status_code=401, detail="Invalid credentials")
# Crear nueva instancia de KeycloakOpenID con el realm del tenant
keycloak_client = KeycloakOpenID(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=tenant.keycloak_realm,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
# PASO 1: Primero actualizamos los atributos del usuario ANTES de autenticar
# Esto es necesario para que los Protocol Mappers incluyan los valores correctos
# en el token que se generará a continuación
# Para obtener el user_id, necesitamos hacer una autenticación temporal
# o buscar el usuario por username
try:
keycloak_admin = KeycloakAdmin(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=tenant.keycloak_realm,
user_realm_name="master",
verify=True,
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{settings.HUB_URL}api/v1/auth/login",
json=login_data.model_dump()
)
# Buscar usuario por username
users = keycloak_admin.get_users({"username": login_data.username})
if users and len(users) > 0:
user_id = users[0]["id"]
# Verificar si el usuario tiene acceso a este tenant
has_access = user_tenant_service.user_has_access_to_tenant(
user_id, tenant.id
if response.status_code == 200:
data = response.json()
# Si el Hub devolvió una lista de tenants (hubo login exitoso pero falta seleccionar tenant)
if "tenants" in data:
from .dto import LoginChoiceResponseDTO, TenantInfoDTO
return LoginChoiceResponseDTO(
tenants=[TenantInfoDTO(**t) for t in data["tenants"]]
)
if not has_access:
logger.warning(
f"User {user_id} tried to access tenant {tenant.id} without permission"
)
raise HTTPException(
status_code=401,
detail="Invalid credentials",
)
# Si devolvió tokens
# AUDIT LOG: Login Success
try:
from api.v1.modules.a76.audit_log.services.service import AuditService
AuditService.log_login(
db=self.db,
username=login_data.username,
ip_address=ip_address,
user_agent=user_agent
)
except Exception as e:
logger.error(f"Failed to audit login: {e}")
# Obtener los datos actuales del usuario
current_user = keycloak_admin.get_user(user_id)
current_attributes = current_user.get("attributes", {})
# Actualizar los atributos de tenant
current_attributes["tenant_id"] = [str(tenant.id)]
current_attributes["tenant_slug"] = [tenant.slug]
# Actualizar el usuario con los nuevos atributos
update_payload = {
"email": current_user.get("email"),
"firstName": current_user.get("firstName"),
"lastName": current_user.get("lastName"),
"enabled": current_user.get("enabled", True),
"emailVerified": current_user.get("emailVerified", False),
"attributes": current_attributes,
}
keycloak_admin.update_user(user_id=user_id, payload=update_payload)
except KeycloakError as e:
logger.warning(f"Could not pre-update user attributes: {str(e)}")
# Continuamos con el login aunque falle la actualización
except HTTPException:
raise # Re-lanzamos las excepciones HTTP (como acceso denegado)
except Exception as e:
logger.warning(f"Error pre-updating user attributes: {str(e)}")
# PASO 2: Ahora autenticamos al usuario
token_response = keycloak_client.token(
username=login_data.username,
password=login_data.password,
grant_type=["password"],
)
return TokenResponseDTO(**data)
# AUDIT LOG: Login Success
# Pasar el mensaje de error real del Hub al cliente
try:
from api.v1.modules.a76.audit_log.services.service import AuditService
AuditService.log_login(
db=self.db,
username=login_data.username,
ip_address=ip_address,
user_agent=user_agent
)
except Exception as e:
logger.error(f"Failed to audit login: {e}")
hub_detail = response.json().get("detail", None)
except Exception:
hub_detail = None
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type="bearer",
expires_in=token_response["expires_in"],
)
if response.status_code == 401:
raise HTTPException(status_code=401, detail=hub_detail or "Credenciales inválidas")
except KeycloakError as e:
logger.warning(f"Keycloak authentication failed: {str(e)}")
raise HTTPException(status_code=401, detail="Invalid credentials")
logger.error(f"Hub login failed with status {response.status_code}: {response.text}")
raise HTTPException(status_code=response.status_code, detail=hub_detail or "Error en el servidor de autenticación")
except httpx.HTTPError as e:
logger.error(f"Hub unreachable during login: {str(e)}")
raise HTTPException(status_code=503, detail="Authentication service unavailable")
except HTTPException:
raise
except Exception as e:
logger.error(f"Login error: {str(e)}")
logger.error(f"Unexpected login error: {str(e)}")
raise HTTPException(status_code=500, detail="Authentication error")
def refresh_token(self, refresh_data: RefreshTokenRequestDTO) -> TokenResponseDTO:
async def refresh_token(self, refresh_data: RefreshTokenRequestDTO) -> TokenResponseDTO:
"""
Refresca el access token usando refresh token
Args:
refresh_data: Refresh token
Returns:
TokenResponseDTO con nuevos tokens
Refresca el access token usando el Hub
"""
try:
token_response = self.keycloak_openid.refresh_token(
refresh_data.refresh_token
)
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{settings.HUB_URL}api/v1/auth/refresh",
json=refresh_data.model_dump()
)
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type="bearer",
expires_in=token_response["expires_in"],
)
if response.status_code == 200:
return TokenResponseDTO(**response.json())
raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
except KeycloakError as e:
logger.warning(f"Token refresh failed: {str(e)}")
raise HTTPException(
status_code=401, detail="Invalid or expired refresh token"
)
except Exception as e:
logger.error(f"Token refresh error: {str(e)}")
raise HTTPException(status_code=500, detail="Token refresh error")
def get_user_info(self, access_token: str) -> UserInfoResponseDTO:
async def get_user_info(self, access_token: str) -> UserInfoResponseDTO:
"""
Obtiene información del usuario desde el token
Obtiene información del usuario desde el Hub
"""
from core.security import verify_token
# Aprovechamos la verificación (y cache) de security.py
user_info = await verify_token(access_token)
return UserInfoResponseDTO(**user_info)
Args:
access_token: Access token JWT
Returns:
UserInfoResponseDTO con información del usuario
async def logout(self, logout_data: LogoutRequestDTO) -> dict:
"""
Cierra sesión a través del Hub
"""
try:
user_info = self.keycloak_openid.userinfo(access_token)
# Extraer roles
roles = []
if "realm_access" in user_info:
roles = user_info["realm_access"].get("roles", [])
roles = ["admin"]
# Extraer tenant_id y tenant_slug si están presentes
tenant_id = user_info.get("tenant_id")
if not tenant_id and "attributes" in user_info:
tenant_id = user_info["attributes"].get("tenant_id")
tenant_slug = user_info.get("tenant_slug")
if not tenant_slug and "attributes" in user_info:
tenant_slug = user_info["attributes"].get("tenant_slug")
# Puede venir como lista de Keycloak attributes
if isinstance(tenant_slug, list):
tenant_slug = tenant_slug[0] if tenant_slug else None
# Obtener permisos del usuario en todas las compañías permitidas
permissions = set()
user_sub = user_info.get("sub")
if user_sub:
from api.v1.modules.core.permissions.service import PermissionService
from api.v1.modules.core.permissions.models import UserCompanyRole
perm_service = PermissionService(self.db)
# Obtener todas las compañías a las que el usuario tiene acceso
user_roles = self.db.query(UserCompanyRole.company_id).filter(
UserCompanyRole.user_id == user_sub,
UserCompanyRole.is_active == True
).distinct().all()
# Unir los permisos de todas las compañías para alimentar la UI
for (cid,) in user_roles:
permissions.update(perm_service.get_user_permissions(user_sub, cid))
# 🛡️ MEJORA DEV: Si es admin de Keycloak O estamos en desarrollo y no tiene permisos locales aún.
# Esto evita el "lockout" cuando se reinicia el proyecto para todos los usuarios.
from core.config import settings
if "admin" in roles or (settings.ENVIRONMENT == "development" and not permissions):
try:
from api.v1.modules.core.permissions.registry import registry as perm_registry
# Asegurar que los permisos core estén registrados
from api.v1.modules.core.permissions import seed_v2
all_registered = [p.code for p in perm_registry.get_all()]
permissions.update(all_registered)
logger.info(f"God Mode (Dev): Otorgando {len(all_registered)} permisos al usuario {user_sub}")
except Exception as e:
logger.error(f"Error in God Mode bootstrap: {e}")
permissions = list(permissions)
return UserInfoResponseDTO(
sub=user_info.get("sub"),
email=user_info.get("email"),
name=user_info.get("name"),
preferred_username=user_info.get("preferred_username"),
tenant_id=int(tenant_id) if tenant_id else None,
tenant_slug=tenant_slug,
roles=roles,
permissions=permissions,
)
except KeycloakError as e:
logger.warning(f"Get user info failed: {str(e)}")
raise HTTPException(status_code=401, detail="Invalid token")
except Exception as e:
logger.error(f"Get user info error: {str(e)}")
raise HTTPException(status_code=500, detail="Error retrieving user info")
def logout(self, logout_data: LogoutRequestDTO) -> dict:
"""
Cierra sesión invalidando el refresh token
Args:
logout_data: Refresh token a invalidar
Returns:
Dict con mensaje de éxito
"""
try:
self.keycloak_openid.logout(logout_data.refresh_token)
async with httpx.AsyncClient(timeout=10.0) as client:
await client.post(
f"{settings.HUB_URL}api/v1/auth/logout",
json=logout_data.model_dump()
)
return {"message": "Logged out successfully"}
except KeycloakError as e:
logger.warning(f"Logout failed: {str(e)}")
# No lanzamos error aquí, el logout puede fallar si el token ya expiró
return {"message": "Logged out"}
except Exception as e:
logger.error(f"Logout error: {str(e)}")
raise HTTPException(status_code=500, detail="Logout error")
return {"message": "Logged out"}
def register(self, register_data: RegisterRequestDTO) -> RegisterResponseDTO:
async def register(self, register_data: Any) -> Any:
"""
Registra un nuevo usuario en Keycloak
Args:
register_data: Datos del usuario a registrar
Returns:
RegisterResponseDTO con información del usuario creado
Raises:
HTTPException: Si el registro falla
Registra un usuario a través del Hub
"""
try:
# Verificar que el tenant existe
from api.v1.modules.core.tenants.service import TenantService
tenant_service = TenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(register_data.tenant_slug)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant is not active")
# Crear instancia de KeycloakAdmin para gestión de usuarios
keycloak_admin = KeycloakAdmin(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=tenant.keycloak_realm,
user_realm_name="master", # El admin suele estar en master realm
verify=True,
)
# Preparar datos del usuario para Keycloak
user_data = {
"username": register_data.username,
"email": register_data.email,
"firstName": register_data.first_name,
"lastName": register_data.last_name,
"enabled": True,
"emailVerified": False,
"credentials": [
{
"type": "password",
"value": register_data.password,
"temporary": False,
}
],
"attributes": {"tenant_id": str(tenant.id), "tenant_slug": tenant.slug},
}
# Crear usuario en Keycloak
user_id = keycloak_admin.create_user(user_data)
# Asignar rol por defecto (user) - opcional, solo si existe
try:
user_role = keycloak_admin.get_realm_role("user")
if user_role:
keycloak_admin.assign_realm_roles(user_id, [user_role])
except KeycloakError as e:
# El rol 'user' no existe, no es un error crítico
logger.warning(f"Could not assign 'user' role: {str(e)}")
# Agregar el usuario al tenant en la base de datos
try:
from api.v1.modules.core.user_tenant.service import UserTenantService
user_tenant_service = UserTenantService(self.db)
user_tenant_service.add_user_to_tenant(
keycloak_user_id=user_id,
tenant_id=tenant.id,
role="user", # Rol por defecto
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{settings.HUB_URL}api/v1/auth/register",
json=register_data.model_dump()
)
except Exception as e:
# Si falla, hacer rollback del usuario en Keycloak
logger.error(f"Failed to add user to tenant in database: {str(e)}")
try:
keycloak_admin.delete_user(user_id)
except Exception as e:
pass
raise HTTPException(
status_code=500, detail="Failed to register user in database"
)
return RegisterResponseDTO(
user_id=user_id,
username=register_data.username,
email=register_data.email,
message="User registered successfully",
)
except KeycloakError as e:
error_message = str(e)
logger.warning(f"Keycloak registration failed: {error_message}")
# Mensajes de error más específicos
if "User exists" in error_message or "409" in error_message:
raise HTTPException(
status_code=409, detail="Username or email already exists"
)
elif "Invalid" in error_message:
raise HTTPException(status_code=400, detail="Invalid user data")
else:
raise HTTPException(status_code=500, detail="Registration error")
except HTTPException:
raise
if response.status_code == 201:
return response.json()
raise HTTPException(status_code=response.status_code, detail=response.text)
except Exception as e:
logger.error(f"Registration error: {str(e)}")
raise HTTPException(status_code=500, detail="Registration error")
def exchange_code(self, exchange_data) -> TokenResponseDTO:
async def exchange_code(self, exchange_data: Any) -> TokenResponseDTO:
"""
Intercambia un authorization code por tokens
Este método se usa cuando el frontend recibe un código de autorización
después de un login con proveedor externo (Microsoft, Google, etc.)
a través de Keycloak.
Args:
exchange_data: Datos del código y redirect_uri
Returns:
TokenResponseDTO con access_token y refresh_token
Raises:
HTTPException: Si el código es inválido o expiró
Intercambia código por tokens a través del Hub
"""
try:
# Importar el DTO aquí para evitar referencias circulares
# Intercambiar código por tokens usando Keycloak
token_response = self.keycloak_openid.token(
grant_type="authorization_code",
code=exchange_data.code,
redirect_uri=exchange_data.redirect_uri,
)
# Si se proporciona tenant_slug, podríamos validar que el usuario pertenece a ese tenant
# Por ahora simplemente retornamos los tokens
if exchange_data.tenant_slug:
# Validar que el tenant existe y está activo
tenant_service = TenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(exchange_data.tenant_slug)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant is not active")
# Opcional: Verificar que el usuario pertenece al tenant
# Esto depende de cómo manejes los tenants en tu aplicación
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type=token_response.get("token_type", "bearer"),
expires_in=token_response.get("expires_in", 3600),
)
except KeycloakError as e:
error_message = str(e)
logger.warning(f"Code exchange failed: {error_message}")
if "invalid_grant" in error_message.lower():
raise HTTPException(
status_code=400, detail="Invalid or expired authorization code"
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{settings.HUB_URL}api/v1/auth/exchange-code",
json=exchange_data.model_dump()
)
elif "invalid_client" in error_message.lower():
raise HTTPException(
status_code=401, detail="Invalid client credentials"
)
else:
raise HTTPException(status_code=500, detail="Token exchange error")
if response.status_code == 200:
return TokenResponseDTO(**response.json())
raise HTTPException(status_code=response.status_code, detail="Code exchange failed")
except Exception as e:
logger.error(f"Exchange code error: {str(e)}")
raise HTTPException(status_code=500, detail="Exchange code error")
async def switch_tenant(self, **kwargs) -> TokenResponseDTO:
"""
Cambia de tenant a través del Hub
"""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{settings.HUB_URL}api/v1/auth/switch-tenant",
json=kwargs
)
if response.status_code == 200:
return TokenResponseDTO(**response.json())
raise HTTPException(status_code=response.status_code, detail="Switch tenant failed")
except Exception as e:
logger.error(f"Switch tenant error: {str(e)}")
raise HTTPException(status_code=500, detail="Switch tenant error")
async def sso_exchange(self, relay_token: str) -> TokenResponseDTO:
"""
Canjea un relay token de un solo uso por KC tokens.
Llama al Hub backend (server-to-server), sin Bearer requerido en el Hub.
"""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{settings.HUB_URL}api/v1/auth/sso-exchange",
json={"relay_token": relay_token},
)
if response.status_code == 200:
data = response.json()
return TokenResponseDTO(
access_token=data["access_token"],
refresh_token=data["refresh_token"],
token_type=data.get("token_type", "bearer"),
expires_in=data.get("expires_in", 3600),
tenant_id=data.get("tenant_id"),
tenant_slug=data.get("tenant_slug"),
)
raise HTTPException(
status_code=response.status_code,
detail=response.json().get("detail", "SSO exchange failed"),
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Code exchange error: {str(e)}")
raise HTTPException(status_code=500, detail="Code exchange error")
def switch_tenant(
self,
keycloak_user_id: str,
keycloak_realm: str,
tenant_slug: str,
refresh_token: str,
) -> TokenResponseDTO:
"""
Cambia el tenant activo de un usuario autenticado sin requerir su contraseña.
Pasos:
1. Verifica que el tenant existe y está activo.
2. Verifica que el usuario tiene acceso a ese tenant.
3. Actualiza los atributos tenant_id/tenant_slug del usuario en Keycloak.
4. Usa el refresh_token para emitir nuevos tokens que ya contienen los atributos actualizados.
"""
from api.v1.modules.core.tenants.models import Tenant
tenant_service = TenantService(self.db)
user_tenant_service = UserTenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(tenant_slug)
if not tenant or not tenant.is_active:
raise HTTPException(status_code=403, detail="Access denied")
# Verificar acceso
has_access = user_tenant_service.user_has_access_to_tenant(keycloak_user_id, tenant.id)
if not has_access:
raise HTTPException(status_code=403, detail="Access denied")
# Actualizar atributos en Keycloak antes de emitir el nuevo token
try:
keycloak_admin = KeycloakAdmin(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=keycloak_realm,
user_realm_name="master",
verify=True,
)
current_user = keycloak_admin.get_user(keycloak_user_id)
attrs = current_user.get("attributes", {})
attrs["tenant_id"] = [str(tenant.id)]
attrs["tenant_slug"] = [tenant.slug]
keycloak_admin.update_user(
user_id=keycloak_user_id,
payload={
"email": current_user.get("email"),
"firstName": current_user.get("firstName"),
"lastName": current_user.get("lastName"),
"enabled": current_user.get("enabled", True),
"emailVerified": current_user.get("emailVerified", False),
"attributes": attrs,
},
)
except KeycloakError as e:
logger.warning(f"switch_tenant: could not update user attributes: {e}")
raise HTTPException(status_code=500, detail="Could not update tenant attributes")
# Emitir nuevos tokens usando el refresh_token existente
keycloak_client = KeycloakOpenID(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=keycloak_realm,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
try:
token_response = keycloak_client.refresh_token(refresh_token)
except KeycloakError as e:
logger.warning(f"switch_tenant: token refresh failed: {e}")
raise HTTPException(status_code=401, detail="Token refresh failed; please log in again")
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type="bearer",
expires_in=token_response["expires_in"],
)
def _verify_credentials_and_list_tenants(self, username: str, password: str) -> list:
"""
Verifica las credenciales del usuario contra Keycloak y, solo si son válidas,
devuelve la lista de tenants a los que tiene acceso.
Esto evita el oráculo de enumeración de usuarios del antiguo endpoint
/discover-tenants que no requería contraseña.
Args:
username: Nombre de usuario o email
password: Contraseña en texto plano
Returns:
Lista de dicts {id, name, slug} con los tenants del usuario
Raises:
HTTPException 401: Si las credenciales son inválidas
"""
from api.v1.modules.core.tenants.models import Tenant
from api.v1.modules.core.user_tenant.models import UserTenant
from sqlalchemy import and_
tenants = self.db.query(Tenant).filter(Tenant.is_active).all()
if not tenants:
raise HTTPException(status_code=401, detail="Invalid credentials")
realms: dict[str, list] = {}
for tenant in tenants:
realms.setdefault(tenant.keycloak_realm, []).append(tenant)
credentials_verified = False
matched_tenants = []
for realm_name, realm_tenants in realms.items():
try:
keycloak_admin = KeycloakAdmin(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=realm_name,
user_realm_name="master",
verify=True,
)
users = keycloak_admin.get_users({"username": username, "exact": True})
if not users:
users = keycloak_admin.get_users({"email": username, "exact": True})
if not users:
continue
keycloak_user_id = users[0]["id"]
# Verificar la contraseña contra este realm (una sola vez)
if not credentials_verified:
keycloak_client = KeycloakOpenID(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=realm_name,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
try:
keycloak_client.token(
username=username,
password=password,
grant_type=["password"],
)
credentials_verified = True
except KeycloakError:
# Contraseña incorrecta — no revelar que el usuario existe
raise HTTPException(status_code=401, detail="Invalid credentials")
# Recopilar tenants con acceso confirmado
for tenant in realm_tenants:
has_access = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.tenant_id == tenant.id,
UserTenant.is_active,
)
)
.first()
)
if has_access:
matched_tenants.append(
{"id": tenant.id, "name": tenant.name, "slug": tenant.slug}
)
except HTTPException:
raise
except Exception as e:
logger.warning(f"Could not query realm '{realm_name}' during credential check: {e}")
continue
if not credentials_verified:
raise HTTPException(status_code=401, detail="Invalid credentials")
return matched_tenants
def discover_user_tenants(self, username: str) -> list:
"""
[DEPRECATED] Usa _verify_credentials_and_list_tenants en su lugar.
Descubre los tenants activos a los que pertenece un usuario dado su username.
"""
from api.v1.modules.core.tenants.models import Tenant
from api.v1.modules.core.user_tenant.models import UserTenant
from sqlalchemy import and_
# 1. Obtener todos los tenants activos
tenants = self.db.query(Tenant).filter(Tenant.is_active).all()
if not tenants:
return []
# 2. Agrupar tenants por keycloak_realm para no repetir consultas admin
realms: dict[str, list] = {}
for tenant in tenants:
realms.setdefault(tenant.keycloak_realm, []).append(tenant)
matched_tenants = []
for realm_name, realm_tenants in realms.items():
try:
keycloak_admin = KeycloakAdmin(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=realm_name,
user_realm_name="master",
verify=True,
)
# Buscar por username exacto
users = keycloak_admin.get_users({"username": username, "exact": True})
if not users:
# Intentar por email
users = keycloak_admin.get_users({"email": username, "exact": True})
if not users:
continue
keycloak_user_id = users[0]["id"]
# 3. Para cada tenant en este realm, verificar UserTenant
for tenant in realm_tenants:
has_access = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.tenant_id == tenant.id,
UserTenant.is_active,
)
)
.first()
)
if has_access:
matched_tenants.append(
{"id": tenant.id, "name": tenant.name, "slug": tenant.slug}
)
except Exception as e:
logger.warning(
f"Could not query realm '{realm_name}' during tenant discovery: {e}"
)
continue
return matched_tenants
logger.error(f"SSO exchange error: {str(e)}")
raise HTTPException(status_code=500, detail="SSO exchange error")

9
backend/api/v1/modules/core/permissions/routes.py
View File

@@ -8,7 +8,7 @@ from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from core.security import collect_user_role_names, get_current_user, validate_access_to_resource
from .dependencies import (
PermissionChecker,
RequirePermission,
@@ -64,11 +64,8 @@ async def get_my_permissions(
user_id = current_user.get("sub") or current_user.get("id")
# 2. Determinar si es un admin de Keycloak para forzar bootstrap si es necesario
realm_roles = current_user.get("realm_access", {}).get("roles", [])
client_roles = []
for client in current_user.get("resource_access", {}).values():
client_roles.extend(client.get("roles", []))
is_keycloak_admin = "admin" in realm_roles or "admin" in client_roles
# Roles vienen del Hub (/auth/me vía verify_token); ver collect_user_role_names.
is_keycloak_admin = "admin" in collect_user_role_names(current_user)
# 3. Bootstrap: si la compañía no tiene roles, o si el usuario es admin, o si estamos en desarrollo y el usuario no tiene roles
from .models import CompanyRole, UserCompanyRole

77
backend/api/v1/modules/core/users/routes.py
View File

@@ -37,28 +37,22 @@ _AVATAR_EXT = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
@router.get("/stats", response_model=UserStatsDTO)
def get_user_statistics(
async def get_user_statistics(
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtiene estadísticas de usuarios del tenant actual
Muestra:
- Total de usuarios
- Usuarios activos e inactivos
- Límite de licencia
- Usuarios disponibles
- Porcentaje de uso
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.view"])
service = UserService(db, tenant_id, company_id)
return service.get_user_stats()
return service.get_user_stats() # Este no es async en service.py
@router.get("/", response_model=UserListResponseDTO)
def list_users(
async def list_users(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(20, ge=1, le=100, description="Tamaño de página"),
@@ -68,12 +62,10 @@ def list_users(
):
"""
Lista todos los usuarios del tenant con paginación
Se puede filtrar por término de búsqueda (busca en username, email, nombre)
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.view"])
service = UserService(db, tenant_id, company_id)
result = service.get_tenant_users(page=page, page_size=page_size, search=search)
result = await service.get_tenant_users(page=page, page_size=page_size, search=search)
return result
@@ -124,7 +116,7 @@ def get_user_avatar_image(
@router.get("/{user_id}", response_model=UserResponseDTO)
def get_user(
async def get_user_detail(
user_id: str,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
@@ -132,34 +124,25 @@ def get_user(
):
"""
Obtiene información detallada de un usuario específico
El usuario debe pertenecer al tenant actual
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.view"])
service = UserService(db, tenant_id, company_id)
return service.get_user(user_id)
return await service.get_user(user_id)
@router.post("/", response_model=UserResponseDTO, status_code=201)
def create_user(
async def create_new_user(
data: CreateUserRequestDTO,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Crea un nuevo usuario en Keycloak y lo asocia al tenant
Validaciones:
- Verifica que no se exceda el límite de usuarios de la licencia
- Verifica que el email y username sean únicos
- Crea el usuario con contraseña temporal
Nota: El tenant_id se obtiene automáticamente del servicio (del token del usuario actual)
Crea un nuevo usuario a través del Hub y lo asocia al tenant
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.create"])
service = UserService(db, tenant_id, company_id)
user = service.create_user(
user = await service.create_user(
email=data.email,
username=data.username,
first_name=data.first_name,
@@ -173,7 +156,7 @@ def create_user(
@router.put("/{user_id}", response_model=UserResponseDTO)
def update_user(
async def update_user_detail(
user_id: str,
data: UpdateUserRequestDTO,
company_id: int = Query(..., description="Company ID"),
@@ -182,17 +165,10 @@ def update_user(
):
"""
Actualiza información de un usuario
Puede actualizar:
- Datos personales (nombre, apellido, email)
- Estado (habilitado/deshabilitado)
- Verificación de email
- Rol en el tenant
- Perfil (avatar, teléfono, bio, preferencias)
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.update"])
service = UserService(db, tenant_id, company_id)
user = service.update_user(
user = await service.update_user(
user_id=user_id,
first_name=data.first_name,
last_name=data.last_name,
@@ -209,7 +185,7 @@ def update_user(
@router.delete("/{user_id}")
def delete_user(
async def delete_user_route(
user_id: str,
company_id: int = Query(..., description="Company ID"),
soft_delete: bool = Query(
@@ -221,18 +197,15 @@ def delete_user(
):
"""
Elimina un usuario del tenant
- soft_delete=True: Solo desactiva la relación (recomendado)
- soft_delete=False: Elimina permanentemente de Keycloak
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.delete"])
service = UserService(db, tenant_id, company_id)
service.delete_user(user_id, soft_delete=soft_delete)
await service.delete_user(user_id, soft_delete=soft_delete)
return {"message": "User deleted successfully"}
@router.post("/{user_id}/change-password")
def change_user_password(
async def change_user_password(
user_id: str,
data: ChangePasswordRequestDTO,
company_id: int = Query(..., description="Company ID"),
@@ -240,14 +213,11 @@ def change_user_password(
current_user: dict = Depends(get_current_user),
):
"""
Cambia la contraseña de un usuario
- temporary=True: Usuario debe cambiar la contraseña en el próximo login
- temporary=False: Contraseña permanente
Cambia la contraseña de un usuario a través del Hub
"""
tenant_id = validate_access_to_resource(db, company_id, current_user, required_permissions=["user.update"])
service = UserService(db, tenant_id, company_id)
service.change_password(user_id, data.password, data.temporary)
await service.change_password(user_id, data.password, data.temporary)
return {"message": "Password changed successfully"}
@@ -255,13 +225,12 @@ def change_user_password(
@router.get("/me/profile", response_model=UserResponseDTO)
def get_my_profile(
async def get_my_profile(
current_user: dict = Depends(get_current_user),
db: Session = Depends(get_core_db),
):
"""
Obtiene el perfil completo del usuario actual
Incluye datos de Keycloak y datos de perfil (avatar, bio, etc.)
"""
keycloak_user_id = current_user.get("sub")
if not keycloak_user_id:
@@ -283,21 +252,17 @@ def get_my_profile(
)
service = UserService(db, user_tenant.tenant_id, user_tenant.company_id)
return service.get_current_user_profile(keycloak_user_id)
return await service.get_current_user_profile(keycloak_user_id)
@router.put("/me/profile", response_model=UserResponseDTO)
def update_my_profile(
async def update_my_profile(
data: UpdateUserRequestDTO,
current_user: dict = Depends(get_current_user),
db: Session = Depends(get_core_db),
):
"""
Actualiza el perfil del usuario actual
Puede actualizar:
- Datos de Keycloak: nombre, apellido, email
- Datos de perfil: avatar, teléfono, biografía, preferencias
"""
keycloak_user_id = current_user.get("sub")
if not keycloak_user_id:
@@ -319,7 +284,7 @@ def update_my_profile(
)
service = UserService(db, user_tenant.tenant_id, user_tenant.company_id)
return service.update_current_user_profile(
return await service.update_current_user_profile(
keycloak_user_id=keycloak_user_id,
first_name=data.first_name,
last_name=data.last_name,

504
backend/api/v1/modules/core/users/service.py
View File

@@ -1,13 +1,9 @@
"""
Servicio para gestionar usuarios de Keycloak con validación de licencias
"""
import logging
import httpx
from datetime import datetime
from typing import Any, Dict, List, Optional
from fastapi import HTTPException
from keycloak import KeycloakAdmin, KeycloakError
from sqlalchemy import and_, func
from sqlalchemy.orm import Session
@@ -19,24 +15,22 @@ from ..user_tenant.models import UserTenant
logger = logging.getLogger(__name__)
def _normalize_keycloak_user(
def _normalize_user(
user_data: Dict[str, Any],
role: Optional[str] = None,
user_tenant: Optional[Any] = None,
) -> Dict[str, Any]:
"""
Normaliza los datos de usuario de Keycloak al formato esperado por el DTO
Keycloak usa camelCase, nuestro DTO usa snake_case
Normaliza los datos de usuario al formato esperado por el DTO
"""
normalized = {
"id": user_data.get("id"),
"username": user_data.get("username", ""),
"id": user_data.get("id") or user_data.get("sub"),
"username": user_data.get("username") or user_data.get("preferred_username", ""),
"email": user_data.get("email", ""),
"first_name": user_data.get("firstName", ""),
"last_name": user_data.get("lastName", ""),
"enabled": user_data.get("enabled", False),
"email_verified": user_data.get("emailVerified", False),
"first_name": user_data.get("firstName") or user_data.get("name", "").split(" ")[0],
"last_name": user_data.get("lastName") or (" ".join(user_data.get("name", "").split(" ")[1:]) if " " in user_data.get("name", "") else ""),
"enabled": user_data.get("enabled", True),
"email_verified": user_data.get("emailVerified") or user_data.get("email_verified", False),
"created_timestamp": user_data.get("createdTimestamp"),
"role": role,
}
@@ -63,22 +57,13 @@ def _normalize_keycloak_user(
class UserService:
"""Servicio para gestionar usuarios en Keycloak"""
"""Servicio para gestionar usuarios vía Hub"""
def __init__(self, db: Session, tenant_id: int, company_id: int = None):
def __init__(self, db: Session, tenant_id: int = None, company_id: int = None):
self.db = db
self.tenant_id = tenant_id
self.company_id = company_id
# Inicializar cliente admin de Keycloak
self.keycloak_admin = KeycloakAdmin(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=settings.KEYCLOAK_REALM,
verify=True,
)
def _get_license(self) -> License:
"""Obtiene la licencia del tenant actual"""
license = (
@@ -126,7 +111,7 @@ class UserService:
f"Currently active: {active_users}. Please upgrade your license.",
)
def create_user(
async def create_user(
self,
email: str,
username: str,
@@ -138,74 +123,44 @@ class UserService:
email_verified: bool = False,
) -> Dict[str, Any]:
"""
Crea un nuevo usuario en Keycloak y lo asocia al tenant
Args:
email: Email del usuario
username: Nombre de usuario
first_name: Nombre
last_name: Apellido
password: Contraseña inicial
role: Rol en el tenant
enabled: Si el usuario está habilitado
email_verified: Si el email está verificado
Returns:
Información del usuario creado
Raises:
HTTPException: Si se alcanza el límite de usuarios o falla la creación
Crea un nuevo usuario a través del Hub y lo asocia localmente
"""
# Verificar límite de usuarios
self._check_user_limit()
try:
# Crear usuario en Keycloak
new_user = {
"email": email,
"username": username,
"enabled": enabled,
"emailVerified": email_verified,
"firstName": first_name,
"lastName": last_name,
"attributes": {
"tenant_id": [
str(self.tenant_id)
] # Atributo requerido por Keycloak
},
"credentials": [
{
"type": "password",
"value": password,
"temporary": True, # Usuario debe cambiar en primer login
# Mandar al Hub para creación en Keycloak
async with httpx.AsyncClient(timeout=10.0) as client:
hub_response = await client.post(
f"{settings.HUB_URL}api/v1/auth/register",
json={
"email": email,
"username": username,
"first_name": first_name,
"last_name": last_name,
"password": password,
"tenant_slug": "default", # TODO: Get real slug if needed
}
],
}
)
if hub_response.status_code != 201:
logger.error(f"Hub registration failed: {hub_response.text}")
raise HTTPException(status_code=hub_response.status_code, detail="Failed to create user in Hub")
user_id = self.keycloak_admin.create_user(new_user)
logger.info(f"User created in Keycloak: {user_id}")
user_data = hub_response.json()
user_id = user_data.get("user_id")
# Obtener company_id si no se proporcionó
if not self.company_id:
# Obtener la primera company del tenant
from api.v1.modules.a76.general_catalogs.company.models import Company
company = (
self.db.query(Company)
.filter(Company.tenant_id == self.tenant_id)
.first()
)
company = self.db.query(Company).filter(Company.tenant_id == self.tenant_id).first()
if not company:
raise HTTPException(
status_code=400,
detail="No company found for this tenant. Please create a company first.",
)
raise HTTPException(status_code=400, detail="No company found")
company_id = company.id
else:
company_id = self.company_id
# Crear relación con el tenant
# Crear relación local
user_tenant = UserTenant(
keycloak_user_id=user_id,
tenant_id=self.tenant_id,
@@ -216,36 +171,16 @@ class UserService:
self.db.add(user_tenant)
self.db.commit()
# Obtener información completa del usuario
user_info = self.keycloak_admin.get_user(user_id)
return _normalize_user(user_data, role, user_tenant)
return _normalize_keycloak_user(user_info, role, user_tenant)
except KeycloakError as e:
logger.error(f"Keycloak error creating user: {str(e)}")
self.db.rollback()
# Manejar errores específicos
if "User exists with same email" in str(e):
raise HTTPException(
status_code=409, detail="A user with this email already exists"
)
elif "User exists with same username" in str(e):
raise HTTPException(
status_code=409, detail="A user with this username already exists"
)
raise HTTPException(
status_code=500, detail=f"Error creating user in Keycloak: {str(e)}"
)
except Exception as e:
logger.error(f"Unexpected error creating user: {str(e)}")
logger.error(f"Error creating user: {str(e)}")
self.db.rollback()
raise HTTPException(
status_code=500, detail=f"Error creating user: {str(e)}"
)
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=str(e))
def get_tenant_users(
async def get_tenant_users(
self, page: int = 1, page_size: int = 20, search: Optional[str] = None
) -> Dict[str, Any]:
"""
@@ -295,41 +230,18 @@ class UserService:
user_roles_map[user_role.user_id] = []
user_roles_map[user_role.user_id].append(user_role.company_role.name)
# Obtener información de Keycloak para cada usuario
users = []
for ut in user_tenants:
try:
user_info = self.keycloak_admin.get_user(ut.keycloak_user_id)
# Obtener roles del usuario
roles = user_roles_map.get(ut.keycloak_user_id, [])
role_str = ", ".join(roles) if roles else None
normalized_user = _normalize_keycloak_user(user_info, role_str, ut)
# En lugar de consultar Keycloak uno a uno (lento y sin API directa ahora),
# devolvemos la info local mínima o consultamos un endpoint de "buscar varios" en el Hub si existiera.
# Por ahora, minimizamos el impacto devolviendo lo que tenemos local.
normalized_user = _normalize_user({
"id": ut.keycloak_user_id,
"username": "User", # Placeholder si no tenemos el dato local
}, role_str, ut)
# Filtrar por búsqueda si se proporciona
if search:
search_lower = search.lower()
if (
search_lower in normalized_user.get("username", "").lower()
or search_lower in normalized_user.get("email", "").lower()
or search_lower
in normalized_user.get("first_name", "").lower()
or search_lower
in normalized_user.get("last_name", "").lower()
or search_lower
in normalized_user.get("phone", "").lower()
or search_lower
in normalized_user.get("bio", "").lower()
):
users.append(normalized_user)
else:
users.append(normalized_user)
except KeycloakError as e:
logger.warning(
f"Could not fetch user {ut.keycloak_user_id} from Keycloak: {str(e)}"
)
users.append(normalized_user)
except Exception as e:
logger.warning(f"Error processing user {ut.keycloak_user_id}: {e}")
continue
total_pages = (total + page_size - 1) // page_size
@@ -348,31 +260,24 @@ class UserService:
status_code=500, detail=f"Error getting users: {str(e)}"
)
def get_user(self, user_id: str) -> Dict[str, Any]:
"""Obtiene un usuario específico del tenant"""
async def get_user(self, user_id: str) -> Dict[str, Any]:
"""Obtiene un usuario específico"""
from ..permissions.models import UserCompanyRole
from sqlalchemy.orm import joinedload
# Verificar que el usuario pertenece al tenant
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == user_id,
UserTenant.tenant_id == self.tenant_id,
UserTenant.is_active == True,
)
user_tenant = self.db.query(UserTenant).filter(
and_(
UserTenant.keycloak_user_id == user_id,
UserTenant.tenant_id == self.tenant_id,
UserTenant.is_active == True,
)
.first()
)
).first()
if not user_tenant:
raise HTTPException(status_code=404, detail="User not found in this tenant")
raise HTTPException(status_code=404, detail="User not found")
# Obtener roles del usuario en la compañía actual
user_roles = self.db.query(UserCompanyRole).options(
joinedload(UserCompanyRole.company_role)
).filter(
# Roles locales
user_roles = self.db.query(UserCompanyRole).options(joinedload(UserCompanyRole.company_role)).filter(
and_(
UserCompanyRole.user_id == user_id,
UserCompanyRole.company_id == self.company_id,
@@ -380,165 +285,58 @@ class UserService:
UserCompanyRole.is_active == True
)
).all()
roles = [ur.company_role.name for ur in user_roles]
role_str = ", ".join(roles) if roles else None
try:
user_info = self.keycloak_admin.get_user(user_id)
return _normalize_keycloak_user(user_info, role_str, user_tenant)
except KeycloakError as e:
logger.error(f"Error getting user from Keycloak: {str(e)}")
raise HTTPException(status_code=404, detail="User not found in Keycloak")
# TODO: Call Hub if more info is needed
return _normalize_user({"id": user_id}, role_str, user_tenant)
def update_user(
self,
user_id: str,
first_name: Optional[str] = None,
last_name: Optional[str] = None,
email: Optional[str] = None,
enabled: Optional[bool] = None,
email_verified: Optional[bool] = None,
role: Optional[str] = None,
avatar_url: Optional[str] = None,
phone: Optional[str] = None,
bio: Optional[str] = None,
preferences: Optional[dict] = None,
) -> Dict[str, Any]:
"""Actualiza información de un usuario"""
# Verificar que el usuario pertenece al tenant
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == user_id,
UserTenant.tenant_id == self.tenant_id,
)
)
.first()
)
async def update_user(self, user_id: str, **kwargs) -> Dict[str, Any]:
"""Actualiza información local del usuario (e identidad vía Hub si se implementa)"""
user_tenant = self.db.query(UserTenant).filter(
and_(UserTenant.keycloak_user_id == user_id, UserTenant.tenant_id == self.tenant_id)
).first()
if not user_tenant:
raise HTTPException(status_code=404, detail="User not found in this tenant")
raise HTTPException(status_code=404, detail="User not found")
try:
# Preparar datos de actualización para Keycloak
update_data = {}
if first_name is not None:
update_data["firstName"] = first_name
if last_name is not None:
update_data["lastName"] = last_name
if email is not None:
update_data["email"] = email
if enabled is not None:
update_data["enabled"] = enabled
if email_verified is not None:
update_data["emailVerified"] = email_verified
# Actualizar campos locales
for field in ["role", "avatar_url", "phone", "bio", "preferences"]:
if field in kwargs and kwargs[field] is not None:
setattr(user_tenant, field, kwargs[field])
# Actualizar en Keycloak si hay cambios
if update_data:
self.keycloak_admin.update_user(user_id, update_data)
self.db.commit()
self.db.refresh(user_tenant)
return _normalize_user({"id": user_id}, user_tenant.role, user_tenant)
# Actualizar campos en UserTenant
if role is not None:
user_tenant.role = role
if avatar_url is not None and not str(avatar_url).startswith(
"/api/v1/core/users/avatar/"
):
user_tenant.avatar_url = avatar_url
if phone is not None:
user_tenant.phone = phone
if bio is not None:
user_tenant.bio = bio
if preferences is not None:
user_tenant.preferences = preferences
async def delete_user(self, user_id: str, soft_delete: bool = True) -> None:
"""Elimina/Desactiva usuario"""
user_tenant = self.db.query(UserTenant).filter(
and_(UserTenant.keycloak_user_id == user_id, UserTenant.tenant_id == self.tenant_id)
).first()
if not user_tenant:
raise HTTPException(status_code=404, detail="User not found")
if soft_delete:
user_tenant.is_active = False
self.db.commit()
else:
# TODO: Call Hub to delete from Keycloak
self.db.delete(user_tenant)
self.db.commit()
self.db.refresh(user_tenant)
# Obtener información actualizada
user_info = self.keycloak_admin.get_user(user_id)
return _normalize_keycloak_user(user_info, user_tenant.role, user_tenant)
except KeycloakError as e:
logger.error(f"Error updating user in Keycloak: {str(e)}")
self.db.rollback()
raise HTTPException(
status_code=500, detail=f"Error updating user: {str(e)}"
)
def delete_user(self, user_id: str, soft_delete: bool = True) -> None:
"""
Elimina un usuario del tenant
Args:
user_id: ID del usuario en Keycloak
soft_delete: Si es True, solo desactiva. Si es False, elimina de Keycloak
"""
# Verificar que el usuario pertenece al tenant
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == user_id,
UserTenant.tenant_id == self.tenant_id,
)
)
.first()
)
if not user_tenant:
raise HTTPException(status_code=404, detail="User not found in this tenant")
async def change_password(self, user_id: str, password: str, temporary: bool = True) -> None:
"""Cambia contraseña vía Hub"""
try:
if soft_delete:
# Solo desactivar la relación
user_tenant.is_active = False
self.db.commit()
else:
# Eliminar permanentemente de Keycloak
self.keycloak_admin.delete_user(user_id)
# Eliminar relación
self.db.delete(user_tenant)
self.db.commit()
except KeycloakError as e:
logger.error(f"Error deleting user from Keycloak: {str(e)}")
self.db.rollback()
raise HTTPException(
status_code=500, detail=f"Error deleting user: {str(e)}"
)
def change_password(
self, user_id: str, password: str, temporary: bool = True
) -> None:
"""Cambia la contraseña de un usuario"""
# Verificar que el usuario pertenece al tenant
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == user_id,
UserTenant.tenant_id == self.tenant_id,
UserTenant.is_active == True,
async with httpx.AsyncClient(timeout=10.0) as client:
await client.post(
f"{settings.HUB_URL}api/v1/auth/change-password",
json={"user_id": user_id, "password": password, "temporary": temporary}
)
)
.first()
)
if not user_tenant:
raise HTTPException(status_code=404, detail="User not found in this tenant")
try:
self.keycloak_admin.set_user_password(
user_id, password, temporary=temporary
)
except KeycloakError as e:
logger.error(f"Error changing user password: {str(e)}")
raise HTTPException(
status_code=500, detail=f"Error changing password: {str(e)}"
)
except Exception as e:
logger.error(f"Error changing password: {e}")
raise HTTPException(status_code=500, detail="Error changing password")
def get_user_stats(self) -> Dict[str, Any]:
"""Obtiene estadísticas de usuarios del tenant"""
@@ -582,95 +380,19 @@ class UserService:
"usage_percentage": round(usage_percentage, 2),
}
def get_current_user_profile(self, keycloak_user_id: str) -> Dict[str, Any]:
"""
Obtiene el perfil completo del usuario actual
Combina datos de Keycloak con datos de UserTenant
"""
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.is_active == True,
)
)
.first()
)
async def get_current_user_profile(self, keycloak_user_id: str) -> Dict[str, Any]:
"""Obtiene el perfil completo del usuario actual"""
# Reutilizamos verify_token para obtener info del Hub
from core.security import verify_token
user_info = await verify_token(keycloak_user_id) # keycloak_user_id es el token en este contexto, o el ID
# Nota: en routes.py se pasa el ID. Si necesitamos info real, pedimos al Hub.
user_tenant = self.db.query(UserTenant).filter(
and_(UserTenant.keycloak_user_id == keycloak_user_id, UserTenant.is_active == True)
).first()
if not user_tenant:
raise HTTPException(status_code=404, detail="User profile not found")
return _normalize_user(user_info, user_tenant.role if user_tenant else None, user_tenant)
try:
user_info = self.keycloak_admin.get_user(keycloak_user_id)
return _normalize_keycloak_user(user_info, user_tenant.role, user_tenant)
except KeycloakError as e:
logger.error(f"Error getting user from Keycloak: {str(e)}")
raise HTTPException(status_code=404, detail="User not found")
def update_current_user_profile(
self,
keycloak_user_id: str,
first_name: Optional[str] = None,
last_name: Optional[str] = None,
email: Optional[str] = None,
avatar_url: Optional[str] = None,
phone: Optional[str] = None,
bio: Optional[str] = None,
preferences: Optional[dict] = None,
) -> Dict[str, Any]:
"""
Actualiza el perfil del usuario actual
"""
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.is_active == True,
)
)
.first()
)
if not user_tenant:
raise HTTPException(status_code=404, detail="User profile not found")
try:
# Actualizar Keycloak
update_data = {}
if first_name is not None:
update_data["firstName"] = first_name
if last_name is not None:
update_data["lastName"] = last_name
if email is not None:
update_data["email"] = email
if update_data:
self.keycloak_admin.update_user(keycloak_user_id, update_data)
# Actualizar campos de perfil en UserTenant
if avatar_url is not None and not str(avatar_url).startswith(
"/api/v1/core/users/avatar/"
):
user_tenant.avatar_url = avatar_url
if phone is not None:
user_tenant.phone = phone
if bio is not None:
user_tenant.bio = bio
if preferences is not None:
user_tenant.preferences = preferences
self.db.commit()
self.db.refresh(user_tenant)
# Retornar perfil actualizado
user_info = self.keycloak_admin.get_user(keycloak_user_id)
return _normalize_keycloak_user(user_info, user_tenant.role, user_tenant)
except KeycloakError as e:
logger.error(f"Error updating user profile: {str(e)}")
self.db.rollback()
raise HTTPException(
status_code=500, detail=f"Error updating profile: {str(e)}"
)
async def update_current_user_profile(self, keycloak_user_id: str, **kwargs) -> Dict[str, Any]:
"""Actualiza el perfil del usuario actual"""
return await self.update_user(keycloak_user_id, **kwargs)

29
backend/core/config.py
View File

@@ -24,15 +24,7 @@ class Settings(BaseSettings):
CORE_DB_USER: str = "postgres"
CORE_DB_PASSWORD: str = "postgres"
TEST_DATABASE_URL: str = "postgresql://postgres:postgres@localhost:5432/anexo76_core"
# Keycloak
KEYCLOAK_SERVER_URL: str = "http://localhost:8080/kcauth"
KEYCLOAK_REALM: str = "master"
KEYCLOAK_CLIENT_ID: str = "anexo76-backend"
KEYCLOAK_CLIENT_SECRET: str = ""
KEYCLOAK_ADMIN_USERNAME: str = "admin"
KEYCLOAK_ADMIN_PASSWORD: str = "admin"
# Security
SECRET_KEY: str = "change-this-secret-key-in-production"
@@ -47,9 +39,19 @@ class Settings(BaseSettings):
# CORS
CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000"
# License
LICENSE_CHECK_ENABLED: bool = True
# Hub de Aduanasoft — requerido siempre (SaaS y self-hosted)
HUB_URL: str = "http://localhost:8001"
@field_validator("CENTRAL_SERVER_URL", "SPOKE_URLS", "HUB_URL", mode="before")
@classmethod
def strip_quotes(cls, v: str) -> str:
if v and isinstance(v, str):
v = v.strip().strip('"').strip("'")
if not v.endswith("/"):
v += "/"
return v
return v
# External APIs
SITAR_API_URL: str = "api.sitar.aduanasoft.com:880"
COVE_API_URL: str = "https://api.vu.aduanasoft.com"
@@ -85,13 +87,6 @@ class Settings(BaseSettings):
env_file_encoding="utf-8",
)
@field_validator("CENTRAL_SERVER_URL", "SPOKE_URLS", mode="before")
@classmethod
def strip_quotes(cls, v: str) -> str:
if v:
return v.strip().strip('"').strip("'")
return v
@property
def core_database_url(self) -> str:
"""URL de conexión a la base de datos core"""

256
backend/core/middleware.py
View File

@@ -1,18 +1,42 @@
import logging
import time
import httpx
from datetime import datetime, timezone
from typing import Callable, Optional
from fastapi import Request
from fastapi import Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from .config import settings
from .database import scoped_core_db
from .security import get_tenant_from_token, verify_token
logger = logging.getLogger(__name__)
def _normalize_text(value: str | None) -> str:
if not value:
return ""
return str(value).strip().lower()
def _is_token_issue_message(*values: str | None) -> bool:
text = " ".join(_normalize_text(v) for v in values if v)
if not text:
return False
token_markers = ["token", "jwt", "bearer", "access"]
invalid_markers = [
"invalido", "inválido", "invalid", "not valid", "malformed", "signature", "unauthorized"
]
expired_markers = ["expirado", "expirada", "expired", "has expired", "caducado", "vencido"]
has_token_context = any(marker in text for marker in token_markers)
has_invalid_marker = any(marker in text for marker in invalid_markers)
has_expired_marker = any(marker in text for marker in expired_markers)
return has_expired_marker or (has_token_context and has_invalid_marker)
def _extract_company_id(request: Request) -> Optional[int]:
"""Obtiene ``company_id`` activa desde header ``X-Company-Id`` o cookie.
@@ -29,8 +53,10 @@ def _extract_company_id(request: Request) -> Optional[int]:
except (TypeError, ValueError):
return None
class TenantMiddleware(BaseHTTPMiddleware):
"""
Middleware original para extraer tenant_id y user_info del token.
"""
async def dispatch(self, request: Request, call_next: Callable):
doc_prefixes = ["/api/redoc", "/api/openapi.json"]
public_prefixes = [
@@ -64,7 +90,7 @@ class TenantMiddleware(BaseHTTPMiddleware):
token = auth_header.split(" ")[1]
try:
user_info = verify_token(token)
user_info = await verify_token(token)
tenant_id = get_tenant_from_token(user_info)
request.state.tenant_id = tenant_id
@@ -85,107 +111,203 @@ class TenantMiddleware(BaseHTTPMiddleware):
class LicenseValidationMiddleware(BaseHTTPMiddleware):
"""Middleware para validar la licencia del tenant antes de procesar requests."""
"""
Middleware que valida la licencia contra el Hub de Aduanasoft.
El Hub siempre es requerido — tanto en SaaS como en self-hosted.
Fail-closed: si el Hub no responde o la licencia es inválida, se bloquea el acceso.
"""
async def dispatch(self, request: Request, call_next: Callable):
if not settings.LICENSE_CHECK_ENABLED or settings.ENVIRONMENT == "development":
return await call_next(request)
exempt_paths = [
"/api/docs",
"/api/redoc",
"/openapi.json",
"/api/v1/auth",
"/api/v1/auth",
"/api/v1/status",
"/api/v1/status",
"/api/health",
"/api/",
"/api/docs", "/api/redoc", "/openapi.json",
"/api/v1/auth", "/api/v1/status", "/api/health",
"/api/v1/core/help-center",
"/api/v1/core/users/avatar",
]
is_exempt = False
for path in exempt_paths:
if request.url.path == path or (
path != "/" and request.url.path.startswith(path)
):
is_exempt = True
break
is_exempt = any(
request.url.path == path or (path != "/" and request.url.path.startswith(path))
for path in exempt_paths
)
if is_exempt:
return await call_next(request)
tenant_id = getattr(request.state, "tenant_id", None)
if not tenant_id:
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
# Permitimos pasar para que TenantMiddleware maneje el 401
return await call_next(request)
token = auth_header.split(" ")[1]
tenant_override = request.headers.get("X-Tenant-Override")
if not tenant_override:
# Fallback para flujos SSO cuando el override no viaja en header.
tenant_override = request.cookies.get("sso_tenant_id") or request.cookies.get("sso_tenant_pub")
# TenantMiddleware (corre antes) ya resolvió el token y dejó tenant en user_info.
# Sin esto, Swagger/curl sin cookies SSO llaman verify-license sin contexto y el Hub
# puede devolver 401 aunque /auth/me con el mismo Bearer responda 200.
if not tenant_override:
user_info = getattr(request.state, "user_info", None)
if isinstance(user_info, dict):
tid = user_info.get("tenant_id")
if tid is not None and str(tid).strip() != "":
tenant_override = str(tid)
hub_headers = {"Authorization": f"Bearer {token}"}
if tenant_override:
hub_headers["X-Tenant-Override"] = str(tenant_override)
logger.info("[license] tenant override propagated to Hub: %s", tenant_override)
# core.licenses / core.license_usage están bajo RLS por tenant_id:
# se abre la sesión con contexto explícito para que LicenseService
# vea las filas del tenant actual.
try:
with scoped_core_db(tenant_id=tenant_id) as db:
from api.v1.modules.core.licenses.service import LicenseService
# Validación contra el Hub Central
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(
f"{settings.HUB_URL}api/v1/auth/verify-license",
headers=hub_headers
)
license_service = LicenseService(db)
license_info = license_service.validate_license(tenant_id)
logger.info(f"🔑 verify-license → status={response.status_code} body={response.text[:300]}")
if not license_info["is_valid"]:
if response.status_code == 404:
# Endpoint no existe en este Hub — dejar pasar
return await call_next(request)
if response.status_code == 200:
data = response.json()
# Escenario 1: sin licencia asignada o licencia inactiva
if not data.get("valid", False):
message = data.get("message", "Sin licencia asignada para este tenant")
detail = data.get("detail")
reason = data.get("reason")
# Si el Hub reporta token inválido/expirado, devolver 401 para que
# el frontend dispare el auto-refresh (solo se activa con 401/403, no 402).
if _is_token_issue_message(message, detail, reason):
logger.warning(
"[license] token expirado/invalido detectado por verify-license; devolviendo 401 para silent refresh | message=%s detail=%s reason=%s",
message,
detail,
reason,
)
return JSONResponse(
status_code=401,
content={
"error": "TOKEN_EXPIRED",
"message": message,
"status_code": 401,
}
)
logger.warning(
"[license] licencia invalida para tenant=%s | message=%s",
data.get("tenant_slug"),
message,
)
return JSONResponse(
status_code=402,
content={
"error": "HTTP_ERROR",
"message": f"License validation failed: {license_info['reason']}",
"error": "LICENSE_ERROR",
"message": message,
"status_code": 402,
}
)
request.state.license_info = license_info
except Exception as e:
logger.error(f"License validation error: {str(e)}")
# Escenario 2: licencia vencida (verificación local de expires_at)
expires_at_str = data.get("expires_at")
if expires_at_str:
try:
expires_at = datetime.fromisoformat(expires_at_str.replace("Z", "+00:00"))
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
if expires_at < datetime.now(timezone.utc):
logger.warning(
"[license] licencia expirada para tenant=%s | expires_at=%s",
data.get("tenant_slug"),
expires_at_str,
)
return JSONResponse(
status_code=402,
content={
"error": "LICENSE_EXPIRED",
"message": f"La licencia venció el {expires_at.strftime('%d/%m/%Y')}. Renueva tu suscripción.",
"status_code": 402,
}
)
except (ValueError, TypeError):
pass # Si no se puede parsear, dejamos pasar — el Hub es la fuente de verdad
request.state.license_info = data
return await call_next(request) # <--- Único camino al éxito
elif response.status_code == 401:
logger.warning("[license] Hub verify-license devolvio 401 (token invalido/expirado)")
return JSONResponse(
status_code=401,
content={
"error": "TOKEN_EXPIRED",
"message": "Token inválido o expirado.",
"status_code": 401,
}
)
elif response.status_code == 403:
return JSONResponse(
status_code=403,
content={
"error": "FORBIDDEN",
"message": "El Tenant no tiene permisos en el Hub central.",
"status_code": 403,
}
)
else:
logger.error(f"Hub error status: {response.status_code}")
return JSONResponse(
status_code=503,
content={
"error": "HUB_ERROR",
"message": "Error en el servidor de licencias.",
"status_code": 503,
}
)
except (httpx.ConnectError, httpx.TimeoutException) as e:
logger.critical(f"❌ CRITICAL: Hub unreachable: {str(e)}")
return JSONResponse(
status_code=500,
status_code=503,
content={
"error": "HTTP_ERROR",
"message": "License validation error",
"status_code": 500,
"error": "HUB_OFFLINE",
"message": "Servicio de licencias fuera de línea. Acceso denegado.",
"status_code": 503,
}
)
return await call_next(request)
except Exception as e:
logger.error(f"Unexpected license error: {str(e)}")
return JSONResponse(
status_code=500,
content={"error": "VALIDATION_ERROR", "message": "Error interno de validación.", "status_code": 500}
)
class RequestLoggingMiddleware(BaseHTTPMiddleware):
"""Middleware para logging de requests."""
"""
Middleware original para logging de performance.
"""
async def dispatch(self, request: Request, call_next: Callable):
start_time = time.time()
excluded_paths = [
"/api/docs",
"/api/redoc",
"/openapi.json",
"/api/v1/status",
"/api/health",
]
if any(
request.url.path == path or request.url.path.startswith(path + "/")
for path in excluded_paths
):
excluded_paths = ["/api/docs", "/api/redoc", "/openapi.json", "/api/v1/status", "/api/health"]
if any(request.url.path == path or request.url.path.startswith(path + "/") for path in excluded_paths):
return await call_next(request)
logger.info(f"Request: {request.method} {request.url.path}")
response = await call_next(request)
process_time = time.time() - start_time
logger.info(
f"Response: {request.method} {request.url.path} "
f"Status: {response.status_code} "
f"Duration: {process_time:.3f}s"
)
response.headers["X-Process-Time"] = str(process_time)
return response
return response

256
backend/core/security.py
View File

@@ -3,79 +3,205 @@ Utilidades de seguridad y autenticación con Keycloak
"""
import logging
from typing import Any, Dict, Optional
from typing import Any, Dict, Optional, Set
from fastapi import Depends, HTTPException, Security
from fastapi import Depends, HTTPException, Request, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from keycloak import KeycloakOpenID
import httpx
from cachetools import TTLCache
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from .config import settings
from .database import get_core_db
logger = logging.getLogger(__name__)
# Configuración de Keycloak
keycloak_openid = KeycloakOpenID(
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=settings.KEYCLOAK_REALM,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
# Cache para tokens verificados (1 minuto de TTL, máximo 1000 tokens)
token_cache = TTLCache(maxsize=1000, ttl=60)
# IDs de tenants ya sincronizados en este proceso (evita consultas repetidas)
_synced_tenant_ids: Set[int] = set()
# Alias Hub tenant_id -> tenant_id local cuando existe drift histórico de IDs
# (mismo slug, diferente id).
_tenant_id_aliases: Dict[int, int] = {}
# Security scheme
security = HTTPBearer()
def verify_token(token: str) -> Dict[str, Any]:
async def verify_token(token: str, tenant_id_override: str = None) -> Dict[str, Any]:
"""
Verifica y decodifica un token JWT de Keycloak
Args:
token: Token JWT
Returns:
Payload del token decodificado
Raises:
HTTPException: Si el token es inválido
Verifica un token JWT llamando al Hub central.
"""
# Cache key incluye el override para que distintos tenants no se mezclen
cache_key = (token, tenant_id_override)
if cache_key in token_cache:
return token_cache[cache_key]
try:
# Obtener clave pública de Keycloak
KEYCLOAK_PUBLIC_KEY = (
"-----BEGIN PUBLIC KEY-----\n"
+ keycloak_openid.public_key()
+ "\n-----END PUBLIC KEY-----"
)
headers: Dict[str, str] = {"Authorization": f"Bearer {token}"}
if tenant_id_override:
headers["X-Tenant-Override"] = tenant_id_override
# Decodificar y verificar token
options = {"verify_signature": True, "verify_aud": False, "verify_exp": True}
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(
f"{settings.HUB_URL}api/v1/auth/me",
headers=headers
)
decoded_token = jwt.decode(
token, KEYCLOAK_PUBLIC_KEY, algorithms=["RS256"], options=options
)
return decoded_token
except JWTError as e:
logger.error(f"Token verification failed: {str(e)}")
if response.status_code == 200:
user_info = response.json()
token_cache[cache_key] = user_info
return user_info
logger.error(f"Hub token verification failed with status {response.status_code}")
raise HTTPException(status_code=401, detail="Could not validate credentials")
except httpx.HTTPError as e:
logger.error(f"Hub unreachable or error during token verification: {str(e)}")
raise HTTPException(status_code=503, detail="Authentication service unavailable")
except Exception as e:
logger.error(f"Unexpected error during token verification: {str(e)}")
raise HTTPException(status_code=401, detail="Authentication error")
def _ensure_company_exists(db: Session, tenant_id: int, tenant_name: str) -> None:
"""
Garantiza que exista al menos una empresa en a76.company para el tenant.
El tenant IS la empresa — se crea automáticamente al primer login.
"""
try:
from api.v1.modules.a76.general_catalogs.company.models import Company
exists = db.query(Company).filter(Company.tenant_id == tenant_id).first()
if not exists:
company = Company(tenant_id=tenant_id, name=tenant_name)
db.add(company)
db.commit()
logger.info(f"Empresa creada automáticamente para tenant id={tenant_id}: '{tenant_name}'")
elif exists.name != tenant_name:
exists.name = tenant_name
db.commit()
logger.info(f"Empresa actualizada para tenant id={tenant_id}: '{tenant_name}'")
except Exception as e:
db.rollback()
logger.warning(f"No se pudo crear empresa automática para tenant {tenant_id}: {e}")
def _ensure_tenant_synced(db: Session, tenant_id: int, tenant_slug: str) -> int:
"""
Garantiza que el tenant del Hub exista en core.tenants local.
Se ejecuta una sola vez por tenant_id por ciclo de vida del proceso.
El Hub es la fuente de verdad — este método solo sincroniza en una dirección.
"""
if tenant_id in _synced_tenant_ids:
return _tenant_id_aliases.get(tenant_id, tenant_id)
try:
# Importación local para evitar imports circulares
from api.v1.modules.core.tenants.models import Tenant, TenantType
name = " ".join(word.capitalize() for word in tenant_slug.replace("-", " ").split())
existing = db.query(Tenant).filter(Tenant.id == tenant_id).first()
if existing:
# Update name/slug/keycloak_realm if they differ (Hub is source of truth)
if existing.slug != tenant_slug or existing.name != name or existing.keycloak_realm != tenant_slug:
existing.slug = tenant_slug
existing.name = name
existing.keycloak_realm = tenant_slug
db.commit()
logger.info(f"Tenant id={tenant_id} actualizado: slug='{tenant_slug}'")
_synced_tenant_ids.add(tenant_id)
# Garantizar empresa aunque el tenant ya existiera
_ensure_company_exists(db, tenant_id, name)
return tenant_id
# Crear el tenant local con los datos disponibles del token.
# El Hub siempre crea el realm de Keycloak con el mismo nombre que el slug.
tenant = Tenant(
id=tenant_id,
name=name,
slug=tenant_slug,
type=TenantType.SHARED,
keycloak_realm=tenant_slug,
is_active=True,
)
db.add(tenant)
db.commit()
_synced_tenant_ids.add(tenant_id)
logger.info(f"Tenant '{tenant_slug}' (id={tenant_id}) sincronizado desde Hub a core.tenants")
# Crear la empresa correspondiente al tenant recién sincronizado
_ensure_company_exists(db, tenant_id, name)
return tenant_id
except IntegrityError:
# Puede ser concurrencia o colisión de slug (id diferente, mismo slug)
db.rollback()
from api.v1.modules.core.tenants.models import Tenant
# Si el slug ya existe con diferente id, el tenant real del Hub no está registrado aún.
# Logueamos el conflicto para depuración; el sistema continuará con tenant_id vacío.
stale = db.query(Tenant).filter(Tenant.slug == tenant_slug).first()
if stale and stale.id != tenant_id:
logger.error(
f"Conflicto de tenant: JWT dice id={tenant_id} slug='{tenant_slug}', "
f"pero core.tenants tiene id={stale.id} mismo slug. "
f"Elimine el registro obsoleto con: "
f"DELETE FROM core.tenants WHERE id={stale.id};"
)
# Auto-heal en runtime: mapear temporalmente al tenant local existente por slug
# para evitar dejar al usuario sin compañías y evitar este conflicto en cada request.
_tenant_id_aliases[tenant_id] = int(stale.id)
_synced_tenant_ids.add(tenant_id)
_ensure_company_exists(db, int(stale.id), stale.name or tenant_slug)
return int(stale.id)
else:
_synced_tenant_ids.add(tenant_id)
return tenant_id
except Exception as e:
db.rollback()
logger.warning(f"No se pudo sincronizar tenant {tenant_id} ({tenant_slug}): {e}")
return _tenant_id_aliases.get(tenant_id, tenant_id)
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Security(security),
db: Session = Depends(get_core_db),
request: Request = None,
) -> Dict[str, Any]:
"""
Dependency para obtener el usuario actual desde el token JWT
Dependency para obtener el usuario actual desde el token JWT.
Auto-sincroniza el tenant en core.tenants si fue creado en el Hub
pero aún no existe en la BD local.
Uso en FastAPI:
current_user: dict = Depends(get_current_user)
"""
token = credentials.credentials
user_info = verify_token(token)
# Leer tenant override del header X-Tenant-Override (pasado por el SvelteKit server
# desde la cookie sso_tenant_id, flujo SSO relay multi-tenant)
tenant_override = request.headers.get('X-Tenant-Override') if request else None
logger.info(f"[get_current_user] X-Tenant-Override={tenant_override!r}")
user_info = await verify_token(token, tenant_id_override=tenant_override)
# Copia local para poder normalizar tenant_id sin mutar el objeto cacheado
user_info = dict(user_info)
# Sincronizar tenant desde Hub a BD local (solo la primera vez por tenant)
tenant_id = user_info.get("tenant_id")
tenant_slug = user_info.get("tenant_slug")
if tenant_id and tenant_slug:
effective_tenant_id = _ensure_tenant_synced(db, int(tenant_id), str(tenant_slug))
if effective_tenant_id != int(tenant_id):
logger.warning(
f"[get_current_user] tenant_id ajustado por alias: hub={tenant_id} local={effective_tenant_id} slug={tenant_slug}"
)
user_info["tenant_id"] = effective_tenant_id
return user_info
@@ -103,13 +229,14 @@ def has_role(required_role: str):
async def role_checker(
current_user: Dict[str, Any] = Depends(get_current_user),
) -> Dict[str, Any]:
user_roles = current_user.get("realm_access", {}).get("roles", [])
user_roles = collect_user_role_names(current_user)
if required_role not in user_roles:
logger.warning(f"Role denied. Required: {required_role}. User actually has: {user_roles}")
# Also check client roles as a debug fallback
client_roles = current_user.get("resource_access", {})
logger.warning(f"User client roles: {client_roles}")
logger.warning(
"Role denied. Required: %s. User has: %s",
required_role,
sorted(user_roles),
)
raise HTTPException(
status_code=403,
detail=f"User does not have required role: {required_role}",
@@ -141,6 +268,29 @@ def get_tenant_from_token(user_info: Dict[str, Any]) -> Optional[int]:
return None
def collect_user_role_names(current_user: Dict[str, Any]) -> Set[str]:
"""
Roles del usuario: primero la lista ``roles`` del Hub (GET /api/v1/auth/me
vía verify_token). Si no hay lista no vacía, se unen realm_access y
resource_access del JWT Keycloak clásico.
"""
names: Set[str] = set()
hub_roles = current_user.get("roles")
if isinstance(hub_roles, list):
names.update(str(r) for r in hub_roles if r is not None)
if names:
return names
realm = current_user.get("realm_access")
if isinstance(realm, dict):
names.update(str(r) for r in (realm.get("roles") or []) if r is not None)
for client in (current_user.get("resource_access") or {}).values():
if isinstance(client, dict):
names.update(str(r) for r in (client.get("roles") or []) if r is not None)
return names
def validate_company_access(
db: Session, company_id: int, current_user: Dict[str, Any]
) -> bool:
@@ -208,23 +358,9 @@ def validate_access_to_resource(
tenant_id = get_tenant_from_token(current_user)
if not tenant_id:
tenant_id = current_user.get("tenant_id")
# 🕵️ DEBUG ULTRA-DETALLADO (Ver en consola del backend)
print("--- TOKEN DEBUG START ---")
print(f"Usuario: {current_user.get('preferred_username')}")
print(f"Sub: {current_user.get('sub')}")
print(f"Realm Roles: {current_user.get('realm_access', {}).get('roles', [])}")
print(f"Resource Access: {current_user.get('resource_access', {})}")
print("--- TOKEN DEBUG END ---")
# 🛡️ DETERMINAR SI ES ADMIN DE KEYCLOAK
realm_roles = current_user.get("realm_access", {}).get("roles", [])
# Buscamos en todos los clientes posibles por si acaso
all_client_roles = []
for client in current_user.get("resource_access", {}).values():
all_client_roles.extend(client.get("roles", []))
all_user_roles = set(realm_roles + all_client_roles)
# Admin global Keycloak / master: lista ``roles`` del Hub (/auth/me), con fallback JWT.
all_user_roles = collect_user_role_names(current_user)
is_keycloak_admin = "admin" in all_user_roles
# 🚪 EXCEPCIÓN ESPECIAL: Si es el endpoint /me, permitimos el paso para el Bootstrap

19
backend/main.py
View File

@@ -61,15 +61,6 @@ async def on_startup():
logger.info("Base de datos inicializada correctamente.")
# Configurar CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Agregar middlewares personalizados
if settings.DEBUG:
app.add_middleware(RequestLoggingMiddleware)
@@ -78,6 +69,16 @@ app.add_middleware(LicenseValidationMiddleware)
app.add_middleware(TenantMiddleware)
app.add_middleware(UserContextMiddleware)
# CORS debe ser el último en añadirse para que sea el más externo
# y cubra todas las respuestas, incluyendo las de los middlewares internos
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Centraliza startup para evitar on_event() (deprecated en FastAPI)

2
backend/requirements.txt
View File

@@ -13,7 +13,7 @@ psycopg2-binary==2.9.11
asyncpg==0.30.0
# Authentication & Authorization
python-keycloak==5.8.1
cachetools==5.5.0
python-jose[cryptography]==3.5.0
passlib[bcrypt]==1.7.4

21
backend/tests/unit/core/test_license_middleware.py Normal file
View File

@@ -0,0 +1,21 @@
from core.middleware import _is_token_issue_message
def test_is_token_issue_message_detects_expired_token_in_spanish():
assert _is_token_issue_message("Token inválido o expirado") is True
def test_is_token_issue_message_detects_expired_token_in_english():
assert _is_token_issue_message("Invalid or expired token") is True
def test_is_token_issue_message_detects_detail_reason_combo():
assert _is_token_issue_message(
"Access denied",
"jwt signature validation failed",
"token malformed",
) is True
def test_is_token_issue_message_does_not_flag_real_license_error():
assert _is_token_issue_message("Sin licencia asignada para este tenant") is False