alice.pds.demo.boltless.dev
This repository has no description
bobbin/crates/knot-proxy: add mirror client & git NSID routes
Lewis: May this revision serve well! <did:plc:3fwecdnvtcscjnrx2p4n7alz>
author
did:plc:3fwecdnvtcscjnrx2p4n7a…
committer
Tangled
date
(Aug 1, 2026, 7:17 PM UTC)
commit
2f888b61
2f888b612f41facb15c0f94afabc544182096d2b
parent
d3d6a888
d3d6a88814063ba1ee36d0c7f62cc526e9725db2
change-id
qwxwpkmu
qwxwpkmuqkunvxpvszonroyoomwztzor
did:plc:3fwecdnvtcscjnrx2p4n7a…
committer
+254
-1
+2
bobbin/containerfiles/bobbin.Containerfile
+2
bobbin/containerfiles/bobbin.Containerfile
+15
-1
bobbin/crates/knot-proxy/src/lib.rs
+15
-1
bobbin/crates/knot-proxy/src/lib.rs
···
8
8
NetworkError, ReqwestHttp, RuntimeHasher,
9
9
};
10
10
use bytes::Bytes;
11
-
use futures::Stream;
11
+
use futures::{Stream, StreamExt};
12
12
use http::{HeaderMap, StatusCode};
13
13
use jacquard_common::BosStr;
14
14
use jacquard_common::types::nsid::Nsid;
···
20
20
mod breaker;
21
21
mod dns;
22
22
mod host;
23
+
mod mirror;
23
24
24
25
pub use breaker::{Breaker, BreakerPermit, CircuitOpen, FailureThreshold, ThresholdError};
25
26
pub use dns::PrivateAddressFilter;
26
27
pub use host::{KnotHost, KnotHostError, PrivateHostReason, RepoSlug, RepoSlugError, classify_ip};
28
+
pub use mirror::{MirrorNsid, MirrorProxy, MirrorProxyError};
27
29
28
30
const USER_AGENT: &str = concat!("bobbin/", env!("CARGO_PKG_VERSION"));
29
31
const HTTPS_SCHEME: &str = "https";
32
+
const DISCARD_BUDGET_BYTES: usize = 64 * 1024;
30
33
31
34
#[derive(Clone, Debug)]
32
35
pub struct KnotProxyConfig {
···
226
229
227
230
pub fn into_body_stream(self) -> BodyStream {
228
231
BodyStream::new(self.body, self.permit)
232
+
}
233
+
234
+
pub async fn discard(self) {
235
+
let mut stream = self.into_body_stream();
236
+
let mut seen = 0usize;
237
+
while seen < DISCARD_BUDGET_BYTES {
238
+
match stream.next().await {
239
+
None | Some(Err(_)) => return,
240
+
Some(Ok(chunk)) => seen = seen.saturating_add(chunk.len().max(1)),
241
+
}
242
+
}
229
243
}
230
244
}
231
245
+237
bobbin/crates/knot-proxy/src/mirror.rs
+237
bobbin/crates/knot-proxy/src/mirror.rs
···
1
+
use std::sync::Arc;
2
+
use std::time::Duration;
3
+
4
+
use bobbin_runtime::{Clock, RuntimeHasher};
5
+
use http::HeaderMap;
6
+
use jacquard_common::types::did::Did;
7
+
use jacquard_common::types::nsid::Nsid;
8
+
use jacquard_common::{BosStr, DefaultStr};
9
+
use thiserror::Error;
10
+
use url::Url;
11
+
12
+
use crate::breaker::FailureThreshold;
13
+
use crate::host::{KnotHost, KnotHostError};
14
+
use crate::{KnotHttpConfig, KnotProxy, KnotProxyConfig, KnotProxyError, ProxyResponse};
15
+
16
+
const REPO_PARAM: &str = "repo";
17
+
const PATH_PARAM: &str = "path";
18
+
19
+
const MIRROR_HTTP: KnotHttpConfig = KnotHttpConfig {
20
+
connect_timeout: Duration::from_secs(3),
21
+
read_timeout: Duration::from_secs(5),
22
+
};
23
+
24
+
#[derive(Debug, Error)]
25
+
pub enum MirrorProxyError {
26
+
#[error("mirror url: {0}")]
27
+
Host(#[from] KnotHostError),
28
+
#[error("mirror http client: {0}")]
29
+
Http(#[from] reqwest::Error),
30
+
#[error("mirror url must be a bare origin, got {0}")]
31
+
NotAnOrigin(String),
32
+
}
33
+
34
+
struct MirrorRoute {
35
+
knot: &'static str,
36
+
mirror: &'static str,
37
+
eligible: fn(&[(&str, &str)]) -> bool,
38
+
}
39
+
40
+
const ROUTES: &[MirrorRoute] = &[
41
+
MirrorRoute {
42
+
knot: "sh.tangled.repo.branches",
43
+
mirror: "sh.tangled.git.temp.listBranches",
44
+
eligible: any_query,
45
+
},
46
+
MirrorRoute {
47
+
knot: "sh.tangled.repo.log",
48
+
mirror: "sh.tangled.git.temp.listCommits",
49
+
eligible: no_path_filter,
50
+
},
51
+
MirrorRoute {
52
+
knot: "sh.tangled.repo.tag",
53
+
mirror: "sh.tangled.git.temp.getTag",
54
+
eligible: any_query,
55
+
},
56
+
MirrorRoute {
57
+
knot: "sh.tangled.repo.tags",
58
+
mirror: "sh.tangled.git.temp.listTags",
59
+
eligible: any_query,
60
+
},
61
+
MirrorRoute {
62
+
knot: "sh.tangled.repo.tree",
63
+
mirror: "sh.tangled.git.temp.getTree",
64
+
eligible: any_query,
65
+
},
66
+
];
67
+
68
+
fn any_query(_: &[(&str, &str)]) -> bool {
69
+
true
70
+
}
71
+
72
+
fn no_path_filter(query: &[(&str, &str)]) -> bool {
73
+
!query.iter().any(|(k, v)| *k == PATH_PARAM && !v.is_empty())
74
+
}
75
+
76
+
#[derive(Debug)]
77
+
pub struct MirrorNsid(Nsid<DefaultStr>);
78
+
79
+
impl MirrorNsid {
80
+
pub fn route(knot_nsid: &str, query: &[(&str, &str)]) -> Option<Self> {
81
+
let route = ROUTES.iter().find(|r| r.knot == knot_nsid)?;
82
+
(route.eligible)(query).then(|| {
83
+
Self(Nsid::new_static(route.mirror).expect("every nsid in ROUTES is a valid literal"))
84
+
})
85
+
}
86
+
87
+
pub fn as_str(&self) -> &str {
88
+
self.0.as_ref()
89
+
}
90
+
}
91
+
92
+
pub struct MirrorProxy {
93
+
proxy: KnotProxy,
94
+
host: KnotHost,
95
+
}
96
+
97
+
impl MirrorProxy {
98
+
pub fn new(
99
+
url: &Url,
100
+
clock: Arc<dyn Clock>,
101
+
hasher: RuntimeHasher,
102
+
) -> Result<Self, MirrorProxyError> {
103
+
let host = KnotHost::parse(url.as_str())?;
104
+
match beyond_origin(url) {
105
+
Some(extra) => Err(MirrorProxyError::NotAnOrigin(extra)),
106
+
None => Ok(Self {
107
+
proxy: KnotProxy::new(
108
+
KnotProxyConfig {
109
+
allow_private_hosts: true,
110
+
require_https: false,
111
+
failure_threshold: FailureThreshold::new(3).expect("nonzero literal"),
112
+
..KnotProxyConfig::default()
113
+
},
114
+
MIRROR_HTTP,
115
+
clock,
116
+
hasher,
117
+
)?,
118
+
host,
119
+
}),
120
+
}
121
+
}
122
+
123
+
pub fn host(&self) -> &KnotHost {
124
+
&self.host
125
+
}
126
+
127
+
pub async fn forward<S: BosStr + AsRef<str>>(
128
+
&self,
129
+
nsid: &MirrorNsid,
130
+
repo: &Did<S>,
131
+
query: &[(&str, &str)],
132
+
headers: HeaderMap,
133
+
) -> Result<ProxyResponse, KnotProxyError> {
134
+
let keyed: Vec<(&str, &str)> = query
135
+
.iter()
136
+
.copied()
137
+
.filter(|(k, _)| *k != REPO_PARAM)
138
+
.chain(std::iter::once((REPO_PARAM, repo.as_ref())))
139
+
.collect();
140
+
self.proxy
141
+
.forward(&self.host, &nsid.0, &keyed, headers)
142
+
.await
143
+
}
144
+
}
145
+
146
+
fn beyond_origin(url: &Url) -> Option<String> {
147
+
[
148
+
(!matches!(url.path(), "" | "/")).then(|| format!("the path {}", url.path())),
149
+
url.query().map(|query| format!("the query {query}")),
150
+
url.fragment()
151
+
.map(|fragment| format!("the fragment {fragment}")),
152
+
(!url.username().is_empty()).then(|| format!("the username {}", url.username())),
153
+
url.password().map(|_| "a password".to_owned()),
154
+
]
155
+
.into_iter()
156
+
.flatten()
157
+
.next()
158
+
}
159
+
160
+
#[cfg(test)]
161
+
mod tests {
162
+
use super::*;
163
+
use bobbin_runtime::SystemClock;
164
+
use wiremock::matchers::{method, path, query_param};
165
+
use wiremock::{Mock, MockServer, ResponseTemplate};
166
+
167
+
fn mirror_at(raw: &str) -> Result<MirrorProxy, MirrorProxyError> {
168
+
MirrorProxy::new(
169
+
&Url::parse(raw).unwrap(),
170
+
Arc::new(SystemClock::new()),
171
+
RuntimeHasher::default(),
172
+
)
173
+
}
174
+
175
+
#[test]
176
+
fn a_mirror_url_with_more_than_an_origin_is_refused_without_echoing_a_password() {
177
+
[
178
+
"https://nel.pet/api",
179
+
"https://nel.pet/?a=1",
180
+
"https://nel.pet/#x",
181
+
"https://nel@nel.pet/",
182
+
"https://nel:hunter2@nel.pet/",
183
+
"https://:hunter2@nel.pet/",
184
+
]
185
+
.iter()
186
+
.for_each(|raw| {
187
+
let Err(err) = mirror_at(raw) else {
188
+
panic!("{raw} must error out, since KnotHost::parse would truncate it");
189
+
};
190
+
assert!(
191
+
matches!(err, MirrorProxyError::NotAnOrigin(_)),
192
+
"{raw} gave {err}",
193
+
);
194
+
assert!(
195
+
!err.to_string().contains("hunter2"),
196
+
"an operator's password must stay out of the log, got {err}",
197
+
);
198
+
});
199
+
assert!(
200
+
mirror_at("https://nel.pet/").is_ok(),
201
+
"a bare origin is what an operator will configure",
202
+
);
203
+
}
204
+
205
+
#[tokio::test]
206
+
async fn forward_replaces_a_slug_the_caller_left_in_the_query() {
207
+
let server = MockServer::start().await;
208
+
Mock::given(method("GET"))
209
+
.and(path("/xrpc/sh.tangled.git.temp.listTags"))
210
+
.and(query_param("repo", "did:plc:conch"))
211
+
.respond_with(ResponseTemplate::new(200).set_body_string("[]"))
212
+
.mount(&server)
213
+
.await;
214
+
215
+
let mirror = mirror_at(&server.uri()).unwrap();
216
+
let nsid = MirrorNsid::route("sh.tangled.repo.tags", &[]).unwrap();
217
+
let resp = mirror
218
+
.forward(
219
+
&nsid,
220
+
&Did::<DefaultStr>::new_static("did:plc:conch").unwrap(),
221
+
&[("repo", "did:plc:nel/scallop")],
222
+
HeaderMap::new(),
223
+
)
224
+
.await
225
+
.expect("the mounted mirror answers");
226
+
assert_eq!(resp.status(), 200);
227
+
}
228
+
229
+
#[test]
230
+
fn a_mirror_dials_where_the_knot_policy_refuses() {
231
+
let mirror = mirror_at("http://127.0.0.1:9/").expect("an operator picks the mirror");
232
+
assert!(
233
+
mirror.proxy.allows_private_hosts() && !mirror.proxy.requires_https(),
234
+
"the knot policy doesn't bind an operator-configured mirror",
235
+
);
236
+
}
237
+
}