create_admin.py

"""
This module contains runtime logic for `create_admin.py`.

In simple terms, this file groups related code so the project can
handle one clear responsibility without mixing unrelated behavior.
"""

import base64
import csv
import os
import re
import secrets
import string
import sys
from datetime import datetime, timezone
from getpass import getpass
from pathlib import Path
from zoneinfo import ZoneInfo

from cryptography.fernet import Fernet, InvalidToken
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from django.db import IntegrityError


TOOL_NAME = "create_admin"
SENTINEL_PLAINTEXT = b"create_admin_tool_unlocked_v1"
LOCK_FILE = Path(__file__).resolve().parent / ".create_admin.lock"
ACCESS_TOKENS_FILE = Path(__file__).resolve().parent / "access_tokens.csv"
MAX_ACCESS_ATTEMPTS = 3
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
TOKENS_HEADERS = ["Name", "email", "access token", "created at", "updated at"]


def setup_django():
    """
    Simple purpose: run `setup_django` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        The updated result, status object, or `None` depending on flow.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "aidra_backend.settings")
    try:
        import django

        django.setup()
    except Exception as exc:
        print(f"Failed to initialize Django: {exc}")
        sys.exit(1)


def _derive_key_from_code(code, salt_b64):
    """
    Simple purpose: run `_derive_key_from_code` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        code: Input value used by this step of the workflow.
        salt_b64: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    salt = base64.urlsafe_b64decode(salt_b64.encode("utf-8"))
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=390000,
    )
    return base64.urlsafe_b64encode(kdf.derive(code.encode("utf-8")))


def _fernet_from_code(code, salt_b64):
    """
    Simple purpose: run `_fernet_from_code` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        code: Input value used by this step of the workflow.
        salt_b64: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    key = _derive_key_from_code(code, salt_b64)
    return Fernet(key)


def _load_tool_lock():
    """
    Simple purpose: run `_load_tool_lock` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from accounts.models import ToolAccessLock

    return ToolAccessLock.objects.filter(tool_name=TOOL_NAME, is_active=True).first()


def _verify_code_against_db(code, tool_lock):
    """
    Simple purpose: run `_verify_code_against_db` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        code: Input value used by this step of the workflow.
        tool_lock: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from django.contrib.auth.hashers import check_password

    if not check_password(code, tool_lock.code_hash):
        return False
    try:
        fernet = _fernet_from_code(code, tool_lock.kdf_salt)
        plaintext = fernet.decrypt(tool_lock.encrypted_sentinel.encode("utf-8"))
        return plaintext == SENTINEL_PLAINTEXT
    except Exception:
        return False


def _verify_or_create_lock_file(code, tool_lock):
    """
    Simple purpose: run `_verify_or_create_lock_file` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        code: Input value used by this step of the workflow.
        tool_lock: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    try:
        fernet = _fernet_from_code(code, tool_lock.kdf_salt)
        if LOCK_FILE.exists():
            encrypted = LOCK_FILE.read_bytes()
            decrypted = fernet.decrypt(encrypted)
            return decrypted == SENTINEL_PLAINTEXT
        LOCK_FILE.write_bytes(fernet.encrypt(SENTINEL_PLAINTEXT))
        return True
    except (InvalidToken, OSError, ValueError):
        return False


def authorize_script_access():
    """
    Simple purpose: run `authorize_script_access` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    print("Admin tool access is protected.")
    tool_lock = _load_tool_lock()
    if not tool_lock:
        print("Tool lock is not configured in database. Run migrations first.")
        return False, None

    for attempt in range(1, MAX_ACCESS_ATTEMPTS + 1):
        provided = getpass("Enter access code: ").strip()
        if _verify_code_against_db(provided, tool_lock) and _verify_or_create_lock_file(provided, tool_lock):
            return True, provided
        remaining = MAX_ACCESS_ATTEMPTS - attempt
        if remaining > 0:
            print(f"Incorrect/invalid access code. Attempts left: {remaining}")
    print("Access denied.")
    return False, None


def generate_access_token(length=32):
    """
    Simple purpose: run `generate_access_token` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        length: Input value used by this step of the workflow.

    Returns:
        A newly created object or generated value.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    alphabet = string.ascii_letters + string.digits
    return "AIDRAx-" + "".join(secrets.choice(alphabet) for _ in range(length))


def _read_access_token_rows():
    """
    Simple purpose: run `_read_access_token_rows` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    if not ACCESS_TOKENS_FILE.exists():
        return []
    try:
        with ACCESS_TOKENS_FILE.open("r", newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            rows = []
            for row in reader:
                rows.append(
                    {
                        "Name": (row.get("Name") or row.get("name") or "").strip(),
                        "email": (row.get("email") or "").strip().lower(),
                        "access token": (
                            row.get("access token")
                            or row.get("email access token")
                            or row.get("access_token")
                            or ""
                        ).strip(),
                        "created at": (
                            row.get("created at")
                            or row.get("created_at")
                            or row.get("created_at_utc")
                            or ""
                        ).strip(),
                        "updated at": (
                            row.get("updated at")
                            or row.get("updated_at_utc")
                            or ""
                        ).strip(),
                    }
                )
            return rows
    except Exception:
        return []


def _write_access_token_rows(rows):
    """
    Simple purpose: run `_write_access_token_rows` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        rows: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    ACCESS_TOKENS_FILE.parent.mkdir(parents=True, exist_ok=True)
    with ACCESS_TOKENS_FILE.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=TOKENS_HEADERS)
        writer.writeheader()
        for row in rows:
            writer.writerow(
                {
                    "Name": (row.get("Name") or "").strip(),
                    "email": (row.get("email") or "").strip().lower(),
                    "access token": (row.get("access token") or "").strip(),
                    "created at": (row.get("created at") or "").strip(),
                    "updated at": (row.get("updated at") or "").strip(),
                }
            )


def _ist_time_label(dt=None):
    """
    Simple purpose: run `_ist_time_label` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        dt: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    if dt is None:
        dt = datetime.now(timezone.utc)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(ZoneInfo("Asia/Kolkata")).strftime("%H:%M:%S IST")


def upsert_access_token_csv(name, email, access_token, created_at=None):
    """
    Simple purpose: run `upsert_access_token_csv` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        name: Input value used by this step of the workflow.
        email: Input value used by this step of the workflow.
        access_token: Input value used by this step of the workflow.
        created_at: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    email = (email or "").strip().lower()
    if not email:
        return
    rows = _read_access_token_rows()
    now_ist = _ist_time_label()
    created_at_ist = _ist_time_label(created_at) if created_at else now_ist
    updated = False
    for row in rows:
        if (row.get("email") or "").strip().lower() == email:
            row["Name"] = (name or row.get("Name") or "").strip()
            row["access token"] = access_token
            row["created at"] = (row.get("created at") or created_at_ist).strip()
            row["updated at"] = now_ist
            updated = True
            break
    if not updated:
        rows.append(
            {
                "Name": (name or "").strip(),
                "email": email,
                "access token": access_token,
                "created at": created_at_ist,
                "updated at": now_ist,
            }
        )
    _write_access_token_rows(rows)


def remove_access_token_csv(email):
    """
    Simple purpose: run `remove_access_token_csv` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        email: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    email = (email or "").strip().lower()
    if not email:
        return
    rows = _read_access_token_rows()
    rows = [r for r in rows if (r.get("email") or "").strip().lower() != email]
    _write_access_token_rows(rows)


def sync_access_tokens_csv_with_db():
    """
    Simple purpose: run `sync_access_tokens_csv_with_db` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        The updated result, status object, or `None` depending on flow.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from accounts.models import User

    existing_rows_by_email = {
        (row.get("email") or "").strip().lower(): row
        for row in _read_access_token_rows()
        if (row.get("email") or "").strip()
    }

    now_ist = _ist_time_label()
    synced_rows = []
    admins = User.objects.filter(role="admin").order_by("created_at")
    for admin in admins:
        email = (admin.email or "").strip().lower()
        if not email:
            continue
        existing = existing_rows_by_email.get(email, {})
        display_name = (
            f"{(admin.first_name or '').strip()} {(admin.last_name or '').strip()}".strip()
            or email
        )
        synced_rows.append(
            {
                "Name": display_name,
                "email": email,
                "access token": (admin.admin_access_token or "").strip(),
                "created at": _ist_time_label(admin.created_at),
                "updated at": now_ist,
            }
        )

    _write_access_token_rows(synced_rows)


def prompt_non_empty(label):
    """
    Simple purpose: run `prompt_non_empty` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        label: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    while True:
        value = input(label).strip()
        if value:
            return value
        print("Value cannot be empty.")


def prompt_password():
    """
    Simple purpose: run `prompt_password` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    while True:
        password = getpass("Password: ")
        confirm = getpass("Confirm password: ")
        if not password:
            print("Password cannot be empty.")
            continue
        if password != confirm:
            print("Passwords do not match.")
            continue
        return password


def rotate_access_code(current_code):
    """
    Simple purpose: run `rotate_access_code` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        current_code: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from django.contrib.auth.hashers import make_password
    from accounts.models import ToolAccessLock

    print("\nRotate tool access code")
    verify_current = getpass("Current access code: ").strip()
    if not secrets.compare_digest(verify_current, current_code):
        print("Current access code does not match.")
        return current_code

    new_code = getpass("New access code: ").strip()
    confirm = getpass("Confirm new access code: ").strip()
    if not new_code:
        print("New code cannot be empty.")
        return current_code
    if new_code != confirm:
        print("New access code confirmation failed.")
        return current_code

    tool_lock = ToolAccessLock.objects.filter(tool_name=TOOL_NAME, is_active=True).first()
    if not tool_lock:
        print("Tool lock not found.")
        return current_code

    salt_b64 = base64.urlsafe_b64encode(os.urandom(16)).decode("utf-8")
    fernet = _fernet_from_code(new_code, salt_b64)
    tool_lock.code_hash = make_password(new_code)
    tool_lock.kdf_salt = salt_b64
    tool_lock.encrypted_sentinel = fernet.encrypt(SENTINEL_PLAINTEXT).decode("utf-8")
    tool_lock.save(update_fields=["code_hash", "kdf_salt", "encrypted_sentinel", "updated_at"])
    LOCK_FILE.write_bytes(fernet.encrypt(SENTINEL_PLAINTEXT))
    print("Access code rotated successfully.")
    return new_code


def create_admin():
    """
    Simple purpose: run `create_admin` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A newly created object or generated value.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from django.contrib.auth.hashers import make_password
    from django.contrib.auth.password_validation import validate_password
    from django.core.exceptions import ValidationError
    from accounts.models import User

    print("\nCreate Admin")
    email = prompt_non_empty("Email: ").lower()
    if not EMAIL_RE.match(email):
        print("Invalid email format.")
        return

    existing_user = User.objects.filter(email__iexact=email).only("role").first()
    if existing_user:
        role = (existing_user.role or "user").strip().lower()
        if role == "student":
            print("Cannot create admin: this email is already registered as a student.")
        elif role == "faculty":
            print("Cannot create admin: this email is already registered as a faculty member.")
        elif role == "admin":
            print("Cannot create admin: this email is already registered as an admin.")
        else:
            print(f"Cannot create admin: this email is already registered as role '{role}'.")
        return

    first_name = input("First name (optional): ").strip()
    last_name = input("Last name (optional): ").strip()
    password = prompt_password()

    try:
        validate_password(password)
    except ValidationError as exc:
        print("Password validation failed:")
        for msg in exc.messages:
            print(f"- {msg}")
        return

    access_token = generate_access_token(32)

    try:
        user = User.objects.create_user(
            username=email,
            email=email,
            password=password,
            role="admin",
            first_name=first_name,
            last_name=last_name,
            is_staff=True,
            is_superuser=True,
        )
    except IntegrityError:
        print("Could not create admin because this email already exists in the database.")
        return

    user.admin_access_token = access_token
    user.secret_key_hash = make_password(access_token)
    user.failed_login_attempts = 0
    user.locked_until = None
    user.is_verified = True
    user.status = "approved"
    user.is_staff = True
    user.is_superuser = True
    user.save()

    print("\nAdmin created successfully.")
    print(f"Email: {user.email}")
    print("Access token (save this securely):")
    print(access_token)
    display_name = f"{(user.first_name or '').strip()} {(user.last_name or '').strip()}".strip() or user.email
    upsert_access_token_csv(display_name, user.email, access_token, created_at=user.created_at)
    sync_access_tokens_csv_with_db()
    print(f"Token updated in {ACCESS_TOKENS_FILE.name}.")
    print("")


def list_admins():
    """
    Simple purpose: run `list_admins` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        Requested data in the shape expected by the caller.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from accounts.models import User

    admins = User.objects.filter(role="admin").order_by("created_at")
    print("\nAdmins")
    if not admins.exists():
        print("No admins found.\n")
        return

    for idx, admin in enumerate(admins, 1):
        name = f"{(admin.first_name or '').strip()} {(admin.last_name or '').strip()}".strip() or "-"
        print(f"{idx}. {admin.email} | name: {name} | superuser: {bool(admin.is_superuser)}")
    print("")


def delete_admin():
    """
    Simple purpose: run `delete_admin` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from accounts.models import User

    admins = list(User.objects.filter(role="admin").order_by("created_at"))
    print("\nDelete Admin")
    if not admins:
        print("No admins found.\n")
        return

    for idx, admin in enumerate(admins, 1):
        name = f"{(admin.first_name or '').strip()} {(admin.last_name or '').strip()}".strip() or "-"
        print(f"{idx}. {admin.email} | name: {name}")
    print("0. Cancel")

    choice = input("Select admin number to delete: ").strip()
    if choice == "0":
        print("Cancelled.\n")
        return
    if not choice.isdigit():
        print("Invalid option.\n")
        return

    idx = int(choice) - 1
    if idx < 0 or idx >= len(admins):
        print("Invalid option.\n")
        return

    if len(admins) == 1:
        print("Cannot delete the last admin account.\n")
        return

    target = admins[idx]
    confirm = input(f"Type DELETE to confirm removing {target.email}: ").strip()
    if confirm != "DELETE":
        print("Deletion cancelled.\n")
        return

    target.delete()
    remove_access_token_csv(target.email)
    sync_access_tokens_csv_with_db()
    print(f"Admin deleted: {target.email}\n")


def ensure_all_admins_are_superusers():
    """
    Simple purpose: run `ensure_all_admins_are_superusers` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    from accounts.models import User

    admins = User.objects.filter(role="admin")
    for admin in admins:
        changed = False
        if not admin.is_staff:
            admin.is_staff = True
            changed = True
        if not admin.is_superuser:
            admin.is_superuser = True
            changed = True
        if changed:
            admin.save(update_fields=["is_staff", "is_superuser"])


def main_menu(access_code):
    """
    Simple purpose: run `main_menu` and complete its job safely.

    This function exists to keep one task focused and easy to reuse.

    Args:
        access_code: Input value used by this step of the workflow.

    Returns:
        A result value that the caller uses in the next step.

    Raises:
        Does not raise intentionally; caller-level validation/handlers manage failures.
    """
    while True:
        print("=== Admin Management Menu ===")
        print("1. Create admin")
        print("2. List admins")
        print("3. Delete admin")
        print("4. Rotate tool access code")
        print("5. Exit")
        choice = input("Choose an option (1-5): ").strip()

        if choice == "1":
            create_admin()
        elif choice == "2":
            list_admins()
        elif choice == "3":
            delete_admin()
        elif choice == "4":
            access_code = rotate_access_code(access_code)
        elif choice == "5":
            print("Exiting.")
            break
        else:
            print("Invalid option.\n")


if __name__ == "__main__":
    setup_django()
    ensure_all_admins_are_superusers()
    sync_access_tokens_csv_with_db()
    ok, active_code = authorize_script_access()
    if not ok:
        sys.exit(1)
    main_menu(active_code)