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/check?hash=<blake3> — check if a hash is blocked
  GET  /v1/blocklist                     — full blocklist (for node sync)
"""

import logging
from datetime import datetime, timezone

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
from meshbay_hub.api.middleware import limiter
from meshbay_hub.api.netutil import client_ip
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


# ── 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 len(body.content_hash) != 64 or not all(c in "0123456789abcdef" for c in 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"
    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"
            log.warning("Content auto-blocked after %d distinct reporters: %s",
                        distinct_reporters, body.content_hash[:16])

    await db.commit()
    return {
        "status":       action,
        "content_hash": body.content_hash,
        "report_count": distinct_reporters,
        "threshold":    AUTO_BLOCK_THRESHOLD,
    }


@router.get("/v1/blocklist/check")
@limiter.limit("120/minute")
async def check_blocklist(
    hash: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    """Check if a single hash is blocked. Used by nodes before serving public content.

    Unauthenticated, because a node consults it before serving public content
    and does so on its own behalf. That makes the shape check worth having:
    without it any string of any length became a primary-key lookup.
    """
    if len(hash) != 64 or not all(c in "0123456789abcdef" for c in hash):
        raise HTTPException(status_code=422, detail="hash must be 64 hex chars (blake3)")
    blocked = await db.get(ContentBlocklist, hash)
    return {
        "blocked": blocked is not None,
        "hash":    hash,
        "reason":  blocked.reason if blocked else None,
    }


@router.get("/v1/blocklist")
async def get_blocklist(
    db: AsyncSession = Depends(get_db),
    # Bounded, like every other list. This one takes no authentication — a
    # node syncs it at startup — and had no ceiling at all, so any stranger
    # could ask for the table in one query, repeatedly. 10 000 is what a node
    # asks for, so it is the default and also the most anyone may have.
    limit: int = Query(default=10000, ge=1, le=10000),
):
    """
    Return the full blocklist. Nodes sync this on startup.
    Returns hashes only (not reasons) to minimize data exposure.
    """
    result = await db.execute(
        select(ContentBlocklist.content_hash)
        .order_by(ContentBlocklist.added_at.desc())
        .limit(limit)
    )
    hashes = [row[0] for row in result.fetchall()]
    return {"count": len(hashes), "hashes": hashes}


# ── 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),
):
    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,
        added_by=current_user.username,
    ))
    await db.commit()
    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()
    return {"status": "unblocked", "hash": content_hash}