feat: Add frontend and backend initialization scripts, implement Keycloak and PostgreSQL setup
- Implemented SvelteKit frontend with authentication callback handling. - Created demo routes and paraglide localization functionality. - Added health check and entrypoint scripts for backend services. - Established PostgreSQL and Keycloak initialization scripts with health checks. - Introduced models for database schema using SQLAlchemy. - Configured Vite and SvelteKit for development and testing environments. - Added health check script to verify service statuses and resource usage. - Created Docker entrypoint scripts for seamless service startup.
This commit is contained in:
166
backend/core/security.py
Normal file
166
backend/core/security.py
Normal file
@@ -0,0 +1,166 @@
|
||||
"""
|
||||
Utilidades de seguridad y autenticación con Keycloak
|
||||
"""
|
||||
from fastapi import HTTPException, Security, Depends
|
||||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||||
from keycloak import KeycloakOpenID
|
||||
from jose import jwt, JWTError
|
||||
from typing import Optional, Dict, Any
|
||||
from .config import settings
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Configuración de Keycloak
|
||||
keycloak_openid = KeycloakOpenID(
|
||||
server_url=settings.KEYCLOAK_SERVER_URL,
|
||||
client_id=settings.KEYCLOAK_CLIENT_ID,
|
||||
realm_name=settings.KEYCLOAK_REALM,
|
||||
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET
|
||||
)
|
||||
|
||||
# Security scheme
|
||||
security = HTTPBearer()
|
||||
|
||||
|
||||
def verify_token(token: str) -> 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
|
||||
"""
|
||||
try:
|
||||
# Obtener clave pública de Keycloak
|
||||
KEYCLOAK_PUBLIC_KEY = (
|
||||
"-----BEGIN PUBLIC KEY-----\n"
|
||||
+ keycloak_openid.public_key()
|
||||
+ "\n-----END PUBLIC KEY-----"
|
||||
)
|
||||
|
||||
# Decodificar y verificar token
|
||||
options = {
|
||||
"verify_signature": True,
|
||||
"verify_aud": False,
|
||||
"verify_exp": True
|
||||
}
|
||||
|
||||
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)}")
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Could not validate credentials"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error during token verification: {str(e)}")
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Authentication error"
|
||||
)
|
||||
|
||||
|
||||
async def get_current_user(
|
||||
credentials: HTTPAuthorizationCredentials = Security(security)
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Dependency para obtener el usuario actual desde el token JWT
|
||||
|
||||
Uso en FastAPI:
|
||||
current_user: dict = Depends(get_current_user)
|
||||
"""
|
||||
token = credentials.credentials
|
||||
user_info = verify_token(token)
|
||||
return user_info
|
||||
|
||||
|
||||
async def get_current_active_user(
|
||||
current_user: Dict[str, Any] = Depends(get_current_user)
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Dependency para obtener usuario activo (puede incluir validaciones adicionales)
|
||||
"""
|
||||
# Aquí se pueden agregar validaciones adicionales
|
||||
# Por ejemplo, verificar si el usuario está activo en la BD
|
||||
return current_user
|
||||
|
||||
|
||||
def has_role(required_role: str):
|
||||
"""
|
||||
Decorator/Dependency para verificar roles de usuario
|
||||
|
||||
Uso:
|
||||
@router.get("/admin")
|
||||
async def admin_endpoint(user = Depends(has_role("admin"))):
|
||||
...
|
||||
"""
|
||||
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", [])
|
||||
|
||||
if required_role not in user_roles:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"User does not have required role: {required_role}"
|
||||
)
|
||||
|
||||
return current_user
|
||||
|
||||
return role_checker
|
||||
|
||||
|
||||
def get_tenant_from_token(user_info: Dict[str, Any]) -> Optional[int]:
|
||||
"""
|
||||
Extrae el tenant_id del token JWT
|
||||
|
||||
El tenant_id puede estar en diferentes lugares según configuración de Keycloak:
|
||||
- En claims personalizados
|
||||
- En el realm
|
||||
- En atributos del usuario
|
||||
"""
|
||||
# Intentar obtener de claims personalizados
|
||||
tenant_id = user_info.get("tenant_id")
|
||||
|
||||
if not tenant_id:
|
||||
# Intentar obtener de atributos
|
||||
tenant_id = user_info.get("attributes", {}).get("tenant_id")
|
||||
|
||||
if tenant_id:
|
||||
return int(tenant_id)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
class KeycloakClient:
|
||||
"""Cliente para interactuar con Keycloak Admin API"""
|
||||
|
||||
def __init__(self):
|
||||
self.openid = keycloak_openid
|
||||
|
||||
def create_user(self, email: str, password: str, tenant_id: int, **kwargs):
|
||||
"""Crea un usuario en Keycloak"""
|
||||
# Implementar lógica para crear usuario usando keycloak admin
|
||||
pass
|
||||
|
||||
def assign_role(self, user_id: str, role: str):
|
||||
"""Asigna un rol a un usuario"""
|
||||
pass
|
||||
|
||||
def create_tenant_realm(self, tenant_name: str):
|
||||
"""Crea un realm para un nuevo tenant"""
|
||||
pass
|
||||
Reference in New Issue
Block a user