maarcadetweet/crates/pds-server/src/main.rs

mod appview_push;
mod car;
mod dag_cbor;
mod firehose;
mod invite;
mod jwt_issuer;
mod keys;
mod password;
mod routes;
mod state;

use crate::routes::types::DescribeServerResp;
use crate::state::AppState;
use axum::extract::State;
use axum::routing::{get, post};
use axum::{Json, Router};
use serde_json::json;
use tracing::{info, warn};
use tracing_subscriber::EnvFilter;

/// Usage line for the binary itself. The subcommands are operator
/// tooling; the no-argument form is the server, which is what every
/// deploy script and systemd unit invokes.
const USAGE: &str = "\
usage: pds-server [command]

  (no command)   run the PDS server
  invite …       manage invite codes (see `pds-server invite help`)
  help           show this message";

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Load `.env` from the working directory (and upwards) if present.
    // Nothing else in the process reads it, so without this
    // `cp .env.example .env && cargo run` fails with `missing env:
    // PDS_HOST`. Real environment variables always win over the file.
    let _ = dotenvy::dotenv();

    // Argument dispatch, by hand.
    //
    // The workspace carries no argument-parsing crate and this does not
    // justify adding one: exactly one subcommand exists, and the
    // overwhelmingly common invocation is the bare binary. Anything we
    // do not recognise is an error rather than being ignored — a
    // mistyped `pds-server invit create` that silently booted a server
    // would look like it worked and mint no codes.
    let args: Vec<String> = std::env::args().skip(1).collect();
    match args.first().map(|s| s.as_str()) {
        None => {}
        Some("invite") => {
            // CLI output is meant to be read and pasted, so keep the
            // log stream quiet unless the operator asked for it. Without
            // this, `sqlx::migrate` chatters over the codes.
            init_tracing("warn");
            return invite::run_cli(&args[1..]).await;
        }
        Some("help") | Some("-h") | Some("--help") => {
            println!("{USAGE}");
            return Ok(());
        }
        Some(other) => {
            eprintln!("pds-server: unknown command: {other}\n\n{USAGE}");
            std::process::exit(2);
        }
    }

    init_tracing("info");

    let cfg = at_shared::config::AppConfig::from_env()?;

    // Announce the relaxed security posture before we bind a port.
    //
    // `PDS_INVITE_REQUIRED` is the only switch in `AppConfig` that
    // defaults to *open* (so the integration suites and dev instances
    // can keep creating throwaway accounts), which makes this warning
    // the only thing standing between "we made the PDS public" and
    // "anyone on the internet can mint repos on our disk". Mirrors
    // `appview`'s `auth::log_startup_posture`.
    if !cfg.pds_invite_required {
        warn!(
            "PDS_INVITE_REQUIRED is not true — com.atproto.server.createAccount accepts \
             ANY caller, and every accepted account allocates a repo, a server-held key \
             pair and firehose events. Fine on a private/dev instance; on a publicly \
             reachable PDS set PDS_INVITE_REQUIRED=true and hand out codes with \
             `pds-server invite create`."
        );
    } else {
        info!("PDS_INVITE_REQUIRED=true — createAccount requires a valid invite code");
    }
    let db = sqlx::postgres::PgPoolOptions::new()
        .max_connections(32)
        .min_connections(2)
        .acquire_timeout(std::time::Duration::from_secs(10))
        .connect(&cfg.database_url_pds)
        .await?;
    sqlx::migrate!("../../migrations/pds").run(&db).await?;

    let blob = at_blob::S3BlobStore::new(
        cfg.s3_endpoint.clone(),
        cfg.s3_region.clone(),
        cfg.s3_access_key.clone(),
        cfg.s3_secret_key.clone(),
        cfg.s3_bucket_pds.clone(),
        cfg.pds_public_url.clone(),
    );

    // Best-effort reachability check for the configured S3 endpoint.
    // The PDS continues to operate if MinIO is unreachable — `uploadBlob`
    // falls back to local-only storage and the S3 push is logged at
    // warn level — but we want this surfaced loudly at startup so
    // operators notice in dev. See `at_blob::s3` for the
    // MinIO-only limitation.
    if !blob.ping().await {
        warn!(
            endpoint = %cfg.s3_endpoint,
            bucket = %cfg.s3_bucket_pds,
            "s3 ping failed at startup; uploadBlob will serve from local blockstore only"
        );
    }

    let state = AppState::new(cfg.clone(), db, blob).await;
    let app = router(state);

    let addr: std::net::SocketAddr = format!("{}:{}", cfg.pds_host, cfg.pds_port).parse()?;
    info!("pds-server listening on http://{addr}");
    let listener = tokio::net::TcpListener::bind(addr).await?;
    axum::serve(listener, app).await?;
    Ok(())
}

/// Install the tracing subscriber, with `default_filter` as the level
/// when `RUST_LOG` says nothing.
///
/// Factored out because the two entry points want different defaults:
/// the server wants `info`, the `invite` subcommand wants `warn` so that
/// migration chatter does not land in the middle of a list of codes the
/// operator is about to copy.
fn init_tracing(default_filter: &str) {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default_filter)),
        )
        .init();
}

pub fn router(state: AppState) -> Router {
    Router::new()
        .route("/", get(root))
        .route("/healthz", get(healthz))
        .route("/.well-known/did.json", get(did_document))
        .route(
            "/xrpc/com.atproto.server.describeServer",
            get(describe_server),
        )
        .route(
            "/xrpc/com.atproto.server.createAccount",
            post(routes::auth::create_account),
        )
        .route(
            "/xrpc/com.atproto.server.createSession",
            post(routes::auth::create_session),
        )
        .route(
            "/xrpc/com.atproto.server.refreshSession",
            post(routes::auth::refresh_session),
        )
        .route(
            "/xrpc/com.atproto.identity.resolveHandle",
            post(routes::identity::resolve_handle),
        )
        .route(
            "/xrpc/com.atproto.repo.createRecord",
            post(routes::repo::create_record),
        )
        .route(
            "/xrpc/com.atproto.repo.deleteRecord",
            post(routes::feed::delete_record),
        )
        .route(
            "/xrpc/com.atproto.feed.like.create",
            post(routes::feed::create_like),
        )
        .route(
            "/xrpc/com.atproto.uploadBlob",
            post(routes::blob::upload_blob)
                .layer(routes::blob::upload_blob_body_limit())
                .layer::<_, std::convert::Infallible>(axum::middleware::from_fn(
                    routes::blob::body_limit_fallback,
                )),
        )
        .route(
            "/xrpc/com.atproto.sync.getRepo",
            get(routes::sync::get_repo),
        )
        .route(
            "/xrpc/com.atproto.sync.getBlocks",
            get(routes::sync::get_blocks),
        )
        .route(
            "/xrpc/com.atproto.sync.getLatestCommit",
            get(routes::sync::get_latest_commit),
        )
        .route(
            "/xrpc/com.atproto.sync.getRecord",
            get(routes::sync::get_record),
        )
        .route(
            "/xrpc/app.bsky.actor.profile.get",
            get(routes::profile::get_profile),
        )
        .route(
            "/xrpc/app.bsky.actor.profile.set",
            post(routes::profile::set_profile),
        )
        .route(
            "/xrpc/com.atproto.sync.listRepos",
            get(routes::sync::list_repos),
        )
        .route(
            "/xrpc/com.atproto.sync.getBlob",
            get(routes::blob::get_blob),
        )
        // The firehose. A WebSocket upgrade arrives as a plain GET, so this
        // is a normal `get` route whose handler happens to return an
        // upgrade response.
        .route(
            "/xrpc/com.atproto.sync.subscribeRepos",
            get(routes::subscribe_repos::subscribe_repos),
        )
        .route(
            "/blob/:cid",
            get(routes::blob::get_blob_by_cid),
        )
        .with_state(state)
}

async fn root() -> Json<serde_json::Value> {
    Json(json!({
        "name": "maarcadetweet-pds",
        "version": env!("CARGO_PKG_VERSION"),
    }))
}

async fn healthz() -> Json<serde_json::Value> {
    Json(json!({ "ok": true }))
}

/// `GET /.well-known/did.json` — the PDS's own DID document.
///
/// This is how the AppView (and any other relying party) learns the
/// P-256 public key that the access JWTs in
/// `Authorization: Bearer …` are signed with. Without it the AppView
/// could not verify a token at all, and the only alternative would be
/// shipping `PDS_JWT_SECRET` to a second service — a private signing
/// key crossing a service boundary, for a check that needs nothing but
/// the public half.
///
/// Nothing in this response is secret. `publicKeyMultibase` is the
/// uncompressed P-256 point derived from `PDS_JWT_SECRET` by
/// [`jwt_issuer::server_p256_public_multibase`]; the secret itself
/// never leaves this process.
///
/// The document id is [`AppConfig::pds_did`], i.e. it follows
/// `PDS_PUBLIC_URL` — so a `did:web:` resolver that starts from the DID,
/// rebuilds the URL and fetches this path lands back here rather than at
/// some other host's document.
async fn did_document(State(state): State<AppState>) -> Result<Json<serde_json::Value>, (axum::http::StatusCode, Json<serde_json::Value>)> {
    let did = state.cfg.pds_did();
    let public_multibase = jwt_issuer::server_p256_public_multibase(&state.cfg).map_err(|e| {
        // A malformed `PDS_JWT_SECRET` is the one way this fails, and
        // it is exactly the failure that also breaks every token this
        // server issues — surface it instead of publishing a document
        // with a missing key.
        (
            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({
                "error": "InternalServerError",
                "message": format!("server key unavailable: {e}"),
            })),
        )
    })?;
    Ok(Json(json!({
        "@context": [
            "https://www.w3.org/ns/did/v1",
            "https://w3id.org/security/multikey/v1",
        ],
        "id": did,
        "verificationMethod": [{
            // `#atproto` is the fragment AT Proto uses for a repo's
            // signing key; we reuse it for the server key so a generic
            // did:web consumer finds it in the usual place.
            "id": format!("{did}#atproto"),
            "type": "Multikey",
            "controller": did,
            "publicKeyMultibase": public_multibase,
        }],
        "service": [{
            "id": "#atproto_pds",
            "type": "AtprotoPersonalDataServer",
            "serviceEndpoint": state.cfg.pds_public_url,
        }],
    })))
}

async fn describe_server(State(state): State<AppState>) -> Json<DescribeServerResp> {
    Json(DescribeServerResp {
        // Derived from `PDS_PUBLIC_URL`, never hardcoded — see
        // `AppConfig::pds_did`. The same value ids the document at
        // `/.well-known/did.json`.
        did: state.cfg.pds_did(),
        available_user_domains: vec![state
            .cfg
            .pds_handle_dns_zone
            .trim_start_matches('.')
            .to_string()],
        // The real switch, not a hardcoded `false`. A client reads this
        // to decide whether to ask the user for a code *before*
        // collecting a handle and password — advertising `false` on a
        // server that then answers `400 InvalidInviteCode` sends the
        // user back to the start of a form they already filled in.
        invite_code_required: state.cfg.pds_invite_required,
        links: json!({
            "termsOfService": null,
            "privacyPolicy": null,
        }),
    })
}