+2

knot2/README.md

View file Reviewed

··· 193 193 194 194 Set `xrpc.trusted_proxy_header = "x-forwarded-for"` when doing this, otherwise every client looks like it comes from the proxy and the ratelimiter wil treat them as one very busy mister. Only set it behind a proxy the operator controls, since a direct client can like, invent that header. 195 195 196 196 + Add `xrpc.trusted_proxies = ["fd00:1::4", "10.89.0.4"]` for example, one entry per address that the proxy connects from, so the knot honors that header from the proxy alone & ratelimits anyone else by the address they connected from. 197 197 + 196 198 The knot can also terminate TLS itself (and that's the only way to get its HTTP3 support) because a plain TCP frontend can't proxy QUIC. Using a certificate the operator already manages: 197 199 198 200 ```toml

+6 -4

knot2/crates/knot-bench/benches/pack.rs

View file Reviewed

··· 202 202 fn archive(bencher: Bencher, commits: u32) { 203 203 let history = build_history(spec_for(commits)); 204 204 let request = build_archive_request(history.tip()); 205 205 - let bytes = upload_archive(history.repo(), &request).unwrap().len(); 206 206 - bencher 207 207 - .counter(BytesCount::new(bytes)) 208 208 - .bench_local(|| upload_archive(history.repo(), &request).unwrap()); 205 205 + let bytes = upload_archive(history.repo(), &request, knot_git::ArchiveLimit::default()) 206 206 + .unwrap() 207 207 + .len(); 208 208 + bencher.counter(BytesCount::new(bytes)).bench_local(|| { 209 209 + upload_archive(history.repo(), &request, knot_git::ArchiveLimit::default()).unwrap() 210 210 + }); 209 211 } 210 212 211 213 struct FreshTarget {

+54 -1

knot2/crates/knot-config/src/lib.rs

View file Reviewed

··· 1 1 use std::fmt; 2 2 - use std::net::SocketAddr; 2 2 + use std::net::{AddrParseError, IpAddr, SocketAddr}; 3 3 use std::path::{Path, PathBuf}; 4 4 use std::sync::OnceLock; 5 5 use std::time::Duration; ··· 237 237 #[config(env = "KNOT_XRPC_MAX_RESPONSE_BYTES", default = 5_242_880)] 238 238 pub max_response_bytes: u64, 239 239 240 240 + /// Upper bound on bytes that a single archive spools, 241 241 + /// across all our surfaces: the sh.tangled.repo.archive query, 242 242 + /// `git archive --remote` over SSH, 243 243 + /// and the smart HTTP archive route. 244 244 + /// The knot will refuse writing smth that would blast an archive past this bound. 240 245 #[config(env = "KNOT_XRPC_MAX_ARCHIVE_BYTES", default = 1_073_741_824)] 241 246 pub max_archive_bytes: u64, 242 247 ··· 302 307 /// can forge it otherwise. 303 308 #[config(env = "KNOT_XRPC_TRUSTED_PROXY_HEADER")] 304 309 pub trusted_proxy_header: Option<String>, 310 310 + 311 311 + /// IP addresses whose `trusted_proxy_header` the knot honors, 312 312 + /// without a port, 313 313 + /// for ex the loopback address of a reverse proxy on the same host. 314 314 + /// The knot rate-limits a request from any other address 315 315 + /// by its own socket address and ignores the header. 316 316 + /// Leave empty to honor the header from every peer, 317 317 + /// which is safe *only* if nothing but the proxy can reach this knot. 318 318 + #[config( 319 319 + env = "KNOT_XRPC_TRUSTED_PROXIES", 320 320 + parse_env = parse_trusted_proxies, 321 321 + default = [] 322 322 + )] 323 323 + pub trusted_proxies: Vec<IpAddr>, 305 324 306 325 #[config(env = "KNOT_XRPC_EVENTS_REPLAY_BUFFER", default = 4096)] 307 326 pub events_replay_buffer: u32, ··· 420 439 .collect() 421 440 } 422 441 442 442 + fn parse_trusted_proxies(raw: &str) -> Result<Vec<IpAddr>, AddrParseError> { 443 443 + raw.split(',') 444 444 + .map(str::trim) 445 445 + .filter(|item| !item.is_empty()) 446 446 + .map(str::parse) 447 447 + .collect() 448 448 + } 449 449 + 423 450 impl KnotConfig { 424 451 pub fn object_format(&self) -> Option<knot_types::ObjectFormat> { 425 452 knot_types::ObjectFormat::from_capability(&self.git.object_format) ··· 803 830 .as_ref() 804 831 .filter(|header| !is_http_token(header)) 805 832 .map(|_| "xrpc.trusted_proxy_header isn't valid HTTP header name".to_string()), 833 833 + check( 834 834 + self.xrpc.trusted_proxy_header.is_some() || self.xrpc.trusted_proxies.is_empty(), 835 835 + "xrpc.trusted_proxies needs xrpc.trusted_proxy_header, the header the knot honors from those addresses", 836 836 + ), 806 837 self.acl 807 838 .legacy_admin_secret_env 808 839 .as_deref() ··· 1188 1219 fork_max_pack_bytes: 1_073_741_824, 1189 1220 fork_fetch_timeout_ms: 600_000, 1190 1221 trusted_proxy_header: None, 1222 1222 + trusted_proxies: Vec::new(), 1191 1223 events_replay_buffer: 4_096, 1192 1224 events_replay_bytes: 67_108_864, 1193 1225 events_max_subscribers: 256, ··· 1542 1574 |config| config.homepage.path = Some(PathBuf::from("homepage.html")), 1543 1575 "homepage.path must be absolute path", 1544 1576 ), 1577 1577 + ( 1578 1578 + "trusted_proxies_without_the_header_the_knot_honors", 1579 1579 + |config| config.xrpc.trusted_proxies = vec!["127.0.0.1".parse().unwrap()], 1580 1580 + "needs xrpc.trusted_proxy_header", 1581 1581 + ), 1545 1582 ]; 1546 1583 cases.iter().for_each(|(label, mutate, expected)| { 1547 1584 let mut config = sample(); ··· 1608 1645 let parsed = parse_admins("did:plc:nel, did:plc:olaren").unwrap(); 1609 1646 assert_eq!(parsed.len(), 2); 1610 1647 assert!(parse_admins("not-a-did").is_err()); 1648 1648 + } 1649 1649 + 1650 1650 + #[test] 1651 1651 + fn trusted_proxies_parse_from_comma_separated_env() { 1652 1652 + assert_eq!( 1653 1653 + parse_trusted_proxies("127.0.0.1, ::1").unwrap(), 1654 1654 + vec![ 1655 1655 + "127.0.0.1".parse::<IpAddr>().unwrap(), 1656 1656 + "::1".parse::<IpAddr>().unwrap() 1657 1657 + ] 1658 1658 + ); 1659 1659 + assert!(parse_trusted_proxies("").unwrap().is_empty()); 1660 1660 + assert!( 1661 1661 + parse_trusted_proxies("127.0.0.1:5555").is_err(), 1662 1662 + "xrpc.trusted_proxies takes bare IP addresses, so a port must fail to parse" 1663 1663 + ); 1611 1664 } 1612 1665 1613 1666 #[test]

+2 -2

knot2/crates/knot-edge/src/lib.rs

View file Reviewed

··· 290 290 RequestTimeout::from_millis(NonZeroU64::new(30_000).unwrap()), 291 291 BodyInactivityTimeout::from_millis(NonZeroU64::new(30_000).unwrap()), 292 292 WriteRequestTimeout::from_millis(NonZeroU64::new(1_800_000).unwrap()), 293 293 - None, 293 293 + knot_types::ProxyTrust::default(), 294 294 ) 295 295 .prepare(&CancellationToken::new()) 296 296 } ··· 312 312 RequestTimeout::from_millis(NonZeroU64::new(30_000).unwrap()), 313 313 BodyInactivityTimeout::from_millis(NonZeroU64::new(30_000).unwrap()), 314 314 WriteRequestTimeout::from_millis(NonZeroU64::new(1_800_000).unwrap()), 315 315 - None, 315 315 + knot_types::ProxyTrust::default(), 316 316 ) 317 317 .prepare(&CancellationToken::new()) 318 318 }

+170 -39

knot2/crates/knot-edge/src/robustness.rs

View file Reviewed

··· 1 1 use std::net::{IpAddr, SocketAddr}; 2 2 use std::num::{NonZeroU32, NonZeroU64}; 3 3 - use std::sync::Arc; 3 3 + use std::sync::{Arc, OnceLock}; 4 4 use std::time::Duration; 5 5 6 6 use axum::Router; ··· 10 10 use axum::middleware::{Next, from_fn_with_state}; 11 11 use axum::response::{IntoResponse, Response}; 12 12 use governor::middleware::NoOpMiddleware; 13 13 - use http::{HeaderName, Method, Request, StatusCode}; 13 13 + use http::{Method, Request, StatusCode}; 14 14 + use knot_types::ProxyTrust; 14 15 use tokio_util::sync::CancellationToken; 15 16 use tower::limit::GlobalConcurrencyLimitLayer; 16 17 use tower::load_shed::LoadShedLayer; ··· 77 78 request_timeout: RequestTimeout, 78 79 body_timeout: BodyInactivityTimeout, 79 80 write_request_timeout: WriteRequestTimeout, 80 80 - proxy_header: Option<HeaderName>, 81 81 + proxy_trust: ProxyTrust, 81 82 } 82 83 83 84 impl EdgeGuards { ··· 88 89 request_timeout: RequestTimeout, 89 90 body_timeout: BodyInactivityTimeout, 90 91 write_request_timeout: WriteRequestTimeout, 91 91 - proxy_header: Option<HeaderName>, 92 92 + proxy_trust: ProxyTrust, 92 93 ) -> Self { 93 94 Self { 94 95 rate, ··· 97 98 request_timeout, 98 99 body_timeout, 99 100 write_request_timeout, 100 100 - proxy_header, 101 101 + proxy_trust, 101 102 } 102 103 } 103 104 104 105 pub(crate) fn prepare(self, shutdown: &CancellationToken) -> GuardLayers { 105 105 - let governor = build_governor(self.rate, self.burst, self.proxy_header); 106 106 + let governor = build_governor(self.rate, self.burst, self.proxy_trust); 106 107 spawn_state_cleanup(Arc::clone(&governor), shutdown.clone()); 107 108 GuardLayers { 108 109 governor, ··· 129 130 struct TimeoutBudget { 130 131 standard: Duration, 131 132 extended: Duration, 133 133 + } 134 134 + 135 135 + #[derive(Clone, Default)] 136 136 + struct IgnoredHeaderNotice(Arc<OnceLock<IpAddr>>); 137 137 + 138 138 + impl IgnoredHeaderNotice { 139 139 + fn report(&self, peer: IpAddr) { 140 140 + if self.0.set(peer).is_ok() { 141 141 + tracing::warn!( 142 142 + %peer, 143 143 + "a peer outside xrpc.trusted_proxies sent xrpc.trusted_proxy_header, so the knot ignored the header and rate-limits that peer by the address it connected from. Add this address to xrpc.trusted_proxies if it is the reverse proxy, since a proxy reaches the knot over one address family and listing the other one silently loses the header. This warning reports the first such peer only." 144 144 + ); 145 145 + } 146 146 + } 132 147 } 133 148 134 149 #[derive(Clone)] 135 150 struct ProxyAwareIp { 136 136 - header: Option<HeaderName>, 151 151 + trust: ProxyTrust, 152 152 + ignored_header: IgnoredHeaderNotice, 137 153 } 138 154 139 155 impl KeyExtractor for ProxyAwareIp { 140 156 type Key = IpAddr; 141 157 142 158 fn extract<T>(&self, request: &Request<T>) -> Result<Self::Key, GovernorError> { 143 143 - let from_header = self 144 144 - .header 145 145 - .as_ref() 146 146 - .and_then(|header| knot_types::forwarded_peer(request.headers(), header)); 147 147 - from_header 148 148 - .or_else(|| { 149 149 - request 150 150 - .extensions() 151 151 - .get::<ConnectInfo<SocketAddr>>() 152 152 - .map(|info| info.0.ip()) 153 153 - }) 154 154 - .ok_or(GovernorError::UnableToExtractKey) 159 159 + let socket = request 160 160 + .extensions() 161 161 + .get::<ConnectInfo<SocketAddr>>() 162 162 + .map(|info| info.0.ip()); 163 163 + let key = self.trust.peer_key(request.headers(), socket); 164 164 + if let Some(peer) = key.ignored_header() { 165 165 + self.ignored_header.report(peer); 166 166 + } 167 167 + key.address().ok_or(GovernorError::UnableToExtractKey) 155 168 } 156 169 } 157 170 158 171 fn build_governor( 159 172 rate: RequestsPerSecond, 160 173 burst: BurstSize, 161 161 - proxy_header: Option<HeaderName>, 174 174 + proxy_trust: ProxyTrust, 162 175 ) -> Arc<GuardGovernor> { 163 176 let mut builder = GovernorConfigBuilder::default(); 164 177 builder.period(rate.period()).burst_size(burst.0.get()); 165 178 let config = builder 166 179 .key_extractor(ProxyAwareIp { 167 167 - header: proxy_header, 180 180 + trust: proxy_trust, 181 181 + ignored_header: IgnoredHeaderNotice::default(), 168 182 }) 169 183 .finish() 170 184 .expect("a non-zero rate period and burst size always yield a governor config"); ··· 257 271 inflight: u32, 258 272 request_timeout_ms: u64, 259 273 body_timeout_ms: u64, 260 260 - proxy_header: Option<&str>, 274 274 + proxy_trust: ProxyTrust, 261 275 ) -> EdgeGuards { 262 276 guards_with_write( 263 277 rate, ··· 266 280 request_timeout_ms, 267 281 body_timeout_ms, 268 282 request_timeout_ms, 269 269 - proxy_header, 283 283 + proxy_trust, 270 284 ) 271 285 } 272 286 ··· 278 292 request_timeout_ms: u64, 279 293 body_timeout_ms: u64, 280 294 write_request_timeout_ms: u64, 281 281 - proxy_header: Option<&str>, 295 295 + proxy_trust: ProxyTrust, 282 296 ) -> EdgeGuards { 283 297 EdgeGuards::new( 284 298 RequestsPerSecond::new(NonZeroU32::new(rate).unwrap()), ··· 287 301 RequestTimeout::from_millis(NonZeroU64::new(request_timeout_ms).unwrap()), 288 302 BodyInactivityTimeout::from_millis(NonZeroU64::new(body_timeout_ms).unwrap()), 289 303 WriteRequestTimeout::from_millis(NonZeroU64::new(write_request_timeout_ms).unwrap()), 290 290 - proxy_header.map(|header| HeaderName::from_bytes(header.as_bytes()).unwrap()), 304 304 + proxy_trust, 305 305 + ) 306 306 + } 307 307 + 308 308 + fn forwarded_for() -> http::HeaderName { 309 309 + http::HeaderName::from_static("x-forwarded-for") 310 310 + } 311 311 + 312 312 + fn trusting_any_peer() -> ProxyTrust { 313 313 + ProxyTrust::new(Some(forwarded_for()), knot_types::TrustedProxies::default()) 314 314 + } 315 315 + 316 316 + fn trusting_loopback() -> ProxyTrust { 317 317 + ProxyTrust::new( 318 318 + Some(forwarded_for()), 319 319 + knot_types::TrustedProxies::new(["127.0.0.1".parse::<IpAddr>().unwrap()]), 291 320 ) 292 321 } 293 322 323 323 + fn extractor(trust: ProxyTrust) -> ProxyAwareIp { 324 324 + ProxyAwareIp { 325 325 + trust, 326 326 + ignored_header: IgnoredHeaderNotice::default(), 327 327 + } 328 328 + } 329 329 + 294 330 fn guarded_router(router: Router, guards: EdgeGuards) -> Router { 295 331 apply(router, guards.prepare(&CancellationToken::new())) 296 332 } ··· 309 345 310 346 #[test] 311 347 fn the_extractor_keys_on_the_trusted_proxy_header_when_configured() { 312 312 - let extractor = ProxyAwareIp { 313 313 - header: Some(HeaderName::from_static("x-forwarded-for")), 314 314 - }; 348 348 + let extractor = extractor(trusting_any_peer()); 315 349 let request = Request::get("/") 316 350 .header("x-forwarded-for", "203.0.113.7, 198.51.100.4") 317 351 .extension(ConnectInfo(SocketAddr::from(([127, 0, 0, 1], 5000)))) ··· 326 360 327 361 #[test] 328 362 fn the_extractor_ignores_a_forgeable_header_when_no_proxy_is_trusted() { 329 329 - let extractor = ProxyAwareIp { header: None }; 363 363 + let extractor = extractor(ProxyTrust::default()); 330 364 let request = Request::get("/") 331 365 .header("x-forwarded-for", "203.0.113.7") 332 366 .extension(ConnectInfo(SocketAddr::from(([10, 0, 0, 9], 5000)))) ··· 340 374 } 341 375 342 376 #[test] 377 377 + fn the_extractor_keys_an_unlisted_peer_on_its_socket_however_it_fills_the_header() { 378 378 + let extractor = extractor(trusting_loopback()); 379 379 + let forged = |host| { 380 380 + Request::get("/") 381 381 + .header("x-forwarded-for", "198.51.100.4") 382 382 + .extension(ConnectInfo(SocketAddr::from(([127, 0, 0, host], 5000)))) 383 383 + .body(()) 384 384 + .unwrap() 385 385 + }; 386 386 + assert_eq!( 387 387 + extractor.extract(&forged(1)).unwrap(), 388 388 + "198.51.100.4".parse::<IpAddr>().unwrap(), 389 389 + "the listed proxy relayed this one, so the extractor keys on the header address" 390 390 + ); 391 391 + assert_eq!( 392 392 + extractor.extract(&forged(9)).unwrap(), 393 393 + "127.0.0.9".parse::<IpAddr>().unwrap(), 394 394 + "an unlisted peer picked its own token bucket by forging the header" 395 395 + ); 396 396 + } 397 397 + 398 398 + #[test] 343 399 fn the_extractor_falls_back_to_the_peer_when_the_trusted_header_is_absent() { 344 344 - let extractor = ProxyAwareIp { 345 345 - header: Some(HeaderName::from_static("x-forwarded-for")), 346 346 - }; 400 400 + let extractor = extractor(trusting_any_peer()); 347 401 let request = Request::get("/") 348 402 .extension(ConnectInfo(SocketAddr::from(([10, 0, 0, 9], 5000)))) 349 403 .body(()) ··· 356 410 357 411 #[test] 358 412 fn the_extractor_fails_when_no_peer_can_be_identified() { 359 359 - let extractor = ProxyAwareIp { header: None }; 413 413 + let extractor = extractor(ProxyTrust::default()); 360 414 let request = Request::get("/").body(()).unwrap(); 361 415 assert!(matches!( 362 416 extractor.extract(&request), ··· 364 418 )); 365 419 } 366 420 421 421 + #[test] 422 422 + fn a_missing_connect_info_fails_closed_rather_than_taking_the_header_on_trust() { 423 423 + let listed = extractor(trusting_loopback()); 424 424 + let headed = || { 425 425 + Request::get("/") 426 426 + .header("x-forwarded-for", "198.51.100.4") 427 427 + .body(()) 428 428 + .unwrap() 429 429 + }; 430 430 + assert!( 431 431 + matches!( 432 432 + listed.extract(&headed()), 433 433 + Err(GovernorError::UnableToExtractKey) 434 434 + ), 435 435 + "with no socket to match against the allowlist the extractor identifies no client" 436 436 + ); 437 437 + assert_eq!( 438 438 + extractor(trusting_any_peer()).extract(&headed()).unwrap(), 439 439 + "198.51.100.4".parse::<IpAddr>().unwrap(), 440 440 + "an operator who lists no proxy already told the knot to take the header from anyone" 441 441 + ); 442 442 + } 443 443 + 444 444 + #[test] 445 445 + fn the_first_unlisted_peer_sending_the_header_is_reported_once() { 446 446 + let extractor = extractor(trusting_loopback()); 447 447 + let forged = |host| { 448 448 + Request::get("/") 449 449 + .header("x-forwarded-for", "198.51.100.4") 450 450 + .extension(ConnectInfo(SocketAddr::from(([203, 0, 113, host], 5000)))) 451 451 + .body(()) 452 452 + .unwrap() 453 453 + }; 454 454 + extractor.extract(&forged(7)).unwrap(); 455 455 + extractor.extract(&forged(9)).unwrap(); 456 456 + assert_eq!( 457 457 + extractor.ignored_header.0.get(), 458 458 + Some(&"203.0.113.7".parse::<IpAddr>().unwrap()), 459 459 + "a wrong-family allowlist sends every request down this path, so only the first peer is reported" 460 460 + ); 461 461 + } 462 462 + 463 463 + #[test] 464 464 + fn a_listed_proxy_and_a_headerless_request_report_nothing() { 465 465 + let listed = extractor(trusting_loopback()); 466 466 + listed 467 467 + .extract( 468 468 + &Request::get("/") 469 469 + .header("x-forwarded-for", "198.51.100.4") 470 470 + .extension(ConnectInfo(SocketAddr::from(([127, 0, 0, 1], 5000)))) 471 471 + .body(()) 472 472 + .unwrap(), 473 473 + ) 474 474 + .unwrap(); 475 475 + listed 476 476 + .extract( 477 477 + &Request::get("/") 478 478 + .extension(ConnectInfo(SocketAddr::from(([203, 0, 113, 7], 5000)))) 479 479 + .body(()) 480 480 + .unwrap(), 481 481 + ) 482 482 + .unwrap(); 483 483 + assert_eq!( 484 484 + listed.ignored_header.0.get(), 485 485 + None, 486 486 + "neither a relayed request or a request without the header says anything about the allowlist" 487 487 + ); 488 488 + } 489 489 + 367 490 #[tokio::test] 368 491 async fn a_well_behaved_request_passes_every_guard() { 369 492 let app = guarded_router( 370 493 Router::new().route("/", get(|| async { "ok" })), 371 371 - guards(50, 200, 1_024, 60_000, 30_000, None), 494 494 + guards(50, 200, 1_024, 60_000, 30_000, ProxyTrust::default()), 372 495 ); 373 496 let status = app 374 497 .oneshot(from_peer(get_request(), 1)) ··· 382 505 async fn a_burst_beyond_the_per_ip_limit_is_rejected_with_429() { 383 506 let app = guarded_router( 384 507 Router::new().route("/", get(|| async { "ok" })), 385 385 - guards(1, 2, 1_024, 60_000, 30_000, None), 508 508 + guards(1, 2, 1_024, 60_000, 30_000, ProxyTrust::default()), 386 509 ); 387 510 let first = app 388 511 .clone() ··· 432 555 "ok" 433 556 }), 434 557 ), 435 435 - guards(10_000, 10_000, 1, 60_000, 30_000, None), 558 558 + guards(10_000, 10_000, 1, 60_000, 30_000, ProxyTrust::default()), 436 559 ); 437 560 let holder = { 438 561 let app = app.clone(); ··· 469 592 "ok" 470 593 }), 471 594 ), 472 472 - guards(10_000, 10_000, 1_024, 80, 30_000, None), 595 595 + guards(10_000, 10_000, 1_024, 80, 30_000, ProxyTrust::default()), 473 596 ); 474 597 let status = app 475 598 .oneshot(from_peer( ··· 500 623 "ok" 501 624 }), 502 625 ), 503 503 - guards_with_write(10_000, 10_000, 1_024, 80, 30_000, 5_000, None), 626 626 + guards_with_write( 627 627 + 10_000, 628 628 + 10_000, 629 629 + 1_024, 630 630 + 80, 631 631 + 30_000, 632 632 + 5_000, 633 633 + ProxyTrust::default(), 634 634 + ), 504 635 ); 505 636 let push = app 506 637 .clone() ··· 539 670 async fn a_stalled_request_body_is_cut_and_never_hangs() { 540 671 let app = guarded_router( 541 672 Router::new().route("/upload", post(|_body: Bytes| async { "ok" })), 542 542 - guards(10_000, 10_000, 1_024, 60_000, 80, None), 673 673 + guards(10_000, 10_000, 1_024, 60_000, 80, ProxyTrust::default()), 543 674 ); 544 675 let body = Body::from_stream( 545 676 futures::stream::once(async {

+157 -15

knot2/crates/knot-git/src/archive.rs

View file Reviewed

··· 1 1 + use std::io::{Seek, SeekFrom, Write}; 1 2 use std::sync::atomic::AtomicBool; 2 3 3 4 use gix::bstr::BString; 4 5 use knot_types::{Oid, ParseError}; 5 6 6 7 use crate::error::{GitError, backend}; 8 8 + use crate::objects::MAX_TREE_DEPTH; 7 9 use crate::repo::Repo; 10 10 + 11 11 + const TAR_BLOCK: u64 = 512; 12 12 + 13 13 + knot_types::scalar_newtype! { 14 14 + pub struct ArchiveLimit(u64); 15 15 + } 16 16 + 17 17 + impl Default for ArchiveLimit { 18 18 + fn default() -> Self { 19 19 + Self::new(1024 * 1024 * 1024) 20 20 + } 21 21 + } 8 22 9 23 #[derive(Debug, Clone, Copy, PartialEq, Eq)] 10 24 pub enum ArchiveFormat { ··· 65 79 tree: Oid, 66 80 format: ArchiveFormat, 67 81 prefix: Option<&ArchivePrefix>, 68 68 - mut out: impl std::io::Write + std::io::Seek, 82 82 + limit: ArchiveLimit, 83 83 + out: impl std::io::Write + std::io::Seek, 69 84 ) -> Result<(), GitError> { 85 85 + self.bound_archive_source(tree.object_id(), limit, MAX_TREE_DEPTH, &mut 0)?; 70 86 let (stream, _index) = self 71 87 .git() 72 88 .worktree_stream(tree.object_id()) 73 89 .map_err(backend)?; 74 90 let interrupt = AtomicBool::new(false); 75 75 - self.git() 76 76 - .worktree_archive( 77 77 - stream, 78 78 - &mut out, 79 79 - gix::progress::Discard, 80 80 - &interrupt, 81 81 - gix_archive::Options { 82 82 - format: format.gix(), 83 83 - tree_prefix: prefix.map(|prefix| BString::from(prefix.as_str())), 84 84 - modification_time: 0, 85 85 - }, 86 86 - ) 87 87 - .map_err(backend) 91 91 + let mut spool = BoundedSpool { 92 92 + inner: out, 93 93 + position: 0, 94 94 + limit, 95 95 + overflowed: false, 96 96 + }; 97 97 + let written = self.git().worktree_archive( 98 98 + stream, 99 99 + &mut spool, 100 100 + gix::progress::Discard, 101 101 + &interrupt, 102 102 + gix_archive::Options { 103 103 + format: format.gix(), 104 104 + tree_prefix: prefix.map(|prefix| BString::from(prefix.as_str())), 105 105 + modification_time: 0, 106 106 + }, 107 107 + ); 108 108 + match (written, spool.overflowed) { 109 109 + (_, true) => Err(GitError::ArchiveTooLarge { limit }), 110 110 + (Ok(()), false) => Ok(()), 111 111 + (Err(error), false) => Err(backend(error)), 112 112 + } 113 113 + } 114 114 + 115 115 + fn bound_archive_source( 116 116 + &self, 117 117 + tree: gix::ObjectId, 118 118 + limit: ArchiveLimit, 119 119 + nesting: usize, 120 120 + spooled: &mut u64, 121 121 + ) -> Result<(), GitError> { 122 122 + if nesting == 0 { 123 123 + return Err(GitError::DepthExceeded("tree nesting")); 124 124 + } 125 125 + if tree == gix::ObjectId::empty_tree(self.git().object_hash()) { 126 126 + return Ok(()); 127 127 + } 128 128 + let object = self.git().find_tree(tree).map_err(backend)?; 129 129 + let decoded = object 130 130 + .decode() 131 131 + .map_err(|error| GitError::Decode(error.to_string()))?; 132 132 + decoded.entries.iter().try_for_each(|entry| { 133 133 + let oid = entry.oid.to_owned(); 134 134 + *spooled = spooled.saturating_add(TAR_BLOCK); 135 135 + match entry.mode.kind() { 136 136 + _ if *spooled > limit.get() => Err(GitError::ArchiveTooLarge { limit }), 137 137 + gix::objs::tree::EntryKind::Commit => Ok(()), 138 138 + gix::objs::tree::EntryKind::Tree => { 139 139 + self.bound_archive_source(oid, limit, nesting - 1, spooled) 140 140 + } 141 141 + _ => { 142 142 + let content = self.blob_size(Oid::from(oid))?; 143 143 + *spooled = spooled.saturating_add(content.next_multiple_of(TAR_BLOCK)); 144 144 + match *spooled > limit.get() { 145 145 + true => Err(GitError::ArchiveTooLarge { limit }), 146 146 + false => Ok(()), 147 147 + } 148 148 + } 149 149 + } 150 150 + }) 151 151 + } 152 152 + } 153 153 + 154 154 + struct BoundedSpool<W> { 155 155 + inner: W, 156 156 + position: u64, 157 157 + limit: ArchiveLimit, 158 158 + overflowed: bool, 159 159 + } 160 160 + 161 161 + impl<W: Write> Write for BoundedSpool<W> { 162 162 + fn write(&mut self, data: &[u8]) -> std::io::Result<usize> { 163 163 + let remaining = self.limit.get().saturating_sub(self.position); 164 164 + if data.len() as u64 > remaining { 165 165 + self.overflowed = true; 166 166 + return Err(std::io::Error::from(std::io::ErrorKind::WriteZero)); 167 167 + } 168 168 + let written = self.inner.write(data)?; 169 169 + self.position = self.position.saturating_add(written as u64); 170 170 + Ok(written) 171 171 + } 172 172 + 173 173 + fn flush(&mut self) -> std::io::Result<()> { 174 174 + self.inner.flush() 175 175 + } 176 176 + } 177 177 + 178 178 + impl<W: Seek> Seek for BoundedSpool<W> { 179 179 + fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> { 180 180 + let position = self.inner.seek(pos)?; 181 181 + self.position = position; 182 182 + Ok(position) 88 183 } 89 184 } 90 185 91 186 #[cfg(test)] 92 187 mod tests { 93 93 - use super::ArchivePrefix; 188 188 + use super::{ArchiveLimit, ArchivePrefix, BoundedSpool}; 189 189 + use std::io::{Seek, SeekFrom, Write}; 190 190 + 191 191 + fn spool(limit: u64) -> BoundedSpool<std::io::Cursor<Vec<u8>>> { 192 192 + BoundedSpool { 193 193 + inner: std::io::Cursor::new(Vec::new()), 194 194 + position: 0, 195 195 + limit: ArchiveLimit::new(limit), 196 196 + overflowed: false, 197 197 + } 198 198 + } 199 199 + 200 200 + #[test] 201 201 + fn the_spool_refuses_the_write_that_would_pass_the_limit() { 202 202 + let mut spool = spool(8); 203 203 + assert!(spool.write_all(b"12345678").is_ok()); 204 204 + assert!(!spool.overflowed); 205 205 + assert!(spool.write_all(b"9").is_err()); 206 206 + assert!(spool.overflowed); 207 207 + assert_eq!( 208 208 + spool.inner.into_inner(), 209 209 + b"12345678", 210 210 + "the refused write never reaches the inner writer" 211 211 + ); 212 212 + } 213 213 + 214 214 + #[test] 215 215 + fn a_seek_backwards_re_credits_the_budget_the_zip_writer_rewinds_over() { 216 216 + let mut spool = spool(8); 217 217 + spool.write_all(b"12345678").unwrap(); 218 218 + spool.seek(SeekFrom::Start(4)).unwrap(); 219 219 + assert_eq!(spool.position, 4); 220 220 + spool 221 221 + .write_all(b"abcd") 222 222 + .expect("rewriting bytes already counted stays within the limit"); 223 223 + assert!(!spool.overflowed); 224 224 + } 225 225 + 226 226 + #[test] 227 227 + fn a_write_whose_length_would_overflow_the_position_is_refused() { 228 228 + let mut spool = spool(u64::MAX); 229 229 + spool.position = u64::MAX; 230 230 + assert!( 231 231 + spool.write_all(b"1").is_err(), 232 232 + "the position saturates at u64::MAX, so the spool must refuse the write" 233 233 + ); 234 234 + assert!(spool.overflowed); 235 235 + } 94 236 95 237 #[test] 96 238 fn a_plain_nested_prefix_is_accepted() {

+2

knot2/crates/knot-git/src/error.rs

View file Reviewed

··· 39 39 ReservedDid(String), 40 40 #[error("{0} exceeds maximum supported depth")] 41 41 DepthExceeded(&'static str), 42 42 + #[error("archive exceeds the {} byte limit", limit.get())] 43 43 + ArchiveTooLarge { limit: crate::ArchiveLimit }, 42 44 #[error("revision walk: {0}")] 43 45 RevWalk(String), 44 46 #[error("upload-pack selection exceeded its {0}")]

+1 -1

knot2/crates/knot-git/src/lib.rs

View file Reviewed

··· 12 12 mod repo; 13 13 mod staging; 14 14 15 15 - pub use archive::{ArchiveFormat, ArchivePrefix}; 15 15 + pub use archive::{ArchiveFormat, ArchiveLimit, ArchivePrefix}; 16 16 pub use bitmap::{reachable_via_bitmap, verbatim_clone_pack, write_bitmap, write_midx_bitmap}; 17 17 pub use error::{GitError, SelectionLimit}; 18 18 pub use maintenance::{PackRefsReport, ReflogReport};

+43 -3

knot2/crates/knot-git/tests/config_isolation.rs

View file Reviewed

··· 1 1 use std::path::Path; 2 2 use std::sync::atomic::AtomicBool; 3 3 4 4 - use knot_git::ArchiveFormat; 4 4 + use knot_git::{ArchiveFormat, ArchiveLimit}; 5 5 use knot_types::Oid; 6 6 7 7 mod common; ··· 80 80 let bare = layout.open(&did).unwrap(); 81 81 let tree = bare.peel_to_tree(head).unwrap(); 82 82 let mut out = std::io::Cursor::new(Vec::new()); 83 83 - bare.write_archive(tree, ArchiveFormat::Tar, None, &mut out) 84 84 - .unwrap(); 83 83 + bare.write_archive( 84 84 + tree, 85 85 + ArchiveFormat::Tar, 86 86 + None, 87 87 + ArchiveLimit::new(u64::MAX), 88 88 + &mut out, 89 89 + ) 90 90 + .unwrap(); 85 91 let served = out.into_inner(); 86 92 87 93 assert!( ··· 100 106 "the knot ran a filter driver defined by config outside the repository" 101 107 ); 102 108 } 109 109 + 110 110 + #[test] 111 111 + fn a_pushed_replace_ref_never_substitutes_an_object_the_knot_reads() { 112 112 + let (_scan, work_dir, layout, did) = seeded(); 113 113 + let work = work_dir.path(); 114 114 + let bare_path = layout.repo_path(&did).unwrap(); 115 115 + 116 116 + commit_file(work, "payload.txt", "kelp\n", "seed"); 117 117 + git(work, &["push", "-q", bare_path.to_str().unwrap(), "main"]); 118 118 + let original = Oid::from_hex(&git(work, &["rev-parse", "HEAD:payload.txt"])).unwrap(); 119 119 + commit_file(work, "payload.txt", "pwned\n", "second"); 120 120 + git(work, &["push", "-q", bare_path.to_str().unwrap(), "main"]); 121 121 + let substitute = Oid::from_hex(&git(work, &["rev-parse", "HEAD:payload.txt"])).unwrap(); 122 122 + 123 123 + git( 124 124 + &bare_path, 125 125 + &[ 126 126 + "update-ref", 127 127 + &format!("refs/replace/{}", original.to_hex()), 128 128 + &substitute.to_hex(), 129 129 + ], 130 130 + ); 131 131 + 132 132 + assert_eq!( 133 133 + git(&bare_path, &["cat-file", "blob", &original.to_hex()]), 134 134 + "pwned", 135 135 + "git read the replaced object as itself, so this fixture never armed the substitution" 136 136 + ); 137 137 + assert_eq!( 138 138 + layout.open(&did).unwrap().read_blob(original).unwrap(), 139 139 + b"kelp\n", 140 140 + "a pushed replace ref rewrote what the knot serves for an object" 141 141 + ); 142 142 + }

+76

knot2/crates/knot-git/tests/reads.rs

View file Reviewed

··· 628 628 tree, 629 629 knot_git::ArchiveFormat::TarGz, 630 630 Some(&knot_git::ArchivePrefix::new("squid-main/").unwrap()), 631 631 + knot_git::ArchiveLimit::new(u64::MAX), 631 632 &mut out, 632 633 ) 633 634 .unwrap(); ··· 644 645 assert!( 645 646 contains(&tar, b"squid-main/src/lib.rs"), 646 647 "tar contains prefixed entries" 648 648 + ); 649 649 + } 650 650 + 651 651 + #[test] 652 652 + fn an_archive_stops_at_its_limit_instead_of_spooling_the_whole_tree() { 653 653 + let (_scan, work_dir, layout, did) = seed_rich(); 654 654 + let work = work_dir.path(); 655 655 + let bare = layout.open(&did).unwrap(); 656 656 + let head = Oid::from_hex(&git(work, &["rev-parse", "HEAD"])).unwrap(); 657 657 + let tree = bare.peel_to_tree(head).unwrap(); 658 658 + 659 659 + [ 660 660 + knot_git::ArchiveFormat::Tar, 661 661 + knot_git::ArchiveFormat::TarGz, 662 662 + knot_git::ArchiveFormat::Zip, 663 663 + ] 664 664 + .iter() 665 665 + .for_each(|format| { 666 666 + let mut out = std::io::Cursor::new(Vec::new()); 667 667 + let refused = bare.write_archive( 668 668 + tree, 669 669 + *format, 670 670 + None, 671 671 + knot_git::ArchiveLimit::new(512), 672 672 + &mut out, 673 673 + ); 674 674 + assert!( 675 675 + matches!(refused, Err(knot_git::GitError::ArchiveTooLarge { .. })), 676 676 + "a {format:?} archive past its limit must be refused, got {refused:?}" 677 677 + ); 678 678 + assert!( 679 679 + out.into_inner().len() <= 512, 680 680 + "the {format:?} writer took bytes past the limit before the refusal" 681 681 + ); 682 682 + }); 683 683 + } 684 684 + 685 685 + #[test] 686 686 + fn a_compressible_tree_is_measured_before_the_compressor_ever_sees_it() { 687 687 + let scan = tempfile::tempdir().unwrap(); 688 688 + let layout = Layout::new(scan.path()); 689 689 + let did = RepoDid::new("did:plc:whelk").unwrap(); 690 690 + layout.create(&did).unwrap(); 691 691 + let bare_path = layout.repo_path(&did).unwrap(); 692 692 + 693 693 + let work_dir = tempfile::tempdir().unwrap(); 694 694 + let work = work_dir.path(); 695 695 + git(work, &["init", "-q", "-b", "main"]); 696 696 + commit_file( 697 697 + work, 698 698 + "kelp.txt", 699 699 + &"kelp\n".repeat(200_000), 700 700 + "one very compressible blob", 701 701 + ); 702 702 + git(work, &["push", "-q", bare_path.to_str().unwrap(), "main"]); 703 703 + 704 704 + let bare = layout.open(&did).unwrap(); 705 705 + let head = Oid::from_hex(&git(work, &["rev-parse", "HEAD"])).unwrap(); 706 706 + let tree = bare.peel_to_tree(head).unwrap(); 707 707 + 708 708 + let mut out = std::io::Cursor::new(Vec::new()); 709 709 + let refused = bare.write_archive( 710 710 + tree, 711 711 + knot_git::ArchiveFormat::TarGz, 712 712 + None, 713 713 + knot_git::ArchiveLimit::new(64 * 1024), 714 714 + &mut out, 715 715 + ); 716 716 + assert!( 717 717 + matches!(refused, Err(knot_git::GitError::ArchiveTooLarge { .. })), 718 718 + "a tree that gzips under the limit still costs its full size to read, so it must be refused, got {refused:?}" 719 719 + ); 720 720 + assert!( 721 721 + out.into_inner().is_empty(), 722 722 + "the knot must refuse before the compressor writes a byte" 647 723 ); 648 724 } 649 725

+12 -5

knot2/crates/knot-pack/src/archive.rs

View file Reviewed

··· 1 1 use std::io::{self, Read, Seek, SeekFrom}; 2 2 3 3 - use knot_git::{ArchiveFormat, ArchivePrefix, Repo}; 3 3 + use knot_git::{ArchiveFormat, ArchiveLimit, ArchivePrefix, Repo}; 4 4 use knot_types::Oid; 5 5 6 6 use crate::error::PackError; ··· 15 15 pub fn stream( 16 16 repo: &Repo, 17 17 request: &[u8], 18 18 + limit: ArchiveLimit, 18 19 sink: &mut dyn FnMut(&[u8]) -> io::Result<()>, 19 20 ) -> Result<(), PackError> { 20 21 let args = parse_arguments(request)?; 21 21 - match build(repo, &args) { 22 22 + match build(repo, &args, limit) { 22 23 Ok(mut spool) => { 23 24 let mut head = Vec::new(); 24 25 pkt::write_data(&mut head, b"ACK\n")?; ··· 109 110 } 110 111 } 111 112 112 112 - fn build(repo: &Repo, args: &[String]) -> Result<std::fs::File, PackError> { 113 113 + fn build(repo: &Repo, args: &[String], limit: ArchiveLimit) -> Result<std::fs::File, PackError> { 113 114 let request = interpret(args)?; 114 115 let id = repo 115 116 .resolve_revision(&request.treeish) ··· 119 120 .peel_to_tree(commit) 120 121 .map_err(|error| PackError::Pack(error.to_string()))?; 121 122 let mut spool = tempfile::tempfile().map_err(|error| PackError::Pack(error.to_string()))?; 122 122 - repo.write_archive(tree, request.format, request.prefix.as_ref(), &mut spool) 123 123 - .map_err(|error| PackError::Pack(error.to_string()))?; 123 123 + repo.write_archive( 124 124 + tree, 125 125 + request.format, 126 126 + request.prefix.as_ref(), 127 127 + limit, 128 128 + &mut spool, 129 129 + ) 130 130 + .map_err(|error| PackError::Pack(error.to_string()))?; 124 131 spool 125 132 .seek(SeekFrom::Start(0)) 126 133 .map_err(|error| PackError::Pack(error.to_string()))?;

+67 -52

knot2/crates/knot-pack/src/lib.rs

View file Reviewed

··· 219 219 pub fn upload_archive_streamed( 220 220 repo: &Repo, 221 221 request: &[u8], 222 222 + limit: knot_git::ArchiveLimit, 222 223 sink: &mut dyn FnMut(&[u8]) -> io::Result<()>, 223 224 ) -> Result<(), PackError> { 224 224 - archive::stream(repo, request, sink) 225 225 + archive::stream(repo, request, limit, sink) 225 226 } 226 227 227 227 - pub fn upload_archive(repo: &Repo, request: &[u8]) -> Result<Vec<u8>, PackError> { 228 228 + pub fn upload_archive( 229 229 + repo: &Repo, 230 230 + request: &[u8], 231 231 + limit: knot_git::ArchiveLimit, 232 232 + ) -> Result<Vec<u8>, PackError> { 228 233 let mut buf = Vec::new(); 229 229 - upload_archive_streamed(repo, request, &mut |chunk| { 234 234 + upload_archive_streamed(repo, request, limit, &mut |chunk| { 230 235 buf.extend_from_slice(chunk); 231 236 Ok(()) 232 237 })?; ··· 340 345 cache: Arc<cache::PackCache>, 341 346 catalog: Arc<Catalog>, 342 347 hostname: KnotHostname, 348 348 + archive_limit: knot_git::ArchiveLimit, 349 349 + } 350 350 + 351 351 + pub struct EdgeConfig { 352 352 + pub layout: Layout, 353 353 + pub resolver: Arc<dyn RepoResolver>, 354 354 + pub receive: Option<Arc<dyn ReceiveAdvertiser>>, 355 355 + pub handle_resolver: Option<Arc<dyn HandleResolver>>, 356 356 + pub pack_slots: PackSlots, 357 357 + pub cache: CacheConfig, 358 358 + pub catalog: Arc<Catalog>, 359 359 + pub hostname: KnotHostname, 360 360 + pub clock: Arc<dyn Clock>, 361 361 + pub archive_limit: knot_git::ArchiveLimit, 362 362 + } 363 363 + 364 364 + impl EdgeConfig { 365 365 + pub fn serving(layout: Layout, resolver: Arc<dyn RepoResolver>, clock: Arc<dyn Clock>) -> Self { 366 366 + Self { 367 367 + layout, 368 368 + resolver, 369 369 + receive: None, 370 370 + handle_resolver: None, 371 371 + pack_slots: PackSlots::new(knot_resource::threads().get()), 372 372 + cache: CacheConfig::default(), 373 373 + catalog: Arc::new(Catalog::defaults()), 374 374 + hostname: default_hostname().clone(), 375 375 + clock, 376 376 + archive_limit: knot_git::ArchiveLimit::default(), 377 377 + } 378 378 + } 379 379 + 380 380 + pub fn with_pack_slots(self, pack_slots: PackSlots) -> Self { 381 381 + Self { pack_slots, ..self } 382 382 + } 343 383 } 344 384 345 385 pub fn router(layout: Layout, resolver: Arc<dyn RepoResolver>, clock: Arc<dyn Clock>) -> Router { 346 346 - router_with_pack_slots( 347 347 - layout, 348 348 - resolver, 349 349 - PackSlots::new(knot_resource::threads().get()), 350 350 - clock, 351 351 - ) 386 386 + serving_router(EdgeConfig::serving(layout, resolver, clock)) 352 387 } 353 388 354 389 pub fn router_with_pack_slots( ··· 357 392 pack_slots: PackSlots, 358 393 clock: Arc<dyn Clock>, 359 394 ) -> Router { 360 360 - let state = pack_state( 361 361 - layout, 362 362 - resolver, 363 363 - None, 364 364 - None, 365 365 - pack_slots, 366 366 - CacheConfig::default(), 367 367 - Arc::new(Catalog::defaults()), 368 368 - default_hostname().clone(), 369 369 - clock, 370 370 - ); 395 395 + serving_router(EdgeConfig::serving(layout, resolver, clock).with_pack_slots(pack_slots)) 396 396 + } 397 397 + 398 398 + fn serving_router(config: EdgeConfig) -> Router { 399 399 + let state = pack_state(config); 371 400 write_routes(state.clone()).merge(advertisement_routes(state).into_router()) 372 401 } 373 402 374 374 - #[allow(clippy::too_many_arguments)] 375 375 - pub fn edge_routes( 376 376 - layout: Layout, 377 377 - resolver: Arc<dyn RepoResolver>, 378 378 - receive: Option<Arc<dyn ReceiveAdvertiser>>, 379 379 - handle_resolver: Option<Arc<dyn HandleResolver>>, 380 380 - pack_slots: PackSlots, 381 381 - cache: CacheConfig, 382 382 - catalog: Arc<Catalog>, 383 383 - hostname: KnotHostname, 384 384 - clock: Arc<dyn Clock>, 385 385 - ) -> (Router, knot_edge::ZeroRttRoutes) { 386 386 - let state = pack_state( 403 403 + pub fn edge_routes(config: EdgeConfig) -> (Router, knot_edge::ZeroRttRoutes) { 404 404 + let state = pack_state(config); 405 405 + (write_routes(state.clone()), advertisement_routes(state)) 406 406 + } 407 407 + 408 408 + fn pack_state(config: EdgeConfig) -> PackState { 409 409 + let EdgeConfig { 387 410 layout, 388 411 resolver, 389 412 receive, ··· 393 416 catalog, 394 417 hostname, 395 418 clock, 396 396 - ); 397 397 - (write_routes(state.clone()), advertisement_routes(state)) 398 398 - } 399 399 - 400 400 - #[allow(clippy::too_many_arguments)] 401 401 - fn pack_state( 402 402 - layout: Layout, 403 403 - resolver: Arc<dyn RepoResolver>, 404 404 - receive: Option<Arc<dyn ReceiveAdvertiser>>, 405 405 - handle_resolver: Option<Arc<dyn HandleResolver>>, 406 406 - pack_slots: PackSlots, 407 407 - cache: CacheConfig, 408 408 - catalog: Arc<Catalog>, 409 409 - hostname: KnotHostname, 410 410 - clock: Arc<dyn Clock>, 411 411 - ) -> PackState { 419 419 + archive_limit, 420 420 + } = config; 412 421 PackState { 413 422 layout, 414 423 resolver, ··· 418 427 cache: cache::PackCache::new(cache, clock), 419 428 catalog, 420 429 hostname, 430 430 + archive_limit, 421 431 } 422 432 } 423 433 ··· 830 840 body: Vec<u8>, 831 841 ) -> Result<Response, PackError> { 832 842 let permit = state.pack_slots.acquire().await; 833 833 - Ok(archive_response(repo, body, permit)) 843 843 + Ok(archive_response(repo, body, state.archive_limit, permit)) 834 844 } 835 845 836 836 - fn archive_response(repo: Repo, body: Vec<u8>, permit: SlotPermit) -> Response { 846 846 + fn archive_response( 847 847 + repo: Repo, 848 848 + body: Vec<u8>, 849 849 + limit: knot_git::ArchiveLimit, 850 850 + permit: SlotPermit, 851 851 + ) -> Response { 837 852 let (tx, rx) = mpsc::channel::<Result<Bytes, io::Error>>(16); 838 853 tokio::task::spawn_blocking(move || { 839 854 let _permit = permit; ··· 841 856 tx.blocking_send(Ok(Bytes::copy_from_slice(chunk))) 842 857 .map_err(|_| io::Error::other("client disconnected")) 843 858 }; 844 844 - if let Err(error) = upload_archive_streamed(&repo, &body, &mut sink) { 859 859 + if let Err(error) = upload_archive_streamed(&repo, &body, limit, &mut sink) { 845 860 let _ = tx.blocking_send(Err(io::Error::other(error.to_string()))); 846 861 } 847 862 });

+58 -1

knot2/crates/knot-pack/tests/git_client.rs

View file Reviewed

··· 483 483 args.iter() 484 484 .for_each(|arg| request.extend(pkt(arg.as_bytes()))); 485 485 request.extend_from_slice(b"0000"); 486 486 - knot_pack::upload_archive(&repo, &request).unwrap() 486 486 + knot_pack::upload_archive(&repo, &request, knot_git::ArchiveLimit::default()).unwrap() 487 487 }; 488 488 489 489 let raw_arg = format!("argument {}\n", tree.to_hex()); ··· 524 524 assert!( 525 525 !contains(&cob, b"README.md"), 526 526 "refused archive mustn't leak the hidden tree's contents" 527 527 + ); 528 528 + } 529 529 + 530 530 + #[tokio::test(flavor = "multi_thread")] 531 531 + async fn http_upload_archive_honors_the_configured_archive_limit() { 532 532 + use tower::ServiceExt as _; 533 533 + 534 534 + let scan = tempfile::tempdir().unwrap(); 535 535 + let layout = Layout::new(scan.path()); 536 536 + let did = RepoDid::new("did:plc:limpet").unwrap(); 537 537 + layout.create(&did).unwrap(); 538 538 + let bare = layout.repo_path(&did).unwrap(); 539 539 + 540 540 + let scratch = tempfile::tempdir().unwrap(); 541 541 + let work = scratch.path().join("work"); 542 542 + seed_repo(&work, bare.to_str().unwrap(), "README.md", "archive me\n"); 543 543 + 544 544 + let (write_routes, _advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig { 545 545 + pack_slots: knot_resource::PackSlots::new(1), 546 546 + archive_limit: knot_git::ArchiveLimit::new(512), 547 547 + ..knot_pack::EdgeConfig::serving( 548 548 + layout.clone(), 549 549 + serve_dids(), 550 550 + Arc::new(knot_runtime::SystemClock), 551 551 + ) 552 552 + }); 553 553 + 554 554 + let mut framed = Vec::new(); 555 555 + framed.extend(pkt(b"argument --format=tar\n")); 556 556 + framed.extend(pkt(b"argument HEAD\n")); 557 557 + framed.extend_from_slice(b"0000"); 558 558 + let response = write_routes 559 559 + .oneshot( 560 560 + axum::http::Request::builder() 561 561 + .method("POST") 562 562 + .uri(format!("/{}/git-upload-archive", did.as_str())) 563 563 + .header( 564 564 + header::CONTENT_TYPE, 565 565 + "application/x-git-upload-archive-request", 566 566 + ) 567 567 + .body(Body::from(framed)) 568 568 + .unwrap(), 569 569 + ) 570 570 + .await 571 571 + .unwrap(); 572 572 + let body = http_body_util::BodyExt::collect(response.into_body()) 573 573 + .await 574 574 + .unwrap() 575 575 + .to_bytes(); 576 576 + let text = String::from_utf8_lossy(&body); 577 577 + assert!( 578 578 + text.contains("NACK") && text.contains("archive exceeds the 512 byte limit"), 579 579 + "an archive past the state's limit must be declined, got {text:?}" 580 580 + ); 581 581 + assert!( 582 582 + !contains(&body, b"README.md"), 583 583 + "the declined archive mustn't leak the tree it refused to serve" 527 584 ); 528 585 } 529 586

+10 -13

knot2/crates/knot-pack/tests/h3_conformance.rs

View file Reviewed

··· 13 13 RequiresFullHandshake, StaticCertPaths, TlsSetup, WriteRequestTimeout, 14 14 }; 15 15 use knot_git::Layout; 16 16 - use knot_pack::{CacheConfig, RepoLookup, RepoResolver, RepoTarget}; 16 16 + use knot_pack::{RepoLookup, RepoResolver, RepoTarget}; 17 17 use knot_types::{ObjectFormat, RepoDid}; 18 18 use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; 19 19 use rustls::crypto::aws_lc_rs; ··· 92 92 RequestTimeout::from_millis(nz64(120_000)), 93 93 BodyInactivityTimeout::from_millis(nz64(120_000)), 94 94 WriteRequestTimeout::from_millis(nz64(1_800_000)), 95 95 - None, 95 95 + knot_types::ProxyTrust::default(), 96 96 ), 97 97 tls: Some(TlsSetup { 98 98 source: CertSource::Static(StaticCertPaths { ··· 164 164 for _ in 0..8 { 165 165 let addr: SocketAddr = format!("127.0.0.1:{}", free_port()).parse().unwrap(); 166 166 let (cert, key, cert_der) = write_self_signed(certdir); 167 167 - let (write_routes, advertisement) = knot_pack::edge_routes( 168 168 - layout.clone(), 169 169 - serve_dids(), 170 170 - None, 171 171 - None, 172 172 - knot_resource::PackSlots::new(4), 173 173 - CacheConfig::default(), 174 174 - Arc::new(knot_messages::Catalog::defaults()), 175 175 - knot_pack::default_hostname().clone(), 176 176 - Arc::new(knot_runtime::SystemClock), 177 177 - ); 167 167 + let (write_routes, advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig { 168 168 + pack_slots: knot_resource::PackSlots::new(4), 169 169 + ..knot_pack::EdgeConfig::serving( 170 170 + layout.clone(), 171 171 + serve_dids(), 172 172 + Arc::new(knot_runtime::SystemClock), 173 173 + ) 174 174 + }); 178 175 let log: Arc<Mutex<Vec<Captured>>> = Arc::new(Mutex::new(Vec::new())); 179 176 let sink = log.clone(); 180 177 let recorded = write_routes.layer(axum::middleware::from_fn(

+9 -11

knot2/crates/knot-pack/tests/handle_owner.rs

View file Reviewed

··· 6 6 use axum::http::{Request, StatusCode}; 7 7 use http_body_util::BodyExt; 8 8 use knot_git::Layout; 9 9 - use knot_pack::{CacheConfig, HandleResolver, RepoLookup, RepoResolver, RepoTarget}; 9 9 + use knot_pack::{HandleResolver, RepoLookup, RepoResolver, RepoTarget}; 10 10 use knot_types::{AccountDid, Handle, OwnerDid, RepoDid}; 11 11 use tower::ServiceExt; 12 12 ··· 38 38 } 39 39 40 40 fn build(layout: &Layout, handle_resolver: Option<Arc<dyn HandleResolver>>) -> axum::Router { 41 41 - let (_write, advertisement) = knot_pack::edge_routes( 42 42 - layout.clone(), 43 43 - repo_resolver(), 44 44 - None, 41 41 + let (_write, advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig { 45 42 handle_resolver, 46 46 - knot_resource::PackSlots::new(4), 47 47 - CacheConfig::default(), 48 48 - Arc::new(knot_messages::Catalog::defaults()), 49 49 - knot_pack::default_hostname().clone(), 50 50 - Arc::new(knot_runtime::SystemClock), 51 51 - ); 43 43 + pack_slots: knot_resource::PackSlots::new(4), 44 44 + ..knot_pack::EdgeConfig::serving( 45 45 + layout.clone(), 46 46 + repo_resolver(), 47 47 + Arc::new(knot_runtime::SystemClock), 48 48 + ) 49 49 + }); 52 50 advertisement.into_router() 53 51 } 54 52

+36 -27

knot2/crates/knot-server/src/main.rs

View file Reviewed

··· 279 279 .context("server.listen_max_connections must be greater than zero")?, 280 280 ); 281 281 // A header name that doesn't parse will never match, 282 282 - // `effective_peer` falls back to socket, 282 282 + // `ProxyTrust::client_peer` falls back to socket, 283 283 // and every request in the world shares 284 284 // the proxy's address + its one ratelimit bucket. 285 285 // So... better to refuse to start. ··· 290 290 .map(|header| axum::http::HeaderName::from_bytes(header.as_bytes())) 291 291 .transpose() 292 292 .context("xrpc.trusted_proxy_header isn't a valid HTTP header name")?; 293 293 + let trusted_proxies = 294 294 + knot_types::TrustedProxies::new(config.xrpc.trusted_proxies.iter().copied()); 295 295 + let proxy_trust = knot_types::ProxyTrust::new(trusted_proxy_header, trusted_proxies); 296 296 + if proxy_trust.trusts_any_peer() && !http_addr.ip().is_loopback() { 297 297 + tracing::warn!( 298 298 + bind = %http_addr, 299 299 + "xrpc.trusted_proxy_header is set without xrpc.trusted_proxies while the HTTP surface takes connections from off-host, so a client that reaches this knot without passing the proxy can forge the header and pick its own rate-limit bucket. List the proxy's address in xrpc.trusted_proxies." 300 300 + ); 301 301 + } 293 302 let edge_guards = knot_edge::EdgeGuards::new( 294 303 knot_edge::RequestsPerSecond::new( 295 304 NonZeroU32::new(config.server.listen_rate_limit_per_second) ··· 315 324 NonZeroU64::new(config.server.listen_write_request_timeout_ms) 316 325 .context("server.listen_write_request_timeout_ms must be greater than zero")?, 317 326 ), 318 318 - trusted_proxy_header.clone(), 327 327 + proxy_trust.clone(), 319 328 ); 320 329 let tls_setup = build_tls_setup(&config, &hostname).context("assemble TLS configuration")?; 321 330 if config.tls.http3 && tls_setup.is_none() { ··· 325 334 } 326 335 if tls_setup.is_none() && config.xrpc.trusted_proxy_header.is_none() { 327 336 tracing::warn!( 328 328 - "running plaintext behind a reverse proxy without xrpc.trusted_proxy_header. Per-IP rate limiting will key on the proxy socket address, throttling all clients as one. Set xrpc.trusted_proxy_header to the header your proxy appends." 337 337 + "running plaintext behind a reverse proxy without xrpc.trusted_proxy_header. Per-IP rate limiting will key on the proxy socket address, throttling all clients as one. Set xrpc.trusted_proxy_header to the header your proxy appends, and xrpc.trusted_proxies to the address it connects from." 329 338 ); 330 339 } 331 340 if config.tls.acme_enabled && http_addr.port() != 443 { ··· 502 511 503 512 let slots = knot_resource::Slots::for_machine(); 504 513 505 505 - let ssh_base = knot_ssh::SshState::new( 506 506 - layout.clone(), 507 507 - Arc::clone(&index), 508 508 - Arc::clone(&atproto), 514 514 + let ssh_base = knot_ssh::SshState::new(knot_ssh::SshConfig { 515 515 + layout: layout.clone(), 516 516 + index: Arc::clone(&index), 517 517 + atproto: Arc::clone(&atproto), 509 518 knot_actor, 510 510 - Arc::clone(&events), 511 511 - hostname.clone(), 512 512 - appview_endpoint.clone(), 513 513 - admins.clone(), 519 519 + events: Arc::clone(&events), 520 520 + hostname: hostname.clone(), 521 521 + appview: appview_endpoint.clone(), 522 522 + admins: admins.clone(), 514 523 admission, 515 515 - byte_limits.pack, 516 516 - budgets.languages_push, 517 517 - ci_logs.clone(), 518 518 - ) 524 524 + max_pack_bytes: byte_limits.pack, 525 525 + archive_limit: byte_limits.archive, 526 526 + languages_push_budget: budgets.languages_push, 527 527 + ci_logs: ci_logs.clone(), 528 528 + }) 519 529 .with_maintenance(maintenance_handle.clone()) 520 530 .with_limits(pack_limits) 521 531 .with_slots(slots.clone()) ··· 541 551 limiter: Arc::new(knot_xrpc::PreAuthLimiter::with_config(xrpc_limits)), 542 552 cob_locks: Arc::new(knot_xrpc::CobLocks::default()), 543 553 reservations, 544 544 - trusted_proxy_header, 554 554 + proxy_trust, 545 555 committer, 546 556 byte_limits, 547 557 budgets, ··· 573 583 let handle_resolver: Arc<dyn knot_pack::HandleResolver> = Arc::new(AtprotoHandleResolver { 574 584 atproto: Arc::clone(&atproto), 575 585 }); 576 576 - let (write_routes, early_data_safe) = knot_pack::edge_routes( 577 577 - layout, 578 578 - resolver, 579 579 - Some(receive_advertiser), 580 580 - Some(handle_resolver), 581 581 - slots.pack.clone(), 582 582 - pack_cache_config, 583 583 - Arc::clone(&catalog), 584 584 - xrpc_state.knot_hostname.clone(), 585 585 - Arc::new(SystemClock), 586 586 - ); 586 586 + let (write_routes, early_data_safe) = knot_pack::edge_routes(knot_pack::EdgeConfig { 587 587 + receive: Some(receive_advertiser), 588 588 + handle_resolver: Some(handle_resolver), 589 589 + pack_slots: slots.pack.clone(), 590 590 + cache: pack_cache_config, 591 591 + catalog: Arc::clone(&catalog), 592 592 + hostname: xrpc_state.knot_hostname.clone(), 593 593 + archive_limit: byte_limits.archive, 594 594 + ..knot_pack::EdgeConfig::serving(layout, resolver, Arc::new(SystemClock)) 595 595 + }); 587 596 let legacy_admin_routes = legacy_admin.map(|secret| { 588 597 tracing::warn!( 589 598 route = knot_xrpc::legacy_admin::ADD_MEMBER_ROUTE,

+1 -1

knot2/crates/knot-sim/src/harness.rs

View file Reviewed

··· 732 732 PerActorQuota::new(256), 733 733 GlobalQuota::new(256), 734 734 )), 735 735 - trusted_proxy_header: None, 735 735 + proxy_trust: knot_types::ProxyTrust::default(), 736 736 committer: Committer { 737 737 name: AuthorName::new("knot"), 738 738 email: Email::new("knot@nel.pet"),

+1 -1

knot2/crates/knot-sim/tests/common/mod.rs

View file Reviewed

··· 73 73 RequestTimeout::from_millis(nz64(120_000)), 74 74 BodyInactivityTimeout::from_millis(nz64(120_000)), 75 75 WriteRequestTimeout::from_millis(nz64(1_800_000)), 76 76 - None, 76 76 + knot_types::ProxyTrust::default(), 77 77 ), 78 78 tls: Some(TlsSetup { 79 79 source: CertSource::Static(StaticCertPaths {

+9 -12

knot2/crates/knot-sim/tests/h3.rs

View file Reviewed

··· 11 11 use http::Method; 12 12 use knot_edge::RequiresFullHandshake; 13 13 use knot_git::Layout; 14 14 - use knot_pack::{CacheConfig, RepoLookup, RepoResolver, RepoTarget}; 14 14 + use knot_pack::{RepoLookup, RepoResolver, RepoTarget}; 15 15 use knot_types::{ObjectFormat, RepoDid}; 16 16 17 17 const PINNED_DATE: &str = "2026-06-20T12:00:00+00:00"; ··· 192 192 let certdir = tempfile::tempdir().unwrap(); 193 193 let clonedir = tempfile::tempdir().unwrap(); 194 194 let edge = common::serve_edge(certdir.path(), || { 195 195 - let (write_routes, advertisement) = knot_pack::edge_routes( 196 196 - layout.clone(), 197 197 - serve_dids(), 198 198 - None, 199 199 - None, 200 200 - knot_resource::PackSlots::new(4), 201 201 - CacheConfig::default(), 202 202 - Arc::new(knot_messages::Catalog::defaults()), 203 203 - knot_pack::default_hostname().clone(), 204 204 - Arc::new(knot_runtime::SystemClock), 205 205 - ); 195 195 + let (write_routes, advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig { 196 196 + pack_slots: knot_resource::PackSlots::new(4), 197 197 + ..knot_pack::EdgeConfig::serving( 198 198 + layout.clone(), 199 199 + serve_dids(), 200 200 + Arc::new(knot_runtime::SystemClock), 201 201 + ) 202 202 + }); 206 203 (RequiresFullHandshake::new(write_routes), advertisement) 207 204 }) 208 205 .await;

+34 -37

knot2/crates/knot-sim/tests/lfs_roundtrip.rs

View file Reviewed

··· 308 308 ), 309 309 )); 310 310 let ssh_state = Arc::new( 311 311 - knot_ssh::SshState::new( 312 312 - layout.clone(), 313 313 - Arc::clone(&index), 314 314 - Arc::clone(&atproto), 315 315 - knot_types::ActorId::from_secp256k1(actor_signer().public_key().as_bytes()), 316 316 - Arc::clone(&events), 317 317 - KnotHostname::new("nel.pet").unwrap(), 318 318 - knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 319 319 - BTreeSet::from([AccountDid::new(OWNER_DID).unwrap()]), 320 320 - AdmissionPolicy::Closed, 321 321 - knot_xrpc::MaxWireBytes::new(1 << 30), 322 322 - knot_xrpc::LanguagesPushBudget::new(Duration::from_secs(2)), 323 323 - None, 324 324 - ) 311 311 + knot_ssh::SshState::new(knot_ssh::SshConfig { 312 312 + layout: layout.clone(), 313 313 + index: Arc::clone(&index), 314 314 + atproto: Arc::clone(&atproto), 315 315 + knot_actor: knot_types::ActorId::from_secp256k1(actor_signer().public_key().as_bytes()), 316 316 + events: Arc::clone(&events), 317 317 + hostname: KnotHostname::new("nel.pet").unwrap(), 318 318 + appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 319 319 + admins: BTreeSet::from([AccountDid::new(OWNER_DID).unwrap()]), 320 320 + admission: AdmissionPolicy::Closed, 321 321 + max_pack_bytes: knot_xrpc::MaxWireBytes::new(1 << 30), 322 322 + archive_limit: knot_git::ArchiveLimit::default(), 323 323 + languages_push_budget: knot_xrpc::LanguagesPushBudget::new(Duration::from_secs(2)), 324 324 + ci_logs: None, 325 325 + }) 325 326 .with_lfs(lfs.clone(), 16), 326 327 ); 327 328 let ssh_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); ··· 356 357 knot_xrpc::PerActorQuota::new(16), 357 358 knot_xrpc::GlobalQuota::new(16), 358 359 )), 359 359 - trusted_proxy_header: None, 360 360 + proxy_trust: knot_types::ProxyTrust::default(), 360 361 committer: knot_xrpc::Committer { 361 362 name: AuthorName::new("Tangled"), 362 363 email: Email::new("noreply@tangled.sh"), ··· 405 406 }) 406 407 }; 407 408 let advertiser = knot_xrpc::receive_advertiser(Arc::clone(&xrpc_state)); 408 408 - let (write_routes, advertisement) = knot_pack::edge_routes( 409 409 - layout.clone(), 410 410 - Arc::clone(&resolver), 411 411 - Some(Arc::clone(&advertiser)), 412 412 - None, 413 413 - knot_resource::PackSlots::new(4), 414 414 - knot_pack::CacheConfig::default(), 415 415 - Arc::new(knot_messages::Catalog::defaults()), 416 416 - knot_pack::default_hostname().clone(), 417 417 - Arc::new(knot_runtime::SystemClock), 418 418 - ); 409 409 + let (write_routes, advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig { 410 410 + receive: Some(Arc::clone(&advertiser)), 411 411 + pack_slots: knot_resource::PackSlots::new(4), 412 412 + ..knot_pack::EdgeConfig::serving( 413 413 + layout.clone(), 414 414 + Arc::clone(&resolver), 415 415 + Arc::new(knot_runtime::SystemClock), 416 416 + ) 417 417 + }); 419 418 let router = write_routes 420 419 .merge(advertisement.into_router()) 421 420 .merge(knot_xrpc::router(Arc::clone(&xrpc_state))); ··· 428 427 true => { 429 428 let certdir = tempfile::tempdir().unwrap(); 430 429 let edge = common::serve_edge(certdir.path(), || { 431 431 - let (write_routes, advertisement) = knot_pack::edge_routes( 432 432 - layout.clone(), 433 433 - Arc::clone(&resolver), 434 434 - Some(Arc::clone(&advertiser)), 435 435 - None, 436 436 - knot_resource::PackSlots::new(4), 437 437 - knot_pack::CacheConfig::default(), 438 438 - Arc::new(knot_messages::Catalog::defaults()), 439 439 - knot_pack::default_hostname().clone(), 440 440 - Arc::new(knot_runtime::SystemClock), 441 441 - ); 430 430 + let (write_routes, advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig { 431 431 + receive: Some(Arc::clone(&advertiser)), 432 432 + pack_slots: knot_resource::PackSlots::new(4), 433 433 + ..knot_pack::EdgeConfig::serving( 434 434 + layout.clone(), 435 435 + Arc::clone(&resolver), 436 436 + Arc::new(knot_runtime::SystemClock), 437 437 + ) 438 438 + }); 442 439 let app = RequiresFullHandshake::new( 443 440 write_routes.merge(knot_xrpc::router(Arc::clone(&xrpc_state))), 444 441 );

+15 -12

knot2/crates/knot-sim/tests/ssh.rs

View file Reviewed

··· 183 183 knot_events::ReplayBytes::new(16 << 20).unwrap(), 184 184 ), 185 185 )); 186 186 - let state = Arc::new(knot_ssh::SshState::new( 187 187 - layout.clone(), 186 186 + let state = Arc::new(knot_ssh::SshState::new(knot_ssh::SshConfig { 187 187 + layout: layout.clone(), 188 188 index, 189 189 atproto, 190 190 - actor_for_seed(1), 191 191 - Arc::clone(&events), 192 192 - knot_types::KnotHostname::new("knot.test").unwrap(), 193 193 - knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 194 194 - std::collections::BTreeSet::new(), 195 195 - knot_types::AdmissionPolicy::Closed, 196 196 - knot_xrpc::MaxWireBytes::new(1 << 30), 197 197 - knot_xrpc::LanguagesPushBudget::new(std::time::Duration::from_secs(2)), 198 198 - None, 199 199 - )); 190 190 + knot_actor: actor_for_seed(1), 191 191 + events: Arc::clone(&events), 192 192 + hostname: knot_types::KnotHostname::new("knot.test").unwrap(), 193 193 + appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 194 194 + admins: std::collections::BTreeSet::new(), 195 195 + admission: knot_types::AdmissionPolicy::Closed, 196 196 + max_pack_bytes: knot_xrpc::MaxWireBytes::new(1 << 30), 197 197 + archive_limit: knot_git::ArchiveLimit::default(), 198 198 + languages_push_budget: knot_xrpc::LanguagesPushBudget::new(std::time::Duration::from_secs( 199 199 + 2, 200 200 + )), 201 201 + ci_logs: None, 202 202 + })); 200 203 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); 201 204 let port = listener.local_addr().unwrap().port(); 202 205 tokio::spawn(async move {

+11 -10

knot2/crates/knot-ssh/examples/ephemeral_knot.rs

View file Reviewed

··· 205 205 .public_key() 206 206 .as_bytes(), 207 207 ); 208 208 - let state = Arc::new(knot_ssh::SshState::new( 208 208 + let state = Arc::new(knot_ssh::SshState::new(knot_ssh::SshConfig { 209 209 layout, 210 210 index, 211 211 atproto, 212 212 - actor, 212 212 + knot_actor: actor, 213 213 events, 214 214 - KnotHostname::new("knot.test").unwrap(), 215 215 - knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 216 216 - BTreeSet::new(), 217 217 - AdmissionPolicy::Closed, 218 218 - knot_pack::MaxWireBytes::new(1 << 34), 219 219 - knot_postreceive::LanguagesPushBudget::new(Duration::from_secs(2)), 220 220 - None, 221 221 - )); 214 214 + hostname: KnotHostname::new("knot.test").unwrap(), 215 215 + appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 216 216 + admins: BTreeSet::new(), 217 217 + admission: AdmissionPolicy::Closed, 218 218 + max_pack_bytes: knot_pack::MaxWireBytes::new(1 << 34), 219 219 + archive_limit: knot_git::ArchiveLimit::default(), 220 220 + languages_push_budget: knot_postreceive::LanguagesPushBudget::new(Duration::from_secs(2)), 221 221 + ci_logs: None, 222 222 + })); 222 223 223 224 let listener = TcpListener::bind(("127.0.0.1", port)).await.unwrap(); 224 225 let bound = listener.local_addr().unwrap().port();

+2 -1

knot2/crates/knot-ssh/src/exec.rs

View file Reviewed

··· 422 422 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(16); 423 423 let layout = state.layout.clone(); 424 424 let did = repo_did.clone(); 425 425 + let archive_limit = state.archive_limit; 425 426 let handle = tokio::task::spawn_blocking(move || -> Result<(), PackError> { 426 427 let _permit = permit; 427 428 let repo = layout.open(&did)?; ··· 429 430 tx.blocking_send(chunk.to_vec()) 430 431 .map_err(|_| std::io::Error::other("client disconnected")) 431 432 }; 432 432 - knot_pack::upload_archive_streamed(&repo, &request, &mut sink) 433 433 + knot_pack::upload_archive_streamed(&repo, &request, archive_limit, &mut sink) 433 434 }); 434 435 435 436 let mut writer = channel.make_writer();

+35 -16

knot2/crates/knot-ssh/src/lib.rs

View file Reviewed

··· 11 11 12 12 use knot_atproto::Atproto; 13 13 use knot_events::EventLog; 14 14 - use knot_git::Layout; 14 14 + use knot_git::{ArchiveLimit, Layout}; 15 15 use knot_index::Index; 16 16 use knot_maintenance::MaintenanceHandle; 17 17 use knot_pack::{MaxWireBytes, PackLimits}; ··· 57 57 admission: AdmissionPolicy, 58 58 limits: PackLimits, 59 59 max_pack_bytes: MaxWireBytes, 60 60 + archive_limit: ArchiveLimit, 60 61 languages_push_budget: LanguagesPushBudget, 61 62 ci_logs: Option<CiLogsAddr>, 62 63 slots: Slots, ··· 74 75 pub(crate) peer_slots: Arc<PreAuthLimiter>, 75 76 } 76 77 78 78 + pub struct SshConfig<H, C> { 79 79 + pub layout: Layout, 80 80 + pub index: Arc<Index>, 81 81 + pub atproto: Arc<Atproto<H, C>>, 82 82 + pub knot_actor: ActorId, 83 83 + pub events: Arc<EventLog<C>>, 84 84 + pub hostname: KnotHostname, 85 85 + pub appview: AppviewEndpoint, 86 86 + pub admins: BTreeSet<AccountDid>, 87 87 + pub admission: AdmissionPolicy, 88 88 + pub max_pack_bytes: MaxWireBytes, 89 89 + pub archive_limit: ArchiveLimit, 90 90 + pub languages_push_budget: LanguagesPushBudget, 91 91 + pub ci_logs: Option<CiLogsAddr>, 92 92 + } 93 93 + 77 94 impl<H: HttpTransport, C: Clock> SshState<H, C> { 78 78 - #[allow(clippy::too_many_arguments)] 79 79 - pub fn new( 80 80 - layout: Layout, 81 81 - index: Arc<Index>, 82 82 - atproto: Arc<Atproto<H, C>>, 83 83 - knot_actor: ActorId, 84 84 - events: Arc<EventLog<C>>, 85 85 - hostname: KnotHostname, 86 86 - appview: AppviewEndpoint, 87 87 - admins: BTreeSet<AccountDid>, 88 88 - admission: AdmissionPolicy, 89 89 - max_pack_bytes: MaxWireBytes, 90 90 - languages_push_budget: LanguagesPushBudget, 91 91 - ci_logs: Option<CiLogsAddr>, 92 92 - ) -> Self { 95 95 + pub fn new(config: SshConfig<H, C>) -> Self { 96 96 + let SshConfig { 97 97 + layout, 98 98 + index, 99 99 + atproto, 100 100 + knot_actor, 101 101 + events, 102 102 + hostname, 103 103 + appview, 104 104 + admins, 105 105 + admission, 106 106 + max_pack_bytes, 107 107 + archive_limit, 108 108 + languages_push_budget, 109 109 + ci_logs, 110 110 + } = config; 93 111 Self { 94 112 layout, 95 113 index, ··· 102 120 admission, 103 121 limits: PackLimits::default(), 104 122 max_pack_bytes, 123 123 + archive_limit, 105 124 languages_push_budget, 106 125 ci_logs, 107 126 slots: Slots::for_machine(),

+77 -26

knot2/crates/knot-ssh/tests/ssh_push.rs

View file Reviewed

··· 7 7 use knot_atproto::Atproto; 8 8 use knot_cob::{CobHome, CobStore}; 9 9 use knot_cobs::{CollaboratorsChange, Grant, MembersChange, Registration, RegistryChange}; 10 10 - use knot_git::{Layout, Repo}; 10 10 + use knot_git::{ArchiveLimit, Layout, Repo}; 11 11 use knot_index::Index; 12 12 use knot_pack::MaxWireBytes; 13 13 use knot_postreceive::LanguagesPushBudget; ··· 205 205 max_pack_bytes: MaxWireBytes, 206 206 warm: bool, 207 207 ) -> (Server, Arc<Index>) { 208 208 - let (server, index, _, _) = spawn_server_core(published_line, max_pack_bytes, warm, None).await; 208 208 + let (server, index, _, _) = spawn_server_core( 209 209 + published_line, 210 210 + max_pack_bytes, 211 211 + ArchiveLimit::default(), 212 212 + warm, 213 213 + None, 214 214 + ) 215 215 + .await; 209 216 (server, index) 210 217 } 211 218 212 219 async fn spawn_server_core( 213 220 published_line: String, 214 221 max_pack_bytes: MaxWireBytes, 222 222 + archive_limit: ArchiveLimit, 215 223 warm: bool, 216 224 lfs: Option<knot_lfs::LfsHandle>, 217 225 ) -> ( ··· 291 299 knot_events::ReplayBytes::new(16 << 20).unwrap(), 292 300 ), 293 301 )); 294 294 - let base = knot_ssh::SshState::new( 295 295 - layout.clone(), 296 296 - Arc::clone(&index), 302 302 + let base = knot_ssh::SshState::new(knot_ssh::SshConfig { 303 303 + layout: layout.clone(), 304 304 + index: Arc::clone(&index), 297 305 atproto, 298 298 - actor_for_seed(1), 299 299 - Arc::clone(&events), 300 300 - knot_types::KnotHostname::new("knot.test").unwrap(), 301 301 - knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 302 302 - std::collections::BTreeSet::new(), 303 303 - knot_types::AdmissionPolicy::Closed, 306 306 + knot_actor: actor_for_seed(1), 307 307 + events: Arc::clone(&events), 308 308 + hostname: knot_types::KnotHostname::new("knot.test").unwrap(), 309 309 + appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 310 310 + admins: std::collections::BTreeSet::new(), 311 311 + admission: knot_types::AdmissionPolicy::Closed, 304 312 max_pack_bytes, 305 305 - LanguagesPushBudget::new(std::time::Duration::from_secs(2)), 306 306 - None, 307 307 - ); 313 313 + archive_limit, 314 314 + languages_push_budget: LanguagesPushBudget::new(std::time::Duration::from_secs(2)), 315 315 + ci_logs: None, 316 316 + }); 308 317 let state = Arc::new(match lfs { 309 318 Some(handle) => base.with_lfs(handle, 2), 310 319 None => base, ··· 470 479 } 471 480 472 481 async fn fixture() -> Fixture { 482 482 + fixture_with_archive_limit(ArchiveLimit::default()).await 483 483 + } 484 484 + 485 485 + async fn fixture_with_archive_limit(archive_limit: ArchiveLimit) -> Fixture { 473 486 let scratch = tempfile::tempdir().unwrap(); 474 487 let (key_path, public_line) = keygen(scratch.path(), "client"); 475 475 - let (server, index) = spawn_server(public_line, MaxWireBytes::new(1 << 30)).await; 488 488 + let (server, index, _, _) = spawn_server_core( 489 489 + public_line, 490 490 + MaxWireBytes::new(1 << 30), 491 491 + archive_limit, 492 492 + true, 493 493 + None, 494 494 + ) 495 495 + .await; 476 496 let url = format!( 477 497 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}", 478 498 server.port ··· 1081 1101 knot_events::ReplayBytes::new(16 << 20).unwrap(), 1082 1102 ), 1083 1103 )); 1084 1084 - let state = Arc::new(knot_ssh::SshState::new( 1104 1104 + let state = Arc::new(knot_ssh::SshState::new(knot_ssh::SshConfig { 1085 1105 layout, 1086 1106 index, 1087 1107 atproto, 1088 1088 - actor_for_seed(77), 1108 1108 + knot_actor: actor_for_seed(77), 1089 1109 events, 1090 1090 - knot_types::KnotHostname::new("knot.test").unwrap(), 1091 1091 - knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 1092 1092 - std::collections::BTreeSet::new(), 1093 1093 - knot_types::AdmissionPolicy::Closed, 1094 1094 - MaxWireBytes::new(1 << 30), 1095 1095 - LanguagesPushBudget::new(std::time::Duration::from_secs(2)), 1096 1096 - None, 1097 1097 - )); 1110 1110 + hostname: knot_types::KnotHostname::new("knot.test").unwrap(), 1111 1111 + appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), 1112 1112 + admins: std::collections::BTreeSet::new(), 1113 1113 + admission: knot_types::AdmissionPolicy::Closed, 1114 1114 + max_pack_bytes: MaxWireBytes::new(1 << 30), 1115 1115 + archive_limit: ArchiveLimit::default(), 1116 1116 + languages_push_budget: LanguagesPushBudget::new(std::time::Duration::from_secs(2)), 1117 1117 + ci_logs: None, 1118 1118 + })); 1098 1119 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); 1099 1120 let port = listener.local_addr().unwrap().port(); 1100 1121 tokio::spawn(async move { ··· 1339 1360 let fx = fixture().await; 1340 1361 seed_work(&fx.work); 1341 1362 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; 1342 1342 - assert!(ok, "seeding push must land before archiving:\n{out}"); 1363 1363 + assert!(ok, "the seeding push must succeed before archiving:\n{out}"); 1343 1364 1344 1365 let out_tar = fx.scratch.path().join("archive.tar"); 1345 1366 let (ok, out) = git_ssh( ··· 1362 1383 assert!( 1363 1384 knot_fixtures::contains(&tar, b"README.md"), 1364 1385 "archived tar must contain the README.md entry" 1386 1386 + ); 1387 1387 + } 1388 1388 + 1389 1389 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1390 1390 + async fn git_archive_remote_over_ssh_honors_the_configured_archive_limit() { 1391 1391 + let fx = fixture_with_archive_limit(ArchiveLimit::new(512)).await; 1392 1392 + seed_work(&fx.work); 1393 1393 + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; 1394 1394 + assert!(ok, "the seeding push must succeed before archiving:\n{out}"); 1395 1395 + 1396 1396 + let out_tar = fx.scratch.path().join("archive.tar"); 1397 1397 + let (ok, out) = git_ssh( 1398 1398 + &fx.work, 1399 1399 + &fx.key_path, 1400 1400 + &[ 1401 1401 + "archive", 1402 1402 + "--format=tar", 1403 1403 + "--remote", 1404 1404 + &fx.url, 1405 1405 + "-o", 1406 1406 + out_tar.to_str().unwrap(), 1407 1407 + "HEAD", 1408 1408 + ], 1409 1409 + ) 1410 1410 + .await; 1411 1411 + assert!(!ok, "git archive --remote past the limit must fail:\n{out}"); 1412 1412 + assert!( 1413 1413 + out.contains("archive exceeds the 512 byte limit"), 1414 1414 + "the refusal must reach the client over the ssh channel:\n{out}" 1365 1415 ); 1366 1416 } 1367 1417 ··· 1480 1530 let (server, _index, shutdown, serve_task) = spawn_server_core( 1481 1531 public_line, 1482 1532 MaxWireBytes::new(1 << 20), 1533 1533 + ArchiveLimit::default(), 1483 1534 true, 1484 1535 Some(handle.clone()), 1485 1536 )

+1 -1

knot2/crates/knot-types/src/lib.rs

View file Reviewed

··· 20 20 pub use hex::{decode_hex, lowercase_hex}; 21 21 22 22 mod net; 23 23 - pub use net::forwarded_peer; 23 23 + pub use net::{PeerKey, ProxyTrust, TrustedProxies}; 24 24 25 25 pub use jacquard_common::CowStr; 26 26 pub use jacquard_common::DefaultStr;

+318 -11

knot2/crates/knot-types/src/net.rs

View file Reviewed

··· 1 1 + use std::collections::BTreeSet; 1 2 use std::net::IpAddr; 2 3 3 3 - use http::HeaderMap; 4 4 - use http::header::AsHeaderName; 4 4 + use http::{HeaderMap, HeaderName}; 5 5 + 6 6 + #[derive(Debug, Clone, Default, PartialEq, Eq)] 7 7 + pub struct TrustedProxies(BTreeSet<IpAddr>); 8 8 + 9 9 + impl TrustedProxies { 10 10 + pub fn new(addresses: impl IntoIterator<Item = IpAddr>) -> Self { 11 11 + Self( 12 12 + addresses 13 13 + .into_iter() 14 14 + .map(|peer| peer.to_canonical()) 15 15 + .collect(), 16 16 + ) 17 17 + } 18 18 + 19 19 + pub fn trusts(&self, peer: Option<IpAddr>) -> bool { 20 20 + match peer { 21 21 + _ if self.0.is_empty() => true, 22 22 + Some(peer) => self.0.contains(&peer.to_canonical()), 23 23 + None => false, 24 24 + } 25 25 + } 26 26 + } 27 27 + 28 28 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 29 29 + pub enum PeerKey { 30 30 + Relayed(IpAddr), 31 31 + Socket(IpAddr), 32 32 + SocketWithIgnoredHeader(IpAddr), 33 33 + Unidentified, 34 34 + } 35 35 + 36 36 + impl PeerKey { 37 37 + pub fn address(self) -> Option<IpAddr> { 38 38 + match self { 39 39 + Self::Relayed(peer) | Self::Socket(peer) | Self::SocketWithIgnoredHeader(peer) => { 40 40 + Some(peer) 41 41 + } 42 42 + Self::Unidentified => None, 43 43 + } 44 44 + } 45 45 + 46 46 + pub fn ignored_header(self) -> Option<IpAddr> { 47 47 + match self { 48 48 + Self::SocketWithIgnoredHeader(peer) => Some(peer), 49 49 + _ => None, 50 50 + } 51 51 + } 52 52 + } 53 53 + 54 54 + #[derive(Debug, Clone, Default)] 55 55 + pub struct ProxyTrust { 56 56 + header: Option<HeaderName>, 57 57 + proxies: TrustedProxies, 58 58 + } 59 59 + 60 60 + impl ProxyTrust { 61 61 + pub fn new(header: Option<HeaderName>, proxies: TrustedProxies) -> Self { 62 62 + Self { header, proxies } 63 63 + } 64 64 + 65 65 + pub fn trusts_any_peer(&self) -> bool { 66 66 + self.header.is_some() && self.proxies.trusts(None) 67 67 + } 68 68 + 69 69 + pub fn peer_key(&self, headers: &HeaderMap, socket: Option<IpAddr>) -> PeerKey { 70 70 + match (self.relayed_peer(headers, socket), socket) { 71 71 + (Some(relayed), _) => PeerKey::Relayed(relayed.to_canonical()), 72 72 + (None, None) => PeerKey::Unidentified, 73 73 + (None, Some(socket)) => match self.ignores_header_from(headers, socket) { 74 74 + true => PeerKey::SocketWithIgnoredHeader(socket.to_canonical()), 75 75 + false => PeerKey::Socket(socket.to_canonical()), 76 76 + }, 77 77 + } 78 78 + } 79 79 + 80 80 + pub fn client_peer(&self, headers: &HeaderMap, socket: Option<IpAddr>) -> Option<IpAddr> { 81 81 + self.peer_key(headers, socket).address() 82 82 + } 83 83 + 84 84 + pub fn client_peer_of(&self, headers: &HeaderMap, socket: IpAddr) -> IpAddr { 85 85 + self.peer_key(headers, Some(socket)) 86 86 + .address() 87 87 + .unwrap_or(socket.to_canonical()) 88 88 + } 89 89 + 90 90 + fn relayed_peer(&self, headers: &HeaderMap, socket: Option<IpAddr>) -> Option<IpAddr> { 91 91 + self.header 92 92 + .as_ref() 93 93 + .filter(|_| self.proxies.trusts(socket)) 94 94 + .and_then(|header| forwarded_peer(headers, header)) 95 95 + } 96 96 + 97 97 + fn ignores_header_from(&self, headers: &HeaderMap, socket: IpAddr) -> bool { 98 98 + self.header 99 99 + .as_ref() 100 100 + .is_some_and(|header| headers.contains_key(header)) 101 101 + && !self.proxies.trusts(Some(socket)) 102 102 + } 103 103 + } 5 104 6 6 - pub fn forwarded_peer<K: AsHeaderName>(headers: &HeaderMap, header: K) -> Option<IpAddr> { 105 105 + fn forwarded_peer(headers: &HeaderMap, header: &HeaderName) -> Option<IpAddr> { 7 106 headers 8 107 .get(header) 9 108 .and_then(|value| value.to_str().ok()) ··· 16 115 mod tests { 17 116 use super::*; 18 117 19 19 - use http::HeaderName; 20 20 - 21 118 fn headers(value: Option<&str>) -> HeaderMap { 22 119 value 23 120 .map(|value| { 24 121 let mut map = HeaderMap::new(); 25 25 - map.insert( 26 26 - HeaderName::from_bytes(b"x-forwarded-for").unwrap(), 27 27 - value.parse().unwrap(), 28 28 - ); 122 122 + map.insert(forwarded_for(), value.parse().unwrap()); 29 123 map 30 124 }) 31 125 .unwrap_or_default() 32 126 } 33 127 128 128 + fn forwarded_for() -> HeaderName { 129 129 + HeaderName::from_static("x-forwarded-for") 130 130 + } 131 131 + 132 132 + fn ip(value: &str) -> IpAddr { 133 133 + value.parse().unwrap() 134 134 + } 135 135 + 34 136 #[test] 35 137 fn forwarded_peer_takes_the_rightmost_parseable_entry() { 36 138 [ ··· 42 144 .iter() 43 145 .for_each(|&(header, expected)| { 44 146 assert_eq!( 45 45 - forwarded_peer(&headers(header), "x-forwarded-for"), 46 46 - expected.map(|ip| ip.parse::<IpAddr>().unwrap()), 147 147 + forwarded_peer(&headers(header), &forwarded_for()), 148 148 + expected.map(ip), 47 149 "{header:?}" 48 150 ); 49 151 }); 152 152 + } 153 153 + 154 154 + #[test] 155 155 + fn an_empty_allowlist_trusts_every_peer() { 156 156 + let anyone = TrustedProxies::default(); 157 157 + assert!(anyone.trusts(Some(ip("203.0.113.7")))); 158 158 + assert!(anyone.trusts(None)); 159 159 + } 160 160 + 161 161 + #[test] 162 162 + fn the_header_applies_only_to_a_peer_on_the_allowlist() { 163 163 + let proxy = ip("127.0.0.1"); 164 164 + let forged = headers(Some("198.51.100.4")); 165 165 + let trust = ProxyTrust::new(Some(forwarded_for()), TrustedProxies::new([proxy])); 166 166 + let peer = |socket| trust.client_peer(&forged, Some(socket)); 167 167 + 168 168 + assert_eq!( 169 169 + peer(proxy), 170 170 + Some(ip("198.51.100.4")), 171 171 + "a request relayed by the listed proxy is limited by the address the proxy recorded" 172 172 + ); 173 173 + assert_eq!( 174 174 + peer(ip("203.0.113.7")), 175 175 + Some(ip("203.0.113.7")), 176 176 + "a client reaching the knot directly forged the header and must answer for its socket" 177 177 + ); 178 178 + } 179 179 + 180 180 + #[test] 181 181 + fn a_caller_with_a_socket_address_gets_the_same_answer_without_an_option() { 182 182 + let listed = TrustedProxies::new([ip("127.0.0.1")]); 183 183 + [ 184 184 + (ProxyTrust::default(), Some("198.51.100.4")), 185 185 + ( 186 186 + ProxyTrust::new(Some(forwarded_for()), listed), 187 187 + Some("198.51.100.4"), 188 188 + ), 189 189 + ( 190 190 + ProxyTrust::new(Some(forwarded_for()), TrustedProxies::default()), 191 191 + None, 192 192 + ), 193 193 + ] 194 194 + .into_iter() 195 195 + .for_each(|(trust, value)| { 196 196 + let headers = headers(value); 197 197 + [ip("127.0.0.1"), ip("203.0.113.7"), ip("::ffff:203.0.113.7")] 198 198 + .into_iter() 199 199 + .for_each(|socket| { 200 200 + assert_eq!( 201 201 + Some(trust.client_peer_of(&headers, socket)), 202 202 + trust.client_peer(&headers, Some(socket)), 203 203 + "{trust:?} disagreed with itself for {socket} and header {value:?}" 204 204 + ); 205 205 + }); 206 206 + }); 207 207 + } 208 208 + 209 209 + #[test] 210 210 + fn a_listed_ipv4_proxy_still_matches_the_v4_mapped_address_a_dual_stack_listener_reports() { 211 211 + let mapped = ip("::ffff:127.0.0.1"); 212 212 + assert!( 213 213 + TrustedProxies::new([ip("127.0.0.1")]).trusts(Some(mapped)), 214 214 + "binding [::] turns an IPv4 proxy into ::ffff:127.0.0.1 and the allowlist must still match it" 215 215 + ); 216 216 + assert!( 217 217 + TrustedProxies::new([mapped]).trusts(Some(ip("127.0.0.1"))), 218 218 + "an operator who writes the mapped form must match a plain IPv4 peer too" 219 219 + ); 220 220 + assert!( 221 221 + !TrustedProxies::new([ip("127.0.0.1")]).trusts(Some(ip("::1"))), 222 222 + "the IPv6 loopback is a different address from the IPv4 one" 223 223 + ); 224 224 + } 225 225 + 226 226 + #[test] 227 227 + fn client_peer_falls_back_to_the_socket_whenever_no_header_applies() { 228 228 + let socket = ip("203.0.113.7"); 229 229 + [ 230 230 + (None, Some("198.51.100.4")), 231 231 + (Some(forwarded_for()), None), 232 232 + (Some(forwarded_for()), Some("not-an-ip")), 233 233 + ] 234 234 + .into_iter() 235 235 + .for_each(|(header_name, header_value)| { 236 236 + let trust = ProxyTrust::new(header_name.clone(), TrustedProxies::default()); 237 237 + assert_eq!( 238 238 + trust.client_peer(&headers(header_value), Some(socket)), 239 239 + Some(socket), 240 240 + "{header_name:?} with {header_value:?}" 241 241 + ); 242 242 + }); 243 243 + } 244 244 + 245 245 + #[test] 246 246 + fn one_address_gets_one_bucket_however_the_listener_spelled_it() { 247 247 + let trust = ProxyTrust::default(); 248 248 + assert_eq!( 249 249 + trust.client_peer(&headers(None), Some(ip("::ffff:203.0.113.7"))), 250 250 + trust.client_peer(&headers(None), Some(ip("203.0.113.7"))), 251 251 + "a v4-mapped socket and the plain v4 address are one client, so they share a key" 252 252 + ); 253 253 + } 254 254 + 255 255 + #[test] 256 256 + fn client_peer_reports_no_peer_when_an_allowlist_leaves_it_with_neither_source() { 257 257 + let trust = ProxyTrust::new( 258 258 + Some(forwarded_for()), 259 259 + TrustedProxies::new([ip("127.0.0.1")]), 260 260 + ); 261 261 + assert_eq!( 262 262 + trust.client_peer(&headers(Some("198.51.100.4")), None), 263 263 + None, 264 264 + "with no socket to check against the allowlist there is no client to key on" 265 265 + ); 266 266 + } 267 267 + 268 268 + #[test] 269 269 + fn the_peer_key_separates_an_ignored_header_from_a_request_that_never_sent_one() { 270 270 + let listed = ProxyTrust::new( 271 271 + Some(forwarded_for()), 272 272 + TrustedProxies::new([ip("127.0.0.1")]), 273 273 + ); 274 274 + assert_eq!( 275 275 + listed.peer_key(&headers(Some("198.51.100.4")), Some(ip("203.0.113.7"))), 276 276 + PeerKey::SocketWithIgnoredHeader(ip("203.0.113.7")), 277 277 + "an unlisted peer sent the header, which is the address an operator has to see" 278 278 + ); 279 279 + assert_eq!( 280 280 + listed.peer_key(&headers(None), Some(ip("203.0.113.7"))), 281 281 + PeerKey::Socket(ip("203.0.113.7")), 282 282 + "a request without the header says nothing about the allowlist" 283 283 + ); 284 284 + assert_eq!( 285 285 + listed.peer_key(&headers(Some("198.51.100.4")), Some(ip("127.0.0.1"))), 286 286 + PeerKey::Relayed(ip("198.51.100.4")), 287 287 + "the listed proxy relayed this one" 288 288 + ); 289 289 + assert_eq!( 290 290 + listed.peer_key(&headers(Some("198.51.100.4")), None), 291 291 + PeerKey::Unidentified 292 292 + ); 293 293 + } 294 294 + 295 295 + #[test] 296 296 + fn only_an_ignored_header_reports_an_address_to_warn_about() { 297 297 + assert_eq!( 298 298 + PeerKey::SocketWithIgnoredHeader(ip("203.0.113.7")).ignored_header(), 299 299 + Some(ip("203.0.113.7")) 300 300 + ); 301 301 + [ 302 302 + PeerKey::Relayed(ip("198.51.100.4")), 303 303 + PeerKey::Socket(ip("203.0.113.7")), 304 304 + PeerKey::Unidentified, 305 305 + ] 306 306 + .into_iter() 307 307 + .for_each(|key| { 308 308 + assert_eq!( 309 309 + key.ignored_header(), 310 310 + None, 311 311 + "{key:?} is not a misconfigured allowlist" 312 312 + ); 313 313 + }); 314 314 + } 315 315 + 316 316 + #[test] 317 317 + fn an_ignored_header_still_keys_the_peer_on_its_socket() { 318 318 + let listed = ProxyTrust::new( 319 319 + Some(forwarded_for()), 320 320 + TrustedProxies::new([ip("127.0.0.1")]), 321 321 + ); 322 322 + let forged = headers(Some("198.51.100.4")); 323 323 + assert_eq!( 324 324 + listed.client_peer(&forged, Some(ip("203.0.113.7"))), 325 325 + Some(ip("203.0.113.7")) 326 326 + ); 327 327 + assert_eq!( 328 328 + listed.client_peer_of(&forged, ip("::ffff:203.0.113.7")), 329 329 + ip("203.0.113.7"), 330 330 + "the reported address stays canonical so the warning and the bucket agree" 331 331 + ); 332 332 + } 333 333 + 334 334 + #[test] 335 335 + fn a_populated_allowlist_trusts_only_the_addresses_it_lists() { 336 336 + let proxies = TrustedProxies::new([ip("127.0.0.1"), ip("::1")]); 337 337 + assert!(proxies.trusts(Some(ip("127.0.0.1")))); 338 338 + assert!(proxies.trusts(Some(ip("::1")))); 339 339 + assert!( 340 340 + !proxies.trusts(Some(ip("203.0.113.7"))), 341 341 + "a client reaching the knot directly would pick its own rate-limit bucket" 342 342 + ); 343 343 + assert!( 344 344 + !proxies.trusts(None), 345 345 + "a peer of None has no address to match against the list" 346 346 + ); 347 347 + } 348 348 + 349 349 + #[test] 350 350 + fn only_a_header_without_an_allowlist_trusts_any_peer() { 351 351 + let listed = TrustedProxies::new([ip("127.0.0.1")]); 352 352 + assert!( 353 353 + ProxyTrust::new(Some(forwarded_for()), TrustedProxies::default()).trusts_any_peer() 354 354 + ); 355 355 + assert!(!ProxyTrust::new(Some(forwarded_for()), listed).trusts_any_peer()); 356 356 + assert!(!ProxyTrust::default().trusts_any_peer()); 50 357 } 51 358 }

+1

knot2/crates/knot-xrpc/src/error.rs

View file Reviewed

··· 120 120 GitError::AtomicRefs(_) => Self::conflict(message), 121 121 GitError::UnsafeRepoDid(_) | GitError::ReservedDid(_) => Self::invalid_request(message), 122 122 GitError::DepthExceeded(_) => Self::invalid_request(message), 123 123 + GitError::ArchiveTooLarge { .. } => Self::request_too_large(message), 123 124 GitError::Selection(_) => Self::overloaded(message), 124 125 // Every oid passed to the object database here came from a ref this 125 126 // knot already resolved or a tree it already read, so a miss means

+1 -5

knot2/crates/knot-xrpc/src/events.rs

View file Reviewed

··· 39 39 Query(query): Query<EventsQuery>, 40 40 upgrade: WebSocketUpgrade, 41 41 ) -> Response { 42 42 - let peer = state 43 43 - .trusted_proxy_header 44 44 - .as_ref() 45 45 - .and_then(|header| knot_types::forwarded_peer(&headers, header)) 46 46 - .unwrap_or_else(|| socket_peer.ip()); 42 42 + let peer = state.proxy_trust.client_peer_of(&headers, socket_peer.ip()); 47 43 let Some(permit) = state.subscriber_gate.try_admit(peer) else { 48 44 return XrpcError::overloaded( 49 45 "knot is serving its maximum number of event subscribers, retry shortly",

+7 -17

knot2/crates/knot-xrpc/src/lib.rs

View file Reviewed

··· 57 57 58 58 use knot_atproto::{Atproto, AtprotoError, ServiceJwt}; 59 59 use knot_events::{EventLog, SubscriberGate}; 60 60 + pub use knot_git::ArchiveLimit; 60 61 use knot_git::Layout; 61 62 use knot_index::{Index, Resolved}; 62 63 use knot_maintenance::MaintenanceHandle; ··· 97 98 pub struct PatchLimit(usize); 98 99 pub struct PatchDecompressedLimit(u64); 99 100 pub struct ResponseLimit(usize); 100 100 - pub struct ArchiveLimit(u64); 101 101 pub struct ForkPackLimit(u64); 102 102 pub struct TreeReadBudget(ReadBudget); 103 103 pub struct BlobReadBudget(ReadBudget); ··· 122 122 patch: PatchLimit::new(16 * 1024 * 1024), 123 123 patch_decompressed: PatchDecompressedLimit::new(128 * 1024 * 1024), 124 124 response: ResponseLimit::new(5 * 1024 * 1024), 125 125 - archive: ArchiveLimit::new(1024 * 1024 * 1024), 125 125 + archive: ArchiveLimit::default(), 126 126 fork_pack: ForkPackLimit::new(1024 * 1024 * 1024), 127 127 pack: MaxWireBytes::new(8 * 1024 * 1024 * 1024), 128 128 } ··· 164 164 pub limiter: Arc<PreAuthLimiter>, 165 165 pub cob_locks: Arc<CobLocks>, 166 166 pub reservations: Arc<Reservations>, 167 167 - pub trusted_proxy_header: Option<http::HeaderName>, 167 167 + pub proxy_trust: knot_types::ProxyTrust, 168 168 pub committer: Committer, 169 169 pub byte_limits: ByteLimits, 170 170 pub budgets: Budgets, ··· 336 336 request: Request, 337 337 next: Next, 338 338 ) -> Response { 339 339 - let peer = effective_peer(&state, socket, request.headers()); 339 339 + let peer = state 340 340 + .proxy_trust 341 341 + .client_peer(request.headers(), socket.ip()); 340 342 match admit_pre_auth(&state, peer) { 341 343 Ok(guard) => { 342 344 let response = next.run(request).await; ··· 347 349 } 348 350 } 349 351 350 350 - pub(crate) fn effective_peer<H: HttpTransport, C: Clock>( 351 351 - state: &XrpcState<H, C>, 352 352 - socket: SocketPeer, 353 353 - headers: &HeaderMap, 354 354 - ) -> Option<IpAddr> { 355 355 - state 356 356 - .trusted_proxy_header 357 357 - .as_ref() 358 358 - .and_then(|header| knot_types::forwarded_peer(headers, header)) 359 359 - .or(socket.ip()) 360 360 - } 361 361 - 362 352 pub(crate) fn admit_pre_auth<H: HttpTransport, C: Clock>( 363 353 state: &XrpcState<H, C>, 364 354 peer: Option<IpAddr>, ··· 501 491 repo: &RepoDid, 502 492 denied: &str, 503 493 ) -> Result<AccountDid, XrpcError> { 504 504 - let peer = effective_peer(state, socket, headers); 494 494 + let peer = state.proxy_trust.client_peer(headers, socket.ip()); 505 495 let guard = admit_pre_auth(state, peer)?; 506 496 let actor = state.authenticate_push(headers).await?; 507 497 guard.refund();

+14 -48

knot2/crates/knot-xrpc/src/reads.rs

View file Reviewed

··· 1 1 use std::collections::BTreeMap; 2 2 - use std::io::{Seek, SeekFrom}; 3 2 use std::sync::Arc; 4 3 5 4 use axum::body::Body; ··· 56 55 const LIST_REPOS_MAX: usize = 1000; 57 56 const MAX_BLOB_BYTES: u64 = 25 * 1024 * 1024; 58 57 const MAX_COMPARE_COMMITS: usize = 500; 59 59 - const ARCHIVE_CAP_MESSAGE: &str = "archive exceeds configured maximum size"; 60 58 const RAW_CSP: &str = "default-src 'none'; style-src 'unsafe-inline'; sandbox"; 61 59 62 60 pub(crate) fn repo_not_found() -> XrpcError { ··· 1213 1211 } 1214 1212 } 1215 1213 1216 1216 - struct BoundedSpool { 1217 1217 - file: std::fs::File, 1218 1218 - position: u64, 1219 1219 - limit: u64, 1220 1220 - tripped: bool, 1221 1221 - } 1222 1222 - 1223 1223 - impl std::io::Write for BoundedSpool { 1224 1224 - fn write(&mut self, data: &[u8]) -> std::io::Result<usize> { 1225 1225 - if self.position.saturating_add(data.len() as u64) > self.limit { 1226 1226 - self.tripped = true; 1227 1227 - return Err(std::io::Error::new( 1228 1228 - std::io::ErrorKind::WriteZero, 1229 1229 - ARCHIVE_CAP_MESSAGE, 1230 1230 - )); 1231 1231 - } 1232 1232 - let written = self.file.write(data)?; 1233 1233 - self.position += written as u64; 1234 1234 - Ok(written) 1235 1235 - } 1236 1236 - 1237 1237 - fn flush(&mut self) -> std::io::Result<()> { 1238 1238 - self.file.flush() 1239 1239 - } 1240 1240 - } 1241 1241 - 1242 1242 - impl Seek for BoundedSpool { 1243 1243 - fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> { 1244 1244 - let position = self.file.seek(pos)?; 1245 1245 - self.position = position; 1246 1246 - Ok(position) 1247 1247 - } 1248 1248 - } 1249 1249 - 1250 1214 fn archive_etag(did: &RepoDid, commit: Oid, format: ArchiveFormat, prefix: &str) -> String { 1251 1215 let mut hasher = Sha256::new(); 1252 1216 hasher.update(did.as_str().as_bytes()); ··· 1333 1297 let temp = run_blocking({ 1334 1298 let layout = state.layout.clone(); 1335 1299 let did = did.clone(); 1336 1336 - let archive_limit = state.byte_limits.archive.get(); 1300 1300 + let archive_limit = state.byte_limits.archive; 1337 1301 let tree_prefix = knot_git::ArchivePrefix::new(format!("{archive_prefix}/")) 1338 1302 .expect("validated prefix with trailing slash stays valid"); 1339 1303 move || { ··· 1341 1305 let tree = repo.peel_to_tree(resolved)?; 1342 1306 let temp = tempfile::NamedTempFile::new() 1343 1307 .map_err(|error| XrpcError::internal(format!("cannot spool archive: {error}")))?; 1344 1344 - let file = temp 1308 1308 + let mut file = temp 1345 1309 .reopen() 1346 1310 .map_err(|error| XrpcError::internal(format!("cannot spool archive: {error}")))?; 1347 1347 - let mut spool = BoundedSpool { 1348 1348 - file, 1349 1349 - position: 0, 1350 1350 - limit: archive_limit, 1351 1351 - tripped: false, 1352 1352 - }; 1353 1353 - repo.write_archive(tree, format.format(), Some(&tree_prefix), &mut spool) 1354 1354 - .map_err(|error| match spool.tripped { 1355 1355 - true => XrpcError::request_too_large(ARCHIVE_CAP_MESSAGE), 1311 1311 + repo.write_archive( 1312 1312 + tree, 1313 1313 + format.format(), 1314 1314 + Some(&tree_prefix), 1315 1315 + archive_limit, 1316 1316 + &mut file, 1317 1317 + ) 1318 1318 + .map_err(|error| { 1319 1319 + match matches!(error, knot_git::GitError::ArchiveTooLarge { .. }) { 1320 1320 + true => XrpcError::from(error), 1356 1321 false => XrpcError::named( 1357 1322 StatusCode::BAD_REQUEST, 1358 1323 "ArchiveError", 1359 1324 format!("failed to create archive: {error}"), 1360 1325 ), 1361 1361 - })?; 1326 1326 + } 1327 1327 + })?; 1362 1328 temp.as_file() 1363 1329 .set_modified(pinned_modified(modified_secs)) 1364 1330 .map_err(|error| XrpcError::internal(error.to_string()))?;

+1 -1

knot2/crates/knot-xrpc/src/tests.rs

View file Reviewed

··· 331 331 limiter: Arc::new(crate::PreAuthLimiter::default()), 332 332 cob_locks: Arc::new(crate::CobLocks::default()), 333 333 reservations, 334 334 - trusted_proxy_header: None, 334 334 + proxy_trust: knot_types::ProxyTrust::default(), 335 335 committer: crate::Committer { 336 336 name: AuthorName::new("Tangled"), 337 337 email: Email::new("noreply@tangled.sh"),

+1 -1

knot2/crates/knot-xrpc/tests/common/mod.rs

View file Reviewed

··· 184 184 PerActorQuota::new(16), 185 185 GlobalQuota::new(16), 186 186 )), 187 187 - trusted_proxy_header: None, 187 187 + proxy_trust: knot_types::ProxyTrust::default(), 188 188 committer: knot_xrpc::Committer { 189 189 name: AuthorName::new("Tangled"), 190 190 email: Email::new("noreply@tangled.sh"),

+20

knot2/example.toml

View file Reviewed

··· 175 175 # Default value: 5242880 176 176 #max_response_bytes = 5242880 177 177 178 178 + # Upper bound on bytes that a single archive spools, 179 179 + # across all our surfaces: the sh.tangled.repo.archive query, 180 180 + # `git archive --remote` over SSH, 181 181 + # and the smart HTTP archive route. 182 182 + # The knot will refuse writing smth that would blast an archive past this bound. 183 183 + # 178 184 # Can also be specified via environment variable `KNOT_XRPC_MAX_ARCHIVE_BYTES`. 185 185 + # 179 186 # Default value: 1073741824 180 187 #max_archive_bytes = 1073741824 181 188 ··· 265 272 # 266 273 # Can also be specified via environment variable `KNOT_XRPC_TRUSTED_PROXY_HEADER`. 267 274 #trusted_proxy_header = 275 275 + 276 276 + # IP addresses whose `trusted_proxy_header` the knot honors, 277 277 + # without a port, 278 278 + # for ex the loopback address of a reverse proxy on the same host. 279 279 + # The knot rate-limits a request from any other address 280 280 + # by its own socket address and ignores the header. 281 281 + # Leave empty to honor the header from every peer, 282 282 + # which is safe *only* if nothing but the proxy can reach this knot. 283 283 + # 284 284 + # Can also be specified via environment variable `KNOT_XRPC_TRUSTED_PROXIES`. 285 285 + # 286 286 + # Default value: [] 287 287 + #trusted_proxies = [] 268 288 269 289 # Can also be specified via environment variable `KNOT_XRPC_EVENTS_REPLAY_BUFFER`. 270 290 # Default value: 4096