+3 -2

Cargo.lock

··· 803 803 804 804 [[package]] 805 805 name = "bytes" 806 806 - version = "1.11.1" 806 806 + version = "1.10.1" 807 807 source = "registry+https://github.com/rust-lang/crates.io-index" 808 808 - checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" 808 808 + checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" 809 809 810 810 [[package]] 811 811 name = "byteview" ··· 5965 5965 "form_urlencoded", 5966 5966 "idna", 5967 5967 "percent-encoding", 5968 5968 + "serde", 5968 5969 ] 5969 5970 5970 5971 [[package]]

+1 -7

constellation/src/bin/main.rs

··· 45 45 #[arg(short, long)] 46 46 #[clap(value_enum, default_value_t = StorageBackend::Memory)] 47 47 backend: StorageBackend, 48 48 - /// Serve a did:web document for this domain 49 49 - #[arg(long)] 50 50 - did_web_domain: Option<String>, 51 48 /// Initiate a database backup into this dir, if supported by the storage 52 49 #[arg(long)] 53 50 backup: Option<PathBuf>, ··· 106 103 MemStorage::new(), 107 104 fixture, 108 105 None, 109 109 - args.did_web_domain, 110 106 stream, 111 107 bind, 112 108 metrics_bind, ··· 142 138 rocks, 143 139 fixture, 144 140 args.data, 145 145 - args.did_web_domain, 146 141 stream, 147 142 bind, 148 143 metrics_bind, ··· 164 159 mut storage: impl LinkStorage, 165 160 fixture: Option<PathBuf>, 166 161 data_dir: Option<PathBuf>, 167 167 - did_web_domain: Option<String>, 168 162 stream: String, 169 163 bind: SocketAddr, 170 164 metrics_bind: SocketAddr, ··· 217 211 if collect_metrics { 218 212 install_metrics_server(metrics_bind)?; 219 213 } 220 220 - serve(readable, bind, did_web_domain, staying_alive).await 214 214 + serve(readable, bind, staying_alive).await 221 215 }) 222 216 .unwrap(); 223 217 stay_alive.drop_guard();

+7 -32

constellation/src/server/mod.rs

··· 3 3 extract::{Query, Request}, 4 4 http::{self, header}, 5 5 middleware::{self, Next}, 6 6 - response::{IntoResponse, Json, Response}, 6 6 + response::{IntoResponse, Response}, 7 7 routing::get, 8 8 Router, 9 9 }; ··· 37 37 http::StatusCode::INTERNAL_SERVER_ERROR 38 38 } 39 39 40 40 - pub async fn serve<S: LinkReader, A: ToSocketAddrs>( 41 41 - store: S, 42 42 - addr: A, 43 43 - did_web_domain: Option<String>, 44 44 - stay_alive: CancellationToken, 45 45 - ) -> anyhow::Result<()> { 46 46 - let mut app = Router::new(); 47 47 - 48 48 - if let Some(d) = did_web_domain { 49 49 - app = app.route( 50 50 - "/.well-known/did.json", 51 51 - get({ 52 52 - let domain = d.clone(); 53 53 - move || did_web(domain) 54 54 - }), 55 55 - ) 56 56 - } 57 57 - 58 58 - let app = app 40 40 + pub async fn serve<S, A>(store: S, addr: A, stay_alive: CancellationToken) -> anyhow::Result<()> 41 41 + where 42 42 + S: LinkReader, 43 43 + A: ToSocketAddrs, 44 44 + { 45 45 + let app = Router::new() 59 46 .route("/robots.txt", get(robots)) 60 47 .route( 61 48 "/", ··· 217 204 User-agent: * 218 205 Disallow: /links 219 206 Disallow: /links/ 220 220 - Disallow: /xrpc/ 221 207 " 222 222 - } 223 223 - 224 224 - async fn did_web(domain: String) -> impl IntoResponse { 225 225 - Json(serde_json::json!({ 226 226 - "id": format!("did:web:{domain}"), 227 227 - "service": [{ 228 228 - "id": "#constellation", 229 229 - "type": "ConstellationGraphService", 230 230 - "serviceEndpoint": format!("https://{domain}") 231 231 - }] 232 232 - })) 233 208 } 234 209 235 210 #[derive(Template, Serialize, Deserialize)]

+1 -1

slingshot/Cargo.toml

··· 28 28 tokio = { version = "1.47.0", features = ["full"] } 29 29 tokio-util = "0.7.15" 30 30 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } 31 31 - url = "2.5.4" 31 31 + url = { version = "2.5.4", features = ["serde"] }

+10

slingshot/src/error.rs

··· 91 91 #[error("upstream non-atproto bad request")] 92 92 UpstreamBadBadNotGoodRequest(reqwest::Error), 93 93 } 94 94 + 95 95 + #[derive(Debug, Error)] 96 96 + pub enum ProxyError { 97 97 + #[error("failed to parse path: {0}")] 98 98 + PathParseError(String), 99 99 + #[error(transparent)] 100 100 + UrlParseError(#[from] url::ParseError), 101 101 + #[error(transparent)] 102 102 + ReqwestError(#[from] reqwest::Error), 103 103 + }

+2

slingshot/src/lib.rs

··· 3 3 mod firehose_cache; 4 4 mod healthcheck; 5 5 mod identity; 6 6 + mod proxy; 6 7 mod record; 7 8 mod server; 8 9 ··· 10 11 pub use firehose_cache::firehose_cache; 11 12 pub use healthcheck::healthcheck; 12 13 pub use identity::{Identity, IdentityKey}; 14 14 + pub use proxy::Proxy; 13 15 pub use record::{CachedRecord, ErrorResponseObject, Repo}; 14 16 pub use server::serve;

+4 -3

slingshot/src/main.rs

··· 2 2 // use foyer::{Engine, DirectFsDeviceOptions, HybridCacheBuilder}; 3 3 use metrics_exporter_prometheus::PrometheusBuilder; 4 4 use slingshot::{ 5 5 - Identity, Repo, consume, error::MainTaskError, firehose_cache, healthcheck, serve, 5 5 + Identity, Proxy, Repo, consume, error::MainTaskError, firehose_cache, healthcheck, serve, 6 6 }; 7 7 use std::net::SocketAddr; 8 8 use std::path::PathBuf; ··· 143 143 ) 144 144 .await 145 145 .map_err(|e| format!("identity setup failed: {e:?}"))?; 146 146 - 147 147 - log::info!("identity service ready."); 148 146 let identity_refresher = identity.clone(); 149 147 let identity_shutdown = shutdown.clone(); 150 148 tasks.spawn(async move { 151 149 identity_refresher.run_refresher(identity_shutdown).await?; 152 150 Ok(()) 153 151 }); 152 152 + log::info!("identity service ready."); 154 153 155 154 let repo = Repo::new(identity.clone()); 155 155 + let proxy = Proxy::new(repo.clone()); 156 156 157 157 let identity_for_server = identity.clone(); 158 158 let server_shutdown = shutdown.clone(); ··· 163 163 server_cache_handle, 164 164 identity_for_server, 165 165 repo, 166 166 + proxy, 166 167 args.acme_domain, 167 168 args.acme_contact, 168 169 args.acme_cache_path,

+505

slingshot/src/proxy.rs

··· 1 1 + use serde::Deserialize; 2 2 + use url::Url; 3 3 + use std::{collections::HashMap, time::Duration}; 4 4 + use crate::{Repo, server::HydrationSource, error::ProxyError}; 5 5 + use reqwest::Client; 6 6 + use serde_json::{Map, Value}; 7 7 + 8 8 + pub enum ParamValue { 9 9 + String(Vec<String>), 10 10 + Int(Vec<i64>), 11 11 + Bool(Vec<bool>), 12 12 + } 13 13 + pub struct Params(HashMap<String, ParamValue>); 14 14 + 15 15 + impl TryFrom<Map<String, Value>> for Params { 16 16 + type Error = (); // TODO 17 17 + fn try_from(val: Map<String, Value>) -> Result<Self, Self::Error> { 18 18 + let mut out = HashMap::new(); 19 19 + for (k, v) in val { 20 20 + match v { 21 21 + Value::String(s) => out.insert(k, ParamValue::String(vec![s])), 22 22 + Value::Bool(b) => out.insert(k, ParamValue::Bool(vec![b])), 23 23 + Value::Number(n) => { 24 24 + let Some(i) = n.as_i64() else { 25 25 + return Err(()); 26 26 + }; 27 27 + out.insert(k, ParamValue::Int(vec![i])) 28 28 + } 29 29 + Value::Array(a) => { 30 30 + let Some(first) = a.first() else { 31 31 + continue; 32 32 + }; 33 33 + if first.is_string() { 34 34 + let mut vals = Vec::with_capacity(a.len()); 35 35 + for v in a { 36 36 + let Some(v) = v.as_str() else { 37 37 + return Err(()); 38 38 + }; 39 39 + vals.push(v.to_string()); 40 40 + } 41 41 + out.insert(k, ParamValue::String(vals)); 42 42 + } else if first.is_i64() { 43 43 + let mut vals = Vec::with_capacity(a.len()); 44 44 + for v in a { 45 45 + let Some(v) = v.as_i64() else { 46 46 + return Err(()); 47 47 + }; 48 48 + vals.push(v); 49 49 + } 50 50 + out.insert(k, ParamValue::Int(vals)); 51 51 + } else if first.is_boolean() { 52 52 + let mut vals = Vec::with_capacity(a.len()); 53 53 + for v in a { 54 54 + let Some(v) = v.as_bool() else { 55 55 + return Err(()); 56 56 + }; 57 57 + vals.push(v); 58 58 + } 59 59 + out.insert(k, ParamValue::Bool(vals)); 60 60 + } 61 61 + todo!(); 62 62 + } 63 63 + _ => return Err(()), 64 64 + }; 65 65 + } 66 66 + 67 67 + Ok(Self(out)) 68 68 + } 69 69 + } 70 70 + 71 71 + #[derive(Clone)] 72 72 + pub struct Proxy { 73 73 + repo: Repo, 74 74 + client: Client, 75 75 + } 76 76 + 77 77 + impl Proxy { 78 78 + pub fn new(repo: Repo) -> Self { 79 79 + let client = Client::builder() 80 80 + .user_agent(format!( 81 81 + "microcosm slingshot v{} (contact: @bad-example.com)", 82 82 + env!("CARGO_PKG_VERSION") 83 83 + )) 84 84 + .no_proxy() 85 85 + .timeout(Duration::from_secs(6)) 86 86 + .build() 87 87 + .unwrap(); 88 88 + Self { repo, client } 89 89 + } 90 90 + 91 91 + pub async fn proxy( 92 92 + &self, 93 93 + xrpc: String, 94 94 + service: String, 95 95 + params: Option<Map<String, Value>>, 96 96 + ) -> Result<Value, ProxyError> { 97 97 + 98 98 + // hackin it to start 99 99 + 100 100 + // 1. assume did-web (TODO) and get the did doc 101 101 + #[derive(Debug, Deserialize)] 102 102 + struct ServiceDoc { 103 103 + id: String, 104 104 + service: Vec<ServiceItem>, 105 105 + } 106 106 + #[derive(Debug, Deserialize)] 107 107 + struct ServiceItem { 108 108 + id: String, 109 109 + #[expect(unused)] 110 110 + r#type: String, 111 111 + #[serde(rename = "serviceEndpoint")] 112 112 + service_endpoint: Url, 113 113 + } 114 114 + let dw = service.strip_prefix("did:web:").expect("a did web"); 115 115 + let (dw, service_id) = dw.split_once("#").expect("whatever"); 116 116 + let mut dw_url = Url::parse(&format!("https://{dw}"))?; 117 117 + dw_url.set_path("/.well-known/did.json"); 118 118 + let doc: ServiceDoc = self.client 119 119 + .get(dw_url) 120 120 + .send() 121 121 + .await? 122 122 + .error_for_status()? 123 123 + .json() 124 124 + .await?; 125 125 + 126 126 + assert_eq!(doc.id, format!("did:web:{}", dw)); 127 127 + 128 128 + let mut upstream = None; 129 129 + for ServiceItem { id, service_endpoint, .. } in doc.service { 130 130 + let Some((_, id)) = id.split_once("#") else { continue; }; 131 131 + if id != service_id { continue; }; 132 132 + upstream = Some(service_endpoint); 133 133 + break; 134 134 + } 135 135 + 136 136 + // 2. proxy the request forward 137 137 + let mut upstream = upstream.expect("to find it"); 138 138 + upstream.set_path(&format!("/xrpc/{xrpc}")); // TODO: validate nsid 139 139 + 140 140 + if let Some(params) = params { 141 141 + let mut query = upstream.query_pairs_mut(); 142 142 + let Params(ps) = params.try_into().expect("valid params"); 143 143 + for (k, pvs) in ps { 144 144 + match pvs { 145 145 + ParamValue::String(s) => { 146 146 + for s in s { 147 147 + query.append_pair(&k, &s); 148 148 + } 149 149 + } 150 150 + ParamValue::Int(i) => { 151 151 + for i in i { 152 152 + query.append_pair(&k, &i.to_string()); 153 153 + } 154 154 + } 155 155 + ParamValue::Bool(b) => { 156 156 + for b in b { 157 157 + query.append_pair(&k, &b.to_string()); 158 158 + } 159 159 + } 160 160 + } 161 161 + } 162 162 + } 163 163 + 164 164 + // TODO: other headers to proxy 165 165 + Ok(self.client 166 166 + .get(upstream) 167 167 + .send() 168 168 + .await? 169 169 + .error_for_status()? 170 170 + .json() 171 171 + .await?) 172 172 + } 173 173 + } 174 174 + 175 175 + #[derive(Debug, PartialEq)] 176 176 + pub enum PathPart { 177 177 + Scalar(String), 178 178 + Vector(String, Option<String>), // key, $type 179 179 + } 180 180 + 181 181 + pub fn parse_record_path(input: &str) -> Result<Vec<PathPart>, String> { 182 182 + let mut out = Vec::new(); 183 183 + 184 184 + let mut key_acc = String::new(); 185 185 + let mut type_acc = String::new(); 186 186 + let mut in_bracket = false; 187 187 + let mut chars = input.chars().enumerate(); 188 188 + while let Some((i, c)) = chars.next() { 189 189 + match c { 190 190 + '[' if in_bracket => return Err(format!("nested opening bracket not allowed, at {i}")), 191 191 + '[' if key_acc.is_empty() => return Err(format!("missing key before opening bracket, at {i}")), 192 192 + '[' => in_bracket = true, 193 193 + ']' if in_bracket => { 194 194 + in_bracket = false; 195 195 + let key = std::mem::take(&mut key_acc); 196 196 + let r#type = std::mem::take(&mut type_acc); 197 197 + let t = if r#type.is_empty() { None } else { Some(r#type) }; 198 198 + out.push(PathPart::Vector(key, t)); 199 199 + // peek ahead because we need a dot after array if there's more and i don't want to add more loop state 200 200 + let Some((i, c)) = chars.next() else { 201 201 + break; 202 202 + }; 203 203 + if c != '.' { 204 204 + return Err(format!("expected dot after close bracket, found {c:?} at {i}")); 205 205 + } 206 206 + } 207 207 + ']' => return Err(format!("unexpected close bracket at {i}")), 208 208 + '.' if in_bracket => type_acc.push(c), 209 209 + '.' if key_acc.is_empty() => return Err(format!("missing key before next segment, at {i}")), 210 210 + '.' => { 211 211 + let key = std::mem::take(&mut key_acc); 212 212 + assert!(type_acc.is_empty()); 213 213 + out.push(PathPart::Scalar(key)); 214 214 + } 215 215 + _ if in_bracket => type_acc.push(c), 216 216 + _ => key_acc.push(c), 217 217 + } 218 218 + } 219 219 + if in_bracket { 220 220 + return Err("unclosed bracket".into()); 221 221 + } 222 222 + if !key_acc.is_empty() { 223 223 + out.push(PathPart::Scalar(key_acc)); 224 224 + } 225 225 + Ok(out) 226 226 + } 227 227 + 228 228 + #[derive(Debug, Clone, PartialEq)] 229 229 + pub enum RefShape { 230 230 + StrongRef, 231 231 + AtUri, 232 232 + AtUriParts, 233 233 + Did, 234 234 + Handle, 235 235 + AtIdentifier, 236 236 + Blob, 237 237 + // TODO: blob with type? 238 238 + } 239 239 + 240 240 + impl TryFrom<&str> for RefShape { 241 241 + type Error = String; 242 242 + fn try_from(s: &str) -> Result<Self, Self::Error> { 243 243 + match s { 244 244 + "strong-ref" => Ok(Self::StrongRef), 245 245 + "at-uri" => Ok(Self::AtUri), 246 246 + "at-uri-parts" => Ok(Self::AtUriParts), 247 247 + "did" => Ok(Self::Did), 248 248 + "handle" => Ok(Self::Handle), 249 249 + "at-identifier" => Ok(Self::AtIdentifier), 250 250 + "blob" => Ok(Self::Blob), 251 251 + _ => Err(format!("unknown shape: {s}")), 252 252 + } 253 253 + } 254 254 + } 255 255 + 256 256 + #[derive(Debug, PartialEq)] 257 257 + pub enum MatchedRef { 258 258 + AtUri { 259 259 + uri: String, 260 260 + cid: Option<String>, 261 261 + }, 262 262 + Identifier(String), 263 263 + Blob { 264 264 + link: String, 265 265 + mime: String, 266 266 + size: u64, 267 267 + } 268 268 + } 269 269 + 270 270 + pub fn match_shape(shape: &RefShape, val: &Value) -> Option<MatchedRef> { 271 271 + // TODO: actually validate at-uri format 272 272 + // TODO: actually validate everything else also 273 273 + // TODO: should this function normalize identifiers to DIDs probably? 274 274 + // or just return at-uri parts so the caller can resolve and reassemble 275 275 + match shape { 276 276 + RefShape::StrongRef => { 277 277 + let o = val.as_object()?; 278 278 + let uri = o.get("uri")?.as_str()?.to_string(); 279 279 + let cid = o.get("cid")?.as_str()?.to_string(); 280 280 + Some(MatchedRef::AtUri { uri, cid: Some(cid) }) 281 281 + } 282 282 + RefShape::AtUri => { 283 283 + let uri = val.as_str()?.to_string(); 284 284 + Some(MatchedRef::AtUri { uri, cid: None }) 285 285 + } 286 286 + RefShape::AtUriParts => { 287 287 + let o = val.as_object()?; 288 288 + let identifier = o.get("repo").or(o.get("did"))?.as_str()?.to_string(); 289 289 + let collection = o.get("collection")?.as_str()?.to_string(); 290 290 + let rkey = o.get("rkey")?.as_str()?.to_string(); 291 291 + let uri = format!("at://{identifier}/{collection}/{rkey}"); 292 292 + let cid = o.get("cid").and_then(|v| v.as_str()).map(str::to_string); 293 293 + Some(MatchedRef::AtUri { uri, cid }) 294 294 + } 295 295 + RefShape::Did => { 296 296 + let id = val.as_str()?; 297 297 + if !id.starts_with("did:") { 298 298 + return None; 299 299 + } 300 300 + Some(MatchedRef::Identifier(id.to_string())) 301 301 + } 302 302 + RefShape::Handle => { 303 303 + let id = val.as_str()?; 304 304 + if id.contains(':') { 305 305 + return None; 306 306 + } 307 307 + Some(MatchedRef::Identifier(id.to_string())) 308 308 + } 309 309 + RefShape::AtIdentifier => { 310 310 + Some(MatchedRef::Identifier(val.as_str()?.to_string())) 311 311 + } 312 312 + RefShape::Blob => { 313 313 + let o = val.as_object()?; 314 314 + if o.get("$type")? != "blob" { 315 315 + return None; 316 316 + } 317 317 + let link = o.get("ref")?.as_object()?.get("$link")?.as_str()?.to_string(); 318 318 + let mime = o.get("mimeType")?.as_str()?.to_string(); 319 319 + let size = o.get("size")?.as_u64()?; 320 320 + Some(MatchedRef::Blob { link, mime, size }) 321 321 + } 322 322 + } 323 323 + } 324 324 + 325 325 + // TODO: send back metadata about the matching 326 326 + pub fn extract_links( 327 327 + sources: Vec<HydrationSource>, 328 328 + skeleton: &Value, 329 329 + ) -> Result<Vec<MatchedRef>, String> { 330 330 + // collect early to catch errors from the client 331 331 + // (TODO maybe the handler should do this and pass in the processed stuff probably definitely yeah) 332 332 + let sources = sources 333 333 + .into_iter() 334 334 + .map(|HydrationSource { path, shape }| { 335 335 + let path_parts = parse_record_path(&path)?; 336 336 + let shape: RefShape = shape.as_str().try_into()?; 337 337 + Ok((path_parts, shape)) 338 338 + }) 339 339 + .collect::<Result<Vec<_>, String>>()?; 340 340 + 341 341 + // lazy first impl, just re-walk the skeleton as many times as needed 342 342 + // not deduplicating for now 343 343 + let mut out = Vec::new(); 344 344 + for (path_parts, shape) in sources { 345 345 + for val in PathWalker::new(&path_parts, skeleton) { 346 346 + if let Some(matched) = match_shape(&shape, val) { 347 347 + out.push(matched); 348 348 + } 349 349 + } 350 350 + } 351 351 + 352 352 + Ok(out) 353 353 + } 354 354 + 355 355 + struct PathWalker<'a> { 356 356 + todo: Vec<(&'a [PathPart], &'a Value)>, 357 357 + } 358 358 + impl<'a> PathWalker<'a> { 359 359 + fn new(path_parts: &'a [PathPart], skeleton: &'a Value) -> Self { 360 360 + Self { todo: vec![(path_parts, skeleton)] } 361 361 + } 362 362 + } 363 363 + impl<'a> Iterator for PathWalker<'a> { 364 364 + type Item = &'a Value; 365 365 + fn next(&mut self) -> Option<Self::Item> { 366 366 + loop { 367 367 + let (parts, val) = self.todo.pop()?; 368 368 + let Some((part, rest)) = parts.split_first() else { 369 369 + return Some(val); 370 370 + }; 371 371 + let Some(o) = val.as_object() else { 372 372 + continue; 373 373 + }; 374 374 + match part { 375 375 + PathPart::Scalar(k) => { 376 376 + let Some(v) = o.get(k) else { 377 377 + continue; 378 378 + }; 379 379 + self.todo.push((rest, v)); 380 380 + } 381 381 + PathPart::Vector(k, t) => { 382 382 + let Some(a) = o.get(k).and_then(|v| v.as_array()) else { 383 383 + continue; 384 384 + }; 385 385 + for v in a 386 386 + .iter() 387 387 + .rev() 388 388 + .filter(|c| { 389 389 + let Some(t) = t else { return true }; 390 390 + c 391 391 + .as_object() 392 392 + .and_then(|o| o.get("$type")) 393 393 + .and_then(|v| v.as_str()) 394 394 + .map(|s| s == t) 395 395 + .unwrap_or(false) 396 396 + }) 397 397 + { 398 398 + self.todo.push((rest, v)) 399 399 + } 400 400 + } 401 401 + } 402 402 + } 403 403 + } 404 404 + } 405 405 + 406 406 + 407 407 + #[cfg(test)] 408 408 + mod tests { 409 409 + use super::*; 410 410 + use serde_json::json; 411 411 + 412 412 + #[test] 413 413 + fn test_parse_record_path() -> Result<(), Box<dyn std::error::Error>> { 414 414 + let cases = [ 415 415 + ("", vec![]), 416 416 + ("subject", vec![PathPart::Scalar("subject".into())]), 417 417 + ("authorDid", vec![PathPart::Scalar("authorDid".into())]), 418 418 + ("subject.uri", vec![PathPart::Scalar("subject".into()), PathPart::Scalar("uri".into())]), 419 419 + ("members[]", vec![PathPart::Vector("members".into(), None)]), 420 420 + ("add[].key", vec![ 421 421 + PathPart::Vector("add".into(), None), 422 422 + PathPart::Scalar("key".into()), 423 423 + ]), 424 424 + ("a[b]", vec![PathPart::Vector("a".into(), Some("b".into()))]), 425 425 + ("a[b.c]", vec![PathPart::Vector("a".into(), Some("b.c".into()))]), 426 426 + ("facets[app.bsky.richtext.facet].features[app.bsky.richtext.facet#mention].did", vec![ 427 427 + PathPart::Vector("facets".into(), Some("app.bsky.richtext.facet".into())), 428 428 + PathPart::Vector("features".into(), Some("app.bsky.richtext.facet#mention".into())), 429 429 + PathPart::Scalar("did".into()), 430 430 + ]), 431 431 + ]; 432 432 + 433 433 + for (path, expected) in cases { 434 434 + let parsed = parse_record_path(path)?; 435 435 + assert_eq!(parsed, expected, "path: {path:?}"); 436 436 + } 437 437 + 438 438 + Ok(()) 439 439 + } 440 440 + 441 441 + #[test] 442 442 + fn test_match_shape() { 443 443 + let cases = [ 444 444 + ("strong-ref", json!(""), None), 445 445 + ("strong-ref", json!({}), None), 446 446 + ("strong-ref", json!({ "uri": "abc" }), None), 447 447 + ("strong-ref", json!({ "cid": "def" }), None), 448 448 + ( 449 449 + "strong-ref", 450 450 + json!({ "uri": "abc", "cid": "def" }), 451 451 + Some(MatchedRef::AtUri { uri: "abc".to_string(), cid: Some("def".to_string()) }), 452 452 + ), 453 453 + ("at-uri", json!({ "uri": "abc" }), None), 454 454 + ("at-uri", json!({ "uri": "abc", "cid": "def" }), None), 455 455 + ( 456 456 + "at-uri", 457 457 + json!("abc"), 458 458 + Some(MatchedRef::AtUri { uri: "abc".to_string(), cid: None }), 459 459 + ), 460 460 + ("at-uri-parts", json!("abc"), None), 461 461 + ("at-uri-parts", json!({}), None), 462 462 + ( 463 463 + "at-uri-parts", 464 464 + json!({"repo": "a", "collection": "b", "rkey": "c"}), 465 465 + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: None }), 466 466 + ), 467 467 + ( 468 468 + "at-uri-parts", 469 469 + json!({"did": "a", "collection": "b", "rkey": "c"}), 470 470 + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: None }), 471 471 + ), 472 472 + ( 473 473 + "at-uri-parts", 474 474 + // 'repo' takes precedence over 'did' 475 475 + json!({"did": "a", "repo": "z", "collection": "b", "rkey": "c"}), 476 476 + Some(MatchedRef::AtUri { uri: "at://z/b/c".to_string(), cid: None }), 477 477 + ), 478 478 + ( 479 479 + "at-uri-parts", 480 480 + json!({"repo": "a", "collection": "b", "rkey": "c", "cid": "def"}), 481 481 + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: Some("def".to_string()) }), 482 482 + ), 483 483 + ( 484 484 + "at-uri-parts", 485 485 + json!({"repo": "a", "collection": "b", "rkey": "c", "cid": {}}), 486 486 + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: None }), 487 487 + ), 488 488 + ("did", json!({}), None), 489 489 + ("did", json!(""), None), 490 490 + ("did", json!("bad-example.com"), None), 491 491 + ("did", json!("did:plc:xyz"), Some(MatchedRef::Identifier("did:plc:xyz".to_string()))), 492 492 + ("handle", json!({}), None), 493 493 + ("handle", json!("bad-example.com"), Some(MatchedRef::Identifier("bad-example.com".to_string()))), 494 494 + ("handle", json!("did:plc:xyz"), None), 495 495 + ("at-identifier", json!({}), None), 496 496 + ("at-identifier", json!("bad-example.com"), Some(MatchedRef::Identifier("bad-example.com".to_string()))), 497 497 + ("at-identifier", json!("did:plc:xyz"), Some(MatchedRef::Identifier("did:plc:xyz".to_string()))), 498 498 + ]; 499 499 + for (shape, val, expected) in cases { 500 500 + let s = shape.try_into().unwrap(); 501 501 + let matched = match_shape(&s, &val); 502 502 + assert_eq!(matched, expected, "shape: {shape:?}, val: {val:?}"); 503 503 + } 504 504 + } 505 505 + }

+2 -2

slingshot/src/record.rs

··· 11 11 12 12 #[derive(Debug, Serialize, Deserialize)] 13 13 pub struct RawRecord { 14 14 - cid: Cid, 15 15 - record: String, 14 14 + pub cid: Cid, 15 15 + pub record: String, 16 16 } 17 17 18 18 // TODO: should be able to do typed CID

+326 -12

slingshot/src/server.rs

··· 1 1 use crate::{ 2 2 - CachedRecord, ErrorResponseObject, Identity, Repo, 2 2 + CachedRecord, ErrorResponseObject, Identity, Proxy, Repo, 3 3 error::{RecordError, ServerError}, 4 4 + proxy::{extract_links, MatchedRef}, 5 5 + record::RawRecord, 4 6 }; 5 7 use atrium_api::types::string::{Cid, Did, Handle, Nsid, RecordKey}; 6 8 use foyer::HybridCache; 7 9 use links::at_uri::parse_at_uri as normalize_at_uri; 8 10 use serde::Serialize; 9 9 - use std::path::PathBuf; 10 10 - use std::str::FromStr; 11 11 - use std::sync::Arc; 12 12 - use std::time::Instant; 11 11 + use std::{path::PathBuf, str::FromStr, sync::Arc, time::Instant, collections::HashMap}; 12 12 + use tokio::sync::mpsc; 13 13 use tokio_util::sync::CancellationToken; 14 14 15 15 use poem::{ ··· 24 24 }; 25 25 use poem_openapi::{ 26 26 ApiResponse, ContactObject, ExternalDocumentObject, Object, OpenApi, OpenApiService, Tags, 27 27 + Union, 27 28 param::Query, payload::Json, types::Example, 28 29 }; 29 30 ··· 54 55 "zQ3shpq1g134o7HGDb86CtQFxnHqzx5pZWknrVX2Waum3fF6j".to_string() 55 56 } 56 57 57 57 - #[derive(Object)] 58 58 + #[derive(Debug, Object)] 58 59 #[oai(example = true)] 59 60 struct XrpcErrorResponseObject { 60 61 /// Should correspond an error `name` in the lexicon errors array ··· 87 88 88 89 fn bad_request_handler_resolve_mini(err: poem::Error) -> ResolveMiniIDResponse { 89 90 ResolveMiniIDResponse::BadRequest(Json(XrpcErrorResponseObject { 91 91 + error: "InvalidRequest".to_string(), 92 92 + message: format!("Bad request, here's some info that maybe should not be exposed: {err}"), 93 93 + })) 94 94 + } 95 95 + 96 96 + fn bad_request_handler_proxy_query(err: poem::Error) -> ProxyHydrateResponse { 97 97 + ProxyHydrateResponse::BadRequest(Json(XrpcErrorResponseObject { 90 98 error: "InvalidRequest".to_string(), 91 99 message: format!("Bad request, here's some info that maybe should not be exposed: {err}"), 92 100 })) ··· 191 199 } 192 200 193 201 #[derive(Object)] 202 202 + struct ProxyHydrationError { 203 203 + reason: String, 204 204 + } 205 205 + 206 206 + #[derive(Object)] 207 207 + struct ProxyHydrationPending { 208 208 + url: String, 209 209 + } 210 210 + 211 211 + #[derive(Object)] 212 212 + struct ProxyHydrationRecordFound { 213 213 + record: serde_json::Value, 214 214 + } 215 215 + 216 216 + #[derive(Object)] 217 217 + struct ProxyHydrationIdentifierFound { 218 218 + mini_doc: MiniDocResponseObject, 219 219 + } 220 220 + 221 221 + #[derive(Object)] 222 222 + #[oai(rename_all = "camelCase")] 223 223 + struct ProxyHydrationBlobFound { 224 224 + /// cdn url 225 225 + link: String, 226 226 + mime_type: String, 227 227 + size: u64, 228 228 + } 229 229 + 230 230 + // todo: there's gotta be a supertrait that collects these? 231 231 + use poem_openapi::types::{Type, ToJSON, ParseFromJSON, IsObjectType}; 232 232 + 233 233 + #[derive(Union)] 234 234 + #[oai(discriminator_name = "status", rename_all = "camelCase")] 235 235 + enum Hydration<T: Send + Sync + Type + ToJSON + ParseFromJSON + IsObjectType> { 236 236 + Error(ProxyHydrationError), 237 237 + Pending(ProxyHydrationPending), 238 238 + Found(T), 239 239 + } 240 240 + 241 241 + #[derive(Object)] 242 242 + #[oai(example = true)] 243 243 + struct ProxyHydrateResponseObject { 244 244 + /// The original upstream response content 245 245 + output: serde_json::Value, 246 246 + /// Any hydrated records 247 247 + records: HashMap<String, Hydration<ProxyHydrationRecordFound>>, 248 248 + /// Any hydrated identifiers 249 249 + /// 250 250 + /// TODO: "identifiers" feels wrong as the name, probably "identities"? 251 251 + identifiers: HashMap<String, Hydration<ProxyHydrationIdentifierFound>>, 252 252 + /// Any hydrated blob CDN urls 253 253 + blobs: HashMap<String, Hydration<ProxyHydrationBlobFound>>, 254 254 + } 255 255 + impl Example for ProxyHydrateResponseObject { 256 256 + fn example() -> Self { 257 257 + Self { 258 258 + output: serde_json::json!({}), 259 259 + records: HashMap::from([ 260 260 + ("asdf".into(), Hydration::Pending(ProxyHydrationPending { url: "todo".into() })), 261 261 + ]), 262 262 + identifiers: HashMap::new(), 263 263 + blobs: HashMap::new(), 264 264 + } 265 265 + } 266 266 + } 267 267 + 268 268 + #[derive(ApiResponse)] 269 269 + #[oai(bad_request_handler = "bad_request_handler_proxy_query")] 270 270 + enum ProxyHydrateResponse { 271 271 + #[oai(status = 200)] 272 272 + Ok(Json<ProxyHydrateResponseObject>), 273 273 + #[oai(status = 400)] 274 274 + BadRequest(XrpcError) 275 275 + } 276 276 + 277 277 + #[derive(Object)] 278 278 + pub struct HydrationSource { 279 279 + /// Record Path syntax for locating fields 280 280 + pub path: String, 281 281 + /// What to expect at the path: 'strong-ref', 'at-uri', 'at-uri-parts', 'did', 'handle', or 'at-identifier'. 282 282 + /// 283 283 + /// - `strong-ref`: object in the shape of `com.atproto.repo.strongRef` with `uri` and `cid` keys. 284 284 + /// - `at-uri`: string, must have all segments present (identifier, collection, rkey) 285 285 + /// - `at-uri-parts`: object with keys (`repo` or `did`), `collection`, `rkey`, and optional `cid`. Other keys may be present and will be ignored. 286 286 + /// - `did`: string, `did` format 287 287 + /// - `handle`: string, `handle` format 288 288 + /// - `at-identifier`: string, `did` or `handle` format 289 289 + pub shape: String, 290 290 + } 291 291 + 292 292 + #[derive(Object)] 293 293 + #[oai(example = true)] 294 294 + struct ProxyQueryPayload { 295 295 + /// The NSID of the XRPC you wish to forward 296 296 + xrpc: String, 297 297 + /// The destination service the request will be forwarded to 298 298 + atproto_proxy: String, 299 299 + /// The `params` for the destination service XRPC endpoint 300 300 + /// 301 301 + /// Currently this will be passed along unchecked, but a future version of 302 302 + /// slingshot may attempt to do lexicon resolution to validate `params` 303 303 + /// based on the upstream service 304 304 + params: Option<serde_json::Value>, 305 305 + /// Paths within the response to look for at-uris that can be hydrated 306 306 + hydration_sources: Vec<HydrationSource>, 307 307 + // todo: deadline thing 308 308 + 309 309 + } 310 310 + impl Example for ProxyQueryPayload { 311 311 + fn example() -> Self { 312 312 + Self { 313 313 + xrpc: "app.bsky.feed.getFeedSkeleton".to_string(), 314 314 + atproto_proxy: "did:web:blue.mackuba.eu#bsky_fg".to_string(), 315 315 + params: Some(serde_json::json!({ 316 316 + "feed": "at://did:plc:oio4hkxaop4ao4wz2pp3f4cr/app.bsky.feed.generator/atproto", 317 317 + })), 318 318 + hydration_sources: vec![ 319 319 + HydrationSource { 320 320 + path: "feed[].post".to_string(), 321 321 + shape: "at-uri".to_string(), 322 322 + } 323 323 + ], 324 324 + } 325 325 + } 326 326 + } 327 327 + 328 328 + #[derive(Object)] 194 329 #[oai(example = true)] 195 330 struct FoundDidResponseObject { 196 331 /// the DID, bi-directionally verified if using Slingshot ··· 221 356 struct Xrpc { 222 357 cache: HybridCache<String, CachedRecord>, 223 358 identity: Identity, 359 359 + proxy: Arc<Proxy>, 224 360 repo: Arc<Repo>, 225 361 } 226 362 ··· 475 611 #[oai(example = "example_handle")] 476 612 Query(identifier): Query<String>, 477 613 ) -> ResolveMiniIDResponse { 614 614 + Self::resolve_mini_doc_impl(&identifier, self.identity.clone()).await 615 615 + } 616 616 + 617 617 + async fn resolve_mini_doc_impl(identifier: &str, identity: Identity) -> ResolveMiniIDResponse { 478 618 let invalid = |reason: &'static str| { 479 619 ResolveMiniIDResponse::BadRequest(xrpc_error("InvalidRequest", reason)) 480 620 }; 481 621 482 622 let mut unverified_handle = None; 483 483 - let did = match Did::new(identifier.clone()) { 623 623 + let did = match Did::new(identifier.to_string()) { 484 624 Ok(did) => did, 485 625 Err(_) => { 486 626 let Ok(alleged_handle) = Handle::new(identifier.to_lowercase()) else { 487 627 return invalid("Identifier was not a valid DID or handle"); 488 628 }; 489 629 490 490 - match self.identity.handle_to_did(alleged_handle.clone()).await { 630 630 + match identity.handle_to_did(alleged_handle.clone()).await { 491 631 Ok(res) => { 492 632 if let Some(did) = res { 493 633 // we did it joe ··· 505 645 } 506 646 } 507 647 }; 508 508 - let Ok(partial_doc) = self.identity.did_to_partial_mini_doc(&did).await else { 648 648 + let Ok(partial_doc) = identity.did_to_partial_mini_doc(&did).await else { 509 649 return invalid("Failed to get DID doc"); 510 650 }; 511 651 let Some(partial_doc) = partial_doc else { ··· 525 665 "handle.invalid".to_string() 526 666 } 527 667 } else { 528 528 - let Ok(handle_did) = self 529 529 - .identity 668 668 + let Ok(handle_did) = identity 530 669 .handle_to_did(partial_doc.unverified_handle.clone()) 531 670 .await 532 671 else { ··· 550 689 })) 551 690 } 552 691 692 692 + /// com.bad-example.proxy.hydrateQueryResponse 693 693 + /// 694 694 + /// > [!important] 695 695 + /// > Unstable! This endpoint is experimental and may change. 696 696 + /// 697 697 + /// Fetch + include records referenced from an upstream xrpc query response 698 698 + #[oai( 699 699 + path = "/com.bad-example.proxy.hydrateQueryResponse", 700 700 + method = "post", 701 701 + tag = "ApiTags::Custom" 702 702 + )] 703 703 + async fn proxy_hydrate_query( 704 704 + &self, 705 705 + Json(payload): Json<ProxyQueryPayload>, 706 706 + ) -> ProxyHydrateResponse { 707 707 + // TODO: the Accept request header, if present, gotta be json 708 708 + // TODO: find any Authorization header and verify it. TBD about `aud`. 709 709 + 710 710 + let params = if let Some(p) = payload.params { 711 711 + let serde_json::Value::Object(map) = p else { 712 712 + panic!("params have to be an object"); 713 713 + }; 714 714 + Some(map) 715 715 + } else { None }; 716 716 + 717 717 + match self.proxy.proxy( 718 718 + payload.xrpc, 719 719 + payload.atproto_proxy, 720 720 + params, 721 721 + ).await { 722 722 + Ok(skeleton) => { 723 723 + let links = match extract_links(payload.hydration_sources, &skeleton) { 724 724 + Ok(l) => l, 725 725 + Err(e) => { 726 726 + log::warn!("problem extracting: {e:?}"); 727 727 + return ProxyHydrateResponse::BadRequest(xrpc_error("oop", "sorry, error extracting")) 728 728 + } 729 729 + }; 730 730 + let mut records = HashMap::new(); 731 731 + let mut identifiers = HashMap::new(); 732 732 + let mut blobs = HashMap::new(); 733 733 + 734 734 + enum GetThing { 735 735 + Record(String, Hydration<ProxyHydrationRecordFound>), 736 736 + Identifier(String, Hydration<ProxyHydrationIdentifierFound>), 737 737 + Blob(String, Hydration<ProxyHydrationBlobFound>), 738 738 + } 739 739 + 740 740 + let (tx, mut rx) = mpsc::channel(1); 741 741 + 742 742 + for link in links { 743 743 + match link { 744 744 + MatchedRef::AtUri { uri, cid } => { 745 745 + if records.contains_key(&uri) { 746 746 + log::warn!("skipping duplicate record without checking cid"); 747 747 + continue; 748 748 + } 749 749 + let mut u = url::Url::parse("https://example.com").unwrap(); 750 750 + u.query_pairs_mut().append_pair("at_uri", &uri); // BLEH todo 751 751 + records.insert(uri.clone(), Hydration::Pending(ProxyHydrationPending { 752 752 + url: format!("/xrpc/blue.microcosm.repo.getRecordByUri?{}", u.query().unwrap()), // TODO better; with cid, etc. 753 753 + })); 754 754 + let tx = tx.clone(); 755 755 + let identity = self.identity.clone(); 756 756 + let repo = self.repo.clone(); 757 757 + tokio::task::spawn(async move { 758 758 + let rest = uri.strip_prefix("at://").unwrap(); 759 759 + let (identifier, rest) = rest.split_once('/').unwrap(); 760 760 + let (collection, rkey) = rest.split_once('/').unwrap(); 761 761 + 762 762 + let did = if identifier.starts_with("did:") { 763 763 + Did::new(identifier.to_string()).unwrap() 764 764 + } else { 765 765 + let handle = Handle::new(identifier.to_string()).unwrap(); 766 766 + identity.handle_to_did(handle).await.unwrap().unwrap() 767 767 + }; 768 768 + 769 769 + let res = match repo.get_record( 770 770 + &did, 771 771 + &Nsid::new(collection.to_string()).unwrap(), 772 772 + &RecordKey::new(rkey.to_string()).unwrap(), 773 773 + &cid.as_ref().map(|s| Cid::from_str(s).unwrap()), 774 774 + ).await { 775 775 + Ok(CachedRecord::Deleted) => 776 776 + Hydration::Error(ProxyHydrationError { 777 777 + reason: "record deleted".to_string(), 778 778 + }), 779 779 + Ok(CachedRecord::Found(RawRecord { cid: found_cid, record })) => { 780 780 + if let Some(c) = cid && found_cid.as_ref().to_string() != c { 781 781 + log::warn!("ignoring cid mismatch"); 782 782 + } 783 783 + let value = serde_json::from_str(&record).unwrap(); 784 784 + Hydration::Found(ProxyHydrationRecordFound { 785 785 + record: value, 786 786 + }) 787 787 + } 788 788 + Err(e) => { 789 789 + log::warn!("finally oop {e:?}"); 790 790 + Hydration::Error(ProxyHydrationError { 791 791 + reason: "failed to fetch record".to_string(), 792 792 + }) 793 793 + } 794 794 + }; 795 795 + tx.send(GetThing::Record(uri, res)).await 796 796 + }); 797 797 + } 798 798 + MatchedRef::Identifier(id) => { 799 799 + if identifiers.contains_key(&id) { 800 800 + continue; 801 801 + } 802 802 + let mut u = url::Url::parse("https://example.com").unwrap(); 803 803 + u.query_pairs_mut().append_pair("identifier", &id); 804 804 + identifiers.insert(id.clone(), Hydration::Pending(ProxyHydrationPending { 805 805 + url: format!("/xrpc/blue.microcosm.identity.resolveMiniDoc?{}", u.query().unwrap()), // gross 806 806 + })); 807 807 + let tx = tx.clone(); 808 808 + let identity = self.identity.clone(); 809 809 + tokio::task::spawn(async move { 810 810 + let res = match Self::resolve_mini_doc_impl(&id, identity).await { 811 811 + ResolveMiniIDResponse::Ok(Json(mini_doc)) => Hydration::Found(ProxyHydrationIdentifierFound { 812 812 + mini_doc 813 813 + }), 814 814 + ResolveMiniIDResponse::BadRequest(e) => { 815 815 + log::warn!("minidoc fail: {:?}", e.0); 816 816 + Hydration::Error(ProxyHydrationError { 817 817 + reason: "failed to resolve mini doc".to_string(), 818 818 + }) 819 819 + } 820 820 + }; 821 821 + tx.send(GetThing::Identifier(id, res)).await 822 822 + }); 823 823 + } 824 824 + MatchedRef::Blob { link, mime, size: _ } => { 825 825 + if blobs.contains_key(&link) { 826 826 + continue; 827 827 + } 828 828 + if mime != "image/jpeg" { 829 829 + Hydration::<ProxyHydrationBlobFound>::Error(ProxyHydrationError { 830 830 + reason: "only image/jpeg supported for now".to_string(), 831 831 + }); 832 832 + } 833 833 + todo!("oops we need to know the account too") 834 834 + } 835 835 + } 836 836 + } 837 837 + // so the channel can close when all are completed 838 838 + // (we shoudl be doing a timeout...) 839 839 + drop(tx); 840 840 + 841 841 + while let Some(hydration) = rx.recv().await { 842 842 + match hydration { 843 843 + GetThing::Record(uri, h) => { records.insert(uri, h); } 844 844 + GetThing::Identifier(uri, md) => { identifiers.insert(uri, md); } 845 845 + GetThing::Blob(cid, asdf) => { blobs.insert(cid, asdf); } 846 846 + }; 847 847 + } 848 848 + 849 849 + ProxyHydrateResponse::Ok(Json(ProxyHydrateResponseObject { 850 850 + output: skeleton, 851 851 + records, 852 852 + identifiers, 853 853 + blobs, 854 854 + })) 855 855 + } 856 856 + Err(e) => { 857 857 + log::warn!("oh no: {e:?}"); 858 858 + ProxyHydrateResponse::BadRequest(xrpc_error("oop", "sorry")) 859 859 + } 860 860 + } 861 861 + 862 862 + } 863 863 + 553 864 async fn get_record_impl( 554 865 &self, 555 866 repo: String, ··· 748 1059 cache: HybridCache<String, CachedRecord>, 749 1060 identity: Identity, 750 1061 repo: Repo, 1062 1062 + proxy: Proxy, 751 1063 acme_domain: Option<String>, 752 1064 acme_contact: Option<String>, 753 1065 acme_cache_path: Option<PathBuf>, ··· 756 1068 bind: std::net::SocketAddr, 757 1069 ) -> Result<(), ServerError> { 758 1070 let repo = Arc::new(repo); 1071 1071 + let proxy = Arc::new(proxy); 759 1072 let api_service = OpenApiService::new( 760 1073 Xrpc { 761 1074 cache, 762 1075 identity, 1076 1076 + proxy, 763 1077 repo, 764 1078 }, 765 1079 "Slingshot", ··· 823 1137 .with( 824 1138 Cors::new() 825 1139 .allow_origin_regex("*") 826 826 - .allow_methods([Method::GET]) 1140 1140 + .allow_methods([Method::GET, Method::POST]) 827 1141 .allow_credentials(false), 828 1142 ) 829 1143 .with(CatchPanic::new())