A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253use axum::body::Bytes;use mlua::{Lua, LuaSerdeExt, Result as LuaResult};use std::sync::Arc;
use crate::AppState;use crate::db::{adapt_sql, now_rfc3339};use crate::profile;
/// Opaque handle to blob bytes stored on the Rust side./// Lua scripts receive this from `atproto.blob_download()` and pass it/// to `atproto.blob_upload()` — the binary data never enters the Lua VM.#[derive(Clone)]pub(crate) struct BlobHandle { pub data: Bytes, pub mime_type: String,}
impl mlua::UserData for BlobHandle { fn add_methods<M: mlua::UserDataMethods<Self>>(methods: &mut M) { methods.add_method("size", |_, this, ()| Ok(this.data.len())); methods.add_method("mime_type", |_, this, ()| Ok(this.mime_type.clone())); }}
/// Register the `atproto` table with AT Protocol utility functions.////// When `caller_did` is provided, the `atproto.sign(record)` function is/// available for inline attestation signing.pub fn register_atproto_api( lua: &Lua, state: Arc<AppState>, caller_did: Option<&str>,) -> LuaResult<()> { let atproto_table = lua.create_table()?;
let state_clone = state.clone(); let resolve_fn = lua.create_async_function(move |_lua, did: String| { let state = state_clone.clone(); async move { let result = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, &did).await;
match result { Ok(endpoint) => Ok(Some(endpoint)), Err(_) => Ok(None), } } })?;
atproto_table.set("resolve_service_endpoint", resolve_fn)?;
// atproto.blob_download(did, cid) -> { handle = BlobHandle, mimeType = string, size = number } { let state_clone = state.clone(); let blob_download_fn = lua.create_async_function(move |lua, (did, cid): (String, String)| { let state = state_clone.clone(); async move { let pds_endpoint = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, &did) .await .map_err(|e| { mlua::Error::runtime(format!( "blob_download: failed to resolve PDS for {did}: {e}" )) })?;
let url = format!( "{}/xrpc/com.atproto.sync.getBlob?did={}&cid={}", pds_endpoint, urlencoding::encode(&did), urlencoding::encode(&cid), );
let response = state.http.get(&url).send().await.map_err(|e| { mlua::Error::runtime(format!("blob_download: request failed: {e}")) })?;
let status = response.status(); if !status.is_success() { return Err(mlua::Error::runtime(format!( "blob_download: PDS returned {status} for did={did} cid={cid}" ))); }
let mime_type = response .headers() .get("content-type") .and_then(|v| v.to_str().ok()) .unwrap_or("application/octet-stream") .to_string();
let bytes = response.bytes().await.map_err(|e| { mlua::Error::runtime(format!("blob_download: failed to read body: {e}")) })?;
let size = bytes.len(); let handle = BlobHandle { data: bytes, mime_type: mime_type.clone(), };
let result = lua.create_table()?; result.set("handle", lua.create_userdata(handle)?)?; result.set("mimeType", mime_type)?; result.set("size", size)?;
Ok(mlua::Value::Table(result)) } })?; atproto_table.set("blob_download", blob_download_fn)?; }
// get_labels(uri) -> array of { src, uri, val, cts } let state_clone = state.clone(); let get_labels_fn = lua.create_async_function(move |lua, uri: String| { let state = state_clone.clone(); async move { let backend = state.db_backend; let now = now_rfc3339(); let sql = adapt_sql( "SELECT src, uri, val, cts FROM happyview_labels WHERE uri = ? AND (exp IS NULL OR exp > ?)", backend, ); let rows: Vec<(String, String, String, String)> = crate::db::query_as(&sql) .bind(&uri) .bind(&now) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("label query failed: {e}")))?;
let result = lua.create_table()?; let mut idx = 1;
for (src, label_uri, val, cts) in &rows { let label = lua.create_table()?; label.set("src", src.as_str())?; label.set("uri", label_uri.as_str())?; label.set("val", val.as_str())?; label.set("cts", cts.as_str())?; result.set(idx, label)?; idx += 1; }
// Check for self-labels in the record itself. let record_sql = adapt_sql("SELECT did, record FROM happyview_records WHERE uri = ?", backend); let record: Option<(String, String)> = crate::db::query_as(&record_sql) .bind(&uri) .fetch_optional(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("record query failed: {e}")))?;
if let Some((did, record_str)) = record { let record_val: serde_json::Value = serde_json::from_str(&record_str).unwrap_or(serde_json::json!({})); if let Some(labels) = record_val.get("labels") && let Some(values) = labels.get("values") && let Some(arr) = values.as_array() { for item in arr { if let Some(val) = item.get("val").and_then(|v| v.as_str()) { let label = lua.create_table()?; label.set("src", did.as_str())?; label.set("uri", uri.as_str())?; label.set("val", val)?; label.set("cts", "")?; result.set(idx, label)?; idx += 1; } } } }
Ok(mlua::Value::Table(result)) } })?; atproto_table.set("get_labels", get_labels_fn)?;
// get_labels_batch(uris) -> table keyed by URI let state_clone = state.clone(); let get_labels_batch_fn = lua.create_async_function(move |lua, uris: mlua::Table| { let state = state_clone.clone(); async move { let backend = state.db_backend; // Collect URIs from the Lua table. let uri_list: Vec<String> = uris .sequence_values::<String>() .collect::<Result<Vec<_>, _>>()?;
let now = now_rfc3339();
// Query labels for all URIs (one query per URI since AnyPool doesn't support array binding). let label_sql = adapt_sql( "SELECT src, uri, val, cts FROM happyview_labels WHERE uri = ? AND (exp IS NULL OR exp > ?)", backend, ); let mut rows: Vec<(String, String, String, String)> = Vec::new(); for uri in &uri_list { let mut uri_rows: Vec<(String, String, String, String)> = crate::db::query_as(&label_sql) .bind(uri) .bind(&now) .fetch_all(&state.db) .await .map_err(|e| { mlua::Error::runtime(format!("label batch query failed: {e}")) })?; rows.append(&mut uri_rows); }
// Query records for self-labels. let record_sql = adapt_sql( "SELECT uri, did, record FROM happyview_records WHERE uri = ?", backend, ); let mut records: Vec<(String, String, String)> = Vec::new(); for uri in &uri_list { let mut uri_records: Vec<(String, String, String)> = crate::db::query_as(&record_sql) .bind(uri) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("record batch query failed: {e}")))?; records.append(&mut uri_records); }
// Build result table keyed by URI. let result = lua.create_table()?;
// Initialize empty arrays for each URI. let mut counters: std::collections::HashMap<String, i32> = std::collections::HashMap::new(); for uri in &uri_list { result.set(uri.as_str(), lua.create_table()?)?; counters.insert(uri.clone(), 1); }
// Add external labels. for (src, uri, val, cts) in &rows { let label = lua.create_table()?; label.set("src", src.as_str())?; label.set("uri", uri.as_str())?; label.set("val", val.as_str())?; label.set("cts", cts.as_str())?;
let uri_table: mlua::Table = result.get(uri.as_str())?; let idx = counters.get(uri).copied().unwrap_or(1); uri_table.set(idx, label)?; counters.insert(uri.clone(), idx + 1); }
// Add self-labels from records. for (uri, did, record_str) in &records { let record_val: serde_json::Value = serde_json::from_str(record_str).unwrap_or(serde_json::json!({})); if let Some(labels) = record_val.get("labels") && let Some(values) = labels.get("values") && let Some(arr) = values.as_array() { for item in arr { if let Some(val) = item.get("val").and_then(|v| v.as_str()) { let label = lua.create_table()?; label.set("src", did.as_str())?; label.set("uri", uri.as_str())?; label.set("val", val)?; label.set("cts", "")?;
let uri_table: mlua::Table = result.get(uri.as_str())?; let idx = counters.get(uri).copied().unwrap_or(1); uri_table.set(idx, label)?; counters.insert(uri.clone(), idx + 1); } } } }
Ok(mlua::Value::Table(result)) } })?; atproto_table.set("get_labels_batch", get_labels_batch_fn)?;
// atproto.sign(record_table) -> inline signature object or nil // // Signs a record using the attestation signer and returns the inline // signature object ({ $type, key, signature: { $bytes } }). // Returns nil if no signer is configured. if let Some(signer) = &state.attestation_signer { let signer = signer.clone(); let did = caller_did.unwrap_or("").to_string(); let sign_fn = lua.create_function(move |lua, table: mlua::Value| { let mut record: serde_json::Value = lua .from_value(table) .map_err(|e| mlua::Error::runtime(format!("atproto.sign: {e}")))?;
signer .sign_record(&mut record, &did) .map_err(|e| mlua::Error::runtime(format!("atproto.sign: {e}")))?;
// Extract the last signature (the one we just added) let sig = record .get("signatures") .and_then(|s| s.as_array()) .and_then(|arr| arr.last()) .cloned() .ok_or_else(|| mlua::Error::runtime("atproto.sign: no signature produced"))?;
lua.to_value(&sig) .map_err(|e| mlua::Error::runtime(format!("atproto.sign: {e}"))) })?; atproto_table.set("sign", sign_fn)?; }
// atproto.verify_signature(record_table, sig_table, repository_did) -> boolean // // Verifies that an inline signature was produced by this HappyView instance. // Recomputes the CID and verifies the ECDSA signature. if let Some(signer) = &state.attestation_signer { let signer = signer.clone(); let verify_fn = lua.create_function( move |lua, (record, sig, repo_did): (mlua::Value, mlua::Value, String)| { let record_json: serde_json::Value = lua .from_value(record) .map_err(|e| mlua::Error::runtime(format!("atproto.verify_signature: {e}")))?; let sig_json: serde_json::Value = lua .from_value(sig) .map_err(|e| mlua::Error::runtime(format!("atproto.verify_signature: {e}")))?;
// `false` and an error are different facts, and only `false` // is a statement about the record: it means we checked and the // signature does not match. An error means we could not check // — malformed signature bytes, a missing field, a record that // will not encode. Collapsing the second into the first would // let any fault in this path present to a script as "this user // forged their records", with nothing in the logs to say // otherwise. Callers that want the old behaviour can `pcall`. signer .verify_record_signature(&record_json, &sig_json, &repo_did) .map_err(|e| { tracing::warn!( repository = %repo_did, error = %e, "atproto.verify_signature could not check the signature — \ this is not a statement that the record is forged" ); mlua::Error::runtime(format!("atproto.verify_signature: {e}")) }) }, )?; atproto_table.set("verify_signature", verify_fn)?; }
// atproto.spaces sub-table let spaces_table = lua.create_table()?;
// atproto.spaces.is_member(space_uri, did) -> boolean let state_clone = state.clone(); let is_member_fn = lua.create_async_function(move |_lua, (space_uri, did): (String, String)| { let state = state_clone.clone(); async move { if !crate::feature_flags::is_enabled( &state.db, crate::feature_flags::FeatureFlag::SPACES_ENABLED, state.db_backend, ) .await { return Err(mlua::Error::runtime("spaces feature is not enabled")); } let uri = crate::spaces::SpaceUri::parse(&space_uri) .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, &uri.did, &uri.type_nsid, &uri.skey, ) .await .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; let space = match space { Some(s) => s, None => return Ok(false), }; let access = crate::spaces::members::is_member(&state.db, state.db_backend, &space.id, &did) .await .map_err(|e| { mlua::Error::runtime(format!("membership check failed: {e}")) })?; Ok(access.is_some()) } })?; spaces_table.set("is_member", is_member_fn)?;
// atproto.spaces.get_access(space_uri, did) -> 'read' | 'write' | nil let state_clone = state.clone(); let get_access_fn = lua.create_async_function(move |_lua, (space_uri, did): (String, String)| { let state = state_clone.clone(); async move { if !crate::feature_flags::is_enabled( &state.db, crate::feature_flags::FeatureFlag::SPACES_ENABLED, state.db_backend, ) .await { return Err(mlua::Error::runtime("spaces feature is not enabled")); } let uri = crate::spaces::SpaceUri::parse(&space_uri) .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, &uri.did, &uri.type_nsid, &uri.skey, ) .await .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; let space = match space { Some(s) => s, None => return Ok(None), }; let access = crate::spaces::members::is_member(&state.db, state.db_backend, &space.id, &did) .await .map_err(|e| { mlua::Error::runtime(format!("membership check failed: {e}")) })?; Ok(access.map(|a| a.as_str().to_string())) } })?; spaces_table.set("get_access", get_access_fn)?;
// atproto.spaces.list_members(space_uri) -> array of { did, access } let state_clone = state.clone(); let list_members_fn = lua.create_async_function(move |lua, space_uri: String| { let state = state_clone.clone(); async move { if !crate::feature_flags::is_enabled( &state.db, crate::feature_flags::FeatureFlag::SPACES_ENABLED, state.db_backend, ) .await { return Err(mlua::Error::runtime("spaces feature is not enabled")); } let uri = crate::spaces::SpaceUri::parse(&space_uri) .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, &uri.did, &uri.type_nsid, &uri.skey, ) .await .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; let space = match space { Some(s) => s, None => { return Err(mlua::Error::runtime("space not found")); } }; let members = crate::spaces::members::resolve_members(&state.db, state.db_backend, &space.id) .await .map_err(|e| mlua::Error::runtime(format!("member resolution failed: {e}")))?;
let result = lua.create_table()?; for (i, member) in members.iter().enumerate() { let entry = lua.create_table()?; entry.set("did", member.did.as_str())?; entry.set("access", member.access.as_str())?; result.set(i + 1, entry)?; } Ok(mlua::Value::Table(result)) } })?; spaces_table.set("list_members", list_members_fn)?;
// atproto.spaces.query({ space_uri, collection, limit, cursor }) -> { records, cursor } let state_clone = state.clone(); let query_fn = lua.create_async_function(move |lua, opts: mlua::Table| { let state = state_clone.clone(); async move { if !crate::feature_flags::is_enabled( &state.db, crate::feature_flags::FeatureFlag::SPACES_ENABLED, state.db_backend, ) .await { return Err(mlua::Error::runtime("spaces feature is not enabled")); } let space_uri: String = opts .get("space_uri") .map_err(|_| mlua::Error::runtime("space_uri is required"))?; let collection: Option<String> = opts.get("collection").ok(); let limit: i64 = opts.get("limit").unwrap_or(50); let cursor: Option<String> = opts.get("cursor").ok();
let uri = crate::spaces::SpaceUri::parse(&space_uri) .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, &uri.did, &uri.type_nsid, &uri.skey, ) .await .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; let space = match space { Some(s) => s, None => { return Err(mlua::Error::runtime("space not found")); } };
let (records, next_cursor) = crate::spaces::db::list_space_records( &state.db, state.db_backend, &space.id, None, collection.as_deref(), limit.min(100), cursor.as_deref(), false, ) .await .map_err(|e| mlua::Error::runtime(format!("record query failed: {e}")))?;
let result = lua.create_table()?; let records_table = lua.create_table()?; for (i, record) in records.iter().enumerate() { let entry = lua.to_value(&serde_json::json!({ "uri": record.uri, "collection": record.collection, "rkey": record.rkey, "record": record.record, "cid": record.cid, "authorDid": record.author_did, }))?; records_table.set(i + 1, entry)?; } result.set("records", records_table)?; match next_cursor { Some(c) => result.set("cursor", c)?, None => result.set("cursor", mlua::Value::Nil)?, }
Ok(mlua::Value::Table(result)) } })?; spaces_table.set("query", query_fn)?;
atproto_table.set("spaces", spaces_table)?;
lua.globals().set("atproto", atproto_table)?; Ok(())}
/// Register blob upload capability on the existing `atproto` table.////// Called only in procedure execution contexts where PDS auth is/// available. `blob_download` is registered in `register_atproto_api`/// (available everywhere); `blob_upload` needs caller credentials to/// write to the caller's PDS.pub(crate) fn register_atproto_blob_api( lua: &Lua, state: Arc<AppState>, claims: Arc<crate::auth::Claims>, pds_auth: Arc<crate::repo::PdsAuth>,) -> LuaResult<()> { let atproto_table: mlua::Table = lua.globals().get("atproto")?;
let upload_fn = lua.create_async_function( move |lua, (handle, content_type): (mlua::AnyUserData, String)| { let state = state.clone(); let claims = claims.clone(); let pds_auth = pds_auth.clone(); async move { let blob_handle = handle.borrow::<BlobHandle>().map_err(|_| { mlua::Error::runtime( "blob_upload: first argument must be a BlobHandle from blob_download()", ) })?; let blob_bytes = blob_handle.data.clone(); drop(blob_handle);
let result = upload_blob_to_pds(&state, claims.did(), &pds_auth, &content_type, blob_bytes) .await .map_err(|e| mlua::Error::runtime(format!("blob_upload: {e}")))?;
lua.to_value(&result) } }, )?; atproto_table.set("blob_upload", upload_fn)?;
Ok(())}
async fn upload_blob_to_pds( state: &AppState, caller_did: &str, pds_auth: &crate::repo::PdsAuth, content_type: &str, blob_bytes: Bytes,) -> Result<serde_json::Value, crate::error::AppError> { use crate::error::AppError; use crate::repo::PdsAuth;
match pds_auth { PdsAuth::OAuth(session) => { use atrium_xrpc::{ InputDataOrBytes, OutputDataOrBytes, XrpcClient, XrpcRequest, http::Method, };
let request = XrpcRequest { method: Method::POST, nsid: "com.atproto.repo.uploadBlob".to_string(), parameters: None::<()>, input: Some(InputDataOrBytes::<()>::Bytes(blob_bytes.to_vec())), encoding: Some(content_type.to_string()), };
let result: Result< OutputDataOrBytes<serde_json::Value>, atrium_xrpc::Error<serde_json::Value>, > = session.send_xrpc(&request).await;
match result { Ok(OutputDataOrBytes::Data(data)) => Ok(data), Ok(OutputDataOrBytes::Bytes(bytes)) => serde_json::from_slice(&bytes) .map_err(|e| AppError::Internal(format!("invalid uploadBlob response: {e}"))), Err(e) => Err(AppError::Internal(format!("PDS uploadBlob failed: {e}"))), } } PdsAuth::Dpop { api_client_id, dpop_key_id, encryption_key, } => { let resp = crate::oauth::pds_write::dpop_pds_post_blob( &state.http, &state.db, state.db_backend, encryption_key, &state.oauth, &state.config.plc_url, api_client_id, caller_did, dpop_key_id, content_type, blob_bytes, ) .await?;
let status = resp.status(); let body = resp .bytes() .await .map_err(|e| AppError::Internal(format!("failed to read upload response: {e}")))?;
if !status.is_success() { let body_str = String::from_utf8_lossy(&body); return Err(AppError::Internal(format!( "PDS uploadBlob returned {status}: {body_str}" ))); }
serde_json::from_slice(&body) .map_err(|e| AppError::Internal(format!("invalid uploadBlob response: {e}"))) } }}
#[cfg(test)]mod tests { use super::*; use crate::config::Config; use crate::db::DatabaseBackend; use crate::lexicon::LexiconRegistry; use tokio::sync::watch;
fn test_state_with_plc(plc_url: &str) -> AppState { let config = Config { host: "127.0.0.1".into(), port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), user_agent: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), relay_url: String::new(), plc_url: plc_url.to_string(), static_dir: String::new(), base_path: None, event_log_retention_days: 30, app_name: None, logo_uri: None, tos_uri: None, policy_uri: None, token_encryption_key: None, default_rate_limit_capacity: 100, default_rate_limit_refill_rate: 2.0, telemetry_collector_url: String::new(), }; let (tx, _) = watch::channel(vec![]); let (labeler_tx, _) = watch::channel(()); sqlx::any::install_default_drivers(); let test_db = sqlx::AnyPool::connect_lazy("sqlite::memory:").unwrap(); let atrium_http = std::sync::Arc::new(crate::http_retry::HappyViewHttpClient::default()); let did_resolver = atrium_identity::did::CommonDidResolver::new( atrium_identity::did::CommonDidResolverConfig { plc_directory_url: "https://plc.directory".into(), http_client: std::sync::Arc::clone(&atrium_http), }, ); let handle_resolver = atrium_identity::handle::AtprotoHandleResolver::new( atrium_identity::handle::AtprotoHandleResolverConfig { dns_txt_resolver: crate::dns::NativeDnsResolver::new(), http_client: atrium_http, }, ); let oauth = atrium_oauth::OAuthClient::new(atrium_oauth::OAuthClientConfig { client_metadata: atrium_oauth::AtprotoLocalhostClientMetadata { redirect_uris: Some(vec!["http://127.0.0.1:0/auth/callback".into()]), scopes: Some(vec![atrium_oauth::Scope::Known( atrium_oauth::KnownScope::Atproto, )]), }, keys: None, state_store: crate::auth::oauth_store::DbStateStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), session_store: crate::auth::oauth_store::DbSessionStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), resolver: atrium_oauth::OAuthResolverConfig { did_resolver, handle_resolver, authorization_server_metadata: Default::default(), protected_resource_metadata: Default::default(), }, http_client: crate::http_retry::HappyViewHttpClient::default(), }) .expect("Failed to create test OAuth client"); AppState { config, http: reqwest::Client::new(), db: test_db.clone(), backfill_db: test_db.clone(), db_backend: DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), collections_tx: tx, labeler_subscriptions_tx: labeler_tx, rate_limiter: crate::rate_limit::RateLimiter::new( crate::rate_limit::RateLimitDefaults { query_cost: 1, procedure_cost: 1, proxy_cost: 1, }, ), oauth: std::sync::Arc::new(crate::auth::OAuthClientRegistry::new(std::sync::Arc::new( oauth, ))), oauth_state_store: crate::auth::oauth_store::DbStateStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), linked_repos_client: std::sync::Arc::new( crate::linked_repos::client::build( "https://plc.directory", "http://127.0.0.1:0/oauth-client-metadata.json", "http://127.0.0.1:0", "http://127.0.0.1:0/auth/callback".into(), true, vec![atrium_oauth::Scope::Known( atrium_oauth::KnownScope::Atproto, )], crate::auth::oauth_store::DbStateStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), test_db.clone(), crate::db::DatabaseBackend::Sqlite, None, ) .expect("Failed to create test linked-repo OAuth client"), ), linked_repos_client_kid: None, cookie_key: axum_extra::extract::cookie::Key::derive_from( b"test-secret-for-tests-only-not-production", ), plugin_registry: std::sync::Arc::new(crate::plugin::PluginRegistry::new()), wasm_runtime: std::sync::Arc::new( crate::plugin::WasmRuntime::new().expect("wasm runtime"), ), attestation_signer: None, official_registry: std::sync::Arc::new(tokio::sync::RwLock::new( crate::plugin::official_registry::OfficialRegistryState::default(), )), official_registry_config: crate::plugin::official_registry::RegistryConfig::production( ), proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), backfill_events_tx: tokio::sync::broadcast::channel(16).0, verbose_event_logging: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), client_jwks: Vec::new(), telemetry_counters: std::sync::Arc::new(crate::telemetry::counters::Counters::new()), } }
#[tokio::test] async fn resolve_service_endpoint_returns_endpoint() { let mock = wiremock::MockServer::start().await;
let did_doc = serde_json::json!({ "id": "did:plc:test123", "alsoKnownAs": ["at://test.example.com"], "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": "https://pds.example.com" }] });
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:test123")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(&did_doc)) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let chunk = r#"return atproto.resolve_service_endpoint("did:plc:test123")"#; let result: String = lua.load(chunk).eval_async().await.unwrap(); assert_eq!(result, "https://pds.example.com"); }
#[tokio::test] async fn resolve_service_endpoint_returns_nil_on_failure() { let mock = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:unknown")) .respond_with(wiremock::ResponseTemplate::new(404)) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let chunk = r#"return atproto.resolve_service_endpoint("did:plc:unknown")"#; let result: mlua::Value = lua.load(chunk).eval_async().await.unwrap(); assert!(matches!(result, mlua::Value::Nil)); }
#[tokio::test] async fn resolve_did_web() { let mock = wiremock::MockServer::start().await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let chunk = r#"return type(atproto.resolve_service_endpoint)"#; let result: String = lua.load(chunk).eval_async().await.unwrap(); assert_eq!(result, "function"); }
fn test_state_with_signer(plc_url: &str) -> AppState { let mut state = test_state_with_plc(plc_url); state.attestation_signer = Some(Arc::new( crate::plugin::attestation::AttestationSigner::for_testing( "did:web:test.example#signing".to_string(), "test.signature".to_string(), ), )); state }
#[tokio::test] async fn sign_returns_signature_object() { let state = test_state_with_signer(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), Some("did:plc:caller")).unwrap();
let chunk = r#" local record = { contributionType = "correction", changes = { name = "Test" } } local sig = atproto.sign(record) return sig.key "#; let result: String = lua.load(chunk).eval_async().await.unwrap(); assert_eq!(result, "did:web:test.example#signing"); }
#[tokio::test] async fn sign_returns_nil_without_signer() { let state = test_state_with_plc(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), Some("did:plc:caller")).unwrap();
let chunk = r#"return atproto.sign ~= nil"#; let result: bool = lua.load(chunk).eval_async().await.unwrap(); // sign should not be registered when no signer is configured assert!(!result); }
#[tokio::test] async fn verify_signature_roundtrip() { let state = test_state_with_signer(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), Some("did:plc:caller")).unwrap();
let chunk = r#" local record = { contributionType = "correction", changes = { name = "Test" } } local sig = atproto.sign(record) return atproto.verify_signature(record, sig, "did:plc:caller") "#; let result: bool = lua.load(chunk).eval_async().await.unwrap(); assert!(result); }
#[tokio::test] async fn verify_signature_rejects_wrong_did() { let state = test_state_with_signer(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), Some("did:plc:caller")).unwrap();
let chunk = r#" local record = { contributionType = "correction", changes = { name = "Test" } } local sig = atproto.sign(record) return atproto.verify_signature(record, sig, "did:plc:wrong") "#; let result: bool = lua.load(chunk).eval_async().await.unwrap(); assert!(!result); }
#[tokio::test] async fn verify_signature_rejects_tampered_record() { let state = test_state_with_signer(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), Some("did:plc:caller")).unwrap();
let chunk = r#" local record = { contributionType = "correction", changes = { name = "Original" } } local sig = atproto.sign(record) record.changes.name = "Tampered" return atproto.verify_signature(record, sig, "did:plc:caller") "#; let result: bool = lua.load(chunk).eval_async().await.unwrap(); assert!(!result); }
/// "We checked and it does not match" and "we could not check" are /// different facts, and only the first is a statement about the record. /// A script that cannot tell them apart will accuse a user of forgery /// because of a decode bug. #[tokio::test] async fn verify_signature_distinguishes_unverifiable_from_invalid() { let state = test_state_with_signer(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), Some("did:plc:caller")).unwrap();
let chunk = r#" local record = { contributionType = "correction", changes = { name = "Original" } } local sig = atproto.sign(record)
-- A well-formed signature over a different payload: genuinely invalid. local tampered = { contributionType = "correction", changes = { name = "Tampered" } } local mismatch_ok, mismatch = pcall(atproto.verify_signature, tampered, sig, "did:plc:caller")
-- Signature bytes that are not base64 at all: unverifiable. sig.signature["$bytes"] = "not!valid!base64" local undecodable_ok, undecodable = pcall(atproto.verify_signature, record, sig, "did:plc:caller")
return mismatch_ok, mismatch, undecodable_ok, tostring(undecodable) "#; let (mismatch_ok, mismatch, undecodable_ok, undecodable): (bool, bool, bool, String) = lua.load(chunk).eval_async().await.unwrap();
assert!(mismatch_ok, "a mismatch is an answer, not a failure"); assert!(!mismatch, "a signature over a different payload is invalid"); assert!( !undecodable_ok, "an unverifiable signature must not be reported as invalid" ); assert!( undecodable.contains("invalid base64"), "the error must name the cause, got: {undecodable}" ); }
#[tokio::test] async fn spaces_api_is_registered() { let state = test_state_with_plc(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let chunk = r#" return type(atproto.spaces) == "table" and type(atproto.spaces.is_member) == "function" and type(atproto.spaces.get_access) == "function" and type(atproto.spaces.list_members) == "function" and type(atproto.spaces.query) == "function" "#; let result: bool = lua.load(chunk).eval_async().await.unwrap(); assert!(result); }
#[tokio::test] async fn blob_handle_exposes_size_and_mime() { let state = test_state_with_plc(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let handle = BlobHandle { data: axum::body::Bytes::from_static(b"hello world"), mime_type: "text/plain".to_string(), }; lua.globals() .set("test_handle", lua.create_userdata(handle).unwrap()) .unwrap();
let size: usize = lua .load("return test_handle:size()") .eval_async() .await .unwrap(); assert_eq!(size, 11);
let mime: String = lua .load("return test_handle:mime_type()") .eval_async() .await .unwrap(); assert_eq!(mime, "text/plain"); }
#[tokio::test] async fn blob_download_returns_handle_and_metadata() { let mock = wiremock::MockServer::start().await;
let did_doc = serde_json::json!({ "id": "did:plc:blobsource", "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": mock.uri() }] }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:blobsource")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(&did_doc)) .mount(&mock) .await;
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.sync.getBlob")) .and(wiremock::matchers::query_param("did", "did:plc:blobsource")) .and(wiremock::matchers::query_param("cid", "bafytest123")) .respond_with( wiremock::ResponseTemplate::new(200) .insert_header("content-type", "image/png") .set_body_bytes(vec![0x89, 0x50, 0x4E, 0x47]), ) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let chunk = r#" local result = atproto.blob_download("did:plc:blobsource", "bafytest123") return { size = result.handle:size(), mimeType = result.mimeType, } "#; let result: mlua::Table = lua.load(chunk).eval_async().await.unwrap(); assert_eq!(result.get::<usize>("size").unwrap(), 4); assert_eq!(result.get::<String>("mimeType").unwrap(), "image/png"); }
#[tokio::test] async fn blob_download_throws_on_404() { let mock = wiremock::MockServer::start().await;
let did_doc = serde_json::json!({ "id": "did:plc:blobsource", "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": mock.uri() }] }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:blobsource")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(&did_doc)) .mount(&mock) .await;
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.sync.getBlob")) .respond_with(wiremock::ResponseTemplate::new(404)) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let result: Result<mlua::Value, _> = lua .load(r#"return atproto.blob_download("did:plc:blobsource", "bafymissing")"#) .eval_async() .await; assert!(result.is_err()); }
#[tokio::test] async fn blob_download_defaults_mime_to_octet_stream() { let mock = wiremock::MockServer::start().await;
let did_doc = serde_json::json!({ "id": "did:plc:blobsource", "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": mock.uri() }] }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:blobsource")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(&did_doc)) .mount(&mock) .await;
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.sync.getBlob")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(vec![0xFF, 0xD8])) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let chunk = r#" local result = atproto.blob_download("did:plc:blobsource", "bafynoheader") return result.mimeType "#; let result: String = lua.load(chunk).eval_async().await.unwrap(); assert_eq!(result, "application/octet-stream"); }
#[tokio::test] async fn blob_download_throws_on_429() { let mock = wiremock::MockServer::start().await;
let did_doc = serde_json::json!({ "id": "did:plc:blobsource", "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": mock.uri() }] }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:blobsource")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(&did_doc)) .mount(&mock) .await;
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.sync.getBlob")) .respond_with(wiremock::ResponseTemplate::new(429)) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let result: Result<mlua::Value, _> = lua .load(r#"return atproto.blob_download("did:plc:blobsource", "bafyratelimit")"#) .eval_async() .await; assert!(result.is_err()); let err_msg = result.unwrap_err().to_string(); assert!( err_msg.contains("429"), "error should mention 429 status: {err_msg}" ); }
#[tokio::test] async fn blob_download_throws_on_did_resolution_failure() { let mock = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/did:plc:nonexistent")) .respond_with(wiremock::ResponseTemplate::new(404)) .mount(&mock) .await;
let state = test_state_with_plc(&mock.uri()); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let result: Result<mlua::Value, _> = lua .load(r#"return atproto.blob_download("did:plc:nonexistent", "bafytest")"#) .eval_async() .await; assert!(result.is_err()); let err_msg = result.unwrap_err().to_string(); assert!( err_msg.contains("resolve PDS"), "error should mention PDS resolution failure: {err_msg}" ); }
#[tokio::test] async fn blob_upload_throws_without_auth() { let state = test_state_with_plc(""); let lua = mlua::Lua::new(); register_atproto_api(&lua, Arc::new(state), None).unwrap();
let has_upload: bool = lua .load("return atproto.blob_upload ~= nil") .eval_async() .await .unwrap(); assert!(!has_upload); }}