Something went wrong. Try again.
⚘ use your pds as a git remote if you want to ⚘
Something went wrong. Try again.
16 kB · 489 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490//! HTTP client for the PDS XRPC API.//!//! Wraps the AT Protocol XRPC endpoints needed for git backup://! record CRUD, blob upload/download, and session creation.
use std::cell::RefCell;
use atproto_identity::key::KeyData;
use crate::dpop;use crate::types::{BlobRef, CidLink};use serde::{Deserialize, Serialize};
/// Authentication mode for PDS requests.enum AuthMode { /// No authentication (unauthenticated reads). None, /// Classic Bearer token (from createSession). Bearer(String), /// DPoP-bound OAuth token (requires proof on every request). DPoP { access_token: String, key_data: KeyData, /// Cached nonce from the PDS (updated after each nonce challenge). last_nonce: RefCell<Option<String>>, },}
/// Client for interacting with a PDS server over XRPC.////// Supports three auth modes:/// - Unauthenticated (clone/fetch)/// - Bearer token (createSession-based push)/// - DPoP (OAuth-based push with proof-of-possession)pub struct PdsClient { /// base URL of the PDS, e.g. "https://bsky.social" base_url: String, auth: AuthMode, http: reqwest::Client,}
/// Response from `com.atproto.repo.getRecord`.#[derive(Debug, Deserialize)]pub struct GetRecordResponse { pub uri: String, pub cid: Option<String>, pub value: serde_json::Value,}
/// Response from `com.atproto.repo.uploadBlob`.#[derive(Debug, Deserialize)]pub struct UploadBlobResponse { pub blob: UploadedBlob,}
/// The blob object returned inside an `uploadBlob` response.#[derive(Debug, Deserialize)]pub struct UploadedBlob { #[serde(rename = "$type")] pub blob_type: String, #[serde(rename = "ref")] pub link: CidLink, #[serde(rename = "mimeType")] pub mime_type: String, pub size: u64,}
/// Request body for `com.atproto.repo.putRecord`.#[derive(Debug, Serialize)]pub struct PutRecordRequest { pub repo: String, pub collection: String, pub rkey: String, pub record: serde_json::Value, /// swap with the current CID to prevent race conditions #[serde(rename = "swapRecord", skip_serializing_if = "Option::is_none")] pub swap_record: Option<String>,}
/// Response from `com.atproto.repo.putRecord`.#[derive(Debug, Deserialize)]pub struct PutRecordResponse { pub uri: String, pub cid: String,}
/// Response from `com.atproto.server.createSession`.#[derive(Debug, Deserialize)]pub struct CreateSessionResponse { pub did: String, #[serde(rename = "accessJwt")] pub access_jwt: String, #[serde(rename = "refreshJwt")] pub refresh_jwt: String, pub handle: String,}
/// Response from `com.atproto.server.refreshSession`.#[derive(Debug, Deserialize)]pub struct RefreshSessionResponse { pub did: String, #[serde(rename = "accessJwt")] pub access_jwt: String, #[serde(rename = "refreshJwt")] pub refresh_jwt: String, pub handle: String,}
/// Error response from PDS XRPC endpoints.#[derive(Debug, Deserialize)]pub struct XrpcError { pub error: String, #[serde(default)] pub message: String,}
impl std::fmt::Display for XrpcError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}: {}", self.error, self.message) }}
impl PdsClient { /// Creates an unauthenticated client for the given PDS base URL. pub fn new(base_url: impl Into<String>) -> Self { Self { base_url: base_url.into().trim_end_matches('/').to_string(), auth: AuthMode::None, http: reqwest::Client::new(), } }
/// Creates an authenticated client with a bearer token. pub fn with_auth(base_url: impl Into<String>, token: impl Into<String>) -> Self { Self { base_url: base_url.into().trim_end_matches('/').to_string(), auth: AuthMode::Bearer(token.into()), http: reqwest::Client::new(), } }
/// Creates an authenticated client with a DPoP-bound OAuth token. pub fn with_dpop_auth( base_url: impl Into<String>, access_token: impl Into<String>, key_data: KeyData, ) -> Self { Self { base_url: base_url.into().trim_end_matches('/').to_string(), auth: AuthMode::DPoP { access_token: access_token.into(), key_data, last_nonce: RefCell::new(None), }, http: reqwest::Client::new(), } }
/// Sets or replaces the auth token (Bearer mode). pub fn set_auth(&mut self, token: impl Into<String>) { self.auth = AuthMode::Bearer(token.into()); }
/// Returns the base URL of the PDS. pub fn base_url(&self) -> &str { &self.base_url }
/// Logs in with handle/password and stores the resulting access token. /// /// Calls `com.atproto.server.createSession`. pub async fn login( &mut self, identifier: &str, password: &str, ) -> Result<CreateSessionResponse, String> { let url = format!("{}/xrpc/com.atproto.server.createSession", self.base_url);
let body = serde_json::json!({ "identifier": identifier, "password": password, });
let resp = self .http .post(&url) .json(&body) .send() .await .map_err(|e| format!("createSession request failed: {}", e))?;
if !resp.status().is_success() { let err = parse_xrpc_error(resp).await; return Err(format!("createSession failed: {}", err)); }
let session: CreateSessionResponse = resp .json() .await .map_err(|e| format!("failed to parse createSession response: {}", e))?;
self.auth = AuthMode::Bearer(session.access_jwt.clone()); Ok(session) }
/// Refreshes a Bearer session using a refresh JWT. /// /// Calls `com.atproto.server.refreshSession` with the refresh token /// in the Authorization header. Returns new access and refresh JWTs. pub async fn refresh_session( http: &reqwest::Client, base_url: &str, refresh_jwt: &str, ) -> Result<RefreshSessionResponse, String> { let url = format!( "{}/xrpc/com.atproto.server.refreshSession", base_url.trim_end_matches('/') );
let resp = http .post(&url) .bearer_auth(refresh_jwt) .send() .await .map_err(|e| format!("refreshSession request failed: {}", e))?;
if !resp.status().is_success() { let err = parse_xrpc_error(resp).await; return Err(format!("refreshSession failed: {}", err)); }
resp.json() .await .map_err(|e| format!("failed to parse refreshSession response: {}", e)) }
/// Fetches a record from the PDS. /// /// Calls `com.atproto.repo.getRecord`. Returns `Ok(None)` if the /// record does not exist (RecordNotFound). pub async fn get_record( &self, did: &str, collection: &str, rkey: &str, ) -> Result<Option<GetRecordResponse>, String> { let url = format!( "{}/xrpc/com.atproto.repo.getRecord?repo={}&collection={}&rkey={}", self.base_url, did, collection, rkey );
let resp = self .http .get(&url) .send() .await .map_err(|e| format!("getRecord request failed: {}", e))?;
// 400 with RecordNotFound means the record doesn't exist yet if resp.status() == reqwest::StatusCode::BAD_REQUEST { let body = resp.text().await.unwrap_or_default(); if body.contains("RecordNotFound") { return Ok(None); } return Err(format!("getRecord failed: {}", body)); }
if !resp.status().is_success() { let err = parse_xrpc_error(resp).await; return Err(format!("getRecord failed: {}", err)); }
let record: GetRecordResponse = resp .json() .await .map_err(|e| format!("failed to parse getRecord response: {}", e))?;
Ok(Some(record)) }
/// Creates or updates a record on the PDS. /// /// Calls `com.atproto.repo.putRecord`. Requires authentication. pub async fn put_record( &self, did: &str, collection: &str, rkey: &str, record: serde_json::Value, swap_record: Option<String>, ) -> Result<PutRecordResponse, String> { let url = format!("{}/xrpc/com.atproto.repo.putRecord", self.base_url);
let body = PutRecordRequest { repo: did.to_string(), collection: collection.to_string(), rkey: rkey.to_string(), record, swap_record, };
let json_body = serde_json::to_vec(&body) .map_err(|e| format!("failed to serialize putRecord body: {}", e))?;
let resp = self .send_authenticated("POST", &url, json_body.clone(), Some("application/json")) .await?;
if !resp.status().is_success() { let err = parse_xrpc_error(resp).await; return Err(format!("putRecord failed: {}", err)); }
resp.json() .await .map_err(|e| format!("failed to parse putRecord response: {}", e)) }
/// Uploads a blob to the PDS. /// /// Calls `com.atproto.repo.uploadBlob`. Requires authentication. /// Returns a `BlobRef` that can be embedded in a record. pub async fn upload_blob(&self, data: Vec<u8>) -> Result<BlobRef, String> { let url = format!("{}/xrpc/com.atproto.repo.uploadBlob", self.base_url);
let resp = self .send_authenticated("POST", &url, data.clone(), Some("application/octet-stream")) .await?;
if !resp.status().is_success() { let err = parse_xrpc_error(resp).await; return Err(format!("uploadBlob failed: {}", err)); }
let upload: UploadBlobResponse = resp .json() .await .map_err(|e| format!("failed to parse uploadBlob response: {}", e))?;
Ok(BlobRef { blob_type: upload.blob.blob_type, link: upload.blob.link, mime_type: upload.blob.mime_type, size: upload.blob.size, }) }
/// Downloads a blob from the PDS. /// /// Calls `com.atproto.sync.getBlob`. This is unauthenticated — /// anyone can download blobs if they know the DID and CID. pub async fn get_blob(&self, did: &str, cid: &str) -> Result<Vec<u8>, String> { let url = format!( "{}/xrpc/com.atproto.sync.getBlob?did={}&cid={}", self.base_url, did, cid );
let resp = self .http .get(&url) .send() .await .map_err(|e| format!("getBlob request failed: {}", e))?;
if !resp.status().is_success() { let err = parse_xrpc_error(resp).await; return Err(format!("getBlob failed: {}", err)); }
resp.bytes() .await .map(|b| b.to_vec()) .map_err(|e| format!("failed to read getBlob response body: {}", e)) }
/// Sends an authenticated request, handling both Bearer and DPoP auth modes. /// /// For DPoP auth, generates proof headers and automatically retries once /// if the PDS responds with a nonce challenge (400/401 with DPoP-Nonce header). async fn send_authenticated( &self, method: &str, url: &str, body: Vec<u8>, content_type: Option<&str>, ) -> Result<reqwest::Response, String> { match &self.auth { AuthMode::None => Err("authenticated request requires auth".to_string()), AuthMode::Bearer(token) => { let mut req = self.http.post(url).bearer_auth(token); if let Some(ct) = content_type { req = req.header("Content-Type", ct); } req.body(body) .send() .await .map_err(|e| format!("request failed: {}", e)) } AuthMode::DPoP { access_token, key_data, last_nonce, } => { let nonce = last_nonce.borrow().clone(); let proof = dpop::make_dpop_proof(key_data, method, url, access_token, nonce.as_deref())?;
let mut req = self .http .post(url) .header("Authorization", format!("DPoP {}", access_token)) .header("DPoP", &proof); if let Some(ct) = content_type { req = req.header("Content-Type", ct); }
let resp = req .body(body.clone()) .send() .await .map_err(|e| format!("request failed: {}", e))?;
// Check for nonce challenge: 400/401 with DPoP-Nonce header let status = resp.status(); if (status == reqwest::StatusCode::BAD_REQUEST || status == reqwest::StatusCode::UNAUTHORIZED) && resp.headers().contains_key("dpop-nonce") { let new_nonce = resp .headers() .get("dpop-nonce") .and_then(|v| v.to_str().ok()) .map(|s| s.to_string());
if let Some(ref nonce_val) = new_nonce { tracing::debug!("DPoP nonce challenge received, retrying with nonce");
// Cache the nonce for subsequent requests *last_nonce.borrow_mut() = new_nonce.clone();
// Regenerate proof with nonce and retry let retry_proof = dpop::make_dpop_proof( key_data, method, url, access_token, Some(nonce_val), )?;
let mut retry_req = self .http .post(url) .header("Authorization", format!("DPoP {}", access_token)) .header("DPoP", &retry_proof); if let Some(ct) = content_type { retry_req = retry_req.header("Content-Type", ct); }
return retry_req .body(body) .send() .await .map_err(|e| format!("request failed (retry): {}", e)); } }
// Cache any nonce from successful responses too if let Some(nonce_val) = resp .headers() .get("dpop-nonce") .and_then(|v| v.to_str().ok()) { *last_nonce.borrow_mut() = Some(nonce_val.to_string()); }
Ok(resp) } } }}
/// Extracts an error message from a non-success XRPC response.async fn parse_xrpc_error(resp: reqwest::Response) -> String { let status = resp.status(); match resp.json::<XrpcError>().await { Ok(err) => format!("{} ({})", err, status), Err(_) => format!("HTTP {}", status), }}