281 lines
10 KiB
Python
281 lines
10 KiB
Python
"""
|
|
Creates:
|
|
1. All predefined accesses
|
|
2. Superadmin role with all accesses
|
|
3. Superadmin user with the role
|
|
|
|
Usage:
|
|
python scripts/seed_superadmin.py
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
backend_dir = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(backend_dir))
|
|
|
|
from sqlalchemy.orm import Session
|
|
from app.config.database import SessionLocal
|
|
from app.config.settings import settings
|
|
from app.config.security import security
|
|
from app.models.auth.user_model import User
|
|
from app.models.auth.access_model import Access
|
|
from app.models.auth.role_model import Role
|
|
from app.models.auth.role_access_model import RoleAccess
|
|
|
|
# Predefined accesses (access_code, category, name, parent_code)
|
|
PREDEFINED_ACCESSES = [
|
|
# Superadmin category
|
|
("superadmin.main.view", "Superadmin", "Allow access to superadmin view", None),
|
|
("superadmin.tenant.create", "Superadmin", "Allow access to create tenants", None),
|
|
("superadmin.tenant.read", "Superadmin", "Allow access to view all tenants", None),
|
|
("superadmin.tenant.update", "Superadmin", "Allow access to update tenants", None),
|
|
("superadmin.tenant.delete", "Superadmin", "Allow access to delete tenants", None),
|
|
("superadmin.role.create", "Superadmin", "Allow access to create roles", None),
|
|
("superadmin.role.read", "Superadmin", "Allow access to view all roles", None),
|
|
("superadmin.role.update", "Superadmin", "Allow access to update any role", None),
|
|
("superadmin.role.delete", "Superadmin", "Allow access to delete any role", None),
|
|
("superadmin.user.create", "Superadmin", "Allow access to create any user", None),
|
|
("superadmin.user.read", "Superadmin", "Allow access to view all users", None),
|
|
("superadmin.user.update", "Superadmin", "Allow access to update any user", None),
|
|
("superadmin.user.delete", "Superadmin", "Allow access to delete any user", None),
|
|
("superadmin.access.read", "Superadmin", "Allow access to view all accesses", None),
|
|
|
|
# Subscription Plans
|
|
("superadmin.plan.create", "Superadmin", "Allow access to create subscription plans", None),
|
|
("superadmin.plan.read", "Superadmin", "Allow access to view subscription plans", None),
|
|
("superadmin.plan.update", "Superadmin", "Allow access to update subscription plans", None),
|
|
("superadmin.plan.delete", "Superadmin", "Allow access to delete subscription plans", None),
|
|
|
|
# Theme/Palette
|
|
("superadmin.palette.read", "Superadmin", "Allow access to view color palettes", None),
|
|
("superadmin.palette.create", "Superadmin", "Allow access to create color palettes", None),
|
|
("superadmin.palette.update", "Superadmin", "Allow access to update color palettes", None),
|
|
("superadmin.palette.delete", "Superadmin", "Allow access to delete color palettes", None),
|
|
|
|
# Module Registry
|
|
("modules.view", "Superadmin", "Allow access to view module registry", None),
|
|
("modules.manage", "Superadmin", "Allow access to manage modules", None),
|
|
("tenants.manage", "Superadmin", "Allow access to manage tenant module assignments", None),
|
|
|
|
# All Accesses hereafter are applicable for a Tenant Admin
|
|
|
|
# Administration category
|
|
("admin.role.create", "Administration", "Allow access to create roles", None),
|
|
("admin.role.read", "Administration", "Allow access to view roles", None),
|
|
("admin.role.update", "Administration", "Allow access to update roles", None),
|
|
("admin.role.delete", "Administration", "Allow access to delete roles", None),
|
|
("admin.user.create", "Administration", "Allow access to create users", None),
|
|
("admin.user.read", "Administration", "Allow access to view users", None),
|
|
("admin.user.update", "Administration", "Allow access to update users", None),
|
|
("admin.user.delete", "Administration", "Allow access to delete users", None),
|
|
("admin.access.read", "Administration", "Allow access to view accesses", None),
|
|
("admin.logs.read", "Administration", "Allow access to view system audit logs", None),
|
|
|
|
]
|
|
|
|
|
|
def seed_accesses(db: Session):
|
|
"""Create all predefined accesses with hierarchy"""
|
|
print("Seeding accesses...")
|
|
created_count = 0
|
|
|
|
# First pass: Create all accesses without parents
|
|
for access_code, category, name, parent_code in PREDEFINED_ACCESSES:
|
|
existing = db.query(Access).filter(Access.access_code == access_code).first()
|
|
if not existing:
|
|
access = Access(
|
|
access_code=access_code, category=category, name=name, parent_id=None
|
|
)
|
|
db.add(access)
|
|
created_count += 1
|
|
|
|
db.commit()
|
|
print(
|
|
f" ✓ Created {created_count} new accesses (total: {len(PREDEFINED_ACCESSES)})"
|
|
)
|
|
|
|
# Second pass: Set parent relationships
|
|
parent_count = 0
|
|
for access_code, category, name, parent_code in PREDEFINED_ACCESSES:
|
|
if parent_code:
|
|
child = db.query(Access).filter(Access.access_code == access_code).first()
|
|
parent = db.query(Access).filter(Access.access_code == parent_code).first()
|
|
|
|
if child and parent:
|
|
if child.parent_id != parent.id:
|
|
child.parent_id = parent.id
|
|
parent_count += 1
|
|
elif child and not parent:
|
|
print(
|
|
f" ⚠ Warning: Parent '{parent_code}' not found for '{access_code}'"
|
|
)
|
|
|
|
db.commit()
|
|
|
|
if parent_count > 0:
|
|
print(f" ✓ Set {parent_count} parent relationships")
|
|
|
|
|
|
def seed_superadmin_role(db: Session) -> Role:
|
|
"""Create superadmin role with ALL accesses"""
|
|
print("Creating superadmin role...")
|
|
|
|
role = (
|
|
db.query(Role)
|
|
.filter(Role.role_name == "superadmin", Role.tenant_id == None)
|
|
.first()
|
|
)
|
|
|
|
if not role:
|
|
role = Role(role_name="superadmin", tenant_id=None)
|
|
db.add(role)
|
|
db.commit()
|
|
db.refresh(role)
|
|
print(" ✓ Superadmin role created")
|
|
else:
|
|
print(" ✓ Superadmin role already exists")
|
|
|
|
all_accesses = db.query(Access).all()
|
|
existing_access_ids = {ra.access_id for ra in role.role_accesses}
|
|
|
|
new_accesses_count = 0
|
|
for access in all_accesses:
|
|
if access.id not in existing_access_ids:
|
|
role_access = RoleAccess(role_id=role.id, access_id=access.id)
|
|
db.add(role_access)
|
|
new_accesses_count += 1
|
|
|
|
db.commit()
|
|
|
|
if new_accesses_count > 0:
|
|
print(f" ✓ Added {new_accesses_count} accesses to superadmin role")
|
|
|
|
print(f" ✓ Superadmin role has {len(all_accesses)} total accesses")
|
|
|
|
return role
|
|
|
|
|
|
def create_superadmin_user(db: Session, role: Role) -> bool:
|
|
print("Creating superadmin user...")
|
|
|
|
existing_superadmin = (
|
|
db.query(User).filter(User.email == settings.SUPER_ADMIN_EMAIL).first()
|
|
)
|
|
|
|
if existing_superadmin:
|
|
print(f" ✓ Superadmin already exists")
|
|
print(f" Email: {existing_superadmin.email}")
|
|
print(
|
|
f" Role: {existing_superadmin.role.role_name if existing_superadmin.role else 'None'}"
|
|
)
|
|
print(
|
|
f" Tenant: {'None (Superadmin)' if existing_superadmin.tenant_id is None else existing_superadmin.tenant_id}"
|
|
)
|
|
|
|
if not existing_superadmin.role_id or existing_superadmin.role_id != role.id:
|
|
existing_superadmin.role_id = role.id
|
|
db.commit()
|
|
print(" ✓ Updated superadmin role")
|
|
|
|
return False
|
|
|
|
if not security.validate_password_strength(settings.SUPER_ADMIN_PASSWORD):
|
|
print(" ✗ Error: Superadmin password does not meet strength requirements")
|
|
print(
|
|
" Password must be 8+ characters with uppercase, lowercase, digit, and special character"
|
|
)
|
|
sys.exit(1)
|
|
|
|
hashed_password = security.hash_password(settings.SUPER_ADMIN_PASSWORD)
|
|
|
|
superadmin = User(
|
|
email=settings.SUPER_ADMIN_EMAIL,
|
|
password=hashed_password,
|
|
first_name=settings.SUPER_ADMIN_FIRST_NAME,
|
|
last_name=settings.SUPER_ADMIN_LAST_NAME,
|
|
tenant_id=None,
|
|
role_id=role.id,
|
|
status="active",
|
|
)
|
|
|
|
db.add(superadmin)
|
|
db.commit()
|
|
db.refresh(superadmin)
|
|
|
|
print(f" ✓ Superadmin created successfully!")
|
|
print(f" Email: {superadmin.email}")
|
|
print(f" Name: {superadmin.first_name} {superadmin.last_name}")
|
|
print(f" Role: {superadmin.role.role_name}")
|
|
print(f" Accesses: {len(superadmin.role.role_accesses)}")
|
|
print(f" Tenant: None (Superadmin)")
|
|
print(f" Status: {superadmin.status}")
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
"""Main execution function."""
|
|
print("=" * 60)
|
|
print("Superadmin Seed Script with RBAC")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
required_vars = [
|
|
"SUPER_ADMIN_EMAIL",
|
|
"SUPER_ADMIN_PASSWORD",
|
|
"SUPER_ADMIN_FIRST_NAME",
|
|
"SUPER_ADMIN_LAST_NAME",
|
|
]
|
|
|
|
missing_vars = []
|
|
for var in required_vars:
|
|
if not getattr(settings, var, None):
|
|
missing_vars.append(var)
|
|
|
|
if missing_vars:
|
|
print("✗ Error: Missing required environment variables:")
|
|
for var in missing_vars:
|
|
print(f" - {var}")
|
|
print()
|
|
print("Please add these to your .env file and try again.")
|
|
sys.exit(1)
|
|
|
|
db = SessionLocal()
|
|
|
|
try:
|
|
seed_accesses(db)
|
|
print()
|
|
|
|
superadmin_role = seed_superadmin_role(db)
|
|
print()
|
|
|
|
created = create_superadmin_user(db, superadmin_role)
|
|
|
|
print()
|
|
print("=" * 60)
|
|
if created:
|
|
print("Setup complete! You can now login with:")
|
|
print(f" Email: {settings.SUPER_ADMIN_EMAIL}")
|
|
print(f" Password: {settings.SUPER_ADMIN_PASSWORD}")
|
|
else:
|
|
print("Seed script completed successfully!")
|
|
print("All accesses and roles are up to date.")
|
|
print("=" * 60)
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
print()
|
|
print("=" * 60)
|
|
print(f"✗ Failed to seed data: {str(e)}")
|
|
print("=" * 60)
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
main() |