Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
16 kB · 497 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498use std::future::Future;use std::pin::Pin;use std::sync::Arc;use std::time::Duration;
use bobbin_knot_proxy::{KnotHost, KnotHostError, PrivateAddressFilter, PrivateHostReason};use bobbin_runtime::{HttpRequest, HttpResponseHead, HttpTransport, NetworkError, ReqwestHttp};use bytes::{Bytes, BytesMut};use chrono::{DateTime, Utc};use futures::TryStreamExt;use http::{HeaderMap, StatusCode};use jacquard_common::DefaultStr;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;use serde::Deserialize;use thiserror::Error;use url::Url;
const USER_AGENT: &str = concat!("bobbin/", env!("CARGO_PKG_VERSION"));const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);const MAX_BODY_BYTES: u64 = 4 * 1024 * 1024;const LIST_PAGE_LIMIT: i64 = 1000;const MAX_LIST_PAGES: usize = 256;
const VERSION_NSID: &str = "sh.tangled.knot.version";const LIST_MEMBERS_NSID: &str = "sh.tangled.knot.listMembers";const LIST_COLLABORATORS_NSID: &str = "sh.tangled.repo.listCollaborators";
#[derive(Clone)]pub struct KnotClient { http: Arc<dyn HttpTransport>,}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize)]#[serde(rename_all = "camelCase")]pub struct AclEntry { pub subject: Did<DefaultStr>, pub created_at: DateTime<Utc>,}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub enum Completeness { Complete, Truncated,}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct AclListing { pub entries: Vec<AclEntry>, pub completeness: Completeness,}
#[derive(Debug, Error)]pub enum KnotClientError { #[error("knot host: {0}")] Host(#[from] KnotHostError), #[error("blocked: knot {host} resolves to {reason} address space")] PrivateHost { host: String, reason: PrivateHostReason, }, #[error("http client build: {0}")] Build(String), #[error("network: {0}")] Network(#[from] NetworkError), #[error("xrpc not found")] NotFound, #[error("upstream returned status {0}")] Upstream(StatusCode), #[error("response body exceeded {limit} bytes")] BodyTooLarge { limit: u64 }, #[error("decode response: {0}")] Decode(#[from] serde_json::Error),}
pub fn knot_endpoint( host: &str, dev: bool, allow_private: bool,) -> Result<KnotHost, KnotClientError> { let raw = if dev { format!("http://{host}") } else { host.to_owned() }; let knot = KnotHost::parse(&raw)?; if !allow_private && let Some(reason) = knot.private_literal_reason() { return Err(KnotClientError::PrivateHost { host: host.to_owned(), reason, }); } Ok(knot)}
fn default_http_client(allow_private: bool) -> Result<reqwest::Client, reqwest::Error> { reqwest::Client::builder() .user_agent(USER_AGENT) .timeout(REQUEST_TIMEOUT) .connect_timeout(CONNECT_TIMEOUT) .redirect(reqwest::redirect::Policy::none()) .dns_resolver(Arc::new(PrivateAddressFilter::new(allow_private))) .build()}
fn nsid(s: &'static str) -> Nsid<DefaultStr> { Nsid::new_static(s).expect("static nsid literal must validate")}
pub(crate) fn authority(host: &KnotHost) -> String { let url = host.url(); match (url.host_str(), url.port()) { (Some(h), Some(p)) => format!("{h}:{p}"), (Some(h), None) => h.to_owned(), (None, _) => String::new(), }}
impl KnotClient { pub fn new(http: Arc<dyn HttpTransport>) -> Self { Self { http } }
pub fn with_default_http(allow_private: bool) -> Result<Self, KnotClientError> { let client = default_http_client(allow_private) .map_err(|e| KnotClientError::Build(e.to_string()))?; Ok(Self::new(ReqwestHttp::shared(client))) }
pub async fn capabilities(&self, host: &KnotHost) -> Result<Vec<String>, KnotClientError> { let mut url = host.xrpc_url(&nsid(VERSION_NSID)); url.set_query(None); let bytes = self.get_json(url).await?; let resp: VersionWire = serde_json::from_slice(&bytes)?; Ok(resp.capabilities.unwrap_or_default()) }
pub async fn list_members(&self, host: &KnotHost) -> Result<AclListing, KnotClientError> { let subject = authority(host); self.drain(host, LIST_MEMBERS_NSID, subject, None, 0, Vec::new()) .await }
pub async fn list_collaborators( &self, host: &KnotHost, repo: &Did<DefaultStr>, ) -> Result<AclListing, KnotClientError> { self.drain( host, LIST_COLLABORATORS_NSID, repo.as_ref().to_owned(), None, 0, Vec::new(), ) .await }
fn drain<'a>( &'a self, host: &'a KnotHost, endpoint: &'static str, subject: String, cursor: Option<String>, page: usize, mut acc: Vec<AclEntry>, ) -> Pin<Box<dyn Future<Output = Result<AclListing, KnotClientError>> + Send + 'a>> { Box::pin(async move { if page >= MAX_LIST_PAGES { tracing::warn!( host = %authority(host), endpoint, pages = page, "knot list truncated at page cap" ); return Ok(AclListing { entries: acc, completeness: Completeness::Truncated, }); } let resp = self .fetch_page(host, endpoint, &subject, cursor.as_deref()) .await?; acc.extend(resp.items); match resp.cursor.filter(|c| !c.is_empty()) { Some(next) => { self.drain(host, endpoint, subject, Some(next), page + 1, acc) .await } None => Ok(AclListing { entries: acc, completeness: Completeness::Complete, }), } }) }
async fn fetch_page( &self, host: &KnotHost, endpoint: &'static str, subject: &str, cursor: Option<&str>, ) -> Result<ListWire, KnotClientError> { let mut url = host.xrpc_url(&nsid(endpoint)); { let mut q = url.query_pairs_mut(); q.clear(); q.append_pair("subject", subject); q.append_pair("limit", &LIST_PAGE_LIMIT.to_string()); if let Some(c) = cursor { q.append_pair("cursor", c); } } let bytes = self.get_json(url).await?; Ok(serde_json::from_slice(&bytes)?) }
async fn get_json(&self, url: Url) -> Result<Bytes, KnotClientError> { let resp = self .http .execute(HttpRequest { url, headers: HeaderMap::new(), }) .await?; match resp.status { StatusCode::OK => read_bounded(resp).await, StatusCode::NOT_FOUND => Err(KnotClientError::NotFound), other => Err(KnotClientError::Upstream(other)), } }}
#[derive(Deserialize)]struct VersionWire { #[serde(default)] capabilities: Option<Vec<String>>,}
#[derive(Deserialize)]struct ListWire { #[serde(default)] items: Vec<AclEntry>, #[serde(default)] cursor: Option<String>,}
async fn read_bounded(resp: HttpResponseHead) -> Result<Bytes, KnotClientError> { if resp.content_length.is_some_and(|len| len > MAX_BODY_BYTES) { return Err(KnotClientError::BodyTooLarge { limit: MAX_BODY_BYTES, }); } let buf = resp .body .map_err(KnotClientError::Network) .try_fold(BytesMut::new(), |mut acc, chunk| async move { if (acc.len() as u64).saturating_add(chunk.len() as u64) > MAX_BODY_BYTES { return Err(KnotClientError::BodyTooLarge { limit: MAX_BODY_BYTES, }); } acc.extend_from_slice(&chunk); Ok(acc) }) .await?; Ok(buf.freeze())}
#[cfg(test)]mod tests { use super::*; use serde_json::json; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate};
fn did(s: &str) -> Did<DefaultStr> { Did::new_owned(s).unwrap() }
fn client() -> KnotClient { KnotClient::new(ReqwestHttp::shared(default_http_client(true).unwrap())) }
fn endpoint(server: &MockServer) -> KnotHost { KnotHost::parse(&server.uri()).unwrap() }
#[tokio::test] async fn capabilities_returns_declared_tokens() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.version")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "version": "1.0.0 (deadbeef)", "capabilities": ["knot-acl"] }))) .mount(&server) .await;
let caps = client().capabilities(&endpoint(&server)).await.unwrap(); assert_eq!(caps, vec!["knot-acl".to_owned()]); }
#[tokio::test] async fn capabilities_empty_when_field_absent() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.version")) .respond_with( ResponseTemplate::new(200).set_body_json(json!({ "version": "1.0.0 (cafe)" })), ) .mount(&server) .await;
let caps = client().capabilities(&endpoint(&server)).await.unwrap(); assert!(caps.is_empty()); }
#[tokio::test] async fn list_members_drains_single_page() { let server = MockServer::start().await; let host = endpoint(&server); Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.listMembers")) .and(query_param("subject", authority(&host).as_str())) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "items": [ {"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"}, {"subject": "did:plc:akshay", "addedBy": "did:plc:akshay", "createdAt": "2026-06-02T12:00:00Z"} ] }))) .mount(&server) .await;
let listing = client().list_members(&host).await.unwrap(); assert_eq!(listing.completeness, Completeness::Complete); assert_eq!( listing.entries, vec![ AclEntry { subject: did("did:plc:boltless"), created_at: "2026-06-01T00:00:00Z".parse().unwrap(), }, AclEntry { subject: did("did:plc:akshay"), created_at: "2026-06-02T12:00:00Z".parse().unwrap(), }, ] ); }
#[tokio::test] async fn list_members_drains_multiple_pages() { let server = MockServer::start().await; let host = endpoint(&server); Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.listMembers")) .and(query_param("cursor", "p2")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "items": [{"subject": "did:plc:akshay", "addedBy": "did:plc:akshay", "createdAt": "2026-06-02T00:00:00Z"}] }))) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.listMembers")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "items": [{"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"}], "cursor": "p2" }))) .mount(&server) .await;
let listing = client().list_members(&host).await.unwrap(); assert_eq!(listing.completeness, Completeness::Complete); let subjects: Vec<_> = listing.entries.into_iter().map(|m| m.subject).collect(); assert_eq!( subjects, vec![did("did:plc:boltless"), did("did:plc:akshay")] ); }
#[tokio::test] async fn list_collaborators_uses_repo_did_subject() { let server = MockServer::start().await; let host = endpoint(&server); Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.repo.listCollaborators")) .and(query_param("subject", "did:plc:scallop")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "items": [{"subject": "did:plc:olaren", "addedBy": "did:plc:boltless", "createdAt": "2026-06-03T00:00:00Z"}] }))) .mount(&server) .await;
let listing = client() .list_collaborators(&host, &did("did:plc:scallop")) .await .unwrap(); assert_eq!(listing.completeness, Completeness::Complete); assert_eq!(listing.entries.len(), 1); assert_eq!(listing.entries[0].subject, did("did:plc:olaren")); }
#[tokio::test] async fn drain_reports_truncation_at_page_cap() { let server = MockServer::start().await; let host = endpoint(&server); Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.listMembers")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "items": [{"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"}], "cursor": "more" }))) .mount(&server) .await;
let listing = client().list_members(&host).await.unwrap(); assert_eq!( listing.completeness, Completeness::Truncated, "a never-terminating cursor must surface as a truncated listing" ); assert_eq!(listing.entries.len(), MAX_LIST_PAGES); }
#[tokio::test] async fn maps_404_to_not_found() { let server = MockServer::start().await; Mock::given(method("GET")) .respond_with(ResponseTemplate::new(404)) .mount(&server) .await;
let err = client() .capabilities(&endpoint(&server)) .await .expect_err("404 must surface"); assert!(matches!(err, KnotClientError::NotFound)); }
#[tokio::test] async fn maps_5xx_to_upstream() { let server = MockServer::start().await; Mock::given(method("GET")) .respond_with(ResponseTemplate::new(503)) .mount(&server) .await;
let err = client() .capabilities(&endpoint(&server)) .await .expect_err("5xx must surface"); match err { KnotClientError::Upstream(s) => assert_eq!(s.as_u16(), 503), other => panic!("wrong variant: {other:?}"), } }
#[tokio::test] async fn does_not_follow_redirects() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.knot.version")) .respond_with(ResponseTemplate::new(301).insert_header( "location", "https://kt.tngl.oyster.cafe/xrpc/sh.tangled.knot.version", )) .mount(&server) .await;
let err = client() .capabilities(&endpoint(&server)) .await .expect_err("a redirect must surface as an error, not be followed to another knot"); match err { KnotClientError::Upstream(s) => assert_eq!(s.as_u16(), 301), other => panic!("expected Upstream(301), got {other:?}"), } }
#[test] fn knot_endpoint_rejects_private_host() { let err = knot_endpoint("127.0.0.1:9", true, false).expect_err("private host must be refused"); assert!(matches!(err, KnotClientError::PrivateHost { .. })); }
#[test] fn knot_endpoint_allows_private_when_permitted() { let knot = knot_endpoint("127.0.0.1:9", true, true).expect("private host allowed"); assert_eq!(knot.url().scheme(), "http"); }}