97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
# #region CreateAdminScript [C:5] [TYPE Module] [SEMANTICS admin, search, user, cli]
|
|
#
|
|
# @BRIEF CLI tool for creating the initial admin user.
|
|
# @LAYER Infrastructure
|
|
# @RELATION CALLS -> [AuthSecurityModule]
|
|
# @RELATION CALLS -> [DatabaseModule]
|
|
# @RELATION CALLS -> [AuthModels]
|
|
#
|
|
# @INVARIANT Admin user must have the "Admin" role.
|
|
# @SIDE_EFFECT Writes admin user to database
|
|
# @DATA_CONTRACT CLIArgs -> AdminUser
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
# Add src to path
|
|
sys.path.append(str(Path(__file__).parent.parent.parent))
|
|
|
|
from src.core.auth.security import get_password_hash
|
|
from src.core.cot_logger import seed_trace_id
|
|
from src.core.database import AuthSessionLocal, init_db
|
|
from src.core.logger import belief_scope, logger
|
|
from src.models.auth import Role, User
|
|
|
|
|
|
# #region create_admin [TYPE Function]
|
|
# @BRIEF Creates an admin user and necessary roles/permissions.
|
|
# @PRE username and password provided via CLI.
|
|
# @POST Admin user exists in auth.db.
|
|
#
|
|
def create_admin(username, password, email=None):
|
|
seed_trace_id()
|
|
with belief_scope("create_admin"):
|
|
db = AuthSessionLocal()
|
|
try:
|
|
normalized_email = (
|
|
email.strip() if isinstance(email, str) and email.strip() else None
|
|
)
|
|
|
|
# 1. Ensure Admin role exists
|
|
admin_role = db.query(Role).filter(Role.name == "Admin").first()
|
|
if not admin_role:
|
|
logger.info("Creating Admin role...")
|
|
admin_role = Role(name="Admin", description="System Administrator")
|
|
db.add(admin_role)
|
|
db.commit()
|
|
db.refresh(admin_role)
|
|
|
|
# 2. Check if user already exists
|
|
existing_user = db.query(User).filter(User.username == username).first()
|
|
if existing_user:
|
|
logger.warning(f"User {username} already exists.")
|
|
return "exists"
|
|
|
|
# 3. Create Admin user
|
|
logger.info(f"Creating admin user: {username}")
|
|
new_user = User(
|
|
username=username,
|
|
email=normalized_email,
|
|
password_hash=get_password_hash(password),
|
|
auth_source="LOCAL",
|
|
is_active=True,
|
|
)
|
|
new_user.roles.append(admin_role)
|
|
db.add(new_user)
|
|
db.commit()
|
|
logger.info(f"Admin user {username} created successfully.")
|
|
return "created"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to create admin user: {e}")
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# #endregion create_admin
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Create initial admin user")
|
|
parser.add_argument("--username", required=True, help="Admin username")
|
|
parser.add_argument("--password", required=True, help="Admin password")
|
|
parser.add_argument("--email", required=False, help="Admin email")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
# Ensure DB is initialized before creating admin
|
|
init_db()
|
|
create_admin(args.username, args.password, args.email)
|
|
sys.exit(0)
|
|
except Exception:
|
|
sys.exit(1)
|
|
|
|
# #endregion CreateAdminScript
|