From 5fe2b21877a68df55fda709e4182e35f9efc7daf Mon Sep 17 00:00:00 2001 From: Chris van der Loo Date: Sat, 5 Sep 2026 14:01:38 -0400 Subject: [PATCH] feat: access private ATProto Spaces without public fallback --- packages/server/src/files/repository.rs | 132 +++++++++++++++++++++ packages/server/src/pds/mod.rs | 2 + packages/server/src/pds/rpc.rs | 146 ++++++++++++++++++++++++ packages/server/src/pds/space.rs | 71 ++++++++++++ 4 files changed, 351 insertions(+) create mode 100644 packages/server/src/files/repository.rs create mode 100644 packages/server/src/pds/mod.rs create mode 100644 packages/server/src/pds/rpc.rs create mode 100644 packages/server/src/pds/space.rs diff --git a/packages/server/src/files/repository.rs b/packages/server/src/files/repository.rs new file mode 100644 index 0000000..e2c84b8 --- /dev/null +++ b/packages/server/src/files/repository.rs @@ -0,0 +1,132 @@ +use crate::{ + auth::Session, + error::{ApiError, Result}, + pds::rpc, +}; +use async_trait::async_trait; +use crate_protocol::Space; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +#[derive(Clone, Debug)] +pub struct Record { + pub id: String, + pub revision: String, + pub value: Value, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "$type")] +pub enum Write { + #[serde(rename = "com.atproto.space.applyWrites#create")] + Create { + collection: String, + rkey: String, + value: Value, + }, + #[serde(rename = "com.atproto.space.applyWrites#update")] + Update { + collection: String, + rkey: String, + value: Value, + }, + #[serde(rename = "com.atproto.space.applyWrites#delete")] + Delete { collection: String, rkey: String }, +} + +#[async_trait] +pub trait Repository: Sync { + async fn get(&self, collection: &str, id: &str) -> Result; + async fn list(&self, collection: &str) -> Result>; + async fn apply(&self, writes: Vec) -> Result<()>; +} + +pub struct PdsRepository<'a> { + pub session: &'a Session, + pub space: &'a Space, + pub did: &'a str, +} + +#[derive(Deserialize)] +struct RecordResponse { + cid: String, + value: Value, +} +#[derive(Deserialize)] +struct ListedRecord { + rkey: String, + cid: String, + value: Value, +} +#[derive(Deserialize)] +struct Listing { + records: Vec, + cursor: Option, +} + +#[async_trait] +impl Repository for PdsRepository<'_> { + async fn get(&self, collection: &str, id: &str) -> Result { + atrium_api::types::string::RecordKey::new(id.into()).map_err(ApiError::bad)?; + let output: RecordResponse = rpc::query( + self.session, + "com.atproto.space.getRecord", + json!({"space":self.space.uri,"repo":self.did,"collection":collection,"rkey":id}), + ) + .await?; + Ok(Record { + id: id.into(), + revision: output.cid, + value: output.value, + }) + } + + async fn list(&self, collection: &str) -> Result> { + let mut records = Vec::new(); + let mut cursor = None::; + loop { + let mut params = + json!({"space":self.space.uri,"repo":self.did,"collection":collection,"limit":100}); + if let Some(cursor) = &cursor { + params["cursor"] = json!(cursor); + } + let output = + rpc::query::(self.session, "com.atproto.space.listRecords", params).await; + let page = match output { + Ok(page) => page, + Err(error) if error.code == "RepoNotFound" || error.code == "SpaceNotFound" => { + return Ok(records); + } + Err(error) => return Err(error.into()), + }; + records.extend(page.records.into_iter().map(|record| Record { + id: record.rkey, + revision: record.cid, + value: record.value, + })); + if page.cursor.is_none() { + return Ok(records); + } + if page.cursor == cursor { + return Err(ApiError::upstream("The PDS repeated a pagination cursor.")); + } + cursor = page.cursor; + } + } + + async fn apply(&self, writes: Vec) -> Result<()> { + if writes.len() > 200 { + return Err(ApiError::too_large()); + } + let body = json!({"space":self.space.uri,"repo":self.did,"writes":writes}); + if serde_json::to_vec(&body) + .map_err(|_| ApiError::bad("Invalid write"))? + .len() + > 1_000_000 + { + return Err(ApiError::too_large()); + } + rpc::procedure::(self.session, "com.atproto.space.applyWrites", body).await?; + Ok(()) + } +} diff --git a/packages/server/src/pds/mod.rs b/packages/server/src/pds/mod.rs new file mode 100644 index 0000000..7f1fed0 --- /dev/null +++ b/packages/server/src/pds/mod.rs @@ -0,0 +1,2 @@ +pub mod rpc; +pub mod space; diff --git a/packages/server/src/pds/rpc.rs b/packages/server/src/pds/rpc.rs new file mode 100644 index 0000000..232da7d --- /dev/null +++ b/packages/server/src/pds/rpc.rs @@ -0,0 +1,146 @@ +use crate::{auth::Session, error::ApiError}; +use atrium_xrpc::{ + InputDataOrBytes, OutputDataOrBytes, XrpcClient, XrpcRequest, + error::{ErrorResponseBody, XrpcErrorKind}, + http::Method, +}; +use serde::{Serialize, de::DeserializeOwned}; +use serde_json::Value; + +#[derive(Debug)] +pub struct RpcError { + pub code: String, + pub message: String, + pub status: u16, +} + +pub async fn query( + session: &Session, + nsid: &str, + params: Value, +) -> Result { + let result = session + .send_xrpc::<_, (), T, ErrorResponseBody>(&XrpcRequest { + method: Method::GET, + nsid: nsid.into(), + parameters: Some(params), + input: None, + encoding: None, + }) + .await; + match result.map_err(convert)? { + OutputDataOrBytes::Data(data) => Ok(data), + _ => Err(malformed()), + } +} + +pub async fn procedure( + session: &Session, + nsid: &str, + body: impl Serialize + Send + Sync, +) -> Result { + let result = session + .send_xrpc::<(), _, T, ErrorResponseBody>(&XrpcRequest { + method: Method::POST, + nsid: nsid.into(), + parameters: None, + input: Some(InputDataOrBytes::Data(body)), + encoding: Some("application/json".into()), + }) + .await; + match result.map_err(convert)? { + OutputDataOrBytes::Data(data) => Ok(data), + _ => Err(malformed()), + } +} + +pub async fn upload( + session: &Session, + bytes: Vec, + mime: &str, +) -> Result { + let result = session + .send_xrpc::<(), (), T, ErrorResponseBody>(&XrpcRequest { + method: Method::POST, + nsid: "com.atproto.repo.uploadBlob".into(), + parameters: None, + input: Some(InputDataOrBytes::Bytes(bytes)), + encoding: Some(mime.into()), + }) + .await; + match result.map_err(convert)? { + OutputDataOrBytes::Data(data) => Ok(data), + _ => Err(malformed()), + } +} + +pub async fn download(session: &Session, params: Value) -> Result, RpcError> { + let result = session + .send_xrpc::<_, (), Value, ErrorResponseBody>(&XrpcRequest { + method: Method::GET, + nsid: "com.atproto.space.getBlob".into(), + parameters: Some(params), + input: None, + encoding: None, + }) + .await; + match result.map_err(convert)? { + OutputDataOrBytes::Bytes(data) => Ok(data), + _ => Err(malformed()), + } +} + +fn malformed() -> RpcError { + RpcError { + code: "InvalidPdsResponse".into(), + message: "The PDS returned an invalid response.".into(), + status: 502, + } +} + +fn convert(error: atrium_xrpc::Error) -> RpcError { + if let atrium_xrpc::Error::XrpcResponse(response) = error { + let details = match response.error { + Some(XrpcErrorKind::Custom(value) | XrpcErrorKind::Undefined(value)) => value, + None => ErrorResponseBody { + error: None, + message: None, + }, + }; + return RpcError { + code: details.error.unwrap_or_default(), + message: details + .message + .unwrap_or_else(|| "The PDS request failed.".into()), + status: response.status.as_u16(), + }; + } + RpcError { + code: "PdsUnavailable".into(), + message: "The PDS could not be reached. Your session is retained; retry shortly.".into(), + status: 502, + } +} + +impl From for ApiError { + fn from(error: RpcError) -> Self { + match (error.status, error.code.as_str()) { + (_, "RecordAlreadyExists") => Self::conflict( + "This revision or filename changed on another client. Refresh and retry.", + ), + (_, "RecordNotFound" | "BlobNotFound") => Self::not_found(), + (_, "MethodNotImplemented" | "XRPCNotSupported" | "NotImplemented") | (501, _) => { + Self::unavailable( + "This PDS does not support the required ATProto Spaces API. No data was made public.", + ) + } + (401, _) => Self::unauthorized(), + (403, _) => Self::forbidden("The PDS denied access to this private space."), + (413, _) => Self::too_large(), + (_, "SpaceNotFound" | "SpaceDeleted") => Self::unavailable( + "The private Crate space is unavailable. No public storage fallback is used.", + ), + _ => Self::upstream(error.message), + } + } +} diff --git a/packages/server/src/pds/space.rs b/packages/server/src/pds/space.rs new file mode 100644 index 0000000..56c58f0 --- /dev/null +++ b/packages/server/src/pds/space.rs @@ -0,0 +1,71 @@ +use super::rpc; +use crate::{ + auth::Session, + error::{ApiError, Result}, +}; +use crate_protocol::Space; +use serde::Deserialize; +use serde_json::{Value, json}; + +pub fn personal(did: &str) -> Space { + Space { + uri: format!("at://{did}/space/network.crate.drive/self"), + authority_did: did.into(), + space_type: "network.crate.drive".into(), + key: "self".into(), + } +} + +#[derive(Deserialize)] +struct SpaceInfo { + uri: String, + policy: Value, + #[serde(rename = "appAccess")] + app_access: Value, +} + +pub async fn exists(session: &Session, space: &Space) -> Result { + let response = rpc::query::( + session, + "com.atproto.simplespace.getSpace", + json!({"space":space.uri}), + ) + .await; + match response { + Ok(info) => { + if info.uri != space.uri + || info.policy["$type"] != "com.atproto.simplespace.defs#memberListPolicy" + { + return Err(ApiError::forbidden( + "Crate requires an explicit member-list space; public spaces are not accepted.", + )); + } + if info.app_access["$type"] != "com.atproto.simplespace.defs#open" { + return Err(ApiError::forbidden( + "This space restricts clients. Open app access is required for interoperable Crate clients.", + )); + } + Ok(true) + } + Err(error) if error.code == "SpaceNotFound" => Ok(false), + Err(error) => Err(error.into()), + } +} + +pub async fn ensure(session: &Session, space: &Space) -> Result<()> { + if exists(session, space).await? { + return Ok(()); + } + let response = rpc::procedure::(session, "com.atproto.simplespace.createSpace", json!({"type":space.space_type,"skey":space.key,"policy":{"$type":"com.atproto.simplespace.defs#memberListPolicy"},"appAccess":{"$type":"com.atproto.simplespace.defs#open"}})).await; + match response { + Ok(_) => {} + Err(error) if error.code == "SpaceAlreadyExists" => {} + Err(error) => return Err(error.into()), + } + if !exists(session, space).await? { + return Err(ApiError::unavailable( + "The PDS did not create the private space.", + )); + } + Ok(()) +} -- 2.51.2