nexgen_mirrors
f16bb88ad2
Add a configurable job runner script that executes Flask CLI commands at specified intervals within Docker containers. Supports graceful shutdown via SIGTERM/SIGINT signals and includes built-in job commands for ZeroTier reconciliation and MFA compliance checks.
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""Seed script for creating superadmin user.
|
|
|
|
This script reads SUPERADMIN_EMAIL and SUPERADMIN_SECRET environment variables.
|
|
If both are present:
|
|
- Creates a superadmin with the given email and hashed credential
|
|
- Idempotent: updates existing superadmin if email already exists
|
|
If either is absent:
|
|
- No-op (logs that env vars are not set)
|
|
|
|
Usage:
|
|
export SUPERADMIN_EMAIL="admin@example.com"
|
|
export SUPERADMIN_SECRET="[SetYourSecureSecretHere]"
|
|
python scripts/seed_superadmin.py
|
|
"""
|
|
import sys
|
|
import os
|
|
import logging
|
|
from dotenv import load_dotenv
|
|
|
|
# Load environment variables FIRST before any app imports
|
|
load_dotenv()
|
|
|
|
from gatehouse_app import create_app
|
|
from gatehouse_app.extensions import db, bcrypt
|
|
from gatehouse_app.models.superadmin import Superadmin
|
|
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
app = create_app()
|
|
|
|
with app.app_context():
|
|
# Read environment variables
|
|
email = os.environ.get("SUPERADMIN_EMAIL")
|
|
credential = os.environ.get("SUPERADMIN_SECRET")
|
|
|
|
# Check if env vars are set
|
|
if not email or not credential:
|
|
logger.info("SUPERADMIN_EMAIL and/or SUPERADMIN_SECRET not set. No-op.")
|
|
logger.info("To create a superadmin, set both environment variables:")
|
|
logger.info(" export SUPERADMIN_EMAIL='admin@example.com'")
|
|
logger.info(" export SUPERADMIN_SECRET='[SetYourSecureSecretHere]'")
|
|
return
|
|
|
|
# Normalize email
|
|
email = email.lower().strip()
|
|
|
|
logger.info(f"Processing superadmin: {email}")
|
|
|
|
# Check if superadmin already exists
|
|
existing = Superadmin.query.filter_by(email=email).first()
|
|
|
|
if existing:
|
|
# Update existing superadmin
|
|
logger.info(" → Existing superadmin found, updating credential")
|
|
existing.password_hash = bcrypt.generate_password_hash(credential).decode("utf-8")
|
|
existing.is_active = True
|
|
existing.full_name = existing.full_name or "Super Admin"
|
|
db.session.commit()
|
|
logger.info(f" → Updated superadmin: {email} (id={existing.id})")
|
|
print(f"Updated existing superadmin: {email}")
|
|
else:
|
|
# Create new superadmin
|
|
logger.info(" → Creating new superadmin")
|
|
password_hash = bcrypt.generate_password_hash(credential).decode("utf-8")
|
|
superadmin = Superadmin(
|
|
email=email,
|
|
password_hash=password_hash,
|
|
full_name="Super Admin",
|
|
is_active=True,
|
|
)
|
|
db.session.add(superadmin)
|
|
db.session.commit()
|
|
logger.info(f" → Created superadmin: {email} (id={superadmin.id})")
|
|
print(f"Created new superadmin: {email}")
|
|
|
|
logger.info("Seed complete.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|