Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 149 additions & 0 deletions bootstrap_admin.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
#!/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())
3 changes: 3 additions & 0 deletions services/analysis/Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ COPY services/analysis/app /app/app
COPY alembic.ini /app/alembic.ini
COPY alembic /app/alembic
COPY seed_demo.py /app/seed_demo.py
# Production first-admin bootstrap (seed_demo refuses production by design):
# docker compose run --rm analysis python bootstrap_admin.py --org … --email …
COPY bootstrap_admin.py /app/bootstrap_admin.py

# /app/services/shared is on PYTHONPATH for the legacy `from db.base import ...`
# imports used by older analysis modules. /app is on PYTHONPATH for
Expand Down
36 changes: 36 additions & 0 deletions services/shared/tests/test_bootstrap_admin.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
"""bootstrap_admin.create_admin — the production first-admin path.

seed_demo refuses ENVIRONMENT=production, so this command is the ONLY way
to create the first admin in a prod database. Pin its safety properties:
creates org+admin, refuses a second admin, validates inputs.
"""

import pytest

from bootstrap_admin import create_admin
from services.shared.db.models import Organization, User, UserRole


def test_creates_org_and_admin(db_session):
summary = create_admin(db_session, "Prod Fab", "boss@prodfab.example", "a-long-password-123")
assert summary["created_org"] is True
org = db_session.query(Organization).filter_by(name="Prod Fab").one()
user = db_session.query(User).filter_by(organization_id=org.id).one()
assert user.role == UserRole.ADMIN
assert user.email == "boss@prodfab.example"
assert user.password_hash and user.password_hash != "a-long-password-123"


def test_refuses_second_admin(db_session):
create_admin(db_session, "Prod Fab", "boss@prodfab.example", "a-long-password-123")
with pytest.raises(ValueError, match="already has an admin"):
create_admin(db_session, "Prod Fab", "other@prodfab.example", "another-long-pass-456")


def test_rejects_bad_inputs(db_session):
with pytest.raises(ValueError, match="valid email"):
create_admin(db_session, "X Fab", "not-an-email", "a-long-password-123")
with pytest.raises(ValueError, match="at least 12"):
create_admin(db_session, "X Fab", "a@b.co", "short")
with pytest.raises(ValueError, match="not be empty"):
create_admin(db_session, " ", "a@b.co", "a-long-password-123")
Loading