implement purge-did admin route properly

Cargo.toml

@@ -7,7 +7,7 @@ edition = "2021"
 anyhow = "1.0.93"
 atrium-api = { version = "0.24.8", default-features = false, features = ["tokio"] }
 bytes = { version = "1.8.0", features = ["serde"] }
-clap = { version = "4.5.21", features = ["derive"] }
+clap = { version = "4.5.21", features = ["derive", "env"] }
 ecdsa = { version = "0.16.9", features = ["verifying"] }
 fastwebsockets = { version = "0.8.0", features = ["hyper", "unstable-split", "upgrade"] }
 http-body-util = "0.1.2"

src/admin.rs

@@ -1,12 +1,19 @@
 use std::{io::Cursor, sync::Arc};
 
-use anyhow::{bail, Result};
+use anyhow::{Context, Result};
 use atrium_api::com::atproto::sync::subscribe_repos;
-use hyper::{body::Incoming, Request};
+use bytes::Buf;
+use http_body_util::BodyExt;
+use hyper::{body::Incoming, Request, Response, StatusCode};
 use ipld_core::ipld::Ipld;
+use serde::Deserialize;
 use serde_ipld_dagcbor::DecodeError;
 
-use crate::{http::ServerResponse, wire_proto::StreamEventHeader, AppState};
+use crate::{
+    http::{body_full, ServerResponse},
+    wire_proto::StreamEventHeader,
+    AppState,
+};
 
 pub fn purge_did(app: &AppState, did: &str) -> Result<()> {
     // drop commits
@@ -45,16 +52,55 @@ pub fn purge_did(app: &AppState, did: &str) -> Result<()> {
     Ok(())
 }
 
-// TODO: ban host
-
 pub async fn handle_purge_did(
-    _app: Arc<AppState>,
-    _req: Request<Incoming>,
+    app: Arc<AppState>,
+    req: Request<Incoming>,
 ) -> Result<ServerResponse> {
-    // TODO:
-    //   - validate admin Authorization header
-    //   - parse JSON body for target did
-    //   - run purge_did function
+    // TODO: we should abstract all of the response building tbh its annoyinngggg.
+    // u should be able to return some Err and get back a nicely-formatted error response
 
-    bail!("not yet implemented")
+    if app.admin_password.is_none() {
+        return Ok(Response::builder()
+            .status(StatusCode::SERVICE_UNAVAILABLE)
+            .header("Content-Type", "text/plain")
+            .body(body_full("admin routes are unavailable :("))?);
+    }
+
+    let Some(auth_header) = req.headers().get(hyper::header::AUTHORIZATION) else {
+        return Ok(Response::builder()
+            .status(StatusCode::UNAUTHORIZED)
+            .header("Content-Type", "text/plain")
+            .body(body_full("missing authorization header :c"))?);
+    };
+
+    let given_password = auth_header.to_str()?.strip_prefix("Bearer ");
+    if given_password.is_none() || given_password != app.admin_password.as_deref() {
+        return Ok(Response::builder()
+            .status(StatusCode::FORBIDDEN)
+            .header("Content-Type", "text/plain")
+            .body(body_full("invalid credentials for admin route >:("))?);
+    }
+
+    #[derive(Deserialize)]
+    struct PurgeDidBody {
+        did: String,
+    }
+
+    let body = req.collect().await?.aggregate();
+    let body = match serde_json::from_reader::<_, PurgeDidBody>(body.reader()) {
+        Ok(body) => body,
+        Err(_) => {
+            // TODO: surely we can build out an XRPC abstraction or something
+            return Ok(Response::builder()
+                .status(400)
+                .header("Content-Type", "text/plain")
+                .body(body_full("failed to parse request body as JSON :("))?);
+        }
+    };
+    purge_did(&app, &body.did)?;
+
+    Ok(Response::builder()
+        .status(200)
+        .header("Content-Type", "text/plain")
+        .body(body_full("o7"))?)
 }

src/app_state.rs

@@ -14,6 +14,8 @@ pub struct AppState {
     pub db_banned_hosts: sled::Tree,
 
     pub plc_resolver: Cow<'static, str>,
+    pub admin_password: Option<String>,
+
     pub known_good_hosts: Mutex<BTreeSet<String>>,
     pub active_indexers: Mutex<BTreeSet<String>>,
 
@@ -36,6 +38,7 @@ impl AppState {
             raw_block_tx,
 
             plc_resolver: Cow::Borrowed("plc.directory"),
+            admin_password: None,
             known_good_hosts: Mutex::new(hosts.into_iter().collect()),
             active_indexers: Default::default(),
 

src/main.rs

@@ -28,6 +28,9 @@ struct Args {
 
     #[arg(long)]
     drop_user_cache: bool,
+
+    #[arg(env = "ADMIN_PASSWORD")]
+    admin_password: Option<String>,
 }
 
 #[tokio::main]
@@ -56,6 +59,7 @@ async fn main() -> Result<()> {
     if let Some(plc_directory) = args.plc_resolver {
         server.plc_resolver = Cow::Owned(plc_directory);
     }
+    server.admin_password = args.admin_password;
 
     let server = Arc::new(server);
     let initial_hosts: Vec<String> = {