+2 -3

Cargo.lock

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

+7 -1

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

+32 -7

constellation/src/server/mod.rs

··· 3 3 extract::{Query, Request}, 4 4 http::{self, header}, 5 5 middleware::{self, Next}, 6 6 - response::{IntoResponse, Response}, 6 6 + response::{IntoResponse, Json, 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, 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() 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 46 59 .route("/robots.txt", get(robots)) 47 60 .route( 48 61 "/", ··· 204 217 User-agent: * 205 218 Disallow: /links 206 219 Disallow: /links/ 220 220 + Disallow: /xrpc/ 207 221 " 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 + })) 208 233 } 209 234 210 235 #[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 = { version = "2.5.4", features = ["serde"] } 31 31 + url = "2.5.4"

-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; 7 6 mod record; 8 7 mod server; 9 8 ··· 11 10 pub use firehose_cache::firehose_cache; 12 11 pub use healthcheck::healthcheck; 13 12 pub use identity::{Identity, IdentityKey}; 14 14 - pub use proxy::Proxy; 15 13 pub use record::{CachedRecord, ErrorResponseObject, Repo}; 16 14 pub use server::serve;

+3 -4

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, Proxy, Repo, consume, error::MainTaskError, firehose_cache, healthcheck, serve, 5 5 + Identity, 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."); 146 148 let identity_refresher = identity.clone(); 147 149 let identity_shutdown = shutdown.clone(); 148 150 tasks.spawn(async move { 149 151 identity_refresher.run_refresher(identity_shutdown).await?; 150 152 Ok(()) 151 153 }); 152 152 - log::info!("identity service ready."); 153 154 154 155 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, 167 166 args.acme_domain, 168 167 args.acme_contact, 169 168 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 - pub cid: Cid, 15 15 - pub record: String, 14 14 + cid: Cid, 15 15 + record: String, 16 16 } 17 17 18 18 // TODO: should be able to do typed CID

+12 -326

slingshot/src/server.rs

··· 1 1 use crate::{ 2 2 - CachedRecord, ErrorResponseObject, Identity, Proxy, Repo, 2 2 + CachedRecord, ErrorResponseObject, Identity, Repo, 3 3 error::{RecordError, ServerError}, 4 4 - proxy::{extract_links, MatchedRef}, 5 5 - record::RawRecord, 6 4 }; 7 5 use atrium_api::types::string::{Cid, Did, Handle, Nsid, RecordKey}; 8 6 use foyer::HybridCache; 9 7 use links::at_uri::parse_at_uri as normalize_at_uri; 10 8 use serde::Serialize; 11 11 - use std::{path::PathBuf, str::FromStr, sync::Arc, time::Instant, collections::HashMap}; 12 12 - use tokio::sync::mpsc; 9 9 + use std::path::PathBuf; 10 10 + use std::str::FromStr; 11 11 + use std::sync::Arc; 12 12 + use std::time::Instant; 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, 28 27 param::Query, payload::Json, types::Example, 29 28 }; 30 29 ··· 55 54 "zQ3shpq1g134o7HGDb86CtQFxnHqzx5pZWknrVX2Waum3fF6j".to_string() 56 55 } 57 56 58 58 - #[derive(Debug, Object)] 57 57 + #[derive(Object)] 59 58 #[oai(example = true)] 60 59 struct XrpcErrorResponseObject { 61 60 /// Should correspond an error `name` in the lexicon errors array ··· 88 87 89 88 fn bad_request_handler_resolve_mini(err: poem::Error) -> ResolveMiniIDResponse { 90 89 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 { 98 90 error: "InvalidRequest".to_string(), 99 91 message: format!("Bad request, here's some info that maybe should not be exposed: {err}"), 100 92 })) ··· 199 191 } 200 192 201 193 #[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)] 329 194 #[oai(example = true)] 330 195 struct FoundDidResponseObject { 331 196 /// the DID, bi-directionally verified if using Slingshot ··· 356 221 struct Xrpc { 357 222 cache: HybridCache<String, CachedRecord>, 358 223 identity: Identity, 359 359 - proxy: Arc<Proxy>, 360 224 repo: Arc<Repo>, 361 225 } 362 226 ··· 611 475 #[oai(example = "example_handle")] 612 476 Query(identifier): Query<String>, 613 477 ) -> 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 { 618 478 let invalid = |reason: &'static str| { 619 479 ResolveMiniIDResponse::BadRequest(xrpc_error("InvalidRequest", reason)) 620 480 }; 621 481 622 482 let mut unverified_handle = None; 623 623 - let did = match Did::new(identifier.to_string()) { 483 483 + let did = match Did::new(identifier.clone()) { 624 484 Ok(did) => did, 625 485 Err(_) => { 626 486 let Ok(alleged_handle) = Handle::new(identifier.to_lowercase()) else { 627 487 return invalid("Identifier was not a valid DID or handle"); 628 488 }; 629 489 630 630 - match identity.handle_to_did(alleged_handle.clone()).await { 490 490 + match self.identity.handle_to_did(alleged_handle.clone()).await { 631 491 Ok(res) => { 632 492 if let Some(did) = res { 633 493 // we did it joe ··· 645 505 } 646 506 } 647 507 }; 648 648 - let Ok(partial_doc) = identity.did_to_partial_mini_doc(&did).await else { 508 508 + let Ok(partial_doc) = self.identity.did_to_partial_mini_doc(&did).await else { 649 509 return invalid("Failed to get DID doc"); 650 510 }; 651 511 let Some(partial_doc) = partial_doc else { ··· 665 525 "handle.invalid".to_string() 666 526 } 667 527 } else { 668 668 - let Ok(handle_did) = identity 528 528 + let Ok(handle_did) = self 529 529 + .identity 669 530 .handle_to_did(partial_doc.unverified_handle.clone()) 670 531 .await 671 532 else { ··· 689 550 })) 690 551 } 691 552 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 - 864 553 async fn get_record_impl( 865 554 &self, 866 555 repo: String, ··· 1059 748 cache: HybridCache<String, CachedRecord>, 1060 749 identity: Identity, 1061 750 repo: Repo, 1062 1062 - proxy: Proxy, 1063 751 acme_domain: Option<String>, 1064 752 acme_contact: Option<String>, 1065 753 acme_cache_path: Option<PathBuf>, ··· 1068 756 bind: std::net::SocketAddr, 1069 757 ) -> Result<(), ServerError> { 1070 758 let repo = Arc::new(repo); 1071 1071 - let proxy = Arc::new(proxy); 1072 759 let api_service = OpenApiService::new( 1073 760 Xrpc { 1074 761 cache, 1075 762 identity, 1076 1076 - proxy, 1077 763 repo, 1078 764 }, 1079 765 "Slingshot", ··· 1137 823 .with( 1138 824 Cors::new() 1139 825 .allow_origin_regex("*") 1140 1140 - .allow_methods([Method::GET, Method::POST]) 826 826 + .allow_methods([Method::GET]) 1141 827 .allow_credentials(false), 1142 828 ) 1143 829 .with(CatchPanic::new())