fix(rbac): scope tenant roles to subscribed modules

.env.development

@@ -1,62 +0,0 @@
-# Project Configuration
-PROJECT_NAME=SaaS Architecture
-VERSION=1.0.0
-PORT=10001
-APP_ENV=development
-SECRET_KEY="Usu9Qmg4ppRexR6Xp657MMMHsoOaiV8cPqlY_THWNaPhGT6DN9Xd8UO4zG3kWjwIqW9hPa5bYwQUoDhyRlzv_w"
-ALLOWED_HOSTS=*
-HOST=127.0.0.1
-FRONTEND_URL=https://saas-dev.maskantech.in
-CORS_ALLOWED_ORIGINS=https://saas-dev.maskantech.in
- 
-# Security
-ENCRYPTION_KEY="1cd1dc2d42afc5606e224df1108162db2d6ca372a45a9b8d278162f6006236d5"
- 
-# Super Admin Configuration
-SUPER_ADMIN_EMAIL=admin@maskantech.in
-SUPER_ADMIN_PASSWORD=SuperAdmin@123
-SUPER_ADMIN_FIRST_NAME=Super
-SUPER_ADMIN_LAST_NAME=Admin
- 
-#Database Configuration
-DB_SSL=False
-DATABASE_URL=postgresql://saas_user:K9uR3mZpQ7~W4F2YH8A_tLxD@106.51.105.22:5432/saas_development
-
-# Redis Configuration
-REDIS_HOST=106.51.105.22
-REDIS_PORT=6382
-REDIS_PASSWORD=8haSTisAqop8ChAs
- 
-# Email Configuration
-SMTP_HOST=smtp.hostinger.com
-SMTP_PORT=465
-SMTP_SECURE=true
-SMTP_USER=info@maskantech.in
-SMTP_PASSWORD=Infomaskan@123
-EMAIL_FROM=info@maskantech.in
- 
-# JWT Configuration
-ACCESS_TOKEN_SECRET="L_ByN0_FIuwsQnDo4sdrOEdJvqlPjKfhVJmqhf76D13v3IWu3mbvzb8hQRnPxHMlr9Y8A9IcOHZZWSs7Kfofpg"
-ACCESS_TOKEN_EXPIRES=900
-REFRESH_TOKEN_SECRET="6Z0yOfkhPjfLH77WTnBh3Iv0JU_gWIfIqpGuGU41GFRV4fnLZMfKN3gAsPTfsqFKv1rRc6szUJRngW8Py0UYUQ"
-REFRESH_TOKEN_EXPIRES=864000
-JWT_ALGORITHM=HS256
-ADMIN_JWT="upRCbNd-3Ex3sG2aEHxcrCx7LZu91BkiNGPIs-vxNXp7YBHyU-0jMpGUYA5dJHLcyOIMLBk4HCw1cpI4WOlIzA"
- 
-# AWS S3 Configuration
-AWS_ACCESS_KEY_ID=""
-AWS_SECRET_ACCESS_KEY=""
-AWS_REGION=""
-S3_BUCKET_NAME=""
- 
-# PayPal Integration
-PAYPAL_CLIENT_ID=""
-PAYPAL_CLIENT_SECRET=""
-PAYPAL_MODE=sandbox
- 
-# Logging
-LOG_LEVEL=info
- 
-# Development flags
-DEBUG_MODE=true
-ENABLE_DEBUG_LOGGING=true

.env.example

@@ -0,0 +1,62 @@
+# SaaS Core Backend Environment Template
+# Duplicate this file to .env.local and populate with your credentials.
+
+PROJECT_NAME=
+VERSION=
+PORT=
+APP_ENV=
+SECRET_KEY=
+ALLOWED_HOSTS=
+HOST=
+FRONTEND_URL=
+CORS_ALLOWED_ORIGINS=
+ENCRYPTION_KEY=
+LOG_LEVEL=
+DEBUG_MODE=
+ENABLE_DEBUG_LOGGING=
+
+# Database Configuration
+DATABASE_URL=
+DB_SSL=
+
+# Super Administrator
+SUPER_ADMIN_EMAIL=
+SUPER_ADMIN_PASSWORD=
+SUPER_ADMIN_FIRST_NAME=
+SUPER_ADMIN_LAST_NAME=
+
+# Redis & Caching
+REDIS_HOST=
+REDIS_PORT=
+REDIS_PASSWORD=
+
+# SaaS Trust Fabric & SSO
+SAAS_TRUST_SECRET=
+SSO_SECRET=
+
+# SMTP / Email
+SMTP_HOST=
+SMTP_PORT=
+SMTP_SECURE=
+SMTP_USER=
+SMTP_PASSWORD=
+EMAIL_FROM=
+
+# JWT & Authentication
+ACCESS_TOKEN_SECRET=
+ACCESS_TOKEN_EXPIRES=
+REFRESH_TOKEN_SECRET=
+REFRESH_TOKEN_EXPIRES=
+JWT_ALGORITHM=
+ADMIN_JWT=
+
+# Object Storage / AWS S3
+AWS_ACCESS_KEY_ID=
+AWS_SECRET_ACCESS_KEY=
+AWS_REGION=
+S3_BUCKET_NAME=
+
+# Payment Provider
+PAYPAL_CLIENT_ID=
+PAYPAL_CLIENT_SECRET=
+PAYPAL_MODE=

.env.local

@@ -1,62 +0,0 @@
-# Project Configuration
-PROJECT_NAME=SaaS Architecture
-VERSION=1.0.0
-PORT=8000
-APP_ENV=local
-SECRET_KEY="Usu9Qmg4ppRexR6Xp657MMMHsoOaiV8cPqlY_THWNaPhGT6DN9Xd8UO4zG3kWjwIqW9hPa5bYwQUoDhyRlzv_w"
-ALLOWED_HOSTS=*
-HOST=127.0.0.1
-FRONTEND_URL=http://localhost:3000
-CORS_ALLOWED_ORIGINS=http://localhost:5173,http://localhost:5174
- 
-# Security
-ENCRYPTION_KEY="1cd1dc2d42afc5606e224df1108162db2d6ca372a45a9b8d278162f6006236d5"
- 
-# Super Admin Configuration
-SUPER_ADMIN_EMAIL=admin@maskantech.in
-SUPER_ADMIN_PASSWORD=SuperAdmin@123
-SUPER_ADMIN_FIRST_NAME=Super
-SUPER_ADMIN_LAST_NAME=Admin
- 
-#Database Configuration
-DB_SSL=False
-DATABASE_URL=postgresql://saas_user:K9uR3mZpQ7~W4F2YH8A_tLxD@106.51.105.22:5432/saas_local
-
-# Redis Configuration
-REDIS_HOST=106.51.105.22
-REDIS_PORT=6381
-REDIS_PASSWORD=8haSTisAqop8ChAs
- 
-# Email Configuration
-SMTP_HOST=smtp.hostinger.com
-SMTP_PORT=465
-SMTP_SECURE=true
-SMTP_USER=info@maskantech.in
-SMTP_PASSWORD=Infomaskan@123
-EMAIL_FROM=info@maskantech.in
- 
-# JWT Configuration
-ACCESS_TOKEN_SECRET="L_ByN0_FIuwsQnDo4sdrOEdJvqlPjKfhVJmqhf76D13v3IWu3mbvzb8hQRnPxHMlr9Y8A9IcOHZZWSs7Kfofpg"
-ACCESS_TOKEN_EXPIRES=900
-REFRESH_TOKEN_SECRET="6Z0yOfkhPjfLH77WTnBh3Iv0JU_gWIfIqpGuGU41GFRV4fnLZMfKN3gAsPTfsqFKv1rRc6szUJRngW8Py0UYUQ"
-REFRESH_TOKEN_EXPIRES=864000
-JWT_ALGORITHM=HS256
-ADMIN_JWT="upRCbNd-3Ex3sG2aEHxcrCx7LZu91BkiNGPIs-vxNXp7YBHyU-0jMpGUYA5dJHLcyOIMLBk4HCw1cpI4WOlIzA"
- 
-# AWS S3 Configuration
-AWS_ACCESS_KEY_ID=""
-AWS_SECRET_ACCESS_KEY=""
-AWS_REGION=""
-S3_BUCKET_NAME=""
- 
-# PayPal Integration
-PAYPAL_CLIENT_ID=""
-PAYPAL_CLIENT_SECRET=""
-PAYPAL_MODE=sandbox
- 
-# Logging
-LOG_LEVEL=info
- 
-# Development flags
-DEBUG_MODE=true
-ENABLE_DEBUG_LOGGING=true

.env.production

@@ -1,67 +0,0 @@
-# Project Configuration
-PROJECT_NAME=SaaS Architecture
-VERSION=1.0.0
-PORT=8000
-APP_ENV=production
-SECRET_KEY="Usu9Qmg4ppRexR6Xp657MMMHsoOaiV8cPqlY_THWNaPhGT6DN9Xd8UO4zG3kWjwIqW9hPa5bYwQUoDhyRlzv_w"
-ALLOWED_HOSTS=*
-HOST=127.0.0.1
-FRONTEND_URL=http://localhost:3000
-CORS_ALLOWED_ORIGINS=http://localhost:3000,http://localhost:5173
- 
-# Security
-ENCRYPTION_KEY="1cd1dc2d42afc5606e224df1108162db2d6ca372a45a9b8d278162f6006236d5"
- 
-# Super Admin Configuration
-SUPER_ADMIN_EMAIL=admin@maskantech.in
-SUPER_ADMIN_PASSWORD=SuperAdmin@123
-SUPER_ADMIN_FIRST_NAME=Super
-SUPER_ADMIN_LAST_NAME=Admin
- 
-#Database Configuration
-DB_SSL=False
-DATABASE_URL=postgresql://postgres:azeem3105@localhost:5434/
- 
-# Redis Configuration
-# REDIS_URL=""
-# port=""
-# host=""    
-# password=""
- 
-# Email Configuration
-SMTP_HOST=smtp.hostinger.com
-SMTP_PORT=465
-SMTP_SECURE=true
-SMTP_USER=info@maskantech.in
-SMTP_PASSWORD=Infomaskan@123
-EMAIL_FROM=info@maskantech.in
- 
-# JWT Configuration
-ACCESS_TOKEN_SECRET="L_ByN0_FIuwsQnDo4sdrOEdJvqlPjKfhVJmqhf76D13v3IWu3mbvzb8hQRnPxHMlr9Y8A9IcOHZZWSs7Kfofpg"
-ACCESS_TOKEN_EXPIRES=900
-REFRESH_TOKEN_SECRET="6Z0yOfkhPjfLH77WTnBh3Iv0JU_gWIfIqpGuGU41GFRV4fnLZMfKN3gAsPTfsqFKv1rRc6szUJRngW8Py0UYUQ"
-REFRESH_TOKEN_EXPIRES=864000
-JWT_ALGORITHM=HS256
-COOKIE_SECURE=true
-ADMIN_JWT="upRCbNd-3Ex3sG2aEHxcrCx7LZu91BkiNGPIs-vxNXp7YBHyU-0jMpGUYA5dJHLcyOIMLBk4HCw1cpI4WOlIzA"
- 
-# External SaaS Webhook
-EXTERNAL_SAAS_WEBHOOK_SECRET=your-webhook-secret-key-change-in-production
- 
-# AWS S3 Configuration
-AWS_ACCESS_KEY_ID=""
-AWS_SECRET_ACCESS_KEY=""
-AWS_REGION=""
-S3_BUCKET_NAME=""
- 
-# PayPal Integration
-PAYPAL_CLIENT_ID=""
-PAYPAL_CLIENT_SECRET=""
-PAYPAL_MODE=sandbox
- 
-# Logging
-LOG_LEVEL=info
- 
-# Development flags
-DEBUG_MODE=true
-ENABLE_DEBUG_LOGGING=true

.env.testing

@@ -1,62 +0,0 @@
-# Project Configuration
-PROJECT_NAME=SaaS Architecture
-VERSION=1.0.0
-PORT=10002
-APP_ENV=testing
-SECRET_KEY="Usu9Qmg4ppRexR6Xp657MMMHsoOaiV8cPqlY_THWNaPhGT6DN9Xd8UO4zG3kWjwIqW9hPa5bYwQUoDhyRlzv_w"
-ALLOWED_HOSTS=*
-HOST=127.0.0.1
-FRONTEND_URL=https://saas-test.maskantech.in
-CORS_ALLOWED_ORIGINS=https://saas-test.maskantech.in
- 
-# Security
-ENCRYPTION_KEY="1cd1dc2d42afc5606e224df1108162db2d6ca372a45a9b8d278162f6006236d5"
- 
-# Super Admin Configuration
-SUPER_ADMIN_EMAIL=admin@maskantech.in
-SUPER_ADMIN_PASSWORD=SuperAdmin@123
-SUPER_ADMIN_FIRST_NAME=Super
-SUPER_ADMIN_LAST_NAME=Admin
- 
-#Database Configuration
-DB_SSL=False
-DATABASE_URL=postgresql://saas_user:K9uR3mZpQ7~W4F2YH8A_tLxD@106.51.105.22:5432/saas_test
-
-# Redis Configuration
-REDIS_HOST=106.51.105.22
-REDIS_PORT=6383
-REDIS_PASSWORD=8haSTisAqop8ChAs
- 
-# Email Configuration
-SMTP_HOST=smtp.hostinger.com
-SMTP_PORT=465
-SMTP_SECURE=true
-SMTP_USER=info@maskantech.in
-SMTP_PASSWORD=Infomaskan@123
-EMAIL_FROM=info@maskantech.in
- 
-# JWT Configuration
-ACCESS_TOKEN_SECRET="L_ByN0_FIuwsQnDo4sdrOEdJvqlPjKfhVJmqhf76D13v3IWu3mbvzb8hQRnPxHMlr9Y8A9IcOHZZWSs7Kfofpg"
-ACCESS_TOKEN_EXPIRES=900
-REFRESH_TOKEN_SECRET="6Z0yOfkhPjfLH77WTnBh3Iv0JU_gWIfIqpGuGU41GFRV4fnLZMfKN3gAsPTfsqFKv1rRc6szUJRngW8Py0UYUQ"
-REFRESH_TOKEN_EXPIRES=864000
-JWT_ALGORITHM=HS256
-ADMIN_JWT="upRCbNd-3Ex3sG2aEHxcrCx7LZu91BkiNGPIs-vxNXp7YBHyU-0jMpGUYA5dJHLcyOIMLBk4HCw1cpI4WOlIzA"
- 
-# AWS S3 Configuration
-AWS_ACCESS_KEY_ID=""
-AWS_SECRET_ACCESS_KEY=""
-AWS_REGION=""
-S3_BUCKET_NAME=""
- 
-# PayPal Integration
-PAYPAL_CLIENT_ID=""
-PAYPAL_CLIENT_SECRET=""
-PAYPAL_MODE=sandbox
- 
-# Logging
-LOG_LEVEL=info
- 
-# Development flags
-DEBUG_MODE=true
-ENABLE_DEBUG_LOGGING=true

.gitignore

@@ -6,7 +6,11 @@ __pycache__/
 *.pyc
 
 # Environment
-.env
+.env*
+!.env.example
+keys/
+*.log
+
 node_modules
 
 # IDE

alembic/env.py

@@ -11,26 +11,6 @@ from alembic import context
 # Add parent directory to path to import app modules
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
-# Load environment variables before importing app
-from dotenv import load_dotenv
-
-app_env = os.getenv("APP_ENV", "local")
-env_filename = f".env.{app_env}"
-
-# Define paths
-base_path = Path(__file__).resolve().parent.parent
-backend_path = base_path
-
-# Load environment variables
-load_dotenv(dotenv_path=base_path / '.env')
-load_dotenv(dotenv_path=backend_path / '.env')
-
-# Override with specific environment config
-if (base_path / env_filename).exists():
-    load_dotenv(dotenv_path=base_path / env_filename, override=True)
-if (backend_path / env_filename).exists():
-    load_dotenv(dotenv_path=backend_path / env_filename, override=True)
-
 # Import app settings and database
 from app.config.settings import settings
 from app.config.database import Base
@@ -53,8 +33,8 @@ config = context.config
 if config.config_file_name is not None:
     fileConfig(config.config_file_name)
 
-# Set the database URL from app settings
+# Set the database URL from app settings (escape % for ConfigParser interpolation)
-config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)
+config.set_main_option("sqlalchemy.url", settings.DATABASE_URL.replace("%", "%%"))
 
 # add your model's MetaData object here
 # for 'autogenerate' support

app/__init__.py

@@ -167,12 +167,15 @@ def create_app() -> FastAPI:
             for origin in settings.CORS_ALLOWED_ORIGINS.split(",")
             if origin.strip()
         ]
+    if settings.FRONTEND_URL and settings.FRONTEND_URL not in origins:
+        origins.append(settings.FRONTEND_URL)
 
     if not origins:
         raise RuntimeError(
             "CORS_ALLOWED_ORIGINS must be set when allow_credentials=True"
         )
 
+
     app.add_middleware(
         CORSMiddleware,
         allow_origins=origins,

app/config/database.py

@@ -1,4 +1,4 @@
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, event, text
 from sqlalchemy.orm import sessionmaker, declarative_base
 from sqlalchemy.exc import SQLAlchemyError
 from fastapi import HTTPException
@@ -31,6 +31,20 @@ engine = create_engine(
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 Base = declarative_base()
 
+@event.listens_for(SessionLocal, "after_begin")
+def restore_request_rls_context(session, transaction, connection):
+    """Reapply request identity after an internal service commit starts a new transaction."""
+    if session.info.get("rls_bypass") is True:
+        connection.execute(text("SELECT set_config('app.bypass_rls', 'on', true)"))
+        return
+
+    tenant_id = session.info.get("rls_tenant_id")
+    if tenant_id:
+        connection.execute(
+            text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
+            {"tenant_id": str(tenant_id)},
+        )
+
 def get_db():
     db = SessionLocal()
     try:
@@ -50,4 +64,4 @@ def get_db():
         db.rollback()
         raise
     finally:
-        db.close()
+        db.close()

app/config/security.py

@@ -155,6 +155,24 @@ class SecurityUtils:
 
         return True
 
+    @staticmethod
+    def get_saas_private_key() -> Optional[str]:
+        """Load RSA private key from settings or persistent file (local/testing only)."""
+        if settings.SAAS_PRIVATE_KEY and settings.SAAS_PRIVATE_KEY.strip():
+            return settings.SAAS_PRIVATE_KEY
+
+        # Check persistent keys directory only in local or testing environments
+        env = (settings.APP_ENV or "").lower().strip()
+        if env in {"local", "testing"}:
+            from pathlib import Path
+            key_file = Path(__file__).resolve().parent.parent.parent / "keys" / "saas_private_key.pem"
+            if key_file.exists():
+                try:
+                    return key_file.read_text().strip()
+                except Exception:
+                    pass
+        return None
+
     @staticmethod
     def generate_module_token(data: Dict[str, Any], module_id: str, ttl_seconds: int = 900) -> str:
         """Generate short-lived module-scoped JWT (15 min default)."""
@@ -166,17 +184,19 @@ class SecurityUtils:
             "aud": str(module_id),
             "iat": datetime.now(timezone.utc).timestamp()
         })
-        
+
-        if not settings.SAAS_PRIVATE_KEY:
+        private_key = SecurityUtils.get_saas_private_key()
-             raise ValueError("SAAS_PRIVATE_KEY is not configured. Cannot sign module identity tokens.")
+        if not private_key:
-             
+            raise ValueError("SAAS_PRIVATE_KEY is not configured. Cannot sign module identity tokens.")
+
         return jwt.encode(
-            to_encode, 
+            to_encode,
-            settings.SAAS_PRIVATE_KEY, 
+            private_key,
             algorithm="RS256",
             headers={"kid": settings.SAAS_KEY_ID}
         )
 
+
     @staticmethod
     def validate_email(email: str) -> bool:
         """Validate email format."""

app/config/settings.py

@@ -1,7 +1,6 @@
-from pydantic_settings import BaseSettings
+from pydantic_settings import BaseSettings, SettingsConfigDict
 from typing import Optional
 from pathlib import Path
-from dotenv import load_dotenv
 import os
 
 app_env = os.getenv("APP_ENV", "local")
@@ -10,15 +9,25 @@ env_filename = f".env.{app_env}"
 base_path = Path(__file__).resolve().parent.parent.parent
 backend_path = Path(__file__).resolve().parent.parent
 
-load_dotenv(dotenv_path=base_path / '.env')
+env_files = []
-load_dotenv(dotenv_path=backend_path / '.env') 
+if (base_path / ".env").is_file():
+    env_files.append(str(base_path / ".env"))
+if (backend_path / ".env").is_file() and (backend_path / ".env") != (base_path / ".env"):
+    env_files.append(str(backend_path / ".env"))
+if (base_path / env_filename).is_file():
+    env_files.append(str(base_path / env_filename))
+if (backend_path / env_filename).is_file() and (backend_path / env_filename) != (base_path / env_filename):
+    env_files.append(str(backend_path / env_filename))
 
-if (base_path / env_filename).exists():
-    load_dotenv(dotenv_path=base_path / env_filename, override=True)
-if (backend_path / env_filename).exists():
-    load_dotenv(dotenv_path=backend_path / env_filename, override=True)
 
 class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=env_files,
+        env_file_encoding="utf-8",
+        case_sensitive=True,
+        extra="ignore",
+    )
+
     PROJECT_NAME: str
     VERSION: str
     
@@ -117,9 +126,5 @@ class Settings(BaseSettings):
     def MAIL_FROM(self) -> str:
         return self.EMAIL_FROM
 
-    model_config = {
-        "case_sensitive": True,
-        "extra": "ignore",
-    }
         
 settings = Settings()

app/controllers/auth/access_controller.py

@@ -1,9 +1,14 @@
 from sqlalchemy.orm import Session
-from typing import List
+from typing import List, Optional
+import uuid
 from app.services.auth.access_service import AccessService
 from app.schemas.auth.access_schema import AccessResponse
 
 class AccessController:
     @staticmethod
-    def get_all_accesses(db: Session, category: str = None) -> List[AccessResponse]:
+    def get_all_accesses(
-        return AccessService.get_all_accesses(db, category)
+        db: Session,
+        category: str = None,
+        tenant_id: Optional[uuid.UUID] = None,
+    ) -> List[AccessResponse]:
+        return AccessService.get_all_accesses(db, category, tenant_id=tenant_id)

app/controllers/auth/module_controller.py

@@ -18,7 +18,9 @@ class ModuleController:
     @staticmethod
     def get_available_modules(db: Session, current_user) -> List[ModuleResponse]:
         from app.schemas.auth.module_schema import ModuleAvailableResponse
-        results = ModuleService.get_available_modules(db, current_user.tenant_id)
+        results = ModuleService.get_available_modules(
+            db, current_user.tenant_id, current_user.role_id
+        )
         
         response = []
         for item in results:
@@ -46,4 +48,4 @@ class ModuleController:
     @staticmethod
     def delete_module(db: Session, module_id: str):
         ModuleService.delete_module(db, module_id)
-        return {"message": "Module deleted successfully"}
+        return {"message": "Module deleted successfully"}

app/controllers/auth/sso_controller.py

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
 class SSOController:
     @staticmethod
     def initiate_sso(db: Session, request: SSOInitiateRequest, current_user: User):
-        return SSOService.generate_signed_payload(
+        return SSOService.generate_grant(
             db=db,
             user_id=current_user.id,
             module_id=request.module_id,
@@ -37,10 +37,16 @@ class SSOController:
             ModuleEnvironment.module_id == module.id,
             ModuleEnvironment.slug == payload.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="Environment not found")
-            
+
         headers = {}
         if x_module_signature:
             headers["X-Module-Signature"] = x_module_signature
@@ -70,4 +76,4 @@ class SSOController:
             grant_code=payload.grant_code,
             module_id=payload.module_id,
             environment_slug=payload.environment_slug
-        )
+        )

app/controllers/auth/tenant_controller.py

@@ -7,8 +7,8 @@ from typing import List, Optional
 class TenantController:
     
     @staticmethod
-    def create_tenant(db: Session, tenant_data: TenantCreate):
+    def create_tenant(db: Session, tenant_data: TenantCreate, commit: bool = True):
-        return TenantService.create_tenant(db, tenant_data)
+        return TenantService.create_tenant(db, tenant_data, commit=commit)
     
     @staticmethod
     def get_all_tenants(db: Session):

app/middleware/auth_middleware.py

@@ -1,5 +1,6 @@
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy import text
 from sqlalchemy.orm import Session
 from typing import List
 from datetime import datetime, timezone
@@ -33,6 +34,25 @@ def get_current_user(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 detail="Invalid token payload"
             )
+
+        # The verified access token carries the canonical tenant UUID. Scope
+        # the database transaction before reading any tenant-owned records.
+        token_tenant_id = payload.get("tenant_id")
+        if token_tenant_id:
+            db.info.pop("rls_bypass", None)
+            db.info["rls_tenant_id"] = str(token_tenant_id)
+            db.execute(
+                text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
+                {"tenant_id": str(token_tenant_id)},
+            )
+        else:
+            # A platform user's signed token intentionally has no tenant claim.
+            # Enable transaction-local discovery only long enough to resolve the
+            # exact token subject; the result is validated below before platform
+            # access remains enabled for the rest of this request.
+            db.info.pop("rls_tenant_id", None)
+            db.info["rls_bypass"] = True
+            db.execute(text("SELECT set_config('app.bypass_rls', 'on', true)"))
     except HTTPException:
         raise
     except Exception:
@@ -43,11 +63,30 @@ def get_current_user(
 
     user = db.query(User).filter(User.id == user_id).first()
     if user is None:
+        if not token_tenant_id:
+            db.info.pop("rls_bypass", None)
+            db.execute(text("SELECT set_config('app.bypass_rls', 'off', true)"))
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="User not found"
         )
 
+    if token_tenant_id:
+        if str(user.tenant_id) != str(token_tenant_id):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Token tenant does not match user tenant"
+            )
+    elif user.tenant_id is not None:
+        # Fail closed for a validly signed tenant-user token that is malformed,
+        # stale, or missing its canonical tenant claim.
+        db.info.pop("rls_bypass", None)
+        db.execute(text("SELECT set_config('app.bypass_rls', 'off', true)"))
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Tenant claim required for tenant user"
+        )
+
     setattr(user, "_saas_db_session", db)
     
     if user.status != "active":

app/models/auth/role_model.py

@@ -28,5 +28,13 @@ class Role(Base):
         "RoleModuleAccess", back_populates="role", cascade="all, delete-orphan"
     )
 
+    @property
+    def assigned_modules(self):
+        names = set()
+        for rma in self.role_module_accesses:
+            if rma.module_access and rma.module_access.module:
+                names.add(rma.module_access.module.module_name)
+        return sorted(list(names))
+
     def __repr__(self):
         return f"<Role {self.role_name}>"

app/models/auth/user_model.py

@@ -27,6 +27,11 @@ class User(Base):
     tenant = relationship("Tenant", back_populates="users")
     role = relationship("Role", back_populates="users")
 
+    @property
+    def canonical_user_id(self) -> uuid.UUID:
+        """authoritative canonical user identity across the SaaS platform."""
+        return self.id
+
     def __repr__(self):
         return f"<User {self.email}>"
 

app/routes/auth/access.py

@@ -16,4 +16,6 @@ def get_accesses(
     current_user: User = Depends(get_current_user),
     _ = Depends(require_access("admin.role.read")),
 ):
-    return AccessController.get_all_accesses(db, category)
+    return AccessController.get_all_accesses(
+        db, category, tenant_id=current_user.tenant_id
+    )

app/routes/auth/subscription_plan.py

@@ -74,21 +74,10 @@ def get_plan(
     current_user: User = Depends(get_current_user),
     _ = Depends(require_access("superadmin.plan.read"))
 ):
+    from app.services.auth.subscription_plan_service import SubscriptionPlanService
     plan = SubscriptionPlanController.get_plan(db, plan_id)
-    return SubscriptionPlanDetailResponse(
+    return SubscriptionPlanService.serialize_plan_response(db, plan, detail=True)
-        id=plan.id,
+
-        name=plan.name,
-        description=plan.description,
-        price=float(plan.price) if plan.price is not None else None,
-        duration_days=plan.duration_days,
-        max_users_allowed=plan.max_users_allowed,
-        is_public=plan.is_public,
-        status=plan.status,
-        created_at=plan.created_at,
-        updated_at=plan.updated_at,
-        access_ids=[pa.access_id for pa in plan.plan_accesses],
-        module_access_ids=[pma.module_access_id for pma in plan.plan_module_accesses]
-    )
 
 @router.get("/list", response_model=SubscriptionPlanPaginatedResponse)
 def list_plans(

app/routes/auth/tenant.py

@@ -9,6 +9,7 @@ from app.schemas.auth.tenant_schema import TenantCreate, TenantUpdate, TenantRes
 from app.middleware.auth_middleware import get_current_user, require_access, User
 from app.middleware.tenant_middleware import get_tenant_id_from_user
 from app.services.system.audit_log_service import AuditLogService
+from app.services.auth.event_service import EventService
 from app.helper.helpers import get_client_ip
 
 router = APIRouter()
@@ -21,21 +22,44 @@ def create_tenant(
     current_user: User = Depends(get_current_user),
     _ = Depends(require_access("superadmin.tenant.create"))
 ):
-    result = TenantController.create_tenant(db, tenant_data)
+    try:
+        # 1. Create tenant/owner/modules/events using flush
+        result = TenantController.create_tenant(db, tenant_data, commit=False)
 
-    AuditLogService.log(
+        # 2. Sanitize payload for audit log (never store raw passwords or hashes)
-        db=db,
+        audit_values = tenant_data.model_dump(mode='json')
-        module_name="Tenants",
+        if "owner" in audit_values and isinstance(audit_values["owner"], dict):
-        action_type="CREATE",
+            audit_values["owner"]["password"] = "[PROTECTED]"
-        entity_id=str(result.id),
+
-        entity_name=result.tenant_name,
+        # 3. Create sanitized audit record using flush (strict=True)
-        description=f"Tenant '{result.tenant_name}' onboarded",
+        AuditLogService.log(
-        performed_by_id=str(current_user.id),
+            db=db,
-        performed_by_email=current_user.email,
+            module_name="Tenants",
-        ip_address=get_client_ip(request),
+            action_type="CREATE",
-        new_values=tenant_data.model_dump(mode='json')
+            entity_id=str(result.id),
-    )
+            entity_name=result.tenant_name,
-    return result
+            description=f"Tenant '{result.tenant_name}' onboarded",
+            performed_by_id=str(current_user.id),
+            performed_by_email=current_user.email,
+            ip_address=get_client_ip(request),
+            new_values=audit_values,
+            commit=False,
+            strict=True,
+        )
+
+        # 4. Perform single final commit
+        db.commit()
+        db.refresh(result)
+
+        # 5. Best-effort post-commit Redis wake-up notification (never breaks response)
+        pending_ids = getattr(result, "pending_event_ids", [])
+        if pending_ids:
+            EventService.enqueue_wakeups(pending_ids)
+
+        return result
+    except Exception:
+        db.rollback()
+        raise
 
 @router.get("/get", response_model=List[TenantResponse])
 def get_all_tenants(

app/schemas/auth/role_schema.py

@@ -8,7 +8,7 @@ class RoleBase(BaseModel):
     tenant_id: Optional[uuid.UUID] = None
 
 class RoleCreate(RoleBase):
-    access_ids: List[uuid.UUID] = []
+    access_ids: List[uuid.UUID] = Field(default_factory=list)
     is_default: Optional[bool] = False
 
 class RoleUpdate(BaseModel):
@@ -19,6 +19,7 @@ class RoleUpdate(BaseModel):
 class RoleResponse(RoleBase):
     id: uuid.UUID
     is_default: bool = False
+    assigned_modules: List[str] = Field(default_factory=list)
     created_at: datetime
     updated_at: datetime
 
@@ -31,7 +32,7 @@ class RoleResponse(RoleBase):
         from_attributes = True
 
 class RoleWithAccessesResponse(RoleResponse):
-    accesses: List[dict] = []
+    accesses: List[dict] = Field(default_factory=list)
 
     class Config:
         from_attributes = True
@@ -41,4 +42,4 @@ class RolePaginatedResponse(BaseModel):
     total: int
     page: int
     page_size: int
-    total_pages: int
+    total_pages: int

app/schemas/auth/subscription_plan_schema.py

@@ -27,10 +27,35 @@ class SubscriptionPlanUpdate(BaseModel):
     access_ids: Optional[List[uuid.UUID]] = None
     module_access_ids: Optional[List[uuid.UUID]] = None
 
+class PlanApplicationEnvironmentSummary(BaseModel):
+    id: uuid.UUID
+    slug: str
+    is_default: bool = False
+    is_active: bool = True
+    frontend_base_url: str
+    sso_entry_path: Optional[str] = "/sso/callback"
+    backend_base_url: str
+
+    class Config:
+        from_attributes = True
+
+class PlanApplicationSummary(BaseModel):
+    module_id: uuid.UUID
+    module_code: str
+    module_name: str
+    description: Optional[str] = None
+    icon_url: Optional[str] = None
+    display_order: int = 0
+    environments: List[PlanApplicationEnvironmentSummary] = []
+
+    class Config:
+        from_attributes = True
+
 class SubscriptionPlanResponse(SubscriptionPlanBase):
     id: uuid.UUID
     created_at: datetime
     updated_at: datetime
+    applications: List[PlanApplicationSummary] = []
 
     class Config:
         from_attributes = True
@@ -38,6 +63,7 @@ class SubscriptionPlanResponse(SubscriptionPlanBase):
 class SubscriptionPlanDetailResponse(SubscriptionPlanResponse):
     access_ids: List[uuid.UUID] = []
     module_access_ids: List[uuid.UUID] = []
+    applications: List[PlanApplicationSummary] = []
 
 class SubscriptionPlanPaginatedResponse(BaseModel):
     items: List[SubscriptionPlanResponse]

app/schemas/auth/tenant_schema.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, EmailStr
 from typing import Optional, List
 from datetime import date, datetime
 import uuid
@@ -12,14 +12,23 @@ class ModuleEnvironmentAssignment(BaseModel):
     module_id: uuid.UUID
     environment_slug: str
 
+class TenantOwnerCreate(BaseModel):
+    first_name: str = Field(..., min_length=1, max_length=50)
+    last_name: str = Field(..., min_length=1, max_length=50)
+    email: EmailStr
+    password: str = Field(..., min_length=8)
+    phone_number: Optional[str] = None
+
 class TenantCreate(TenantBase):
     plan_id: uuid.UUID
+    selected_module_ids: List[uuid.UUID] = Field(..., min_length=1)
+    module_environments: List[ModuleEnvironmentAssignment] = Field(default_factory=list)
+    owner: TenantOwnerCreate
     start_date: Optional[date] = None
     end_date: Optional[date] = None
     status: Optional[str] = "ACTIVE"
 
-    module_environments: Optional[List[ModuleEnvironmentAssignment]] = []
+
-    default_environment_slug: str = "prod"
 
 class TenantUpdate(BaseModel):
     tenant_name: Optional[str] = Field(None, min_length=2, max_length=100)

app/services/auth/access_service.py

@@ -1,11 +1,13 @@
 from sqlalchemy.orm import Session, joinedload
 from app.models.auth.access_model import Access
-from typing import List, Any
+from typing import List, Any, Optional
+import uuid
 from app.core.redis import sync_redis_client
 import json
 import logging
 from app.models.auth.module_access_model import ModuleAccess
 from app.models.auth.module_model import Module
+from app.models.auth.tenant_module_model import TenantModule
 from datetime import datetime
 
 logger = logging.getLogger(__name__)
@@ -13,11 +15,11 @@ logger = logging.getLogger(__name__)
 class AccessService:
     
     @staticmethod
-    def get_all_accesses(db: Session, category: str = None) -> List[any]:
+    def get_all_accesses(db: Session, category: str = None, tenant_id: Optional[uuid.UUID] = None) -> List[any]:
-        
+
-        cache_key = f"saas:access:v2:all:{category if category else 'full'}"
+        cache_key = f"saas:access:v4:active:{category if category else 'full'}:{str(tenant_id) if tenant_id else 'global'}"
-        cached_data = sync_redis_client.client.get(cache_key) if sync_redis_client.client else None
+        cached_data = None
-        
+
         if cached_data:
             try:
                 data_list = json.loads(cached_data)
@@ -44,15 +46,35 @@ class AccessService:
         if category:
             query = query.filter(Access.category == category)
         
+        if tenant_id:
+            query = query.filter(Access.category != "Superadmin")
+
         saas_accesses = query.all()
         for access in saas_accesses:
             access.module_name = "SaaS (Internal)"
         
-        module_query = db.query(ModuleAccess).options(joinedload(ModuleAccess.module))
+        module_query = (
+            db.query(ModuleAccess)
+            .join(Module, ModuleAccess.module_id == Module.id)
+            .filter(Module.status == "active")
+            .options(joinedload(ModuleAccess.module))
+        )
         
         if category:
             module_query = module_query.filter(ModuleAccess.category == category)
         
+        if tenant_id:
+            active_tms = (
+                db.query(TenantModule.module_id)
+                .filter(
+                    TenantModule.tenant_id == tenant_id,
+                    TenantModule.is_active == True
+                )
+                .all()
+            )
+            active_module_ids = [tm[0] for tm in active_tms]
+            module_query = module_query.filter(ModuleAccess.module_id.in_(active_module_ids))
+
         module_accesses = module_query.all()
         
         for ma in module_accesses:
@@ -87,4 +109,4 @@ class AccessService:
         module_categories = db.query(ModuleAccess.category).distinct().all()
         
         all_cats = set([cat[0] for cat in categories] + [cat[0] for cat in module_categories])
-        return list(all_cats)
+        return list(all_cats)

app/services/auth/auth_service.py

@@ -1,3 +1,4 @@
+from sqlalchemy import text
 from sqlalchemy.orm import Session
 from fastapi import HTTPException, status
 from app.models.auth.user_model import User
@@ -46,8 +47,20 @@ class AuthService:
 
     @staticmethod
     def signin(db: Session, signin_data: UserSignin):
+        # Sign-in starts before the caller's tenant is known. Use the database's
+        # transaction-local discovery bypass only for the globally unique email
+        # lookup, then immediately restore tenant isolation for the rest of the
+        # authentication request.
+        db.execute(text("SELECT set_config('app.bypass_rls', 'on', true)"))
         user = db.query(User).filter(User.email == signin_data.email).first()
 
+        db.execute(text("SELECT set_config('app.bypass_rls', 'off', true)"))
+        if user and user.tenant_id:
+            db.execute(
+                text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
+                {"tenant_id": str(user.tenant_id)},
+            )
+
         if not user or not security.verify_password(
             signin_data.password, user.password
         ):
@@ -98,6 +111,13 @@ class AuthService:
     def refresh_access_token(db: Session, refresh_token: str):
         payload = security.verify_refresh_token(refresh_token)
         user_id = payload.get("sub")
+        token_tenant_id = payload.get("tenant_id")
+
+        if token_tenant_id:
+            db.execute(
+                text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
+                {"tenant_id": str(token_tenant_id)},
+            )
 
         user = db.query(User).filter(User.id == user_id).first()
         if not user:

app/services/auth/event_service.py

@@ -2,6 +2,7 @@ import uuid
 import httpx
 import json
 import logging
+import time
 from datetime import datetime, timezone, timedelta
 from typing import Dict, Any, List, Optional
 from sqlalchemy.orm import Session
@@ -25,10 +26,11 @@ class EventService:
         payload: Dict[str, Any],
         tenant_id: Optional[uuid.UUID] = None,
         follow_up_event: Optional[Dict[str, Any]] = None
-    ):
+    ) -> Optional[str]:
         """
-        Emits an event by writing it to the Outbox (event_logs).
+        Emits an event by writing it to the Outbox (event_logs) and flushing.
-        Scopes delivery to relevant modules based on tenant_id.
+        Does NOT notify Redis directly before database transaction commit.
+        Returns the logical event_id on success, or None if no targets exist.
         """
         event_id = str(uuid.uuid4())
         timestamp = datetime.now(timezone.utc).isoformat()
@@ -56,12 +58,14 @@ class EventService:
                 env_slug = target.get("environment_slug")
                 
                 if module_id and env_slug:
+                    mod_uuid = uuid.UUID(str(module_id)) if not isinstance(module_id, uuid.UUID) else module_id
                     env = db.query(ModuleEnvironment).filter(
-                        ModuleEnvironment.module_id == module_id,
+                        ModuleEnvironment.module_id == mod_uuid,
                         ModuleEnvironment.slug == env_slug
                     ).first()
                     if env:
                         targets.append(env)
+
                         
         elif tenant_id:
             tenant_modules = db.query(TenantModule).filter(
@@ -85,19 +89,16 @@ class EventService:
                     targets.append(env)
         
         if not targets:
-            logger.warning(f"Event {event_type} emitted with no resolved targets. Payload scoping: {'Explicit' if 'targets' in payload.get('data', {}) else 'Implicit'}")
+            logger.warning(f"Event {event_type} (id={event_id}) emitted with no resolved targets.")
-            return
+            return None
 
         for env in targets:
             base = env.backend_base_url.rstrip('/')
             
             if event_type in {"TENANT_PROVISION_REQUESTED", "TENANT_UPDATED", "TENANT_STATUS_CHANGED", "TENANT_DEPROVISION_REQUESTED"} and env.provisioning_endpoint:
                 endpoint = env.provisioning_endpoint.lstrip('/')
-                logger.info(f"Trace: base='{base}', endpoint='{endpoint}'") 
                 target_url = f"{base}/{endpoint}"
-                logger.info(f"Trace: Calculated target_url='{target_url}'")
             else:
-                logger.info(f"Using default event stream for env '{env.slug}'. ProvEndpoint: '{env.provisioning_endpoint}'")
                 target_url = f"{base}/api/internal/events"
             
             log = EventLog(
@@ -113,12 +114,24 @@ class EventService:
             )
             db.add(log)
         
-        db.flush() 
+        db.flush()
-        
+        return event_id
-        try:
+
-             sync_redis_client.rpush("saas:events:queue", event_id)
+    @staticmethod
-        except Exception as e:
+    def enqueue_wakeups(event_ids: List[str]) -> None:
-             logger.error(f"Failed to push event to Redis queue: {e}")
+        """
+        Best-effort Redis notification to wake up workers after DB commit.
+        A Redis failure here must never break the main caller since durable Outbox exists.
+        """
+        if not event_ids:
+            return
+        for eid in event_ids:
+            if not eid:
+                continue
+            try:
+                sync_redis_client.rpush("saas:events:queue", eid)
+            except Exception as e:
+                logger.warning(f"Failed to push wake-up event {eid} to Redis queue: {e}")
 
     @staticmethod
     def process_queue_item(db: Session, event_id: str):
@@ -146,26 +159,36 @@ class EventService:
                     log.error_log = "Target environment config missing"
                     continue
                 
-                payload_json = json.dumps(log.payload)
+                raw_bytes = json.dumps(log.payload, separators=(',', ':')).encode("utf-8")
                 secret = env.trust_credentials.get("hmac_secret") if env.trust_credentials else None
+                key_id = env.trust_credentials.get("key_id", "v1") if env.trust_credentials else "v1"
                 
+                timestamp = str(int(time.time() * 1000))
                 if secret:
+                    body_hash = hashlib.sha256(raw_bytes).hexdigest()
+                    canonical = f"{timestamp}.{body_hash}"
                     signature = hmac.new(
                         secret.encode("utf-8"),
-                        payload_json.encode("utf-8"),
+                        canonical.encode("utf-8"),
                         hashlib.sha256
                     ).hexdigest()
                 else:
-                     signature = ""
+                    signature = ""
                 
                 headers = {
                     "Content-Type": "application/json",
+                    "X-Integration-Timestamp": timestamp,
+                    "X-Integration-Signature": signature,
+                    "X-Integration-Key-Id": key_id,
+                    "X-Integration-Source": "saas-core",
+                    "X-SaaS-Timestamp": timestamp,
                     "X-SaaS-Signature": signature,
+                    "X-Signature": signature,
                     "X-SaaS-Event-Source": "saas-core"
                 }
-                
+
                 logger.info(f"Sending event {log.event_type} to {log.target_url}")
-                response = httpx.post(log.target_url, content=payload_json, headers=headers, timeout=5)
+                response = httpx.post(log.target_url, content=raw_bytes, headers=headers, timeout=5)
                 
                 if response.status_code in range(200, 300):
                     log.status = EventStatus.COMPLETED
@@ -226,26 +249,36 @@ class EventService:
                     log.error_log = "Target environment config missing"
                     continue
                 
-                payload_json = json.dumps(log.payload)
+                raw_bytes = json.dumps(log.payload, separators=(',', ':')).encode("utf-8")
-                secret = env.trust_credentials.get("hmac_secret")
+                secret = env.trust_credentials.get("hmac_secret") if env.trust_credentials else None
+                key_id = env.trust_credentials.get("key_id", "v1") if env.trust_credentials else "v1"
                 
+                timestamp = str(int(time.time() * 1000))
                 if secret:
+                    body_hash = hashlib.sha256(raw_bytes).hexdigest()
+                    canonical = f"{timestamp}.{body_hash}"
                     signature = hmac.new(
                         secret.encode("utf-8"),
-                        payload_json.encode("utf-8"),
+                        canonical.encode("utf-8"),
                         hashlib.sha256
                     ).hexdigest()
                 else:
-                     signature = ""
+                    signature = ""
                 
                 headers = {
                     "Content-Type": "application/json",
+                    "X-Integration-Timestamp": timestamp,
+                    "X-Integration-Signature": signature,
+                    "X-Integration-Key-Id": key_id,
+                    "X-Integration-Source": "saas-core",
+                    "X-SaaS-Timestamp": timestamp,
                     "X-SaaS-Signature": signature,
+                    "X-Signature": signature,
                     "X-SaaS-Event-Source": "saas-core"
                 }
-                
+
-                logger.info(f"Sending event {log.event_type} to {log.target_url}. Payload: {payload_json}")
+                logger.info(f"Sending event {log.event_type} (id={log.event_id}) to {log.target_url}")
-                response = httpx.post(log.target_url, content=payload_json, headers=headers, timeout=5)
+                response = httpx.post(log.target_url, content=raw_bytes, headers=headers, timeout=5)
                 
                 if response.status_code in range(200, 300):
                     log.status = EventStatus.COMPLETED

app/services/auth/module_environment_service.py

@@ -73,9 +73,13 @@ class ModuleEnvironmentService:
         for key, value in update_data.items():
             setattr(environment, key, value)
         
-        db.commit()
+        try:
-        db.refresh(environment)
+            db.commit()
-        return environment
+            db.refresh(environment)
+            return environment
+        except IntegrityError:
+            db.rollback()
+            raise HTTPException(status_code=409, detail="Environment slug already exists for this module")
 
     @staticmethod
     def set_default_environment(db: Session, module_id: str, env_id: str):

app/services/auth/module_permission_service.py

@@ -52,23 +52,20 @@ class ModulePermissionService:
             
             secret = env.trust_credentials.get("hmac_secret")
             if not secret:
-                 pass 
+                raise ValueError("Module environment is missing its HMAC trust secret")
-            
+
-            if secret:
+            signature = hmac.new(
-                signature = hmac.new(
+                secret.encode("utf-8"),
-                    secret.encode("utf-8"),
+                payload_body.encode("utf-8"),
-                    payload_body.encode("utf-8"),
+                hashlib.sha256
-                    hashlib.sha256
+            ).hexdigest()
-                ).hexdigest()
-            else:
-                signature = ""
 
             headers = {
                 "Content-Type": "application/json",
                 "X-SaaS-Signature": signature
             }
             
-            url = f"{env.backend_base_url}{env.permission_sync_endpoint}"
+            url = f"{env.backend_base_url.rstrip('/')}/{env.permission_sync_endpoint.lstrip('/')}"
             response = requests.post(url, headers=headers, data=payload_body, timeout=10)
             response.raise_for_status()
             data = response.json()
@@ -80,18 +77,26 @@ class ModulePermissionService:
         
         synced_count = 0
         timestamp = datetime.now(timezone.utc)
-        
+
-        permission_map = {}
+        incoming_codes = {
+            perm.get("permission_code")
+            for perm in permissions
+            if perm.get("permission_code")
+        }
+        existing_accesses = db.query(ModuleAccess).filter(
+            ModuleAccess.module_id == module.id,
+            ModuleAccess.access_code.in_(incoming_codes),
+        ).all() if incoming_codes else []
+        permission_map = {
+            access.access_code: access for access in existing_accesses
+        }
         
         for perm in permissions:
             code = perm.get("permission_code")
             if not code:
                 continue
                 
-            access = db.query(ModuleAccess).filter(
+            access = permission_map.get(code)
-                ModuleAccess.module_id == module.id,
-                ModuleAccess.access_code == code
-            ).first()
             
             if not access:
                 access = ModuleAccess(
@@ -101,6 +106,7 @@ class ModulePermissionService:
                     category=perm.get("category", "General"),
                 )
                 db.add(access)
+                permission_map[code] = access
             else:
                 access.name = perm.get("name", access.name)
                 access.category = perm.get("category", access.category)
@@ -108,7 +114,6 @@ class ModulePermissionService:
             access.last_synced_at = timestamp
             access.sync_checksum = perm.get("hash")
             
-            permission_map[code] = access
             synced_count += 1
             
         db.flush()
@@ -123,12 +128,6 @@ class ModulePermissionService:
             access = permission_map.get(code)
             parent_access = permission_map.get(parent_code)
             
-            if not parent_access:
-                parent_access = db.query(ModuleAccess).filter(
-                    ModuleAccess.module_id == module.id,
-                    ModuleAccess.access_code == parent_code
-                ).first()
-                
             if access and parent_access:
                 access.parent_id = parent_access.id
             
@@ -159,4 +158,4 @@ class ModulePermissionService:
              
         return db.query(ModuleAccess).filter(
             ModuleAccess.module_id == module.id
-        ).all()
+        ).all()

app/services/auth/module_service.py

@@ -5,6 +5,8 @@ from typing import List, Optional
 import uuid
 from app.models.auth.tenant_module_model import TenantModule
 from app.models.auth.module_model import Module
+from app.models.auth.module_access_model import ModuleAccess
+from app.models.auth.role_module_access_model import RoleModuleAccess
 from app.schemas.auth.module_schema import ModuleCreate, ModuleUpdate
 
 class ModuleService:
@@ -39,15 +41,25 @@ class ModuleService:
         return module
 
     @staticmethod
-    def get_available_modules(db: Session, tenant_id: Optional[uuid.UUID]) -> List[dict]:
+    def get_available_modules(
+        db: Session,
+        tenant_id: Optional[uuid.UUID],
+        role_id: Optional[uuid.UUID] = None,
+    ) -> List[dict]:
         if tenant_id:
             results = db.query(Module, TenantModule.is_active).join(
                 TenantModule, 
                 (TenantModule.module_id == Module.id) & (TenantModule.tenant_id == tenant_id)
+            ).join(
+                ModuleAccess, ModuleAccess.module_id == Module.id
+            ).join(
+                RoleModuleAccess,
+                (RoleModuleAccess.module_access_id == ModuleAccess.id) &
+                (RoleModuleAccess.role_id == role_id)
             ).filter(
                 Module.status == "active",
                 TenantModule.is_active == True
-            ).order_by(Module.display_order).all()
+            ).distinct().order_by(Module.display_order).all()
             
             return [{"module": mod, "is_active": is_active} for mod, is_active in results]
         else:
@@ -58,6 +70,17 @@ class ModuleService:
     def get_module_by_module_id(db: Session, module_id_str: str) -> Optional[Module]:
         return db.query(Module).filter(Module.module_id == module_id_str).first()
 
+    @staticmethod
+    def role_can_launch_module(db: Session, role_id: Optional[uuid.UUID], module_id: uuid.UUID) -> bool:
+        if not role_id:
+            return False
+        return db.query(RoleModuleAccess.id).join(
+            ModuleAccess, ModuleAccess.id == RoleModuleAccess.module_access_id
+        ).filter(
+            RoleModuleAccess.role_id == role_id,
+            ModuleAccess.module_id == module_id,
+        ).first() is not None
+
     @staticmethod
     def update_module(db: Session, module_id: str, module_data: ModuleUpdate) -> Module:
         module = ModuleService.get_module(db, module_id)
@@ -74,4 +97,4 @@ class ModuleService:
     def delete_module(db: Session, module_id: str):
         module = ModuleService.get_module(db, module_id)
         db.delete(module)
-        db.commit()
+        db.commit()

app/services/auth/role_service.py

@@ -19,7 +19,12 @@ logger = logging.getLogger(__name__)
 class RoleService:
 
     @staticmethod
-    def create_role(db: Session, role_data: RoleCreate, emit_events: bool = True) -> Role:
+    def create_role(
+        db: Session,
+        role_data: RoleCreate,
+        emit_events: bool = True,
+        commit: bool = True,
+    ) -> Role:
         existing = (
             db.query(Role)
             .filter(
@@ -42,11 +47,16 @@ class RoleService:
         )
 
         db.add(role)
-        db.commit()
+        if commit:
-        db.refresh(role)
+            db.commit()
+            db.refresh(role)
+        else:
+            db.flush()
 
         if role_data.access_ids:
-            RoleService.assign_accesses(db, role.id, role_data.access_ids)
+            RoleService.assign_accesses(
+                db, role.id, role_data.access_ids, commit=commit
+            )
             
         if emit_events:
             assigned_modules = (
@@ -104,12 +114,22 @@ class RoleService:
                         tenant_id=role.tenant_id
                     )
 
-                    db.commit()
+                    if commit:
+                        db.commit()
+                    else:
+                        db.flush()
 
         return role
 
     @staticmethod
-    def assign_accesses(db: Session, role_id: uuid.UUID, access_ids: List[uuid.UUID]):
+    def assign_accesses(
+        db: Session,
+        role_id: uuid.UUID,
+        access_ids: List[uuid.UUID],
+        commit: bool = True,
+    ):
+        role = db.query(Role).filter(Role.id == role_id).first()
+
         db.query(RoleAccess).filter(RoleAccess.role_id == role_id).delete()
         db.query(RoleModuleAccess).filter(RoleModuleAccess.role_id == role_id).delete()
 
@@ -119,6 +139,13 @@ class RoleService:
         saas_accesses = db.query(Access).filter(Access.id.in_(access_ids)).all()
         saas_ids = {a.id for a in saas_accesses}
         
+        if role and role.tenant_id:
+            if any(a.category == "Superadmin" for a in saas_accesses):
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Cannot assign superadmin permissions to a tenant role.",
+                )
+
         for access in saas_accesses:
             db.add(RoleAccess(role_id=role_id, access_id=access.id))
             
@@ -126,10 +153,31 @@ class RoleService:
         
         if remaining_ids:
             module_accesses = db.query(ModuleAccess).filter(ModuleAccess.id.in_(remaining_ids)).all()
+            if role and role.tenant_id:
+                from app.models.auth.tenant_module_model import TenantModule
+                active_tms = (
+                    db.query(TenantModule.module_id)
+                    .filter(
+                        TenantModule.tenant_id == role.tenant_id,
+                        TenantModule.is_active == True,
+                    )
+                    .all()
+                )
+                allowed_module_ids = {tm[0] for tm in active_tms}
+                for ma in module_accesses:
+                    if ma.module_id not in allowed_module_ids:
+                        raise HTTPException(
+                            status_code=status.HTTP_400_BAD_REQUEST,
+                            detail=f"Cannot assign permission '{ma.name}' for unsubscribed module to a tenant role.",
+                        )
+
             for access in module_accesses:
                 db.add(RoleModuleAccess(role_id=role_id, module_access_id=access.id))
 
-        db.commit()
+        if commit:
+            db.commit()
+        else:
+            db.flush()
 
     @staticmethod
     def get_all_roles(db: Session, tenant_id: uuid.UUID = None):
@@ -361,4 +409,4 @@ class RoleService:
             page=page,
             page_size=page_size,
             total_pages=total_pages,
-        )
+        )

app/services/auth/sso_service.py

@@ -2,6 +2,7 @@ import uuid
 import logging
 from datetime import datetime, timedelta, timezone
 from typing import Dict, Any, Optional
+from sqlalchemy import text
 from sqlalchemy.orm import Session
 from fastapi import HTTPException, status
 from app.models.auth.sso_grant_model import SSOGrant
@@ -17,6 +18,7 @@ import time
 from app.services.auth.subscription_entitlement_service import (
     SubscriptionEntitlementService,
 )
+from app.services.auth.module_service import ModuleService
 
 logger = logging.getLogger(__name__)
 
@@ -49,6 +51,10 @@ class SSOService:
             
             if not tm or not tm.is_active:
                  raise HTTPException(status_code=403, detail="Tenant does not have access to this module")
+
+            user = db.query(User).filter(User.id == user_id).first()
+            if not user or not ModuleService.role_can_launch_module(db, user.role_id, module.id):
+                raise HTTPException(status_code=403, detail="User is not assigned to this module")
             
             if tm.assigned_environment_slug:
                 environment_slug = tm.assigned_environment_slug
@@ -69,7 +75,11 @@ class SSOService:
 
         grant_code = str(uuid.uuid4().hex)
 
-        redirect_url = f"{env.frontend_base_url}{env.sso_entry_path}?grant={grant_code}"
+        frontend_base_url = env.frontend_base_url.rstrip("/")
+        entry_path = env.sso_entry_path or "/sso/callback"
+        if not entry_path.startswith("/"):
+            entry_path = f"/{entry_path}"
+        redirect_url = f"{frontend_base_url}{entry_path}?grant={grant_code}"
 
         grant_data = json.dumps({
             "user_id": str(user_id),
@@ -109,7 +119,12 @@ class SSOService:
         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:
+            if not tm or not tm.is_active:
+                raise HTTPException(status_code=403, detail="Tenant does not have access to this module")
+            user = db.query(User).filter(User.id == user_id).first()
+            if not user or not ModuleService.role_can_launch_module(db, user.role_id, module.id):
+                raise HTTPException(status_code=403, detail="User is not assigned 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()
@@ -123,19 +138,9 @@ class SSOService:
         if not user:
              raise HTTPException(status_code=404, detail="User not found")
 
-        permissions = sorted(
+        # SaaS controls identity and module entry. Operational authorization is
-            SubscriptionEntitlementService.get_plan_module_access_codes(
+        # deliberately owned by the destination module.
-                db, tenant_id, module.id
+        permissions = []
-            )
-        )
-        if not permissions and user.role and user.role.role_module_accesses:
-            permissions = sorted(
-                {
-                    rma.module_access.access_code
-                    for rma in user.role.role_module_accesses
-                    if rma.module_access and rma.module_access.module_id == module.id
-                }
-            )
 
         timestamp = int(time.time() * 1000)
         subscription_details = SubscriptionEntitlementService.get_subscription_summary(
@@ -219,11 +224,19 @@ class SSOService:
         if grant_data["environment_slug"] != environment_slug:
             raise HTTPException(status_code=401, detail="Grant invalid for this environment")
 
+        grant_tenant_id = grant_data.get("tenant_id")
+        if grant_tenant_id:
+            db.execute(
+                text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
+                {"tenant_id": str(grant_tenant_id)},
+            )
+        else:
+            db.execute(text("SELECT set_config('app.bypass_rls', 'on', true)"))
+
         user = db.query(User).filter(User.id == grant_data["user_id"]).first()
         if not user:
-            raise HTTPException(status_code=401, detail="User not found")
+             raise HTTPException(status_code=401, detail="User not found")
 
-        grant_tenant_id = grant_data.get("tenant_id")
         if grant_tenant_id:
             if str(user.tenant_id) != grant_tenant_id:
                 raise HTTPException(
@@ -232,28 +245,23 @@ class SSOService:
                 )
 
         permissions = sorted(
-            SubscriptionEntitlementService.get_plan_module_access_codes(
+            SubscriptionEntitlementService.get_effective_module_access_codes(
-                db,
+                db, user, module.id
-                uuid.UUID(grant_tenant_id) if grant_tenant_id else None,
-                module.id,
             )
         )
-        if not permissions and user.role and user.role.role_module_accesses:
-            permissions = sorted(
-                {
-                    rma.module_access.access_code
-                    for rma in user.role.role_module_accesses
-                    if rma.module_access and rma.module_access.module_id == module.id
-                }
-            )
 
         token_payload = {
             "sub": str(user.id),
             "email": user.email,
+            "name": " ".join(
+                part for part in [user.first_name, user.last_name] if part
+            ) or user.email,
             "tenant_id": grant_tenant_id,
+            "tenant_name": user.tenant.tenant_name if user.tenant else None,
             "module_id": module_id,
             "environment": environment_slug,
             "permissions": permissions,
+            "role_id": str(user.role.id) if user.role else None,
             "roles": [user.role.role_name] if user.role else []
         }
 

app/services/auth/subscription_entitlement_service.py

@@ -61,15 +61,56 @@ class SubscriptionEntitlementService:
         rows = query.all()
         return {row[0] for row in rows}
 
+    @staticmethod
+    def get_effective_module_access_codes(
+        db: Session,
+        user: User,
+        module_id: uuid.UUID,
+    ) -> Set[str]:
+        """Return permissions both purchased by the tenant and granted to the user role."""
+        if not user.role:
+            return set()
+
+        role_codes = {
+            row.module_access.access_code
+            for row in user.role.role_module_accesses
+            if row.module_access and row.module_access.module_id == module_id
+        }
+        tenant = SubscriptionEntitlementService.get_tenant(db, user.tenant_id)
+        if not tenant or not tenant.plan_id:
+            return role_codes
+
+        plan_codes = SubscriptionEntitlementService.get_plan_module_access_codes(
+            db, user.tenant_id, module_id
+        )
+        return role_codes & plan_codes
+
     @staticmethod
     def get_effective_access_codes(db: Session, user: User) -> Set[str]:
         role_codes = set()
+        active_module_ids = set()
+        if user.tenant_id:
+            from app.models.auth.tenant_module_model import TenantModule
+            active_tms = (
+                db.query(TenantModule.module_id)
+                .filter(
+                    TenantModule.tenant_id == user.tenant_id,
+                    TenantModule.is_active == True,
+                )
+                .all()
+            )
+            active_module_ids = {tm[0] for tm in active_tms}
+
         if user.role:
             role_codes = {
                 ra.access.access_code
                 for ra in user.role.role_accesses
                 if ra.access is not None
             }
+            for rma in user.role.role_module_accesses:
+                if rma.module_access is not None:
+                    if not user.tenant_id or rma.module_access.module_id in active_module_ids:
+                        role_codes.add(rma.module_access.access_code)
 
         plan_codes = SubscriptionEntitlementService.get_plan_access_codes(
             db, user.tenant_id

app/services/auth/subscription_plan_service.py

@@ -14,14 +14,123 @@ from app.models.auth.module_environment_model import ModuleEnvironment
 from app.schemas.auth.role_schema import RoleUpdate
 from app.services.auth.role_service import RoleService
 from app.services.auth.event_service import EventService
+from app.models.auth.module_model import Module
+from app.models.auth.module_access_model import ModuleAccess
 from app.schemas.auth.subscription_plan_schema import (
     SubscriptionPlanCreate, 
     SubscriptionPlanUpdate, 
     SubscriptionPlanPaginatedResponse,
-    SubscriptionPlanResponse
+    SubscriptionPlanResponse,
+    SubscriptionPlanDetailResponse,
+    PlanApplicationSummary,
+    PlanApplicationEnvironmentSummary,
 )
 
+ONBOARDING_MODULE_CODES = {"pim", "inventory", "fulfillment"}
+
 class SubscriptionPlanService:
+    @staticmethod
+    def resolve_plan_applications(
+        db: Session,
+        plan: SubscriptionPlan,
+        onboarding_only: bool = True,
+    ) -> List[PlanApplicationSummary]:
+        """
+        Directly resolves active applications linked to the plan via PlanModuleAccess.
+        Excludes legacy internal submodules and sanitizes environment configs (omitting secrets).
+        """
+        plan_module_accesses = db.query(PlanModuleAccess).filter(PlanModuleAccess.plan_id == plan.id).all()
+        distinct_module_ids = []
+        if plan_module_accesses:
+            access_ids = [pma.module_access_id for pma in plan_module_accesses]
+            module_accesses = db.query(ModuleAccess).filter(ModuleAccess.id.in_(access_ids)).all()
+            distinct_module_ids = list({ma.module_id for ma in module_accesses})
+
+        # Check if plan has explicit onboarding suite modules (PIM, Inventory, Fulfillment)
+        onboarding_mods_in_plan = []
+        if distinct_module_ids:
+            onboarding_mods_in_plan = db.query(Module).filter(
+                Module.id.in_(distinct_module_ids),
+                Module.status == "active",
+                Module.module_id.in_(ONBOARDING_MODULE_CODES)
+            ).all()
+
+        if not onboarding_mods_in_plan:
+            # Fail closed: plans without explicit module links entitle zero applications
+            return []
+
+        modules_query = db.query(Module).filter(
+            Module.id.in_([m.id for m in onboarding_mods_in_plan]),
+            Module.status == "active"
+        )
+
+        modules = modules_query.order_by(Module.display_order, Module.module_name).all()
+
+        app_summaries = []
+        for mod in modules:
+            if onboarding_only and mod.module_id not in ONBOARDING_MODULE_CODES:
+                continue
+
+            envs = (
+                db.query(ModuleEnvironment)
+                .filter(ModuleEnvironment.module_id == mod.id, ModuleEnvironment.is_active == True)
+                .order_by(ModuleEnvironment.is_default.desc(), ModuleEnvironment.slug.asc())
+                .all()
+            )
+
+            env_summaries = [
+                PlanApplicationEnvironmentSummary(
+                    id=e.id,
+                    slug=e.slug,
+                    is_default=e.is_default,
+                    is_active=e.is_active,
+                    frontend_base_url=e.frontend_base_url,
+                    sso_entry_path=e.sso_entry_path or "/sso/callback",
+                    backend_base_url=e.backend_base_url,
+                )
+                for e in envs
+            ]
+
+            app_summaries.append(
+                PlanApplicationSummary(
+                    module_id=mod.id,
+                    module_code=mod.module_id,
+                    module_name=mod.module_name,
+                    description=mod.description,
+                    icon_url=mod.icon_url,
+                    display_order=mod.display_order or 0,
+                    environments=env_summaries,
+                )
+            )
+
+        return app_summaries
+
+    @staticmethod
+    def serialize_plan_response(
+        db: Session,
+        plan: SubscriptionPlan,
+        detail: bool = False,
+    ) -> SubscriptionPlanResponse | SubscriptionPlanDetailResponse:
+        apps = SubscriptionPlanService.resolve_plan_applications(db, plan, onboarding_only=True)
+        base_data = {
+            "id": plan.id,
+            "name": plan.name,
+            "description": plan.description,
+            "price": float(plan.price) if plan.price is not None else None,
+            "duration_days": plan.duration_days,
+            "max_users_allowed": plan.max_users_allowed,
+            "is_public": plan.is_public,
+            "status": plan.status,
+            "created_at": plan.created_at,
+            "updated_at": plan.updated_at,
+            "applications": apps,
+        }
+        if detail:
+            base_data["access_ids"] = [pa.access_id for pa in plan.plan_accesses]
+            base_data["module_access_ids"] = [pma.module_access_id for pma in plan.plan_module_accesses]
+            return SubscriptionPlanDetailResponse(**base_data)
+        return SubscriptionPlanResponse(**base_data)
+
     @staticmethod
     def _resolve_environment_slug(
         db: Session,
@@ -213,7 +322,8 @@ class SubscriptionPlanService:
             query = query.filter(SubscriptionPlan.is_public == is_public)
         if status is not None:
             query = query.filter(SubscriptionPlan.status == status)
-        return query.all()
+        plans = query.all()
+        return [SubscriptionPlanService.serialize_plan_response(db, p) for p in plans]
 
     @staticmethod
     def get_paginated_plans(
@@ -263,13 +373,14 @@ class SubscriptionPlanService:
         total_pages = (total + page_size - 1) // page_size if total > 0 else 0
         
         return SubscriptionPlanPaginatedResponse(
-            items=[SubscriptionPlanResponse.from_orm(p) for p in plans],
+            items=[SubscriptionPlanService.serialize_plan_response(db, p) for p in plans],
             total=total,
             page=page,
             page_size=page_size,
             total_pages=total_pages
         )
 
+
     @staticmethod
     def delete_plan(db: Session, plan_id: uuid.UUID):
         plan = SubscriptionPlanService.get_plan(db, plan_id)

app/services/auth/tenant_service.py

@@ -10,9 +10,13 @@ from app.models.auth.plan_module_access_model import PlanModuleAccess
 from app.models.auth.module_access_model import ModuleAccess
 from app.models.auth.role_model import Role
 from app.models.auth.role_module_access_model import RoleModuleAccess
+from app.models.auth.user_model import User
+from app.models.auth.module_model import Module
+from app.models.auth.module_environment_model import ModuleEnvironment
 from app.schemas.auth.tenant_schema import TenantCreate, TenantUpdate, TenantPaginatedResponse, TenantResponse
 from app.schemas.auth.role_schema import RoleCreate, RoleUpdate
 from app.services.auth.role_service import RoleService
+from app.config.security import security
 import uuid
 from typing import List, Optional
 from app.services.auth.event_service import EventService   
@@ -20,6 +24,9 @@ import logging
 
 logger = logging.getLogger(__name__)
 
+ONBOARDING_MODULE_CODES = {"pim", "inventory", "fulfillment"}
+
+
 class TenantService:
     STATUS_ACTIVE = "ACTIVE"
     STATUS_INACTIVE = "INACTIVE"
@@ -79,7 +86,8 @@ class TenantService:
         return changed
 
     @staticmethod
-    def create_tenant(db: Session, tenant_data: TenantCreate) -> Tenant:
+    def create_tenant(db: Session, tenant_data: TenantCreate, commit: bool = True) -> Tenant:
+        # Pre-validation 1: Name and Domain Uniqueness
         existing = db.query(Tenant).filter(Tenant.tenant_name == tenant_data.tenant_name).first()
         if existing:
             raise HTTPException(
@@ -94,13 +102,127 @@ class TenantService:
                 detail="Tenant domain already exists"
             )
             
+        # Pre-validation 2: Plan existence
         plan = db.query(SubscriptionPlan).filter(SubscriptionPlan.id == tenant_data.plan_id).first()
         if not plan:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail="Invalid plan_id provided"
             )
+
+        # Pre-validation 3: Owner Credentials Validation (Mandatory)
+        if not tenant_data.owner:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Tenant owner information is required"
+            )
+
+        owner_email = str(tenant_data.owner.email).strip().lower()
+        # Email uniqueness check across system
+        existing_user = db.query(User).filter(User.email == owner_email).first()
+        if existing_user:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Owner email '{owner_email}' is already registered"
+            )
+        # Password strength check
+        if not security.validate_password_strength(tenant_data.owner.password):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Owner password is too weak. Must be at least 8 characters with uppercase, lowercase, number, and special character."
+            )
+
+        # Pre-validation 4: Resolve Plan Modules and Validate Selection (Mandatory non-empty)
+        if not tenant_data.selected_module_ids:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="At least one application must be selected"
+            )
+
+        # Check duplicate module IDs
+        if len(tenant_data.selected_module_ids) != len(set(tenant_data.selected_module_ids)):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Duplicate module IDs in selected_module_ids"
+            )
+
+        plan_module_accesses = db.query(PlanModuleAccess).filter(PlanModuleAccess.plan_id == plan.id).all()
+        plan_mod_access_ids = [pma.module_access_id for pma in plan_module_accesses]
         
+        plan_distinct_module_ids = set()
+        if plan_mod_access_ids:
+            modules_query = db.query(ModuleAccess.module_id).filter(ModuleAccess.id.in_(plan_mod_access_ids)).distinct().all()
+            plan_distinct_module_ids = {m[0] for m in modules_query}
+
+        # Check if plan has explicit onboarding suite modules (PIM, Inventory, Fulfillment)
+        onboarding_mods_in_plan = []
+        if plan_distinct_module_ids:
+            onboarding_mods_in_plan = db.query(Module).filter(
+                Module.id.in_(plan_distinct_module_ids),
+                Module.status == "active",
+                Module.module_id.in_(ONBOARDING_MODULE_CODES)
+            ).all()
+
+        active_onboarding_modules = onboarding_mods_in_plan
+        allowed_module_ids = {m.id for m in active_onboarding_modules}
+
+        selected_set = set(tenant_data.selected_module_ids)
+        invalid_modules = selected_set - allowed_module_ids
+        if invalid_modules:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Selected application is not included in the selected plan or is inactive"
+            )
+        effective_selected_module_ids = [m.id for m in active_onboarding_modules if m.id in selected_set]
+
+        # Pre-validation 5: Validate Module Environments (Explicit 1-to-1 required, reject duplicates)
+        module_env_map = {}
+        seen_env_modules = set()
+        if tenant_data.module_environments:
+            for me in tenant_data.module_environments:
+                if me.module_id in seen_env_modules:
+                    raise HTTPException(
+                        status_code=status.HTTP_400_BAD_REQUEST,
+                        detail=f"Duplicate environment assignment for module '{me.module_id}'"
+                    )
+                seen_env_modules.add(me.module_id)
+
+                if me.module_id not in selected_set:
+                    raise HTTPException(
+                        status_code=status.HTTP_400_BAD_REQUEST,
+                        detail=f"Environment assignment specified for module '{me.module_id}' which is not in selected applications"
+                    )
+                # Verify environment exists, belongs to this module, and is active
+                env_exists = db.query(ModuleEnvironment).filter(
+                    ModuleEnvironment.module_id == me.module_id,
+                    ModuleEnvironment.slug == me.environment_slug,
+                    ModuleEnvironment.is_active == True
+                ).first()
+                if not env_exists:
+                    raise HTTPException(
+                        status_code=status.HTTP_400_BAD_REQUEST,
+                        detail=f"Environment '{me.environment_slug}' does not exist or is inactive for module '{me.module_id}'"
+                    )
+                module_env_map[me.module_id] = me.environment_slug
+
+        # Require an explicit environment assignment for EVERY selected application
+        resolved_event_targets = []
+        for mod_id in effective_selected_module_ids:
+            env_slug = module_env_map.get(mod_id)
+            if not env_slug:
+                mod_obj = db.query(Module).filter(Module.id == mod_id).first()
+                mod_name = mod_obj.module_name if mod_obj else str(mod_id)
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=f"Explicit environment assignment is required for selected application '{mod_name}'"
+                )
+
+            resolved_event_targets.append({
+                "module_id": str(mod_id),
+                "environment_slug": env_slug
+            })
+
+
         provisioning_id = str(uuid.uuid4())
         
         try:
@@ -122,22 +244,9 @@ class TenantService:
             db.add(tenant)
             db.flush()
             
-            plan_module_accesses = db.query(PlanModuleAccess).filter(PlanModuleAccess.plan_id == plan.id).all()
+            # 2. Create TenantModule assignments for selected modules only
-            mod_access_ids = [pma.module_access_id for pma in plan_module_accesses]
+            for mod_id in effective_selected_module_ids:
-            
+                env_slug = module_env_map[mod_id]
-            distinct_module_ids = []
-            if mod_access_ids:
-                modules_query = db.query(ModuleAccess.module_id).filter(ModuleAccess.id.in_(mod_access_ids)).distinct().all()
-                distinct_module_ids = [m[0] for m in modules_query]
-                
-            module_env_map = {}
-            if tenant_data.module_environments:
-                for me in tenant_data.module_environments:
-                    module_env_map[me.module_id] = me.environment_slug
-
-            event_targets = []
-            for mod_id in distinct_module_ids:
-                env_slug = module_env_map.get(mod_id, tenant_data.default_environment_slug)
                 tm = TenantModule(
                     tenant_id=tenant.id,
                     module_id=mod_id,
@@ -146,24 +255,52 @@ class TenantService:
                 )
                 db.add(tm)
 
-                event_targets.append({
-                    "module_id": str(mod_id),
-                    "environment_slug": env_slug
-                })
-
             db.flush()
 
+            # 3. Create Primary Admin Role
+            # SaaS accesses from plan
             plan_saas_accesses = db.query(PlanAccess).filter(PlanAccess.plan_id == plan.id).all()
-            all_access_ids = [a.access_id for a in plan_saas_accesses] + mod_access_ids
+            saas_access_ids = [a.access_id for a in plan_saas_accesses]
+
+            # Module launch accesses ONLY for selected modules
+
+            selected_module_accesses = db.query(ModuleAccess.id).filter(
+                ModuleAccess.id.in_(plan_mod_access_ids),
+                ModuleAccess.module_id.in_(effective_selected_module_ids)
+            ).all()
+            selected_mod_access_ids = [ma[0] for ma in selected_module_accesses]
 
             role_create_data = RoleCreate(
                 role_name="Primary Admin",
                 tenant_id=tenant.id,
                 is_default=True,
-                access_ids=all_access_ids
+                access_ids=saas_access_ids + selected_mod_access_ids
             )
-            role = RoleService.create_role(db, role_create_data, emit_events=False)
+            role = RoleService.create_role(
+                db,
+                role_create_data,
+                emit_events=False,
+                commit=False,
+            )
+            db.flush()
 
+            # 4. Create Owner User if requested
+            owner_user = None
+            if tenant_data.owner:
+                owner_user = User(
+                    email=tenant_data.owner.email.strip().lower(),
+                    password=security.hash_password(tenant_data.owner.password),
+                    first_name=tenant_data.owner.first_name.strip(),
+                    last_name=tenant_data.owner.last_name.strip(),
+                    phone_number=tenant_data.owner.phone_number,
+                    status="active",
+                    tenant_id=tenant.id,
+                    role_id=role.id
+                )
+                db.add(owner_user)
+                db.flush()
+
+            # 5. Build Outbox Events (durable log rows)
             role_module_perms = (
                 db.query(RoleModuleAccess, ModuleAccess)
                 .join(ModuleAccess, RoleModuleAccess.module_access_id == ModuleAccess.id)
@@ -181,7 +318,7 @@ class TenantService:
                     module_map[mid].append(ma.access_code)
 
                 role_targets = []
-                for target in event_targets:
+                for target in resolved_event_targets:
                     mid = target["module_id"]
                     if mid in module_map:
                         role_targets.append({
@@ -197,39 +334,79 @@ class TenantService:
                         "payload": {
                             "role_id": str(role.id),
                             "role_name": role.role_name,
+                            "role_code": "TENANT_OWNER",
+                            "is_owner": True,
                             "tenant_id": str(tenant.id),
                             "provisioning_id": str(uuid.uuid4()),
                             "targets": role_targets
                         }
                     }
 
-            if event_targets:
+            emitted_event_ids = []
-                logger.info(f"Creating tenant {tenant.tenant_name}. Processing {len(event_targets)} event targets.")
+            if resolved_event_targets:
+                logger.info(f"Emitting TENANT_PROVISION_REQUESTED for tenant '{tenant.tenant_name}' across {len(resolved_event_targets)} targets.")
 
                 payload = {
                     "tenant_id": str(tenant.id),
+                    "canonical_tenant_id": str(tenant.id),
                     "tenant_name": tenant.tenant_name,
                     "tenant_domain": tenant.tenant_domain,
                     "tenant_logo_url": tenant.tenant_logo_url,
                     "max_users_allowed": plan.max_users_allowed,
                     "provisioning_id": provisioning_id,
-                    "targets": event_targets
+                    "targets": resolved_event_targets
                 }
 
-                EventService.emit_event(
+                tenant_event_id = EventService.emit_event(
                     db,
                     event_type="TENANT_PROVISION_REQUESTED",
                     payload=payload,
                     tenant_id=tenant.id,
                     follow_up_event=role_follow_up
                 )
-                logger.info(f"Event TENANT_PROVISION_REQUESTED emitted to outbox{' (with ROLE_PROVISION_REQUESTED follow-up)' if role_follow_up else ''}.")
+                if tenant_event_id:
+                    emitted_event_ids.append(tenant_event_id)
+
+            # If owner user created, emit USER_PROVISION_REQUESTED outbox event with canonical_user_id
+            if owner_user:
+                user_payload = {
+                    "user_id": str(owner_user.id),
+                    "canonical_user_id": str(owner_user.id),
+                    "email": owner_user.email,
+                    "first_name": owner_user.first_name,
+                    "last_name": owner_user.last_name,
+                    "phone_number": owner_user.phone_number,
+                    "tenant_id": str(tenant.id),
+                    "canonical_tenant_id": str(tenant.id),
+                    "role_id": str(role.id),
+                    "role_name": role.role_name,
+                    "role_code": "TENANT_OWNER",
+                    "is_owner": True,
+                    "status": owner_user.status,
+                    "targets": role_targets if role_module_perms else []
+                }
+                user_event_id = EventService.emit_event(
+                    db,
+                    event_type="USER_PROVISION_REQUESTED",
+                    payload=user_payload,
+                    tenant_id=tenant.id
+                )
+                if user_event_id:
+                    emitted_event_ids.append(user_event_id)
+
+            tenant.pending_event_ids = emitted_event_ids
+
+            # Controlled final commit or flush
+            if commit:
+                db.commit()
+                db.refresh(tenant)
+                # Best-effort wake-up notification to Redis after successful commit
+                EventService.enqueue_wakeups(emitted_event_ids)
+            else:
+                db.flush()
 
-            db.commit()
-            db.refresh(tenant)
-            
             return tenant
-            
+
         except Exception as e:
             db.rollback()
             raise e

app/services/system/audit_log_service.py

@@ -23,12 +23,15 @@ class AuditLogService:
         ip_address: Optional[str] = None,
         old_values: Optional[Dict[str, Any]] = None,
         new_values: Optional[Dict[str, Any]] = None,
-    ) -> None:
+        commit: bool = True,
+        strict: bool = False,
+    ) -> Optional[AuditLog]:
         """
         Write an audit log entry.
  
-        Failures are silently swallowed and logged — they must NEVER break
+        If commit=True, commits the session. If commit=False, flushes to the active transaction.
-        the main operation that triggered them.
+        If strict=True, exceptions are re-raised to the caller.
+        If strict=False, failures are swallowed and logged.
         """
         try:
             entry = AuditLog(
@@ -46,7 +49,15 @@ class AuditLogService:
                 new_values=new_values,
             )
             db.add(entry)
-            db.commit()
+            if commit:
+                db.commit()
+            else:
+                db.flush()
+            return entry
         except Exception as exc:
-            db.rollback()
+            if strict:
-            logger.error("AuditLogService.log failed: %s", exc, exc_info=True)
+                raise
+            if commit:
+                db.rollback()
+            logger.error("AuditLogService.log failed: %s", exc, exc_info=True)
+            return None

requirements.txt

@@ -7,8 +7,9 @@ alembic==1.17.2
 pydantic-settings==2.12.0
 bcrypt>=4.0.1
 pyjwt>=2.8.0
+cryptography>=42.0.0
 email-validator>=2.1.0
 redis==7.1.0
 requests==2.32.5
 typer==0.21.1
-httpx==0.28.1
+httpx==0.28.1

scripts/seed_local_ecosystem.py

@@ -0,0 +1,414 @@
+"""
+Idempotent localhost-only seeder for SaaS ecosystem applications:
+- PIM
+- Inventory
+- Fulfillment & Logistics
+
+Usage:
+    python scripts/seed_local_ecosystem.py
+"""
+
+import sys
+import os
+import uuid
+import logging
+from urllib.parse import urlparse
+from pathlib import Path
+from typing import Dict, Any, List, Optional
+from sqlalchemy.engine import make_url
+from sqlalchemy.orm import Session
+
+# Setup path
+backend_dir = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(backend_dir))
+
+from app.config.settings import settings
+from app.config.database import SessionLocal
+from app.models.auth.module_model import Module
+from app.models.auth.module_environment_model import ModuleEnvironment
+from app.models.auth.module_access_model import ModuleAccess
+from app.models.auth.subscription_plan_model import SubscriptionPlan
+from app.models.auth.plan_module_access_model import PlanModuleAccess
+
+logger = logging.getLogger(__name__)
+
+# Constants for the 3 sellable applications
+ECOSYSTEM_MODULES = [
+    {
+        "module_id": "pim",
+        "module_name": "Product Information Management",
+        "description": "Enterprise Product Information Management (PIM)",
+        "display_order": 1,
+        "default_frontend_url": "http://127.0.0.1:5173",
+        "default_backend_url": "http://127.0.0.1:5002",
+        "trust_secret_env": "MODULE_TRUST_SECRET_PIM",
+        "access_codes": [
+            ("pim.view", "PIM", "View Products"),
+            ("pim.manage", "PIM", "Manage Products"),
+        ],
+    },
+    {
+        "module_id": "inventory",
+        "module_name": "Inventory",
+        "description": "Warehouse and Inventory Management System",
+        "display_order": 2,
+        "default_frontend_url": "http://127.0.0.1:5174",
+        "default_backend_url": "http://127.0.0.1:12001",
+        "trust_secret_env": "MODULE_TRUST_SECRET_INVENTORY",
+        "access_codes": [
+            ("inventory.view", "Inventory", "View Inventory"),
+            ("inventory.manage", "Inventory", "Manage Inventory"),
+        ],
+    },
+    {
+        "module_id": "fulfillment",
+        "module_name": "Fulfillment & Logistics",
+        "description": "Fulfillment, Logistics and Dispatch Management",
+        "display_order": 3,
+        "default_frontend_url": "http://127.0.0.1:5175",
+        "default_backend_url": "http://127.0.0.1:8080",
+        "trust_secret_env": "MODULE_TRUST_SECRET_FULFILLMENT",
+        "access_codes": [
+            ("fulfillment.view", "Fulfillment", "View Orders & Shipments"),
+            ("fulfillment.manage", "Fulfillment", "Manage Fulfillment"),
+        ],
+    },
+]
+
+TEST_PLAN = {
+    "name": "Local Ecosystem Test",
+    "description": "Test subscription plan entitling PIM, Inventory, and Fulfillment",
+    "price": 0.0,
+    "duration_days": 365,
+    "max_users_allowed": 50,
+    "is_public": True,
+    "status": "active",
+}
+
+
+def is_local_hostname(hostname: Optional[str]) -> bool:
+    """Validate that hostname is strictly a local loopback."""
+    if not hostname:
+        return True  # in-memory or socket
+    clean_host = hostname.lower().strip("[]")
+    return clean_host in ["127.0.0.1", "localhost", "::1"]
+
+
+def validate_http_url(url_str: str, name: str) -> str:
+    """Ensure HTTP URL is valid and strictly targets local loopback in local seeder."""
+    parsed = urlparse(url_str)
+    if parsed.scheme not in ["http", "https"]:
+        raise ValueError(f"Invalid URL scheme for {name}: '{url_str}'")
+    if not is_local_hostname(parsed.hostname):
+        raise ValueError(
+            f"Non-local host in {name} '{url_str}'. Local seeder requires localhost/127.0.0.1/::1."
+        )
+    return url_str
+
+
+def verify_safety_gate(db_url: Optional[str] = None, app_env: Optional[str] = None) -> None:
+    """Refuse to run unless on local/testing environment and strictly local database target."""
+    effective_env = app_env or getattr(settings, "APP_ENV", "local")
+    effective_url = db_url or getattr(settings, "DATABASE_URL", "")
+
+    if effective_env not in ["local", "testing"]:
+        raise RuntimeError(f"Safety gate failed: APP_ENV '{effective_env}' is not local or testing.")
+
+    try:
+        url_obj = make_url(effective_url)
+        backend_name = url_obj.get_backend_name()
+    except Exception as e:
+        raise RuntimeError(f"Safety gate failed: Malformed DATABASE_URL: {e}")
+
+    if backend_name not in ["sqlite", "postgresql", "postgres"]:
+        raise RuntimeError(f"Safety gate failed: Unsupported database dialect '{backend_name}'.")
+
+    if backend_name == "sqlite":
+        return
+
+    host = url_obj.host
+    if not is_local_hostname(host):
+        raise RuntimeError(
+            f"Safety gate failed: Non-local database host '{host}'. Only localhost/127.0.0.1/::1 allowed."
+        )
+    db_name = url_obj.database or ""
+    if db_name not in ["saas_local", "saas_test"] and "test" not in db_name:
+        raise RuntimeError(
+            f"Safety gate failed: Database name '{db_name}' is not saas_local or an isolated test DB."
+        )
+
+
+def validate_and_extract_secrets() -> Dict[str, str]:
+    """
+    Validate that all three module trust secrets exist in explicit process environment variables.
+    Fails immediately without inserting partial rows if any secret is missing.
+    """
+    secrets = {}
+    missing = []
+
+    for mod_info in ECOSYSTEM_MODULES:
+        env_var = mod_info["trust_secret_env"]
+        val = os.getenv(env_var)
+        if not val or not val.strip():
+            missing.append(env_var)
+        else:
+            secrets[mod_info["module_id"]] = val.strip()
+
+    if missing:
+        raise ValueError(
+            f"Missing required module trust secret environment variables: {', '.join(missing)}. "
+            f"Each module requires its explicit variable."
+        )
+
+    return secrets
+
+
+def seed_ecosystem_modules(db: Session) -> List[Module]:
+    """Seed or update the 3 top-level sellable modules."""
+    seeded_modules = []
+
+    for mod_info in ECOSYSTEM_MODULES:
+        mod_code = mod_info["module_id"]
+        module = db.query(Module).filter(Module.module_id == mod_code).first()
+
+        if not module:
+            module = Module(
+                id=uuid.uuid4(),
+                module_id=mod_code,
+                module_name=mod_info["module_name"],
+                description=mod_info["description"],
+                status="active",
+                display_order=mod_info["display_order"],
+            )
+            db.add(module)
+            db.flush()
+        else:
+            module.module_name = mod_info["module_name"]
+            module.description = mod_info["description"]
+            module.display_order = mod_info["display_order"]
+            module.status = "active"
+
+        seeded_modules.append(module)
+
+    return seeded_modules
+
+
+def seed_module_environments(
+    db: Session,
+    modules: List[Module],
+    secrets_map: Dict[str, str],
+) -> List[ModuleEnvironment]:
+    """Seed or update the default local environment for each module with correct SSO routes."""
+    seeded_envs = []
+    mod_lookup = {m.module_id: m for m in modules}
+
+    for mod_info in ECOSYSTEM_MODULES:
+        mod_code = mod_info["module_id"]
+        module = mod_lookup.get(mod_code)
+        if not module:
+            raise ValueError(f"Module {mod_code} was not found in seeded modules")
+
+        trust_secret = secrets_map.get(mod_code)
+        if not trust_secret:
+            raise ValueError(f"Missing validated trust secret for module '{mod_code}'")
+
+        raw_frontend = os.getenv(f"MODULE_URL_{mod_code.upper()}", mod_info["default_frontend_url"])
+        raw_backend = os.getenv(f"MODULE_BACKEND_URL_{mod_code.upper()}", mod_info["default_backend_url"])
+
+        frontend_url = validate_http_url(raw_frontend, f"{mod_code} frontend URL")
+        backend_url = validate_http_url(raw_backend, f"{mod_code} backend URL")
+
+        # Check existing environment by unique (module_id, slug)
+        env = db.query(ModuleEnvironment).filter(
+            ModuleEnvironment.module_id == module.id,
+            ModuleEnvironment.slug == "local",
+        ).first()
+
+        # Check for multiple active default environments for the same module
+        other_defaults = db.query(ModuleEnvironment).filter(
+            ModuleEnvironment.module_id == module.id,
+            ModuleEnvironment.is_default == True,
+            ModuleEnvironment.slug != "local",
+        ).all()
+        if other_defaults:
+            raise ValueError(
+                f"Conflict: Module '{mod_code}' has multiple conflicting default environments configured."
+            )
+
+        if not env:
+            env = ModuleEnvironment(
+                id=uuid.uuid4(),
+                module_id=module.id,
+                slug="local",
+                frontend_base_url=frontend_url,
+                sso_entry_path="/sso/callback",
+                backend_base_url=backend_url,
+                sso_exchange_endpoint="/sso/exchange",
+                permission_sync_endpoint="/internal/permissions/sync",
+                provisioning_endpoint="/internal/tenants/provision",
+                trust_type="hmac",
+                trust_credentials={"hmac_secret": trust_secret},
+                is_default=True,
+                is_active=True,
+            )
+            db.add(env)
+            db.flush()
+        else:
+            # Check ownership integrity
+            if env.module_id != module.id:
+                raise ValueError(f"Integrity violation: environment {env.id} ownership mismatch.")
+
+            env.frontend_base_url = frontend_url
+            env.sso_entry_path = "/sso/callback"
+            env.backend_base_url = backend_url
+            env.sso_exchange_endpoint = "/sso/exchange"
+            env.permission_sync_endpoint = "/internal/permissions/sync"
+            env.provisioning_endpoint = "/internal/tenants/provision"
+            env.trust_type = "hmac"
+            env.trust_credentials = {"hmac_secret": trust_secret}
+            env.is_default = True
+            env.is_active = True
+
+        seeded_envs.append(env)
+
+    return seeded_envs
+
+
+def seed_module_accesses(db: Session, modules: List[Module]) -> List[ModuleAccess]:
+    """Seed baseline module accesses for each application."""
+    seeded_accesses = []
+    mod_lookup = {m.module_id: m for m in modules}
+
+    for mod_info in ECOSYSTEM_MODULES:
+        mod_code = mod_info["module_id"]
+        module = mod_lookup.get(mod_code)
+        if not module:
+            continue
+
+        for code, category, name in mod_info["access_codes"]:
+            access = db.query(ModuleAccess).filter(
+                ModuleAccess.module_id == module.id,
+                ModuleAccess.access_code == code,
+            ).first()
+
+            if not access:
+                access = ModuleAccess(
+                    id=uuid.uuid4(),
+                    module_id=module.id,
+                    access_code=code,
+                    category=category,
+                    name=name,
+                )
+                db.add(access)
+                db.flush()
+            else:
+                access.category = category
+                access.name = name
+
+            seeded_accesses.append(access)
+
+    return seeded_accesses
+
+
+def seed_test_subscription_plan(db: Session, module_accesses: List[ModuleAccess]) -> SubscriptionPlan:
+    """Seed the test plan and reconcile all fields completely on every run."""
+    plan = db.query(SubscriptionPlan).filter(
+        SubscriptionPlan.name == TEST_PLAN["name"]
+    ).first()
+
+    if not plan:
+        plan = SubscriptionPlan(
+            id=uuid.uuid4(),
+            name=TEST_PLAN["name"],
+            description=TEST_PLAN["description"],
+            price=TEST_PLAN["price"],
+            duration_days=TEST_PLAN["duration_days"],
+            max_users_allowed=TEST_PLAN["max_users_allowed"],
+            is_public=TEST_PLAN["is_public"],
+            status=TEST_PLAN["status"],
+        )
+        db.add(plan)
+        db.flush()
+    else:
+        # Full field reconciliation on reruns
+        plan.description = TEST_PLAN["description"]
+        plan.price = TEST_PLAN["price"]
+        plan.duration_days = TEST_PLAN["duration_days"]
+        plan.max_users_allowed = TEST_PLAN["max_users_allowed"]
+        plan.is_public = TEST_PLAN["is_public"]
+        plan.status = TEST_PLAN["status"]
+
+    # Link module accesses to plan via PlanModuleAccess
+    existing_pmas = db.query(PlanModuleAccess).filter(
+        PlanModuleAccess.plan_id == plan.id
+    ).all()
+    existing_access_ids = {pma.module_access_id for pma in existing_pmas}
+
+    for ma in module_accesses:
+        if ma.id not in existing_access_ids:
+            pma = PlanModuleAccess(
+                id=uuid.uuid4(),
+                plan_id=plan.id,
+                module_access_id=ma.id,
+            )
+            db.add(pma)
+
+    db.flush()
+    return plan
+
+
+def run_seeder(db: Session, db_url: Optional[str] = None) -> Dict[str, Any]:
+    """Execute complete ecosystem seeding within a single transaction."""
+    # If db_url not provided, inspect from session bind engine URL if available
+    effective_url = db_url
+    if not effective_url and db.bind:
+        try:
+            effective_url = str(db.bind.url)
+        except Exception:
+            pass
+
+    verify_safety_gate(db_url=effective_url)
+
+    # Pre-validate all secrets before any DB mutations
+    secrets_map = validate_and_extract_secrets()
+
+    try:
+        modules = seed_ecosystem_modules(db)
+        envs = seed_module_environments(db, modules, secrets_map)
+        accesses = seed_module_accesses(db, modules)
+        plan = seed_test_subscription_plan(db, accesses)
+
+        db.commit()
+
+        return {
+            "modules_count": len(modules),
+            "environments_count": len(envs),
+            "module_accesses_count": len(accesses),
+            "plan_name": plan.name,
+            "status": "success",
+        }
+    except Exception as e:
+        db.rollback()
+        logger.error(f"Ecosystem seeding failed: {e}")
+        raise
+
+
+def main():
+    print("=" * 60)
+    print("SaaS Local Ecosystem Seeder (PIM, Inventory, Fulfillment)")
+    print("=" * 60)
+
+    db = SessionLocal()
+    try:
+        result = run_seeder(db)
+        print(f"✓ Seed completed successfully!")
+        print(f"  Modules seeded: {result['modules_count']}")
+        print(f"  Environments seeded: {result['environments_count']}")
+        print(f"  Plan seeded: {result['plan_name']}")
+        print("=" * 60)
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()

tests/test_local_ecosystem_seed.py

@@ -0,0 +1,251 @@
+import os
+import sys
+import uuid
+import pytest
+from pathlib import Path
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+# Setup path to import app modules
+backend_dir = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(backend_dir))
+
+from app.config.database import Base
+from app.models.auth.module_model import Module
+from app.models.auth.module_environment_model import ModuleEnvironment
+from app.models.auth.module_access_model import ModuleAccess
+from app.models.auth.subscription_plan_model import SubscriptionPlan
+from app.models.auth.plan_module_access_model import PlanModuleAccess
+from app.services.auth.subscription_plan_service import SubscriptionPlanService
+import scripts.seed_local_ecosystem as seeder
+
+
+@pytest.fixture
+def isolated_db():
+    """Create a completely isolated in-memory SQLite database session for target ecosystem models."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    target_tables = [
+        Module.__table__,
+        ModuleEnvironment.__table__,
+        ModuleAccess.__table__,
+        SubscriptionPlan.__table__,
+        PlanModuleAccess.__table__,
+    ]
+    Base.metadata.create_all(bind=engine, tables=target_tables)
+    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    session = TestingSessionLocal()
+    try:
+        yield session
+    finally:
+        session.close()
+        Base.metadata.drop_all(bind=engine, tables=target_tables)
+
+
+def test_first_run_creates_exact_ecosystem_records(isolated_db, monkeypatch):
+    """1. First run creates exactly 3 modules, 3 local environments, 6 accesses, 1 plan, 6 plan-access links, and exact SSO routes."""
+    monkeypatch.setenv("APP_ENV", "testing")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_PIM", "test-pim-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_INVENTORY", "test-inv-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_FULFILLMENT", "test-fl-secret")
+
+    # Run seeder with pre-validation
+    secrets_map = seeder.validate_and_extract_secrets()
+    modules = seeder.seed_ecosystem_modules(isolated_db)
+    envs = seeder.seed_module_environments(isolated_db, modules, secrets_map)
+    accesses = seeder.seed_module_accesses(isolated_db, modules)
+    plan = seeder.seed_test_subscription_plan(isolated_db, accesses)
+    isolated_db.commit()
+
+    # Assert exact counts
+    assert isolated_db.query(Module).count() == 3
+    assert isolated_db.query(ModuleEnvironment).count() == 3
+    assert isolated_db.query(ModuleAccess).count() == 6
+    assert isolated_db.query(SubscriptionPlan).count() == 1
+    assert isolated_db.query(PlanModuleAccess).count() == 6
+
+    # Assert exact module codes
+    mod_codes = {m.module_id for m in isolated_db.query(Module).all()}
+    assert mod_codes == {"pim", "inventory", "fulfillment"}
+
+    # Assert all 3 environments have correct SSO entry & exchange endpoints
+    for env in isolated_db.query(ModuleEnvironment).all():
+        assert env.slug == "local"
+        assert env.sso_entry_path == "/sso/callback"
+        assert env.sso_exchange_endpoint == "/sso/exchange"
+        assert env.trust_type == "hmac"
+        assert "hmac_secret" in env.trust_credentials
+        assert env.is_default is True
+        assert env.is_active is True
+
+    # Assert plan resolution through real SubscriptionPlanService
+    db_plan = isolated_db.query(SubscriptionPlan).first()
+    entitled_apps = SubscriptionPlanService.resolve_plan_applications(isolated_db, db_plan)
+    entitled_codes = {app.module_code for app in entitled_apps}
+    assert entitled_codes == {"pim", "inventory", "fulfillment"}
+
+
+def test_second_run_is_fully_idempotent_and_reconciles_plan(isolated_db, monkeypatch):
+    """2. Second run preserves IDs/counts and reconciles modified plan metadata."""
+    monkeypatch.setenv("APP_ENV", "testing")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_PIM", "test-pim-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_INVENTORY", "test-inv-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_FULFILLMENT", "test-fl-secret")
+
+    # Pass 1
+    secrets_map_1 = seeder.validate_and_extract_secrets()
+    modules_1 = seeder.seed_ecosystem_modules(isolated_db)
+    envs_1 = seeder.seed_module_environments(isolated_db, modules_1, secrets_map_1)
+    accesses_1 = seeder.seed_module_accesses(isolated_db, modules_1)
+    plan_1 = seeder.seed_test_subscription_plan(isolated_db, accesses_1)
+    isolated_db.commit()
+
+    mod_ids_1 = {m.module_id: m.id for m in modules_1}
+    env_ids_1 = {e.module_id: e.id for e in envs_1}
+    plan_id_1 = plan_1.id
+
+    # Mutate plan description in DB to verify reconciliation on second pass
+    plan_1.description = "Old dirty description"
+    plan_1.max_users_allowed = 5
+    isolated_db.commit()
+
+    # Pass 2
+    secrets_map_2 = seeder.validate_and_extract_secrets()
+    modules_2 = seeder.seed_ecosystem_modules(isolated_db)
+    envs_2 = seeder.seed_module_environments(isolated_db, modules_2, secrets_map_2)
+    accesses_2 = seeder.seed_module_accesses(isolated_db, modules_2)
+    plan_2 = seeder.seed_test_subscription_plan(isolated_db, accesses_2)
+    isolated_db.commit()
+
+    # Verify counts remain identical
+    assert isolated_db.query(Module).count() == 3
+    assert isolated_db.query(ModuleEnvironment).count() == 3
+    assert isolated_db.query(SubscriptionPlan).count() == 1
+    assert isolated_db.query(ModuleAccess).count() == 6
+    assert isolated_db.query(PlanModuleAccess).count() == 6
+
+    # Verify stable IDs
+    for m in modules_2:
+        assert m.id == mod_ids_1[m.module_id]
+    for e in envs_2:
+        assert e.id == env_ids_1[e.module_id]
+    assert plan_2.id == plan_id_1
+
+    # Verify plan reconciliation
+    assert plan_2.description == seeder.TEST_PLAN["description"]
+    assert plan_2.max_users_allowed == 50
+
+
+def test_missing_any_one_secret_fails_prevalidation_with_zero_records(isolated_db, monkeypatch):
+    """3. Missing any single module secret raises error before insertion, leaving 0 rows."""
+    monkeypatch.setenv("APP_ENV", "testing")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_PIM", "test-pim-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_INVENTORY", "test-inv-secret")
+    # Leave MODULE_TRUST_SECRET_FULFILLMENT unset
+    monkeypatch.delenv("MODULE_TRUST_SECRET_FULFILLMENT", raising=False)
+    monkeypatch.setenv("SAAS_TRUST_SECRET", "shared-fallback-which-must-be-ignored")
+
+    with pytest.raises(ValueError, match="MODULE_TRUST_SECRET_FULFILLMENT"):
+        seeder.run_seeder(isolated_db)
+
+    # Prove zero partial rows created
+    assert isolated_db.query(Module).count() == 0
+    assert isolated_db.query(ModuleEnvironment).count() == 0
+    assert isolated_db.query(SubscriptionPlan).count() == 0
+
+
+def test_safety_gate_rejects_malformed_and_remote_database_urls():
+    """4. Safety gate strictly rejects malformed or remote database URLs."""
+    # Malformed / Unsupported dialect URL
+    with pytest.raises(RuntimeError, match="Unsupported database dialect|Malformed DATABASE_URL"):
+        seeder.verify_safety_gate(db_url="not_a_valid_url://", app_env="local")
+
+    # Remote database host
+    with pytest.raises(RuntimeError, match="Non-local database host"):
+        seeder.verify_safety_gate(
+            db_url="postgresql://user:pass@106.51.105.22:5432/saas_local",
+            app_env="local",
+        )
+
+    # Disallowed DB name on localhost
+    with pytest.raises(RuntimeError, match="Database name 'saas_production' is not saas_local"):
+        seeder.verify_safety_gate(
+            db_url="postgresql://user:pass@127.0.0.1:5432/saas_production",
+            app_env="local",
+        )
+
+
+def test_url_validation_rejects_remote_frontend_backend_urls():
+    """5. Frontend/Backend URL validation strictly rejects non-local hosts."""
+    # Remote frontend URL
+    with pytest.raises(ValueError, match="Non-local host in PIM frontend URL"):
+        seeder.validate_http_url("http://app.maskantech.in/pim", "PIM frontend URL")
+
+    # Remote backend URL
+    with pytest.raises(ValueError, match="Non-local host in Inventory backend URL"):
+        seeder.validate_http_url("http://api.maskantech.in:12001", "Inventory backend URL")
+
+    # Invalid scheme
+    with pytest.raises(ValueError, match="Invalid URL scheme"):
+        seeder.validate_http_url("ftp://127.0.0.1:5173", "PIM frontend URL")
+
+    # Valid localhost URLs pass
+    assert seeder.validate_http_url("http://127.0.0.1:5173", "PIM") == "http://127.0.0.1:5173"
+    assert seeder.validate_http_url("http://localhost:8080", "FL") == "http://localhost:8080"
+
+
+def test_multiple_default_environments_conflict_rejected(isolated_db, monkeypatch):
+    """6. Reject multiple default environments for the same module."""
+    monkeypatch.setenv("APP_ENV", "testing")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_PIM", "test-pim-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_INVENTORY", "test-inv-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_FULFILLMENT", "test-fl-secret")
+
+    # Seed modules first
+    modules = seeder.seed_ecosystem_modules(isolated_db)
+    pim_mod = modules[0]
+
+    # Pre-insert a conflicting default environment for PIM with different slug
+    conflict_env = ModuleEnvironment(
+        id=uuid.uuid4(),
+        module_id=pim_mod.id,
+        slug="staging_default",
+        frontend_base_url="http://127.0.0.1:5173",
+        backend_base_url="http://127.0.0.1:5002",
+        trust_type="hmac",
+        trust_credentials={"hmac_secret": "secret"},
+        is_default=True,
+        is_active=True,
+    )
+    isolated_db.add(conflict_env)
+    isolated_db.commit()
+
+    secrets_map = seeder.validate_and_extract_secrets()
+    with pytest.raises(ValueError, match="multiple conflicting default environments"):
+        seeder.seed_module_environments(isolated_db, modules, secrets_map)
+
+
+def test_failure_after_flush_rolls_back_entire_run(isolated_db, monkeypatch):
+    """7. Failure during execution triggers full rollback leaving no partial changes."""
+    monkeypatch.setenv("APP_ENV", "testing")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_PIM", "test-pim-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_INVENTORY", "test-inv-secret")
+    monkeypatch.setenv("MODULE_TRUST_SECRET_FULFILLMENT", "test-fl-secret")
+
+    # Monkeypatch seed_test_subscription_plan to deliberately fail after modules are flushed
+    def broken_plan_seeder(db, accesses):
+        raise RuntimeError("Simulated mid-transaction failure")
+
+    monkeypatch.setattr(seeder, "seed_test_subscription_plan", broken_plan_seeder)
+
+    with pytest.raises(RuntimeError, match="Simulated mid-transaction failure"):
+        seeder.run_seeder(isolated_db)
+
+    # Assert rollback: zero modules exist
+    assert isolated_db.query(Module).count() == 0
+    assert isolated_db.query(ModuleEnvironment).count() == 0
+    assert isolated_db.query(SubscriptionPlan).count() == 0

tests/test_saas_settings_precedence.py

@@ -0,0 +1,314 @@
+import os
+import sys
+import tempfile
+import textwrap
+import subprocess
+from pathlib import Path
+import pytest
+
+# Complete set of synthetic required process variables for saas_backend
+SYNTHETIC_PROCESS_ENV = {
+    "PROJECT_NAME": "Synthetic SaaS Platform",
+    "VERSION": "1.0.0-synthetic",
+    "PORT": "8001",
+    "HOST": "127.0.0.1",
+    "APP_ENV": "synthetic_env",
+    "SECRET_KEY": "synthetic-secret-key-12345",
+    "FRONTEND_URL": "http://localhost:3000",
+    "CORS_ALLOWED_ORIGINS": "http://localhost:3000",
+    "DATABASE_URL": "postgresql://mock_user:mock_pass@127.0.0.1:5432/mock_saas_db",
+    "REDIS_HOST": "localhost",
+    "REDIS_PORT": "6379",
+    "REDIS_PASSWORD": "synthetic-redis-pass",
+    "SMTP_HOST": "smtp.synthetic.local",
+    "SMTP_PORT": "587",
+    "SMTP_USER": "synthetic-smtp-user",
+    "SMTP_PASSWORD": "synthetic-smtp-pass",
+    "EMAIL_FROM": "synthetic@fake.local",
+    "ACCESS_TOKEN_SECRET": "synthetic-access-secret-12345678901234567890",
+    "REFRESH_TOKEN_SECRET": "synthetic-refresh-secret-12345678901234567890",
+    "SUPER_ADMIN_EMAIL": "admin@synthetic.local",
+    "SUPER_ADMIN_PASSWORD": "SyntheticAdminPass123!",
+    "SUPER_ADMIN_FIRST_NAME": "Super",
+    "SUPER_ADMIN_LAST_NAME": "Admin",
+    "PAYPAL_CLIENT_ID": "synthetic-paypal-id",
+    "PAYPAL_CLIENT_SECRET": "synthetic-paypal-secret",
+}
+
+FAKE_REQUIRED_ENV = """
+PROJECT_NAME=Fake SaaS
+VERSION=1.0.0
+PORT=8001
+HOST=127.0.0.1
+APP_ENV={app_env}
+SECRET_KEY=fake-secret-key-12345
+FRONTEND_URL=http://localhost:3000
+CORS_ALLOWED_ORIGINS=http://localhost:3000
+DATABASE_URL={database_url}
+REDIS_HOST=localhost
+REDIS_PORT=6379
+REDIS_PASSWORD=fake-redis-pass
+SMTP_HOST=smtp.fake.local
+SMTP_PORT=587
+SMTP_USER=fake-smtp-user
+SMTP_PASSWORD=fake-smtp-pass
+EMAIL_FROM=test@fake.local
+ACCESS_TOKEN_SECRET=fake-access-secret-12345678901234567890
+REFRESH_TOKEN_SECRET=fake-refresh-secret-12345678901234567890
+SUPER_ADMIN_EMAIL=admin@fake.local
+SUPER_ADMIN_PASSWORD=FakeAdminPass123!
+SUPER_ADMIN_FIRST_NAME=Super
+SUPER_ADMIN_LAST_NAME=Admin
+PAYPAL_CLIENT_ID=fake-paypal-id
+PAYPAL_CLIENT_SECRET=fake-paypal-secret
+"""
+
+def run_in_isolated_process(env_vars: dict, script: str) -> subprocess.CompletedProcess:
+    """Run python script in an isolated subprocess with clean synthetic environment."""
+    clean_env = {
+        "PATH": os.environ.get("PATH", ""),
+        "PYTHONPATH": str(Path(__file__).resolve().parent.parent),
+    }
+    clean_env.update(env_vars)
+    
+    return subprocess.run(
+        [sys.executable, "-c", textwrap.dedent(script)],
+        env=clean_env,
+        capture_output=True,
+        text=True
+    )
+
+
+def test_process_database_url_overrides_dotenv_local(tmp_path):
+    """1. Process DATABASE_URL overrides .env.local DATABASE_URL."""
+    fake_backend = tmp_path / "saas_backend"
+    fake_backend.mkdir()
+    (fake_backend / ".env.local").write_text(
+        FAKE_REQUIRED_ENV.format(app_env="local", database_url="postgresql://dotenv_user:pass@127.0.0.1/dotenv_db")
+    )
+    
+    script = f"""
+    import importlib.util
+    from pathlib import Path
+    
+    spec = importlib.util.spec_from_file_location("settings_module", "{Path(__file__).resolve().parent.parent}/app/config/settings.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    
+    settings = mod.Settings(_env_file=str(Path("{fake_backend}") / ".env.local"))
+    print("EFFECTIVE_DB_URL=" + settings.DATABASE_URL)
+    """
+    
+    proc_env = dict(SYNTHETIC_PROCESS_ENV)
+    proc_env["DATABASE_URL"] = "postgresql://process_user:pass@127.0.0.1/process_db"
+    proc_env["APP_ENV"] = "local"
+    
+    proc = run_in_isolated_process(proc_env, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "EFFECTIVE_DB_URL=postgresql://process_user:pass@127.0.0.1/process_db" in proc.stdout
+
+
+def test_app_env_testing_selects_env_testing(tmp_path):
+    """2. APP_ENV=testing selects .env.testing."""
+    fake_backend = tmp_path / "saas_backend"
+    fake_backend.mkdir()
+    
+    (fake_backend / ".env.testing").write_text(
+        FAKE_REQUIRED_ENV.format(app_env="testing", database_url="postgresql://test_user:pass@127.0.0.1/testing_db")
+    )
+    (fake_backend / ".env.local").write_text(
+        FAKE_REQUIRED_ENV.format(app_env="local", database_url="postgresql://local_user:pass@127.0.0.1/local_db")
+    )
+    
+    script = f"""
+    import os, importlib.util
+    from pathlib import Path
+    
+    app_env = os.getenv("APP_ENV", "local")
+    env_file = str(Path("{fake_backend}") / f".env.{{app_env}}")
+    
+    spec = importlib.util.spec_from_file_location("settings_module", "{Path(__file__).resolve().parent.parent}/app/config/settings.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    
+    settings = mod.Settings(_env_file=env_file)
+    print("EFFECTIVE_APP_ENV=" + settings.APP_ENV)
+    print("EFFECTIVE_DB_URL=" + settings.DATABASE_URL)
+    """
+    
+    proc = run_in_isolated_process({"APP_ENV": "testing"}, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "EFFECTIVE_APP_ENV=testing" in proc.stdout
+    assert "EFFECTIVE_DB_URL=postgresql://test_user:pass@127.0.0.1/testing_db" in proc.stdout
+
+
+def test_environment_specific_file_overrides_base_env(tmp_path):
+    """3. Environment-specific file overrides base .env."""
+    fake_backend = tmp_path / "saas_backend"
+    fake_backend.mkdir()
+    
+    (fake_backend / ".env").write_text(
+        FAKE_REQUIRED_ENV.format(app_env="base", database_url="postgresql://base_user:pass@127.0.0.1/base_db") + "\nPORT=8001\n"
+    )
+    (fake_backend / ".env.custom").write_text(
+        "DATABASE_URL=postgresql://custom_user:pass@127.0.0.1/custom_db\nPORT=9000\n"
+    )
+    
+    script = f"""
+    import importlib.util
+    from pathlib import Path
+    
+    spec = importlib.util.spec_from_file_location("settings_module", "{Path(__file__).resolve().parent.parent}/app/config/settings.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    
+    env_files = [
+        str(Path("{fake_backend}") / ".env"),
+        str(Path("{fake_backend}") / ".env.custom"),
+    ]
+    settings = mod.Settings(_env_file=env_files)
+    print("EFFECTIVE_PORT=" + str(settings.PORT))
+    print("EFFECTIVE_DB_URL=" + settings.DATABASE_URL)
+    """
+    
+    proc = run_in_isolated_process({}, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "EFFECTIVE_PORT=9000" in proc.stdout
+    assert "EFFECTIVE_DB_URL=postgresql://custom_user:pass@127.0.0.1/custom_db" in proc.stdout
+
+
+def test_base_env_fills_value_absent_from_env_specific(tmp_path):
+    """4. Base .env fills a value absent from the environment-specific file."""
+    fake_backend = tmp_path / "saas_backend"
+    fake_backend.mkdir()
+    
+    (fake_backend / ".env").write_text(
+        FAKE_REQUIRED_ENV.format(app_env="base", database_url="postgresql://base_user:pass@127.0.0.1/base_db")
+        + "\nENCRYPTION_KEY=base-encryption-key\n"
+    )
+    (fake_backend / ".env.special").write_text(
+        "DATABASE_URL=postgresql://special_user:pass@127.0.0.1/special_db\n"
+    )
+    
+    script = f"""
+    import importlib.util
+    from pathlib import Path
+    
+    spec = importlib.util.spec_from_file_location("settings_module", "{Path(__file__).resolve().parent.parent}/app/config/settings.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    
+    env_files = [
+        str(Path("{fake_backend}") / ".env"),
+        str(Path("{fake_backend}") / ".env.special"),
+    ]
+    settings = mod.Settings(_env_file=env_files)
+    print("EFFECTIVE_DB_URL=" + settings.DATABASE_URL)
+    print("EFFECTIVE_ENC_KEY=" + str(settings.ENCRYPTION_KEY))
+    """
+    
+    proc = run_in_isolated_process({}, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "EFFECTIVE_DB_URL=postgresql://special_user:pass@127.0.0.1/special_db" in proc.stdout
+    assert "EFFECTIVE_ENC_KEY=base-encryption-key" in proc.stdout
+
+
+def test_importing_alembic_config_does_not_overwrite_process_variables():
+    """5. Importing Alembic configuration does not overwrite process variables."""
+    sentinel_env = dict(SYNTHETIC_PROCESS_ENV)
+    sentinel_env["DATABASE_URL"] = "postgresql://sentinel_user:pass@127.0.0.1/sentinel_db"
+    sentinel_env["PROJECT_NAME"] = "SENTINEL_SAAS"
+    
+    script = f"""
+    import os, sys
+    from pathlib import Path
+    sys.path.insert(0, "{Path(__file__).resolve().parent.parent}")
+    
+    # Import Alembic Config and inspect sqlalchemy.url
+    from alembic.config import Config
+    alembic_cfg = Config("{Path(__file__).resolve().parent.parent}/alembic.ini")
+    
+    # Ensure os.environ is untouched
+    assert os.environ.get("DATABASE_URL") == "postgresql://sentinel_user:pass@127.0.0.1/sentinel_db"
+    assert os.environ.get("PROJECT_NAME") == "SENTINEL_SAAS"
+    print("ENV_PRESERVED=TRUE")
+    """
+    
+    proc = run_in_isolated_process(sentinel_env, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "ENV_PRESERVED=TRUE" in proc.stdout
+
+
+def test_effective_alembic_url_comes_from_settings():
+    """6. The effective Alembic URL comes from Settings."""
+    sentinel_env = dict(SYNTHETIC_PROCESS_ENV)
+    sentinel_env["DATABASE_URL"] = "postgresql://saas_user:saas_pass@127.0.0.1:5432/saas_local"
+    
+    script = f"""
+    import sys
+    from pathlib import Path
+    sys.path.insert(0, "{Path(__file__).resolve().parent.parent}")
+    
+    from app.config.settings import settings
+    # Escaping logic in alembic/env.py
+    escaped_url = settings.DATABASE_URL.replace("%", "%%")
+    print("ALEMBIC_URL=" + escaped_url)
+    """
+    
+    proc = run_in_isolated_process(sentinel_env, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "ALEMBIC_URL=postgresql://saas_user:saas_pass@127.0.0.1:5432/saas_local" in proc.stdout
+
+
+def test_percent_encoded_database_urls_handled_safely():
+    """7. Percent-encoded database URLs are handled safely without ConfigParser interpolation error."""
+    sentinel_env = dict(SYNTHETIC_PROCESS_ENV)
+    # Password containing percent encoded chars %40, %21
+    sentinel_env["DATABASE_URL"] = "postgresql://user:p%40ss%21word@127.0.0.1:5432/saas_db"
+    
+    script = f"""
+    import sys, configparser
+    from pathlib import Path
+    sys.path.insert(0, "{Path(__file__).resolve().parent.parent}")
+    
+    from app.config.settings import settings
+    raw_url = settings.DATABASE_URL
+    escaped_url = raw_url.replace("%", "%%")
+    
+    # Simulate ConfigParser reading this option as alembic does
+    parser = configparser.ConfigParser()
+    parser.add_section("alembic")
+    parser.set("alembic", "sqlalchemy.url", escaped_url)
+    
+    resolved = parser.get("alembic", "sqlalchemy.url")
+    assert resolved == raw_url, f"Resolved {{resolved}} does not match raw {{raw_url}}"
+    print("CONFIGPARSER_PERCENT_SAFE=TRUE")
+    """
+    
+    proc = run_in_isolated_process(sentinel_env, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "CONFIGPARSER_PERCENT_SAFE=TRUE" in proc.stdout
+
+
+def test_no_repo_dotenv_value_leaked_and_zero_network():
+    """8. Prove no real repository dotenv value is loaded or printed, and zero network calls occur."""
+    sentinel_env = dict(SYNTHETIC_PROCESS_ENV)
+    sentinel_env["DATABASE_URL"] = "postgresql://mock_iso:mock_iso@127.0.0.1:5432/mock_db"
+    
+    script = f"""
+    import sys
+    from pathlib import Path
+    sys.path.insert(0, "{Path(__file__).resolve().parent.parent}")
+    
+    from app.config.settings import settings
+    print("SETTINGS_APP_ENV=" + settings.APP_ENV)
+    print("SETTINGS_DB_URL=" + settings.DATABASE_URL)
+    """
+    
+    proc = run_in_isolated_process(sentinel_env, script)
+    assert proc.returncode == 0, f"Error: {proc.stderr}"
+    assert "SETTINGS_APP_ENV=synthetic_env" in proc.stdout
+    assert "SETTINGS_DB_URL=postgresql://mock_iso:mock_iso@127.0.0.1:5432/mock_db" in proc.stdout
+    # Confirm 106.51.105.22 never leaks
+    assert "106.51.105.22" not in proc.stdout
+    assert "106.51.105.22" not in proc.stderr

tests/test_sso_controller.py

@@ -0,0 +1,37 @@
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from app.controllers.auth.sso_controller import SSOController
+from app.schemas.auth.sso_schema import SSOInitiateRequest
+
+
+class SSOControllerTest(unittest.TestCase):
+    def test_initiate_sso_uses_one_time_grant_flow(self):
+        user = SimpleNamespace(id="user-1", tenant_id="tenant-1")
+        expected = {
+            "grant_code": "a" * 32,
+            "redirect_url": "http://127.0.0.1:5173/sso/callback?grant=" + "a" * 32,
+        }
+
+        with patch(
+            "app.controllers.auth.sso_controller.SSOService.generate_grant",
+            return_value=expected,
+        ) as generate_grant:
+            result = SSOController.initiate_sso(
+                db="db-session",
+                request=SSOInitiateRequest(module_id="pim"),
+                current_user=user,
+            )
+
+        self.assertEqual(result, expected)
+        generate_grant.assert_called_once_with(
+            db="db-session",
+            user_id="user-1",
+            module_id="pim",
+            tenant_id="tenant-1",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_tenant_onboarding.py

@@ -0,0 +1,966 @@
+import unittest
+import uuid
+from datetime import date, datetime, timezone
+from unittest.mock import patch, MagicMock
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.dialects.postgresql import JSONB
+
+@compiles(JSONB, 'sqlite')
+def compile_jsonb_sqlite(type_, compiler, **kw):
+    return 'TEXT'
+
+from fastapi import HTTPException
+from app.config.database import Base
+
+from app.models.auth.tenant_model import Tenant
+from app.models.auth.user_model import User
+from app.models.auth.role_model import Role
+from app.models.auth.subscription_plan_model import SubscriptionPlan
+from app.models.auth.plan_access_model import PlanAccess
+from app.models.auth.plan_module_access_model import PlanModuleAccess
+from app.models.auth.module_model import Module
+from app.models.auth.module_access_model import ModuleAccess
+from app.models.auth.module_environment_model import ModuleEnvironment
+from app.models.auth.tenant_module_model import TenantModule
+from app.models.auth.access_model import Access
+from app.models.auth.role_access_model import RoleAccess
+from app.models.auth.role_module_access_model import RoleModuleAccess
+from app.models.system.event_log_model import EventLog
+from app.models.system.audit_log import AuditLog
+from app.schemas.auth.tenant_schema import TenantCreate, TenantOwnerCreate, ModuleEnvironmentAssignment, TenantResponse
+
+from app.services.auth.tenant_service import TenantService
+from app.services.auth.subscription_plan_service import SubscriptionPlanService
+from app.services.auth.role_service import RoleService
+from app.services.system.audit_log_service import AuditLogService
+from app.services.auth.event_service import EventService
+from app.controllers.auth.tenant_controller import TenantController
+from app.config.security import SecurityUtils, security
+from app.config.settings import settings
+from app.middleware.auth_middleware import can_access
+
+
+class TenantOnboardingUnitTest(unittest.TestCase):
+    def setUp(self):
+        # In-memory SQLite database for isolated unit testing
+        self.engine = create_engine("sqlite:///:memory:")
+        tables = [
+            Tenant.__table__,
+            User.__table__,
+            Role.__table__,
+            SubscriptionPlan.__table__,
+            PlanAccess.__table__,
+            PlanModuleAccess.__table__,
+            Module.__table__,
+            ModuleAccess.__table__,
+            ModuleEnvironment.__table__,
+            TenantModule.__table__,
+            EventLog.__table__,
+            AuditLog.__table__,
+            Access.__table__,
+            RoleAccess.__table__,
+            RoleModuleAccess.__table__,
+        ]
+        Base.metadata.create_all(self.engine, tables=tables)
+        self.Session = sessionmaker(bind=self.engine)
+        self.db = self.Session()
+
+        # Seed SaaS accesses
+        self.tenant_create_access = Access(
+            id=uuid.uuid4(),
+            access_code="superadmin.tenant.create",
+            category="saas",
+            name="Create Tenant",
+        )
+        self.db.add(self.tenant_create_access)
+        self.db.commit()
+
+        # Seed test modules: PIM, Inventory, Fulfillment, and legacy Sales
+        self.pim_mod = Module(
+            id=uuid.uuid4(),
+            module_id="pim",
+            module_name="PIM",
+            status="active",
+            display_order=0,
+        )
+        self.inv_mod = Module(
+            id=uuid.uuid4(),
+            module_id="inventory",
+            module_name="Inventory",
+            status="active",
+            display_order=1,
+        )
+        self.fl_mod = Module(
+            id=uuid.uuid4(),
+            module_id="fulfillment",
+            module_name="Fulfillment & Logistics",
+            status="active",
+            display_order=2,
+        )
+        self.legacy_sales_mod = Module(
+            id=uuid.uuid4(),
+            module_id="sales",
+            module_name="Sales Management",
+            status="active",
+            display_order=3,
+        )
+        self.db.add_all([self.pim_mod, self.inv_mod, self.fl_mod, self.legacy_sales_mod])
+        self.db.commit()
+
+        # Seed environments
+        self.pim_env = ModuleEnvironment(
+            id=uuid.uuid4(),
+            module_id=self.pim_mod.id,
+            slug="local",
+            frontend_base_url="http://127.0.0.1:5173",
+            backend_base_url="http://localhost:5002",
+            trust_type="rsa",
+            trust_credentials={"test": "key"},
+            is_default=True,
+            is_active=True,
+        )
+        self.inv_env = ModuleEnvironment(
+            id=uuid.uuid4(),
+            module_id=self.inv_mod.id,
+            slug="local",
+            frontend_base_url="http://127.0.0.1:5174",
+            backend_base_url="http://localhost:12001",
+            trust_type="rsa",
+            trust_credentials={"test": "key"},
+            is_default=True,
+            is_active=True,
+        )
+        self.fl_env = ModuleEnvironment(
+            id=uuid.uuid4(),
+            module_id=self.fl_mod.id,
+            slug="local",
+            frontend_base_url="http://127.0.0.1:5175",
+            backend_base_url="http://localhost:8000",
+            trust_type="rsa",
+            trust_credentials={"test": "key"},
+            is_default=True,
+            is_active=True,
+        )
+        self.db.add_all([self.pim_env, self.inv_env, self.fl_env])
+        self.db.commit()
+
+        # Seed module accesses (for plan linking)
+        self.pim_access = ModuleAccess(
+            id=uuid.uuid4(),
+            module_id=self.pim_mod.id,
+            access_code="pim.app.launch",
+            name="Launch PIM",
+            category="launcher",
+        )
+        self.inv_access = ModuleAccess(
+            id=uuid.uuid4(),
+            module_id=self.inv_mod.id,
+            access_code="inventory.app.launch",
+            name="Launch Inventory",
+            category="launcher",
+        )
+        self.sales_access = ModuleAccess(
+            id=uuid.uuid4(),
+            module_id=self.legacy_sales_mod.id,
+            access_code="sales.app.launch",
+            name="Launch Sales",
+            category="launcher",
+        )
+        self.db.add_all([self.pim_access, self.inv_access, self.sales_access])
+        self.db.commit()
+
+        # Seed subscription plan with PIM + Inventory + legacy Sales
+        self.plan = SubscriptionPlan(
+            id=uuid.uuid4(),
+            name="Standard Business Plan",
+            duration_days=30,
+            max_users_allowed=10,
+            is_public=True,
+            status="active",
+        )
+        self.db.add(self.plan)
+        self.db.commit()
+
+        self.pma1 = PlanModuleAccess(plan_id=self.plan.id, module_access_id=self.pim_access.id)
+        self.pma2 = PlanModuleAccess(plan_id=self.plan.id, module_access_id=self.inv_access.id)
+        self.pma3 = PlanModuleAccess(plan_id=self.plan.id, module_access_id=self.sales_access.id)
+        self.db.add_all([self.pma1, self.pma2, self.pma3])
+        self.db.commit()
+
+    def tearDown(self):
+        self.db.close()
+
+    def test_plan_applications_response_includes_only_onboarding_apps_without_secrets(self):
+        apps = SubscriptionPlanService.resolve_plan_applications(self.db, self.plan, onboarding_only=True)
+        mod_codes = [a.module_code for a in apps]
+        self.assertIn("pim", mod_codes)
+        self.assertIn("inventory", mod_codes)
+        self.assertNotIn("sales", mod_codes)  # Legacy internal submodule excluded from onboarding contract
+
+        # Verify environment serialization omits trust_credentials
+        for app in apps:
+            for env in app.environments:
+                self.assertFalse(hasattr(env, "trust_credentials"))
+                self.assertIn(env.slug, ["local"])
+
+    def test_plan_with_no_application_links_returns_empty_list_fail_closed(self):
+        # Create a new plan with NO PlanModuleAccess links
+        empty_plan = SubscriptionPlan(
+            id=uuid.uuid4(),
+            name="Empty Unlinked Plan",
+            duration_days=30,
+            max_users_allowed=5,
+            is_public=True,
+            status="active",
+        )
+        self.db.add(empty_plan)
+        self.db.commit()
+
+        # Must return empty list, NOT fallback to all active applications
+        apps = SubscriptionPlanService.resolve_plan_applications(self.db, empty_plan, onboarding_only=True)
+        self.assertEqual(apps, [])
+
+        apps_all = SubscriptionPlanService.resolve_plan_applications(self.db, empty_plan, onboarding_only=False)
+        self.assertEqual(apps_all, [])
+
+    def test_tenant_creation_on_plan_with_no_links_rejected(self):
+        # Create a plan with zero links
+        empty_plan = SubscriptionPlan(
+            id=uuid.uuid4(),
+            name="Zero Link Plan",
+            duration_days=30,
+            max_users_allowed=5,
+            is_public=True,
+            status="active",
+        )
+        self.db.add(empty_plan)
+        self.db.commit()
+
+        tenant_create = TenantCreate(
+            tenant_name="Empty Plan Tenant",
+            tenant_domain="emptyplan.local",
+            plan_id=empty_plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Zero",
+                last_name="Plan",
+                email="zeroplan@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("not included in the selected plan", ctx.exception.detail)
+
+    def test_successful_tenant_onboarding_with_owner_and_applications(self):
+        owner_data = TenantOwnerCreate(
+            first_name="Ash",
+            last_name="Admin",
+            email="owner@example.com",
+            password="StrongPassword@123",
+            phone_number="+919876543210",
+        )
+        tenant_create = TenantCreate(
+            tenant_name="Ash Traders",
+            tenant_domain="ashtraders.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id, self.inv_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local"),
+                ModuleEnvironmentAssignment(module_id=self.inv_mod.id, environment_slug="local"),
+            ],
+            owner=owner_data,
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush"):
+            tenant = TenantService.create_tenant(self.db, tenant_create)
+
+        self.assertIsNotNone(tenant.id)
+        self.assertEqual(tenant.tenant_name, "Ash Traders")
+        self.assertEqual(tenant.tenant_domain, "ashtraders.local")
+
+        # Verify Owner user created and mapped to Primary Admin role
+        owner = self.db.query(User).filter(User.email == "owner@example.com").first()
+        self.assertIsNotNone(owner)
+        self.assertEqual(owner.tenant_id, tenant.id)
+        self.assertNotEqual(owner.password, "StrongPassword@123")  # Password must be hashed
+
+        # Verify Primary Admin role created with launcher permissions ONLY for selected modules
+        role = self.db.query(Role).filter(Role.id == owner.role_id).first()
+        self.assertIsNotNone(role)
+        self.assertEqual(role.role_name, "Primary Admin")
+
+        role_module_accesses = self.db.query(RoleModuleAccess).filter(RoleModuleAccess.role_id == role.id).all()
+        role_mod_acc_ids = {rma.module_access_id for rma in role_module_accesses}
+        self.assertIn(self.pim_access.id, role_mod_acc_ids)
+        self.assertIn(self.inv_access.id, role_mod_acc_ids)
+        # Assert the unselected legacy Sales module is ABSENT from RoleModuleAccess
+        self.assertNotIn(self.sales_access.id, role_mod_acc_ids)
+
+        # Verify TenantModule assignments
+        tms = self.db.query(TenantModule).filter(TenantModule.tenant_id == tenant.id).all()
+        tm_mod_ids = {tm.module_id for tm in tms}
+        self.assertEqual(tm_mod_ids, {self.pim_mod.id, self.inv_mod.id})
+
+        # Verify outbox event logs created durably
+        events = self.db.query(EventLog).all()
+        event_types = [e.event_type for e in events]
+        self.assertIn("TENANT_PROVISION_REQUESTED", event_types)
+        self.assertIn("USER_PROVISION_REQUESTED", event_types)
+
+        # Confirm every event payload contains NO 'password' key
+        for event in events:
+            payload = event.payload if isinstance(event.payload, dict) else {}
+            self.assertNotIn("password", payload)
+            payload_str = str(event.payload)
+            self.assertNotIn("StrongPassword@123", payload_str)
+
+    def test_selected_application_outside_plan_rejected(self):
+        # F&L is not in self.plan
+        tenant_create = TenantCreate(
+            tenant_name="Illegal Plan Tenant",
+            tenant_domain="illegal.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.fl_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.fl_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Illegal",
+                last_name="Admin",
+                email="illegal@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("not included in the selected plan", ctx.exception.detail)
+
+    def test_duplicate_owner_email_rejected_before_db_write(self):
+        # Pre-create a user with this email
+        pre_user = User(
+            email="existing_owner@test.com",
+            password="hashedpassword",
+            first_name="Existing",
+            last_name="User",
+        )
+        self.db.add(pre_user)
+        self.db.commit()
+
+        tenant_create = TenantCreate(
+            tenant_name="Duplicate Email Tenant",
+            tenant_domain="dupemail.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Duplicate",
+                last_name="Owner",
+                email="existing_owner@test.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("already registered", ctx.exception.detail)
+
+        # Confirm tenant was never created
+        t = self.db.query(Tenant).filter(Tenant.tenant_domain == "dupemail.local").first()
+        self.assertIsNone(t)
+
+    def test_weak_password_rejected(self):
+        tenant_create = TenantCreate(
+            tenant_name="Weak Password Tenant",
+            tenant_domain="weakpass.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Weak",
+                last_name="Pass",
+                email="weak@example.com",
+                password="weakpassword1",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("too weak", ctx.exception.detail)
+
+    def test_duplicate_tenant_domain_rejected(self):
+        t1 = Tenant(
+            tenant_name="First Tenant",
+            tenant_domain="same-domain.local",
+            plan_id=self.plan.id,
+        )
+        self.db.add(t1)
+        self.db.commit()
+
+        tenant_create = TenantCreate(
+            tenant_name="Second Tenant",
+            tenant_domain="same-domain.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Second",
+                last_name="Owner",
+                email="second@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("already exists", ctx.exception.detail)
+
+    def test_invalid_environment_slug_rejected(self):
+        tenant_create = TenantCreate(
+            tenant_name="Invalid Env Tenant",
+            tenant_domain="invalid-env.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="nonexistent-env")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Invalid",
+                last_name="Env",
+                email="invalid@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("does not exist or is inactive", ctx.exception.detail)
+
+    def test_duplicate_environment_assignments_rejected(self):
+        tenant_create = TenantCreate(
+            tenant_name="Duplicate Env Tenant",
+            tenant_domain="dupenv.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local"),
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local"),
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Dup",
+                last_name="Env",
+                email="dupenv@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("Duplicate environment assignment", ctx.exception.detail)
+
+    def test_selected_application_without_environment_rejected(self):
+        # PIM and Inventory selected, but environment provided only for PIM
+        tenant_create = TenantCreate(
+            tenant_name="Missing Env Tenant",
+            tenant_domain="missingenv.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id, self.inv_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Missing",
+                last_name="Env",
+                email="missingenv@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        with self.assertRaises(HTTPException) as ctx:
+            TenantService.create_tenant(self.db, tenant_create)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("Explicit environment assignment is required", ctx.exception.detail)
+
+    def test_transaction_rollback_on_owner_creation_failure(self):
+        tenant_create = TenantCreate(
+            tenant_name="Rollback Owner Tenant",
+            tenant_domain="rollback-owner.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Rollback",
+                last_name="Owner",
+                email="rollbackowner@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        # Mock password hashing to throw an unexpected runtime error during Owner creation
+        with patch("app.config.security.security.hash_password", side_effect=RuntimeError("Simulated Hash Error")):
+            with self.assertRaises(RuntimeError):
+                TenantService.create_tenant(self.db, tenant_create)
+
+        # Assert full rollback: tenant, roles, and modules were not persisted
+        t = self.db.query(Tenant).filter(Tenant.tenant_domain == "rollback-owner.local").first()
+        self.assertIsNone(t)
+        roles = self.db.query(Role).all()
+        self.assertEqual(len(roles), 0)
+        tms = self.db.query(TenantModule).all()
+        self.assertEqual(len(tms), 0)
+
+    def test_transaction_rollback_on_outbox_event_failure(self):
+        tenant_create = TenantCreate(
+            tenant_name="Rollback Outbox Tenant",
+            tenant_domain="rollback-outbox.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Rollback",
+                last_name="Outbox",
+                email="rollbackoutbox@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+        # Mock EventService.emit_event to throw an unexpected error
+        with patch("app.services.auth.event_service.EventService.emit_event", side_effect=RuntimeError("Outbox write failed")):
+            with self.assertRaises(RuntimeError):
+                TenantService.create_tenant(self.db, tenant_create)
+
+        # Assert full rollback: tenant, owner user, roles, and modules were not persisted
+        t = self.db.query(Tenant).filter(Tenant.tenant_domain == "rollback-outbox.local").first()
+        self.assertIsNone(t)
+        u = self.db.query(User).filter(User.email == "rollbackoutbox@example.com").first()
+        self.assertIsNone(u)
+        roles = self.db.query(Role).all()
+        self.assertEqual(len(roles), 0)
+
+    def test_authorization_platform_admin_vs_tenant_user(self):
+        # 1. Platform Admin role with superadmin.tenant.create
+        platform_role = Role(id=uuid.uuid4(), role_name="SuperAdmin", is_default=False)
+        self.db.add(platform_role)
+        self.db.commit()
+
+        ra = RoleAccess(role_id=platform_role.id, access_id=self.tenant_create_access.id)
+        self.db.add(ra)
+        self.db.commit()
+
+        platform_admin = User(
+            id=uuid.uuid4(),
+            email="platform@maskantech.in",
+            password="hashed",
+            first_name="Platform",
+            last_name="Admin",
+            role_id=platform_role.id,
+            status="active",
+            tenant_id=None,
+        )
+        self.db.add(platform_admin)
+        self.db.commit()
+
+        # 2. Tenant User with standard role (no superadmin permissions)
+        existing_tenant = Tenant(
+            id=uuid.uuid4(),
+            tenant_name="Existing Tenant",
+            tenant_domain="existing.local",
+            plan_id=self.plan.id,
+        )
+        self.db.add(existing_tenant)
+        self.db.commit()
+
+        tenant_role = Role(id=uuid.uuid4(), role_name="Staff", tenant_id=existing_tenant.id)
+        self.db.add(tenant_role)
+        self.db.commit()
+
+        tenant_user = User(
+            id=uuid.uuid4(),
+            email="staff@example.com",
+            password="hashed",
+            first_name="Staff",
+            last_name="User",
+            role_id=tenant_role.id,
+            tenant_id=existing_tenant.id,
+            status="active",
+        )
+        self.db.add(tenant_user)
+        self.db.commit()
+
+        # Check authorization using SaaS access authorization helpers
+        platform_admin_has_access = can_access(platform_admin, "superadmin.tenant.create", self.db)
+        tenant_user_has_access = can_access(tenant_user, "superadmin.tenant.create", self.db)
+
+        self.assertTrue(platform_admin_has_access)
+        self.assertFalse(tenant_user_has_access)
+
+    def test_audit_log_failure_triggers_complete_rollback_with_zero_records(self):
+        tenant_create = TenantCreate(
+            tenant_name="Audit Fail Tenant",
+            tenant_domain="audit-fail.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Audit",
+                last_name="Fail",
+                email="auditfail@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush"):
+            try:
+                # 1. Create tenant/owner/modules/events using flush (commit=False)
+                result = TenantController.create_tenant(self.db, tenant_create, commit=False)
+
+                # 2. Prepare sanitized audit record
+                audit_values = tenant_data_dict = tenant_create.model_dump(mode='json')
+                if "owner" in audit_values and isinstance(audit_values["owner"], dict):
+                    audit_values["owner"]["password"] = "[PROTECTED]"
+
+                # 3. Simulate strict audit log failure
+                with patch("app.services.system.audit_log_service.AuditLog", side_effect=RuntimeError("Simulated Audit Failure")):
+                    AuditLogService.log(
+                        db=self.db,
+                        module_name="Tenants",
+                        action_type="CREATE",
+                        entity_id=str(result.id),
+                        entity_name=result.tenant_name,
+                        description=f"Tenant '{result.tenant_name}' onboarded",
+                        performed_by_id=str(uuid.uuid4()),
+                        performed_by_email="superadmin@maskan.local",
+                        new_values=audit_values,
+                        commit=False,
+                        strict=True,
+                    )
+
+                self.db.commit()
+            except Exception:
+                self.db.rollback()
+
+        # Assert complete atomicity: zero records in all tables
+        self.assertEqual(self.db.query(Tenant).count(), 0)
+        self.assertEqual(self.db.query(User).count(), 0)
+        self.assertEqual(self.db.query(Role).count(), 0)
+        self.assertEqual(self.db.query(TenantModule).count(), 0)
+        self.assertEqual(self.db.query(EventLog).count(), 0)
+        self.assertEqual(self.db.query(AuditLog).count(), 0)
+
+    def test_password_sanitization_and_orchestration_contract(self):
+        raw_password = "SuperSecretPassword#2026"
+        tenant_create = TenantCreate(
+            tenant_name="Sanitize Corp",
+            tenant_domain="sanitize.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id, self.inv_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local"),
+                ModuleEnvironmentAssignment(module_id=self.inv_mod.id, environment_slug="local"),
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Secure",
+                last_name="Owner",
+                email="secureowner@example.com",
+                password=raw_password,
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush"):
+            # Execute full route orchestration sequence
+            result = TenantController.create_tenant(self.db, tenant_create, commit=False)
+
+            audit_values = tenant_create.model_dump(mode='json')
+            if "owner" in audit_values and isinstance(audit_values["owner"], dict):
+                audit_values["owner"]["password"] = "[PROTECTED]"
+
+            AuditLogService.log(
+                db=self.db,
+                module_name="Tenants",
+                action_type="CREATE",
+                entity_id=str(result.id),
+                entity_name=result.tenant_name,
+                description=f"Tenant '{result.tenant_name}' onboarded",
+                performed_by_id=str(uuid.uuid4()),
+                performed_by_email="superadmin@maskan.local",
+                new_values=audit_values,
+                commit=False,
+                strict=True,
+            )
+
+            self.db.commit()
+            self.db.refresh(result)
+
+        # 1. Stored owner password is hashed (never plaintext, verifiable)
+        owner_user = self.db.query(User).filter(User.email == "secureowner@example.com").first()
+        self.assertIsNotNone(owner_user)
+        self.assertNotEqual(owner_user.password, raw_password)
+        self.assertTrue(security.verify_password(raw_password, owner_user.password))
+
+        # 2. AuditLog contains [PROTECTED], never plaintext or hash
+        audit_entry = self.db.query(AuditLog).filter(AuditLog.entity_id == str(result.id)).first()
+        self.assertIsNotNone(audit_entry)
+        self.assertEqual(audit_entry.new_values["owner"]["password"], "[PROTECTED]")
+        audit_str = str(audit_entry.new_values)
+        self.assertNotIn(raw_password, audit_str)
+        self.assertNotIn(owner_user.password, audit_str)
+
+        # 3. EventLog payload JSON contains neither plaintext password nor 'password' key
+        events = self.db.query(EventLog).all()
+        self.assertGreater(len(events), 0)
+        for ev in events:
+            payload_str = str(ev.payload)
+            self.assertNotIn(raw_password, payload_str)
+            self.assertNotIn(owner_user.password, payload_str)
+            # Check nested data
+            data = ev.payload.get("data", {}) if isinstance(ev.payload, dict) else {}
+            self.assertNotIn("password", data)
+
+        # 4. API response contains no password field
+        response_model = TenantResponse.model_validate(result)
+        response_dict = response_model.model_dump()
+        self.assertNotIn("password", response_dict)
+        self.assertNotIn("owner_password", response_dict)
+
+    def test_rsa_file_fallback_environment_isolation(self):
+        # 1. When APP_ENV is production and SAAS_PRIVATE_KEY is unset -> must return None (refuse file fallback)
+        with patch.object(settings, "APP_ENV", "production"):
+            with patch.object(settings, "SAAS_PRIVATE_KEY", None):
+                with patch("pathlib.Path.exists", return_value=True):
+                    with patch("pathlib.Path.read_text", return_value="FAKE_RSA_KEY"):
+                        key = SecurityUtils.get_saas_private_key()
+                        self.assertIsNone(key)
+
+        # 2. When APP_ENV is production and SAAS_PRIVATE_KEY is set in settings -> returns config key
+        with patch.object(settings, "APP_ENV", "production"):
+            with patch.object(settings, "SAAS_PRIVATE_KEY", "PROD_CONFIGURED_RSA_KEY"):
+                key = SecurityUtils.get_saas_private_key()
+                self.assertEqual(key, "PROD_CONFIGURED_RSA_KEY")
+
+        # 3. When APP_ENV is local and SAAS_PRIVATE_KEY is unset -> allows file fallback
+        with patch.object(settings, "APP_ENV", "local"):
+            with patch.object(settings, "SAAS_PRIVATE_KEY", None):
+                with patch("pathlib.Path.exists", return_value=True):
+                    with patch("pathlib.Path.read_text", return_value="LOCAL_FILE_RSA_KEY"):
+                        key = SecurityUtils.get_saas_private_key()
+                        self.assertEqual(key, "LOCAL_FILE_RSA_KEY")
+
+        # 4. When APP_ENV is testing and SAAS_PRIVATE_KEY is unset -> allows file fallback
+        with patch.object(settings, "APP_ENV", "testing"):
+            with patch.object(settings, "SAAS_PRIVATE_KEY", None):
+                with patch("pathlib.Path.exists", return_value=True):
+                    with patch("pathlib.Path.read_text", return_value="TESTING_FILE_RSA_KEY"):
+                        key = SecurityUtils.get_saas_private_key()
+                        self.assertEqual(key, "TESTING_FILE_RSA_KEY")
+
+    def test_no_redis_enqueue_before_commit_and_successful_commit_triggers_notification(self):
+        tenant_create = TenantCreate(
+            tenant_name="Redis Timing Corp",
+            tenant_domain="redistiming.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Timing",
+                last_name="Owner",
+                email="timingowner@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush") as mock_rpush:
+            # 1. Uncommitted phase: tenant creation with flush only
+            result = TenantController.create_tenant(self.db, tenant_create, commit=False)
+            
+            # Assert zero Redis enqueue occurred before DB commit
+            mock_rpush.assert_not_called()
+
+            # Pending event IDs are tracked
+            self.assertTrue(hasattr(result, "pending_event_ids"))
+            self.assertGreater(len(result.pending_event_ids), 0)
+
+            # 2. Database Commit phase
+            self.db.commit()
+            self.db.refresh(result)
+
+            # 3. Post-commit wake-up phase
+            EventService.enqueue_wakeups(result.pending_event_ids)
+
+            # Assert Redis was notified after commit
+            self.assertEqual(mock_rpush.call_count, len(result.pending_event_ids))
+
+    def test_rollback_produces_no_database_event_and_no_redis_notification(self):
+        tenant_create = TenantCreate(
+            tenant_name="Rollback Event Corp",
+            tenant_domain="rollbackevent.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Rollback",
+                last_name="Owner",
+                email="rollbackevent@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush") as mock_rpush:
+            TenantController.create_tenant(self.db, tenant_create, commit=False)
+            self.db.rollback()
+
+            # Assert zero Redis calls and zero EventLog rows in DB
+            mock_rpush.assert_not_called()
+            self.assertEqual(self.db.query(EventLog).count(), 0)
+
+    def test_redis_failure_after_commit_still_returns_tenant_success(self):
+        tenant_create = TenantCreate(
+            tenant_name="Redis Down Corp",
+            tenant_domain="redisdown.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Down",
+                last_name="Owner",
+                email="downowner@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush", side_effect=Exception("Redis Connection Refused")):
+            # Simulate full route execution
+            result = TenantController.create_tenant(self.db, tenant_create, commit=False)
+
+            audit_values = tenant_create.model_dump(mode='json')
+            if "owner" in audit_values and isinstance(audit_values["owner"], dict):
+                audit_values["owner"]["password"] = "[PROTECTED]"
+
+            AuditLogService.log(
+                db=self.db,
+                module_name="Tenants",
+                action_type="CREATE",
+                entity_id=str(result.id),
+                entity_name=result.tenant_name,
+                description=f"Tenant '{result.tenant_name}' onboarded",
+                performed_by_id=str(uuid.uuid4()),
+                performed_by_email="superadmin@maskan.local",
+                new_values=audit_values,
+                commit=False,
+                strict=True,
+            )
+
+            self.db.commit()
+            self.db.refresh(result)
+
+            # Redis wake-up fails, but must not throw or block the return
+            EventService.enqueue_wakeups(getattr(result, "pending_event_ids", []))
+
+            self.assertIsNotNone(result.id)
+            self.assertEqual(result.tenant_name, "Redis Down Corp")
+
+    def test_user_provisioning_contains_canonical_user_id(self):
+        tenant_create = TenantCreate(
+            tenant_name="Canonical Identity Corp",
+            tenant_domain="canonical.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="Canon",
+                last_name="User",
+                email="canonuser@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush"):
+            result = TenantController.create_tenant(self.db, tenant_create, commit=True)
+
+        owner = self.db.query(User).filter(User.email == "canonuser@example.com").first()
+        self.assertIsNotNone(owner)
+        self.assertEqual(owner.canonical_user_id, owner.id)
+
+        user_event = self.db.query(EventLog).filter(EventLog.event_type == "USER_PROVISION_REQUESTED").first()
+        self.assertIsNotNone(user_event)
+        data = user_event.payload.get("data", {})
+        self.assertEqual(data.get("user_id"), str(owner.id))
+        self.assertEqual(data.get("canonical_user_id"), str(owner.id))
+        self.assertEqual(data.get("canonical_tenant_id"), str(result.id))
+
+    def test_worker_outbox_processing_and_hmac_no_nameerror(self):
+        # Update environment with HMAC secret
+        self.pim_env.trust_credentials = {"hmac_secret": "test-hmac-secret-123"}
+        self.db.commit()
+
+        tenant_create = TenantCreate(
+            tenant_name="Worker HMAC Corp",
+            tenant_domain="workerhmac.local",
+            plan_id=self.plan.id,
+            selected_module_ids=[self.pim_mod.id],
+            module_environments=[
+                ModuleEnvironmentAssignment(module_id=self.pim_mod.id, environment_slug="local")
+            ],
+            owner=TenantOwnerCreate(
+                first_name="HMAC",
+                last_name="Worker",
+                email="hmacworker@example.com",
+                password="StrongPassword@123",
+            ),
+        )
+
+        with patch("app.services.auth.event_service.sync_redis_client.rpush"):
+            TenantController.create_tenant(self.db, tenant_create, commit=True)
+
+        # Mock httpx.post to simulate successful module webhook delivery
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = "OK"
+
+        with patch("httpx.post", return_value=mock_response) as mock_post:
+            processed = 0
+            while True:
+                batch_count = EventService.process_outbox(self.db, batch_size=10)
+                if batch_count == 0:
+                    break
+                processed += batch_count
+
+            self.assertGreater(processed, 0)
+
+            # Check that httpx.post was invoked with HMAC headers containing timestamp and signature
+            call_kwargs = mock_post.call_args[1]
+            headers = call_kwargs["headers"]
+            self.assertIn("X-SaaS-Timestamp", headers)
+            self.assertIn("X-SaaS-Signature", headers)
+            self.assertGreater(len(headers["X-SaaS-Signature"]), 0)
+
+        # Confirm all pending logs for this tenant transitioned to COMPLETED
+        pending = self.db.query(EventLog).filter(EventLog.status == "PENDING").count()
+        self.assertEqual(pending, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()