leocrm/scripts/seed_admin.py

#!/usr/bin/env python3
"""Seed a default tenant and admin user for LeoCRM.

Usage: python scripts/seed_admin.py

Creates:
  - Tenant: "Default Org" (slug: default)
  - Admin role with full permissions
  - Admin user: admin@media-on.de / Admin123!
  - UserTenant link with admin role

If tenant or user already exists, skips creation.

Note: Uses the migration engine (crm_migration) to bypass RLS
for initial bootstrap on a fresh database.
"""

import asyncio
import os
import sys

# Ensure app is importable
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker

from app.core.auth import hash_password
from app.core.db import get_migration_engine, set_tenant_context
from app.models.role import Role
from app.models.tenant import Tenant
from app.models.user import User, UserTenant


async def seed():
    # Use migration engine to bypass RLS for bootstrap
    engine = get_migration_engine()
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as db:
        # Check if default tenant exists
        result = await db.execute(select(Tenant).where(Tenant.slug == "default"))
        tenant = result.scalar_one_or_none()

        if tenant is None:
            tenant = Tenant(name="Default Org", slug="default")
            db.add(tenant)
            await db.flush()
            print(f"Created tenant: {tenant.name} (slug: {tenant.slug}, id: {tenant.id})")
        else:
            print(f"Tenant exists: {tenant.name} (id: {tenant.id})")

        # Set tenant context for RLS
        await set_tenant_context(db, tenant.id)

        # Create admin role if not exists
        result = await db.execute(select(Role).where(Role.name == "admin", Role.tenant_id == tenant.id))
        role = result.scalar_one_or_none()

        if role is None:
            role = Role(
                tenant_id=tenant.id,
                name="admin",
                permissions={"*:*": True},
            )
            db.add(role)
            await db.flush()
            print(f"Created admin role: {role.id}")
        else:
            print(f"Admin role exists: {role.id}")

        # Check if admin user exists
        result = await db.execute(select(User).where(User.email == os.environ.get("ADMIN_EMAIL", "admin@media-on.de")))
        user = result.scalar_one_or_none()

        if user is None:
            # F30 (Astra P1): no known default password on fresh installs.
            # - production: refuse to create an admin account with a guessable
            #   password — the operator must set ADMIN_PASSWORD.
            # - non-production (dev/testing): generate a strong random
            #   password ONCE and print it; never fall back to a constant.
            admin_password = os.environ.get("ADMIN_PASSWORD", "").strip()
            from app.config import get_settings

            if not admin_password:
                if get_settings().environment == "production":
                    print(
                        "ERROR: ADMIN_PASSWORD is not set. Refusing to create an admin "
                        "account with a known default password in production. Set the "
                        "ADMIN_PASSWORD environment variable and restart.",
                        file=sys.stderr,
                    )
                    raise SystemExit(1)
                import secrets

                admin_password = secrets.token_urlsafe(16)
                print(
                    "NOTICE: ADMIN_PASSWORD not set — generated a one-time random "
                    f"password for the new admin account: {admin_password}",
                )
            user = User(
                email=os.environ.get("ADMIN_EMAIL", "admin@media-on.de"),
                name="Administrator",
                password_hash=hash_password(admin_password),
                is_active=True,
                is_system_admin=True,
                preferences={},
            )
            db.add(user)
            await db.flush()
            print(f"Created user: {user.email} (id: {user.id})")

            # Link user to tenant with admin role
            ut = UserTenant(
                user_id=user.id,
                tenant_id=tenant.id,
                role_id=role.id,
                role="admin",
                is_default=True,
            )
            db.add(ut)
            await db.flush()
            print("Created user_tenant link with admin role")
        else:
            print(f"User exists: {user.email} (id: {user.id})")
            # Ensure existing admin has is_system_admin=True
            if not user.is_system_admin:
                user.is_system_admin = True
                await db.flush()
                print(f"Updated is_system_admin=True for {user.email}")

        # Seed default workspace if none exists for this tenant
        from app.services.workspace_service import seed_default_workspace
        ws_result = await seed_default_workspace(db, tenant.id, user.id)
        if ws_result:
            print(f"Created default workspace: {ws_result['name']} (id: {ws_result['id']})")
        else:
            print("Workspace already exists — skipping workspace seed")

        await db.commit()
        print("Seed completed successfully.")


if __name__ == "__main__":
    asyncio.run(seed())