path: root/packages/meshbay-hub/src/meshbay_hub/api/admin.py

"""
MeshBay Hub — admin/moderation panel endpoints.

All endpoints require moderator or admin role.
Separate from moderation.py (which handles public reporting and content blocklist).
"""

import logging
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from meshbay_hub.auth import decrypt_email
from meshbay_hub.api.deps import require_admin, require_moderator, user_is_admin
from meshbay_hub.api.revocation import get_connected_node_count, is_node_connected
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import Group, GroupMember, IPLog, Node, User
from meshbay_hub import hub_settings

log = logging.getLogger(__name__)

router = APIRouter(prefix="/v1/admin", tags=["admin"])


# ── Request models ───────────────────────────────────────────────────────────

class UserPatchRequest(BaseModel):
    role: str | None = None
    status: str | None = None


class GroupPatchRequest(BaseModel):
    status: str | None = None


class SettingsPatchRequest(BaseModel):
    allow_public_groups: bool | None = None
    # Every mail bound, each optional: the panel sends only what changed.
    mail: dict[str, int] | None = None
    # The sign-in lockout's two numbers, each optional, as for mail.
    login: dict[str, int] | None = None
    # Session lifetime, in hours, each optional.
    session: dict[str, int] | None = None


# ── Instance settings ────────────────────────────────────────────────────────

async def _settings_payload(db: AsyncSession) -> dict:
    return {
        "allow_public_groups": await hub_settings.public_groups_allowed(db),
        "mail": await hub_settings.mail_limits(db),
        # So the panel can show what a field falls back to, and label the
        # bounds it will refuse — rather than the operator finding out by
        # having a value silently clamped.
        "mail_defaults": {k: hub_settings.mail_default(k)
                          for k in hub_settings.MAIL_KEYS},
        "mail_bounds": {k: list(v) for k, v in hub_settings.MAIL_BOUNDS.items()},
        "login": await hub_settings.login_limits(db),
        "login_defaults": dict(hub_settings.LOGIN_DEFAULTS),
        "login_bounds": {k: list(v) for k, v in hub_settings.LOGIN_BOUNDS.items()},
        "session": await hub_settings.session_limits(db),
        "session_defaults": dict(hub_settings.SESSION_DEFAULTS),
        "session_bounds": {k: list(v) for k, v in hub_settings.SESSION_BOUNDS.items()},
    }


@router.get("/settings")
async def admin_get_settings(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
):
    """Instance-wide policy an admin controls from the panel. Moderators may read."""
    return await _settings_payload(db)


@router.patch("/settings")
async def admin_patch_settings(
    body: SettingsPatchRequest,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """
    Change instance policy. Admin only — moderators get the read above.

    The enforcement lives where the thing being restricted happens (public-group
    creation is refused in `groups.create_group`), so flipping this here is the
    whole change: a client that keeps drawing the option still cannot use it.
    """
    if body.allow_public_groups is not None:
        await hub_settings.set_raw(
            db, hub_settings.ALLOW_PUBLIC_GROUPS,
            "true" if body.allow_public_groups else "false")
        log.info("Instance setting allow_public_groups=%s by %s",
                 body.allow_public_groups, current_user.username)
        db.add(IPLog(
            user_id=current_user.id,
            event="admin_settings_update",
            ip_address="admin",
            detail=f"allow_public_groups={body.allow_public_groups}",
        ))
        await db.commit()

    if body.mail:
        unknown = sorted(set(body.mail) - set(hub_settings.MAIL_KEYS))
        if unknown:
            raise HTTPException(
                status_code=422, detail=f"Unknown mail setting(s): {unknown}")
        changed = []
        for key, value in body.mail.items():
            clamped = hub_settings.clamp_mail_value(key, value)
            await hub_settings.set_raw(db, f"mail.{key}", str(clamped))
            changed.append(f"{key}={clamped}")
        log.info("Mail bounds changed by %s: %s",
                 current_user.username, ", ".join(changed))
        db.add(IPLog(
            user_id=current_user.id,
            event="admin_mail_limits_update",
            ip_address="admin",
            detail=", ".join(changed)[:255],
        ))
        await db.commit()

    if body.login:
        unknown = sorted(set(body.login) - set(hub_settings.LOGIN_KEYS))
        if unknown:
            raise HTTPException(
                status_code=422, detail=f"Unknown login setting(s): {unknown}")
        changed = []
        for key, value in body.login.items():
            clamped = hub_settings.clamp_login_value(key, value)
            await hub_settings.set_raw(db, f"login.{key}", str(clamped))
            changed.append(f"{key}={clamped}")
        log.info("Sign-in lockout changed by %s: %s",
                 current_user.username, ", ".join(changed))
        db.add(IPLog(
            user_id=current_user.id,
            event="admin_login_lockout_update",
            ip_address="admin",
            detail=", ".join(changed)[:255],
        ))
        await db.commit()

    if body.session:
        unknown = sorted(set(body.session) - set(hub_settings.SESSION_KEYS))
        if unknown:
            raise HTTPException(
                status_code=422, detail=f"Unknown session setting(s): {unknown}")
        changed = []
        for key, value in body.session.items():
            clamped = hub_settings.clamp_session_value(key, value)
            await hub_settings.set_raw(db, f"session.{key}", str(clamped))
            changed.append(f"{key}={clamped}")
        log.info("Session lifetime changed by %s: %s",
                 current_user.username, ", ".join(changed))
        db.add(IPLog(
            user_id=current_user.id,
            event="admin_session_update",
            ip_address="admin",
            detail=", ".join(changed)[:255],
        ))
        await db.commit()

    return await _settings_payload(db)


@router.get("/mail")
async def admin_mail_status(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
):
    """Is the hub still sending, and how much of the hour is left.

    There was no way to see this at all: a refusal was a line in the journal,
    so an instance that had stopped sending registration codes looked, from the
    panel, exactly like one that had no sign-ups.
    """
    from meshbay_hub import mail

    return await mail.status(db)


# ── Stats ────────────────────────────────────────────────────────────────────

@router.get("/stats")
async def admin_stats(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
):
    # Deleted accounts are tombstoned rather than dropped, so that the
    # connection log stays readable. They are not users any more and must not be
    # counted as any: a hub whose user count only ever rises is measuring its
    # own history, not its population.
    user_count = (await db.execute(
        select(func.count()).select_from(User)
        .where(User.status != "deleted"))).scalar_one()
    # Groups are not tombstoned — deleting one removes the row — so every group
    # here is a group. A revoked one is suspended by moderation and still shown
    # in the list, so counting it keeps the two consistent.
    group_count = (await db.execute(
        select(func.count()).select_from(Group))).scalar_one()
    node_count = (await db.execute(select(func.count()).select_from(Node))).scalar_one()
    return {
        "users": user_count,
        "groups": group_count,
        "nodes": node_count,
        "online_nodes": get_connected_node_count(),
    }


# ── Users ────────────────────────────────────────────────────────────────────

@router.get("/users")
async def admin_list_users(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
    q: str = "",
    offset: int = 0,
    limit: int = Query(default=50, le=200),
):
    query = (select(User).where(User.status != "deleted")
             .order_by(User.created_at.desc()))
    if q:
        query = query.where(User.username.ilike(f"%{q}%"))
    query = query.offset(offset).limit(limit)
    result = await db.execute(query)
    users = result.scalars().all()

    total_query = (select(func.count()).select_from(User)
                   .where(User.status != "deleted"))
    if q:
        total_query = total_query.where(User.username.ilike(f"%{q}%"))
    total = (await db.execute(total_query)).scalar_one()

    return {
        "users": [
            {
                "id": u.id,
                "username": u.username,
                "role": u.role,
                "status": u.status,
                "created_at": u.created_at.isoformat(),
            }
            for u in users
        ],
        "total": total,
    }


@router.get("/users/{user_id}")
async def admin_get_user(
    user_id: str,
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    group_count = (await db.execute(
        select(func.count()).where(GroupMember.user_id == user_id)
    )).scalar_one()

    try:
        email = decrypt_email(user.email)
    except Exception:
        email = "(encrypted)"

    return {
        "id": user.id,
        "username": user.username,
        "email": email,
        "role": user.role,
        "status": user.status,
        "created_at": user.created_at.isoformat(),
        "group_count": group_count,
    }


@router.patch("/users/{user_id}")
async def admin_patch_user(
    user_id: str,
    body: UserPatchRequest,
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    if user.id == current_user.id:
        raise HTTPException(status_code=400, detail="Cannot modify your own account")

    # A moderator suspends and restores accounts — reversible content moderation.
    # Changing what someone *is* (their role), and the one irreversible status
    # (`revoked`, which is signed and broadcast to every node), are administrative.
    # Without this split a moderator could promote an accomplice to admin, or
    # revoke every admin, entirely from the moderation role. `admin_delete_user`
    # already draws this exact line for the same reason.
    privileged = body.role is not None or body.status == "revoked"
    if privileged and not user_is_admin(current_user):
        raise HTTPException(
            status_code=403,
            detail="Changing a role, or revoking an account, requires admin rights")

    # An admin's account is not a moderator's to touch at all — not their role,
    # not their status.
    if user_is_admin(user) and not user_is_admin(current_user):
        raise HTTPException(
            status_code=403,
            detail="Only an admin can change another admin's account")

    from meshbay_hub.api.notifications import create_notification

    if body.role is not None:
        if body.role not in ("user", "moderator", "admin"):
            raise HTTPException(status_code=422, detail="role must be user, moderator, or admin")
        user.role = body.role
        log.info("User %s role changed to %s by %s", user.username, body.role, current_user.username)
        await create_notification(
            db, user.id, "role_change",
            f"Your role has been changed to {body.role}",
        )

    if body.status is not None:
        if body.status not in ("active", "suspended", "revoked"):
            raise HTTPException(status_code=422, detail="status must be active, suspended, or revoked")
        user.status = body.status
        log.info("User %s status changed to %s by %s",
                 user.username, body.status, current_user.username)
        if body.status == "suspended":
            await create_notification(
                db, user.id, "account_suspended",
                "Your account has been suspended",
            )
        elif body.status == "active":
            await create_notification(
                db, user.id, "account_restored",
                "Your account has been restored",
            )

    db.add(IPLog(
        user_id=current_user.id,
        event="admin_user_update",
        ip_address="admin",
        detail=f"{user.username}: role={user.role} status={user.status}",
    ))
    await db.commit()

    return {
        "id": user.id,
        "username": user.username,
        "role": user.role,
        "status": user.status,
    }


# ── Groups ───────────────────────────────────────────────────────────────────

@router.delete("/users/{user_id}")
async def admin_delete_user(
    user_id: str,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """
    Erase an account, and every group it owns.

    The same erasure a user performs on themselves, with one difference: a user
    is asked to hand their groups over first, an administrator is not. This is
    the route an erasure ordered by an authority goes through, and it cannot
    wait on the person it is about.

    Then a signed revocation goes to every connected node, for the account and
    for each group deleted with it. The hub's records are gone at that point,
    but an access token already issued stays valid on a node until it expires;
    the revocation is what makes the nodes refuse the account and close the
    groups' sessions now. A node that is offline misses it — the hub cannot
    reach a machine it does not command.

    Admin rather than moderator: suspension is reversible and is the moderation
    tool; this is not. Refused for one's own account — an administrator locking
    themselves out is a support incident, and there is `DELETE /v1/users/me` for
    someone who means it.
    """
    from meshbay_hub.api import revocation
    from meshbay_hub.api.users import erase_account

    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    if user.id == current_user.id:
        raise HTTPException(
            status_code=400,
            detail="Use your own account settings to delete your account")
    if user.status == "deleted":
        raise HTTPException(status_code=410, detail="Account already deleted")

    groups = (await db.execute(
        select(Group.name).where(Group.admin_id == user.id))).scalars().all()
    db.add(IPLog(
        user_id=current_user.id,
        event="admin_user_delete",
        ip_address="admin",
        detail=f"{user.username} ({user.id}); groups deleted: {', '.join(groups) or 'none'}"[:256],
    ))
    result = await erase_account(db, user, owned_groups="delete")

    reason = "account deleted by an administrator"
    sent = await revocation.broadcast_revocation(
        revocation._sign_revocation("user", result["user_id"], reason))
    for g in result["groups_deleted"]:
        await revocation.broadcast_revocation(
            revocation._sign_revocation("group", g["id"], reason))
    log.warning("Account %s erased by admin %s, %d owned group(s) deleted, "
                "revocations sent to %d node(s)", result["username"],
                current_user.username, len(result["groups_deleted"]), sent)
    return {**result, "nodes_notified": sent}


@router.get("/groups")
async def admin_list_groups(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
    offset: int = 0,
    limit: int = Query(default=50, le=200),
):
    query = (
        select(
            Group,
            func.count(User.id).label("member_count"),
        )
        # Members, not rows: a deleted account's membership is removed with it,
        # but joining through User keeps the count honest if one ever survives.
        .outerjoin(GroupMember, Group.id == GroupMember.group_id)
        .outerjoin(User, (User.id == GroupMember.user_id)
                   & (User.status != "deleted"))
        .group_by(Group.id)
        .order_by(Group.created_at.desc())
        .offset(offset)
        .limit(limit)
    )
    result = await db.execute(query)
    rows = result.all()

    total = (await db.execute(select(func.count()).select_from(Group))).scalar_one()

    owner_ids = {g.admin_id for g, _ in rows}
    owner_by_id = dict((await db.execute(
        select(User.id, User.username).where(User.id.in_(owner_ids)))).all()) \
        if owner_ids else {}

    return {
        "groups": [
            {
                "id": g.id,
                "name": g.name,
                "admin_id": g.admin_id,
                "owner_username": owner_by_id.get(g.admin_id),
                "visibility": g.visibility,
                "description": g.description or "",
                "status": g.status,
                "created_at": g.created_at.isoformat(),
                "member_count": mc,
            }
            for g, mc in rows
        ],
        "total": total,
    }


@router.patch("/groups/{group_id}")
async def admin_patch_group(
    group_id: str,
    body: GroupPatchRequest,
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
):
    group = await db.get(Group, group_id)
    if not group:
        raise HTTPException(status_code=404, detail="Group not found")

    if body.status is not None:
        if body.status not in ("active", "suspended", "revoked"):
            raise HTTPException(status_code=422, detail="status must be active, suspended, or revoked")
        group.status = body.status
        log.info("Group %s status changed to %s by %s",
                 group.name, body.status, current_user.username)

    db.add(IPLog(
        user_id=current_user.id,
        event="admin_group_update",
        ip_address="admin",
        detail=f"{group.name}: status={group.status}",
    ))
    await db.commit()

    return {
        "id": group.id,
        "name": group.name,
        "status": group.status,
    }


# ── IP Audit Logs ────────────────────────────────────────────────────────────

@router.get("/nodes")
async def admin_list_nodes(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
    limit: int = Query(default=100, le=200),
):
    """
    Registered nodes, with the address the hub saw them announce from.

    `observed_ip` is the one to answer a question with: it comes from the
    connection that carried a valid Ed25519 signature over a fresh timestamp, so
    it is the address of whoever holds the node key. `endpoint_hint` is what the
    node believes its own address to be, discovered through a STUN server and
    sent to us — useful for reaching it, and not evidence of anything.
    """
    rows = (await db.execute(
        select(Node, User.username)
        .outerjoin(User, User.id == Node.user_id)
        .order_by(Node.announced_at.desc())
        .limit(limit))).all()
    return {
        "nodes": [
            {
                "id": n.id,
                "user_id": n.user_id,
                "username": uname or "",
                "pk_node": n.pk_node,
                "observed_ip": n.observed_ip or "",
                "endpoint_hint": n.endpoint_hint or "",
                "last_seen": n.last_seen.isoformat() if n.last_seen else "",
                "announced_at": n.announced_at.isoformat(),
                "online": is_node_connected(n.id),
            }
            for n, uname in rows
        ],
    }


@router.get("/logs")
async def admin_list_logs(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
    user_id: str | None = None,
    event: str | None = None,
    offset: int = 0,
    limit: int = Query(default=50, le=200),
):
    query = (
        select(IPLog, User.username)
        .outerjoin(User, IPLog.user_id == User.id)
        .order_by(IPLog.timestamp.desc())
    )
    if user_id:
        query = query.where(IPLog.user_id == user_id)
    if event:
        query = query.where(IPLog.event == event)
    query = query.offset(offset).limit(limit)
    result = await db.execute(query)
    rows = result.all()

    return {
        "logs": [
            {
                "id": lg.id,
                "user_id": lg.user_id,
                # The kept name wins: it is only ever written when an account is
                # deleted, and the join still answers then — with the tombstone,
                # `deleted-3f9a1c`, which is the one answer that helps nobody.
                "username": lg.username or uname or "",
                "event": lg.event,
                "ip_address": lg.ip_address,
                "detail": lg.detail,
                "timestamp": lg.timestamp.isoformat(),
            }
            for lg, uname in rows
        ],
    }