chadtmiller.com/knotless

//! The moderation service: reports in, takedowns out, as atproto has them.
//!
//! Anyone sends a report with a service token from their PDS
//! (`com.atproto.moderation.createReport`). The operator, by the admin
//! password (`KNOTLESS_ADMIN_PASSWORD`), reads them (`dev.knotless.listReports`)
//! and takes accounts and records down (`com.atproto.admin.updateSubjectStatus`).
//! A takedown leaves the index, the pages and the clones here at once, and
//! goes out as a `!takedown` label (`com.atproto.label.subscribeLabels`,
//! `queryLabels`) for other appviews that follow this one; this one follows
//! the labelers it is given the same way (`--labeler`). Labels go out
//! unsigned: a follower reads them from the endpoint the labeler's DID
//! document names.

use crate::gateway::{caller, gateway};
use crate::index::Index;
use crate::{App, AppError};
use anyhow::Context;
use axum::Json;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::Response;
use base64::Engine;
use futures::{SinkExt, StreamExt};
use knotless::moderation::{GET_STATUS, LABELER, REPORT};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::Duration;

fn index(app: &App) -> Result<&Index, AppError> {
    app.index
        .as_deref()
        .ok_or_else(|| AppError(StatusCode::NOT_FOUND, "this server keeps no index".into()))
}

fn bad(msg: impl Into<String>) -> AppError {
    AppError(StatusCode::BAD_REQUEST, msg.into())
}

/// Lets the request through if it carries the admin password, as a PDS
/// takes it: `Basic` for the user `admin`.
fn admin(app: &App, headers: &HeaderMap) -> Result<(), AppError> {
    let denied = || {
        AppError(
            StatusCode::UNAUTHORIZED,
            "the admin password is required".into(),
        )
    };
    let password = app.ctx.env.admin_password.as_deref().ok_or_else(denied)?;
    let expected = base64::engine::general_purpose::STANDARD.encode(format!("admin:{password}"));
    let given = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Basic "));
    match given {
        Some(given) if given == expected => Ok(()),
        _ => Err(denied()),
    }
}

/// The account or record a subject names: a DID or an `at://` URI.
fn subject_key(subject: &Value) -> Result<String, AppError> {
    let key = match subject["$type"].as_str() {
        Some("com.atproto.admin.defs#repoRef") => subject["did"].as_str(),
        Some("com.atproto.repo.strongRef") => subject["uri"].as_str(),
        _ => None,
    };
    key.map(str::to_string).ok_or_else(|| {
        bad("a subject is a com.atproto.admin.defs#repoRef or a com.atproto.repo.strongRef")
    })
}

pub async fn create_report(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, AppError> {
    let index = index(&app)?;
    let by = caller(&app, &headers, REPORT).await?;
    subject_key(&body["subject"])?;
    let reason_type = body["reasonType"]
        .as_str()
        .ok_or_else(|| bad("a report has a reasonType"))?;
    let mut report = json!({
        "reasonType": reason_type,
        "subject": body["subject"],
        "reportedBy": by,
        "createdAt": knotless::records::now(),
    });
    if let Some(reason) = body["reason"].as_str() {
        report["reason"] = reason.into();
    }
    Ok(Json(
        index
            .report(report)
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?,
    ))
}

#[derive(Deserialize)]
pub struct ReportsParams {
    limit: Option<usize>,
    cursor: Option<i64>,
}

pub async fn list_reports(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(p): Query<ReportsParams>,
) -> Result<Json<Value>, AppError> {
    admin(&app, &headers)?;
    let limit = p.limit.unwrap_or(50).clamp(1, 500);
    let reports = index(&app)?
        .reports(limit, p.cursor)
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    let mut out = json!({ "reports": reports });
    if reports.len() == limit {
        out["cursor"] = reports.last().map(|r| r["id"].to_string()).into();
    }
    Ok(Json(out))
}

pub async fn update_subject_status(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, AppError> {
    admin(&app, &headers)?;
    let index = index(&app)?;
    let src = &gateway(&app)?.did;
    let subject = &body["subject"];
    let key = subject_key(subject)?;
    let mut out = json!({ "subject": subject });
    if let Some(applied) = body["takedown"]["applied"].as_bool() {
        let (seq, label) = index
            .take_down(src, &key, subject, applied)
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
        let _ = app.labels.send((seq, label));
        out["takedown"] = json!({ "applied": applied });
    }
    Ok(Json(out))
}

#[derive(Deserialize)]
pub struct StatusParams {
    did: Option<String>,
    uri: Option<String>,
}

pub async fn get_subject_status(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(p): Query<StatusParams>,
) -> Result<Json<Value>, AppError> {
    admin(&app, &headers)?;
    let (key, named) = match (p.did, p.uri) {
        (Some(did), _) => (
            did.clone(),
            json!({"$type": "com.atproto.admin.defs#repoRef", "did": did}),
        ),
        (None, Some(uri)) => (
            uri.clone(),
            json!({"$type": "com.atproto.repo.strongRef", "uri": uri}),
        ),
        (None, None) => return Err(bad(format!("{GET_STATUS} takes a did or a uri"))),
    };
    let found = index(&app)?
        .taken_down(&key)
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    let applied = found.is_some();
    Ok(Json(json!({
        "subject": found.unwrap_or(named),
        "takedown": { "applied": applied },
    })))
}

/// Takedowns standing here, as labels, for each of the `uriPatterns`: a
/// URI, or a prefix ending in `*`.
pub async fn query_labels(
    State(app): State<Arc<App>>,
    Query(params): Query<Vec<(String, String)>>,
) -> Result<Json<Value>, AppError> {
    let index = index(&app)?;
    let src = &gateway(&app)?.did;
    let mut labels = Vec::new();
    for (_, pattern) in params.iter().filter(|(k, _)| k == "uriPatterns") {
        let (prefix, exact) = match pattern.strip_suffix('*') {
            Some(prefix) => (prefix, false),
            None => (pattern.as_str(), true),
        };
        let found = index
            .labels(src, prefix)
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
        labels.extend(found.into_iter().filter(|l| !exact || l["uri"] == *pattern));
    }
    Ok(Json(json!({ "labels": labels })))
}

#[derive(Deserialize)]
pub struct SubscribeParams {
    cursor: Option<i64>,
}

/// The labels this server gives out, as they are given: those after
/// `cursor` first, when there is one.
pub async fn subscribe_labels(
    State(app): State<Arc<App>>,
    Query(p): Query<SubscribeParams>,
    ws: WebSocketUpgrade,
) -> Result<Response, AppError> {
    index(&app)?;
    gateway(&app)?;
    Ok(ws.on_upgrade(move |socket| stream(app, socket, p.cursor)))
}

async fn stream(app: Arc<App>, socket: WebSocket, cursor: Option<i64>) {
    let (mut tx, mut rx) = socket.split();
    let mut live = app.labels.subscribe();
    let mut past: VecDeque<(i64, Value)> = match (cursor, &app.index) {
        (Some(cursor), Some(index)) => index.emitted(cursor).unwrap_or_default().into(),
        _ => VecDeque::new(),
    };
    // What was given out while the past was read arrives live as well.
    let mut sent = cursor.unwrap_or(i64::MIN);
    let mut open = true;
    while open {
        let next = match past.pop_front() {
            Some(next) => Some(next),
            None => tokio::select! {
                got = live.recv() => got.ok(),
                _ = closed(&mut rx) => None,
            },
        };
        let Some((seq, label)) = next else { return };
        if seq > sent {
            sent = seq;
            open = tx
                .send(Message::Binary(frame(seq, &label).into()))
                .await
                .is_ok();
        }
    }
}

/// Resolves when the subscriber hangs up: it has nothing to say.
async fn closed(rx: &mut futures::stream::SplitStream<WebSocket>) {
    while let Some(Ok(_)) = rx.next().await {}
}

/// A `#labels` frame, as `subscribeLabels` sends one: a CBOR header and body.
pub fn frame(seq: i64, label: &Value) -> Vec<u8> {
    let mut out = Vec::new();
    let header = json!({"op": 1, "t": "#labels"});
    let body = json!({"seq": seq, "labels": [label]});
    ciborium::into_writer(&header, &mut out).expect("JSON encodes as CBOR");
    ciborium::into_writer(&body, &mut out).expect("JSON encodes as CBOR");
    out
}

/// Reads a `#labels` frame: its sequence number and labels, each with the
/// fields that matter here. `None` for any other frame.
pub fn parse(frame: &[u8]) -> Option<(i64, Vec<Value>)> {
    use ciborium::Value as V;
    let mut reader = frame;
    let header: V = ciborium::from_reader(&mut reader).ok()?;
    let body: V = ciborium::from_reader(&mut reader).ok()?;
    let field = |v: &V, name: &str| -> Option<V> {
        v.as_map()?
            .iter()
            .find(|(k, _)| k.as_text() == Some(name))
            .map(|(_, v)| v.clone())
    };
    if field(&header, "t")?.as_text() != Some("#labels") {
        return None;
    }
    let seq = i64::try_from(field(&body, "seq")?.as_integer()?).ok()?;
    let labels = field(&body, "labels")?
        .as_array()?
        .iter()
        .map(|l| {
            let mut out = json!({});
            for name in ["src", "uri", "val", "exp"] {
                if let Some(text) = field(l, name).and_then(|v| v.as_text().map(str::to_string)) {
                    out[name] = text.into();
                }
            }
            out["neg"] = field(l, "neg")
                .and_then(|v| v.as_bool())
                .unwrap_or(false)
                .into();
            out
        })
        .collect();
    Some((seq, labels))
}

/// Follows the labeler `did` until the process ends: its `!takedown` and
/// `!hide` labels take things out of view here, as this server's own
/// takedowns do. Reconnects from where it read to.
pub async fn follow(app: Arc<App>, did: String) {
    loop {
        if let Err(err) = follow_once(&app, &did).await {
            eprintln!("following the labeler {did}: {err:#}");
        }
        tokio::time::sleep(Duration::from_millis(500)).await;
    }
}

async fn follow_once(app: &App, did: &str) -> anyhow::Result<()> {
    let index = app
        .index
        .as_ref()
        .ok_or_else(|| anyhow::anyhow!("following a labeler takes an index"))?;
    let doc = knotless::identity::document(&app.ctx, did).await?;
    let endpoint = knotless::identity::service(&doc, LABELER)
        .ok_or_else(|| anyhow::anyhow!("{did} names no {LABELER} service"))?;
    let cursor = index.labeler_seq(did)?.unwrap_or(0);
    let url = format!(
        "{}/xrpc/com.atproto.label.subscribeLabels?cursor={cursor}",
        endpoint.replacen("http", "ws", 1)
    );
    let (mut ws, _) = tokio_tungstenite::connect_async(url).await?;
    loop {
        let msg = ws.next().await.context("the labeler ended the stream")??;
        let (seq, labels) = parse(&msg.into_data()).unwrap_or_default();
        for label in &labels {
            index.label(did, seq, label)?;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn frames() {
        let label = json!({"src": "did:web:l", "uri": "did:plc:a", "val": "!takedown", "cts": "t"});
        let (seq, labels) = parse(&frame(7, &label)).unwrap();
        assert_eq!(seq, 7);
        assert_eq!(
            labels,
            [json!({"src": "did:web:l", "uri": "did:plc:a", "val": "!takedown", "neg": false})]
        );
        let mut other = Vec::new();
        ciborium::into_writer(&json!({"op": 1, "t": "#info"}), &mut other).unwrap();
        ciborium::into_writer(&json!({"name": "OutdatedCursor"}), &mut other).unwrap();
        assert_eq!(parse(&other), None);
        assert_eq!(parse(b"\xff"), None);
    }

    #[test]
    fn subjects() {
        assert_eq!(
            subject_key(&json!({"$type": "com.atproto.admin.defs#repoRef", "did": "did:a"}))
                .unwrap(),
            "did:a"
        );
        assert!(subject_key(&json!({"$type": "com.atproto.repo.strongRef"})).is_err());
        assert!(subject_key(&json!({"did": "did:a"})).is_err());
    }
}