from sqlalchemy.orm import Session from fastapi import Request, HTTPException from typing import Optional, Dict, Any import uuid import logging from app.services.auth.sso_service import SSOService from app.services.auth.trust_service import TrustService from app.schemas.auth.sso_schema import SSOInitiateRequest, SSOExchangeRequest from app.models.auth.module_model import Module from app.models.auth.module_environment_model import ModuleEnvironment from app.models.auth.user_model import User logger = logging.getLogger(__name__) class SSOController: @staticmethod def initiate_sso(db: Session, request: SSOInitiateRequest, current_user: User): return SSOService.generate_signed_payload( db=db, user_id=current_user.id, module_id=request.module_id, tenant_id=current_user.tenant_id ) @staticmethod def exchange_grant( db: Session, payload: SSOExchangeRequest, x_module_signature: Optional[str] = None, x_module_key: Optional[str] = None ): module = db.query(Module).filter(Module.module_id == payload.module_id).first() if not module: raise HTTPException(status_code=404, detail="Module not found") env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == module.id, ModuleEnvironment.slug == payload.environment_slug ).first() if not env: raise HTTPException(status_code=404, detail="Environment not found") headers = {} if x_module_signature: headers["X-Module-Signature"] = x_module_signature if x_module_key: headers["X-Module-Key"] = x_module_key actual_body = payload.model_dump_json() try: TrustService.validate_module_trust( environment=env, request_headers=headers, request_body=actual_body ) except HTTPException: logger.warning( "HMAC verify with body failed for %s, trying empty fallback (DEPRECATED)", payload.module_id, ) TrustService.validate_module_trust( environment=env, request_headers=headers, request_body="" ) return SSOService.exchange_grant( db=db, grant_code=payload.grant_code, module_id=payload.module_id, environment_slug=payload.environment_slug )