Author SHA1 Message Date
Inamul-hasan-tec 75b23efaf4 fix(rbac): scope tenant roles to subscribed modules 2026-09-08 16:24:58 +05:30
Inamul-hasan-tec f87b70099e feat(saas): canonical tenant identity provisioning and asymmetric sso signing 2026-09-05 13:17:31 +05:30
Inamul-hasan-tec 1bd3050474 chore(security): stop tracking environment files and add sanitized template 2026-09-05 13:12:25 +05:30
Inamul-hasan-tec 9178041844 feat(onboarding): add atomic tenant owner and application provisioning 2026-09-04 15:06:42 +05:30
Inamul-hasan-tec 5073833831 feat(local): add isolated ecosystem module seeder 2026-09-04 12:11:31 +05:30
Inamul-hasan-tec de1d35c18f fix(config): preserve SaaS process environment precedence 2026-09-04 11:42:38 +05:30
Inamul-hasan-tec 66d21563c0 fix(tenants): preserve RLS context and atomic onboarding 2026-09-02 16:11:58 +05:30
Inamul-hasan-tec 6b57c0b19f fix(auth): allow verified platform users through RLS 2026-09-02 15:55:35 +05:30
Inamul-hasan-tec 552fcd7be5 feat(auth): gate module SSO by user assignment 2026-09-02 15:11:09 +05:30
Inamul-hasan-tec 1e9da77555 fix(auth): separate entitlements from role permissions 2026-09-02 11:23:24 +05:30
Inamul-hasan-tec 9d322c579b fix(sso): align module trust and identity claims 2026-09-02 10:47:00 +05:30
Inamul-hasan-tec bb51048958 fix(rbac): synchronize signed module permissions 2026-09-01 12:45:18 +05:30
Inamul-hasan-tec bd6477784c fix(auth): establish tenant context for authentication 2026-09-01 11:37:14 +05:30
Inamul-hasan-tec fffe605d53 feat(sso): issue one-time grants for module launch 2026-09-01 11:21:47 +05:30
42 changed files with 2920 additions and 496 deletions
-62
View File
@@ -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
+62
View File
@@ -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=
-63
View File
@@ -1,63 +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,http://localhost:5179
CORS_ALLOW_ORIGIN_REGEX=https?://(localhost|127\.0\.0\.1):\d+
# 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
-67
View File
@@ -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
-62
View File
@@ -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
+5 -1
View File
@@ -6,7 +6,11 @@ __pycache__/
*.pyc
# Environment
.env
.env*
!.env.example
keys/
*.log
node_modules
# IDE
+2 -22
View File
@@ -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
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)
# Set the database URL from app settings (escape % for ConfigParser interpolation)
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL.replace("%", "%%"))
# add your model's MetaData object here
# for 'autogenerate' support
+2 -5
View File
@@ -167,21 +167,18 @@ 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"
)
# Vite hops 5173+ when ports are taken. Local regex covers that without listing every port.
origin_regex = settings.CORS_ALLOW_ORIGIN_REGEX
if settings.APP_ENV == "local" and not origin_regex:
origin_regex = r"https?://(localhost|127\.0\.0\.1):\d+"
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_origin_regex=origin_regex,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
+16 -2
View File
@@ -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()
+26 -6
View File
@@ -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:
raise ValueError("SAAS_PRIVATE_KEY is not configured. Cannot sign module identity tokens.")
private_key = SecurityUtils.get_saas_private_key()
if not private_key:
raise ValueError("SAAS_PRIVATE_KEY is not configured. Cannot sign module identity tokens.")
return jwt.encode(
to_encode,
settings.SAAS_PRIVATE_KEY,
to_encode,
private_key,
algorithm="RS256",
headers={"kid": settings.SAAS_KEY_ID}
)
@staticmethod
def validate_email(email: str) -> bool:
"""Validate email format."""
+17 -12
View File
@@ -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')
load_dotenv(dotenv_path=backend_path / '.env')
env_files = []
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()
+8 -3
View File
@@ -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]:
return AccessService.get_all_accesses(db, category)
def get_all_accesses(
db: Session,
category: str = None,
tenant_id: Optional[uuid.UUID] = None,
) -> List[AccessResponse]:
return AccessService.get_all_accesses(db, category, tenant_id=tenant_id)
+4 -2
View File
@@ -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"}
+10 -4
View File
@@ -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
)
)
+2 -2
View File
@@ -7,8 +7,8 @@ from typing import List, Optional
class TenantController:
@staticmethod
def create_tenant(db: Session, tenant_data: TenantCreate):
return TenantService.create_tenant(db, tenant_data)
def create_tenant(db: Session, tenant_data: TenantCreate, commit: bool = True):
return TenantService.create_tenant(db, tenant_data, commit=commit)
@staticmethod
def get_all_tenants(db: Session):
+39
View File
@@ -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":
+8
View File
@@ -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}>"
+5
View File
@@ -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}>"
+3 -1
View File
@@ -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
)
+3 -14
View File
@@ -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(
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]
)
return SubscriptionPlanService.serialize_plan_response(db, plan, detail=True)
@router.get("/list", response_model=SubscriptionPlanPaginatedResponse)
def list_plans(
+38 -14
View File
@@ -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(
db=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(current_user.id),
performed_by_email=current_user.email,
ip_address=get_client_ip(request),
new_values=tenant_data.model_dump(mode='json')
)
return result
# 2. Sanitize payload for audit log (never store raw passwords or hashes)
audit_values = tenant_data.model_dump(mode='json')
if "owner" in audit_values and isinstance(audit_values["owner"], dict):
audit_values["owner"]["password"] = "[PROTECTED]"
# 3. Create sanitized audit record using flush (strict=True)
AuditLogService.log(
db=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(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(
+4 -3
View File
@@ -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
@@ -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]
+12 -3
View File
@@ -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)
+30 -8
View File
@@ -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]:
cache_key = f"saas:access:v2:all:{category if category else 'full'}"
cached_data = sync_redis_client.client.get(cache_key) if sync_redis_client.client else None
def get_all_accesses(db: Session, category: str = None, tenant_id: Optional[uuid.UUID] = None) -> List[any]:
cache_key = f"saas:access:v4:active:{category if category else 'full'}:{str(tenant_id) if tenant_id else 'global'}"
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)
+20
View File
@@ -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:
+60 -27
View File
@@ -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).
Scopes delivery to relevant modules based on tenant_id.
Emits an event by writing it to the Outbox (event_logs) and flushing.
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'}")
return
logger.warning(f"Event {event_type} (id={event_id}) emitted with no resolved targets.")
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()
try:
sync_redis_client.rpush("saas:events:queue", event_id)
except Exception as e:
logger.error(f"Failed to push event to Redis queue: {e}")
db.flush()
return event_id
@staticmethod
def enqueue_wakeups(event_ids: List[str]) -> None:
"""
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)
secret = env.trust_credentials.get("hmac_secret")
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}. Payload: {payload_json}")
response = httpx.post(log.target_url, content=payload_json, headers=headers, timeout=5)
logger.info(f"Sending event {log.event_type} (id={log.event_id}) to {log.target_url}")
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
@@ -73,9 +73,13 @@ class ModuleEnvironmentService:
for key, value in update_data.items():
setattr(environment, key, value)
db.commit()
db.refresh(environment)
return environment
try:
db.commit()
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):
+24 -25
View File
@@ -52,23 +52,20 @@ class ModulePermissionService:
secret = env.trust_credentials.get("hmac_secret")
if not secret:
pass
if secret:
signature = hmac.new(
secret.encode("utf-8"),
payload_body.encode("utf-8"),
hashlib.sha256
).hexdigest()
else:
signature = ""
raise ValueError("Module environment is missing its HMAC trust secret")
signature = hmac.new(
secret.encode("utf-8"),
payload_body.encode("utf-8"),
hashlib.sha256
).hexdigest()
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(
ModuleAccess.module_id == module.id,
ModuleAccess.access_code == code
).first()
access = permission_map.get(code)
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()
+26 -3
View File
@@ -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()
+56 -8
View File
@@ -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()
db.refresh(role)
if commit:
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,
)
)
+37 -29
View File
@@ -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(
SubscriptionEntitlementService.get_plan_module_access_codes(
db, tenant_id, 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
}
)
# SaaS controls identity and module entry. Operational authorization is
# deliberately owned by the destination module.
permissions = []
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(
db,
uuid.UUID(grant_tenant_id) if grant_tenant_id else None,
module.id,
SubscriptionEntitlementService.get_effective_module_access_codes(
db, user, 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 []
}
@@ -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
+114 -3
View File
@@ -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)
+212 -35
View File
@@ -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()
mod_access_ids = [pma.module_access_id for pma in plan_module_accesses]
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)
# 2. Create TenantModule assignments for selected modules only
for mod_id in effective_selected_module_ids:
env_slug = module_env_map[mod_id]
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:
logger.info(f"Creating tenant {tenant.tenant_name}. Processing {len(event_targets)} event targets.")
emitted_event_ids = []
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
+17 -6
View File
@@ -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
the main operation that triggered them.
If commit=True, commits the session. If commit=False, flushes to the active transaction.
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()
logger.error("AuditLogService.log failed: %s", exc, exc_info=True)
if strict:
raise
if commit:
db.rollback()
logger.error("AuditLogService.log failed: %s", exc, exc_info=True)
return None
+2 -1
View File
@@ -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
+414
View File
@@ -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()
+251
View File
@@ -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
+314
View File
@@ -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
+37
View File
@@ -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()
+966
View File
@@ -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()