superadmin / scripts /bootstrap_admin.py
suhailshafi007's picture
feat(backend): centralized environment configuration, Razorpay standard web checkout, Google Play keyless auth & security hardening
bbd07c7
Raw History Blame Contribute Delete
3.79 kB
#!/usr/bin/env python3
"""
Sahih Islam Administrative Bootstrap Script
===========================================
Securely bootstraps or promotes the initial system administrator directly via database access.
Used during infrastructure provisioning or initial production deployment.
Usage:
python bootstrap_admin.py --uid <firebase_uid>
python bootstrap_admin.py --email <user_email>
"""
import sys
import os
import argparse
import asyncio
from datetime import datetime
# Add BACKEND directory to sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.core.config import settings
from motor.motor_asyncio import AsyncIOMotorClient
async def bootstrap_admin_cli(uid: str = None, email: str = None):
if not uid and not email:
print("❌ Error: Either --uid or --email must be provided.")
sys.exit(1)
print("==================================================")
print(" Sahih Islam Administrator Bootstrap Utility")
print(f" Environment : {settings.APP_ENV}")
print(f" Target DB : {settings.MONGO_DB_NAME}")
print("==================================================")
client = AsyncIOMotorClient(settings.MONGO_URI)
db = client[settings.MONGO_DB_NAME]
users_col = db["users"]
lock_col = db["admin_bootstrap_lock"]
# 1. Locate the target user
query = {"uid": uid} if uid else {"email": email.strip().lower()}
user = await users_col.find_one(query)
if not user:
target_str = f"UID='{uid}'" if uid else f"Email='{email}'"
print(f"❌ User not found in database with {target_str}.")
print(" Ensure the user has completed registration or first login before running bootstrap.")
sys.exit(1)
target_uid = user.get("uid")
target_email = user.get("email", "N/A")
current_role = user.get("role", "USER")
print(f"👤 Found User: UID={target_uid}, Email={target_email}, Current Role={current_role}")
if current_role == "ADMIN" and user.get("is_admin") is True:
print("ℹ️ User is already an active administrator.")
sys.exit(0)
# 2. Acquire or update singleton lock
now = datetime.utcnow()
try:
await lock_col.update_one(
{"_id": "admin_singleton_lock"},
{
"$set": {
"bootstrapped_by_uid": target_uid,
"bootstrapped_via": "CLI_BOOTSTRAP",
"updated_at": now
},
"$setOnInsert": {
"created_at": now
}
},
upsert=True
)
except Exception as e:
print(f"⚠️ Warning: Could not update admin_singleton_lock: {e}")
# 3. Promote user atomically
await users_col.update_one(
{"uid": target_uid},
{
"$set": {
"role": "ADMIN",
"is_admin": True,
"updated_at": now
}
}
)
print("==================================================")
print("✅ SUCCESS: Administrator successfully bootstrapped!")
print(f" UID : {target_uid}")
print(f" Email : {target_email}")
print(" Role : ADMIN")
print(" Timestamp: " + now.isoformat())
print("==================================================")
def main():
parser = argparse.ArgumentParser(description="Bootstrap initial Sahih Islam administrator.")
parser.add_argument("--uid", type=str, help="Firebase UID of the user to promote to ADMIN")
parser.add_argument("--email", type=str, help="Email of the user to promote to ADMIN")
args = parser.parse_args()
asyncio.run(bootstrap_admin_cli(uid=args.uid, email=args.email))
if __name__ == "__main__":
main()