diff --git a/Cargo.lock b/Cargo.lock --- a/Cargo.lock +++ b/Cargo.lock @@ -488,6 +488,7 @@ checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", + "axum-macros", "base64", "bytes", "form_urlencoded", @@ -557,6 +558,17 @@ "tower-layer", "tower-service", "tracing", +] + +[[package]] +name = "axum-macros" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", ] [[package]] @@ -984,6 +996,7 @@ version = "0.0.1" dependencies = [ "axum", + "base64", "bobbin-edge-index", "bobbin-ingest", "bobbin-knot-proxy", @@ -1000,6 +1013,7 @@ "jacquard-common", "jacquard-identity", "reqwest 0.13.1", + "scc", "serde", "serde_json", "thiserror 2.0.18", diff --git a/Cargo.toml b/Cargo.toml --- a/Cargo.toml +++ b/Cargo.toml @@ -121,7 +121,7 @@ arc-swap = "1" reqwest = { version = "0.13", default-features = false, features = ["rustls", "webpki-roots", "http2", "json", "gzip", "stream"] } -axum = "0.8" +axum = { version = "0.8", features = ["macros"] } hyper = { version = "1", features = ["server", "http1", "http2"] } hyper-util = { version = "0.1", features = ["server", "server-auto", "tokio", "service"] } tower = { version = "0.5", features = ["util", "limit", "load-shed"] } diff --git a/docker-compose.yml b/docker-compose.yml --- a/docker-compose.yml +++ b/docker-compose.yml @@ -453,6 +453,7 @@ BOBBIN_BIND: 0.0.0.0:8090 BOBBIN_HYDRANT_URL: http://hydrant:3000 BOBBIN_SLINGSHOT_URL: http://hydrant:3000 + BOBBIN_SERVICE_DID: did:web:bobbin.tngl.boltless.dev BOBBIN_KNOT_ALLOW_PRIVATE: "true" BOBBIN_KNOT_REQUIRE_HTTPS: "false" BOBBIN_MIRROR_V2_URL: http://knotmirror:7000 diff --git a/gitmirror/src/git.rs b/gitmirror/src/git.rs --- a/gitmirror/src/git.rs +++ b/gitmirror/src/git.rs @@ -39,7 +39,7 @@ let mut object_dirs = Vec::new(); for &did in dids { if seen.contains(&did) { - continue + continue; } // A DID has no `/`, so this is also what keeps `repo_base.join(did)` inside `repo_base`. validate_did(did) diff --git a/gitmirror/src/merge.rs b/gitmirror/src/merge.rs --- a/gitmirror/src/merge.rs +++ b/gitmirror/src/merge.rs @@ -168,7 +168,9 @@ /// Mirror of `crate::git::open_scratch`: a throwaway repo whose alternates point at the /// given object dirs, opened in-memory. - fn scratch_with_alternates(object_dirs: &[std::path::PathBuf]) -> (tempfile::TempDir, gix::Repository) { + fn scratch_with_alternates( + object_dirs: &[std::path::PathBuf], + ) -> (tempfile::TempDir, gix::Repository) { let scratch = tempfile::tempdir().unwrap(); gix::init_bare(scratch.path()).unwrap(); let info = scratch.path().join("objects").join("info"); diff --git a/gitmirror/src/xrpc.rs b/gitmirror/src/xrpc.rs --- a/gitmirror/src/xrpc.rs +++ b/gitmirror/src/xrpc.rs @@ -11,8 +11,8 @@ use bobbin_types::sh_tangled::git::temp2::{get_diff, get_interdiff, list_commits, merge_check}; use gix::ObjectId; use jacquard_axum::{ExtractXrpc, XrpcResponse}; -use jacquard_common::types::string::Datetime; use jacquard_common::ToSmolStr; +use jacquard_common::types::string::Datetime; use serde_json::json; use tracing::{error, info}; @@ -145,7 +145,9 @@ size: f.size as i64, is_binary: f.is_binary, is_submodule: f.is_submodule, - content: f.content.map(|b| String::from_utf8_lossy(&b).into_owned().into()), + content: f + .content + .map(|b| String::from_utf8_lossy(&b).into_owned().into()), extra_data: Default::default(), } } @@ -199,9 +201,10 @@ fn find_commit(repo: &gix::Repository, sha: &str) -> Result { let oid = gix::ObjectId::from_hex(sha.as_bytes()) .map_err(|e| XrpcError::InvalidRequest(format!("bad commit sha {sha:?}: {e}")))?; - repo.find_commit(oid).map_err(|_| XrpcError::RevisionNotFound { - rev: sha.to_owned(), - })?; + repo.find_commit(oid) + .map_err(|_| XrpcError::RevisionNotFound { + rev: sha.to_owned(), + })?; Ok(oid) } @@ -286,10 +289,14 @@ State(state): State, ExtractXrpc(args): ExtractXrpc, ) -> Result, XrpcError> { - let from_base_id = ObjectId::from_hex(args.base_commit1.as_bytes()).map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; - let from_head_id = ObjectId::from_hex(args.base_commit2.as_bytes()).map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; - let to_base_id = ObjectId::from_hex(args.head_commit1.as_bytes()).map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; - let to_head_id = ObjectId::from_hex(args.head_commit2.as_bytes()).map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; + let from_base_id = ObjectId::from_hex(args.base_commit1.as_bytes()) + .map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; + let from_head_id = ObjectId::from_hex(args.base_commit2.as_bytes()) + .map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; + let to_base_id = ObjectId::from_hex(args.head_commit1.as_bytes()) + .map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; + let to_head_id = ObjectId::from_hex(args.head_commit2.as_bytes()) + .map_err(|e| XrpcError::InvalidRequest(e.to_string()))?; let scratch = open_scratch( &state.repo_base, @@ -297,7 +304,11 @@ )?; tokio::task::spawn_blocking(move || { - get_interdiff_inner(&scratch, (from_base_id, from_head_id), (to_base_id, to_head_id)) + get_interdiff_inner( + &scratch, + (from_base_id, from_head_id), + (to_base_id, to_head_id), + ) }) .await .map_err(|e| XrpcError::Internal(e.to_string()))? @@ -327,15 +338,17 @@ } })?; - let limit = args.limit.map(|limit| limit as u32).unwrap_or(DEFAULT_LIMIT); + let limit = args + .limit + .map(|limit| limit as u32) + .unwrap_or(DEFAULT_LIMIT); if limit == 0 || limit > MAX_LIMIT { return Err(XrpcError::InvalidRequest(format!( "limit must be between 1 and {MAX_LIMIT}" ))); } - let mut commits: Vec = - Vec::with_capacity(limit as usize); + let mut commits: Vec = Vec::with_capacity(limit as usize); let mut skipped = 0usize; for info in walk { @@ -367,26 +380,29 @@ ) -> Result, XrpcError> { let path = state.repo_base.join(args.repo.as_str()); let repo = gix::open(path) - .map_err(|e| XrpcError::RepoNotFound{ detail: e.to_string() })? + .map_err(|e| XrpcError::RepoNotFound { + detail: e.to_string(), + })? .into_sync(); - tokio::task::spawn_blocking(move || { - list_commits_inner(&repo.to_thread_local(), args) - }) - .await - .map_err(|e| XrpcError::Internal(e.to_string()))? - .map(|commits| { - XrpcResponse(list_commits::ListCommitsOutput { - commits, - extra_data: Default::default(), + tokio::task::spawn_blocking(move || list_commits_inner(&repo.to_thread_local(), args)) + .await + .map_err(|e| XrpcError::Internal(e.to_string()))? + .map(|commits| { + XrpcResponse(list_commits::ListCommitsOutput { + commits, + extra_data: Default::default(), + }) }) - }) } pub async fn serve(addr: SocketAddr, repo_base: PathBuf) -> anyhow::Result<()> { let app = Router::new() .route("/xrpc/sh.tangled.git.temp2.getDiff", get(get_diff)) - .route("/xrpc/sh.tangled.git.temp2.getInterdiff", get(get_interdiff)) + .route( + "/xrpc/sh.tangled.git.temp2.getInterdiff", + get(get_interdiff), + ) .route("/xrpc/sh.tangled.git.temp2.listCommits", get(list_commits)) .route("/xrpc/sh.tangled.git.temp2.mergeCheck", get(merge_check)) .with_state(XrpcState { @@ -449,7 +465,12 @@ let fork = root.path().join("fork"); git( root.path(), - &["clone", "-q", upstream.to_str().unwrap(), fork.to_str().unwrap()], + &[ + "clone", + "-q", + upstream.to_str().unwrap(), + fork.to_str().unwrap(), + ], ); std::fs::write(upstream.join("upstream.txt"), "theirs\n").unwrap(); diff --git a/bobbin/crates/xrpc/Cargo.toml b/bobbin/crates/xrpc/Cargo.toml --- a/bobbin/crates/xrpc/Cargo.toml +++ b/bobbin/crates/xrpc/Cargo.toml @@ -22,6 +22,7 @@ chrono = { workspace = true } futures = { workspace = true } http = { workspace = true } +scc = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } @@ -36,6 +37,7 @@ [dev-dependencies] bobbin-ingest = { workspace = true } tokio-util = { workspace = true } +base64 = { workspace = true } bobbin-runtime = { workspace = true } http = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } diff --git a/bobbin/crates/bobbin/src/config.rs b/bobbin/crates/bobbin/src/config.rs --- a/bobbin/crates/bobbin/src/config.rs +++ b/bobbin/crates/bobbin/src/config.rs @@ -5,6 +5,7 @@ use anyhow::{Context, anyhow}; use confique::Config; +use jacquard_common::types::did::Did; use trusted_proxies::{ProxyNetError, TrustedProxies}; use url::Url; @@ -25,6 +26,7 @@ "backpressure.tighten_above_ratio", "backpressure.reserved_index_bytes", "slingshot.url", + "service_auth.did", "record_cache.lru_bytes", "search.heap_bytes", "knot.allow_private", @@ -50,6 +52,7 @@ "BOBBIN_BACKPRESSURE_TIGHTEN_ABOVE_RATIO", "BOBBIN_BACKPRESSURE_RESERVED_INDEX_BYTES", "BOBBIN_SLINGSHOT_URL", + "BOBBIN_SERVICE_DID", "BOBBIN_RECORD_LRU_BYTES", "BOBBIN_SEARCH_HEAP_BYTES", "BOBBIN_KNOT_ALLOW_PRIVATE", @@ -76,6 +79,9 @@ #[config(nested)] pub slingshot: SlingshotConfig, + + #[config(nested)] + pub service_auth: ServiceAuthConfig, #[config(nested)] pub record_cache: RecordCacheConfig, @@ -231,6 +237,12 @@ /// Base URL of a slingshot instance. Used for record bodies and identity. #[config(env = "BOBBIN_SLINGSHOT_URL", default = "http://127.0.0.1:13011")] pub url: Url, +} + +#[derive(Debug, Config)] +pub struct ServiceAuthConfig { + #[config(env = "BOBBIN_SERVICE_DID")] + pub did: Did, } #[derive(Debug, Config)] diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -258,9 +258,9 @@ mirror_v2 = %m.host().url(), "we will forward few sh.tangled.git.* to the v2 knotmirror", ), - None => tracing::info!( - "some sh.tangled.git.* methods will fail, since mirror_v2.url is unset", - ), + None => { + tracing::info!("some sh.tangled.git.* methods will fail, since mirror_v2.url is unset",) + } } let search_heap = usize::try_from(search_heap_cap) .with_context(|| format!("search heap {search_heap_cap} exceeds usize"))?; @@ -391,6 +391,7 @@ .with_mirror(mirror) .with_mirror_v2(mirror_v2) .with_proxies(trusted_proxies) + .with_service_did(cfg.service_auth.did.clone()) .with_settlements(settlements); let app = router(state); diff --git a/bobbin/crates/xrpc/src/feed.rs b/bobbin/crates/xrpc/src/feed.rs --- a/bobbin/crates/xrpc/src/feed.rs +++ b/bobbin/crates/xrpc/src/feed.rs @@ -12,9 +12,10 @@ use bobbin_types::sh_tangled::feed::star::{Star, StarRecord, StarSubject}; use bobbin_types::sh_tangled::graph::follow::{Follow, FollowRecord}; use bobbin_types::sh_tangled::repo::{self, Repo, RepoRecord, RepoViewBasic}; -use jacquard_common::DefaultStr; +use jacquard_axum::service_auth::ExtractOptionalServiceAuth; use jacquard_common::types::string::{AtUri, Datetime, Did, Handle, UriValue}; use jacquard_common::xrpc::XrpcResp; +use jacquard_common::{DefaultStr, IntoStatic as _}; use crate::{ AppState, XrpcError, XrpcQuery, fetch, json_stream, paged_tail, parse_cursor, parse_limit, @@ -31,29 +32,30 @@ #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct GetTimelineQuery { - viewer: Option>, #[serde(default)] following_only: bool, limit: Option, cursor: Option, } +#[axum::debug_handler] pub(crate) async fn get_timeline( State(state): State, + ExtractOptionalServiceAuth(auth): ExtractOptionalServiceAuth, XrpcQuery(q): XrpcQuery, ) -> Result { let limit = parse_limit(q.limit)?; let cursor = parse_cursor(q.cursor.as_deref())?; let permit = state.heavy_permit()?; - let viewer = q.viewer.as_ref(); + let viewer = auth.as_ref().map(|auth| auth.did().into_static()); // Prepare timeline skeleton let page = if q.following_only { // Following feed: the viewer's followed set, then a time-ordered k-way merge. - let viewer = viewer.ok_or_else(|| { - XrpcError::InvalidParams("followingOnly requires a viewer did".into()) + let viewer = viewer.as_ref().ok_or_else(|| { + XrpcError::AuthRequired("followingOnly requires an authenticated viewer".into()) })?; - let followed = followed_dids(&state, viewer).await; + let followed = followed_dids(&state, &viewer).await; following_skeleton(&state, &followed, cursor, limit) } else { global_skeleton(&state, cursor, limit) @@ -63,7 +65,7 @@ let hydrated = stream::iter( page.items .iter() - .map(|it| hydrate_item(&state, viewer, it)) + .map(|it| hydrate_item(&state, viewer.as_ref(), it)) .collect::>(), ) .buffered(HYDRATE_CONCURRENCY) diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -84,6 +84,7 @@ }; use futures::Stream; use futures::stream::{self, StreamExt, TryStreamExt}; +use jacquard_axum::service_auth::{self, ServiceAuthConfig}; use jacquard_common::types::did::Did; use jacquard_common::types::ident::AtIdentifier; use jacquard_common::types::nsid::Nsid; @@ -149,6 +150,7 @@ pub awaiting: Arc, pub client_address: Arc, pub settlements: Arc, + service_auth_config: service_auth::ServiceAuthConfig>, enrich_router: Arc>, } @@ -184,11 +186,15 @@ search, resolver, identity, - directory, + directory: directory.clone(), limiter: None, awaiting: Arc::new(AwaitLimiter::new(MaxAwaiting::default())), client_address: Arc::new(ClientAddress::default()), settlements: Arc::new(Settlements::new()), + service_auth_config: ServiceAuthConfig::new( + Did::new_static("did:web:localhost").unwrap(), + directory, + ), enrich_router: Arc::new(std::sync::OnceLock::new()), } } @@ -228,6 +234,11 @@ self } + pub fn with_service_did(mut self, did: Did) -> Self { + self.service_auth_config = ServiceAuthConfig::new(did, self.directory.clone()); + self + } + /// for internal xrpc dispatch [`enrich`] pub fn self_router(&self) -> Router { self.enrich_router @@ -237,6 +248,29 @@ pub(crate) fn heavy_permit(&self) -> Result, XrpcError> { self.limiter.as_ref().map(|l| l.try_enter()).transpose() + } +} + +impl service_auth::ServiceAuth for AppState { + type Resolver = Directory; + + fn service_did(&self) -> Did<&str> { + self.service_auth_config.service_did() + } + fn resolver(&self) -> &Self::Resolver { + self.service_auth_config.resolver() + } + fn require_lxm(&self) -> bool { + service_auth::ServiceAuth::require_lxm(&self.service_auth_config) + } + fn allowed_services(&self) -> &[jacquard_common::SmolStr] { + self.service_auth_config.allowed_services() + } + fn replay_protection_enabled(&self) -> bool { + self.service_auth_config.replay_protection_enabled() + } + fn replay_store(&self) -> &dyn jacquard_axum::service_auth::ReplayStore { + self.service_auth_config.replay_store() } } @@ -428,7 +462,10 @@ "/xrpc/sh.tangled.repo.pull.countStatusesBy", get(count_pull_statuses_by), ) - .route("/xrpc/sh.tangled.pull.getPullView", get(repo::get_pull_view)) + .route( + "/xrpc/sh.tangled.pull.getPullView", + get(repo::get_pull_view), + ) // .route("/xrpc/sh.tangled.pull.listPullViews", get(repo::list_pull_views)) .route( "/xrpc/sh.tangled.spindle.listMembersBy", @@ -584,11 +621,7 @@ fn knot_proxied_routes() -> Router { let with_repo = register_proxied(Router::new(), REPO_PROXIED_NSIDS, proxy_repo_handler); let with_knot = register_proxied(with_repo, KNOT_PROXIED_NSIDS, proxy_knot_handler); - register_proxied( - with_knot, - MIRROR_V2_PROXIED_NSIDS, - proxy_mirror_v2_handler, - ) + register_proxied(with_knot, MIRROR_V2_PROXIED_NSIDS, proxy_mirror_v2_handler) } fn register_proxied( @@ -818,6 +851,12 @@ pub enum XrpcError { #[error("invalid request: {0}")] InvalidParams(String), + #[error("authentication required: {0}")] + AuthRequired(String), + // /// A service-auth token failed verification. Typed so the jwt-level checks speak jacquard's + // /// vocabulary; adopting `jacquard-axum` later replaces the producer, not this variant. + // #[error(transparent)] + // Auth(#[from] jacquard_common::service_auth::ServiceAuthError), #[error("record not found")] NotFound, #[error("upstream unavailable: {0}")] @@ -850,6 +889,7 @@ fn into_response(self) -> Response { let (status, error) = match &self { Self::InvalidParams(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"), + Self::AuthRequired(_) => (StatusCode::UNAUTHORIZED, "AuthenticationRequired"), Self::NotFound => (StatusCode::NOT_FOUND, "RecordNotFound"), Self::UpstreamUnavailable(_) => (StatusCode::BAD_GATEWAY, "UpstreamFailed"), Self::UpstreamGone(_) => (StatusCode::BAD_GATEWAY, "UpstreamGone"), @@ -1715,9 +1755,12 @@ Ok(None) } }, - Err(err @ (XrpcError::Internal(_) | XrpcError::Overloaded | XrpcError::NotSettled)) => { - Err(err) - } + Err( + err @ (XrpcError::Internal(_) + | XrpcError::Overloaded + | XrpcError::NotSettled + | XrpcError::AuthRequired(_)), + ) => Err(err), } } diff --git a/bobbin/crates/xrpc/src/repo.rs b/bobbin/crates/xrpc/src/repo.rs --- a/bobbin/crates/xrpc/src/repo.rs +++ b/bobbin/crates/xrpc/src/repo.rs @@ -1,6 +1,9 @@ #![allow(unused)] use crate::{ - AppState, SubjectQuery, TypedListQuery, XrpcError, accept_state_source, fetch, filter::ListFilter, record_edge_page, view::{build_profile_basic, build_repo_view_basic} + AppState, SubjectQuery, TypedListQuery, XrpcError, accept_state_source, fetch, + filter::ListFilter, + record_edge_page, + view::{build_profile_basic, build_repo_view_basic}, }; use axum::extract::State; use bobbin_edge_index::{EdgeItem, StateKind}; @@ -12,7 +15,7 @@ repo::pull::{Pull, PullRecord}, }, }; -use futures::{stream, StreamExt as _}; +use futures::{StreamExt as _, stream}; use jacquard_axum::{ExtractXrpc, XrpcResponse}; const HYDRATE_CONCURRENCY: usize = 8; @@ -77,13 +80,17 @@ extra_data: Default::default(), }), target_branch: pull.target.branch, - versions: pull.versions.into_iter().map(|version| { - PatchViewBuilder::new() - .base(version.base) - .head(version.head) - .created_at(version.created_at) - .build() - }).collect(), + versions: pull + .versions + .into_iter() + .map(|version| { + PatchViewBuilder::new() + .base(version.base) + .head(version.head) + .created_at(version.created_at) + .build() + }) + .collect(), state: pull_state, extra_data: Default::default(), }, diff --git a/bobbin/crates/xrpc/src/view.rs b/bobbin/crates/xrpc/src/view.rs --- a/bobbin/crates/xrpc/src/view.rs +++ b/bobbin/crates/xrpc/src/view.rs @@ -1,6 +1,6 @@ use bobbin_edge_index::EdgeStore; use bobbin_types::{ - ids::{nsid_static, owner_did_from_aturi, EdgeKey, SubjectRef}, + ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}, sh_tangled::{ actor, feed::star::StarRecord, @@ -9,13 +9,13 @@ }, }; use jacquard_common::{ + DefaultStr, types::{aturi::AtUri, did::Did, string::Handle}, xrpc::XrpcResp as _, - DefaultStr, }; use tracing::error; -use crate::{fetch, AppState, XrpcError}; +use crate::{AppState, XrpcError, fetch}; pub(crate) async fn build_profile_basic( state: &AppState,