Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 19 additions & 5 deletions AUDIT_MASTER.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,8 @@
- **Total de Hallazgos:** 34
- **Desglose por Severidad:** Crítico: 5 | Alto: 15 | Medio: 13 | Bajo: 1
- **Desglose por Tipo:** Defecto: 6 | Deuda Técnica: 6 | Brecha Funcional: 22
- **Total con Estado "Corregido":** 28
- **Total con Estado "Pendiente":** 6
- **Total con Estado "Corregido":** 29
- **Total con Estado "Pendiente":** 5
- **Desglose por Categoría (A-J):**
- A. Seguridad y Aislamiento de Ejecución: 8 hallazgos
- B. Gobernanza Empresarial y Multi-tenencia: 6 hallazgos
Expand Down Expand Up @@ -38,7 +38,7 @@
| AUD-005 | Medio | Brecha Funcional | A. Seguridad y Aislamiento | P1 Esperado | cognito-backend | Ausencia de metadatos de comportamiento (read-only/destructive/concurrency) en esquema de herramientas | Corregido |
| AUD-006 | Medio | Deuda Técnica | A. Seguridad y Aislamiento | P1 Esperado | cognito-backend / worker | Rango abierto de dependencias Python sin lockfile con hashes integrados | Corregido |
| AUD-007 | Crítico | Brecha Funcional | B. Gobernanza y Multi-tenencia | P0 Bloqueante | cognito-backend | Ausencia de modelo de datos multi-tenant (Org / Tenant / User) | Corregido |
| AUD-008 | Crítico | Brecha Funcional | B. Gobernanza y Multi-tenencia | P0 Bloqueante | cognito-backend | Inexistencia de autenticación SSO/SAML/OIDC para operadores humanos | Pendiente (Plan de diseño disponible) |
| AUD-008 | Crítico | Brecha Funcional | B. Gobernanza y Multi-tenencia | P0 Bloqueante | cognito-backend | Inexistencia de autenticación SSO/SAML/OIDC para operadores humanos | Corregido |
| AUD-009 | Crítico | Brecha Funcional | B. Gobernanza y Multi-tenencia | P0 Bloqueante | cognito-backend | Inexistencia de audit log estructurado exportable hacia sistemas SIEM | Corregido |
| AUD-010 | Alto | Brecha Funcional | B. Gobernanza y Multi-tenencia | P1 Esperado | cognito-backend | Control de presupuesto de tokens restringido al ámbito de sesión individual | Corregido |
| AUD-011 | Medio | Brecha Funcional | B. Gobernanza y Multi-tenencia | P1 Esperado | cognito-backend | Inexistencia de políticas automatizadas de retención y borrado de datos de usuario/sesión | Corregido |
Expand Down Expand Up @@ -319,8 +319,22 @@
- **Descripción del problema:** No existe ninguna integración con esquemas de autenticación federada SSO, SAML 2.0 ni OpenID Connect (OIDC) para validar la identidad de los usuarios humanos que interactúan con el backend o la CLI.
- **Evidencia de Ubicación en Código:** Revisión completa del directorio `very-simplified-stack/cognito-backend/app/api/routes/` (ausencia de módulos de OAuth/OIDC/SAML).
- **Comparación con el estado del arte:** El soporte de SSO/OIDC/SAML es un requisito no negociable en las evaluaciones de seguridad corporativa para permitir el control de acceso centralizado.
- **Estado:** Pendiente (Plan de diseño disponible)
- **Nota de Plan de Diseño:** Se definió la integración SSO/OIDC/SAML con verificación de firmas asimétricas (`PyJWT` / `python-saml`) y reglas de mapeo de claims a usuarios/roles sobre los modelos unificados `Organization`, `Project` y `User` (`app/models/domain.py` / `app/models/db.py`) en `ARCHITECTURE_RFC_GOBERNANZA.md`.
- **Estado:** Corregido
- **Resolución y Evidencia Técnica:**
- Se creó el módulo `very-simplified-stack/cognito-backend/app/core/sso/` definiendo la interfaz abstracta `SSOProvider` (`base.py`) y la implementación real `OIDCProvider` (`oidc.py`).
- Se implementó la verificación real de firmas asimétricas (RS256 / ES256) de los ID Tokens OIDC contra el JWKS (JSON Web Key Set) del proveedor utilizando `PyJWT` y `cryptography`. Tokens con firma manipulada o expirados son explícitamente rechazados (`InvalidTokenSignatureError`).
- Se implementó la clase `SSOService` (`service.py`) que resuelve la vinculación de `Organization` mediante reglas configurables de mapeo por dominio de email (`COGNITO_SSO_DOMAIN_MAP`), realiza el auto-aprovisionamiento de nuevos usuarios (`User`) o actualización de existentes en su primer login, y vincula la sesión de Cognito (`SessionManager` con `auth_type="authenticated_sso"`).
- Se integró el registro de eventos de auditoría SIEM (`auth.sso_login` y `auth.sso_logout`) en el Audit Log estructurado (AUD-009).
- Se expusieron las rutas HTTP REST en `app/api/routes/auth.py` (`GET /api/auth/sso/login`, `GET/POST /api/auth/sso/callback`, `POST /api/auth/sso/logout`) y se montaron en `app/main.py`.
- Se implementó la clase `SAMLProvider` (`saml.py`) como stub documentado siguiendo el mismo patrón arquitectónico de `VaultSecretsProvider` (AUD-003), permitiendo extender el soporte completo a SAML 2.0 en el futuro sin romper contratos de API.
- **Test de Regresión:**
- `very-simplified-stack/cognito-backend/tests/test_sso_oidc.py`:
- `test_oidc_authorization_url_generation`: Valida la generación de URLs de autorización OIDC con `state` y `client_id`.
- `test_oidc_valid_id_token_verification`: Verifica la validación exitosa de firmas asimétricas RS256 usando pares de claves RSA y JWKS.
- `test_oidc_tampered_signature_token_rejected`: Prueba explícitamente que tokens con firma o payload manipulados son rechazados lanzando `InvalidTokenSignatureError`.
- `test_oidc_end_to_end_callback_flow_and_auto_provisioning`: Prueba E2E del callback SSO con mock IdP, verificando el auto-aprovisionamiento de usuario, vinculación multi-tenant a la organización por dominio, emisión de sesión Cognito y registro en el Audit Log.
- `test_sso_logout_endpoint_and_audit`: Confirma el cierre de sesión y la presencia del evento `auth.sso_logout` en el Audit Log.
- `test_saml_provider_stub_instantiation_and_callback`: Verifica el comportamiento e instanciación del stub documentado de SAML 2.0.

#### AUD-009
- **ID:** AUD-009
Expand Down
146 changes: 146 additions & 0 deletions very-simplified-stack/cognito-backend/app/api/routes/auth.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
import os
import uuid
import logging
from typing import Optional, Dict, Any
from fastapi import APIRouter, HTTPException, Query, Request, status
from pydantic import BaseModel

from app.core.sso import sso_manager, InvalidTokenSignatureError
from app.core.sso.service import sso_service
from app.core.audit import audit_logger, AuditLogRecord, ActorInfo

logger = logging.getLogger("cognito.backend.api.auth")

router = APIRouter(prefix="/auth/sso", tags=["SSO Authentication"])


class SSOLoginResponse(BaseModel):
authorization_url: str
state: str
provider_type: str


class SSOCallbackResponse(BaseModel):
session_id: str
user_id: str
org_id: str
email: str
full_name: Optional[str] = None
roles: list[str]


class SSOLogoutRequest(BaseModel):
user_id: str
org_id: str
session_id: Optional[str] = None


@router.get("/login", response_model=SSOLoginResponse)
async def sso_login(
redirect_uri: str = Query(..., description="URI de redirección del cliente tras autenticarse con IdP"),
org_id: Optional[str] = Query(None, description="ID de organización opcional"),
provider_type: Optional[str] = Query(None, description="Tipo de proveedor ('oidc' o 'saml')"),
):
state = f"st-{uuid.uuid4().hex[:12]}"
provider = sso_manager.get_provider_for_org(org_id=org_id, provider_type=provider_type)
auth_url = provider.get_authorization_url(redirect_uri=redirect_uri, state=state)
eff_p_type = provider_type or os.getenv("COGNITO_SSO_PROVIDER", "oidc")

return SSOLoginResponse(
authorization_url=auth_url,
state=state,
provider_type=eff_p_type,
)


@router.get("/callback", response_model=SSOCallbackResponse)
@router.post("/callback", response_model=SSOCallbackResponse)
async def sso_callback(
request: Request,
code: Optional[str] = Query(None),
redirect_uri: Optional[str] = Query(None),
org_id: Optional[str] = Query(None),
session_id: Optional[str] = Query(None),
provider_type: Optional[str] = Query(None),
):
# Procesa tanto query params como JSON body si se envió vía POST
if request.method == "POST":
try:
body = await request.json()
code = code or body.get("code")
redirect_uri = redirect_uri or body.get("redirect_uri")
org_id = org_id or body.get("org_id")
session_id = session_id or body.get("session_id")
provider_type = provider_type or body.get("provider_type")
except Exception:
pass

if not code or not redirect_uri:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Se requieren 'code' y 'redirect_uri' para procesar el callback SSO",
)

provider = sso_manager.get_provider_for_org(org_id=org_id, provider_type=provider_type)

try:
claims = await provider.process_callback(code=code, redirect_uri=redirect_uri)
except InvalidTokenSignatureError as e:
logger.warning(f"Rechazado intento de login SSO con firma inválida: {e}")
audit_logger.record(
AuditLogRecord(
audit_id=f"aud-sso-err-{uuid.uuid4().hex[:12]}",
org_id=org_id or "org-default-local",
action="auth.sso_login",
resource="sso:callback",
status="INVALID_SIGNATURE",
actor=ActorInfo(type="user", id="anonymous_unverified"),
details={"error": str(e)},
)
)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Firma de token SSO inválida o manipulada: {e}",
)
except Exception as e:
logger.error(f"Error procesando callback SSO: {e}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Error al autenticar con el proveedor SSO: {e}",
)

# Auto-aprovisionamiento y mapeo por dominio de email
target_org_id = sso_service.resolve_org_id_for_email(claims.email, explicit_org_id=org_id)
user = sso_service.provision_or_update_user(claims, org_id=target_org_id)

# Emisión/vinculación de sesión Cognito
cognito_session_id = sso_service.bind_or_create_cognito_session(
user=user,
existing_session_id=session_id,
)

# Auditoría SIEM
sso_service.record_sso_login_audit(user=user, session_id=cognito_session_id, status="SUCCESS")

return SSOCallbackResponse(
session_id=cognito_session_id,
user_id=user.user_id,
org_id=user.org_id,
email=user.email,
full_name=user.full_name,
roles=user.roles or ["developer"],
)


@router.post("/logout")
async def sso_logout(payload: SSOLogoutRequest):
provider = sso_manager.get_provider_for_org(org_id=payload.org_id)
await provider.logout(user_id=payload.user_id)

sso_service.record_sso_logout_audit(
user_id=payload.user_id,
org_id=payload.org_id,
session_id=payload.session_id,
)

return {"message": "Sesión SSO cerrada correctamente", "user_id": payload.user_id}
14 changes: 14 additions & 0 deletions very-simplified-stack/cognito-backend/app/core/sso/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
from app.core.sso.base import SSOProvider, SSOUserClaims
from app.core.sso.oidc import OIDCProvider, InvalidTokenSignatureError
from app.core.sso.saml import SAMLProvider
from app.core.sso.manager import SSOManager, sso_manager

__all__ = [
"SSOProvider",
"SSOUserClaims",
"OIDCProvider",
"InvalidTokenSignatureError",
"SAMLProvider",
"SSOManager",
"sso_manager",
]
46 changes: 46 additions & 0 deletions very-simplified-stack/cognito-backend/app/core/sso/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field


class SSOUserClaims(BaseModel):
subject: str
email: str
full_name: Optional[str] = None
roles: list[str] = Field(default_factory=lambda: ["developer"])
raw_claims: Dict[str, Any] = Field(default_factory=dict)


class SSOProvider(ABC):
"""
Abstract interface for SSO Authentication Providers (OIDC, SAML).
"""

@abstractmethod
def get_authorization_url(self, redirect_uri: str, state: str) -> str:
"""
Generates the authorization URL to redirect the user to the IdP.
"""
pass

@abstractmethod
async def process_callback(self, code: str, redirect_uri: str) -> SSOUserClaims:
"""
Handles authorization code exchange callback and returns validated user claims.
"""
pass

@abstractmethod
def validate_id_token(self, id_token: str, jwks: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Validates the signature and claims of an ID token against JWKS.
Lanzará InvalidTokenSignatureError / ValueError si la firma o los claims son inválidos.
"""
pass

@abstractmethod
async def logout(self, user_id: str) -> bool:
"""
Executes logout logic for the SSO session.
"""
pass
63 changes: 63 additions & 0 deletions very-simplified-stack/cognito-backend/app/core/sso/manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
import os
import logging
from typing import Dict, Any, Optional

from app.core.sso.base import SSOProvider
from app.core.sso.oidc import OIDCProvider
from app.core.sso.saml import SAMLProvider

logger = logging.getLogger("cognito.backend.sso.manager")


class SSOManager:
"""
Central SSO Manager to resolve and instantiate SSO Providers (OIDC, SAML)
dynamically based on Organization configuration or global environment defaults.
"""

def __init__(self):
self._providers: Dict[str, SSOProvider] = {}

def get_provider_for_org(
self,
org_id: Optional[str] = None,
provider_type: Optional[str] = None,
org_sso_config: Optional[Dict[str, Any]] = None,
) -> SSOProvider:
"""
Resolves the appropriate SSOProvider for a given Organization or system default.
Layered resolution:
1. Explicit provider_type or org_sso_config parameter
2. Environment variable COGNITO_SSO_PROVIDER ("oidc" default, or "saml")
"""
config = org_sso_config or {}
p_type = (provider_type or config.get("type") or os.getenv("COGNITO_SSO_PROVIDER", "oidc")).lower()

if p_type == "saml":
return SAMLProvider(
entity_id=config.get("entity_id"),
sso_url=config.get("sso_url"),
x509_cert=config.get("x509_cert"),
)

# OIDC default
client_id = config.get("client_id") or os.getenv("COGNITO_OIDC_CLIENT_ID", "cognito-client-id")
client_secret = config.get("client_secret") or os.getenv("COGNITO_OIDC_CLIENT_SECRET", "cognito-client-secret")
issuer = config.get("issuer") or os.getenv("COGNITO_OIDC_ISSUER", "https://auth.example.com")
auth_endpoint = config.get("authorization_endpoint") or os.getenv("COGNITO_OIDC_AUTH_URL", f"{issuer}/oauth2/v1/authorize")
token_endpoint = config.get("token_endpoint") or os.getenv("COGNITO_OIDC_TOKEN_URL", f"{issuer}/oauth2/v1/token")
jwks_uri = config.get("jwks_uri") or os.getenv("COGNITO_OIDC_JWKS_URI", f"{issuer}/oauth2/v1/keys")
jwks_data = config.get("jwks_data")

return OIDCProvider(
client_id=client_id,
client_secret=client_secret,
issuer=issuer,
authorization_endpoint=auth_endpoint,
token_endpoint=token_endpoint,
jwks_uri=jwks_uri,
jwks_data=jwks_data,
)


sso_manager = SSOManager()
Loading
Loading