import uuid from datetime import datetime, timedelta, timezone from typing import Dict, Any, Optional from sqlalchemy.orm import Session from fastapi import HTTPException, status from app.models.auth.sso_grant_model import SSOGrant from app.models.auth.module_model import Module from app.models.auth.module_environment_model import ModuleEnvironment from app.models.auth.tenant_module_model import TenantModule from app.models.auth.user_model import User from app.config.security import security from app.services.auth.trust_service import TrustService import json import time class SSOService: @staticmethod def generate_grant( db: Session, user_id: uuid.UUID, module_id: str, tenant_id: Optional[uuid.UUID] = None ) -> Dict[str, str]: """ Generates a one-time SSO grant code for the specified module. Resolves the correct environment URL based on tenant/user config. """ module = db.query(Module).filter(Module.module_id == module_id).first() if not module: raise HTTPException(status_code=404, detail="Module not found") if module.status != "active": raise HTTPException(status_code=403, detail="Module is disabled") environment_slug = "prod" if tenant_id: tm = db.query(TenantModule).filter( TenantModule.tenant_id == tenant_id, TenantModule.module_id == module.id ).first() if not tm or not tm.is_active: raise HTTPException(status_code=403, detail="Tenant does not have access to this module") if tm.assigned_environment_slug: environment_slug = tm.assigned_environment_slug env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == module.id, ModuleEnvironment.slug == environment_slug ).first() if not env: env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == module.id, ModuleEnvironment.is_default == True ).first() if not env: raise HTTPException(status_code=404, detail="No active environment found for module. Please configure an environment in the Admin Console.") grant_code = str(uuid.uuid4().hex) expires_at = datetime.now(timezone.utc) + timedelta(seconds=60) redirect_url = f"{env.frontend_base_url}{env.sso_entry_path}?grant={grant_code}" grant = SSOGrant( grant_code=grant_code, user_id=user_id, module_id=module.id, tenant_id=tenant_id, environment_slug=env.slug, expires_at=expires_at ) db.add(grant) db.commit() db.refresh(grant) return { "grant_code": grant.grant_code, "redirect_url": redirect_url } @staticmethod def generate_signed_payload( db: Session, user_id: uuid.UUID, module_id: str, tenant_id: Optional[uuid.UUID] = None ) -> Dict[str, Any]: """ Generates a signed payload for the client to POST directly to the module backend. """ module = db.query(Module).filter(Module.module_id == module_id).first() if not module or module.status != "active": raise HTTPException(status_code=404, detail="Module not found or disabled") environment_slug = "prod" if tenant_id: tm = db.query(TenantModule).filter(TenantModule.tenant_id == tenant_id, TenantModule.module_id == module.id).first() if tm and tm.is_active and tm.assigned_environment_slug: environment_slug = tm.assigned_environment_slug env = db.query(ModuleEnvironment).filter(ModuleEnvironment.module_id == module.id, ModuleEnvironment.slug == environment_slug).first() if not env: env = db.query(ModuleEnvironment).filter(ModuleEnvironment.module_id == module.id, ModuleEnvironment.is_default == True).first() if not env: raise HTTPException(status_code=404, detail="No active environment found for module. Please configure an environment in the Admin Console.") user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") permissions = [] if user.role: for ra in user.role.role_accesses: if ra.access: pass if user.role.role_module_accesses: for rma in user.role.role_module_accesses: if rma.module_access and rma.module_access.module_id == module.id: permissions.append(rma.module_access.access_code) timestamp = int(time.time() * 1000) payload_data = { "user_id": str(user.id), "email": user.email, "tenant_id": str(tenant_id) if tenant_id else None, "permissions": permissions, "timestamp": timestamp, "first_name": user.first_name, "last_name": user.last_name, "role": user.role.role_name if user.role else None } tenant_id_str = str(tenant_id) if tenant_id else "" canonical_string = f"user_id={user.id}&email={user.email}&tenant_id={tenant_id_str}×tamp={timestamp}" try: signature = TrustService.sign_payload(env, canonical_string) except ValueError: raise HTTPException(status_code=500, detail="Module trust configuration error (missing HMAC secret)") base_url = env.backend_base_url.rstrip('/') path = env.sso_entry_path if env.sso_entry_path else "/sso/login" if not path.startswith('/'): path = '/' + path target_url = f"{base_url}{path}" return { "target_url": target_url, "payload": payload_data, "headers": { "X-App-Id": "saas", "X-App-Id": module.module_id, "X-Signature": signature }, "redirect_url": env.frontend_base_url } @staticmethod def exchange_grant( db: Session, grant_code: str, module_id: str, environment_slug: str ) -> Dict[str, Any]: """ Validates grant and returns a short-lived module-scoped token. This is called by the Module Backend. """ grant = db.query(SSOGrant).filter(SSOGrant.grant_code == grant_code).first() if not grant: raise HTTPException(status_code=401, detail="Invalid grant code") if grant.is_used: raise HTTPException(status_code=401, detail="Grant code already used") if grant.expires_at < datetime.now(timezone.utc): raise HTTPException(status_code=401, detail="Grant code expired") module = db.query(Module).filter(Module.module_id == module_id).first() if not module or module.id != grant.module_id: raise HTTPException(status_code=401, detail="Grant invalid for this module") if grant.environment_slug != environment_slug: raise HTTPException(status_code=401, detail="Grant invalid for this environment") user = db.query(User).filter(User.id == grant.user_id).first() if not user: raise HTTPException(status_code=401, detail="User not found") if grant.tenant_id: if user.tenant_id != grant.tenant_id: raise HTTPException( status_code=401, detail="Tenant mismatch for SSO grant" ) grant.is_used = True grant.used_at = datetime.now(timezone.utc) db.commit() permissions = [] if user.role: if user.role.role_module_accesses: for rma in user.role.role_module_accesses: if rma.module_access and rma.module_access.module_id == module.id: permissions.append(rma.module_access.access_code) token_payload = { "sub": str(user.id), "email": user.email, "tenant_id": str(grant.tenant_id) if grant.tenant_id else None, "module_id": module_id, "environment": environment_slug, "permissions": permissions, "roles": [user.role.role_name] if user.role else [] } token = security.generate_module_token(token_payload, module_id) return { "access_token": token, "token_type": "bearer", "expires_in": 900, "user": { "id": str(user.id), "email": user.email, "first_name": user.first_name, "last_name": user.last_name } }