bootstrap_admin.py

#!/usr/bin/env python3
"""Production bootstrap: create the first organization + admin user.

seed_demo.py (rightly) refuses to run with ENVIRONMENT=production because it
plants well-known demo credentials — which left a fresh production database
with NO way to create the first admin. This command closes that gap:

    DATABASE_URL=postgresql+psycopg://... python bootstrap_admin.py \
        --org "Acme Fab" --email admin@acme.example

Behavior:
  - Password comes from BOOTSTRAP_ADMIN_PASSWORD (env) or --password; when
    neither is given a cryptographically random one is GENERATED and printed
    exactly once. No credential is ever hardcoded.
  - Idempotent and refuse-to-clobber: if the org already has ANY admin the
    command exits non-zero without touching anything (bootstrap is for empty
    databases, not credential resets).
  - Safe in every environment, INCLUDING production — that is its purpose.
  - In containers:  docker compose run --rm analysis python bootstrap_admin.py …
    (the analysis image bakes this file alongside seed_demo.py).
"""

from __future__ import annotations

import argparse
import os
import re
import secrets
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from sqlalchemy import text
from sqlalchemy.orm import Session

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
MIN_PASSWORD_LEN = 12


def _slugify(name: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
    return slug or "org"


def create_admin(db: Session, org_name: str, email: str, password: str) -> dict:
    """Create (or reuse) the organization and create its first admin.

    Raises ValueError on invalid input or if an admin already exists in the
    target org. Returns a summary dict (no password included).
    """
    from services.shared.auth.jwt import hash_password
    from services.shared.db.models import Organization, User, UserRole

    if not EMAIL_RE.match(email):
        raise ValueError(f"'{email}' is not a valid email address")
    if len(password) < MIN_PASSWORD_LEN:
        raise ValueError(f"password must be at least {MIN_PASSWORD_LEN} characters")
    if not org_name.strip():
        raise ValueError("organization name must not be empty")

    org = db.query(Organization).filter(Organization.name == org_name).first()
    created_org = False
    if org is None:
        org = Organization(name=org_name, slug=_slugify(org_name))
        db.add(org)
        db.flush()
        created_org = True

    existing_admin = (
        db.query(User).filter(User.organization_id == org.id, User.role == UserRole.ADMIN).first()
    )
    if existing_admin is not None:
        raise ValueError(
            f"organization '{org_name}' already has an admin ({existing_admin.email}) — "
            "bootstrap only runs against a fresh org. Use the normal user-management "
            "path (or reset the password directly) instead."
        )

    if db.query(User).filter(User.organization_id == org.id, User.email == email).first():
        raise ValueError(f"user {email} already exists in '{org_name}'")

    user = User(
        organization_id=org.id,
        email=email,
        name=email.split("@")[0],
        role=UserRole.ADMIN,
        password_hash=hash_password(password),
        is_active=True,
    )
    db.add(user)
    db.commit()

    return {
        "organization": org_name,
        "organization_id": str(org.id),
        "created_org": created_org,
        "admin_email": email,
        "admin_id": str(user.id),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--org", required=True, help="Organization name")
    parser.add_argument("--email", required=True, help="Admin email address")
    parser.add_argument(
        "--password",
        default=None,
        help="Admin password (min 12 chars). Prefer BOOTSTRAP_ADMIN_PASSWORD "
        "env; omit both to auto-generate.",
    )
    args = parser.parse_args()

    password = args.password or os.getenv("BOOTSTRAP_ADMIN_PASSWORD")
    generated = False
    if not password:
        password = secrets.token_urlsafe(18)
        generated = True

    from services.shared.db.base import SessionLocal

    db = SessionLocal()
    try:
        # Server-controlled bootstrap: opt out of RLS for this connection
        # (same convention as seed_demo.py; required under the NOBYPASSRLS
        # app role).
        db.execute(text("SET spectra.current_org_id = 'admin'"))
        summary = create_admin(db, args.org, args.email, password)
    except ValueError as exc:
        print(f"REFUSED: {exc}", file=sys.stderr)
        return 2
    except Exception as exc:  # noqa: BLE001 — surface the real failure
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    finally:
        db.close()

    print("Bootstrap complete:")
    for k, v in summary.items():
        print(f"  {k}: {v}")
    if generated:
        print("\nGenerated admin password (shown ONCE — store it in your password manager):")
        print(f"  {password}")
    return 0


if __name__ == "__main__":
    sys.exit(main())