diff --git a/Cargo.lock b/Cargo.lock index 4679a5c..d65d177 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7838,7 +7838,10 @@ dependencies = [ "async-trait", "bytes", "futures", + "serde", + "serde_json", "thiserror 2.0.18", + "tranquil-types", ] [[package]] @@ -7855,9 +7858,9 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", + "tranquil-infra", "tranquil-types", "unicode-segmentation", - "urlencoding", "wiremock", ] @@ -7880,6 +7883,7 @@ dependencies = [ "sqlx", "tokio", "tracing", + "tranquil-infra", "tranquil-types", "uuid", ] @@ -7911,6 +7915,7 @@ dependencies = [ "tranquil-config", "tranquil-crypto", "tranquil-db-traits", + "tranquil-infra", "tranquil-pds", "tranquil-scopes", "tranquil-types", @@ -7991,6 +7996,7 @@ dependencies = [ "tranquil-config", "tranquil-db", "tranquil-db-traits", + "tranquil-infra", "tranquil-lexicon", "tranquil-oauth", "tranquil-oauth-server", @@ -8221,10 +8227,14 @@ dependencies = [ "cid", "jacquard-common", "rand 0.8.5", + "reqwest", "serde", "serde_json", "sqlx", "thiserror 2.0.18", + "tokio", + "tracing", + "url", "uuid", ] diff --git a/Cargo.toml b/Cargo.toml index 85c38c5..cdbf10a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -137,6 +137,7 @@ tower-layer = "0.3" tracing = "0.1" tracing-subscriber = "0.3" urlencoding = "2.1" +url = "2.5" uuid = { version = "1.19", features = ["v4", "v5", "v7", "fast-rng", "serde"] } webauthn-rs = { version = "0.5", features = ["danger-allow-state-serialisation", "danger-user-presence-only-security-keys", "conditional-ui"] } webauthn-rs-proto = "0.5" diff --git a/crates/tranquil-api/src/delegation.rs b/crates/tranquil-api/src/delegation.rs index 2e52499..7262bf4 100644 --- a/crates/tranquil-api/src/delegation.rs +++ b/crates/tranquil-api/src/delegation.rs @@ -12,8 +12,8 @@ use tranquil_pds::api::{ }; use tranquil_pds::auth::{Active, Auth}; use tranquil_pds::delegation::{ - DelegationActionType, SCOPE_PRESETS, ValidatedDelegationScope, verify_can_add_controllers, - verify_can_control_accounts, + DelegationActionType, IdentityResolutionError, SCOPE_PRESETS, ValidatedDelegationScope, + verify_can_add_controllers, verify_can_control_accounts, }; use tranquil_pds::rate_limit::{AccountCreationLimit, RateLimited}; use tranquil_pds::state::AppState; @@ -65,16 +65,16 @@ pub async fn add_controller( ) -> Result, ApiError> { let resolved = tranquil_pds::delegation::resolve_identity(&state, &input.controller_did) .await - .map_err(|_| ApiError::ControllerNotFound)?; + .map_err(|e| match e { + IdentityResolutionError::PdsEndpoint(_) => ApiError::InvalidDelegation( + "Controller PDS endpoint isn't a usable https URL".into(), + ), + IdentityResolutionError::DidResolution(_) => ApiError::ControllerNotFound, + })?; if !resolved.is_local && let Some(ref pds_url) = resolved.pds_url { - if !pds_url.starts_with("https://") { - return Err(ApiError::InvalidDelegation( - "Controller PDS must use HTTPS".into(), - )); - } match state .cross_pds_oauth .check_remote_is_delegated(pds_url, &input.controller_did) @@ -477,7 +477,12 @@ pub async fn resolve_controller( let resolved = tranquil_pds::delegation::resolve_identity(&state, &did) .await - .map_err(|_| ApiError::ControllerNotFound)?; + .map_err(|e| match e { + IdentityResolutionError::PdsEndpoint(_) => ApiError::InvalidDelegation( + "Controller PDS endpoint isn't a usable https URL".into(), + ), + IdentityResolutionError::DidResolution(_) => ApiError::ControllerNotFound, + })?; Ok(Json(resolved)) } diff --git a/crates/tranquil-cache/Cargo.toml b/crates/tranquil-cache/Cargo.toml index de971ff..bbc805e 100644 --- a/crates/tranquil-cache/Cargo.toml +++ b/crates/tranquil-cache/Cargo.toml @@ -9,7 +9,7 @@ valkey = ["dep:redis"] [dependencies] tranquil-config = { workspace = true } -tranquil-infra = { workspace = true } +tranquil-infra = { workspace = true, features = ["cache-keys"] } tranquil-ripple = { workspace = true } async-trait = { workspace = true } diff --git a/crates/tranquil-cache/src/lib.rs b/crates/tranquil-cache/src/lib.rs index 2551cf8..67804cb 100644 --- a/crates/tranquil-cache/src/lib.rs +++ b/crates/tranquil-cache/src/lib.rs @@ -1,4 +1,6 @@ -pub use tranquil_infra::{Cache, CacheError, DistributedRateLimiter}; +pub use tranquil_infra::{ + Cache, CacheError, DistributedRateLimiter, cache_keys, cached_json, read_json, write_json, +}; use async_trait::async_trait; use std::sync::Arc; @@ -173,11 +175,10 @@ pub async fn create_cache( ) -> Result<(Arc, Arc), CacheInitError> { let cache_cfg = tranquil_config::try_get().map(|c| &c.cache); let backend = cache_cfg.map(|c| c.backend.as_str()).unwrap_or("ripple"); - let valkey_url = cache_cfg.and_then(|c| c.valkey_url.as_deref()); #[cfg(feature = "valkey")] if backend == "valkey" { - if let Some(url) = valkey_url { + if let Some(url) = cache_cfg.and_then(|c| c.valkey_url.as_deref()) { match ValkeyCache::new(url).await { Ok(cache) => { tracing::info!("using valkey cache at {url}"); diff --git a/crates/tranquil-infra/Cargo.toml b/crates/tranquil-infra/Cargo.toml index 4c7436e..acef127 100644 --- a/crates/tranquil-infra/Cargo.toml +++ b/crates/tranquil-infra/Cargo.toml @@ -4,8 +4,16 @@ version.workspace = true edition.workspace = true license.workspace = true +[features] +testing = [] +cache-keys = ["dep:tranquil-types"] + [dependencies] +tranquil-types = { workspace = true, optional = true } + async-trait = { workspace = true } bytes = { workspace = true } futures = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/crates/tranquil-infra/src/cache_keys.rs b/crates/tranquil-infra/src/cache_keys.rs new file mode 100644 index 0000000..c01ac62 --- /dev/null +++ b/crates/tranquil-infra/src/cache_keys.rs @@ -0,0 +1,103 @@ +use tranquil_types::{ + CidLink, ClientId, CrossPdsState, Did, EmailTokenPurpose, Handle, Jti, JwksUri, Nsid, PdsUrl, + SsoIssuer, SsoJwksUri, +}; + +pub fn session_key(did: &Did, jti: &Jti) -> String { + format!("auth:session:{}:{}", did, jti) +} + +pub fn signing_key_key(did: &Did) -> String { + format!("auth:key:{}", did) +} + +pub fn user_status_key(did: &Did) -> String { + format!("auth:status:{}", did) +} + +pub fn handle_key(handle: &Handle) -> String { + format!("handle:{}", handle) +} + +pub fn reauth_key(did: &Did) -> String { + format!("reauth:{}", did) +} + +pub fn plc_doc_key(did: &Did) -> String { + format!("plc:doc:{}", did) +} + +pub fn plc_data_key(did: &Did) -> String { + format!("plc:data:{}", did) +} + +pub fn did_web_doc_key(did: &Did) -> String { + format!("did:web:doc:{}", did) +} + +pub fn email_update_key(did: &Did) -> String { + format!("email_update:{}", did) +} + +pub fn email_token_key(did: &Did, purpose: EmailTokenPurpose) -> String { + format!("email_token:{}:{}", purpose, did) +} + +pub fn legacy_2fa_challenge_key(did: &Did) -> String { + format!("legacy_2fa:{}", did) +} + +pub fn legacy_2fa_cooldown_key(did: &Did) -> String { + format!("legacy_2fa_cooldown:{}", did) +} + +pub fn scope_ref_key(cid: &CidLink) -> String { + format!("scope_ref:{}", cid) +} + +pub fn auto_verify_sent_key(did: &Did) -> String { + format!("auto_verify_sent:{}", did) +} + +pub fn permission_set_key(nsid: &Nsid, aud: Option<&str>) -> String { + match aud { + Some(a) => format!("permset:{}:{}", nsid, a), + None => format!("permset:{}", nsid), + } +} + +pub fn oauth_client_meta_key(client_id: &ClientId) -> String { + format!("oauth:client_meta:{}", client_id) +} + +pub fn oauth_client_jwks_key(jwks_uri: &JwksUri) -> String { + format!("oauth:jwks:{}", jwks_uri.canonical()) +} + +pub fn oauth_client_jwks_cooldown_key(jwks_uri: &JwksUri) -> String { + format!("oauth:jwks_cooldown:{}", jwks_uri.canonical()) +} + +pub fn sso_jwks_key(jwks_uri: &SsoJwksUri) -> String { + format!("sso:jwks:{}", jwks_uri.canonical()) +} + +pub fn oidc_discovery_key(issuer: &SsoIssuer) -> String { + format!("oidc:discovery:{}", issuer.canonical()) +} + +pub fn cross_pds_state_key(state: &CrossPdsState) -> String { + format!("cross_pds_state:{}", state) +} + +pub fn cross_pds_oauth_meta_key(pds_url: &PdsUrl) -> String { + format!("cross_pds_oauth_meta:v2:{}", pds_url.canonical()) +} + +pub fn lexicon_doc_key(nsid: &Nsid) -> String { + format!("lexicon:doc:{}", nsid) +} + +pub fn lexicon_negative_key(nsid: &Nsid) -> String { + format!("lexicon:neg:{}", nsid) +} diff --git a/crates/tranquil-infra/src/lib.rs b/crates/tranquil-infra/src/lib.rs index 9928884..76a0783 100644 --- a/crates/tranquil-infra/src/lib.rs +++ b/crates/tranquil-infra/src/lib.rs @@ -1,6 +1,15 @@ +#[cfg(feature = "cache-keys")] +pub mod cache_keys; + +#[cfg(feature = "testing")] +mod memory_cache; +#[cfg(feature = "testing")] +pub use memory_cache::MemoryCache; + use async_trait::async_trait; use bytes::Bytes; use futures::Stream; +use std::future::Future; use std::pin::Pin; use std::time::Duration; @@ -57,6 +66,42 @@ pub trait Cache: Send + Sync { } } +pub async fn read_json(cache: &dyn Cache, key: &str) -> Option { + let json = cache.get(key).await?; + serde_json::from_str(&json).ok() +} + +pub async fn write_json( + cache: &dyn Cache, + key: &str, + value: &T, + ttl: Duration, +) { + if let Ok(json) = serde_json::to_string(value) { + let _ = cache.set(key, &json, ttl).await; + } +} + +pub async fn cached_json( + cache: &dyn Cache, + key: &str, + ttl: Duration, + fetch: impl FnOnce() -> Fut, +) -> Result +where + T: serde::Serialize + serde::de::DeserializeOwned, + Fut: Future>, +{ + match read_json(cache, key).await { + Some(value) => Ok(value), + None => { + let value = fetch().await?; + write_json(cache, key, &value, ttl).await; + Ok(value) + } + } +} + #[async_trait] pub trait DistributedRateLimiter: Send + Sync { async fn check_rate_limit(&self, key: &str, limit: u32, window_ms: u64) -> bool; diff --git a/crates/tranquil-infra/src/memory_cache.rs b/crates/tranquil-infra/src/memory_cache.rs new file mode 100644 index 0000000..eca7692 --- /dev/null +++ b/crates/tranquil-infra/src/memory_cache.rs @@ -0,0 +1,74 @@ +use crate::{Cache, CacheError}; +use async_trait::async_trait; +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +struct Entry { + value: Vec, + expires_at: Instant, +} + +#[derive(Default)] +pub struct MemoryCache { + entries: Mutex>, +} + +impl MemoryCache { + pub fn new() -> Self { + Self::default() + } + + fn read(&self, key: &str) -> Option> { + let now = Instant::now(); + let mut entries = self.entries.lock().unwrap_or_else(|e| e.into_inner()); + match entries.get(key) { + Some(entry) if entry.expires_at > now => Some(entry.value.clone()), + Some(_) => { + entries.remove(key); + None + } + None => None, + } + } + + fn write(&self, key: &str, value: Vec, ttl: Duration) { + let entry = Entry { + value, + expires_at: Instant::now() + ttl, + }; + self.entries + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(key.to_string(), entry); + } +} + +#[async_trait] +impl Cache for MemoryCache { + async fn get(&self, key: &str) -> Option { + self.read(key).and_then(|v| String::from_utf8(v).ok()) + } + + async fn set(&self, key: &str, value: &str, ttl: Duration) -> Result<(), CacheError> { + self.write(key, value.as_bytes().to_vec(), ttl); + Ok(()) + } + + async fn delete(&self, key: &str) -> Result<(), CacheError> { + self.entries + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(key); + Ok(()) + } + + async fn get_bytes(&self, key: &str) -> Option> { + self.read(key) + } + + async fn set_bytes(&self, key: &str, value: &[u8], ttl: Duration) -> Result<(), CacheError> { + self.write(key, value.to_vec(), ttl); + Ok(()) + } +} diff --git a/crates/tranquil-oauth-server/src/endpoints/delegation.rs b/crates/tranquil-oauth-server/src/endpoints/delegation.rs index 748f90a..9f4dbba 100644 --- a/crates/tranquil-oauth-server/src/endpoints/delegation.rs +++ b/crates/tranquil-oauth-server/src/endpoints/delegation.rs @@ -13,7 +13,8 @@ use tranquil_pds::rate_limit::{LoginLimit, OAuthRateLimited, TotpVerifyLimit}; use tranquil_pds::state::AppState; use tranquil_pds::types::PlainPassword; use tranquil_pds::util::ClientIp; -use tranquil_types::did_doc::{extract_handle, extract_pds_endpoint}; +use tranquil_types::did_doc::{PdsEndpointError, extract_handle, extract_pds_endpoint}; +use tranquil_types::url_kind; use tranquil_types::{Did, RequestId}; #[allow(clippy::result_large_err)] @@ -231,11 +232,17 @@ pub async fn delegation_auth( } }; - let pds_url = match extract_pds_endpoint(&did_doc) { - Some(url) => url, - None => { + let pds_url = match extract_pds_endpoint::(&did_doc) { + Ok(url) => url, + Err(PdsEndpointError::Missing) => { return DelegationAuthResponse::err("Controller has no PDS endpoint"); } + Err(PdsEndpointError::Invalid(e)) => { + tracing::warn!(controller = %controller_did, error = %e, "Controller PDS endpoint rejected"); + return DelegationAuthResponse::err( + "Controller PDS endpoint isn't a usable https URL", + ); + } }; let hostname = &tranquil_config::get().server.hostname; @@ -447,7 +454,7 @@ pub async fn delegation_auth_token( #[derive(Debug, Deserialize)] pub struct CrossPdsCallbackParams { pub code: tranquil_types::AuthorizationCode, - pub state: String, + pub state: tranquil_types::CrossPdsState, pub iss: Option, } @@ -474,7 +481,7 @@ pub async fn delegation_callback( if let Some(ref expected_issuer) = auth_state.expected_issuer { match ¶ms.iss { - Some(iss) if iss != expected_issuer => { + Some(iss) if iss.as_str() != expected_issuer.as_str() => { tracing::error!( "Cross-PDS issuer mismatch: expected {}, got {}", expected_issuer, diff --git a/crates/tranquil-oauth/Cargo.toml b/crates/tranquil-oauth/Cargo.toml index 5194281..b7b1d5f 100644 --- a/crates/tranquil-oauth/Cargo.toml +++ b/crates/tranquil-oauth/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true [dependencies] tranquil-types = { workspace = true } +tranquil-infra = { workspace = true, features = ["cache-keys"] } anyhow = { workspace = true } sqlx = { workspace = true } diff --git a/crates/tranquil-oauth/src/types.rs b/crates/tranquil-oauth/src/types.rs index 806268f..5f8f01b 100644 --- a/crates/tranquil-oauth/src/types.rs +++ b/crates/tranquil-oauth/src/types.rs @@ -1,7 +1,7 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; -use tranquil_types::{ClientId, Did}; +use tranquil_types::{AuthServerEndpoint, ClientId, Did, Issuer}; pub use tranquil_types::{AuthorizationCode, DeviceId, RefreshToken, RequestId, TokenId}; @@ -195,9 +195,9 @@ pub struct ProtectedResourceMetadata { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AuthorizationServerMetadata { - pub issuer: String, - pub authorization_endpoint: String, - pub token_endpoint: String, + pub issuer: Issuer, + pub authorization_endpoint: AuthServerEndpoint, + pub token_endpoint: AuthServerEndpoint, pub jwks_uri: String, pub registration_endpoint: Option, pub scopes_supported: Option>, @@ -206,7 +206,7 @@ pub struct AuthorizationServerMetadata { pub grant_types_supported: Option>, pub token_endpoint_auth_methods_supported: Option>, pub code_challenge_methods_supported: Option>, - pub pushed_authorization_request_endpoint: Option, + pub pushed_authorization_request_endpoint: Option, pub require_pushed_authorization_requests: Option, pub dpop_signing_alg_values_supported: Option>, pub authorization_response_iss_parameter_supported: Option, diff --git a/crates/tranquil-pds/Cargo.toml b/crates/tranquil-pds/Cargo.toml index 6e90db3..59fb81a 100644 --- a/crates/tranquil-pds/Cargo.toml +++ b/crates/tranquil-pds/Cargo.toml @@ -15,7 +15,7 @@ tranquil-auth = { workspace = true } tranquil-oauth = { workspace = true } tranquil-comms = { workspace = true } tranquil-signal = { workspace = true } -tranquil-db = { workspace = true } +tranquil-db = { workspace = true, features = ["postgres"] } tranquil-db-traits = { workspace = true } tranquil-store = { workspace = true } tranquil-lexicon = { workspace = true, features = ["resolve"] } @@ -86,6 +86,7 @@ frontend = [] native-tls-roots = ["tranquil-oauth/native-tls-roots"] [dev-dependencies] +tranquil-infra = { workspace = true, features = ["testing"] } tempfile = "3" ciborium = { workspace = true } ctor = { workspace = true } diff --git a/crates/tranquil-pds/src/cache/mod.rs b/crates/tranquil-pds/src/cache/mod.rs index 52ddb33..1386b86 100644 --- a/crates/tranquil-pds/src/cache/mod.rs +++ b/crates/tranquil-pds/src/cache/mod.rs @@ -1,4 +1,6 @@ -pub use tranquil_cache::{Cache, CacheError, DistributedRateLimiter, NoOpCache, create_cache}; +pub use tranquil_cache::{ + Cache, CacheError, DistributedRateLimiter, NoOpCache, cached_json, create_cache, +}; #[cfg(feature = "valkey")] pub use tranquil_cache::{RedisRateLimiter, ValkeyCache}; diff --git a/crates/tranquil-pds/src/cache_keys.rs b/crates/tranquil-pds/src/cache_keys.rs index 3a148ce..b2f2840 100644 --- a/crates/tranquil-pds/src/cache_keys.rs +++ b/crates/tranquil-pds/src/cache_keys.rs @@ -1,48 +1 @@ -use crate::types::{CidLink, Did, Handle, Jti}; - -pub fn session_key(did: &Did, jti: &Jti) -> String { - format!("auth:session:{}:{}", did, jti) -} - -pub fn signing_key_key(did: &Did) -> String { - format!("auth:key:{}", did) -} - -pub fn user_status_key(did: &Did) -> String { - format!("auth:status:{}", did) -} - -pub fn handle_key(handle: &Handle) -> String { - format!("handle:{}", handle) -} - -pub fn reauth_key(did: &Did) -> String { - format!("reauth:{}", did) -} - -pub fn plc_doc_key(did: &Did) -> String { - format!("plc:doc:{}", did) -} - -pub fn plc_data_key(did: &Did) -> String { - format!("plc:data:{}", did) -} - -pub fn email_update_key(did: &Did) -> String { - format!("email_update:{}", did) -} - -pub fn scope_ref_key(cid: &CidLink) -> String { - format!("scope_ref:{}", cid) -} - -pub fn auto_verify_sent_key(did: &Did) -> String { - format!("auto_verify_sent:{}", did) -} - -pub fn permission_set_key(nsid: &tranquil_types::Nsid, aud: Option<&str>) -> String { - match aud { - Some(a) => format!("permset:{}:{}", nsid, a), - None => format!("permset:{}", nsid), - } -} +pub use tranquil_cache::cache_keys::*; diff --git a/crates/tranquil-pds/src/delegation/mod.rs b/crates/tranquil-pds/src/delegation/mod.rs index e793f8d..3bf6e5f 100644 --- a/crates/tranquil-pds/src/delegation/mod.rs +++ b/crates/tranquil-pds/src/delegation/mod.rs @@ -13,6 +13,16 @@ pub use tranquil_db_traits::DelegationActionType; use crate::did::DidResolutionError; use crate::state::AppState; use crate::types::{Did, Handle}; +use tranquil_types::did_doc::{PdsEndpointError, extract_handle, extract_pds_endpoint}; +use tranquil_types::{InvalidHttpUrl, PdsUrl}; + +#[derive(Debug, thiserror::Error)] +pub enum IdentityResolutionError { + #[error(transparent)] + DidResolution(#[from] DidResolutionError), + #[error("remote PDS endpoint is unusable: {0}")] + PdsEndpoint(InvalidHttpUrl), +} #[derive(serde::Serialize)] #[serde(rename_all = "camelCase")] @@ -21,14 +31,14 @@ pub struct ResolvedIdentity { #[serde(skip_serializing_if = "Option::is_none")] pub handle: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub pds_url: Option, + pub pds_url: Option, pub is_local: bool, } pub async fn resolve_identity( state: &AppState, did: &Did, -) -> Result { +) -> Result { let is_local = state .repos .user @@ -38,26 +48,23 @@ pub async fn resolve_identity( .flatten() .is_some(); - let did_doc = state.did_resolver.resolve_did(did).await?; + let did_doc = state.did_resolver.fetch_did_document(did).await?; - let pds_url = did_doc.services.iter().find_map(|svc| { - if (svc.id == "#atproto_pds" || svc.id.ends_with("#atproto_pds")) - && svc.service_type == "AtprotoPersonalDataServer" - { - Some(svc.service_endpoint.clone()) - } else { + let pds_url = match (extract_pds_endpoint(&did_doc), is_local) { + (Ok(url), _) => Some(url), + (Err(PdsEndpointError::Missing), _) => None, + (Err(PdsEndpointError::Invalid(e)), true) => { + tracing::debug!(did = %did, error = %e, "local account has an unusable PDS endpoint"); None } - }); - let handle = did_doc - .also_known_as - .iter() - .find_map(|alias| alias.strip_prefix("at://")) - .and_then(|s| Handle::new(s).ok()); + (Err(PdsEndpointError::Invalid(e)), false) => { + return Err(IdentityResolutionError::PdsEndpoint(e)); + } + }; Ok(ResolvedIdentity { did: did.clone(), - handle, + handle: extract_handle(&did_doc), pds_url, is_local, }) diff --git a/crates/tranquil-pds/src/oauth/client.rs b/crates/tranquil-pds/src/oauth/client.rs index 2f37144..fc65eb7 100644 --- a/crates/tranquil-pds/src/oauth/client.rs +++ b/crates/tranquil-pds/src/oauth/client.rs @@ -10,10 +10,12 @@ use tranquil_oauth::{ AuthorizationServerMetadata, ClientMetadata, compute_es256_jkt, compute_pkce_challenge, create_dpop_proof, }; -use tranquil_types::{AuthorizationCode, ClientId, Did}; +use tranquil_types::{AuthorizationCode, ClientId, CrossPdsState, Did, Issuer, PdsUrl}; use crate::cache::Cache; +const SERVER_METADATA_TTL: Duration = Duration::from_secs(300); + #[derive(Error, Debug)] pub enum CrossPdsError { #[error("failed to fetch OAuth metadata: {0}")] @@ -32,11 +34,11 @@ pub enum CrossPdsError { pub struct CrossPdsAuthState { pub original_request_uri: String, pub controller_did: Did, - pub controller_pds_url: String, + pub controller_pds_url: PdsUrl, pub code_verifier: String, pub dpop_private_key_der: String, pub delegated_did: Did, - pub expected_issuer: Option, + pub expected_issuer: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -70,17 +72,23 @@ impl CrossPdsOAuthClient { let http = Client::builder() .timeout(Duration::from_secs(15)) .connect_timeout(Duration::from_secs(5)) + .redirect(tranquil_types::redirect_policy( + tranquil_types::ReachPolicy::GlobalOnly, + )) + .dns_resolver(tranquil_types::dns_guard( + tranquil_types::ReachPolicy::GlobalOnly, + )) .build() - .unwrap_or_else(|_| Client::new()); + .expect("failed to build cross-PDS OAuth HTTP client"); Self { http, cache } } pub async fn store_auth_state( &self, - state_key: &str, + state_key: &CrossPdsState, auth_state: &CrossPdsAuthState, ) -> Result<(), CrossPdsError> { - let cache_key = format!("cross_pds_state:{}", state_key); + let cache_key = crate::cache_keys::cross_pds_state_key(state_key); let json_bytes = serde_json::to_vec(auth_state) .map_err(|e| CrossPdsError::ParFailed(format!("serialize auth state: {}", e)))?; let encrypted = crate::config::encrypt_key(&json_bytes) @@ -93,9 +101,9 @@ impl CrossPdsOAuthClient { pub async fn retrieve_auth_state( &self, - state_key: &str, + state_key: &CrossPdsState, ) -> Result { - let cache_key = format!("cross_pds_state:{}", state_key); + let cache_key = crate::cache_keys::cross_pds_state_key(state_key); let encrypted_bytes = self.cache.get_bytes(&cache_key).await.ok_or_else(|| { CrossPdsError::TokenExchangeFailed("auth state expired or not found".into()) })?; @@ -110,13 +118,11 @@ impl CrossPdsOAuthClient { }) } - pub async fn check_remote_is_delegated(&self, pds_url: &str, did: &Did) -> Option { - let url = format!( - "{}/oauth/security-status?identifier={}", - pds_url.trim_end_matches('/'), - urlencoding::encode(did.as_str()) - ); - let resp = self.http.get(&url).send().await.ok()?; + pub async fn check_remote_is_delegated(&self, pds_url: &PdsUrl, did: &Did) -> Option { + let mut url = pds_url.endpoint("oauth/security-status"); + url.query_pairs_mut() + .append_pair("identifier", did.as_str()); + let resp = self.http.get(url).send().await.ok()?; if !resp.status().is_success() { return None; } @@ -176,24 +182,12 @@ impl CrossPdsOAuthClient { Ok(resp) } - fn require_https(url: &str, label: &str) -> Result<(), CrossPdsError> { - if !url.starts_with("https://") { - return Err(CrossPdsError::MetadataFetch(format!( - "{} must use HTTPS, got: {}", - label, url - ))); - } - Ok(()) - } - - async fn resolve_authorization_server(&self, pds_url: &str) -> Result { - Self::require_https(pds_url, "PDS URL")?; - - let resource_url = format!( - "{}/.well-known/oauth-protected-resource", - pds_url.trim_end_matches('/') - ); - if let Ok(resp) = self.http.get(&resource_url).send().await + async fn resolve_authorization_server( + &self, + pds_url: &PdsUrl, + ) -> Result { + let resource_url = pds_url.endpoint(".well-known/oauth-protected-resource"); + if let Ok(resp) = self.http.get(resource_url).send().await && resp.status().is_success() { #[derive(Deserialize)] @@ -203,30 +197,36 @@ impl CrossPdsOAuthClient { if let Ok(pr) = resp.json::().await && let Some(server) = pr.authorization_servers.and_then(|s| s.into_iter().next()) { - Self::require_https(&server, "Authorization server")?; - return Ok(server); + return Issuer::new(server) + .map_err(|e| CrossPdsError::MetadataFetch(e.to_string())); } } - Ok(pds_url.trim_end_matches('/').to_string()) + Issuer::new(pds_url.as_str()).map_err(|e| CrossPdsError::MetadataFetch(e.to_string())) } pub async fn fetch_server_metadata( &self, - pds_url: &str, + pds_url: &PdsUrl, ) -> Result { - let cache_key = format!("cross_pds_oauth_meta:{}", pds_url); - if let Some(cached) = self.cache.get(&cache_key).await - && let Ok(meta) = serde_json::from_str(&cached) - { - return Ok(meta); - } + crate::cache::cached_json( + self.cache.as_ref(), + &crate::cache_keys::cross_pds_oauth_meta_key(pds_url), + SERVER_METADATA_TTL, + || self.fetch_verified_server_metadata(pds_url), + ) + .await + } + async fn fetch_verified_server_metadata( + &self, + pds_url: &PdsUrl, + ) -> Result { let auth_server = self.resolve_authorization_server(pds_url).await?; - let url = format!("{}/.well-known/oauth-authorization-server", auth_server); + let url = auth_server.endpoint(".well-known/oauth-authorization-server"); let resp = self .http - .get(&url) + .get(url.clone()) .send() .await .map_err(|e| CrossPdsError::MetadataFetch(e.to_string()))?; @@ -244,11 +244,11 @@ impl CrossPdsOAuthClient { .await .map_err(|e| CrossPdsError::MetadataFetch(e.to_string()))?; - if let Ok(json_str) = serde_json::to_string(&meta) { - let _ = self - .cache - .set(&cache_key, &json_str, Duration::from_secs(300)) - .await; + if meta.issuer != auth_server { + return Err(CrossPdsError::MetadataFetch(format!( + "issuer mismatch: {} serves metadata for {}", + auth_server, meta.issuer + ))); } Ok(meta) @@ -256,22 +256,22 @@ impl CrossPdsOAuthClient { pub async fn initiate_par( &self, - pds_url: &str, + pds_url: &PdsUrl, urls: &DelegationOAuthUrls, login_hint: Option<&str>, original_request_uri: &str, controller_did: &Did, delegated_did: &Did, - ) -> Result<(ParResult, CrossPdsAuthState, String), CrossPdsError> { + ) -> Result<(ParResult, CrossPdsAuthState, CrossPdsState), CrossPdsError> { let meta = self.fetch_server_metadata(pds_url).await?; let par_endpoint = meta .pushed_authorization_request_endpoint - .as_deref() + .as_ref() .ok_or(CrossPdsError::NoParEndpoint)?; let code_verifier = crate::util::generate_random_token(); let code_challenge = compute_pkce_challenge(&code_verifier); - let state = crate::util::generate_random_token(); + let state = CrossPdsState::new(crate::util::generate_random_token()); let signing_key = SigningKey::random(&mut OsRng); let dpop_key_der = URL_SAFE_NO_PAD.encode(signing_key.to_bytes()); @@ -284,7 +284,7 @@ impl CrossPdsOAuthClient { ("client_id", urls.client_id.to_string()), ("redirect_uri", urls.redirect_uri.clone()), ("scope", "atproto".to_string()), - ("state", state.clone()), + ("state", state.to_string()), ("code_challenge", code_challenge), ("code_challenge_method", "S256".to_string()), ("dpop_jkt", dpop_jkt), @@ -294,7 +294,7 @@ impl CrossPdsOAuthClient { } let resp = self - .send_with_dpop_retry(&signing_key, "POST", par_endpoint, ¶ms, None) + .send_with_dpop_retry(&signing_key, "POST", par_endpoint.as_str(), ¶ms, None) .await .map_err(|e| CrossPdsError::ParFailed(e.to_string()))?; @@ -313,17 +313,16 @@ impl CrossPdsOAuthClient { .await .map_err(|e| CrossPdsError::ParFailed(e.to_string()))?; - let authorize_url = format!( - "{}?request_uri={}&client_id={}", - meta.authorization_endpoint, - urlencoding::encode(&par_resp.request_uri), - urlencoding::encode(&urls.client_id) - ); + let mut authorize_url = meta.authorization_endpoint.url().clone(); + authorize_url + .query_pairs_mut() + .append_pair("request_uri", &par_resp.request_uri) + .append_pair("client_id", &urls.client_id); let auth_state = CrossPdsAuthState { original_request_uri: original_request_uri.to_string(), controller_did: controller_did.clone(), - controller_pds_url: pds_url.to_string(), + controller_pds_url: pds_url.clone(), code_verifier, dpop_private_key_der: dpop_key_der, delegated_did: delegated_did.clone(), @@ -333,7 +332,7 @@ impl CrossPdsOAuthClient { Ok(( ParResult { request_uri: par_resp.request_uri, - authorize_url, + authorize_url: authorize_url.into(), }, auth_state, state, @@ -366,7 +365,13 @@ impl CrossPdsOAuthClient { ]; let resp = self - .send_with_dpop_retry(&signing_key, "POST", &meta.token_endpoint, ¶ms, None) + .send_with_dpop_retry( + &signing_key, + "POST", + meta.token_endpoint.as_str(), + ¶ms, + None, + ) .await .map_err(CrossPdsError::TokenExchangeFailed)?; diff --git a/crates/tranquil-signal/Cargo.toml b/crates/tranquil-signal/Cargo.toml index ed8d49a..3db515e 100644 --- a/crates/tranquil-signal/Cargo.toml +++ b/crates/tranquil-signal/Cargo.toml @@ -18,7 +18,7 @@ tokio = { workspace = true } tokio-util = { workspace = true } futures = { workspace = true } serde_json = { workspace = true } -url = "2.5" +url = { workspace = true } uuid = { workspace = true } thiserror = { workspace = true } diff --git a/crates/tranquil-types/Cargo.toml b/crates/tranquil-types/Cargo.toml index 4055d2b..f0c808f 100644 --- a/crates/tranquil-types/Cargo.toml +++ b/crates/tranquil-types/Cargo.toml @@ -10,8 +10,15 @@ chrono = { workspace = true } cid = { workspace = true } jacquard-common = { workspace = true } rand = { workspace = true } +reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } sqlx = { workspace = true } thiserror = { workspace = true } +tokio = { workspace = true, features = ["net", "rt"] } +tracing = { workspace = true } +url = { workspace = true } uuid = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true } diff --git a/crates/tranquil-types/src/lib.rs b/crates/tranquil-types/src/lib.rs index 039fdcd..6fe70a8 100644 --- a/crates/tranquil-types/src/lib.rs +++ b/crates/tranquil-types/src/lib.rs @@ -1,6 +1,8 @@ use serde::{Deserialize, Serialize}; use std::borrow::Cow; use std::fmt; +use std::hash::Hash; +use std::marker::PhantomData; use std::ops::Deref; use std::str::FromStr; @@ -813,6 +815,10 @@ simple_string_newtype! { pub struct Jti; } +simple_string_newtype_no_sqlx! { + pub struct CrossPdsState; +} + simple_string_newtype! { pub struct AuthorizationCode; } @@ -881,6 +887,425 @@ impl fmt::Display for CommsChannel { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HostReach { + Global, + Loopback, + Private, +} + +fn ipv4_reach(ip: std::net::Ipv4Addr) -> HostReach { + let [a, b, c, _] = ip.octets(); + match ip { + _ if ip.is_loopback() => HostReach::Loopback, + _ if ip.is_private() + || ip.is_link_local() + || ip.is_multicast() + || ip.is_documentation() + || a == 0 + || a == 100 && (64..128).contains(&b) + || a == 192 && b == 0 && c == 0 + || a == 192 && b == 88 && c == 99 + || a == 198 && (18..20).contains(&b) + || a & 0xf0 == 240 => + { + HostReach::Private + } + _ => HostReach::Global, + } +} + +fn ipv6_reach(ip: std::net::Ipv6Addr) -> HostReach { + let seg = ip.segments(); + let embedded_ipv4 = + |hi: u16, lo: u16| std::net::Ipv4Addr::from((u32::from(hi) << 16) | u32::from(lo)); + match ip.to_ipv4_mapped() { + Some(mapped) => ipv4_reach(mapped), + None => match ip { + _ if ip.is_loopback() => HostReach::Loopback, + _ if seg[..6] == [0, 0, 0, 0, 0, 0] => ipv4_reach(embedded_ipv4(seg[6], seg[7])), + _ if seg[..2] == [0x2001, 0] => ipv4_reach(embedded_ipv4(!seg[6], !seg[7])), + _ if seg[0] == 0x2002 => ipv4_reach(embedded_ipv4(seg[1], seg[2])), + _ if seg[..6] == [0x64, 0xff9b, 0, 0, 0, 0] => { + ipv4_reach(embedded_ipv4(seg[6], seg[7])) + } + _ if seg[..3] == [0x64, 0xff9b, 1] => HostReach::Private, + _ if ip.is_unspecified() + || ip.is_multicast() + || seg[0] & 0xfe00 == 0xfc00 + || seg[0] & 0xffc0 == 0xfe80 + || seg[..2] == [0x2001, 0x0db8] => + { + HostReach::Private + } + _ => HostReach::Global, + }, + } +} + +fn host_reach(host: url::Host<&str>) -> HostReach { + match host { + url::Host::Ipv4(ip) => ipv4_reach(ip), + url::Host::Ipv6(ip) => ipv6_reach(ip), + url::Host::Domain(name) => { + let name = name.trim_end_matches('.').to_ascii_lowercase(); + match name.as_str() { + "localhost" => HostReach::Loopback, + _ if name.ends_with(".localhost") => HostReach::Loopback, + _ if name.ends_with(".local") + || name.ends_with(".internal") + || name.ends_with(".home.arpa") + || name == "home.arpa" => + { + HostReach::Private + } + _ => HostReach::Global, + } + } + } +} + +pub fn url_reach(url: &url::Url) -> Option { + url.host().map(host_reach) +} + +pub fn ip_reach(ip: std::net::IpAddr) -> HostReach { + match ip { + std::net::IpAddr::V4(v4) => ipv4_reach(v4), + std::net::IpAddr::V6(v6) => ipv6_reach(v6), + } +} + +pub fn reach_permits(reach: HostReach, policy: ReachPolicy) -> bool { + matches!( + (reach, policy), + (HostReach::Global, _) + | ( + HostReach::Loopback, + ReachPolicy::AllowLoopback | ReachPolicy::AllowPrivate, + ) + | (HostReach::Private, ReachPolicy::AllowPrivate) + ) +} + +pub fn url_reach_permits(url: &url::Url, policy: ReachPolicy) -> bool { + let Some(reach) = url_reach(url) else { + return false; + }; + let scheme_permits = matches!( + (url.scheme(), reach), + ("https", _) | ("http", HostReach::Loopback | HostReach::Private) + ); + scheme_permits && reach_permits(reach, policy) +} + +fn parse_http_url(s: &str, policy: ReachPolicy, allow_query: bool) -> Option { + let parsed = url::Url::parse(s).ok()?; + let rejected = (parsed.query().is_some() && !allow_query) + || parsed.fragment().is_some() + || !parsed.username().is_empty() + || parsed.password().is_some() + || !url_reach_permits(&parsed, policy); + match rejected { + true => None, + false => Some(parsed), + } +} + +const REDIRECT_HOP_LIMIT: usize = 5; + +pub fn redirect_policy(policy: ReachPolicy) -> reqwest::redirect::Policy { + reqwest::redirect::Policy::custom(move |attempt| { + let over_limit = attempt.previous().len() > REDIRECT_HOP_LIMIT; + let permitted = url_reach_permits(attempt.url(), policy); + let target = attempt.url().clone(); + match (over_limit, permitted) { + (true, _) => attempt.error(format!("more than {} redirect hops", REDIRECT_HOP_LIMIT)), + (false, false) => attempt.error(format!( + "redirect target {} is outside the allowed host reach", + target + )), + (false, true) => attempt.follow(), + } + }) +} + +pub struct ReachGuardedDns(ReachPolicy); + +impl reqwest::dns::Resolve for ReachGuardedDns { + fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving { + let policy = self.0; + Box::pin(async move { + let host = name.as_str().to_owned(); + let permitted: Vec = tokio::net::lookup_host((host.as_str(), 0)) + .await? + .filter(|addr| reach_permits(ip_reach(addr.ip()), policy)) + .collect(); + match permitted.is_empty() { + true => Err(format!( + "no resolved address for {} is inside the allowed host reach", + host + ) + .into()), + false => Ok(Box::new(permitted.into_iter()) as reqwest::dns::Addrs), + } + }) + } +} + +pub fn dns_guard(policy: ReachPolicy) -> std::sync::Arc { + std::sync::Arc::new(ReachGuardedDns(policy)) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReachPolicy { + AllowLoopback, + AllowPrivate, + GlobalOnly, +} + +impl ReachPolicy { + #[cfg(debug_assertions)] + pub const DEBUG_LOOPBACK: ReachPolicy = ReachPolicy::AllowLoopback; + #[cfg(not(debug_assertions))] + pub const DEBUG_LOOPBACK: ReachPolicy = ReachPolicy::GlobalOnly; +} + +pub trait UrlKind { + const LABEL: &'static str; + const REACH_POLICY: ReachPolicy; + const ALLOW_QUERY: bool; +} + +pub mod url_kind { + use super::{ReachPolicy, UrlKind}; + + pub struct AuthServerEndpoint; + impl UrlKind for AuthServerEndpoint { + const LABEL: &'static str = "authorization server endpoint"; + const REACH_POLICY: ReachPolicy = ReachPolicy::GlobalOnly; + const ALLOW_QUERY: bool = true; + } + + pub struct Issuer; + impl UrlKind for Issuer { + const LABEL: &'static str = "issuer"; + const REACH_POLICY: ReachPolicy = ReachPolicy::GlobalOnly; + const ALLOW_QUERY: bool = false; + } + + pub struct Jwks; + impl UrlKind for Jwks { + const LABEL: &'static str = "JWKS URI"; + const REACH_POLICY: ReachPolicy = ReachPolicy::DEBUG_LOOPBACK; + const ALLOW_QUERY: bool = true; + } + + pub struct Pds; + impl UrlKind for Pds { + const LABEL: &'static str = "PDS URL"; + const REACH_POLICY: ReachPolicy = ReachPolicy::GlobalOnly; + const ALLOW_QUERY: bool = false; + } + + pub struct SchemaHost; + impl UrlKind for SchemaHost { + const LABEL: &'static str = "schema host URL"; + const REACH_POLICY: ReachPolicy = ReachPolicy::DEBUG_LOOPBACK; + const ALLOW_QUERY: bool = false; + } + + pub struct SsoIssuer; + impl UrlKind for SsoIssuer { + const LABEL: &'static str = "SSO issuer"; + const REACH_POLICY: ReachPolicy = ReachPolicy::AllowPrivate; + const ALLOW_QUERY: bool = false; + } + + pub struct SsoJwks; + impl UrlKind for SsoJwks { + const LABEL: &'static str = "SSO JWKS URI"; + const REACH_POLICY: ReachPolicy = ReachPolicy::AllowPrivate; + const ALLOW_QUERY: bool = true; + } +} + +pub struct HttpUrl { + raw: String, + parsed: url::Url, + kind: PhantomData K>, +} + +pub type AuthServerEndpoint = HttpUrl; +pub type Issuer = HttpUrl; +pub type JwksUri = HttpUrl; +pub type PdsUrl = HttpUrl; +pub type SchemaHostUrl = HttpUrl; +pub type SsoIssuer = HttpUrl; +pub type SsoJwksUri = HttpUrl; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct InvalidHttpUrl { + pub kind: &'static str, + pub value: String, +} + +impl fmt::Display for InvalidHttpUrl { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "invalid {}: {}", self.kind, self.value) + } +} + +impl std::error::Error for InvalidHttpUrl {} + +impl HttpUrl { + pub fn new(s: impl Into) -> Result { + let raw = s.into(); + match parse_http_url(&raw, K::REACH_POLICY, K::ALLOW_QUERY) { + Some(parsed) => Ok(Self { + raw, + parsed, + kind: PhantomData, + }), + None => Err(InvalidHttpUrl { + kind: K::LABEL, + value: raw, + }), + } + } + + /// The URL as given. + /// OIDC & OAuth define issuer comparison as an + /// exact string match, + /// so anything sent to or compared against a peer uses this. + /// Give it to us raw & wriggling!! + pub fn as_str(&self) -> &str { + &self.raw + } + + /// The parsed form: lowercased scheme and host, with `/` for a bare authority. + /// Cache keys use this so `https://oyster.cafe` and `https://oyster.cafe/` share one entry. + pub fn canonical(&self) -> &str { + self.parsed.as_str() + } + + pub fn url(&self) -> &url::Url { + &self.parsed + } + + pub fn endpoint(&self, path: &str) -> url::Url { + let mut url = self.parsed.clone(); + let base = url.path().trim_end_matches('/').to_owned(); + url.set_path(&format!("{}/{}", base, path.trim_start_matches('/'))); + url + } +} + +pub mod http_url { + use super::{HttpUrl, UrlKind}; + use serde::Deserialize; + + pub fn deserialize_optional<'de, D, K>(deserializer: D) -> Result>, D::Error> + where + D: serde::Deserializer<'de>, + K: UrlKind, + { + Ok(Option::::deserialize(deserializer)?.and_then(|s| { + HttpUrl::new(s) + .inspect_err(|e| tracing::warn!(error = %e, "discarding unusable URL field")) + .ok() + })) + } +} + +impl FromStr for HttpUrl { + type Err = InvalidHttpUrl; + + fn from_str(s: &str) -> Result { + Self::new(s) + } +} + +impl fmt::Debug for HttpUrl { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}({})", K::LABEL, self.raw) + } +} + +impl fmt::Display for HttpUrl { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.raw) + } +} + +impl Clone for HttpUrl { + fn clone(&self) -> Self { + Self { + raw: self.raw.clone(), + parsed: self.parsed.clone(), + kind: PhantomData, + } + } +} + +impl PartialEq for HttpUrl { + fn eq(&self, other: &Self) -> bool { + self.parsed == other.parsed + } +} + +impl Eq for HttpUrl {} + +impl Hash for HttpUrl { + fn hash(&self, state: &mut H) { + self.parsed.hash(state); + } +} + +impl Serialize for HttpUrl { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(&self.raw) + } +} + +impl<'de, K: UrlKind> Deserialize<'de> for HttpUrl { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + Self::new(s).map_err(|e| serde::de::Error::custom(e.to_string())) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EmailTokenPurpose { + UpdateEmail, + ConfirmEmail, + DeleteAccount, + ResetPassword, + PlcOperation, +} + +impl EmailTokenPurpose { + pub fn as_str(&self) -> &'static str { + match self { + Self::UpdateEmail => "update_email", + Self::ConfirmEmail => "confirm_email", + Self::DeleteAccount => "delete_account", + Self::ResetPassword => "reset_password", + Self::PlcOperation => "plc_operation", + } + } +} + +impl fmt::Display for EmailTokenPurpose { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, sqlx::Type)] #[serde(rename_all = "snake_case")] #[sqlx(type_name = "comms_type", rename_all = "snake_case")] @@ -894,24 +1319,32 @@ pub enum CommsType { } pub mod did_doc { - pub fn extract_pds_endpoint(doc: &serde_json::Value) -> Option { + use crate::{HttpUrl, InvalidHttpUrl, UrlKind}; + + #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] + pub enum PdsEndpointError { + #[error("DID document has no atproto PDS service entry")] + Missing, + #[error(transparent)] + Invalid(#[from] InvalidHttpUrl), + } + + pub fn extract_pds_endpoint( + doc: &serde_json::Value, + ) -> Result, PdsEndpointError> { doc.get("service") .and_then(|s| s.as_array()) .and_then(|services| { services.iter().find_map(|svc| { let id = svc.get("id").and_then(|v| v.as_str()).unwrap_or_default(); let svc_type = svc.get("type").and_then(|v| v.as_str()).unwrap_or_default(); - if (id == "#atproto_pds" || id.ends_with("#atproto_pds")) - && svc_type == "AtprotoPersonalDataServer" - { - svc.get("serviceEndpoint") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()) - } else { - None - } + ((id == "#atproto_pds" || id.ends_with("#atproto_pds")) + && svc_type == "AtprotoPersonalDataServer") + .then(|| svc.get("serviceEndpoint").and_then(|v| v.as_str()))? }) }) + .ok_or(PdsEndpointError::Missing) + .and_then(|endpoint| HttpUrl::new(endpoint).map_err(PdsEndpointError::Invalid)) } pub fn extract_handle(doc: &serde_json::Value) -> Option { @@ -928,6 +1361,187 @@ pub mod did_doc { } } +#[cfg(test)] +mod http_url_tests { + use super::did_doc::{PdsEndpointError, extract_pds_endpoint}; + use super::{ + AuthServerEndpoint, Issuer, JwksUri, PdsUrl, SchemaHostUrl, SsoIssuer, SsoJwksUri, + }; + + #[test] + fn extract_pds_endpoint_selects_the_pds_service_and_reports_missing_or_invalid() { + let labeler = serde_json::json!({ + "id": "#atproto_labeler", + "type": "AtprotoLabeler", + "serviceEndpoint": "https://labeler.nel.pet" + }); + let pds = |endpoint: &str| { + serde_json::json!({ + "id": "#atproto_pds", + "type": "AtprotoPersonalDataServer", + "serviceEndpoint": endpoint + }) + }; + let both = serde_json::json!({ "service": [labeler.clone(), pds("https://oyster.cafe")] }); + assert_eq!( + extract_pds_endpoint::(&both) + .unwrap() + .as_str(), + "https://oyster.cafe" + ); + [ + serde_json::json!({ "service": [labeler] }), + serde_json::json!({}), + ] + .iter() + .for_each(|doc| { + assert_eq!( + extract_pds_endpoint::(doc).unwrap_err(), + PdsEndpointError::Missing + ); + }); + let plain_http = serde_json::json!({ "service": [pds("http://oyster.cafe")] }); + assert!(matches!( + extract_pds_endpoint::(&plain_http), + Err(PdsEndpointError::Invalid(_)) + )); + } + + #[test] + fn pds_and_jwks_kinds_reject_private_and_reserved_addresses() { + [ + "https://10.0.0.1", + "https://192.168.1.1", + "https://172.16.0.1", + "https://169.254.169.254/latest/meta-data", + "https://100.64.0.1", + "https://0.1.2.3", + "https://192.0.0.8", + "https://192.88.99.1", + "https://[fd00::1]", + "https://[fe80::1]", + "https://[ff02::1]", + "https://[::ffff:10.0.0.1]", + "https://[64:ff9b::a00:1]", + "https://[64:ff9b:1::1]", + "https://[2002:a00:1::]", + "https://[::10.0.0.1]", + "https://[2001:0:0:0:0:0:f5ff:fffe]", + "https://kelp.internal", + "https://whelk.local", + "https://limpet.home.arpa", + ] + .iter() + .for_each(|url| { + assert!(PdsUrl::new(*url).is_err(), "PdsUrl must reject {url}"); + assert!(JwksUri::new(*url).is_err(), "JwksUri must reject {url}"); + }); + [ + "https://oyster.cafe", + "https://[64:ff9b::808:808]", + "https://[::8.8.8.8]", + "https://[2001::f7f7:f7f7]", + ] + .iter() + .for_each(|url| assert!(PdsUrl::new(*url).is_ok(), "PdsUrl must accept {url}")); + } + + #[test] + fn each_kind_applies_its_own_local_host_policy() { + assert!(PdsUrl::new("http://127.0.0.1:2583").is_err()); + assert!(PdsUrl::new("https://localhost").is_err()); + assert!(Issuer::new("http://localhost:8080").is_err()); + assert_eq!( + JwksUri::new("http://localhost:8080/keys").is_ok(), + cfg!(debug_assertions) + ); + assert_eq!( + SchemaHostUrl::new("http://127.0.0.1:2583").is_ok(), + cfg!(debug_assertions) + ); + assert!(SsoJwksUri::new("http://127.0.0.1:8080/keys").is_ok()); + assert!(SsoJwksUri::new("http://[::1]:8080/keys").is_ok()); + assert!(SsoJwksUri::new("http://squid.localhost:8080/keys").is_ok()); + assert!(SsoJwksUri::new("https://keycloak.internal/keys?client=squid").is_ok()); + assert!(SsoJwksUri::new("http://oyster.cafe/keys").is_err()); + assert!(SsoIssuer::new("https://keycloak.internal/realms/uni").is_ok()); + assert!(SsoIssuer::new("http://10.0.0.5:8080").is_ok()); + assert!(SsoIssuer::new("http://localhost:8080").is_ok()); + assert!(SsoIssuer::new("http://oyster.cafe").is_err()); + assert!(AuthServerEndpoint::new("https://oyster.cafe/oauth/par?tenant=uni").is_ok()); + [ + "https://169.254.169.254/oauth/par", + "https://[fd00::1]/oauth/par", + "http://127.0.0.1:2583/oauth/par", + "http://oyster.cafe/oauth/par", + ] + .iter() + .for_each(|url| { + assert!( + AuthServerEndpoint::new(*url).is_err(), + "AuthServerEndpoint must reject {url}" + ); + }); + } + + #[test] + fn canonicalization_keeps_identity_and_rejects_query_fragment_and_userinfo() { + assert_eq!( + PdsUrl::new("HTTPS://oyster.cafe").unwrap().canonical(), + "https://oyster.cafe/" + ); + let bare = PdsUrl::new("https://oyster.cafe").unwrap(); + let slashed = PdsUrl::new("https://oyster.cafe/").unwrap(); + assert_eq!(bare, slashed); + assert_eq!(bare.canonical(), slashed.canonical()); + let issuer = Issuer::new("https://accounts.google.com").unwrap(); + assert_eq!(issuer.as_str(), "https://accounts.google.com"); + assert_eq!(issuer.canonical(), "https://accounts.google.com/"); + assert_eq!( + PdsUrl::new("https://oyster.cafe/pds/") + .unwrap() + .endpoint(".well-known/oauth-protected-resource") + .as_str(), + "https://oyster.cafe/pds/.well-known/oauth-protected-resource" + ); + assert_eq!( + JwksUri::new("https://oyster.cafe/keys?appid=abc") + .expect("JwksUri keeps the query") + .canonical(), + "https://oyster.cafe/keys?appid=abc" + ); + assert!(PdsUrl::new("https://oyster.cafe/?x=1").is_err()); + assert!(PdsUrl::new("https://oyster.cafe/#frag").is_err()); + assert!(PdsUrl::new("https://nel:pw@oyster.cafe").is_err()); + assert!(Issuer::new("https://oyster.cafe/?x=1").is_err()); + assert!(JwksUri::new("https://oyster.cafe/keys#frag").is_err()); + } +} + +#[cfg(test)] +mod dns_guard_tests { + use super::{ReachPolicy, dns_guard}; + use reqwest::dns::Resolve; + + #[tokio::test] + async fn the_policy_gates_loopback_resolution() { + let name = |host: &str| host.parse::().expect("valid hostname"); + assert!( + dns_guard(ReachPolicy::GlobalOnly) + .resolve(name("localhost")) + .await + .is_err() + ); + let addrs: Vec<_> = dns_guard(ReachPolicy::AllowLoopback) + .resolve(name("localhost")) + .await + .expect("localhost resolves") + .collect(); + assert!(!addrs.is_empty()); + assert!(addrs.iter().all(|a| a.ip().is_loopback())); + } +} + #[cfg(test)] mod validated_newtype_tests { use super::*;