worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git

RSproxy.rs5.2 KBDownload

1	// Proxying a request to a tenant.
2	//
3	// Streamed in both directions. A clone is a multi-gigabyte response and a push is a
4	// multi-gigabyte request; buffering either would undo the work the transport layer
5	// does to avoid exactly that, and would do it at the hop where there are N tenants
6	// rather than one.
7	//
8	// Every proxied request carries a fresh signed assertion (`account::assertion`). It
9	// is single-use and bound to this method, path and body, so a tenant that logs a
10	// header, or an operator who reads one, holds nothing reusable.
11
12	use crate::account::assertion::{header, Claims, Signer_};
13	use crate::error::{Error, Result};
14	use crate::http::response::Body;
15	use futures::TryStreamExt;
16	use http_body_util::{BodyExt, StreamBody};
17	use hyper::body::{Frame, Incoming};
18	use hyper::{Request, Response, StatusCode};
19	use std::time::Duration;
20
21	/// Headers that must not be forwarded in either direction.
22	fn is_hop_by_hop(name: &str) -> bool {
23	const BLOCKED: &[&str] = &[
24	"connection",
25	"keep-alive",
26	"transfer-encoding",
27	"upgrade",
28	"proxy-authenticate",
29	"proxy-authorization",
30	"te",
31	"trailer",
32	"host",
33	// The caller's credential stops here: a tenant is told who is calling by the
34	// signed assertion, and forwarding the original token would let a tenant
35	// replay it against the control plane.
36	"authorization",
37	];
38	BLOCKED.iter().any(\|b\| name.eq_ignore_ascii_case(b))
39	}
40
41	/// Forward one request to a tenant and stream the answer back.
42	pub async fn forward(
43	client: &reqwest::Client,
44	signer: &Signer_,
45	endpoint: &str,
46	account: &str,
47	request: Request<Incoming>,
48	max_body: u64,
49	) -> Result<Response<Body>> {
50	let method = request.method().clone();
51	let uri = request.uri().clone();
52	let path_and_query = uri
53	.path_and_query()
54	.map(\|p\| p.as_str())
55	.unwrap_or_else(\|\| uri.path());
56	let headers = request.headers().clone();
57
58	// Streamed, not buffered. A push is a multi-gigabyte pack, and this is the hop
59	// with N tenants behind it — buffering here would undo the work the transport
60	// does to avoid exactly that. The assertion is deliberately not bound to the
61	// body for the same reason; its nonce is what stops replay.
62	let _ = max_body;
63	let assertion = signer.sign(&Claims::new(account, method.as_str(), uri.path()));
64
65	let upstream = request
66	.into_body()
67	.into_data_stream()
68	.map_err(std::io::Error::other);
69
70	let mut outbound = client
71	.request(
72	reqwest::Method::from_bytes(method.as_str().as_bytes())
73	.map_err(\|_\| Error::invalid("invalid-method", "unsupported method"))?,
74	format!("{}{}", endpoint.trim_end_matches('/'), path_and_query),
75	)
76	.header(header(), assertion)
77	.body(reqwest::Body::wrap_stream(upstream));
78
79	for (name, value) in headers.iter() {
80	if !is_hop_by_hop(name.as_str()) {
81	outbound = outbound.header(name.as_str(), value.as_bytes());
82	}
83	}
84
85	let response = outbound.send().await.map_err(\|e\| {
86	// A tenant that is up but not answering is not the caller's fault, and is
87	// usually transient — it is still booting, or was just started.
88	eprintln!("[control] proxy to {endpoint} failed: {e}");
89	Error::RateLimited { retry_after: 5 }
90	})?;
91
92	let status =
93	StatusCode::from_u16(response.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
94	let mut builder = Response::builder().status(status);
95	for (name, value) in response.headers().iter() {
96	if !is_hop_by_hop(name.as_str()) {
97	builder = builder.header(name.as_str(), value.as_bytes());
98	}
99	}
100
101	let stream = response
102	.bytes_stream()
103	.map_ok(Frame::data)
104	.map_err(std::io::Error::other);
105
106	builder
107	.body(StreamBody::new(stream).boxed())
108	.map_err(\|e\| Error::Internal(anyhow::Error::from(e).context("build proxy response")))
109	}
110
111	/// A client configured for proxying.
112	///
113	/// No redirect following: a tenant redirecting us is either a bug or an attempt to
114	/// make the control plane fetch something else.
115	pub fn client(timeout: Duration) -> Result<reqwest::Client> {
116	reqwest::Client::builder()
117	.redirect(reqwest::redirect::Policy::none())
118	.connect_timeout(Duration::from_secs(5))
119	.timeout(timeout)
120	.build()
121	.map_err(\|e\| Error::Internal(anyhow::Error::from(e).context("build proxy client")))
122	}
123
124	#[cfg(test)]
125	mod tests {
126	use super::*;
127
128	#[test]
129	fn the_callers_credential_is_not_forwarded_to_the_tenant() {
130	// A tenant learns who is calling from the signed assertion. Forwarding the
131	// original token would let it replay that token against the control plane.
132	assert!(is_hop_by_hop("Authorization"));
133	assert!(is_hop_by_hop("authorization"));
134	}
135
136	#[test]
137	fn hop_by_hop_headers_are_not_forwarded() {
138	for header in ["Connection", "Transfer-Encoding", "Upgrade", "Host", "TE"] {
139	assert!(is_hop_by_hop(header), "{header} must not cross the proxy");
140	}
141	assert!(!is_hop_by_hop("Content-Type"));
142	assert!(!is_hop_by_hop("Git-Protocol"));
143	}
144	}