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

Refactor code structure for improved readability and maintainability

Browse Source
This commit is contained in:
AlexeerCT
2026-03-07 23:09:13 -06:00
parent 2ac07b55b6
commit 5478a88f2b
14 changed files with 799 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,7 +12,9 @@ class LoginRequestDTO(BaseModel):
username: str = Field(..., description="Usuario o email")
password: str = Field(..., min_length=6, description="Contraseña")
tenant_slug: str = Field(..., description="Slug del tenant")
# Opcional en el primer paso: si no se provee, el backend verifica credenciales
# y devuelve la lista de tenants disponibles en lugar de tokens.
tenant_slug: Optional[str] = Field(None, description="Slug del tenant")
class Config:
json_schema_extra = {
@@ -57,6 +59,7 @@ class UserInfoResponseDTO(BaseModel):
name: Optional[str] = None
preferred_username: Optional[str] = None
tenant_id: Optional[int] = None
tenant_slug: Optional[str] = None
roles: list[str] = []
class Config:
@@ -146,10 +149,49 @@ class SetCookieRequestDTO(BaseModel):
access_token: str = Field(..., description="Access token JWT")
refresh_token: str = Field(..., description="Refresh token JWT")
class SwitchTenantRequestDTO(BaseModel):
"""DTO para cambiar de tenant estando autenticado"""
tenant_slug: str = Field(..., description="Slug del tenant destino")
refresh_token: str = Field(..., description="Refresh token actual para emitir nuevos tokens")
class DiscoverTenantsRequestDTO(BaseModel):
"""DTO para descubrir los tenants de un usuario sin necesidad de indicarlo manualmente"""
username: str = Field(..., description="Nombre de usuario o email")
class Config:
json_schema_extra = {
"example": {
"access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"refresh_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"username": "jperez",
}
}
class TenantInfoDTO(BaseModel):
"""Información básica de un tenant para mostrar en el selector de login"""
id: int
name: str
slug: str
class Config:
from_attributes = True
class DiscoverTenantsResponseDTO(BaseModel):
"""Respuesta con los tenants disponibles para un usuario"""
tenants: list[TenantInfoDTO]
class LoginChoiceResponseDTO(BaseModel):
"""
Respuesta del login cuando el usuario pertenece a varios tenants.
Las credenciales ya fueron verificadas; el cliente debe re-enviar con tenant_slug.
"""
status: str = "choose_tenant"
tenants: list[TenantInfoDTO]

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

@@ -10,12 +10,14 @@ from sqlalchemy.orm import Session
from .dto import (
ExchangeCodeRequestDTO,
LoginChoiceResponseDTO,
LoginRequestDTO,
LogoutRequestDTO,
RefreshTokenRequestDTO,
RegisterRequestDTO,
RegisterResponseDTO,
SetCookieRequestDTO,
SwitchTenantRequestDTO,
TokenResponseDTO,
UserInfoResponseDTO,
)
@@ -49,7 +51,7 @@ async def register(
return service.register(register_data)
@router.post("/login", response_model=TokenResponseDTO)
@router.post("/login", response_model=None)
async def login(
login_data: LoginRequestDTO,
request: Request, # Inject Request
@@ -73,6 +75,42 @@ async def login(
)
@router.post("/switch-tenant", response_model=TokenResponseDTO)
async def switch_tenant(
data: SwitchTenantRequestDTO,
db: Session = Depends(get_core_db),
credentials: HTTPAuthorizationCredentials = Depends(security),
):
"""
Cambia el tenant activo de un usuario ya autenticado y retorna nuevos tokens JWT.
Requiere:
- Authorization: Bearer <access_token> (para identificar al usuario)
- Body: { tenant_slug, refresh_token }
"""
service = AuthService(db)
# Obtener info del usuario desde el access token actual
user_info = 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,
# pero lo más directo es dejar que Keycloak lo resuelva usando la config global.
# Todos los tenants comparten el mismo realm en esta arquitectura.
from api.v1.modules.core.tenants.models import Tenant
from core.database import get_core_db as _gcdb
# Obtener el realm del tenant destino (o default)
tenant = db.query(Tenant).filter(Tenant.slug == data.tenant_slug, Tenant.is_active).first()
if not tenant:
raise HTTPException(status_code=403, detail="Access denied")
return service.switch_tenant(
keycloak_user_id=keycloak_user_id,
keycloak_realm=tenant.keycloak_realm,
tenant_slug=data.tenant_slug,
refresh_token=data.refresh_token,
)
@router.post("/refresh", response_model=TokenResponseDTO)
async def refresh_token(
refresh_data: RefreshTokenRequestDTO, db: Session = Depends(get_core_db)

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

@@ -43,32 +43,45 @@ class AuthService:
login_data: LoginRequestDTO,
ip_address: str = None,
user_agent: str = None
) -> TokenResponseDTO:
):
"""
Autentica usuario y obtiene tokens
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
login_data: Credenciales de login (tenant_slug es opcional)
ip_address: Dirección IP del cliente
user_agent: User Agent del cliente
Returns:
TokenResponseDTO con access_token y refresh_token
TokenResponseDTO si tenant_slug fue provisto,
LoginChoiceResponseDTO si no se proveyó tenant_slug.
Raises:
HTTPException: Si las credenciales son inválidas
"""
# 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:
raise HTTPException(status_code=404, detail="Tenant not found")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant is not active")
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(
@@ -110,8 +123,8 @@ class AuthService:
f"User {user_id} tried to access tenant {tenant.id} without permission"
)
raise HTTPException(
status_code=403,
detail="You don't have access to this tenant",
status_code=401,
detail="Invalid credentials",
)
# Obtener los datos actuales del usuario
@@ -228,17 +241,25 @@ class AuthService:
if "realm_access" in user_info:
roles = user_info["realm_access"].get("roles", [])
# Extraer tenant_id si está presente
# 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
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,
)
@@ -455,3 +476,252 @@ class AuthService:
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