feat: CLI to bootstrap first super_admin (no UI for the very first admin)

A fresh deployment has no user at all and login itself needs one, and there
was no seed/bootstrap path for the first super_admin. Add
scripts/bootstrap_super_admin.py that creates/ensures an organization, user,
and super_admin membership via the same IdentityRepository/PasswordService the
app uses.

- Password hashed with argon2 (PasswordService), min 12 chars; never printed.
- Idempotent: an existing user's password is left unchanged; re-run only
  ensures the super_admin membership exists.
- Verified locally: created org/user/membership, then /api/auth/login with the
  bootstrap credentials returned success:true role=super_admin; wrong password
  returned invalid_credentials. Re-run left the password unchanged.
Run in container console with ADMIN_EMAIL/ADMIN_PASSWORD/ADMIN_ORG_SLUG env.
This commit is contained in:
Kunthawat Greethong
2026-09-01 13:08:24 +07:00
parent 1015cafe5f
commit 211b1a6545

View File

@@ -0,0 +1,104 @@
"""Bootstrap the first super_admin user for a fresh CrowdSight deployment.
There is no UI to create the very first admin (login itself needs a user), so
this CLI creates/updates a super_admin user + organization directly against
the durable database through the same services the app uses.
Run inside the container console (or locally with the same DATABASE_URL):
env DATABASE_URL=<...> SECRET_KEY=<...> \
ADMIN_EMAIL='admin@example.com' ADMIN_PASSWORD='...12+ chars...' \
ADMIN_ORG_NAME='Acme' ADMIN_ORG_SLUG='acme' \
uv run python scripts/bootstrap_super_admin.py
If the user already exists it is NOT given a new password; it only ensures a
super_admin membership exists. Password must be at least 12 characters (the
app's PasswordService enforces this) and is hashed with argon2; it is never
printed or stored in plaintext.
"""
import argparse
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from app.db import create_database_engine, create_session_factory # noqa: E402
from app.models.saas import Membership, Organization, User # noqa: E402
from app.security.policy import Role # noqa: E402
from app.services.identity import IdentityRepository, PasswordService # noqa: E402
# The migration's server default for password_hash. Means "not set yet".
INVITE_PENDING = "!invite_pending"
def main() -> int:
parser = argparse.ArgumentParser(description="Bootstrap the first super_admin")
parser.add_argument("--email", default=os.environ.get("ADMIN_EMAIL", ""))
parser.add_argument("--password", default=os.environ.get("ADMIN_PASSWORD", ""))
parser.add_argument("--org", default=os.environ.get("ADMIN_ORG_NAME", "Acme"))
parser.add_argument("--slug", default=os.environ.get("ADMIN_ORG_SLUG", ""))
args = parser.parse_args()
email = (args.email or "").strip()
password = args.password or ""
if not email:
print("ERROR: ADMIN_EMAIL (--email) is required", file=sys.stderr)
return 1
if not password:
print("ERROR: ADMIN_PASSWORD (--password) is required", file=sys.stderr)
return 1
if len(password) < 12:
print("ERROR: password must be at least 12 characters", file=sys.stderr)
return 1
engine = create_database_engine()
factory = create_session_factory(engine)
try:
with factory() as session:
repo = IdentityRepository(session)
# Find or create the organization by slug.
slug = (args.slug.strip().lower() or email.split("@")[0])
org = session.query(Organization).filter_by(slug=slug).first()
if org is None:
org = repo.create_organization(name=args.org.strip() or "Acme", slug=slug)
print(f"created organization '{org.name}' (slug={org.slug})")
# Find or create the user.
normalized = repo.normalize_email(email)
user = session.query(User).filter_by(email_normalized=normalized).first()
created_user = user is None
if created_user:
user = repo.create_user(email=normalized)
# Only set the password for a brand-new or invite-pending user;
# never overwrite an existing, already-set password.
if created_user or user.password_hash == INVITE_PENDING:
user.password_hash = PasswordService.hash_password(password)
print(f"password set for {normalized}")
else:
print(f"user {normalized} already has a password; left unchanged")
# Ensure a super_admin membership exists.
membership = (
session.query(Membership)
.filter_by(user_id=user.id, organization_id=org.id)
.first()
)
if membership is None:
repo.create_membership(user.id, org.id, Role.SUPER_ADMIN)
print("created super_admin membership")
elif membership.role.value != Role.SUPER_ADMIN.value:
membership.role = Role.SUPER_ADMIN
print("promoted membership to super_admin")
session.commit()
print(f"DONE: {normalized} is super_admin on org '{org.slug}'")
return 0
finally:
engine.dispose()
if __name__ == "__main__":
raise SystemExit(main())