""" Seed a vertical's roles (backend//roles.json) into a tenant's RBAC tables — one role + one test user per persona, permissions scoped to that persona's "Primary Modules" column. Super Admin and Admin ARE created here, from the vertical's roles.json: Super Admin — the "*" wildcard the RBAC layer already understands, rather than an enumerated list that goes stale as permissions are added Admin — every permission any role in this vertical holds, plus the platform-management ones (users, settings, modules, security) Because these roles now live here, provisioning must NOT also be given an admin_email: it would create a second super account under a different address (it upserts by email, and the addresses differ), leaving two "super" identities with separately-set passwords — exactly the divergence this file used to avoid by creating none at all. Idempotent: re-running upserts each role's permission set AND resets each user's password to the current value of its tier, so this is safe to run again — and re-running is how a changed password is applied to accounts that already exist. Usage: python seed_documented_personas.py [tenant-slug] [vertical] # tenant-slug default: demo-enterprise ; vertical default: private (private|government) """ import asyncio import json import os from pathlib import Path import sys from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from ams.core.config import settings from ams.services import rbac_admin as svc # Passwords, in three tiers rather than one shared secret. # # One password across every account means the weakest holder of it can sign in # as Super Admin. Splitting the two privileged roles out keeps a compromise of a # role account — of which there are twenty — away from full system access. # # SEED_SUPERADMIN_PASSWORD Super Admin only (all 170 permissions) # SEED_ADMIN_PASSWORD Admin only # SEED_USER_PASSWORD every other role # # Each falls back to SEED_USER_PASSWORD, then to the default below. Those # defaults are published in this repository and identical on every deployment: # bootstrap credentials, to be changed at first login on any server someone else # can reach. They are documented in /README.md. _DEFAULTS = { "SEED_SUPERADMIN_PASSWORD": "Superadmin@1234", "SEED_ADMIN_PASSWORD": "Admin@1234", "SEED_USER_PASSWORD": "Demo@1234", } # The longest password the login form accepts, from its own schema in # ams-frontend/src/features/auth/pages/AuthScreens.tsx. Enforced here because a # longer one seeds an account nobody can ever sign into: the form rejects the # password client-side, so the request is never made and the failure reads as a # validation bug rather than a bad seed. A generated 40-character secret hit # exactly this. MAX_LOGIN_PASSWORD = 32 def _tier(key: str) -> str: """This tier's password: its own variable, then SEED_USER_PASSWORD, then the documented default.""" shared = "" if key == "SEED_USER_PASSWORD" else os.environ.get("SEED_USER_PASSWORD", "") value = os.environ.get(key) or shared or _DEFAULTS[key] if len(value) > MAX_LOGIN_PASSWORD: raise SystemExit( f" ABORTING — {key} is {len(value)} characters.\n" f" The login form accepts at most {MAX_LOGIN_PASSWORD}, so every account seeded\n" f" with it would be impossible to sign into. Nothing was changed.\n" f" Shorten it in .env and run again." ) return value DEFAULT_PASSWORD = _tier("SEED_USER_PASSWORD") SUPERADMIN_PASSWORD = _tier("SEED_SUPERADMIN_PASSWORD") ADMIN_PASSWORD = _tier("SEED_ADMIN_PASSWORD") def _password_for(role_name: str) -> str: """Which of the three tiers this role's user gets.""" key = " ".join(role_name.strip().lower().split()) if key == "super admin": return SUPERADMIN_PASSWORD if key == "admin": return ADMIN_PASSWORD return DEFAULT_PASSWORD # name -> (modules comment, permissions). Permissions verified against the # live routePermissions.ts / backend PERMISSION_CATALOG — see # documentation/rbac-18-persona-audit-report.md for the audit trail behind # each grant, including the post-launch additions (asset:update for Finance # Controller, auditor_dashboard:view + planner_dashboard:view for Executive / # CFO, audit:read + report:view for Document Admin, user:read for HR System # (Integration), asset:tag for Label Designer Admin and Audit Officer, ou:read # + report:view for Asset Manager). def _load_vertical_roles(vertical: str) -> dict[str, list[str]]: """Role -> permissions for a vertical, read from backend//roles.json. The data lives in a folder per vertical rather than in this file so an operator can add a role, or change one permission, without editing Python and without a rebuild — and so `private` and `government` are visibly separate things rather than one dict quietly updating another. government/roles.json already CONTAINS the private set plus the statutory roles, so a vertical is a complete answer on its own. """ path = Path(__file__).parent / vertical / "roles.json" if not path.exists(): raise SystemExit( f"No role data for vertical {vertical!r}: {path} is missing.\n" f"Expected one of: {', '.join(sorted(p.name for p in Path(__file__).parent.glob('*/roles.json')))}" ) return json.loads(path.read_text(encoding="utf-8")) def _slug_email(name: str, tenant_slug: str) -> str: s = name.lower() for ch in ("/", "(", ")", "."): s = s.replace(ch, "") # Was hardcoded to "@demo-enterprise.in" regardless of which tenant this # ran against — every persona seeded into demo-government (or any other # tenant) silently got a demo-enterprise-looking email, so no # asset-manager@demo-government.in-style row was ever produced. return "-".join(s.split()) + f"@{tenant_slug}.in" async def seed_tenant(session: AsyncSession, tenant_slug: str, vertical: str = "private") -> None: row = (await session.execute( text("SELECT id FROM public.tenants WHERE slug = :s"), {"s": tenant_slug} )).fetchone() if not row: print(f" Tenant '{tenant_slug}' not found — skipping") return schema = f"tenant_{str(row.id).replace('-', '_')}" await session.execute(text(f"SET search_path TO {schema}, public")) personas = _load_vertical_roles(vertical) # Check every role's permissions BEFORE writing anything. update_role raises # on the first unknown string, and that exception aborts the whole run — so # one stale name in roles.json used to leave the tenant with some roles, # no users, and a deployment nobody could log into. Better to name all the # bad entries at once and change nothing. from ams.services.rbac_admin import ALL_PERMISSION_NAMES known = set(ALL_PERMISSION_NAMES) # NOT "*": create_role rejects it invalid = {r: sorted(set(p) - known) for r, p in personas.items()} invalid = {r: v for r, v in invalid.items() if v} if invalid: print(f" ABORTING — {vertical}/roles.json has permissions this build does not define:") for role, perms in invalid.items(): print(f" {role}: {', '.join(perms)}") print(" Nothing was changed. Fix those names (see PERMISSION_CATALOG in") print(" ams/services/rbac_admin.py) and run again.") raise SystemExit(1) for name, permissions in personas.items(): existing = (await session.execute( text("SELECT id FROM roles WHERE name = :n"), {"n": name} )).fetchone() if existing: await svc.update_role(session, str(existing.id), permissions=list(permissions)) print(f" role updated: {name}") else: await svc.create_role(session, name, list(permissions)) print(f" role created: {name}") email = _slug_email(name, tenant_slug) existing_user = (await session.execute( text("SELECT id FROM users WHERE email = :e"), {"e": email} )).fetchone() if not existing_user: await svc.create_user( session, name=f"{name} (Test User)", email=email, password=_password_for(name), role=name, department=name, employee_code=None, ) print(f" user created: {email}") else: # Re-running this script must always leave a known-good login — # a user row surviving from an earlier/partial seeding attempt # (with an unknown, unrecoverable password) would otherwise be # silently left untouched forever, unlike the role handling # above which is always refreshed. await svc.update_user( session, str(existing_user.id), password=_password_for(name), role=name, is_active=True, ) print(f" user password reset: {email}") await session.commit() def _available_verticals() -> list[str]: return sorted(d.parent.name for d in Path(__file__).parent.glob("*/roles.json")) def _resolve_vertical() -> str: """Which vertical to seed: CLI arg, then SEED_VERTICAL, then ask. Never guesses. Seeding the wrong vertical gives a government tenant no statutory roles (or a private one four roles it should not have), and the mistake is invisible until someone cannot approve a CAG report. """ choices = _available_verticals() picked = sys.argv[2] if len(sys.argv) > 2 else os.environ.get("SEED_VERTICAL", "").strip() if picked in choices: return picked if picked: raise SystemExit(f"Unknown vertical {picked!r}. Available: {', '.join(choices)}") if not sys.stdin.isatty(): raise SystemExit( "SEED_VERTICAL is not set and there is no terminal to ask on.\n" f"Set SEED_VERTICAL to one of: {', '.join(choices)}" ) print("Which vertical should be seeded?") for i, c in enumerate(choices, 1): print(f" {i}) {c}") while True: answer = input(f"Vertical [{choices[0]}]: ").strip() or choices[0] if answer in choices: return answer if answer.isdigit() and 1 <= int(answer) <= len(choices): return choices[int(answer) - 1] print(f" Pick one of: {', '.join(choices)}") def _database_url_for(tenant_slug: str, vertical: str) -> str: """Which physical database this tenant's schema lives in. TENANT_DB_MAP (JSON: slug -> alias) is the deployment's own answer and wins. Otherwise a government tenant defaults to the dedicated government database and everything else to the shared one — the same aliases ams/core/database.py resolves at request time, so seeding and serving can never disagree about where a tenant's data is. """ aliases = {"shared": settings.DATABASE_URL, "government-dedicated": settings.GOVERNMENT_DB_URL} aliases.update(json.loads(os.environ.get("EXTRA_DB_ALIASES", "{}") or "{}")) mapping = json.loads(os.environ.get("TENANT_DB_MAP", "{}") or "{}") alias = mapping.get(tenant_slug) or ("government-dedicated" if vertical == "government" else "shared") if alias not in aliases: raise SystemExit( f"Tenant {tenant_slug!r} maps to database alias {alias!r}, which is not defined.\n" f"Known aliases: {', '.join(sorted(aliases))}. Add it to EXTRA_DB_ALIASES." ) return aliases[alias] async def main() -> None: tenant_slug = sys.argv[1] if len(sys.argv) > 1 else os.environ.get("SEED_TENANT_SLUG", "demo-enterprise") vertical = _resolve_vertical() db_url = _database_url_for(tenant_slug, vertical) # Say where it is going before it goes there — seeding the wrong database is # the expensive mistake, and the host is the part an operator can check. host = db_url.split("@")[-1] print(f"Seeding '{vertical}' personas into tenant '{tenant_slug}' on {host}") missing = [k for k in _DEFAULTS if not os.environ.get(k)] if missing: print(f" NOTE: {', '.join(missing)} not set.") if os.environ.get("SEED_USER_PASSWORD"): print(" Those tiers share SEED_USER_PASSWORD.") else: print(f" Using the defaults documented in {vertical}/README.md. They are") print(" published in this repository — change them at first login on any") print(" server other people can reach.") engine = create_async_engine(db_url, echo=False) Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) try: async with Session() as session: await seed_tenant(session, tenant_slug, vertical=vertical) finally: await engine.dispose() print("Done.") if __name__ == "__main__": asyncio.run(main())