glo/routes/admin.py

from .common import (
    ROLES,
    STATUSES,
    now_iso,
    parse_iso,
    utc_now,
    csrf_valid,
    rate_limited,
    require_moderator,
    require_admin,
    audit,
    hash_identifier,
    safe_json,
    jsonify,
    get_db,
    db_write,
)


def register_admin_routes(app):
    @app.route("/api/admin/users")
    def admin_list_users():
        moderator, error = require_moderator()
        if error:
            return error

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username,
                    u.role,
                    u.status,
                    u.restricted_until,
                    u.created_at,
                    u.last_login_at,
                    COUNT(m.id) AS message_count
                FROM users AS u
                LEFT JOIN messages AS m
                    ON m.user_id = u.id
                GROUP BY u.id
                ORDER BY
                    CASE u.role
                        WHEN 'admin' THEN 0
                        WHEN 'moderator' THEN 1
                        ELSE 2
                    END,
                    u.username COLLATE NOCASE
                """
            ).fetchall()

        return jsonify(
            {
                "users": [
                    {
                        "id": row["id"],
                        "username": row["username"],
                        "role": row["role"],
                        "status": row["status"],
                        "restricted_until": row["restricted_until"],
                        "created_at": row["created_at"],
                        "last_login_at": row["last_login_at"],
                        "message_count": row["message_count"],
                    }
                    for row in rows
                ],
                "requested_by": moderator["username"],
            }
        )

    @app.route("/api/admin/users/<int:user_id>", methods=["PATCH"])
    def admin_update_user(user_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        with get_db() as db:
            target = db.execute(
                """
                SELECT
                    id,
                    username,
                    role,
                    status,
                    restricted_until
                FROM users
                WHERE id = ?
                """,
                (user_id,),
            ).fetchone()

        if not target:
            return jsonify({"error": "User not found."}), 404

        if target["id"] == admin["id"]:
            return jsonify({"error": "You cannot modify your own account here."}), 400

        role = data.get("role", target["role"])
        status = data.get("status", target["status"])
        restricted_until = data.get("restricted_until", target["restricted_until"])

        if role not in ROLES:
            return jsonify({"error": "Invalid role."}), 400
        if status not in STATUSES:
            return jsonify({"error": "Invalid account status."}), 400

        if status == "restricted" and restricted_until:
            parsed = parse_iso(restricted_until)
            if not parsed or parsed <= utc_now():
                return jsonify({"error": "restricted_until must be a future ISO timestamp."}), 400
        elif status != "restricted":
            restricted_until = None

        if target["role"] == "admin" and role != "admin":
            with get_db() as db:
                count = db.execute(
                    """
                    SELECT COUNT(*) AS count
                    FROM users
                    WHERE role = 'admin'
                      AND status = 'active'
                    """
                ).fetchone()["count"]
            if count <= 1:
                return jsonify({"error": "At least one active administrator is required."}), 400

        is_admin_value = 1 if role == "admin" else 0
        is_active = 1 if status in {"active", "restricted"} else 0

        db_write(
            """
            UPDATE users
            SET
                role = ?,
                status = ?,
                restricted_until = ?,
                is_admin = ?,
                is_active = ?
            WHERE id = ?
            """,
            (role, status, restricted_until, is_admin_value, is_active, user_id),
        )

        if status == "blacklisted":
            db_write(
                """
                INSERT OR REPLACE INTO blacklist_entries (
                    kind,
                    value_hash,
                    reason,
                    created_at,
                    created_by_user_id
                ) VALUES ('user', ?, ?, ?, ?)
                """,
                (
                    hash_identifier(str(user_id)),
                    "Account blacklisted by administrator",
                    now_iso(),
                    admin["id"],
                ),
            )

        if target["status"] == "blacklisted" and status != "blacklisted":
            db_write(
                "DELETE FROM blacklist_entries WHERE kind = 'user' AND value_hash = ?",
                (hash_identifier(str(user_id)),),
            )

        audit(
            "user_updated",
            admin["id"],
            "user",
            str(user_id),
            f"role={role};status={status};restricted_until={restricted_until}",
        )
        return jsonify({"status": "updated"})

    @app.route("/api/admin/rooms")
    def admin_rooms():
        moderator, error = require_moderator()
        if error:
            return error

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    id,
                    slug,
                    name,
                    description,
                    visibility,
                    created_by_user_id,
                    is_active
                FROM rooms
                WHERE is_active = 1
                ORDER BY name COLLATE NOCASE
                """
            ).fetchall()

        return jsonify(
            {
                "rooms": [
                    {
                        "id": row["id"],
                        "slug": row["slug"],
                        "name": row["name"],
                        "description": row["description"],
                        "visibility": row["visibility"],
                    }
                    for row in rows
                ]
            }
        )

    @app.route("/api/admin/blacklist/ip", methods=["POST"])
    def blacklist_ip():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        ip = str(data.get("ip", "")).strip()
        reason = str(data.get("reason", "")).strip()[:300]

        if not ip or len(ip) > 128:
            return jsonify({"error": "A valid IP value is required."}), 400

        db_write(
            """
            INSERT OR REPLACE INTO blacklist_entries (
                kind,
                value_hash,
                reason,
                created_at,
                created_by_user_id
            ) VALUES ('ip', ?, ?, ?, ?)
            """,
            (
                hash_identifier(ip),
                reason or "IP blacklisted by administrator",
                now_iso(),
                admin["id"],
            ),
        )

        audit("ip_blacklisted", admin["id"], "ip", hash_identifier(ip), reason)
        return jsonify({"status": "blacklisted"})