+1

Cargo.lock

··· 5101 5101 name = "slingshot" 5102 5102 version = "0.1.0" 5103 5103 dependencies = [ 5104 5104 + "async-stream", 5104 5105 "atrium-api 0.25.4 (git+https://github.com/uniphil/atrium.git?branch=fix%2Fresolve-handle-https-accept-whitespace)", 5105 5106 "atrium-common 0.1.2 (git+https://github.com/uniphil/atrium.git?branch=fix%2Fresolve-handle-https-accept-whitespace)", 5106 5107 "atrium-identity 0.1.5 (git+https://github.com/uniphil/atrium.git?branch=fix%2Fresolve-handle-https-accept-whitespace)",

+1

slingshot/Cargo.toml

··· 4 4 edition = "2024" 5 5 6 6 [dependencies] 7 7 + async-stream = "0.3.6" 7 8 atrium-api = { git = "https://github.com/uniphil/atrium.git", branch = "fix/resolve-handle-https-accept-whitespace", default-features = false } 8 9 atrium-common = { git = "https://github.com/uniphil/atrium.git", branch = "fix/resolve-handle-https-accept-whitespace" } 9 10 atrium-identity = { git = "https://github.com/uniphil/atrium.git", branch = "fix/resolve-handle-https-accept-whitespace" }

+8

slingshot/src/error.rs

··· 100 100 UrlParseError(#[from] url::ParseError), 101 101 #[error(transparent)] 102 102 ReqwestError(#[from] reqwest::Error), 103 103 + #[error(transparent)] 104 104 + InvalidHeader(#[from] reqwest::header::InvalidHeaderValue), 105 105 + #[error(transparent)] 106 106 + IdentityError(#[from] IdentityError), 107 107 + #[error("upstream service could not be resolved")] 108 108 + ServiceNotFound, 109 109 + #[error("upstream service was found but no services matched")] 110 110 + ServiceNotMatched, 103 111 }

+208 -16

slingshot/src/identity.rs

··· 17 17 18 18 use crate::error::IdentityError; 19 19 use atrium_api::{ 20 20 - did_doc::DidDocument, 20 20 + did_doc::{DidDocument, Service as DidDocServic}, 21 21 types::string::{Did, Handle}, 22 22 }; 23 23 use atrium_common::resolver::Resolver; ··· 41 41 pub enum IdentityKey { 42 42 Handle(Handle), 43 43 Did(Did), 44 44 + ServiceDid(Did), 44 45 } 45 46 46 47 impl IdentityKey { ··· 48 49 let s = match self { 49 50 IdentityKey::Handle(h) => h.as_str(), 50 51 IdentityKey::Did(d) => d.as_str(), 52 52 + IdentityKey::ServiceDid(d) => d.as_str(), 51 53 }; 52 54 std::mem::size_of::<Self>() + std::mem::size_of_val(s) 53 55 } ··· 59 61 #[derive(Debug, Serialize, Deserialize)] 60 62 enum IdentityData { 61 63 NotFound, 62 62 - Did(Did), 63 63 - Doc(PartialMiniDoc), 64 64 + Did(Did), // from handle 65 65 + Doc(PartialMiniDoc), // from did 66 66 + ServiceDoc(MiniServiceDoc), // from service did 64 67 } 65 68 66 69 impl IdentityVal { ··· 71 74 IdentityData::Did(d) => std::mem::size_of_val(d.as_str()), 72 75 IdentityData::Doc(d) => { 73 76 std::mem::size_of_val(d.unverified_handle.as_str()) 74 74 - + std::mem::size_of_val(d.pds.as_str()) 75 75 - + std::mem::size_of_val(d.signing_key.as_str()) 77 77 + + std::mem::size_of_val(&d.pds) 78 78 + + std::mem::size_of_val(&d.signing_key) 79 79 + } 80 80 + IdentityData::ServiceDoc(d) => { 81 81 + let mut s = std::mem::size_of::<MiniServiceDoc>(); 82 82 + s += std::mem::size_of_val(&d.services); 83 83 + for sv in &d.services { 84 84 + s += std::mem::size_of_val(&sv.full_id); 85 85 + s += std::mem::size_of_val(&sv.r#type); 86 86 + s += std::mem::size_of_val(&sv.endpoint); 87 87 + } 88 88 + s 76 89 } 77 90 }; 78 91 wrapping + inner ··· 168 181 } 169 182 } 170 183 184 184 + /// Simplified info from service DID docs 185 185 + #[derive(Debug, Clone, Serialize, Deserialize)] 186 186 + pub struct MiniServiceDoc { 187 187 + services: Vec<MiniService>, 188 188 + } 189 189 + 190 190 + impl MiniServiceDoc { 191 191 + pub fn get(&self, id_fragment: &str, service_type: Option<&str>) -> Option<&MiniService> { 192 192 + self.services.iter().find(|ms| { 193 193 + ms.full_id.ends_with(id_fragment) 194 194 + && service_type.map(|t| t == ms.r#type).unwrap_or(true) 195 195 + }) 196 196 + } 197 197 + } 198 198 + 199 199 + /// The corresponding service info 200 200 + #[derive(Debug, Clone, Serialize, Deserialize)] 201 201 + pub struct MiniService { 202 202 + /// The full id 203 203 + /// 204 204 + /// for informational purposes only -- services are deduplicated by id fragment 205 205 + full_id: String, 206 206 + r#type: String, 207 207 + /// HTTP endpoint for the actual service 208 208 + pub endpoint: String, 209 209 + } 210 210 + 211 211 + impl TryFrom<DidDocument> for MiniServiceDoc { 212 212 + type Error = String; 213 213 + fn try_from(did_doc: DidDocument) -> Result<Self, Self::Error> { 214 214 + let mut services = Vec::new(); 215 215 + let mut seen = HashSet::new(); 216 216 + 217 217 + for DidDocServic { 218 218 + id, 219 219 + r#type, 220 220 + service_endpoint, 221 221 + } in did_doc.service.unwrap_or(vec![]) 222 222 + { 223 223 + let Some((_, id_fragment)) = id.rsplit_once('#') else { 224 224 + continue; 225 225 + }; 226 226 + if !seen.insert((id_fragment.to_string(), r#type.clone())) { 227 227 + continue; 228 228 + } 229 229 + services.push(MiniService { 230 230 + full_id: id, 231 231 + r#type, 232 232 + endpoint: service_endpoint, 233 233 + }); 234 234 + } 235 235 + 236 236 + Ok(Self { services }) 237 237 + } 238 238 + } 239 239 + 171 240 /// multi-producer *single-consumer* queue structures (wrap in arc-mutex plz) 172 241 /// 173 242 /// the hashset allows testing for presense of items in the queue. ··· 296 365 let now = UtcDateTime::now(); 297 366 let IdentityVal(last_fetch, data) = entry.value(); 298 367 match data { 299 299 - IdentityData::Doc(_) => { 300 300 - log::error!("identity value mixup: got a doc from a handle key (should be a did)"); 301 301 - Err(IdentityError::IdentityValTypeMixup(handle.to_string())) 302 302 - } 303 368 IdentityData::NotFound => { 304 369 if (now - *last_fetch) >= MIN_NOT_FOUND_TTL { 305 370 metrics::counter!("identity_handle_refresh_queued", "reason" => "ttl", "found" => "false").increment(1); ··· 313 378 self.queue_refresh(key).await; 314 379 } 315 380 Ok(Some(did.clone())) 381 381 + } 382 382 + _ => { 383 383 + log::error!("identity value mixup: got a doc from a handle key (should be a did)"); 384 384 + Err(IdentityError::IdentityValTypeMixup(handle.to_string())) 316 385 } 317 386 } 318 387 } ··· 362 431 let now = UtcDateTime::now(); 363 432 let IdentityVal(last_fetch, data) = entry.value(); 364 433 match data { 365 365 - IdentityData::Did(_) => { 366 366 - log::error!("identity value mixup: got a did from a did key (should be a doc)"); 367 367 - Err(IdentityError::IdentityValTypeMixup(did.to_string())) 368 368 - } 369 434 IdentityData::NotFound => { 370 435 if (now - *last_fetch) >= MIN_NOT_FOUND_TTL { 371 436 metrics::counter!("identity_did_refresh_queued", "reason" => "ttl", "found" => "false").increment(1); ··· 373 438 } 374 439 Ok(None) 375 440 } 376 376 - IdentityData::Doc(mini_did) => { 441 441 + IdentityData::Doc(mini_doc) => { 377 442 if (now - *last_fetch) >= MIN_TTL { 378 443 metrics::counter!("identity_did_refresh_queued", "reason" => "ttl", "found" => "true").increment(1); 379 444 self.queue_refresh(key).await; 380 445 } 381 381 - Ok(Some(mini_did.clone())) 446 446 + Ok(Some(mini_doc.clone())) 447 447 + } 448 448 + _ => { 449 449 + log::error!("identity value mixup: got a doc from a handle key (should be a did)"); 450 450 + Err(IdentityError::IdentityValTypeMixup(did.to_string())) 451 451 + } 452 452 + } 453 453 + } 454 454 + 455 455 + /// Fetch (and cache) a service mini doc from a did 456 456 + pub async fn did_to_mini_service_doc( 457 457 + &self, 458 458 + did: &Did, 459 459 + ) -> Result<Option<MiniServiceDoc>, IdentityError> { 460 460 + let key = IdentityKey::ServiceDid(did.clone()); 461 461 + metrics::counter!("slingshot_get_service_did_doc").increment(1); 462 462 + let entry = self 463 463 + .cache 464 464 + .get_or_fetch(&key, { 465 465 + let did = did.clone(); 466 466 + let resolver = self.did_resolver.clone(); 467 467 + || async move { 468 468 + let t0 = Instant::now(); 469 469 + let (res, success) = match resolver.resolve(&did).await { 470 470 + Ok(did_doc) if did_doc.id != did.to_string() => ( 471 471 + // TODO: fix in atrium: should verify id is did 472 472 + Err(IdentityError::BadDidDoc( 473 473 + "did doc's id did not match did".to_string(), 474 474 + )), 475 475 + "false", 476 476 + ), 477 477 + Ok(did_doc) => match did_doc.try_into() { 478 478 + Ok(mini_service_doc) => ( 479 479 + Ok(IdentityVal( 480 480 + UtcDateTime::now(), 481 481 + IdentityData::ServiceDoc(mini_service_doc), 482 482 + )), 483 483 + "true", 484 484 + ), 485 485 + Err(e) => (Err(IdentityError::BadDidDoc(e)), "false"), 486 486 + }, 487 487 + Err(atrium_identity::Error::NotFound) => ( 488 488 + Ok(IdentityVal(UtcDateTime::now(), IdentityData::NotFound)), 489 489 + "false", 490 490 + ), 491 491 + Err(other) => (Err(IdentityError::ResolutionFailed(other)), "false"), 492 492 + }; 493 493 + metrics::histogram!("slingshot_fetch_service_did_doc", "success" => success) 494 494 + .record(t0.elapsed()); 495 495 + res 496 496 + } 497 497 + }) 498 498 + .await?; 499 499 + 500 500 + let now = UtcDateTime::now(); 501 501 + let IdentityVal(last_fetch, data) = entry.value(); 502 502 + match data { 503 503 + IdentityData::NotFound => { 504 504 + if (now - *last_fetch) >= MIN_NOT_FOUND_TTL { 505 505 + metrics::counter!("identity_service_did_refresh_queued", "reason" => "ttl", "found" => "false").increment(1); 506 506 + self.queue_refresh(key).await; 507 507 + } 508 508 + Ok(None) 509 509 + } 510 510 + IdentityData::ServiceDoc(mini_service_doc) => { 511 511 + if (now - *last_fetch) >= MIN_TTL { 512 512 + metrics::counter!("identity_service_did_refresh_queued", "reason" => "ttl", "found" => "true").increment(1); 513 513 + self.queue_refresh(key).await; 514 514 + } 515 515 + Ok(Some(mini_service_doc.clone())) 516 516 + } 517 517 + _ => { 518 518 + log::error!( 519 519 + "identity value mixup: got a doc from a different key type (should be a service did)" 520 520 + ); 521 521 + Err(IdentityError::IdentityValTypeMixup(did.to_string())) 382 522 } 383 523 } 384 524 } ··· 519 659 log::warn!( 520 660 "refreshed did doc failed: wrong did doc id. dropping refresh." 521 661 ); 662 662 + self.complete_refresh(&task_key).await?; 522 663 continue; 523 664 } 524 665 let mini_doc = match did_doc.try_into() { ··· 526 667 Err(e) => { 527 668 metrics::counter!("identity_did_refresh", "success" => "false", "reason" => "bad doc").increment(1); 528 669 log::warn!( 529 529 - "converting mini doc failed: {e:?}. dropping refresh." 670 670 + "converting mini doc for {did:?} failed: {e:?}. dropping refresh." 530 671 ); 672 672 + self.complete_refresh(&task_key).await?; 531 673 continue; 532 674 } 533 675 }; ··· 554 696 } 555 697 556 698 self.complete_refresh(&task_key).await?; // failures are bugs, so break loop 699 699 + } 700 700 + IdentityKey::ServiceDid(ref did) => { 701 701 + log::trace!("refreshing service did doc: {did:?}"); 702 702 + 703 703 + match self.did_resolver.resolve(did).await { 704 704 + Ok(did_doc) => { 705 705 + // TODO: fix in atrium: should verify id is did 706 706 + if did_doc.id != did.to_string() { 707 707 + metrics::counter!("identity_service_did_refresh", "success" => "false", "reason" => "wrong did").increment(1); 708 708 + log::warn!( 709 709 + "refreshed did doc failed: wrong did doc id. dropping refresh." 710 710 + ); 711 711 + self.complete_refresh(&task_key).await?; 712 712 + continue; 713 713 + } 714 714 + let mini_service_doc = match did_doc.try_into() { 715 715 + Ok(md) => md, 716 716 + Err(e) => { 717 717 + metrics::counter!("identity_service_did_refresh", "success" => "false", "reason" => "bad doc").increment(1); 718 718 + log::warn!( 719 719 + "converting mini service doc failed: {e:?}. dropping refresh." 720 720 + ); 721 721 + self.complete_refresh(&task_key).await?; 722 722 + continue; 723 723 + } 724 724 + }; 725 725 + metrics::counter!("identity_service_did_refresh", "success" => "true") 726 726 + .increment(1); 727 727 + self.cache.insert( 728 728 + task_key.clone(), 729 729 + IdentityVal( 730 730 + UtcDateTime::now(), 731 731 + IdentityData::ServiceDoc(mini_service_doc), 732 732 + ), 733 733 + ); 734 734 + } 735 735 + Err(atrium_identity::Error::NotFound) => { 736 736 + metrics::counter!("identity_service_did_refresh", "success" => "false", "reason" => "not found").increment(1); 737 737 + self.cache.insert( 738 738 + task_key.clone(), 739 739 + IdentityVal(UtcDateTime::now(), IdentityData::NotFound), 740 740 + ); 741 741 + } 742 742 + Err(err) => { 743 743 + metrics::counter!("identity_service_did_refresh", "success" => "false", "reason" => "other").increment(1); 744 744 + log::warn!( 745 745 + "failed to refresh did doc: {err:?}. leaving stale (should we eventually do something?)" 746 746 + ); 747 747 + } 748 748 + } 557 749 } 558 750 } 559 751 }

+61 -25

slingshot/src/main.rs

··· 1 1 - // use foyer::HybridCache; 2 2 - // use foyer::{Engine, DirectFsDeviceOptions, HybridCacheBuilder}; 3 1 use metrics_exporter_prometheus::PrometheusBuilder; 4 2 use slingshot::{ 5 3 Identity, Proxy, Repo, consume, error::MainTaskError, firehose_cache, healthcheck, serve, ··· 9 7 10 8 use clap::Parser; 11 9 use tokio_util::sync::CancellationToken; 10 10 + use url::Url; 12 11 13 12 /// Slingshot record edge cache 14 13 #[derive(Parser, Debug, Clone)] ··· 48 47 #[arg(long, env = "SLINGSHOT_IDENTITY_CACHE_DISK_DB")] 49 48 #[clap(default_value_t = 1)] 50 49 identity_cache_disk_gb: usize, 50 50 + /// the address of this server 51 51 + /// 52 52 + /// used if --acme-domain is not set, defaulting to `--bind` 53 53 + #[arg(long, conflicts_with("tls_domain"), env = "SLINGSHOT_PUBLIC_HOST")] 54 54 + base_url: Option<Url>, 51 55 /// the domain pointing to this server 52 56 /// 53 57 /// if present: 54 58 /// - a did:web document will be served at /.well-known/did.json 55 55 - /// - an HTTPS certs will be automatically configured with Acme/letsencrypt 59 59 + /// - the server will bind on port 443 60 60 + /// - if `--acme-contact` is present, the server will bind port 80 for http 61 61 + /// challenges and attempt to auto-provision certs for `--tls-domain` 62 62 + /// - if `--acme-contact is absent, the server will load certs from the 63 63 + /// `--tls-certs` folder, and try to reload them twice daily, guarded by 64 64 + /// a lock file called `.cert-lock` in the `--tls-certs` folder. 56 65 /// - TODO: a rate-limiter will be installed 57 66 #[arg( 58 67 long, 59 68 conflicts_with("bind"), 60 60 - requires("acme_cache_path"), 61 61 - env = "SLINGSHOT_ACME_DOMAIN" 69 69 + requires("tls_certs"), 70 70 + env = "SLINGSHOT_TLS_DOMAIN" 71 71 + )] 72 72 + tls_domain: Option<String>, 73 73 + /// a location to find/cache acme or other tls certs 74 74 + /// 75 75 + /// recommended in production, mind the file permissions. 76 76 + #[arg(long, env = "SLINGSHOT_TLS_CERTS_PATH")] 77 77 + tls_certs: Option<PathBuf>, 78 78 + /// listen for ipv6 when using acme or other tls 79 79 + /// 80 80 + /// you must also configure the relevant DNS records for this to work 81 81 + #[arg(long, action, requires("tls_domain"), env = "SLINGSHOT_TLS_IPV6")] 82 82 + tls_ipv6: bool, 83 83 + /// redirect acme http-01 challenges to this url 84 84 + /// 85 85 + /// useful if you're setting up a second instance that synchronizes its 86 86 + /// certs from a main instance doing acme. 87 87 + #[arg( 88 88 + long, 89 89 + conflicts_with("acme_contact"), 90 90 + requires("tls_domain"), 91 91 + env = "SLINGSHOT_ACME_CHALLENGE_REDIRECT" 62 92 )] 63 63 - acme_domain: Option<String>, 93 93 + acme_challenge_redirect: Option<String>, 64 94 /// email address for letsencrypt contact 65 95 /// 66 96 /// recommended in production, i guess? 67 67 - #[arg(long, requires("acme_domain"), env = "SLINGSHOT_ACME_CONTACT")] 97 97 + #[arg(long, requires("tls_domain"), env = "SLINGSHOT_ACME_CONTACT")] 68 98 acme_contact: Option<String>, 69 69 - /// a location to cache acme https certs 99 99 + /// use the staging environment for letsencrypt 70 100 /// 71 71 - /// required when (and only used when) --acme-domain is specified. 72 72 - /// 73 73 - /// recommended in production, but mind the file permissions. 74 74 - #[arg(long, requires("acme_domain"), env = "SLINGSHOT_ACME_CACHE_PATH")] 75 75 - acme_cache_path: Option<PathBuf>, 76 76 - /// listen for ipv6 when using acme 77 77 - /// 78 78 - /// you must also configure the relevant DNS records for this to work 79 79 - #[arg(long, action, requires("acme_domain"), env = "SLINGSHOT_ACME_IPV6")] 80 80 - acme_ipv6: bool, 101 101 + /// recommended to initially test out new deployments with this to avoid 102 102 + /// letsencrypt rate limit problems. 103 103 + #[arg(long, action, requires("acme_contact"), env = "SLINGSHOT_ACME_STAGING")] 104 104 + acme_staging: bool, 81 105 /// an web address to send healtcheck pings to every ~51s or so 82 106 #[arg(long, env = "SLINGSHOT_HEALTHCHECK")] 83 107 healthcheck: Option<String>, ··· 101 125 102 126 let args = Args::parse(); 103 127 128 128 + let base_url: Url = args 129 129 + .base_url 130 130 + .or_else(|| { 131 131 + args.tls_domain 132 132 + .as_ref() 133 133 + .map(|d| Url::parse(&format!("https://{d}")).unwrap()) 134 134 + }) 135 135 + .unwrap_or_else(|| Url::parse(&format!("http://{}", args.bind)).unwrap()); 136 136 + 104 137 if args.collect_metrics { 105 138 log::trace!("installing metrics server..."); 106 139 if let Err(e) = install_metrics_server(args.bind_metrics) { ··· 152 185 log::info!("identity service ready."); 153 186 154 187 let repo = Repo::new(identity.clone()); 155 155 - let proxy = Proxy::new(repo.clone()); 188 188 + let proxy = Proxy::new(identity.clone()); 156 189 157 190 let identity_for_server = identity.clone(); 158 191 let server_shutdown = shutdown.clone(); ··· 164 197 identity_for_server, 165 198 repo, 166 199 proxy, 167 167 - args.acme_domain, 200 200 + base_url, 201 201 + args.tls_domain, 202 202 + args.tls_certs, 203 203 + args.tls_ipv6, 204 204 + args.acme_challenge_redirect, 168 205 args.acme_contact, 169 169 - args.acme_cache_path, 170 170 - args.acme_ipv6, 206 206 + args.acme_staging, 171 207 server_shutdown, 172 208 bind, 173 209 ) ··· 236 272 ) -> Result<(), metrics_exporter_prometheus::BuildError> { 237 273 log::info!("installing metrics server..."); 238 274 PrometheusBuilder::new() 239 239 - .set_quantiles(&[0.5, 0.9, 0.99, 1.0])? 240 240 - .set_bucket_duration(std::time::Duration::from_secs(300))? 241 241 - .set_bucket_count(std::num::NonZero::new(12).unwrap()) // count * duration = 60 mins. stuff doesn't happen that fast here. 275 275 + .set_buckets(&[0.001, 0.006, 0.036, 0.216, 1.296, 7.776, 45.656])? 276 276 + .set_bucket_duration(std::time::Duration::from_secs(15))? 277 277 + .set_bucket_count(std::num::NonZero::new(4).unwrap()) // count * duration = bucket lifetime 242 278 .set_enable_unit_suffix(false) // this seemed buggy for constellation (sometimes wouldn't engage) 243 279 .with_http_listener(bind_metrics) 244 280 .install()?;

+251 -155

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}; 1 1 + use crate::{Identity, error::ProxyError, server::HydrationSource}; 2 2 + use atrium_api::types::string::{AtIdentifier, Cid, Did, Nsid, RecordKey}; 5 3 use reqwest::Client; 6 4 use serde_json::{Map, Value}; 5 5 + use std::{collections::HashMap, time::Duration}; 6 6 + use url::Url; 7 7 8 8 pub enum ParamValue { 9 9 String(Vec<String>), ··· 13 13 pub struct Params(HashMap<String, ParamValue>); 14 14 15 15 impl TryFrom<Map<String, Value>> for Params { 16 16 - type Error = (); // TODO 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 { ··· 70 70 71 71 #[derive(Clone)] 72 72 pub struct Proxy { 73 73 - repo: Repo, 73 73 + identity: Identity, 74 74 client: Client, 75 75 } 76 76 77 77 impl Proxy { 78 78 - pub fn new(repo: Repo) -> Self { 78 78 + pub fn new(identity: Identity) -> Self { 79 79 let client = Client::builder() 80 80 .user_agent(format!( 81 81 "microcosm slingshot v{} (contact: @bad-example.com)", ··· 85 85 .timeout(Duration::from_secs(6)) 86 86 .build() 87 87 .unwrap(); 88 88 - Self { repo, client } 88 88 + Self { client, identity } 89 89 } 90 90 91 91 pub async fn proxy( 92 92 &self, 93 93 - xrpc: String, 94 94 - service: String, 93 93 + service_did: &Did, 94 94 + service_id: &str, 95 95 + xrpc: &Nsid, 96 96 + authorization: Option<&str>, 97 97 + atproto_accept_labelers: Option<&str>, 95 98 params: Option<Map<String, Value>>, 96 99 ) -> 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() 100 100 + let mut upstream: Url = self 101 101 + .identity 102 102 + .did_to_mini_service_doc(service_did) 121 103 .await? 122 122 - .error_for_status()? 123 123 - .json() 124 124 - .await?; 104 104 + .ok_or(ProxyError::ServiceNotFound)? 105 105 + .get(service_id, None) 106 106 + .ok_or(ProxyError::ServiceNotMatched)? 107 107 + .endpoint 108 108 + .parse()?; 125 109 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 110 110 + upstream.set_path(&format!("/xrpc/{}", xrpc.as_str())); 139 111 140 112 if let Some(params) = params { 141 113 let mut query = upstream.query_pairs_mut(); ··· 161 133 } 162 134 } 163 135 164 164 - // TODO: other headers to proxy 165 165 - Ok(self.client 136 136 + // TODO i mean maybe we should look for headers also in our headers but not obviously 137 137 + let mut headers = reqwest::header::HeaderMap::new(); 138 138 + // TODO: check the jwt aud against the upstream!!! 139 139 + if let Some(auth) = authorization { 140 140 + headers.insert("Authorization", auth.try_into()?); 141 141 + } 142 142 + if let Some(aal) = atproto_accept_labelers { 143 143 + headers.insert("atproto-accept-labelers", aal.try_into()?); 144 144 + } 145 145 + 146 146 + let t0 = std::time::Instant::now(); 147 147 + let res = self 148 148 + .client 166 149 .get(upstream) 150 150 + .headers(headers) 167 151 .send() 168 168 - .await? 169 169 - .error_for_status()? 170 170 - .json() 171 171 - .await?) 152 152 + .await 153 153 + .and_then(|r| r.error_for_status()); 154 154 + 155 155 + if res.is_ok() { 156 156 + metrics::histogram!("slingshot_proxy_upstream_request", "success" => "true") 157 157 + .record(t0.elapsed()); 158 158 + } else { 159 159 + metrics::histogram!("slingshot_proxy_upstream_request", "success" => "false") 160 160 + .record(t0.elapsed()); 161 161 + } 162 162 + 163 163 + Ok(res?.json().await?) 172 164 } 173 165 } 174 166 ··· 188 180 while let Some((i, c)) = chars.next() { 189 181 match c { 190 182 '[' 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}")), 183 183 + '[' if key_acc.is_empty() => { 184 184 + return Err(format!("missing key before opening bracket, at {i}")); 185 185 + } 192 186 '[' => in_bracket = true, 193 187 ']' if in_bracket => { 194 188 in_bracket = false; 195 189 let key = std::mem::take(&mut key_acc); 196 190 let r#type = std::mem::take(&mut type_acc); 197 197 - let t = if r#type.is_empty() { None } else { Some(r#type) }; 191 191 + let t = if r#type.is_empty() { 192 192 + None 193 193 + } else { 194 194 + Some(r#type) 195 195 + }; 198 196 out.push(PathPart::Vector(key, t)); 199 197 // peek ahead because we need a dot after array if there's more and i don't want to add more loop state 200 198 let Some((i, c)) = chars.next() else { 201 199 break; 202 200 }; 203 201 if c != '.' { 204 204 - return Err(format!("expected dot after close bracket, found {c:?} at {i}")); 202 202 + return Err(format!( 203 203 + "expected dot after close bracket, found {c:?} at {i}" 204 204 + )); 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}")), 209 209 + '.' if key_acc.is_empty() => { 210 210 + return Err(format!("missing key before next segment, at {i}")); 211 211 + } 210 212 '.' => { 211 213 let key = std::mem::take(&mut key_acc); 212 214 assert!(type_acc.is_empty()); ··· 225 227 Ok(out) 226 228 } 227 229 228 228 - #[derive(Debug, Clone, PartialEq)] 230 230 + #[derive(Debug, Clone, Copy, PartialEq)] 229 231 pub enum RefShape { 230 232 StrongRef, 231 233 AtUri, ··· 233 235 Did, 234 236 Handle, 235 237 AtIdentifier, 236 236 - Blob, 237 237 - // TODO: blob with type? 238 238 } 239 239 240 240 impl TryFrom<&str> for RefShape { ··· 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 250 _ => Err(format!("unknown shape: {s}")), 252 251 } 253 252 } 254 253 } 255 254 256 255 #[derive(Debug, PartialEq)] 256 256 + pub struct FullAtUriParts { 257 257 + pub repo: AtIdentifier, 258 258 + pub collection: Nsid, 259 259 + pub rkey: RecordKey, 260 260 + pub cid: Option<Cid>, 261 261 + } 262 262 + 263 263 + impl FullAtUriParts { 264 264 + pub fn to_uri(&self) -> String { 265 265 + let repo: String = self.repo.clone().into(); // no as_str for AtIdentifier atrium??? 266 266 + let collection = self.collection.as_str(); 267 267 + let rkey = self.rkey.as_str(); 268 268 + format!("at://{repo}/{collection}/{rkey}") 269 269 + } 270 270 + } 271 271 + 272 272 + // TODO: move this to links 273 273 + pub fn split_uri(uri: &str) -> Option<(AtIdentifier, Nsid, RecordKey)> { 274 274 + let rest = uri.strip_prefix("at://")?; 275 275 + let (repo, rest) = rest.split_once("/")?; 276 276 + let repo = repo.parse().ok()?; 277 277 + let (collection, rkey) = rest.split_once("/")?; 278 278 + let collection = collection.parse().ok()?; 279 279 + let rkey = rkey.split_once('#').map(|(k, _)| k).unwrap_or(rkey); 280 280 + let rkey = rkey.split_once('?').map(|(k, _)| k).unwrap_or(rkey); 281 281 + let rkey = rkey.parse().ok()?; 282 282 + Some((repo, collection, rkey)) 283 283 + } 284 284 + 285 285 + #[derive(Debug, PartialEq)] 257 286 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 - } 287 287 + AtUri(FullAtUriParts), 288 288 + Identifier(AtIdentifier), 268 289 } 269 290 270 270 - pub fn match_shape(shape: &RefShape, val: &Value) -> Option<MatchedRef> { 291 291 + pub fn match_shape(shape: RefShape, val: &Value) -> Option<MatchedRef> { 271 292 // TODO: actually validate at-uri format 272 293 // TODO: actually validate everything else also 273 294 // TODO: should this function normalize identifiers to DIDs probably? ··· 276 297 RefShape::StrongRef => { 277 298 let o = val.as_object()?; 278 299 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) }) 300 300 + let cid = o.get("cid")?.as_str()?.parse().ok()?; 301 301 + let (repo, collection, rkey) = split_uri(&uri)?; 302 302 + Some(MatchedRef::AtUri(FullAtUriParts { 303 303 + repo, 304 304 + collection, 305 305 + rkey, 306 306 + cid: Some(cid), 307 307 + })) 281 308 } 282 309 RefShape::AtUri => { 283 310 let uri = val.as_str()?.to_string(); 284 284 - Some(MatchedRef::AtUri { uri, cid: None }) 311 311 + let (repo, collection, rkey) = split_uri(&uri)?; 312 312 + Some(MatchedRef::AtUri(FullAtUriParts { 313 313 + repo, 314 314 + collection, 315 315 + rkey, 316 316 + cid: None, 317 317 + })) 285 318 } 286 319 RefShape::AtUriParts => { 287 320 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 }) 321 321 + let repo = o.get("repo").or(o.get("did"))?.as_str()?.parse().ok()?; 322 322 + let collection = o.get("collection")?.as_str()?.parse().ok()?; 323 323 + let rkey = o.get("rkey")?.as_str()?.parse().ok()?; 324 324 + let cid = o 325 325 + .get("cid") 326 326 + .and_then(|v| v.as_str()) 327 327 + .and_then(|s| s.parse().ok()); 328 328 + Some(MatchedRef::AtUri(FullAtUriParts { 329 329 + repo, 330 330 + collection, 331 331 + rkey, 332 332 + cid, 333 333 + })) 294 334 } 295 335 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())) 336 336 + let did = val.as_str()?.parse().ok()?; 337 337 + Some(MatchedRef::Identifier(AtIdentifier::Did(did))) 301 338 } 302 339 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())) 340 340 + let handle = val.as_str()?.parse().ok()?; 341 341 + Some(MatchedRef::Identifier(AtIdentifier::Handle(handle))) 308 342 } 309 343 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 }) 344 344 + let identifier = val.as_str()?.parse().ok()?; 345 345 + Some(MatchedRef::Identifier(identifier)) 321 346 } 322 347 } 323 348 } ··· 343 368 let mut out = Vec::new(); 344 369 for (path_parts, shape) in sources { 345 370 for val in PathWalker::new(&path_parts, skeleton) { 346 346 - if let Some(matched) = match_shape(&shape, val) { 371 371 + if let Some(matched) = match_shape(shape, val) { 347 372 out.push(matched); 348 373 } 349 374 } ··· 357 382 } 358 383 impl<'a> PathWalker<'a> { 359 384 fn new(path_parts: &'a [PathPart], skeleton: &'a Value) -> Self { 360 360 - Self { todo: vec![(path_parts, skeleton)] } 385 385 + Self { 386 386 + todo: vec![(path_parts, skeleton)], 387 387 + } 361 388 } 362 389 } 363 390 impl<'a> Iterator for PathWalker<'a> { ··· 382 409 let Some(a) = o.get(k).and_then(|v| v.as_array()) else { 383 410 continue; 384 411 }; 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 - { 412 412 + for v in a.iter().rev().filter(|c| { 413 413 + let Some(t) = t else { return true }; 414 414 + c.as_object() 415 415 + .and_then(|o| o.get("$type")) 416 416 + .and_then(|v| v.as_str()) 417 417 + .map(|s| s == t) 418 418 + .unwrap_or(false) 419 419 + }) { 398 420 self.todo.push((rest, v)) 399 421 } 400 422 } ··· 403 425 } 404 426 } 405 427 406 406 - 407 428 #[cfg(test)] 408 429 mod tests { 409 430 use super::*; 410 431 use serde_json::json; 411 432 433 433 + static TEST_CID: &str = "bafyreidffwk5wvh5l76yy7zefiqrovv6yaaegb4wg4zaq35w7nt3quix5a"; 434 434 + 412 435 #[test] 413 436 fn test_parse_record_path() -> Result<(), Box<dyn std::error::Error>> { 414 437 let cases = [ 415 438 ("", vec![]), 416 439 ("subject", vec![PathPart::Scalar("subject".into())]), 417 440 ("authorDid", vec![PathPart::Scalar("authorDid".into())]), 418 418 - ("subject.uri", vec![PathPart::Scalar("subject".into()), PathPart::Scalar("uri".into())]), 441 441 + ( 442 442 + "subject.uri", 443 443 + vec![ 444 444 + PathPart::Scalar("subject".into()), 445 445 + PathPart::Scalar("uri".into()), 446 446 + ], 447 447 + ), 419 448 ("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 - ]), 449 449 + ( 450 450 + "add[].key", 451 451 + vec![ 452 452 + PathPart::Vector("add".into(), None), 453 453 + PathPart::Scalar("key".into()), 454 454 + ], 455 455 + ), 424 456 ("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 - ]), 457 457 + ( 458 458 + "a[b.c]", 459 459 + vec![PathPart::Vector("a".into(), Some("b.c".into()))], 460 460 + ), 461 461 + ( 462 462 + "facets[app.bsky.richtext.facet].features[app.bsky.richtext.facet#mention].did", 463 463 + vec![ 464 464 + PathPart::Vector("facets".into(), Some("app.bsky.richtext.facet".into())), 465 465 + PathPart::Vector( 466 466 + "features".into(), 467 467 + Some("app.bsky.richtext.facet#mention".into()), 468 468 + ), 469 469 + PathPart::Scalar("did".into()), 470 470 + ], 471 471 + ), 431 472 ]; 432 473 433 474 for (path, expected) in cases { ··· 444 485 ("strong-ref", json!(""), None), 445 486 ("strong-ref", json!({}), None), 446 487 ("strong-ref", json!({ "uri": "abc" }), None), 447 447 - ("strong-ref", json!({ "cid": "def" }), None), 488 488 + ("strong-ref", json!({ "cid": TEST_CID }), None), 448 489 ( 449 490 "strong-ref", 450 450 - json!({ "uri": "abc", "cid": "def" }), 451 451 - Some(MatchedRef::AtUri { uri: "abc".to_string(), cid: Some("def".to_string()) }), 491 491 + json!({ "uri": "at://a.com/xx.yy.zz/1", "cid": TEST_CID }), 492 492 + Some(MatchedRef::AtUri(FullAtUriParts { 493 493 + repo: "a.com".parse().unwrap(), 494 494 + collection: "xx.yy.zz".parse().unwrap(), 495 495 + rkey: "1".parse().unwrap(), 496 496 + cid: Some(TEST_CID.parse().unwrap()), 497 497 + })), 452 498 ), 453 499 ("at-uri", json!({ "uri": "abc" }), None), 454 454 - ("at-uri", json!({ "uri": "abc", "cid": "def" }), None), 455 500 ( 456 501 "at-uri", 457 457 - json!("abc"), 458 458 - Some(MatchedRef::AtUri { uri: "abc".to_string(), cid: None }), 502 502 + json!({ "uri": "at://did:web:y.com/xx.yy.zz/1", "cid": TEST_CID }), 503 503 + None, 504 504 + ), 505 505 + ( 506 506 + "at-uri", 507 507 + json!("at://did:web:y.com/xx.yy.zz/1"), 508 508 + Some(MatchedRef::AtUri(FullAtUriParts { 509 509 + repo: "did:web:y.com".parse().unwrap(), 510 510 + collection: "xx.yy.zz".parse().unwrap(), 511 511 + rkey: "1".parse().unwrap(), 512 512 + cid: None, 513 513 + })), 459 514 ), 460 515 ("at-uri-parts", json!("abc"), None), 461 516 ("at-uri-parts", json!({}), None), 462 517 ( 463 518 "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 }), 519 519 + json!({"repo": "a.com", "collection": "xx.yy.zz", "rkey": "1", "cid": TEST_CID}), 520 520 + Some(MatchedRef::AtUri(FullAtUriParts { 521 521 + repo: "a.com".parse().unwrap(), 522 522 + collection: "xx.yy.zz".parse().unwrap(), 523 523 + rkey: "1".parse().unwrap(), 524 524 + cid: Some(TEST_CID.parse().unwrap()), 525 525 + })), 466 526 ), 467 527 ( 468 528 "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 }), 529 529 + json!({"did": "a.com", "collection": "xx.yy.zz", "rkey": "1"}), 530 530 + Some(MatchedRef::AtUri(FullAtUriParts { 531 531 + repo: "a.com".parse().unwrap(), 532 532 + collection: "xx.yy.zz".parse().unwrap(), 533 533 + rkey: "1".parse().unwrap(), 534 534 + cid: None, 535 535 + })), 471 536 ), 472 537 ( 473 538 "at-uri-parts", 474 539 // '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 }), 540 540 + json!({"did": "did:web:a.com", "repo": "z.com", "collection": "xx.yy.zz", "rkey": "1"}), 541 541 + Some(MatchedRef::AtUri(FullAtUriParts { 542 542 + repo: "z.com".parse().unwrap(), 543 543 + collection: "xx.yy.zz".parse().unwrap(), 544 544 + rkey: "1".parse().unwrap(), 545 545 + cid: None, 546 546 + })), 477 547 ), 478 548 ( 479 549 "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()) }), 550 550 + json!({"repo": "a.com", "collection": "xx.yy.zz", "rkey": "1", "cid": TEST_CID}), 551 551 + Some(MatchedRef::AtUri(FullAtUriParts { 552 552 + repo: "a.com".parse().unwrap(), 553 553 + collection: "xx.yy.zz".parse().unwrap(), 554 554 + rkey: "1".parse().unwrap(), 555 555 + cid: Some(TEST_CID.parse().unwrap()), 556 556 + })), 482 557 ), 483 558 ( 484 559 "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 }), 560 560 + json!({"repo": "a.com", "collection": "xx.yy.zz", "rkey": "1", "cid": {}}), 561 561 + Some(MatchedRef::AtUri(FullAtUriParts { 562 562 + repo: "a.com".parse().unwrap(), 563 563 + collection: "xx.yy.zz".parse().unwrap(), 564 564 + rkey: "1".parse().unwrap(), 565 565 + cid: None, 566 566 + })), 487 567 ), 488 568 ("did", json!({}), None), 489 569 ("did", json!(""), None), 490 570 ("did", json!("bad-example.com"), None), 491 491 - ("did", json!("did:plc:xyz"), Some(MatchedRef::Identifier("did:plc:xyz".to_string()))), 571 571 + ( 572 572 + "did", 573 573 + json!("did:plc:xyz"), 574 574 + Some(MatchedRef::Identifier("did:plc:xyz".parse().unwrap())), 575 575 + ), 492 576 ("handle", json!({}), None), 493 493 - ("handle", json!("bad-example.com"), Some(MatchedRef::Identifier("bad-example.com".to_string()))), 577 577 + ( 578 578 + "handle", 579 579 + json!("bad-example.com"), 580 580 + Some(MatchedRef::Identifier("bad-example.com".parse().unwrap())), 581 581 + ), 494 582 ("handle", json!("did:plc:xyz"), None), 495 583 ("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()))), 584 584 + ( 585 585 + "at-identifier", 586 586 + json!("bad-example.com"), 587 587 + Some(MatchedRef::Identifier("bad-example.com".parse().unwrap())), 588 588 + ), 589 589 + ( 590 590 + "at-identifier", 591 591 + json!("did:plc:xyz"), 592 592 + Some(MatchedRef::Identifier("did:plc:xyz".parse().unwrap())), 593 593 + ), 498 594 ]; 499 499 - for (shape, val, expected) in cases { 595 595 + for (i, (shape, val, expected)) in cases.into_iter().enumerate() { 500 596 let s = shape.try_into().unwrap(); 501 501 - let matched = match_shape(&s, &val); 502 502 - assert_eq!(matched, expected, "shape: {shape:?}, val: {val:?}"); 597 597 + let matched = match_shape(s, &val); 598 598 + assert_eq!(matched, expected, "{i}: shape: {shape:?}, val: {val:?}"); 503 599 } 504 600 } 505 601 }

+546 -194

slingshot/src/server.rs

··· 1 1 use crate::{ 2 2 CachedRecord, ErrorResponseObject, Identity, Proxy, Repo, 3 3 error::{RecordError, ServerError}, 4 4 - proxy::{extract_links, MatchedRef}, 4 4 + proxy::{FullAtUriParts, MatchedRef, extract_links, split_uri}, 5 5 record::RawRecord, 6 6 }; 7 7 - use atrium_api::types::string::{Cid, Did, Handle, Nsid, RecordKey}; 7 7 + use atrium_api::types::string::{AtIdentifier, Cid, Did, Handle, Nsid, RecordKey}; 8 8 use foyer::HybridCache; 9 9 use links::at_uri::parse_at_uri as normalize_at_uri; 10 10 use serde::Serialize; 11 11 - use std::{path::PathBuf, str::FromStr, sync::Arc, time::Instant, collections::HashMap}; 11 11 + use std::{collections::HashMap, path::PathBuf, str::FromStr, sync::Arc, time::Instant}; 12 12 use tokio::sync::mpsc; 13 13 use tokio_util::sync::CancellationToken; 14 14 15 15 use poem::{ 16 16 - Endpoint, EndpointExt, IntoResponse, Route, Server, 16 16 + Endpoint, EndpointExt, IntoResponse, Route, RouteScheme, Server, 17 17 endpoint::{StaticFileEndpoint, make_sync}, 18 18 http::Method, 19 19 listener::{ 20 20 Listener, TcpListener, 21 21 - acme::{AutoCert, LETS_ENCRYPT_PRODUCTION}, 21 21 + acme::{AutoCert, ChallengeType, LETS_ENCRYPT_PRODUCTION, LETS_ENCRYPT_STAGING}, 22 22 }, 23 23 middleware::{CatchPanic, Cors, Tracing}, 24 24 }; 25 25 use poem_openapi::{ 26 26 ApiResponse, ContactObject, ExternalDocumentObject, Object, OpenApi, OpenApiService, Tags, 27 27 - Union, 28 28 - param::Query, payload::Json, types::Example, 27 27 + Union, param::Query, payload::Json, types::Example, 29 28 }; 30 29 31 30 fn example_handle() -> String { ··· 33 32 } 34 33 fn example_did() -> String { 35 34 "did:plc:hdhoaan3xa3jiuq4fg4mefid".to_string() 35 35 + } 36 36 + fn example_service_did() -> String { 37 37 + "did:web:constellation.microcosm.blue".to_string() 36 38 } 37 39 fn example_collection() -> String { 38 40 "app.bsky.feed.like".to_string() 39 41 } 40 42 fn example_rkey() -> String { 41 43 "3lv4ouczo2b2a".to_string() 44 44 + } 45 45 + fn example_id_fragment() -> String { 46 46 + "#constellation".to_string() 42 47 } 43 48 fn example_uri() -> String { 44 49 format!( ··· 86 91 })) 87 92 } 88 93 89 89 - fn bad_request_handler_resolve_mini(err: poem::Error) -> ResolveMiniIDResponse { 90 90 - ResolveMiniIDResponse::BadRequest(Json(XrpcErrorResponseObject { 94 94 + fn bad_request_handler_resolve_mini(err: poem::Error) -> ResolveMiniDocResponse { 95 95 + ResolveMiniDocResponse::BadRequest(Json(XrpcErrorResponseObject { 96 96 + error: "InvalidRequest".to_string(), 97 97 + message: format!("Bad request, here's some info that maybe should not be exposed: {err}"), 98 98 + })) 99 99 + } 100 100 + 101 101 + fn bad_request_handler_resolve_service(err: poem::Error) -> ResolveServiceResponse { 102 102 + ResolveServiceResponse::BadRequest(Json(XrpcErrorResponseObject { 91 103 error: "InvalidRequest".to_string(), 92 104 message: format!("Bad request, here's some info that maybe should not be exposed: {err}"), 93 105 })) ··· 189 201 190 202 #[derive(ApiResponse)] 191 203 #[oai(bad_request_handler = "bad_request_handler_resolve_mini")] 192 192 - enum ResolveMiniIDResponse { 204 204 + enum ResolveMiniDocResponse { 193 205 /// Identity resolved 194 206 #[oai(status = 200)] 195 207 Ok(Json<MiniDocResponseObject>), ··· 199 211 } 200 212 201 213 #[derive(Object)] 202 202 - struct ProxyHydrationError { 203 203 - reason: String, 214 214 + #[oai(example = true)] 215 215 + struct ServiceResponseObject { 216 216 + /// The service endpoint URL, if found 217 217 + endpoint: String, 204 218 } 205 205 - 206 206 - #[derive(Object)] 207 207 - struct ProxyHydrationPending { 208 208 - url: String, 219 219 + impl Example for ServiceResponseObject { 220 220 + fn example() -> Self { 221 221 + Self { 222 222 + endpoint: "https://example.com".to_string(), 223 223 + } 224 224 + } 209 225 } 210 226 211 211 - #[derive(Object)] 212 212 - struct ProxyHydrationRecordFound { 213 213 - record: serde_json::Value, 227 227 + #[derive(ApiResponse)] 228 228 + #[oai(bad_request_handler = "bad_request_handler_resolve_service")] 229 229 + enum ResolveServiceResponse { 230 230 + /// Service resolved 231 231 + #[oai(status = 200)] 232 232 + Ok(Json<ServiceResponseObject>), 233 233 + /// Bad request or service not resolved 234 234 + #[oai(status = 400)] 235 235 + BadRequest(XrpcError), 214 236 } 215 237 216 238 #[derive(Object)] 217 217 - struct ProxyHydrationIdentifierFound { 218 218 - mini_doc: MiniDocResponseObject, 239 239 + #[oai(rename_all = "camelCase")] 240 240 + struct ProxyHydrationError { 241 241 + /// Short description of why the hydration failed 242 242 + reason: String, 243 243 + /// Whether or not it's recommended to retry requesting this item 244 244 + should_retry: bool, 245 245 + /// URL to follow up at if retrying 246 246 + follow_up: String, 219 247 } 220 248 221 249 #[derive(Object)] 222 250 #[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, 251 251 + struct ProxyHydrationPending { 252 252 + /// URL you can request to finish hydrating this item 253 253 + follow_up: String, 254 254 + /// Why this item couldn't be hydrated: 'deadline' or 'limit' 255 255 + /// 256 256 + /// - `deadline`: the item fetch didn't complete before the response was 257 257 + /// due, but will continue on slingshot in the background -- `followUp` 258 258 + /// requests are coalesced into the original item fetch to be available as 259 259 + /// early as possible. 260 260 + /// 261 261 + /// - `limit`: slingshot only attempts to hydrate the first 100 items found 262 262 + /// in a proxied response, with the remaining marked `pending`. You can 263 263 + /// request `followUp` to fetch them. 264 264 + /// 265 265 + /// In the future, Slingshot may put pending links after `limit` into a low- 266 266 + /// priority fetch queue, so that these items become available sooner on 267 267 + /// follow-up request as well. 268 268 + reason: String, 228 269 } 229 270 230 271 // todo: there's gotta be a supertrait that collects these? 231 231 - use poem_openapi::types::{Type, ToJSON, ParseFromJSON, IsObjectType}; 272 272 + use poem_openapi::types::{IsObjectType, ParseFromJSON, ToJSON, Type}; 232 273 233 274 #[derive(Union)] 234 275 #[oai(discriminator_name = "status", rename_all = "camelCase")] ··· 244 285 /// The original upstream response content 245 286 output: serde_json::Value, 246 287 /// Any hydrated records 247 247 - records: HashMap<String, Hydration<ProxyHydrationRecordFound>>, 288 288 + records: HashMap<String, Hydration<FoundRecordResponseObject>>, 248 289 /// 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>>, 290 290 + identifiers: HashMap<String, Hydration<MiniDocResponseObject>>, 254 291 } 255 292 impl Example for ProxyHydrateResponseObject { 256 293 fn example() -> Self { 257 294 Self { 258 295 output: serde_json::json!({}), 259 259 - records: HashMap::from([ 260 260 - ("asdf".into(), Hydration::Pending(ProxyHydrationPending { url: "todo".into() })), 261 261 - ]), 296 296 + records: HashMap::from([( 297 297 + "asdf".into(), 298 298 + Hydration::Pending(ProxyHydrationPending { 299 299 + follow_up: "/xrpc/com.atproto.repo.getRecord?...".to_string(), 300 300 + reason: "deadline".to_string(), 301 301 + }), 302 302 + )]), 262 303 identifiers: HashMap::new(), 263 263 - blobs: HashMap::new(), 264 304 } 265 305 } 266 306 } ··· 271 311 #[oai(status = 200)] 272 312 Ok(Json<ProxyHydrateResponseObject>), 273 313 #[oai(status = 400)] 274 274 - BadRequest(XrpcError) 314 314 + BadRequest(XrpcError), 275 315 } 276 316 277 317 #[derive(Object)] ··· 296 336 xrpc: String, 297 337 /// The destination service the request will be forwarded to 298 338 atproto_proxy: String, 339 339 + /// An optional auth token to pass on 340 340 + /// 341 341 + /// the `aud` field must match the upstream atproto_proxy service 342 342 + authorization: Option<String>, 343 343 + /// An optional set of labelers to request be applied by the upstream 344 344 + atproto_accept_labelers: Option<String>, 299 345 /// The `params` for the destination service XRPC endpoint 300 346 /// 301 347 /// Currently this will be passed along unchecked, but a future version of ··· 304 350 params: Option<serde_json::Value>, 305 351 /// Paths within the response to look for at-uris that can be hydrated 306 352 hydration_sources: Vec<HydrationSource>, 307 307 - // todo: deadline thing 308 308 - 353 353 + // todo: let clients pass a hydration deadline? 309 354 } 310 355 impl Example for ProxyQueryPayload { 311 356 fn example() -> Self { 312 357 Self { 313 358 xrpc: "app.bsky.feed.getFeedSkeleton".to_string(), 314 359 atproto_proxy: "did:web:blue.mackuba.eu#bsky_fg".to_string(), 360 360 + authorization: None, 361 361 + atproto_accept_labelers: None, 315 362 params: Some(serde_json::json!({ 316 363 "feed": "at://did:plc:oio4hkxaop4ao4wz2pp3f4cr/app.bsky.feed.generator/atproto", 317 364 })), 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 - ], 365 365 + hydration_sources: vec![HydrationSource { 366 366 + path: "feed[].post".to_string(), 367 367 + shape: "at-uri".to_string(), 368 368 + }], 324 369 } 325 370 } 326 371 } ··· 354 399 } 355 400 356 401 struct Xrpc { 402 402 + base_url: url::Url, 357 403 cache: HybridCache<String, CachedRecord>, 358 404 identity: Identity, 359 405 proxy: Arc<Proxy>, ··· 422 468 /// only retains the most recent version of a record. 423 469 Query(cid): Query<Option<String>>, 424 470 ) -> GetRecordResponse { 425 425 - self.get_record_impl(repo, collection, rkey, cid).await 471 471 + self.get_record_impl(&repo, &collection, &rkey, cid.as_deref()) 472 472 + .await 426 473 } 427 474 428 475 /// blue.microcosm.repo.getRecordByUri ··· 492 539 return bad_at_uri(); 493 540 }; 494 541 495 495 - // TODO: move this to links 496 496 - let Some(rest) = normalized.strip_prefix("at://") else { 497 497 - return bad_at_uri(); 498 498 - }; 499 499 - let Some((repo, rest)) = rest.split_once('/') else { 542 542 + let Some((repo, collection, rkey)) = split_uri(&normalized) else { 500 543 return bad_at_uri(); 501 544 }; 502 502 - let Some((collection, rest)) = rest.split_once('/') else { 503 503 - return bad_at_uri(); 504 504 - }; 505 505 - let rkey = if let Some((rkey, _rest)) = rest.split_once('?') { 506 506 - rkey 507 507 - } else { 508 508 - rest 509 509 - }; 510 545 511 546 self.get_record_impl( 512 512 - repo.to_string(), 513 513 - collection.to_string(), 514 514 - rkey.to_string(), 515 515 - cid, 547 547 + Into::<String>::into(repo).as_str(), 548 548 + collection.as_str(), 549 549 + rkey.as_str(), 550 550 + cid.as_deref(), 516 551 ) 517 552 .await 518 553 } ··· 592 627 /// Handle or DID to resolve 593 628 #[oai(example = "example_handle")] 594 629 Query(identifier): Query<String>, 595 595 - ) -> ResolveMiniIDResponse { 630 630 + ) -> ResolveMiniDocResponse { 596 631 self.resolve_mini_id(Query(identifier)).await 597 632 } 598 633 ··· 610 645 /// Handle or DID to resolve 611 646 #[oai(example = "example_handle")] 612 647 Query(identifier): Query<String>, 613 613 - ) -> ResolveMiniIDResponse { 648 648 + ) -> ResolveMiniDocResponse { 614 649 Self::resolve_mini_doc_impl(&identifier, self.identity.clone()).await 615 650 } 616 651 617 617 - async fn resolve_mini_doc_impl(identifier: &str, identity: Identity) -> ResolveMiniIDResponse { 652 652 + async fn resolve_mini_doc_impl(identifier: &str, identity: Identity) -> ResolveMiniDocResponse { 618 653 let invalid = |reason: &'static str| { 619 619 - ResolveMiniIDResponse::BadRequest(xrpc_error("InvalidRequest", reason)) 654 654 + ResolveMiniDocResponse::BadRequest(xrpc_error("InvalidRequest", reason)) 620 655 }; 621 656 622 657 let mut unverified_handle = None; ··· 681 716 } 682 717 }; 683 718 684 684 - ResolveMiniIDResponse::Ok(Json(MiniDocResponseObject { 719 719 + ResolveMiniDocResponse::Ok(Json(MiniDocResponseObject { 685 720 did: did.to_string(), 686 721 handle, 687 722 pds: partial_doc.pds, ··· 689 724 })) 690 725 } 691 726 727 727 + /// com.bad-example.identity.resolveService 728 728 + /// 729 729 + /// resolve an atproto service did + id to its http endpoint 730 730 + /// 731 731 + /// > [!important] 732 732 + /// > this endpoint is experimental and may change 733 733 + #[oai( 734 734 + path = "/com.bad-example.identity.resolveService", 735 735 + method = "get", 736 736 + tag = "ApiTags::Custom" 737 737 + )] 738 738 + async fn resolve_service( 739 739 + &self, 740 740 + /// the service's did 741 741 + #[oai(example = "example_service_did")] 742 742 + Query(did): Query<String>, 743 743 + /// id fragment, starting with '#' 744 744 + /// 745 745 + /// must be url-encoded! 746 746 + #[oai(example = "example_id_fragment")] 747 747 + Query(id): Query<String>, 748 748 + /// optionally, the exact service type to filter 749 749 + /// 750 750 + /// resolving a pds requires matching the type as well as id. service 751 751 + /// proxying ignores the type. 752 752 + Query(r#type): Query<Option<String>>, 753 753 + ) -> ResolveServiceResponse { 754 754 + let Ok(did) = Did::new(did) else { 755 755 + return ResolveServiceResponse::BadRequest(xrpc_error( 756 756 + "InvalidRequest", 757 757 + "could not parse 'did' into a DID", 758 758 + )); 759 759 + }; 760 760 + let identity = self.identity.clone(); 761 761 + Self::resolve_service_impl(&did, &id, r#type.as_deref(), identity).await 762 762 + } 763 763 + 764 764 + async fn resolve_service_impl( 765 765 + did: &Did, 766 766 + id_fragment: &str, 767 767 + service_type: Option<&str>, 768 768 + identity: Identity, 769 769 + ) -> ResolveServiceResponse { 770 770 + let invalid = |reason: &'static str| { 771 771 + ResolveServiceResponse::BadRequest(xrpc_error("InvalidRequest", reason)) 772 772 + }; 773 773 + let Ok(service_mini_doc) = identity.did_to_mini_service_doc(did).await else { 774 774 + return invalid("Failed to get DID doc"); 775 775 + }; 776 776 + let Some(service_mini_doc) = service_mini_doc else { 777 777 + return invalid("Failed to find DID doc"); 778 778 + }; 779 779 + 780 780 + let Some(matching) = service_mini_doc.get(id_fragment, service_type) else { 781 781 + return invalid("failed to match identity (and maybe type)"); 782 782 + }; 783 783 + 784 784 + ResolveServiceResponse::Ok(Json(ServiceResponseObject { 785 785 + endpoint: matching.endpoint.clone(), 786 786 + })) 787 787 + } 788 788 + 692 789 /// com.bad-example.proxy.hydrateQueryResponse 693 790 /// 694 791 /// > [!important] ··· 704 801 &self, 705 802 Json(payload): Json<ProxyQueryPayload>, 706 803 ) -> 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 804 let params = if let Some(p) = payload.params { 711 805 let serde_json::Value::Object(map) = p else { 712 806 panic!("params have to be an object"); 713 807 }; 714 808 Some(map) 715 715 - } else { None }; 809 809 + } else { 810 810 + None 811 811 + }; 812 812 + 813 813 + let Some((service_did, id_fragment)) = payload.atproto_proxy.rsplit_once("#") else { 814 814 + return ProxyHydrateResponse::BadRequest(xrpc_error( 815 815 + "BadParameter", 816 816 + "atproto_proxy could not be understood", 817 817 + )); 818 818 + }; 716 819 717 717 - match self.proxy.proxy( 718 718 - payload.xrpc, 719 719 - payload.atproto_proxy, 720 720 - params, 721 721 - ).await { 820 820 + let Ok(service_did) = service_did.parse() else { 821 821 + return ProxyHydrateResponse::BadRequest(xrpc_error( 822 822 + "BadParameter", 823 823 + "atproto_proxy service did could not be parsed", 824 824 + )); 825 825 + }; 826 826 + 827 827 + let Ok(xrpc) = payload.xrpc.parse() else { 828 828 + return ProxyHydrateResponse::BadRequest(xrpc_error( 829 829 + "BadParameter", 830 830 + "invalid NSID for xrpc param", 831 831 + )); 832 832 + }; 833 833 + 834 834 + match self 835 835 + .proxy 836 836 + .proxy( 837 837 + &service_did, 838 838 + &format!("#{id_fragment}"), 839 839 + &xrpc, 840 840 + payload.authorization.as_deref(), 841 841 + payload.atproto_accept_labelers.as_deref(), 842 842 + params, 843 843 + ) 844 844 + .await 845 845 + { 722 846 Ok(skeleton) => { 723 847 let links = match extract_links(payload.hydration_sources, &skeleton) { 724 848 Ok(l) => l, 725 849 Err(e) => { 726 850 log::warn!("problem extracting: {e:?}"); 727 727 - return ProxyHydrateResponse::BadRequest(xrpc_error("oop", "sorry, error extracting")) 851 851 + return ProxyHydrateResponse::BadRequest(xrpc_error( 852 852 + "oop", 853 853 + "sorry, error extracting", 854 854 + )); 728 855 } 729 856 }; 730 857 let mut records = HashMap::new(); 731 858 let mut identifiers = HashMap::new(); 732 732 - let mut blobs = HashMap::new(); 733 859 734 860 enum GetThing { 735 735 - Record(String, Hydration<ProxyHydrationRecordFound>), 736 736 - Identifier(String, Hydration<ProxyHydrationIdentifierFound>), 737 737 - Blob(String, Hydration<ProxyHydrationBlobFound>), 861 861 + Record(String, Hydration<FoundRecordResponseObject>), 862 862 + Identifier(String, Hydration<MiniDocResponseObject>), 738 863 } 739 864 740 865 let (tx, mut rx) = mpsc::channel(1); 741 866 742 742 - for link in links { 867 867 + let t0 = Instant::now(); 868 868 + 869 869 + for (i, link) in links.into_iter().enumerate() { 743 870 match link { 744 744 - MatchedRef::AtUri { uri, cid } => { 745 745 - if records.contains_key(&uri) { 871 871 + MatchedRef::AtUri(parts) => { 872 872 + let non_canonical_url = parts.to_uri(); 873 873 + if records.contains_key(&non_canonical_url) { 746 874 log::warn!("skipping duplicate record without checking cid"); 747 875 continue; 748 876 } 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 - })); 877 877 + let mut follow_up = self.base_url.clone(); 878 878 + follow_up.set_path("/xrpc/com.atproto.repo.getRecord"); 879 879 + follow_up 880 880 + .query_pairs_mut() 881 881 + .append_pair("repo", &Into::<String>::into(parts.repo.clone())) 882 882 + .append_pair("collection", parts.collection.as_str()) 883 883 + .append_pair("rkey", parts.rkey.as_str()); 884 884 + if let Some(ref cid) = parts.cid { 885 885 + follow_up 886 886 + .query_pairs_mut() 887 887 + .append_pair("cid", &cid.as_ref().to_string()); 888 888 + } 889 889 + 890 890 + if i >= 100 { 891 891 + records.insert( 892 892 + non_canonical_url.clone(), 893 893 + Hydration::Pending(ProxyHydrationPending { 894 894 + reason: "limit".to_string(), 895 895 + follow_up: follow_up.to_string(), 896 896 + }), 897 897 + ); 898 898 + continue; 899 899 + } else { 900 900 + records.insert( 901 901 + non_canonical_url.clone(), 902 902 + Hydration::Pending(ProxyHydrationPending { 903 903 + reason: "deadline".to_string(), 904 904 + follow_up: follow_up.to_string(), 905 905 + }), 906 906 + ); 907 907 + } 908 908 + 754 909 let tx = tx.clone(); 755 910 let identity = self.identity.clone(); 756 911 let repo = self.repo.clone(); 757 912 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() 913 913 + let FullAtUriParts { 914 914 + repo: ident, 915 915 + collection, 916 916 + rkey, 917 917 + cid, 918 918 + } = parts; 919 919 + let did = match ident { 920 920 + AtIdentifier::Did(did) => did, 921 921 + AtIdentifier::Handle(handle) => { 922 922 + let Ok(Some(did)) = identity.handle_to_did(handle).await 923 923 + else { 924 924 + let res = Hydration::Error(ProxyHydrationError { 925 925 + reason: "could not resolve handle".to_string(), 926 926 + should_retry: true, 927 927 + follow_up: follow_up.to_string(), 928 928 + }); 929 929 + return if tx 930 930 + .send(GetThing::Record(non_canonical_url, res)) 931 931 + .await 932 932 + .is_ok() 933 933 + { 934 934 + metrics::counter!("slingshot_hydrated_one", "type" => "record", "ontime" => "true").increment(1); 935 935 + } else { 936 936 + metrics::counter!("slingshot_hydrated_one", "type" => "record", "ontime" => "false").increment(1); 937 937 + }; 938 938 + }; 939 939 + did 940 940 + } 767 941 }; 768 942 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"); 943 943 + let res = 944 944 + match repo.get_record(&did, &collection, &rkey, &cid).await { 945 945 + Ok(CachedRecord::Deleted) => { 946 946 + Hydration::Error(ProxyHydrationError { 947 947 + reason: "record deleted".to_string(), 948 948 + should_retry: false, 949 949 + follow_up: follow_up.to_string(), 950 950 + }) 951 951 + } 952 952 + Ok(CachedRecord::Found(RawRecord { 953 953 + cid: found_cid, 954 954 + record, 955 955 + })) => { 956 956 + if cid 957 957 + .as_ref() 958 958 + .map(|expected| *expected != found_cid) 959 959 + .unwrap_or(false) 960 960 + { 961 961 + Hydration::Error(ProxyHydrationError { 962 962 + reason: "not found".to_string(), 963 963 + should_retry: false, 964 964 + follow_up: follow_up.to_string(), 965 965 + }) 966 966 + } else if let Ok(value) = serde_json::from_str(&record) 967 967 + { 968 968 + let canonical_uri = FullAtUriParts { 969 969 + repo: AtIdentifier::Did(did), 970 970 + collection, 971 971 + rkey, 972 972 + cid: None, // not used for .to_uri 973 973 + } 974 974 + .to_uri(); 975 975 + Hydration::Found(FoundRecordResponseObject { 976 976 + cid: Some(found_cid.as_ref().to_string()), 977 977 + uri: canonical_uri, 978 978 + value, 979 979 + }) 980 980 + } else { 981 981 + Hydration::Error(ProxyHydrationError { 982 982 + reason: "could not parse upstream response" 983 983 + .to_string(), 984 984 + should_retry: false, 985 985 + follow_up: follow_up.to_string(), 986 986 + }) 987 987 + } 782 988 } 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 989 989 + Err(e) => { 990 990 + log::warn!("finally oop {e:?}"); 991 991 + Hydration::Error(ProxyHydrationError { 992 992 + reason: "failed to fetch record".to_string(), 993 993 + should_retry: true, // TODO 994 994 + follow_up: follow_up.to_string(), 995 995 + }) 996 996 + } 997 997 + }; 998 998 + if tx 999 999 + .send(GetThing::Record(non_canonical_url, res)) 1000 1000 + .await 1001 1001 + .is_ok() 1002 1002 + { 1003 1003 + metrics::counter!("slingshot_hydrated_one", "type" => "record", "ontime" => "true").increment(1); 1004 1004 + } else { 1005 1005 + metrics::counter!("slingshot_hydrated_one", "type" => "record", "ontime" => "false").increment(1); 1006 1006 + } 796 1007 }); 797 1008 } 798 1009 MatchedRef::Identifier(id) => { 799 799 - if identifiers.contains_key(&id) { 1010 1010 + let identifier: String = id.clone().into(); 1011 1011 + if identifiers.contains_key(&identifier) { 800 1012 continue; 801 1013 } 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 - })); 1014 1014 + 1015 1015 + let mut follow_up = self.base_url.clone(); 1016 1016 + follow_up.set_path("/xrpc/blue.microcosm.identity.resolveMiniDoc"); 1017 1017 + 1018 1018 + follow_up 1019 1019 + .query_pairs_mut() 1020 1020 + .append_pair("identifier", &identifier); 1021 1021 + 1022 1022 + if i >= 100 { 1023 1023 + identifiers.insert( 1024 1024 + identifier.clone(), 1025 1025 + Hydration::Pending(ProxyHydrationPending { 1026 1026 + reason: "limit".to_string(), 1027 1027 + follow_up: follow_up.to_string(), 1028 1028 + }), 1029 1029 + ); 1030 1030 + continue; 1031 1031 + } else { 1032 1032 + identifiers.insert( 1033 1033 + identifier.clone(), 1034 1034 + Hydration::Pending(ProxyHydrationPending { 1035 1035 + reason: "deadline".to_string(), 1036 1036 + follow_up: follow_up.to_string(), 1037 1037 + }), 1038 1038 + ); 1039 1039 + } 1040 1040 + 807 1041 let tx = tx.clone(); 808 1042 let identity = self.identity.clone(); 809 1043 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) => { 1044 1044 + let res = match Self::resolve_mini_doc_impl(&identifier, identity) 1045 1045 + .await 1046 1046 + { 1047 1047 + ResolveMiniDocResponse::Ok(Json(mini_doc)) => { 1048 1048 + Hydration::Found(mini_doc) 1049 1049 + } 1050 1050 + ResolveMiniDocResponse::BadRequest(e) => { 815 1051 log::warn!("minidoc fail: {:?}", e.0); 816 1052 Hydration::Error(ProxyHydrationError { 817 1053 reason: "failed to resolve mini doc".to_string(), 1054 1054 + should_retry: false, 1055 1055 + follow_up: follow_up.to_string(), 818 1056 }) 819 1057 } 820 1058 }; 821 821 - tx.send(GetThing::Identifier(id, res)).await 1059 1059 + if tx.send(GetThing::Identifier(identifier, res)).await.is_ok() { 1060 1060 + metrics::counter!("slingshot_hydrated_one", "type" => "identity", "ontime" => "true").increment(1); 1061 1061 + } else { 1062 1062 + metrics::counter!("slingshot_hydrated_one", "type" => "identity", "ontime" => "false").increment(1); 1063 1063 + } 822 1064 }); 823 1065 } 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 1066 } 836 1067 } 837 1068 // so the channel can close when all are completed 838 1069 // (we shoudl be doing a timeout...) 839 1070 drop(tx); 840 1071 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 - }; 1072 1072 + let deadline = t0 + std::time::Duration::from_secs_f64(1.6); 1073 1073 + let res = tokio::time::timeout_at(deadline.into(), async { 1074 1074 + while let Some(hydration) = rx.recv().await { 1075 1075 + match hydration { 1076 1076 + GetThing::Record(uri, h) => { 1077 1077 + if let Some(r) = records.get_mut(&uri) { 1078 1078 + match (&r, &h) { 1079 1079 + (_, Hydration::Found(_)) => *r = h, // always replace if found 1080 1080 + (Hydration::Pending(_), _) => *r = h, // or if it was pending 1081 1081 + _ => {} // else leave it 1082 1082 + } 1083 1083 + } else { 1084 1084 + records.insert(uri, h); 1085 1085 + } 1086 1086 + } 1087 1087 + GetThing::Identifier(identifier, md) => { 1088 1088 + identifiers.insert(identifier.to_string(), md); 1089 1089 + } 1090 1090 + }; 1091 1091 + } 1092 1092 + }) 1093 1093 + .await; 1094 1094 + 1095 1095 + if res.is_ok() { 1096 1096 + metrics::histogram!("slingshot_hydration_all_completed").record(t0.elapsed()); 1097 1097 + } else { 1098 1098 + metrics::counter!("slingshot_hydration_cut_off").increment(1); 847 1099 } 848 1100 849 1101 ProxyHydrateResponse::Ok(Json(ProxyHydrateResponseObject { 850 1102 output: skeleton, 851 1103 records, 852 1104 identifiers, 853 853 - blobs, 854 1105 })) 855 1106 } 856 1107 Err(e) => { ··· 858 1109 ProxyHydrateResponse::BadRequest(xrpc_error("oop", "sorry")) 859 1110 } 860 1111 } 861 861 - 862 1112 } 863 1113 864 1114 async fn get_record_impl( 865 1115 &self, 866 866 - repo: String, 867 867 - collection: String, 868 868 - rkey: String, 869 869 - cid: Option<String>, 1116 1116 + repo: &str, 1117 1117 + collection: &str, 1118 1118 + rkey: &str, 1119 1119 + cid: Option<&str>, 870 1120 ) -> GetRecordResponse { 871 871 - let did = match Did::new(repo.clone()) { 1121 1121 + let did = match Did::new(repo.to_string()) { 872 1122 Ok(did) => did, 873 1123 Err(_) => { 874 1124 let Ok(handle) = Handle::new(repo.to_lowercase()) else { ··· 899 1149 } 900 1150 }; 901 1151 902 902 - let Ok(collection) = Nsid::new(collection) else { 1152 1152 + let Ok(collection) = Nsid::new(collection.to_string()) else { 903 1153 return GetRecordResponse::BadRequest(xrpc_error( 904 1154 "InvalidRequest", 905 1155 "Invalid NSID for collection", 906 1156 )); 907 1157 }; 908 1158 909 909 - let Ok(rkey) = RecordKey::new(rkey) else { 1159 1159 + let Ok(rkey) = RecordKey::new(rkey.to_string()) else { 910 1160 return GetRecordResponse::BadRequest(xrpc_error("InvalidRequest", "Invalid rkey")); 911 1161 }; 912 1162 913 1163 let cid: Option<Cid> = if let Some(cid) = cid { 914 914 - let Ok(cid) = Cid::from_str(&cid) else { 1164 1164 + let Ok(cid) = Cid::from_str(cid) else { 915 1165 return GetRecordResponse::BadRequest(xrpc_error("InvalidRequest", "Invalid CID")); 916 1166 }; 917 1167 Some(cid) ··· 1060 1310 identity: Identity, 1061 1311 repo: Repo, 1062 1312 proxy: Proxy, 1063 1063 - acme_domain: Option<String>, 1313 1313 + base_url: url::Url, 1314 1314 + tls_domain: Option<String>, 1315 1315 + tls_certs: Option<PathBuf>, 1316 1316 + tls_ipv6: bool, 1317 1317 + acme_challenge_redirect: Option<String>, 1064 1318 acme_contact: Option<String>, 1065 1065 - acme_cache_path: Option<PathBuf>, 1066 1066 - acme_ipv6: bool, 1319 1319 + acme_staging: bool, 1067 1320 shutdown: CancellationToken, 1068 1321 bind: std::net::SocketAddr, 1069 1322 ) -> Result<(), ServerError> { ··· 1071 1324 let proxy = Arc::new(proxy); 1072 1325 let api_service = OpenApiService::new( 1073 1326 Xrpc { 1327 1327 + base_url, 1074 1328 cache, 1075 1329 identity, 1076 1330 proxy, ··· 1079 1333 "Slingshot", 1080 1334 env!("CARGO_PKG_VERSION"), 1081 1335 ) 1082 1082 - .server(if let Some(ref h) = acme_domain { 1336 1336 + .server(if let Some(ref h) = tls_domain { 1083 1337 format!("https://{h}") 1084 1338 } else { 1085 1339 format!("http://{bind}") // yeah should probably fix this for reverse-proxy scenarios but it's ok for dev for now ··· 1095 1349 "https://microcosm.blue/slingshot", 1096 1350 )); 1097 1351 1098 1098 - let mut app = Route::new() 1352 1352 + let app = Route::new() 1099 1353 .at("/", StaticFileEndpoint::new("./static/index.html")) 1100 1354 .nest("/openapi", api_service.spec_endpoint()) 1101 1355 .nest("/xrpc/", api_service); 1102 1356 1103 1103 - if let Some(domain) = acme_domain { 1357 1357 + let cors = Cors::new() 1358 1358 + .allow_origin_regex("*") 1359 1359 + .allow_methods([Method::GET, Method::POST]) 1360 1360 + .allow_credentials(false); 1361 1361 + 1362 1362 + if let Some(domain) = tls_domain { 1104 1363 rustls::crypto::aws_lc_rs::default_provider() 1105 1364 .install_default() 1106 1365 .expect("alskfjalksdjf"); 1107 1366 1108 1108 - app = app.at("/.well-known/did.json", get_did_doc(&domain)); 1367 1367 + let app = app 1368 1368 + .at("/.well-known/did.json", get_did_doc(&domain)) 1369 1369 + .with(cors); 1109 1370 1110 1110 - let mut auto_cert = AutoCert::builder() 1111 1111 - .directory_url(LETS_ENCRYPT_PRODUCTION) 1112 1112 - .domain(&domain); 1113 1371 if let Some(contact) = acme_contact { 1114 1114 - auto_cert = auto_cert.contact(contact); 1372 1372 + let (listener, app) = acmify(app, domain, tls_certs, tls_ipv6, contact, acme_staging)?; 1373 1373 + run(listener, app, shutdown).await 1374 1374 + } else { 1375 1375 + let certs = tls_certs.expect("certs path must be set for non-acme tls"); 1376 1376 + let (listener, app) = tlsify(app, domain, certs, tls_ipv6, acme_challenge_redirect)?; 1377 1377 + run(listener, app, shutdown).await 1115 1378 } 1116 1116 - if let Some(cache_path) = acme_cache_path { 1117 1117 - auto_cert = auto_cert.cache_path(cache_path); 1118 1118 - } 1119 1119 - let auto_cert = auto_cert.build().map_err(ServerError::AcmeBuildError)?; 1379 1379 + } else { 1380 1380 + run(TcpListener::bind(bind), app.with(cors), shutdown).await 1381 1381 + } 1382 1382 + } 1120 1383 1121 1121 - run( 1122 1122 - TcpListener::bind(if acme_ipv6 { "[::]:443" } else { "0.0.0.0:443" }).acme(auto_cert), 1123 1123 - app, 1124 1124 - shutdown, 1125 1125 - ) 1126 1126 - .await 1384 1384 + fn acmify( 1385 1385 + app: impl Endpoint + 'static, 1386 1386 + domain: String, 1387 1387 + tls_certs: Option<PathBuf>, 1388 1388 + tls_ipv6: bool, 1389 1389 + acme_contact: String, 1390 1390 + acme_staging: bool, 1391 1391 + ) -> Result<(impl Listener + 'static, impl Endpoint + 'static), ServerError> { 1392 1392 + let mut auto_cert = AutoCert::builder() 1393 1393 + .contact(acme_contact) 1394 1394 + .directory_url(if acme_staging { 1395 1395 + LETS_ENCRYPT_STAGING 1396 1396 + } else { 1397 1397 + LETS_ENCRYPT_PRODUCTION 1398 1398 + }) 1399 1399 + .domain(&domain) 1400 1400 + .challenge_type(ChallengeType::Http01); 1401 1401 + 1402 1402 + if let Some(path) = tls_certs { 1403 1403 + auto_cert = auto_cert.cache_path(path); 1127 1404 } else { 1128 1128 - run(TcpListener::bind(bind), app, shutdown).await 1405 1405 + log::warn!( 1406 1406 + "provisioning acme certs without `--tls-certs` folder configured, you might hit letsencrypt rate limits." 1407 1407 + ); 1129 1408 } 1409 1409 + 1410 1410 + let auto_cert = auto_cert.build().map_err(ServerError::AcmeBuildError)?; 1411 1411 + 1412 1412 + let app = RouteScheme::new() 1413 1413 + .https(app) 1414 1414 + .http(auto_cert.http_01_endpoint()); 1415 1415 + 1416 1416 + let listener = TcpListener::bind(if tls_ipv6 { "[::]:443" } else { "0.0.0.0:443" }) 1417 1417 + .acme(auto_cert) 1418 1418 + .combine(TcpListener::bind(if tls_ipv6 { 1419 1419 + "[::]:80" 1420 1420 + } else { 1421 1421 + "0.0.0.0:80" 1422 1422 + })); 1423 1423 + 1424 1424 + Ok((listener, app)) 1130 1425 } 1131 1426 1132 1132 - async fn run<L>(listener: L, app: Route, shutdown: CancellationToken) -> Result<(), ServerError> 1427 1427 + fn tlsify( 1428 1428 + app: impl Endpoint + 'static, 1429 1429 + domain: String, 1430 1430 + tls_certs: PathBuf, 1431 1431 + tls_ipv6: bool, 1432 1432 + acme_challenge_redirect: Option<String>, 1433 1433 + ) -> Result<(impl Listener + 'static, impl Endpoint + 'static), ServerError> { 1434 1434 + use poem::listener::{RustlsCertificate, RustlsConfig}; 1435 1435 + use std::path::Path; 1436 1436 + 1437 1437 + fn load_tls_config(f: &Path, domain: &str) -> Result<RustlsConfig, std::io::Error> { 1438 1438 + let cert_contents = std::fs::read(f.join("cert.pem")) 1439 1439 + .inspect_err(|e| log::error!("failed to read cert file in {f:?}: {e}"))?; 1440 1440 + 1441 1441 + let key_contents = std::fs::read(f.join("key.pem")) 1442 1442 + .inspect_err(|e| log::error!("failed to read key file in {f:?}: {e}"))?; 1443 1443 + 1444 1444 + let cert = RustlsCertificate::new() 1445 1445 + .cert(cert_contents) 1446 1446 + .key(key_contents); 1447 1447 + Ok(RustlsConfig::new().certificate(domain, cert)) 1448 1448 + } 1449 1449 + 1450 1450 + let listener = TcpListener::bind(if tls_ipv6 { "[::]:443" } else { "0.0.0.0:443" }) 1451 1451 + .rustls(async_stream::stream! { 1452 1452 + loop { 1453 1453 + if let Ok(tls_config) = load_tls_config(&tls_certs, &domain) { 1454 1454 + // TODO: cert reload healthcheck 1455 1455 + yield tls_config; 1456 1456 + } else { 1457 1457 + log::warn!("failed to load tls config."); 1458 1458 + } 1459 1459 + tokio::time::sleep(std::time::Duration::from_secs(3600 * 12)).await; 1460 1460 + } 1461 1461 + }) 1462 1462 + // TODO: should be allowed to run in tls mode without binding port 80 if not forwarding acme challenges 1463 1463 + .combine(TcpListener::bind(if tls_ipv6 { 1464 1464 + "[::]:80" 1465 1465 + } else { 1466 1466 + "0.0.0.0:80" 1467 1467 + })); 1468 1468 + 1469 1469 + let app = if let Some(redir) = acme_challenge_redirect { 1470 1470 + use poem::web; 1471 1471 + 1472 1472 + let redirect = poem::endpoint::make_sync(move |req| { 1473 1473 + let token = req.path_params::<String>().unwrap(); 1474 1474 + metrics::counter!("http_challenge_redirects").increment(1); 1475 1475 + web::Redirect::temporary(format!("{redir}{token}")) 1476 1476 + }); 1477 1477 + 1478 1478 + RouteScheme::new() 1479 1479 + .https(app) 1480 1480 + .http(Route::new().at("/.well-known/acme-challenge/:token", redirect)) 1481 1481 + } else { 1482 1482 + // just uh... 404 for port 80? should probably reply with something. 1483 1483 + RouteScheme::new().https(app).http(Route::new()) 1484 1484 + }; 1485 1485 + 1486 1486 + Ok((listener, app)) 1487 1487 + } 1488 1488 + 1489 1489 + async fn run<L, A>(listener: L, app: A, shutdown: CancellationToken) -> Result<(), ServerError> 1133 1490 where 1134 1491 L: Listener + 'static, 1492 1492 + A: Endpoint + 'static, 1135 1493 { 1136 1494 let app = app 1137 1137 - .with( 1138 1138 - Cors::new() 1139 1139 - .allow_origin_regex("*") 1140 1140 - .allow_methods([Method::GET, Method::POST]) 1141 1141 - .allow_credentials(false), 1142 1142 - ) 1143 1495 .with(CatchPanic::new()) 1144 1496 .around(request_counter) 1145 1497 .with(Tracing);