alice.pds.demo.boltless.dev
This repository has no description
17 kB
509 lines
1use std::future::Future;
2use std::pin::Pin;
3use std::sync::Arc;
4use std::time::Duration;
5
6use bobbin_knot_proxy::{KnotHost, KnotHostError, PrivateAddressFilter, PrivateHostReason};
7use bobbin_runtime::{HttpRequest, HttpResponseHead, HttpTransport, NetworkError, ReqwestHttp};
8use bytes::{Bytes, BytesMut};
9use chrono::{DateTime, Utc};
10use futures::TryStreamExt;
11use http::{HeaderMap, StatusCode};
12use jacquard_common::DefaultStr;
13use jacquard_common::types::did::Did;
14use jacquard_common::types::nsid::Nsid;
15use serde::Deserialize;
16use thiserror::Error;
17use url::Url;
18
19const USER_AGENT: &str = concat!("bobbin/", env!("CARGO_PKG_VERSION"));
20const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
21const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
22const MAX_BODY_BYTES: u64 = 4 * 1024 * 1024;
23const LIST_PAGE_LIMIT: i64 = 1000;
24const MAX_LIST_PAGES: usize = 256;
25
26const VERSION_NSID: &str = "sh.tangled.knot.version";
27const LIST_MEMBERS_NSID: &str = "sh.tangled.knot.listMembers";
28const LIST_COLLABORATORS_NSID: &str = "sh.tangled.repo.listCollaborators";
29
30#[derive(Clone)]
31pub struct KnotClient {
32 http: Arc<dyn HttpTransport>,
33}
34
35#[derive(Clone, Debug, Eq, PartialEq, Deserialize)]
36#[serde(rename_all = "camelCase")]
37pub struct AclEntry {
38 pub subject: Did<DefaultStr>,
39 pub created_at: DateTime<Utc>,
40}
41
42#[derive(Clone, Copy, Debug, Eq, PartialEq)]
43pub enum Completeness {
44 Complete,
45 Truncated,
46}
47
48#[derive(Clone, Debug, Eq, PartialEq)]
49pub struct AclListing {
50 pub entries: Vec<AclEntry>,
51 pub completeness: Completeness,
52}
53
54#[derive(Debug, Error)]
55pub enum KnotClientError {
56 #[error("knot host: {0}")]
57 Host(#[from] KnotHostError),
58 #[error("blocked: knot {host} resolves to {reason} address space")]
59 PrivateHost {
60 host: String,
61 reason: PrivateHostReason,
62 },
63 #[error("http client build: {0}")]
64 Build(String),
65 #[error("network: {0}")]
66 Network(#[from] NetworkError),
67 #[error("xrpc not found")]
68 NotFound,
69 #[error("upstream returned status {0}")]
70 Upstream(StatusCode),
71 #[error("response body exceeded {limit} bytes")]
72 BodyTooLarge { limit: u64 },
73 #[error("decode response: {0}")]
74 Decode(#[from] serde_json::Error),
75 #[error("extra ca cert: {0}")]
76 ExtraCa(#[from] bobbin_runtime::ExtraCaError),
77}
78
79pub fn knot_endpoint(
80 host: &str,
81 dev: bool,
82 allow_private: bool,
83) -> Result<KnotHost, KnotClientError> {
84 let raw = if dev {
85 format!("http://{host}")
86 } else {
87 host.to_owned()
88 };
89 let knot = KnotHost::parse(&raw)?;
90 if !allow_private && let Some(reason) = knot.private_literal_reason() {
91 return Err(KnotClientError::PrivateHost {
92 host: host.to_owned(),
93 reason,
94 });
95 }
96 Ok(knot)
97}
98
99fn default_http_client(
100 allow_private: bool,
101 extra_ca_cert: Option<&std::path::Path>,
102) -> Result<reqwest::Client, KnotClientError> {
103 let mut builder = reqwest::Client::builder()
104 .user_agent(USER_AGENT)
105 .timeout(REQUEST_TIMEOUT)
106 .connect_timeout(CONNECT_TIMEOUT)
107 .redirect(reqwest::redirect::Policy::none())
108 .dns_resolver(Arc::new(PrivateAddressFilter::new(allow_private)));
109 if let Some(path) = extra_ca_cert {
110 builder = builder.add_root_certificate(bobbin_runtime::load_extra_ca_cert(path)?);
111 }
112 builder
113 .build()
114 .map_err(|e| KnotClientError::Build(e.to_string()))
115}
116
117fn nsid(s: &'static str) -> Nsid<DefaultStr> {
118 Nsid::new_static(s).expect("static nsid literal must validate")
119}
120
121pub(crate) fn authority(host: &KnotHost) -> String {
122 let url = host.url();
123 match (url.host_str(), url.port()) {
124 (Some(h), Some(p)) => format!("{h}:{p}"),
125 (Some(h), None) => h.to_owned(),
126 (None, _) => String::new(),
127 }
128}
129
130impl KnotClient {
131 pub fn new(http: Arc<dyn HttpTransport>) -> Self {
132 Self { http }
133 }
134
135 pub fn with_default_http(
136 allow_private: bool,
137 extra_ca_cert: Option<&std::path::Path>,
138 ) -> Result<Self, KnotClientError> {
139 let client = default_http_client(allow_private, extra_ca_cert)?;
140 Ok(Self::new(ReqwestHttp::shared(client)))
141 }
142
143 pub async fn capabilities(&self, host: &KnotHost) -> Result<Vec<String>, KnotClientError> {
144 let mut url = host.xrpc_url(&nsid(VERSION_NSID));
145 url.set_query(None);
146 let bytes = self.get_json(url).await?;
147 let resp: VersionWire = serde_json::from_slice(&bytes)?;
148 Ok(resp.capabilities.unwrap_or_default())
149 }
150
151 pub async fn list_members(&self, host: &KnotHost) -> Result<AclListing, KnotClientError> {
152 let subject = authority(host);
153 self.drain(host, LIST_MEMBERS_NSID, subject, None, 0, Vec::new())
154 .await
155 }
156
157 pub async fn list_collaborators(
158 &self,
159 host: &KnotHost,
160 repo: &Did<DefaultStr>,
161 ) -> Result<AclListing, KnotClientError> {
162 self.drain(
163 host,
164 LIST_COLLABORATORS_NSID,
165 repo.as_ref().to_owned(),
166 None,
167 0,
168 Vec::new(),
169 )
170 .await
171 }
172
173 fn drain<'a>(
174 &'a self,
175 host: &'a KnotHost,
176 endpoint: &'static str,
177 subject: String,
178 cursor: Option<String>,
179 page: usize,
180 mut acc: Vec<AclEntry>,
181 ) -> Pin<Box<dyn Future<Output = Result<AclListing, KnotClientError>> + Send + 'a>> {
182 Box::pin(async move {
183 if page >= MAX_LIST_PAGES {
184 tracing::warn!(
185 host = %authority(host),
186 endpoint,
187 pages = page,
188 "knot list truncated at page cap"
189 );
190 return Ok(AclListing {
191 entries: acc,
192 completeness: Completeness::Truncated,
193 });
194 }
195 let resp = self
196 .fetch_page(host, endpoint, &subject, cursor.as_deref())
197 .await?;
198 acc.extend(resp.items);
199 match resp.cursor.filter(|c| !c.is_empty()) {
200 Some(next) => {
201 self.drain(host, endpoint, subject, Some(next), page + 1, acc)
202 .await
203 }
204 None => Ok(AclListing {
205 entries: acc,
206 completeness: Completeness::Complete,
207 }),
208 }
209 })
210 }
211
212 async fn fetch_page(
213 &self,
214 host: &KnotHost,
215 endpoint: &'static str,
216 subject: &str,
217 cursor: Option<&str>,
218 ) -> Result<ListWire, KnotClientError> {
219 let mut url = host.xrpc_url(&nsid(endpoint));
220 {
221 let mut q = url.query_pairs_mut();
222 q.clear();
223 q.append_pair("subject", subject);
224 q.append_pair("limit", &LIST_PAGE_LIMIT.to_string());
225 if let Some(c) = cursor {
226 q.append_pair("cursor", c);
227 }
228 }
229 let bytes = self.get_json(url).await?;
230 Ok(serde_json::from_slice(&bytes)?)
231 }
232
233 async fn get_json(&self, url: Url) -> Result<Bytes, KnotClientError> {
234 let resp = self
235 .http
236 .execute(HttpRequest {
237 url,
238 headers: HeaderMap::new(),
239 })
240 .await?;
241 match resp.status {
242 StatusCode::OK => read_bounded(resp).await,
243 StatusCode::NOT_FOUND => Err(KnotClientError::NotFound),
244 other => Err(KnotClientError::Upstream(other)),
245 }
246 }
247}
248
249#[derive(Deserialize)]
250struct VersionWire {
251 #[serde(default)]
252 capabilities: Option<Vec<String>>,
253}
254
255#[derive(Deserialize)]
256struct ListWire {
257 #[serde(default)]
258 items: Vec<AclEntry>,
259 #[serde(default)]
260 cursor: Option<String>,
261}
262
263async fn read_bounded(resp: HttpResponseHead) -> Result<Bytes, KnotClientError> {
264 if resp.content_length.is_some_and(|len| len > MAX_BODY_BYTES) {
265 return Err(KnotClientError::BodyTooLarge {
266 limit: MAX_BODY_BYTES,
267 });
268 }
269 let buf = resp
270 .body
271 .map_err(KnotClientError::Network)
272 .try_fold(BytesMut::new(), |mut acc, chunk| async move {
273 if (acc.len() as u64).saturating_add(chunk.len() as u64) > MAX_BODY_BYTES {
274 return Err(KnotClientError::BodyTooLarge {
275 limit: MAX_BODY_BYTES,
276 });
277 }
278 acc.extend_from_slice(&chunk);
279 Ok(acc)
280 })
281 .await?;
282 Ok(buf.freeze())
283}
284
285#[cfg(test)]
286mod tests {
287 use super::*;
288 use serde_json::json;
289 use wiremock::matchers::{method, path, query_param};
290 use wiremock::{Mock, MockServer, ResponseTemplate};
291
292 fn did(s: &str) -> Did<DefaultStr> {
293 Did::new_owned(s).unwrap()
294 }
295
296 fn client() -> KnotClient {
297 KnotClient::new(ReqwestHttp::shared(default_http_client(true, None).unwrap()))
298 }
299
300 fn endpoint(server: &MockServer) -> KnotHost {
301 KnotHost::parse(&server.uri()).unwrap()
302 }
303
304 #[tokio::test]
305 async fn capabilities_returns_declared_tokens() {
306 let server = MockServer::start().await;
307 Mock::given(method("GET"))
308 .and(path("/xrpc/sh.tangled.knot.version"))
309 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
310 "version": "1.0.0 (deadbeef)",
311 "capabilities": ["knot-acl"]
312 })))
313 .mount(&server)
314 .await;
315
316 let caps = client().capabilities(&endpoint(&server)).await.unwrap();
317 assert_eq!(caps, vec!["knot-acl".to_owned()]);
318 }
319
320 #[tokio::test]
321 async fn capabilities_empty_when_field_absent() {
322 let server = MockServer::start().await;
323 Mock::given(method("GET"))
324 .and(path("/xrpc/sh.tangled.knot.version"))
325 .respond_with(
326 ResponseTemplate::new(200).set_body_json(json!({ "version": "1.0.0 (cafe)" })),
327 )
328 .mount(&server)
329 .await;
330
331 let caps = client().capabilities(&endpoint(&server)).await.unwrap();
332 assert!(caps.is_empty());
333 }
334
335 #[tokio::test]
336 async fn list_members_drains_single_page() {
337 let server = MockServer::start().await;
338 let host = endpoint(&server);
339 Mock::given(method("GET"))
340 .and(path("/xrpc/sh.tangled.knot.listMembers"))
341 .and(query_param("subject", authority(&host).as_str()))
342 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
343 "items": [
344 {"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"},
345 {"subject": "did:plc:akshay", "addedBy": "did:plc:akshay", "createdAt": "2026-06-02T12:00:00Z"}
346 ]
347 })))
348 .mount(&server)
349 .await;
350
351 let listing = client().list_members(&host).await.unwrap();
352 assert_eq!(listing.completeness, Completeness::Complete);
353 assert_eq!(
354 listing.entries,
355 vec![
356 AclEntry {
357 subject: did("did:plc:boltless"),
358 created_at: "2026-06-01T00:00:00Z".parse().unwrap(),
359 },
360 AclEntry {
361 subject: did("did:plc:akshay"),
362 created_at: "2026-06-02T12:00:00Z".parse().unwrap(),
363 },
364 ]
365 );
366 }
367
368 #[tokio::test]
369 async fn list_members_drains_multiple_pages() {
370 let server = MockServer::start().await;
371 let host = endpoint(&server);
372 Mock::given(method("GET"))
373 .and(path("/xrpc/sh.tangled.knot.listMembers"))
374 .and(query_param("cursor", "p2"))
375 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
376 "items": [{"subject": "did:plc:akshay", "addedBy": "did:plc:akshay", "createdAt": "2026-06-02T00:00:00Z"}]
377 })))
378 .mount(&server)
379 .await;
380 Mock::given(method("GET"))
381 .and(path("/xrpc/sh.tangled.knot.listMembers"))
382 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
383 "items": [{"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"}],
384 "cursor": "p2"
385 })))
386 .mount(&server)
387 .await;
388
389 let listing = client().list_members(&host).await.unwrap();
390 assert_eq!(listing.completeness, Completeness::Complete);
391 let subjects: Vec<_> = listing.entries.into_iter().map(|m| m.subject).collect();
392 assert_eq!(
393 subjects,
394 vec![did("did:plc:boltless"), did("did:plc:akshay")]
395 );
396 }
397
398 #[tokio::test]
399 async fn list_collaborators_uses_repo_did_subject() {
400 let server = MockServer::start().await;
401 let host = endpoint(&server);
402 Mock::given(method("GET"))
403 .and(path("/xrpc/sh.tangled.repo.listCollaborators"))
404 .and(query_param("subject", "did:plc:scallop"))
405 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
406 "items": [{"subject": "did:plc:olaren", "addedBy": "did:plc:boltless", "createdAt": "2026-06-03T00:00:00Z"}]
407 })))
408 .mount(&server)
409 .await;
410
411 let listing = client()
412 .list_collaborators(&host, &did("did:plc:scallop"))
413 .await
414 .unwrap();
415 assert_eq!(listing.completeness, Completeness::Complete);
416 assert_eq!(listing.entries.len(), 1);
417 assert_eq!(listing.entries[0].subject, did("did:plc:olaren"));
418 }
419
420 #[tokio::test]
421 async fn drain_reports_truncation_at_page_cap() {
422 let server = MockServer::start().await;
423 let host = endpoint(&server);
424 Mock::given(method("GET"))
425 .and(path("/xrpc/sh.tangled.knot.listMembers"))
426 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
427 "items": [{"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"}],
428 "cursor": "more"
429 })))
430 .mount(&server)
431 .await;
432
433 let listing = client().list_members(&host).await.unwrap();
434 assert_eq!(
435 listing.completeness,
436 Completeness::Truncated,
437 "a never-terminating cursor must surface as a truncated listing"
438 );
439 assert_eq!(listing.entries.len(), MAX_LIST_PAGES);
440 }
441
442 #[tokio::test]
443 async fn maps_404_to_not_found() {
444 let server = MockServer::start().await;
445 Mock::given(method("GET"))
446 .respond_with(ResponseTemplate::new(404))
447 .mount(&server)
448 .await;
449
450 let err = client()
451 .capabilities(&endpoint(&server))
452 .await
453 .expect_err("404 must surface");
454 assert!(matches!(err, KnotClientError::NotFound));
455 }
456
457 #[tokio::test]
458 async fn maps_5xx_to_upstream() {
459 let server = MockServer::start().await;
460 Mock::given(method("GET"))
461 .respond_with(ResponseTemplate::new(503))
462 .mount(&server)
463 .await;
464
465 let err = client()
466 .capabilities(&endpoint(&server))
467 .await
468 .expect_err("5xx must surface");
469 match err {
470 KnotClientError::Upstream(s) => assert_eq!(s.as_u16(), 503),
471 other => panic!("wrong variant: {other:?}"),
472 }
473 }
474
475 #[tokio::test]
476 async fn does_not_follow_redirects() {
477 let server = MockServer::start().await;
478 Mock::given(method("GET"))
479 .and(path("/xrpc/sh.tangled.knot.version"))
480 .respond_with(ResponseTemplate::new(301).insert_header(
481 "location",
482 "https://kt.tngl.oyster.cafe/xrpc/sh.tangled.knot.version",
483 ))
484 .mount(&server)
485 .await;
486
487 let err = client()
488 .capabilities(&endpoint(&server))
489 .await
490 .expect_err("a redirect must surface as an error, not be followed to another knot");
491 match err {
492 KnotClientError::Upstream(s) => assert_eq!(s.as_u16(), 301),
493 other => panic!("expected Upstream(301), got {other:?}"),
494 }
495 }
496
497 #[test]
498 fn knot_endpoint_rejects_private_host() {
499 let err =
500 knot_endpoint("127.0.0.1:9", true, false).expect_err("private host must be refused");
501 assert!(matches!(err, KnotClientError::PrivateHost { .. }));
502 }
503
504 #[test]
505 fn knot_endpoint_allows_private_when_permitted() {
506 let knot = knot_endpoint("127.0.0.1:9", true, true).expect("private host allowed");
507 assert_eq!(knot.url().scheme(), "http");
508 }
509}