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

"""
MeshBay Hub — moderation endpoints.

Reporting flow:
  POST /v1/reports            — report a content hash (sign-in required)

  Thresholds (counted as DISTINCT reporting accounts, not raw rows):
    < AUTO_BLOCK_THRESHOLD distinct reporters → logged
    >= AUTO_BLOCK_THRESHOLD distinct reporters → hash added to the blocklist

  The flow only runs while the hub brokers public content: with public groups
  switched off instance-wide there is nothing here to serve a reported hash from,
  so it is refused rather than left open as an unauthenticated write surface.

Admin endpoints:
  GET  /v1/admin/blocklist    — list blocked hashes
  POST /v1/admin/blocklist    — manually add a hash
  DELETE /v1/admin/blocklist/{hash} — remove a hash

Node integration:
  GET  /v1/blocklist?after=<hash>  — the list, paged, for a node's own token
  WebSocket `blocklist_update`      — additions and removals, pushed to nodes
                                      hosting a public group
"""

import logging

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

from meshbay_hub import hub_settings
from meshbay_hub.api.deps import get_current_user, require_admin, require_node_scope
from meshbay_hub.api.middleware import limiter
from meshbay_hub.api.netutil import client_ip
from meshbay_hub.api.revocation import broadcast_blocklist_update
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import ContentBlocklist, ContentReport, User

log = logging.getLogger(__name__)

router = APIRouter(tags=["moderation"])

# Distinct reporting accounts before a hash is auto-blocked. Kept low for a
# responsive community signal, but note it is only as strong as account
# creation: while a bot can register freely (see the reCAPTCHA gap), the real
# control is the admin reviewing `GET /v1/admin/blocklist` and the audit log.
AUTO_BLOCK_THRESHOLD = 3


def _is_hash(value: str) -> bool:
    return len(value) == 64 and all(c in "0123456789abcdef" for c in value)


# ── Models ────────────────────────────────────────────────────────────────────

class ReportRequest(BaseModel):
    content_hash: str   # blake3 hex (64 chars)
    group_id:     str | None = None
    reason:       str = "illegal"
    detail:       str | None = None


class BlocklistAddRequest(BaseModel):
    content_hash: str
    reason:       str


# ── Public endpoints ──────────────────────────────────────────────────────────

@router.post("/v1/reports", status_code=201)
@limiter.limit("10/hour")
async def report_content(
    body: ReportRequest,
    request: Request,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """
    Report a public content hash for moderation.

    Sign-in is required. It used to be anonymous, which made it a censorship
    primitive: two unauthenticated POSTs naming any blake3 id auto-added it to
    the blocklist that nodes enforce, network-wide, with manual admin removal the
    only undo. The threshold now counts *distinct reporting accounts*, one vote
    per account per hash.

    Refused entirely when the hub has public groups switched off: nothing here
    brokers public content then, nothing syncs the blocklist, and an open write
    endpoint would only be abuse surface.
    """
    if not await hub_settings.public_groups_allowed(db):
        raise HTTPException(
            status_code=403,
            detail="This hub does not broker public content, so there is nothing to report here.")

    if not _is_hash(body.content_hash):
        raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)")

    # One vote per account per hash — a single reporter must not be able to walk
    # the threshold up on their own by posting repeatedly.
    already = await db.scalar(
        select(ContentReport.id).where(
            ContentReport.content_hash == body.content_hash,
            ContentReport.reporter_id == current_user.id))

    if not already:
        db.add(ContentReport(
            content_hash=body.content_hash,
            reporter_id=current_user.id,
            group_id=body.group_id,
            reason=body.reason,
            detail=body.detail,
            ip_address=client_ip(request),
        ))
        await db.flush()

    distinct_reporters = await db.scalar(
        select(func.count(func.distinct(ContentReport.reporter_id)))
        .where(ContentReport.content_hash == body.content_hash)) or 0

    action = "already_reported" if already else "logged"
    auto_blocked = False
    if distinct_reporters >= AUTO_BLOCK_THRESHOLD:
        existing = await db.get(ContentBlocklist, body.content_hash)
        if not existing:
            db.add(ContentBlocklist(
                content_hash=body.content_hash,
                reason=f"auto:{body.reason}",
                added_by="auto",
            ))
            action = "auto_blocked"
            auto_blocked = True
            log.warning("Content auto-blocked after %d distinct reporters: %s",
                        distinct_reporters, body.content_hash[:16])

    await db.commit()
    if auto_blocked:
        await broadcast_blocklist_update(db, add=[body.content_hash])
    return {
        "status":       action,
        "content_hash": body.content_hash,
        "report_count": distinct_reporters,
        "threshold":    AUTO_BLOCK_THRESHOLD,
    }


@router.get("/v1/blocklist")
async def get_blocklist(
    current_node: User = Depends(require_node_scope),
    db: AsyncSession = Depends(get_db),
    after: str = Query(default="", max_length=64),
    limit: int = Query(default=10000, ge=1, le=10000),
):
    """
    The content blocklist, a page at a time, for a node hosting a public group.

    Hashes only, never the reasons. Ordered by hash so `after` (the last hash of
    the previous page) is a stable cursor: a list longer than one page used to
    be cut at 10 000 with no way to ask for the rest, and the node applying it
    silently served everything past the cut. A node's own token, because this
    is what a node fetches on its own behalf and nothing else asks for it.
    """
    result = await db.execute(
        select(ContentBlocklist.content_hash)
        .where(ContentBlocklist.content_hash > after)
        .order_by(ContentBlocklist.content_hash)
        .limit(limit)
    )
    hashes = [row[0] for row in result.fetchall()]
    return {"hashes": hashes,
            "next": hashes[-1] if len(hashes) == limit else None}


# ── Admin endpoints ───────────────────────────────────────────────────────────

@router.get("/v1/admin/blocklist")
async def admin_list_blocklist(
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
    limit: int = 500,
):
    result = await db.execute(
        select(ContentBlocklist)
        .order_by(ContentBlocklist.added_at.desc())
        .limit(limit)
    )
    entries = result.scalars().all()
    return {
        "entries": [
            {
                "hash":     e.content_hash,
                "reason":   e.reason,
                "added_at": e.added_at.isoformat(),
                "added_by": e.added_by,
            }
            for e in entries
        ]
    }


@router.post("/v1/admin/blocklist", status_code=201)
async def admin_add_blocklist(
    body: BlocklistAddRequest,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    if not _is_hash(body.content_hash):
        raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)")
    existing = await db.get(ContentBlocklist, body.content_hash)
    if existing:
        raise HTTPException(status_code=409, detail="Hash already blocked")

    db.add(ContentBlocklist(
        content_hash=body.content_hash,
        reason=body.reason[:64],
        added_by=current_user.username,
    ))
    await db.commit()
    await broadcast_blocklist_update(db, add=[body.content_hash])
    return {"status": "blocked", "hash": body.content_hash}


@router.delete("/v1/admin/blocklist/{content_hash}", status_code=200)
async def admin_remove_blocklist(
    content_hash: str,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    entry = await db.get(ContentBlocklist, content_hash)
    if not entry:
        raise HTTPException(status_code=404, detail="Hash not in blocklist")
    await db.delete(entry)
    await db.commit()
    await broadcast_blocklist_update(db, remove=[content_hash])
    return {"status": "unblocked", "hash": content_hash}